ConsumerSeekAware broken with upgrade from 1.1.5 to 1.1.6

[tbinias]

Having a listener which implements ConsumerSeekAware annotated with @KafkaListener works fine in 1.1.5 but not in 1.1.6:

@Component
public class EntityChangedEventConsumer implements ConsumerSeekAware {
@KafkaListener(topics = "${kafka.consumer.topic}")
    public void onMessage(ConsumerRecord<String, EntityChangedEvent> data) { ... }
}
   @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
      ...
    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
...
    }

None of the ConsumerSeekAware-EntryPoints are called in 1.1.6.

Since #324 ConsumerSeekAware implementations are detected on this.genericListener

But this.genericListener is always assigned to listener:
line 354 KafkaMessageListenerContainer -> this.genericListener = listener;

The case of having a GenericAcknowledgingMessageListener is lost:
line 331 -> this.theListener = listener == null ? ackListener : listener;

[artembilan]

Fixed via b02ecd6

[guillaumedrolet21]

Hi, using version 1.2.2.RELEASE here and the methods are still not invoked.

Using a ConcurrentKafkaListenerContainerFactory and a listener with KafkaListener annotation and implementing ConsumerSeekAware like described in this issue.

Like @tbinias has described, when arriving at below condition in KafkaMessageListenerContainer in createRebalanceListener method, genericListener is null.

if (ListenerConsumer.this.genericListener instanceof ConsumerSeekAware) { seekPartitions(partitions, false); }

Also, I added the idle interval config like @garyrussell said in this stack overflow thread: https://stackoverflow.com/questions/46321483/spring-kafka-consumer-seek-aware-methods-not-invoked/46325685?newreg=5487961b397b4c7387b56694c6798884

I am using auto commit and I have auto offset reset set to "latest" if that changes anything.

something I am missing?

Thank you very much for your help!

[artembilan]

Well, the fix is there already after release 1.2.2: https://github.com/spring-projects/spring-kafka/commits/1.2.x .
We may release 1.2.3 if that is a strong requirement.

[guillaumedrolet21]

Ok, thank you for quick response

[garyrussell]

My mistake, sorry; my local 1.2.x branch was out of date; I somehow reasoned it wasn't an issue on that branch since I tested with a snapshot.

@artembilan Please go ahead and release 1.2.3 tomorrow, while I am in transit; thanks.

[artembilan]

Spring Kafka 1.2.3 has been released.
You can obtain it from the https://repo.spring.io/release/org/springframework/kafka/spring-kafka/1.2.3.RELEASE/

Will be in the Maven Central in a couple hours.