When working with Apache Kafka consumers in Spring Kafka, managing message offsets is essential. Offsets tell Kafka where the consumer left off, preventing duplicate processing after restarts or failures. Spring Kafka offers fine-grained control over this process through AckMode settings.

Understanding the Challenge: Batch Commit and Potential Issues

Kafka consumers in Spring Boot traditionally used the BATCH commit strategy by default. In this approach, offsets are committed periodically at predefined intervals. While BATCH works well in many scenarios, it can lead to problems when message processing time exceeds the commit interval.

Here's why BATCH commit can be challenging:

• Delayed Commits and Duplicates: If a message takes longer to process than the commit interval, its offset won't be committed until the next interval. If the consumer crashes or restarts before the commit happens, Kafka will redeliver uncommitted messages. This can result in duplicate processing, potentially causing errors or unexpected behavior in your application.

• Performance Degradation: Redelivery due to delayed commits can put additional load on your consumers, impacting overall performance.

These challenges highlight the importance of understanding your message processing needs and choosing the appropriate AckMode for optimal control over offset commits.

Step-by-Step Guide to Committing Offsets in Spring Kafka

Now, let's explore the different AckMode options and how to configure them in Spring Kafka:

1. A New Default:enable.auto.commit Set to False

A key change in recent Spring Kafka versions (starting from 2.3) is the default behavior of enable.auto.commit. Unlike previous versions that inherited Kafka's default of true (automatic commits), Spring Kafka now sets this property to false by default. This shift empowers you to take explicit control over committing offsets.

This means you'll need to configure your consumer to handle commits manually using AckMode settings.

2. Exploring Your Options: AckMode Settings

Spring Kafka provides various AckMode settings to tailor your offset committing strategy. Here's a breakdown of some key options (assuming no transactions):

• RECORD: Commits the offset for each message after successful processing. This offers the most fine-grained control but requires acknowledging each message individually.

• BATCH (Default): This is the new default! Offsets are committed only after all messages retrieved in a single poll() call are processed. It's a good balance between performance and ensuring at-least-once delivery.

• TIME: Commits after processing all messages from a poll(), but only if a specific time (ackTime) has elapsed since the last commit. Useful for scenarios where processing time can vary.

• COUNT: Commits after processing all messages from a poll(), but only if a certain number (ackCount) of messages have been received since the last commit. Suitable for high-volume message streams.

• COUNT_TIME: A combination, committing if either the time threshold or message count is met. Offers flexibility for diverse processing needs.

• MANUAL: The listener acknowledges each message with Acknowledgment.acknowledge(). Commit behavior then follows BATCH logic. Provides fine-grained control but requires manual acknowledgement in your listener code.

• MANUAL_IMMEDIATE: Commits the offset the moment the listener calls Acknowledgment.acknowledge(). Delivers the strongest guarantee of at-least-once delivery but requires acknowledging each message.

3. Spring Kafka Consumer Configuration with Code Examples:

Let's delve into code to see how to configure consumers for specific AckMode settings. We'll create separate classes for Batch and MANUAL_IMMEDIATE configurations:

3.a) Batch Commit Configuration:

This configuration uses the new default AckMode.BATCH. Offsets are committed after processing all messages from a single poll() call.

@Configuration
@EnableKafkaStreams
public class BatchConsumerConfig {

    @Value("${spring.kafka.consumer.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // Explicitly set

        // Batch Commit Configuration
        props.put(ConsumerConfig.ACK_MODE_CONFIG, AckMode.BATCH.toString());

        // Key and Value Deserializers (replace with your actual classes)
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        return new DefaultConsumerFactory<>(props);
    }

    @Bean
    @KafkaListener(topics = "myTopic", groupId = "${spring.kafka.consumer.group-id}")
    public void consumeBatch(ConsumerRecords<String, String> records) {
        // Process records here for Batch Commit
        for (ConsumerRecord<String, String> record : records) {
            String key = record.key();
            String value = record.value();

            // Process message
            System.out.println("Key: " + key + ", Value: " + value);
        }
    }
}

3.b) MANUAL_IMMEDIATE Commit Configuration:

To address the offset commit delay issue, we can switch to using the MANUAL_IMMEDIATE commit strategy provided by Spring Kafka. This strategy allows consumers to manually commit offsets immediately after processing each message, ensuring timely offset commits and reducing the risk of message reprocessing.

@Configuration
@EnableKafkaStreams
public class ManualImmediateConsumerConfig {

    @Value("${spring.kafka.consumer.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // Explicitly set

        // Manual Commit with Immediate Acknowledgement Configuration
        props.put(ConsumerConfig.ACK_MODE_CONFIG, AckMode.MANUAL_IMMEDIATE.toString());

        // Key and Value Deserializers (replace with your actual classes)
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        return new DefaultConsumerFactory<>(props);
    }

    @Bean
    @KafkaListener(topics = "myTopic", groupId = "${spring.kafka.consumer.group-id}")
    public void consumeManualImmediate(ConsumerRecords<String, String> records) {
        // Process records here for Manual Commit with Immediate Acknowledgement
        for (ConsumerRecord<String, String> record : records) {
            String key = record.key();
            String value = record.value();

            // Process message
            System.out.println("Key: " + key + ", Value: " + value);

            // Acknowledge each message for MANUAL_IMMEDIATE
            record.acknowledge();
        }
    }
}