ResumeRebalanceListener.java

Overview

`ResumeRebalanceListener` is a Java class implementing Kafka's `ConsumerRebalanceListener` interface. It is part of the Apache Camel Kafka component, specifically supporting the **resume strategies** mechanism that handles offset management for Kafka consumers.

The primary responsibility of this class is to listen to partition rebalance events on Kafka consumers and integrate with the resume offset management system. It ensures that offsets are committed appropriately when partitions are revoked and that the consumer resumes from the correct offsets when partitions are assigned.

This class acts as a bridge between Kafka's rebalance lifecycle and the resume strategy logic, coordinating commit operations and resume actions to maintain fault-tolerant and consistent message processing.

Package

package org.apache.camel.component.kafka.consumer.support.resume;

Dependencies

Kafka Classes:
- ConsumerRebalanceListener
- Consumer
- TopicPartition
Apache Camel Kafka Component:
- KafkaConfiguration
- CommitManager
- ResumeStrategy
- KafkaResumeAdapter (adapter for resume strategy integration)
Logging: org.slf4j.Logger and LoggerFactory

Class: ResumeRebalanceListener

public class ResumeRebalanceListener implements ConsumerRebalanceListener

Description

This class listens to Kafka consumer partition rebalance events. It handles:

Partition Revocation: Commits offsets for revoked partitions if auto-commit is disabled.
Partition Assignment: Triggers the resume mechanism to reposition the consumer according to the cached resume offsets.

Fields

Field Name	Type	Description
`LOG`	Logger	Logger instance for debugging and info messages.
`threadId`	String	Identifier for the consumer thread using this listener.
`configuration`	KafkaConfiguration	Kafka consumer configuration, used to determine commit behavior.
`commitManager`	CommitManager	Manages offset commit operations outside Kafka's auto-commit.
`resumeAdapter`	KafkaResumeAdapter	Adapter responsible for handling resume offset state and consumer.

Constructor

public ResumeRebalanceListener(String threadId,
                               KafkaConfiguration configuration,
                               CommitManager commitManager,
                               Consumer<?, ?> consumer,
                               ResumeStrategy resumeStrategy)

Parameters

threadId: String identifier for the thread or consumer instance.
configuration: Kafka consumer configuration object.
commitManager: Component managing manual offset commits.
consumer: Kafka consumer instance to be controlled by the resume adapter.
resumeStrategy: Resume strategy instance providing the adapter.

Behavior

Initializes the listener with the given thread ID and configuration.
Retrieves a KafkaResumeAdapter from the resume strategy.
Sets the Kafka consumer on the resume adapter, enabling resume operations.

Methods

onPartitionsRevoked

@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions)

Description: Called when partitions are revoked from the consumer due to rebalance.
Parameters:
- partitions: Collection of TopicPartition objects revoked.
Behavior:
- Logs debug information for each revoked partition.
- If Kafka auto-commit is disabled (autoCommitEnable == false), commits offsets for each revoked partition using the CommitManager.
Usage:
Ensures that offsets for revoked partitions are safely committed, preventing message reprocessing or data loss.

onPartitionsAssigned

@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions)

Description: Called when partitions are assigned to the consumer after rebalance.
Parameters:
- partitions: Collection of TopicPartition objects assigned.
Behavior:
- Logs debug information for each assigned partition.
- Triggers the resume process on the resumeAdapter to reposition the consumer offsets according to persisted resume state.
Usage:
Ensures consumers start reading from the correct offsets after partition assignment, maintaining continuity.

Important Implementation Details

Integration with Resume Strategy:
The resumeAdapter is a critical component integrating the resume offset logic. It controls the consumer's position on assigned partitions by seeking offsets from a persisted resume cache.
Commit Management:
The listener explicitly commits offsets for revoked partitions only if auto-commit is disabled. This prevents duplicate commits and ensures that custom commit logic (via CommitManager) is applied correctly.
Logging:
Uses the SLF4J logger at debug level to trace partition revocation and assignment events, tagged with the consumer thread ID for clarity in multi-threaded environments.
Thread Identification:
The threadId allows distinguishing logs and operations per consumer thread, which is useful when multiple consumers run concurrently.

Interaction with Other Components

Kafka Consumer:
The listener hooks into the Kafka consumer lifecycle events (rebalance callbacks) and controls offset commit and resume behavior during these events.
CommitManager:
Responsible for committing offsets manually when auto-commit is disabled.
ResumeStrategy & KafkaResumeAdapter:
These components manage offset persistence and resumption, allowing the consumer to restore its position after failures or restarts.
KafkaConfiguration:
Provides configuration flags such as autoCommitEnable that influence whether commits happen automatically or need manual intervention.

Usage Example

// Assume these components are created and configured elsewhere
String threadId = "consumer-thread-1";
KafkaConfiguration kafkaConfig = new KafkaConfiguration();
CommitManager commitManager = new CommitManager();
Consumer<String, String> kafkaConsumer = createKafkaConsumer();
ResumeStrategy resumeStrategy = getResumeStrategyInstance();

// Instantiate the rebalance listener
ResumeRebalanceListener rebalanceListener = new ResumeRebalanceListener(
    threadId,
    kafkaConfig,
    commitManager,
    kafkaConsumer,
    resumeStrategy
);

// Attach listener to Kafka consumer
kafkaConsumer.subscribe(
    topics,
    rebalanceListener
);

In this example, the listener will commit offsets when partitions are revoked and trigger resume logic when partitions are assigned.

Visual Diagram: Class Structure

classDiagram
    class ResumeRebalanceListener {
        - static final Logger LOG
        - String threadId
        - KafkaConfiguration configuration
        - CommitManager commitManager
        - KafkaResumeAdapter resumeAdapter
        + ResumeRebalanceListener(String, KafkaConfiguration, CommitManager, Consumer<?,?>, ResumeStrategy)
        + void onPartitionsRevoked(Collection~TopicPartition~)
        + void onPartitionsAssigned(Collection~TopicPartition~)
    }
    ResumeRebalanceListener ..> KafkaConfiguration : uses
    ResumeRebalanceListener ..> CommitManager : uses
    ResumeRebalanceListener ..> KafkaResumeAdapter : uses
    ResumeRebalanceListener ..> Consumer : uses
    ResumeRebalanceListener ..> ResumeStrategy : uses

Summary

`ResumeRebalanceListener` is a specialized Kafka consumer rebalance listener tailored for the Apache Camel Kafka component's resume offset strategies. It:

Commits offsets manually during partition revocation if auto-commit is off.
Invokes resume logic to restore consumer position on partition assignment.
Integrates Kafka consumer lifecycle events with advanced offset management.
Supports fault-tolerant, resumable consumption scenarios by cooperating with commit managers and resume adapters.

This class plays a crucial role in enabling the fault-tolerant behavior of Kafka consumers by ensuring offsets are safely managed across rebalances, contributing to reliable message processing in distributed systems.