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 -> {
        String record = exchange.getInputPayloadAs(String.class);
        // process the Kafka record
    })
    .toSink("slf4j://log")
    .build();

The KafkaPollingConsumer polls on a dedicated thread via KafkaConsumerService.

URL format

kafka://<topic-name>[?<params>]

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:

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

With Spring Boot, the configurer can be defined as a @Bean.

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.