Spring Kafka is a project that applies core Spring concepts to the development of Kafka-based messaging solutions. It provides a high-level abstraction for Kafka-based messaging solutions and simplifies the use of Kafka APIs for sending and receiving messages.

Kafka Consumer and Polling

In Kafka, a consumer retrieves messages from Kafka servers by calling the poll() method of the Kafka consumer API. This method fetches any available messages from the server(s) for all the topics and partitions that the consumer is subscribed to. The poll() method is pivotal in controlling how and when consumers retrieve messages.

Integration in Spring Kafka

Spring Kafka manages the lifecycle of the Kafka consumer in its listener container. This listener container handles the setup of the Kafka consumer, the receiving of messages, and the delegation of those messages to a listener (usually a method annotated with @KafkaListener). The core component responsible for these operations is the KafkaMessageListenerContainer.

In Spring Kafka, you don’t call the poll() method directly. Instead, this method is invoked internally by Spring's listener container. Here's a simplified view of how this happens:

1. Initialization: Once a KafkaListenerContainer is created and started, it establishes a connection with Kafka brokers and joins the consumer group.
2. Poll Loop: The KafkaMessageListenerContainer enters a polling loop, where it continually calls the poll() method on the Kafka consumer at a specified interval.
3. Message Handling: Each time poll() is called, it returns a batch of messages (if available). Spring Kafka then dispatches these messages to the appropriate message listener.

Configuration

The behavior of the poll() method, such as how long it waits for messages, and how much data it tries to fetch, can be controlled by several consumer configuration settings:

• fetch.min.bytes: This sets the minimum amount of data that the server should return for a fetch request. If not enough data is available, the request will wait until sufficient data is accumulated.
• fetch.max.wait.ms: This sets the maximum time the server will block before answering the fetch request if there isn't sufficient data to meet the fetch.min.bytes requirement.
• max.poll.records: This limits the number of records returned in a single call to poll.

Example Configuration in Application.yml

1spring:
2  kafka:
3    consumer:
4      bootstrap-servers: localhost:9092
5      group-id: my-group
6      auto-offset-reset: earliest
7      enable-auto-commit: true
8      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
9      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
10      properties:
11        max.poll.records: 10
12        fetch.min.bytes: 500
13        fetch.max.wait.ms: 1000

Special Considerations

• Long Polling: If fetch.min.bytes is high compared to the size of messages being produced to the topic, a call to poll() might block for a long duration (up to fetch.max.wait.ms), which could delay message processing.
• Heartbeats: Kafka consumers need to send heartbeats to the Kafka brokers periodically to maintain membership in their consumer groups. If poll() does not get called within a specific interval (set by session.timeout.ms), the consumer can be considered dead by the brokers and it might be kicked out of the group.

Summary Table

Utilizing Spring Kafka abstracts much of the complexity associated with direct Kafka API interaction, allowing developers to focus more on business logic rather than messaging infrastructure details. Understanding when and how Consumer.poll() is invoked can help in tuning your consumer configurations for optimal performance.