This version is still in development and is not considered stable yet.

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-M1</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();

KafkaConsumerService runs its own poll loop against the Kafka client on a dedicated thread, then hands each record to the flow through the same durable-inbox/dispatch machinery every event-driven source uses.

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"))

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 KafkaConsumerService 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.

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.