KafkaResumable.java

Overview

The `KafkaResumable` class is part of the Apache Camel Kafka component, under the package `org.apache.camel.component.kafka.consumer.support`. It implements the `Resumable` interface, providing a mechanism to represent and manage resumable offsets specifically for Kafka message consumption.

This class encapsulates the Kafka partition and offset information, enabling the Camel Kafka consumer to resume message consumption from a specific offset within a partition. It provides methods to retrieve the last processed offset and the key identifying the offset, facilitating reliable and consistent message processing with resume capabilities.

Class: KafkaResumable

Description

`KafkaResumable` is a simple immutable value object that holds Kafka partition and offset information. It implements the `Resumable` interface to integrate with Apache Camel's resume support framework, which manages stateful resumption of message consumption.

Package

org.apache.camel.component.kafka.consumer.support

Implements

org.apache.camel.resume.Resumable

Fields

Constructor

public KafkaResumable(String partition, String offset)

Parameters

• partition - The Kafka partition identifier as a string.

• offset - The Kafka offset for the partition as a string.

Description

Creates a new `KafkaResumable` instance representing a specific partition and offset from which consumption can be resumed.

Usage Example

KafkaResumable resumable = new KafkaResumable("partition-0", "12345");

Methods

getLastOffset

@Override
public Offset<String> getLastOffset()

• Returns: Offset<String>

Returns the last consumed offset wrapped in an `Offset` object. The offset is represented as a string and encapsulated via the utility method `Offsets.of(offset)`.

Usage Example

Offset<String> lastOffset = resumable.getLastOffset();
System.out.println("Last offset: " + lastOffset.getValue());

getOffsetKey

@Override
public OffsetKey<?> getOffsetKey()

• Returns: OffsetKey<?>

Returns an unmodifiable offset key representing the Kafka partition. This key uniquely identifies the offset within the resume framework and is created using `OffsetKeys.unmodifiableOf(partition)`.

Usage Example

OffsetKey<?> offsetKey = resumable.getOffsetKey();
System.out.println("Offset key (partition): " + offsetKey.getKey());

Important Implementation Details

• The class is immutable with final fields to ensure thread safety and consistent state.

• It leverages Apache Camel's resume infrastructure (Offset, OffsetKey, Offsets, and OffsetKeys) to abstract offset handling.

• The offset and partition are represented as strings to maintain flexibility and compatibility with Kafka's offset representations.

• getLastOffset() and getOffsetKey() provide integration points for the resume mechanism to persist and retrieve consumption state reliably.

Interaction with Other Components

• Apache Camel Resume Framework:
  The KafkaResumable class implements Resumable interface, allowing Kafka consumer components in Camel to use this to track and resume consumption offsets.

• Kafka Consumer Component:
  This class is utilized by Kafka consumers within Camel to represent the resumption point for message consumption, coordinating with offset storage and recovery mechanisms.

• Offset and OffsetKey Utilities:
  Uses utility classes from org.apache.camel.support.resume to create immutable and type-safe offset and key objects.

Visual Diagram

classDiagram
    class KafkaResumable {
        -partition: String
        -offset: String
        +KafkaResumable(partition: String, offset: String)
        +getLastOffset(): Offset<String>
        +getOffsetKey(): OffsetKey<?>
    }
    KafkaResumable ..|> Resumable

    class Resumable {
        <<interface>>
        +getLastOffset(): Offset<?>
        +getOffsetKey(): OffsetKey<?>
    }

    class Offset~T~ {
        +getValue(): T
    }

    class OffsetKey~T~ {
        +getKey(): T
    }

    class Offsets {
        +of(value: String): Offset<String>
    }

    class OffsetKeys {
        +unmodifiableOf(key: String): OffsetKey<String>
    }

    KafkaResumable ..> Offset : uses
    KafkaResumable ..> OffsetKey : uses
    KafkaResumable ..> Offsets : uses
    KafkaResumable ..> OffsetKeys : uses

Summary

`KafkaResumable` is a minimalistic, immutable class designed to encapsulate Kafka partition and offset information for Apache Camel's Kafka consumer resume functionality. By implementing the `Resumable` interface, it standardizes how offsets are tracked and resumed, leveraging Camel's offset utilities to maintain clean and reliable state management. This class plays a crucial role in enabling fault-tolerant Kafka consumers within the Camel ecosystem.