How do I custom consumer Interceptor for one record

[rajeshhereforyou]

Hi,

I'm trying to add a custom consumer interceptor using 'org.apache.kafka.clients.consumer.ConsumerInterceptor'. The problem I've here is the 'onConsume' method is taking 'ConsumerRecords<String, Object> records' but i'm looking for intercepting only one record at a time instead of a bunch of records.

How do i do that? Please suggest

[artembilan]

The org.apache.kafka.clients.consumer.ConsumerInterceptor class is not a part of this project. And we don't provide Apache Kafka support here.
More over we don't handle question here on GitHub. Would be better to ask similar questions on the public forums where broader community may assist you.

I don't know the answer for you from Apache Kafka terms, but with Spring Kafka it sounds like a plain @KafkaListener for record-based consumer is the way to go.

[garyrussell]

The Kafka Consumer API only deals with ConsumerRecords so it makes sense that their interceptor does the same.

Exactly what do you want to do in a per-record interceptor?

There may be some other way to do what you want and/or we might want to add a record interceptor to this framework if there is a reasonable use case.

[rajeshhereforyou]

Let me try my best to explain one of the use cases we have for this particular requirement.

We are trying to transport some custom headers for logging stuff like request initiator/ request id and have some monitoring use cases like monitoring/track the request flow across multiple layers of the request. In this case, we need to inject these custom headers per record so that I can carry forward/move them across the complete request chain.

[garyrussell]

OK; we can add a record interceptor to the listener container.

[rajeshhereforyou]

I will try to add more in case of RabbitMq Spring framework provides class
MessagePostProcessor that we can register with Listener Container like
container.setAfterReceivePostProcessors(rabbitListenerMessagePostProcessor); This helps us to achieve movement of tracing information. I believe what you suggested above will solve our problem.

[rajeshhereforyou]

Thanks for the quick response. Meanwhile we will try to figure a workaround and I'm curious to know if you have any design in mind so that we can contribute or any release in your thoughts?

[garyrussell]

It's already done - it's pretty trivial. I was planning on putting it in the upcoming 2.3 release only, but since it's so trivial, I don't see why we can't put it in 2.2.7 which will be out Friday or Monday.

[garyrussell]

As a work around, can't you iterate over the ConsumerRecords instead?