Add seek management support for the Kafka Inbound Channel Adapters

[alpert]

Hi.

This may be a generic question about Kafka and not directly related to spring integration but without manual offset management what will happen if i close and open my application(that using spring integration kafka) again? Will it continue where it left? Will it continue from beginning or end? What should I do if i want it to read from beginning, end of where it left?

[artembilan]

The Spring Integration Kafka is fully based now on the Spring for Apache Kafka: https://spring.io/projects/spring-kafka

The KafkaMessageDrivenChannelAdapter and KafkaInboundGateway fully rely on the MessageListenerContainer therefore all the offset management comes from there: https://docs.spring.io/spring-kafka/docs/2.1.9.RELEASE/reference/html/_reference.html#committing-offsets.
If you use a enable.auto.commit = true , then you should understand that Kafka Client commits offsets in async manner, therefore it is indeed that after closing app and restarting you may start to consume from some earlier offset.

To re-read from the beginning you need to use a ConsumerSeekCallback: https://docs.spring.io/spring-kafka/docs/2.1.9.RELEASE/reference/html/_reference.html#seek.

Well, it is used, but looks like in case of KafkaMessageDrivenChannelAdapter and KafkaInboundGateway we don't provide any hooks for that functionality.
And from here it sounds like we need to provide for those Channel Adapter a functionality like setConsumerSeekAware(). So, you will be able to do the same what can be done now with the @KafkaListener and ConsumerSeekAware impl on that class.

Saying that I'd recommend you as a workaround to use a @KafkaListener for time being and send messages to the channel from the method impl or delegate to some @MessagingGateway.

Does it make sense?

@garyrussell , WDYT?

[garyrussell]

Yes; we should add seek capability to the adapter.

[harshita-mangly-mohan]

Hi, @garyrussell @artembilan,

I am using ConcurrentMessageListenerContainer with KafkaMessageDrivenChannelAdapter for an application Kafka implementation, in a use case to replay older messages from a kafka topic I plan to follow ->
Find offset by timestamp, seek and then continue polling from that offset

I was using the ConsumerSeekCallback.seekToTimestamp​ given by setOnPartitionsAssignedSeekCallback() hook, for which I get a multi thread access on consumer ERROR [ConcurrentModificationException], I believe the consumerCallback returned by setOnPartitionsAssignedSeekCallback() method should be used only in that method scope for seek to be triggered on the respective consumer thread , won't we need the 'ListenerConsumer' reference returned from registerSeekCallback() which is not available with KafkaMessageDrivenChannelAdapter to perform seek from the corresponding consumer thread at a later point in time?

Or can you please suggest if I am missing something in following this example form docs -> https://docs.spring.io/spring-kafka/reference/html/#seek

[garyrussell]

This enhancement was only to support seeks during partition assignment; not at arbitrary times later.

More work is required to support that; I suggest you open a new issue for a new feature.