Spring Kafka -Consumer seek aware methods not invoked

Viewed 2557

I am implementing ConsumerSeekAware in my listener class. But overridden methods registerSeekCallback,onPartitionsAssigned,onIdleContainer never get invoked. Below are my consumer config and consumer files. Could not identify the mistake.

using Spring-kafka : 1.1.6 version.

Consumer:

package com.test.kafka.binder;

import java.util.Map;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.stereotype.Component;

@Component
public class Consumer implements ConsumerSeekAware {
    
  @KafkaListener(topics = "${kafka.topic}", group = "foo")
  public void listen(String message) {
    System.out.println("Received Messasge in group foo: " + message);
  }
    
  @Override
  public void registerSeekCallback(ConsumerSeekCallback callback) {
    System.out.println("registerSeekCallback");
  }

  @Override
  public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
    System.out.println("onPartitionsAssigned");
  }
    
  @Override
  public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
    System.out.println("onIdleContainer");
  }
}

Consumer Config:

package com.test.kafka.binder;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {
 

     @Value("${kafka.bootstrap-servers}")
     private Object bootstrapAddress;

     
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(
          ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 
          bootstrapAddress);
        props.put(
          ConsumerConfig.GROUP_ID_CONFIG, 
          "test");
        props.put(
          ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
          StringDeserializer.class);
        props.put(
          ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
          StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }
 
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> 
      kafkaListenerContainerFactory() {
    
        ConcurrentKafkaListenerContainerFactory<String, String> factory
          = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}
1 Answers

This is a known problem it is fixed on master. We need to release 1.1.7.

However, if you can upgrade to 1.2.1, it's already fixed there.

Note that you also need to enable idle events:

factory.getContainerProperties().setIdleEventInterval(10_000L);
Related