Kafka Connector

Overview

Connecting streams to Kafka with Reactive Messaging is easy to do. There is a standard Kafka client behind the scenes, all the producer and consumer configs can be propagated through messaging config.

Maven Coordinates

To enable Reactive Kafka Connector, add the following dependency to your project’s pom.xml (see Managing Dependencies).

pom.xml

<dependency>
  <groupId>io.helidon.messaging.kafka</groupId>
  <artifactId>helidon-messaging-kafka</artifactId>
</dependency>

Config

Example of connector config:

mp.messaging:

  incoming.from-kafka:
    connector: helidon-kafka
    topic: messaging-test-topic-1
    auto.offset.reset: latest1
    enable.auto.commit: true
    group.id: example-group-id

  outgoing.to-kafka:
    connector: helidon-kafka
    topic: messaging-test-topic-1

  connector:
    helidon-kafka:
      bootstrap.servers: localhost:90922
      key.serializer: org.apache.kafka.common.serialization.StringSerializer
      value.serializer: org.apache.kafka.common.serialization.StringSerializer
      key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value.deserializer: org.apache.kafka.common.serialization.StringDeserializer

Kafka client consumer’s property auto.offset.reset configuration for from-kafka channel only
Kafka client’s property bootstrap.servers configuration for all channels using the connector

Besides the following configuration options, any property from consumer or producer configuration can be passed to the underlying Kafka client.

Configuration options

Key	Type	Description
`value-deserializer`	`Class`	Deserializer class for value that implements the `org.apache.kafka.common.serialization.Deserializer` interface
`group-id`	`String`	A unique string that identifies the consumer group this consumer belongs to
`auto-offset-reset`	`AutoOffsetReset`	What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server (e.g
`topic-pattern`	`Pattern`	Pattern for topic names to consume from
`acks`	`String`	The number of acknowledgments the producer requires the leader to have received before considering a request complete
`bootstrap-servers`	`String`	A list of host/port pairs to use for establishing the initial connection to the Kafka cluster
`key-serializer`	`Class`	Serializer class for key that implements the `org.apache.kafka.common.serialization.Serializer` interface
`enable-auto-commit`	`Boolean`	If true the consumer's offset will be periodically committed in the background
`batch-size`	`Integer`	The producer will attempt to batch records together into fewer requests whenever multiple records are being sent to the same partition
`dlq-topic`	`List<String>`	Names of the "dead letter queue" topics to be used in case message is nacked
`retries`	`Integer`	Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error
`buffer-memory`	`Long`	The total bytes of memory the producer can use to buffer records waiting to be sent to the server
`poll-timeout`	`Long`	The maximum time to block polling loop in milliseconds
`compression-type`	`String`	The compression type for all data generated by the producer
`topic`	`List<String>`	Names of the topics to consume from
`value-serializer`	`Class`	Serializer class for value that implements the `org.apache.kafka.common.serialization.Serializer` interface
`period-executions`	`Long`	Period between successive executions of polling loop
`key-deserializer`	`Class`	Deserializer class for key that implements the `org.apache.kafka.common.serialization.Deserializer` interface

Consuming Messages

Example of consuming from Kafka:

@Incoming("from-kafka")
public void consumeKafka(String msg) {
    System.out.println("Kafka says: " + msg);
}

Producing Messages

Example of producing to Kafka:

@Outgoing("to-kafka")
public PublisherBuilder<String> produceToKafka() {
    return ReactiveStreams.of("test1", "test2");
}

NACK Strategy

Strategy	Description
Kill channel	Nacked message sends error signal and causes channel failure so Messaging Health check can report it as DOWN
DLQ	Nacked messages are sent to specified dead-letter-queue
Log only	Nacked message is logged and channel continues normally

Kill channel

Default NACK strategy for Kafka connector. When

Dead Letter Queue

Sends nacked messages to error topic, DLQ is well known pattern for dealing with unprocessed messages.

Helidon can derive connection settings for DLQ topic automatically if the error topic is present on the same Kafka cluster. Serializers are derived from deserializers used for consumption org.apache.kafka.common.serialization.StringDeserializer > org.apache.kafka.common.serialization.StringSerializer. Note that the name of the error topic is needed only in this case.

Example of derived DLQ config:

mp.messaging:
  incoming:
    my-channel:
      nack-dlq: dql_topic_name

If a custom connection is needed, then use the 'nack-dlq' key for all the producer configuration.

Example of custom DLQ config:

mp.messaging:
  incoming:
    my-channel:
      nack-dlq:
        topic: dql_topic_name
        bootstrap.servers: localhost:9092
        key.serializer: org.apache.kafka.common.serialization.StringSerializer
        value.serializer: org.apache.kafka.common.serialization.StringSerializer

Log only

Only logs nacked messages and throws them away, offset is committed and channel continues normally consuming subsequent messages.

Example of log only enabled nack strategy:

mp.messaging:
  incoming:
    my-channel:
      nack-log-only: true

Examples

Don’t forget to check out the examples with pre-configured Kafka docker image, for easy testing:

Helidon messaging examples

Edit this page or Report an issue

Reactive Messaging

MicroProfile Reactive Messaging

JMS Connector

Reactive Messaging support for JMS