Spring Kafka: Poll for new messages instead of being notified using `onMessage`

user_947526 0
Viewed 5111

I am using Spring Kafka in my project as it seemed a natural choice in a Spring based project to consume Kafka messages. To consume messages, I can make use of the MessageListener interface. Spring Kafka internally takes care to invoke my onMessage method for each new message.

However, in my setting I prefer to explicitly poll for new messages and work on them sequentially (which will take a few seconds). As a workaround, I might just block inside my onMessage implementation, or buffer the messages internally. However, this seems to go against the core idea of Spring Kafka.

Kafka is designed so that consumers have to poll for new messages, which matches my requirements. Is there a way to make use of this "natural" workflow with Spring Kafka?

Should I refrain from using Spring Kafka for this use case?

The KafkaConsumer documentation states:

For use cases where message processing time varies unpredictably, neither of these options may be sufficient. The recommended way to handle these cases is to move message processing to another thread, which allows the consumer to continue calling poll while the processor is still working. Some care must be taken to ensure that committed offsets do not get ahead of the actual position. Typically, you must disable automatic commits and manually commit processed offsets for records only after the thread has finished handling them (depending on the delivery semantics you need). Note also that you will need to pause the partition so that no new records are received from poll until after thread has finished handling those previously returned.

Related issue: https://github.com/spring-projects/spring-kafka/issues/195

spring-kafka kafka-consumer-api apache-kafka spring
2 Answers

The issue with having to keep polling the consumer has now been resolved (in 0.10.1.x by KIP-62) so that's not an issue any more (as long as you don't exceed the max.poll.interval.ms) which is 5 mins by default but can be increased.

However, if you want to poll yourself, you can still use spring-kafka (e.g. to get the Spring Boot auto configuration goodness if you are using Boot), but you can get a Consumer from the DefaultKafkaConsumerFactory and poll() it directly.

user_12407630

Here is how I do it. This is in the context of an integration test configuration class, which I load in my JUnit with:

@Import(IntegrationTestConfiguration.class)

In my Test class I have the following:

@Autowired
Consumer<String, String> consumer;

In my test configuration class, I have:

@Bean
public Consumer<String, String> consumer() {
    String bootstrapAddress = "server:port";       // fix this 
    String groupId = "my.group";                   // fix this.
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    ConsumerFactory<String, String> fact = new DefaultKafkaConsumerFactory<>(props);
    
    // Create the consumer, subscribe to the topic
    Consumer<String, String> consumer = fact.createConsumer();
    String topic = "my.topic";                     // fix this.
    List<String> topics = new ArrayList<String>();
    topics.add(topic);
    consumer.subscribe(topics);
    return consumer;
}

Finally, in my test, I do:

@Test
public void testSomething() {
    // Do stuff that will publish a message to Kafka

    // Repeat a number of times untill you get the message you want...
    // Or you give up  
    Duration d = Duration.ofSeconds(2);
    ConsumerRecords<String, String> records = consumer.poll(d);


}
user_12411150
Related