Kafka Channel Adapter

The kafka-channel-adapter module provides integration with Apache Kafka 3.6.x.

Dependency

<dependency>
    <groupId>io.pipelite</groupId>
    <artifactId>kafka-channel-adapter</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

Usage as source (consumer)

Pipelite.defineFlow("kafka-consumer")
    .fromSource("kafka://my-topic")
    .process("handle", (exchange, contribution) -> {
        String record = exchange.getInputPayloadAs(String.class);
        // process the Kafka record
    })
    .toSink("slf4j://log")
    .build();

The Kafka source runs its own poll loop against the Kafka client on a dedicated thread, then hands each record to the framework, which makes it durable and queues it before the offsets are committed, as it does for every event source. The consumer is closed by that same thread when the flow stops.

The same settings can be applied through the typed KafkaSourceConfigurer instead of a query string (see Custom Channel Adapters for the general mechanism):

.fromSource("kafka://my-topic", (KafkaSourceConfigurer c) -> c
    .groupId("my-consumer-group")
    .autoOffsetReset("earliest"))

Headers of the record

The record key is always put on the exchange, in the header org.apache.kafka.record_key. The other headers of the record do not reach the flow unless the flow asks for them, see Headers from and to the outside. Besides the headers of the record, these are offered with the names below:

Name Value

Kafka-Topic

The topic the record was read from.

Kafka-Partition

The partition.

Kafka-Offset

The offset of the record in its partition.

Kafka-Timestamp

The timestamp of the record, in milliseconds since the epoch.

 
.fromSource("kafka://orders", (KafkaSourceConfigurer c) -> c
    .propagateHeaders(h -> h.filter("Idempotency-Key", "Kafka-Offset")))

The headers of a Kafka record are bytes: they are read as UTF-8 text. When a name is repeated the last value is kept, and a header with no value is left out.

URL format

kafka://<topic-name>[?group.id=<id>&auto.offset.reset=<policy>]
The query-string keys are Kafka’s own native ConsumerConfig property names (group.id, auto.offset.reset) - the KafkaSourceConfigurer methods above are camelCase Java, but lower into these same native keys underneath, so the source reads one consistent shape regardless of which style built the URL.

Usage as sink (producer)

.toSink("kafka://output-topic")

The DefaultKafkaProducer sends the serialized payload as the Kafka record value, and the header org.apache.kafka.record_key, if the exchange has it, as the record key.

The headers of the exchange are not sent unless the flow chooses them:

.toSink("kafka://output-topic", (KafkaSinkConfigurer c) -> c
    .propagateHeaders(h -> h.filter("Idempotency-Key", "X-Tenant-Id")))

They become the headers of the record, written as UTF-8 text. See Headers from and to the outside.

Configuration

Kafka configuration is provided via KafkaChannelConfigurer, registered on the context:

context.addChannelConfigurer((KafkaChannelConfigurer) config -> {
    config.setBootstrapServers("localhost:9092");
    config.setGroupId("pipelite-consumer-group");
    config.setAutoOffsetReset("earliest");
});

With Spring Boot, register it from a bean implementing PipeliteContextInitializer instead - see Configuration Properties. A plain @Bean returning a KafkaChannelConfigurer is not picked up automatically.

Available KafkaChannelConfiguration properties

Property Default Description

bootstrapServers

localhost:9092

Kafka broker address

groupId

pipelite-default

Consumer group ID

autoOffsetReset

latest

Offset reset policy (earliest / latest)

Serialization

The module includes:

  • JsonSerializer — Serialization of Java payload to JSON via Jackson.

  • JsonDeserializer — Deserialization from JSON to the target type.

  • TypeConverterRegistry — Registry of available converters.