Spring Kafka中ConsumerRebalanceListener实现及消费者对象获取咨询

Hey there! Let's break down your questions one by one, based on your Spring Kafka 2.2.2 setup.

1. Is your current configuration compatible with ConsumerRebalanceListener? #

Your base configuration does meet the prerequisites for using ConsumerRebalanceListener to fix rebalance-related message loss:

• You've disabled auto-commit (ENABLE_AUTO_COMMIT_CONFIG=false), which is mandatory for manual offset management—this is exactly what ConsumerRebalanceListener relies on to safely commit processed offsets before rebalances.
• You're using AckMode.MANUAL, which puts you in control of when offsets are acknowledged, aligning with the manual offset handling required for the listener.
• Your poll and heartbeat settings follow Kafka's best practices:
  • HEARTBEAT_INTERVAL_MS_CONFIG (1000ms) is less than 1/3 of SESSION_TIMEOUT_MS_CONFIG (40000ms), ensuring the broker detects consumer liveness reliably.
  • MAX_POLL_INTERVAL_MS_CONFIG (50000ms) is longer than SESSION_TIMEOUT_MS_CONFIG, preventing false consumer death triggers from long-running message processing.

The only missing piece is actually registering your ConsumerRebalanceListener implementation (we'll cover that later), but your existing config provides the necessary foundation.

2. Other solutions to fix rebalance-related message loss #

Besides ConsumerRebalanceListener, here are additional approaches to mitigate or eliminate message loss during rebalances:

• Switch to AckMode.MANUAL_IMMEDIATE:
  Unlike MANUAL (which batches offset commits after a poll cycle completes), MANUAL_IMMEDIATE commits offsets immediately when you call Acknowledgment.acknowledge(). This reduces the window of uncommitted offsets, lowering the risk of loss if a rebalance hits mid-processing.

• Use Kafka Transactions for Exactly-Once Semantics:
  If your use case requires strict no-loss guarantees, enable Kafka transactions. Configure a transaction ID prefix for producers, set consumer isolation.level to read_committed, and tie offset commits to transaction completion. This ensures offsets are only committed if message processing succeeds, and uncompleted transactions are rolled back on rebalance.

• Optimize poll parameters to reduce rebalance triggers:

  • If message processing is slow, increase MAX_POLL_INTERVAL_MS_CONFIG (e.g., to 300000ms = 5 minutes) to avoid consumers being kicked out of the group for taking too long to process a poll batch.
  • Reduce MAX_POLL_RECORDS_CONFIG to limit the number of messages per poll, shortening processing time and minimizing unprocessed messages during rebalances.

• Static Group Membership (if compatible):
  If your Kafka cluster and client version support it (Kafka 2.3+; note Spring Kafka 2.2.2 uses Kafka client 2.2.x, so you may need a minor upgrade), set a unique group.instance.id for each consumer instance. This lets consumers rejoin the group without triggering a full rebalance, as the broker retains their partition assignments.

3. How to implement ConsumerRebalanceListener and get the Consumer object #

In Spring Kafka 2.2.2, the easiest way to work with rebalance events and access the consumer is via ConsumerAwareRebalanceListener (a Spring extension of the native Kafka listener that gives you direct access to the Consumer object). Here are two common implementation patterns:

Option 1: Register the listener with the ConsumerFactory #

This applies the listener to all consumers created by the factory:

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    // Your existing config
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 50000);
    props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 40000);

    DefaultKafkaConsumerFactory<String, String> factory = new DefaultKafkaConsumerFactory<>(props);
    
    // Register custom rebalance listener
    factory.addListener(new ConsumerAwareRebalanceListener() {
        @Override
        public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
            // Commit processed offsets BEFORE rebalance to avoid loss
            // If you track offsets manually, use consumer.commitSync(yourOffsetMap) instead
            consumer.commitSync();
        }

        @Override
        public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
            // Optional: Reset offsets or initialize state for assigned partitions
            // consumer.seekToBeginning(partitions);
        }
    });

    return factory;
}

Option 2: Attach the listener to a specific container factory #

Use this if you want the listener to apply only to certain @KafkaListener methods:

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    
    // Your existing container config
    factory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.INFO);
    factory.getContainerProperties().setAckOnError(false);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

    // Add rebalance listener to the container
    factory.getContainerProperties().setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {
        @Override
        public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
            consumer.commitSync();
        }

        @Override
        public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
            // Custom logic for partition assignment
        }
    });

    return factory;
}

Accessing the Consumer in your @KafkaListener method #

You can directly inject the Consumer object into your listener method for ad-hoc operations:

@KafkaListener(topics = "your-topic", groupId = "your-group", containerFactory = "kafkaListenerContainerFactory")
public void processMessage(ConsumerRecord<String, String> record, Acknowledgment ack, Consumer<String, String> consumer) {
    // Process your message
    ack.acknowledge();
    
    // Use the consumer object directly if needed (e.g., query offsets, pause/resume partitions)
    // consumer.pause(consumer.assignment());
}

内容的提问来源于stack exchange，提问作者manas ranjan Pradhan