Apache Kafka

The Kafka connector adds support for Kafka to Reactive Messaging. With it you can receive Kafka Records as well as write message into Kafka.

Introduction

Apache Kafka is a popular distributed streaming platform. It lets you:

Publish and subscribe to streams of records, similar to a message queue or enterprise messaging system.
Store streams of records in a fault-tolerant durable way.
Process streams of records as they occur.

The Kafka cluster stores streams of records in categories called topics. Each record consists of a key, a value, and a timestamp.

For more details about Kafka, check the documentation.

Using the Kafka Connector

To use the Kafka Connector, add the following dependency to your project:

<dependency>
  <groupId>io.smallrye.reactive</groupId>
  <artifactId>smallrye-reactive-messaging-kafka</artifactId>
  <version>3.12.1</version>
</dependency>

The connector name is: smallrye-kafka.

So, to indicate that a channel is managed by this connector you need:

# Inbound
mp.messaging.incoming.[channel-name].connector=smallrye-kafka

# Outbound
mp.messaging.outgoing.[channel-name].connector=smallrye-kafka

Receiving Kafka Records

The Kafka Connector retrieves Kafka Records from Kafka Brokers and maps each of them to Reactive Messaging Messages.

Example

Let’s imagine you have a Kafka broker running, and accessible using the kafka:9092 address (by default it would use localhost:9092). Configure your application to receive Kafka records from a Kafka topic on the prices channel as follows:

kafka.bootstrap.servers=kafka:9092      (1)

mp.messaging.incoming.prices.connector=smallrye-kafka       (2)
mp.messaging.incoming.prices.value.deserializer=org.apache.kafka.common.serialization.DoubleDeserializer    (3)
mp.messaging.incoming.prices.broadcast=true     (4)

Configure the broker location. You can configure it globally or per channel
Configure the connector to manage the prices channel
Sets the (Kafka) deserializer to read the record’s value
Make sure that we can receive from more that one consumer (see KafkaPriceConsumer and KafkaPriceMessageConsumer below)

You don’t need to set the Kafka topic. By default, it uses the channel name (prices). You can configure the topic attribute to override it.

Then, your application receives Message<Double>. You can consume the payload directly:

package inbound;

import org.eclipse.microprofile.reactive.messaging.Incoming;

import javax.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class KafkaPriceConsumer {

    @Incoming("prices")
    public void consume(double price) {
        // process your price.
    }

}

Or, you can retrieve the Message<Double>:

package inbound;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;

import javax.enterprise.context.ApplicationScoped;
import java.util.concurrent.CompletionStage;

@ApplicationScoped
public class KafkaPriceMessageConsumer {

    @Incoming("prices")
    public CompletionStage<Void> consume(Message<Double> price) {
        // process your price.

        // Acknowledge the incoming message (commit the offset)
        return price.ack();
    }

}

Deserialization

The deserialization is handled by the underlying Kafka Client. You need to configure the:

mp.messaging.incoming.[channel-name].value.deserializer to configure the value deserializer (mandatory)
mp.messaging.incoming.[channel-name].key.deserializer to configure the key deserializer (optional, default to String)

If you want to use a custom deserializer, add it to your CLASSPATH and configure the associate attribute.

In addition, the Kafka Connector also provides a set of message converters. So you can receive payloads representing records from Kafka using:

Record<K,V> - a pair key/value
ConsumerRecord<K,V> - a structure representing the record with all its metadata

@Incoming("topic-a")
public void consume(Record<String, String> record) {
    String key = record.key(); // Can be `null` if the incoming record has no key
    String value = record.value(); // Can be `null` if the incoming record has no value
}

@Incoming("topic-b")
public void consume(ConsumerRecord<String, String> record) {
    String key = record.key(); // Can be `null` if the incoming record has no key
    String value = record.value(); // Can be `null` if the incoming record has no value
    String topic = record.topic();
    int partition = record.partition();
    // ...
}

Inbound Metadata

Messages coming from Kafka contains an instance of IncomingKafkaRecordMetadata<K, T> in the metadata. K is the type of the record’s key. T is the type of the record’s value. It provides the key, topic, partitions, headers and so on:

IncomingKafkaRecordMetadata<String, Double> metadata = incoming.getMetadata(IncomingKafkaRecordMetadata.class)
    .orElse(null);
if (metadata != null) {
    // The topic
    String topic = metadata.getTopic();

    // The key
    String key = metadata.getKey();

    // The timestamp
    Instant timestamp = metadata.getTimestamp();

    // The underlying record
    ConsumerRecord<String, Double> record = metadata.getRecord();

    // ...
}

Acknowledgement

When a message produced from a Kafka record is acknowledged, the connector invokes a commit strategy. These strategies decide when the consumer offset for a specific topic/partition is committed. Committing an offset indicates that all previous records have been processed. It is also the position where the application would restart the processing after a crash recovery or a restart.

Committing every offset has performance penalties as Kafka offset management can be slow. However, not committing the offset often enough may lead to message duplication if the application crashes between two commits.

The Kafka connector supports three strategies:

throttled keeps track of received messages and commit to the next offset after the latest acked message in sequence. This strategy guarantees at-least-once delivery even if the channel performs asynchronous processing. The connector tracks the received records and periodically (period specified by auto.commit.interval.ms (default: 5000)) commits the highest consecutive offset. The connector will be marked as unhealthy if a message associated with a record is not acknowledged in throttled.unprocessed-record-max-age.ms (default: 60000). Indeed, this strategy cannot commit the offset as soon as a single record processing fails (see failure-strategy to configure what happens on failing processing). If throttled.unprocessed-record-max-age.ms is set to less than or equal to 0, it does not perform any health check verification. Such a setting might lead to running out of memory if there are poison pill messages. This strategy is the default if enable.auto.commit is not explicitly set to true.
latest commits the record offset received by the Kafka consumer as soon as the associated message is acknowledged (if the offset is higher than the previously committed offset). This strategy provides at-least-once delivery if the channel processes the message without performing any asynchronous processing. This strategy should not be used on high-load as offset commit is expensive. However, it reduces the risk of duplicates.
ignore performs no commit. This strategy is the default strategy when the consumer is explicitly configured with enable.auto.commit to true. It delegates the offset commit to the Kafka client. When enable.auto.commit is true this strategy DOES NOT guarantee at-least-once delivery. However, if the processing failed between two commits, messages received after the commit and before the failure will be re-processed.

The Kafka connector disables the Kafka auto commit if not explicitly enabled. This behavior differs from the traditional Kafka consumer.

If high-throughout is important for you, and not limited by the downstream, we recommend to either:

Use the throttled policy
or set enable.auto.commit to true and annotate the consuming method with @Acknowledgment(Acknowledgment.Strategy.NONE)

Failure Management

If a message produced from a Kafka record is nacked, a failure strategy is applied. The Kafka connector supports 3 strategies:

fail - fail the application, no more records will be processed. (default) The offset of the record that has not been processed correctly is not committed.
ignore - the failure is logged, but the processing continue. The offset of the record that has not been processed correctly is committed.
dead-letter-queue - the offset of the record that has not been processed correctly is committed, but the record is written to a (Kafka) dead letter queue topic.

The strategy is selected using the failure-strategy attribute.

In the case of dead-letter-queue, you can configure the following attributes:

dead-letter-queue.topic: the topic to use to write the records not processed correctly, default is dead-letter-topic-$channel, with $channel being the name of the channel.
dead-letter-queue.key.serializer: the serializer used to write the record key on the dead letter queue. By default, it deduces the serializer from the key deserializer.
dead-letter-queue.value.serializer: the serializer used to write the record value on the dead letter queue. By default, it deduces the serializer from the value deserializer.

The record written on the dead letter topic contains the original record’s headers, as well as a set of additional headers about the original record:

dead-letter-reason - the reason of the failure (the Throwable passed to nack())
dead-letter-cause - the cause of the failure (the getCause() of the Throwable passed to nack()), if any
dead-letter-topic - the original topic of the record
dead-letter-partition - the original partition of the record (integer mapped to String)
dead-letter-offset - the original offset of the record (long mapped to String)

When using dead-letter-queue, it is also possible to change some metadata of the record that is sent to the dead letter topic. To do that, use the Message.nack(Throwable, Metadata) method:

@Incoming("in")
public CompletionStage<Void> consume(KafkaRecord<String, String> message) {
    return message.nack(new Exception("Failed!"), Metadata.of(
        OutgoingKafkaRecordMetadata.builder()
            .withKey("failed-record")
            .withHeaders(new RecordHeaders()
                .add("my-header", "my-header-value".getBytes(StandardCharsets.UTF_8))
            )
    ));
}

The Metadata may contain an instance of OutgoingKafkaRecordMetadata. If the instance is present, the following properties will be used:

key; if not present, the original record’s key will be used
topic; if not present, the configured dead letter topic will be used
partition; if not present, partition will be assigned automatically
headers; combined with the original record’s headers, as well as the dead-letter-* headers described above

Retrying processing

You can combine Reactive Messaging with SmallRye Fault Tolerance, and retry processing when it fails:

@Incoming("kafka")
@Outgoing("processed")
@Retry(delay = 10, maxRetries = 5)
public String process(String v) {
   // ... retry if this method throws an exception
}

You can configure the delay, the number of retries, the jitter…

If your method returns a Uni, you need to add the @NonBlocking annotation:

@Incoming("kafka")
@Outgoing("processed")
@Retry(delay = 10, maxRetries = 5)
@NonBlocking
public Uni<String> process(String v) {
   // ... retry if this method throws an exception or the returned Uni produce a failure
}

The incoming messages are acknowledged only once the processing completes successfully. So, it commits the offset after the successful processing. If after the retries the processing still failed, the message is nacked and the failure strategy is applied.

You can also use @Retry on methods only consuming incoming messages:

@Incoming("kafka")
@Retry(delay = 10, maxRetries = 5)
public void consume(String v) {
   // ... retry if this method throws an exception
}

Handling deserialization failures

Because deserialization happens before creating a Message, the failure strategy presented above cannot be applied. However, when a deserialization failure occurs, you can intercept it and provide a fallback value. To achieve this, create a CDI bean implementing the DeserializationFailureHandler<T> interface:

@ApplicationScoped
@Identifier("failure-retry") // Set the name of the failure handler
public class MyDeserializationFailureHandler
    implements DeserializationFailureHandler<JsonObject> { // Specify the expected type

    @Override
    public JsonObject decorateDeserialization(Uni<JsonObject> deserialization, String topic, boolean isKey,
            String deserializer, byte[] data, Headers headers) {
        return deserialization
                    .onFailure().retry().atMost(3)
                    .await().atMost(Duration.ofMillis(200));
    }
}

The bean must be exposed with the @Identifier qualifier specifying the name of the bean. Then, in the connector configuration, specify the following attribute:

mp.messaging.incoming.$channel.key-deserialization-failure-handler: name of the bean handling deserialization failures happening for the record’s key
mp.messaging.incoming.$channel.value-deserialization-failure-handler: name of the bean handling deserialization failures happening for the record’s value,

The handler is called with the deserialization action as a Uni<T>, the record’s topic, a boolean indicating whether the failure happened on a key, the class name of the deserializer that throws the exception, the corrupted data, the exception, and the records headers augmented with headers describing the failure (which ease the write to a dead letter). On the deserialization Uni failure strategies like retry, providing a fallback value or applying timeout can be implemented. Note that the method must await on the result and return the deserialized object. Alternatively, the handler can only implement handleDeserializationFailure method and provide a fallback value, which may be null.

If you don’t configure a deserialization failure handlers and a deserialization failure happens, the application is marked unhealthy. You can also ignore the failure, which will log the exception and produce a null value. To enable this behavior, set the mp.messaging.incoming.$channel.fail-on-deserialization-failure attribute to false.

Receiving Cloud Events

The Kafka connector supports Cloud Events. When the connector detects a structured or binary Cloud Events, it adds a IncomingKafkaCloudEventMetadata<K, T> in the metadata of the Message. IncomingKafkaCloudEventMetadata contains the various (mandatory and optional) Cloud Event attributes.

If the connector cannot extract the Cloud Event metadata, it sends the Message without the metadata.

Binary Cloud Events

For binary Cloud Events, all mandatory Cloud Event attributes must be set in the record header, prefixed by ce_ (as mandated by the protocol binding). The connector considers headers starting with the ce_ prefix but not listed in the specification as extensions. You can access them using the getExtension method from IncomingKafkaCloudEventMetadata. You can retrieve them as String.

The datacontenttype attribute is mapped to the content-type header of the record. The partitionkey attribute is mapped to the record’s key, if any.

Note that all headers are read as UTF-8.

With binary Cloud Events, the record’s key and value can use any deserializer.

Structured Cloud Events

For structured Cloud Events, the event is encoded in the record’s value. Only JSON is supported, so your event must be encoded as JSON in the record’s value.

Structured Cloud Event must set the content-type header of the record to application/cloudevents or prefix the value with application/cloudevents such as: application/cloudevents+json; charset=UTF-8.

To receive structured Cloud Events, your value deserializer must be:

org.apache.kafka.common.serialization.StringDeserializer
org.apache.kafka.common.serialization.ByteArrayDeserializer
io.vertx.kafka.client.serialization.JsonObjectDeserializer

As mentioned previously, the value must be a valid JSON object containing at least all the mandatory Cloud Events attributes.

If the record is a structured Cloud Event, the created Message’s payload is the Cloud Event data.

The partitionkey attribute is mapped to the record’s key if any.

Consumer Rebalance Listener

To handle offset commit and assigned partitions yourself, you can provide a consumer rebalance listener. To achieve this, implement the io.smallrye.reactive.messaging.kafka.KafkaConsumerRebalanceListener interface, make the implementing class a bean, and add the @Identifier qualifier. A usual use case is to store offset in a separate data store to implement exactly-once semantic, or starting the processing at a specific offset.

The listener is invoked every time the consumer topic/partition assignment changes. For example, when the application starts, it invokes the partitionsAssigned callback with the initial set of topics/partitions associated with the consumer. If, later, this set changes, it calls the partitionsRevoked and partitionsAssigned callbacks again, so you can implement custom logic.

Note that the rebalance listener methods are called from the Kafka polling thread and must block the caller thread until completion. That’s because the rebalance protocol has synchronization barriers, and using asynchronous code in a rebalance listener may be executed after the synchronization barrier.

When topics/partitions are assigned or revoked from a consumer, it pauses the message delivery and restarts once the rebalance completes.

If the rebalance listener handles offset commit on behalf of the user (using the ignore commit strategy), the rebalance listener must commit the offset synchronously in the partitionsRevoked callback. We also recommend applying the same logic when the application stops.

Unlike the ConsumerRebalanceListener from Apache Kafka, the io.smallrye.reactive.messaging.kafka.KafkaConsumerRebalanceListener methods pass the Kafka Consumer and the set of topics/partitions.

Example

In this example we set-up a consumer that always starts on messages from at most 10 minutes ago (or offset 0). First we need to provide a bean that implements the io.smallrye.reactive.messaging.kafka.KafkaConsumerRebalanceListener interface and is annotated with @Identifier. We then must configure our inbound connector to use this named bean.

package inbound;

import io.smallrye.common.annotation.Identifier;
import io.smallrye.reactive.messaging.kafka.KafkaConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;

import javax.enterprise.context.ApplicationScoped;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Logger;

@ApplicationScoped
@Identifier("rebalanced-example.rebalancer")
public class KafkaRebalancedConsumerRebalanceListener implements KafkaConsumerRebalanceListener {

    private static final Logger LOGGER = Logger.getLogger(KafkaRebalancedConsumerRebalanceListener.class.getName());

    /**
     * When receiving a list of partitions will search for the earliest offset within 10 minutes
     * and seek the consumer to it.
     *
     * @param consumer   underlying consumer
     * @param partitions set of assigned topic partitions
     */
    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer,
        Collection<org.apache.kafka.common.TopicPartition> partitions) {
        long now = System.currentTimeMillis();
        long shouldStartAt = now - 600_000L; //10 minute ago

        Map<org.apache.kafka.common.TopicPartition, Long> request = new HashMap<>();
        for (org.apache.kafka.common.TopicPartition partition : partitions) {
            LOGGER.info("Assigned " + partition);
            request.put(partition, shouldStartAt);
        }
        Map<org.apache.kafka.common.TopicPartition, OffsetAndTimestamp> offsets = consumer
            .offsetsForTimes(request);
        for (Map.Entry<org.apache.kafka.common.TopicPartition, OffsetAndTimestamp> position : offsets.entrySet()) {
            long target = position.getValue() == null ? 0L : position.getValue().offset();
            LOGGER.info("Seeking position " + target + " for " + position.getKey());
            consumer.seek(position.getKey(), target);
        }
    }

}

package inbound;

import io.smallrye.reactive.messaging.kafka.IncomingKafkaRecord;
import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
import org.eclipse.microprofile.reactive.messaging.Incoming;

import javax.enterprise.context.ApplicationScoped;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

@ApplicationScoped
public class KafkaRebalancedConsumer {

    @Incoming("rebalanced-example")
    @Acknowledgment(Acknowledgment.Strategy.NONE)
    public CompletionStage<Void> consume(IncomingKafkaRecord<Integer, String> message) {
        // We don't need to ACK messages because in this example we set offset during consumer re-balance
        return CompletableFuture.completedFuture(null);
    }

}

To configure the inbound connector to use the provided listener we either set the consumer rebalance listener’s name:

mp.messaging.incoming.rebalanced-example.consumer-rebalance-listener.name=rebalanced-example.rebalancer

Or have the listener’s name be the same as the group id:

mp.messaging.incoming.rebalanced-example.group.id=rebalanced-example.rebalancer

Setting the consumer rebalance listener’s name takes precedence over using the group id.

Receiving Kafka Records in Batches

By default, incoming methods receive each Kafka record individually. Under the hood, Kafka consumer clients poll the broker constantly and receive records in batches, presented inside the ConsumerRecords container.

In batch mode, your application can receive all the records returned by the consumer poll in one go.

To achieve this you need to set mp.messaging.incoming.$channel.batch=true and specify a compatible container type to receive all the data:

@Incoming("prices")
public void consume(List<Double> prices) {
    for (double price : prices) {
        // process price
    }
}

The incoming method can also receive Message<List<Payload>, KafkaBatchRecords<Payload> ConsumerRecords<Key, Payload> types, They give access to record details such as offset or timestamp :

@Incoming("prices")
public CompletionStage<Void> consumeMessage(KafkaRecordBatch<String, Double> records) {
    for (KafkaRecord<String, Double> record : records) {
        record.getMetadata(IncomingKafkaRecordMetadata.class).ifPresent(metadata -> {
            int partition = metadata.getPartition();
            long offset = metadata.getOffset();
            Instant timestamp = metadata.getTimestamp();
            //... process messages
        });
    }
    // ack will commit the latest offsets (per partition) of the batch.
    return records.ack();
}

@Incoming("prices")
public void consumeRecords(ConsumerRecords<String, Double> records) {
    for (TopicPartition partition : records.partitions()) {
        for (ConsumerRecord<String, Double> record : records.records(partition)) {
            //... process messages
        }
    }
}

Note that the successful processing of the incoming record batch will commit the latest offsets for each partition received inside the batch. The configured commit strategy will be applied for these records only.

Conversely, if the processing throws an exception, all messages are nacked, applying the failure strategy for all the records inside the batch.

Configuration Reference

Table 1. Incoming Attributes of the 'smallrye-kafka' connector
Attribute (alias)	Description	Mandatory	Default
auto.offset.reset	What to do when there is no initial offset in Kafka.Accepted values are earliest, latest and none Type: string	false	`latest`
batch	Whether the Kafka records are consumed in batch. The channel injection point must consume a compatible type, such as `List<Payload>` or `KafkaRecordBatch<Payload>`. Type: boolean	false	`false`
bootstrap.servers (kafka.bootstrap.servers)	A comma-separated list of host:port to use for establishing the initial connection to the Kafka cluster. Type: string	false	`localhost:9092`
broadcast	Whether the Kafka records should be dispatched to multiple consumer Type: boolean	false	`false`
cloud-events	Enables (default) or disables the Cloud Event support. If enabled on an incoming channel, the connector analyzes the incoming records and try to create Cloud Event metadata. If enabled on an outgoing, the connector sends the outgoing messages as Cloud Event if the message includes Cloud Event Metadata. Type: boolean	false	`true`
commit-strategy	Specify the commit strategy to apply when a message produced from a record is acknowledged. Values can be `latest`, `ignore` or `throttled`. If `enable.auto.commit` is true then the default is `ignore` otherwise it is `throttled` Type: string	false
consumer-rebalance-listener.name	The name set in `@Identifier` of a bean that implements `io.smallrye.reactive.messaging.kafka.KafkaConsumerRebalanceListener`. If set, this rebalance listener is applied to the consumer. Type: string	false
dead-letter-queue.key.serializer	When the `failure-strategy` is set to `dead-letter-queue` indicates the key serializer to use. If not set the serializer associated to the key deserializer is used Type: string	false
dead-letter-queue.topic	When the `failure-strategy` is set to `dead-letter-queue` indicates on which topic the record is sent. Defaults is `dead-letter-topic-$channel` Type: string	false
dead-letter-queue.value.serializer	When the `failure-strategy` is set to `dead-letter-queue` indicates the value serializer to use. If not set the serializer associated to the value deserializer is used Type: string	false
enable.auto.commit	If enabled, consumer’s offset will be periodically committed in the background by the underlying Kafka client, ignoring the actual processing outcome of the records. It is recommended to NOT enable this setting and let Reactive Messaging handles the commit. Type: boolean	false	`false`
fail-on-deserialization-failure	When no deserialization failure handler is set and a deserialization failure happens, report the failure and mark the application as unhealthy. If set to `false` and a deserialization failure happens, a `null` value is forwarded. Type: boolean	false	`true`
failure-strategy	Specify the failure strategy to apply when a message produced from a record is acknowledged negatively (nack). Values can be `fail` (default), `ignore`, or `dead-letter-queue` Type: string	false	`fail`
fetch.min.bytes	The minimum amount of data the server should return for a fetch request. The default setting of 1 byte means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive. Type: int	false	`1`
graceful-shutdown	Whether or not a graceful shutdown should be attempted when the application terminates. Type: boolean	false	`true`
group.id	A unique string that identifies the consumer group the application belongs to. If not set, a unique, generated id is used Type: string	false
health-enabled	Whether health reporting is enabled (default) or disabled Type: boolean	false	`true`
health-readiness-enabled	Whether readiness health reporting is enabled (default) or disabled Type: boolean	false	`true`
health-readiness-timeout	deprecated - During the readiness health check, the connector connects to the broker and retrieves the list of topics. This attribute specifies the maximum duration (in ms) for the retrieval. If exceeded, the channel is considered not-ready. Deprecated: Use 'health-topic-verification-timeout' instead. Type: long	false
health-readiness-topic-verification	deprecated - Whether the readiness check should verify that topics exist on the broker. Default to false. Enabling it requires an admin connection. Deprecated: Use 'health-topic-verification-enabled' instead. Type: boolean	false
health-topic-verification-enabled	Whether the startup and readiness check should verify that topics exist on the broker. Default to false. Enabling it requires an admin client connection. Type: boolean	false	`false`
health-topic-verification-timeout	During the startup and readiness health check, the connector connects to the broker and retrieves the list of topics. This attribute specifies the maximum duration (in ms) for the retrieval. If exceeded, the channel is considered not-ready. Type: long	false	`2000`
kafka-configuration	Identifier of a CDI bean that provides the default Kafka consumer/producer configuration for this channel. The channel configuration can still override any attribute. The bean must have a type of Map<String, Object> and must use the @io.smallrye.common.annotation.Identifier qualifier to set the identifier. Type: string	false
key-deserialization-failure-handler	The name set in `@Identifier` of a bean that implements `io.smallrye.reactive.messaging.kafka.DeserializationFailureHandler`. If set, deserialization failure happening when deserializing keys are delegated to this handler which may retry or provide a fallback value. Type: string	false
key.deserializer	The deserializer classname used to deserialize the record’s key Type: string	false	`org.apache.kafka.common.serialization.StringDeserializer`
max-queue-size-factor	Multiplier factor to determine maximum number of records queued for processing, using `max.poll.records` * `max-queue-size-factor`. Defaults to 2. In `batch` mode `max.poll.records` is considered `1`. Type: int	false	`2`
partitions	The number of partitions to be consumed concurrently. The connector creates the specified amount of Kafka consumers. It should match the number of partition of the targeted topic Type: int	false	`1`
pattern	Indicate that the `topic` property is a regular expression. Must be used with the `topic` property. Cannot be used with the `topics` property Type: boolean	false	`false`
pause-if-no-requests	Whether the polling must be paused when the application does not request items and resume when it does. This allows implementing back-pressure based on the application capacity. Note that polling is not stopped, but will not retrieve any records when paused. Type: boolean	false	`true`
poll-timeout	The polling timeout in milliseconds. When polling records, the poll will wait at most that duration before returning records. Default is 1000ms Type: int	false	`1000`
requests	When `partitions` is greater than 1, this attribute allows configuring how many records are requested by each consumers every time. Type: int	false	`128`
retry	Whether or not the connection to the broker is re-attempted in case of failure Type: boolean	false	`true`
retry-attempts	The maximum number of reconnection before failing. -1 means infinite retry Type: int	false	`-1`
retry-max-wait	The max delay (in seconds) between 2 reconnects Type: int	false	`30`
throttled.unprocessed-record-max-age.ms	While using the `throttled` commit-strategy, specify the max age in milliseconds that an unprocessed message can be before the connector is marked as unhealthy. Setting this attribute to 0 disables this monitoring. Type: int	false	`60000`
topic	The consumed / populated Kafka topic. If neither this property nor the `topics` properties are set, the channel name is used Type: string	false
topics	A comma-separating list of topics to be consumed. Cannot be used with the `topic` or `pattern` properties Type: string	false
tracing-enabled	Whether tracing is enabled (default) or disabled Type: boolean	false	`true`
value-deserialization-failure-handler	The name set in `@Identifier` of a bean that implements `io.smallrye.reactive.messaging.kafka.DeserializationFailureHandler`. If set, deserialization failure happening when deserializing values are delegated to this handler which may retry or provide a fallback value. Type: string	false
value.deserializer	The deserializer classname used to deserialize the record’s value Type: string	true

You can also pass any property supported by the underlying Kafka consumer.

For example, to configure the max.poll.records property, use:

mp.messaging.incoming.[channel].max.poll.records=1000

Some consumer client properties are configured to sensible default values:

If not set, reconnect.backoff.max.ms is set to 10000 to avoid high load on disconnection.

If not set, key.deserializer is set to org.apache.kafka.common.serialization.StringDeserializer.

The consumer client.id is configured according to the number of clients to create using mp.messaging.incoming.[channel].partitions property.

If a client.id is provided, it is used as-is or suffixed with client index if partitions property is set.
If a client.id is not provided, it is generated as kafka-consumer-[channel][-index]. == Writing Kafka Records

The Kafka Connector can write Reactive Messaging Messages as Kafka Records.

Example

Let’s imagine you have a Kafka broker running, and accessible using the kafka:9092 address (by default it would use localhost:9092). Configure your application to write the messages from the prices channel into a Kafka topic as follows:

kafka.bootstrap.servers=kafka:9092      (1)

mp.messaging.outgoing.prices-out.connector=smallrye-kafka   (2)
mp.messaging.outgoing.prices-out.value.serializer=org.apache.kafka.common.serialization.DoubleSerializer  (3)
mp.messaging.outgoing.prices-out.topic=prices   (4)

Configure the broker location. You can configure it globally or per channel
Configure the connector to manage the prices channel
Sets the (Kafka) serializer to encode the message payload into the record’s value
Make sure the topic name is prices (and not the default prices-out)

Then, your application must send Message<Double> to the prices channel. It can use double payloads as in the following snippet:

package outbound;

import io.smallrye.mutiny.Multi;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import javax.enterprise.context.ApplicationScoped;
import java.time.Duration;
import java.util.Random;

@ApplicationScoped
public class KafkaPriceProducer {

    private final Random random = new Random();

    @Outgoing("prices-out")
    public Multi<Double> generate() {
        // Build an infinite stream of random prices
        // It emits a price every second
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
            .map(x -> random.nextDouble());
    }

}

Or, you can send Message<Double>:

package outbound;

import io.smallrye.mutiny.Multi;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import javax.enterprise.context.ApplicationScoped;
import java.time.Duration;
import java.util.Random;

@ApplicationScoped
public class KafkaPriceMessageProducer {

    private final Random random = new Random();

    @Outgoing("prices-out")
    public Multi<Message<Double>> generate() {
        // Build an infinite stream of random prices
        // It emits a price every second
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
            .map(x -> Message.of(random.nextDouble()));
    }

}

Serialization

The serialization is handled by the underlying Kafka Client. You need to configure the:

mp.messaging.outgoing.[channel-name].value.serializer to configure the value serializer (mandatory)
mp.messaging.outgoing.[channel-name].key.serializer to configure the key serializer (optional, default to String)

If you want to use a custom serializer, add it to your CLASSPATH and configure the associate attribute.

By default, the written record contains:

the Message payload as value
no key, or the key configured using the key attribute or the key passed in the metadata attached to the Message
the timestamp computed for the system clock (now) or the timestamp passed in the metadata attached to the Message

Sending key/value pairs

In the Kafka world, it’s often necessary to send records, i.e. a key/value pair. The connector provides the io.smallrye.reactive.messaging.kafka.Record class that you can use to send a pair:

    @Incoming("in")
    @Outgoing("out")
    public Record<String, String> process(String in) {
        return Record.of("my-key", in);
    }

When the connector receives a message with a Record payload, it extracts the key and value from it. The configured serializers for the key and the value must be compatible with the record’s key and value. Note that the key and the value can be null. It is also possible to create a record with a null key AND a null value.

If you need more control on the written records, use OutgoingKafkaRecordMetadata.

Outbound Metadata

When sending Messages, you can add an instance of OutgoingKafkaRecordMetadata to influence how the message is going to written to Kafka. For example, you can add Kafka headers, configure the record key…

        // Creates an OutgoingKafkaRecordMetadata
        // The type parameter is the type of the record's key
        OutgoingKafkaRecordMetadata<String> metadata = OutgoingKafkaRecordMetadata.<String>builder()
            .withKey("my-key")
            .withHeaders(new RecordHeaders().add("my-header", "value".getBytes()))
            .build();

        // Create a new message from the `incoming` message
        // Add `metadata` to the metadata from the `incoming` message.
        return incoming.addMetadata(metadata);

Propagating Record Key

When processing messages, you can propagate incoming record key to the outgoing record.

Consider the following example method, which consumes messages from the channel in, transforms the payload, and writes the result to the channel out.

    @Incoming("in")
    @Outgoing("out")
    public double process(int in) {
        return in * 0.88;
    }

Enabled with mp.messaging.outgoing.[channel-name].propagate-record-key=true configuration, record key propagation produces the outgoing record with the same key as the incoming record.

If the outgoing record already contains a key, it won’t be overridden by the incoming record key. If the incoming record does have a null key, the mp.messaging.outgoing.[channel-name].key property is used.

Dynamic topic names

Sometimes it is desirable to select the destination of a message dynamically. In this case, you should not configure the topic inside your application configuration file, but instead, use the outbound metadata to set the name of the topic.

For example, you can route to a dynamic topic based on the incoming message:

String topicName = selectTopicFromIncommingMessage(incoming);
OutgoingKafkaRecordMetadata<String> metadata = OutgoingKafkaRecordMetadata.<String>builder()
    .withTopic(topicName)
    .build();

// Create a new message from the `incoming` message
// Add `metadata` to the metadata from the `incoming` message.
return incoming.addMetadata(metadata);

Acknowledgement

Kafka acknowledgement can take times depending on the configuration. Also, it stores in-memory the records that cannot be written.

By default, the connector does wait for Kafka to acknowledge the record to continue the processing (acknowledging the received Message). You can disable this by setting the waitForWriteCompletion attribute to false.

Note that the acks attribute has a huge impact on the record acknowledgement.

If a record cannot be written, the message is nacked.

Back-pressure and inflight records

The Kafka outbound connector handles back-pressure monitoring the number of in-flight messages waiting to be written to the Kafka broker. The number of in-flight messages is configured using the max-inflight-messages attribute and defaults to 1024.

The connector only sends that amount of messages concurrently. No other messages will be sent until at least one in-flight message gets acknowledged by the broker. Then, the connector writes a new message to Kafka when one of the broker’s in-flight messages get acknowledged. Be sure to configure Kafka’s batch.size and linger.ms accordingly.

You can also remove the limit of inflight messages by setting max-inflight-messages to 0. However, note that the Kafka Producer may block if the number of requests reaches max.in.flight.requests.per.connection.

Handling serialization failures

For Kafka producer client serialization failures are not recoverable, thus the message dispatch is not retried. However, using schema registries, serialization may intermittently fail to contact the registry. In these cases you may need to apply a failure strategy for the serializer. To achieve this, create a CDI bean implementing the SerializationFailureHandler<T> interface:

@ApplicationScoped
@Identifier("failure-fallback") // Set the name of the failure handler
public class MySerializationFailureHandler
    implements SerializationFailureHandler<JsonObject> { // Specify the expected type

    @Override
    public byte[] decorateSerialization(Uni<byte[]> serialization, String topic, boolean isKey, String serializer,
                Object data, Headers headers) {
        return serialization
                    .onFailure().retry().atMost(3)
                    .await().atMost(Duration.ofMillis(200));
    }
}

The bean must be exposed with the @Identifier qualifier specifying the name of the bean. Then, in the connector configuration, specify the following attribute:

mp.messaging.incoming.$channel.key-serialization-failure-handler: name of the bean handling serialization failures happening for the record’s key
mp.messaging.incoming.$channel.value-serialization-failure-handler: name of the bean handling serialization failures happening for the record’s value,

The handler is called with the serialization action as a Uni, the record’s topic, a boolean indicating whether the failure happened on a key, the class name of the deserializer that throws the exception, the corrupted data, the exception, and the records headers. Failure strategies like retry, providing a fallback value or applying timeout can be implemented. Note that the method must await on the result and return the serialized byte array. Alternatively, the handler can implement decorateSerialization method which can return a fallback value.

Sending Cloud Events

The Kafka connector supports Cloud Events. The connector sends the outbound record as Cloud Events if:

the message metadata contains an io.smallrye.reactive.messaging.ce.OutgoingCloudEventMetadata instance,
the channel configuration defines the cloud-events-type and cloud-events-source attributes.

You can create io.smallrye.reactive.messaging.ce.OutgoingCloudEventMetadata instances using:

package outbound;

import io.smallrye.reactive.messaging.ce.OutgoingCloudEventMetadata;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import javax.enterprise.context.ApplicationScoped;
import java.net.URI;

@ApplicationScoped
public class KafkaCloudEventProcessor {

    @Outgoing("cloud-events")
    public Message<String> toCloudEvents(Message<String> in) {
        return in.addMetadata(OutgoingCloudEventMetadata.builder()
            .withId("id-" + in.getPayload())
            .withType("greetings")
            .withSource(URI.create("http://example.com"))
            .withSubject("greeting-message")
            .build());
    }

}

If the metadata does not contain an id, the connector generates one (random UUID). The type and source can be configured per message or at the channel level using the cloud-events-type and cloud-events-source attributes. Other attributes are also configurable.

The metadata can be contributed by multiple methods, however, you must always retrieve the already existing metadata to avoid overriding the values:

package outbound;

import io.smallrye.reactive.messaging.ce.OutgoingCloudEventMetadata;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import javax.enterprise.context.ApplicationScoped;
import java.net.URI;

@ApplicationScoped
public class KafkaCloudEventMultipleProcessors {

    @Incoming("source")
    @Outgoing("processed")
    public Message<String> process(Message<String> in) {
        return in.addMetadata(OutgoingCloudEventMetadata.builder()
            .withId("id-" + in.getPayload())
            .withType("greeting")
            .build());
    }

    @SuppressWarnings("unchecked")
    @Incoming("processed")
    @Outgoing("cloud-events")
    public Message<String> process2(Message<String> in) {
        OutgoingCloudEventMetadata<String> metadata = in
            .getMetadata(OutgoingCloudEventMetadata.class)
            .orElseGet(() -> OutgoingCloudEventMetadata.builder().build());

        return in.addMetadata(OutgoingCloudEventMetadata.from(metadata)
            .withSource(URI.create("source://me"))
            .withSubject("test")
            .build());
    }

}

By default, the connector sends the Cloud Events using the binary format. You can write structured Cloud Events by setting the cloud-events-mode to structured. Only JSON is supported, so the created records had its content-type header set to application/cloudevents+json; charset=UTF-8 When using the structured mode, the value serializer must be set to org.apache.kafka.common.serialization.StringSerializer, otherwise the connector reports the error. In addition, in structured, the connector maps the message’s payload to JSON, except for String passed directly.

The record’s key can be set in the channel configuration (key attribute), in the OutgoingKafkaRecordMetadata or using the partitionkey Cloud Event attribute.

you can disable the Cloud Event support by setting the cloud-events attribute to false

Using `ProducerRecord`

Kafka built-in type ProducerRecord<K,V> can also be used for producing messages:

@Outgoing("out")
public ProducerRecord<String, String> generate() {
    return new ProducerRecord<>("my-topic", "key", "value");
}

This is an advanced feature. The ProducerRecord is sent to Kafka as is. Any possible metadata attached through Message<ProducerRecord<K, V>> are ignored and lost.

Configuration Reference

Table 2. Outgoing Attributes of the 'smallrye-kafka' connector
Attribute (alias)	Description	Mandatory	Default
acks	The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent. Accepted values are: 0, 1, all Type: string	false	`1`
bootstrap.servers (kafka.bootstrap.servers)	A comma-separated list of host:port to use for establishing the initial connection to the Kafka cluster. Type: string	false	`localhost:9092`
buffer.memory	The total bytes of memory the producer can use to buffer records waiting to be sent to the server. Type: long	false	`33554432`
close-timeout	The amount of milliseconds waiting for a graceful shutdown of the Kafka producer Type: int	false	`10000`
cloud-events	Enables (default) or disables the Cloud Event support. If enabled on an incoming channel, the connector analyzes the incoming records and try to create Cloud Event metadata. If enabled on an outgoing, the connector sends the outgoing messages as Cloud Event if the message includes Cloud Event Metadata. Type: boolean	false	`true`
cloud-events-data-content-type (cloud-events-default-data-content-type)	Configure the default `datacontenttype` attribute of the outgoing Cloud Event. Requires `cloud-events` to be set to `true`. This value is used if the message does not configure the `datacontenttype` attribute itself Type: string	false
cloud-events-data-schema (cloud-events-default-data-schema)	Configure the default `dataschema` attribute of the outgoing Cloud Event. Requires `cloud-events` to be set to `true`. This value is used if the message does not configure the `dataschema` attribute itself Type: string	false
cloud-events-insert-timestamp (cloud-events-default-timestamp)	Whether or not the connector should insert automatically the `time` attribute into the outgoing Cloud Event. Requires `cloud-events` to be set to `true`. This value is used if the message does not configure the `time` attribute itself Type: boolean	false	`true`
cloud-events-mode	The Cloud Event mode (`structured` or `binary` (default)). Indicates how are written the cloud events in the outgoing record Type: string	false	`binary`
cloud-events-source (cloud-events-default-source)	Configure the default `source` attribute of the outgoing Cloud Event. Requires `cloud-events` to be set to `true`. This value is used if the message does not configure the `source` attribute itself Type: string	false
cloud-events-subject (cloud-events-default-subject)	Configure the default `subject` attribute of the outgoing Cloud Event. Requires `cloud-events` to be set to `true`. This value is used if the message does not configure the `subject` attribute itself Type: string	false
cloud-events-type (cloud-events-default-type)	Configure the default `type` attribute of the outgoing Cloud Event. Requires `cloud-events` to be set to `true`. This value is used if the message does not configure the `type` attribute itself Type: string	false
health-enabled	Whether health reporting is enabled (default) or disabled Type: boolean	false	`true`
health-readiness-enabled	Whether readiness health reporting is enabled (default) or disabled Type: boolean	false	`true`
health-readiness-timeout	deprecated - During the readiness health check, the connector connects to the broker and retrieves the list of topics. This attribute specifies the maximum duration (in ms) for the retrieval. If exceeded, the channel is considered not-ready. Deprecated: Use 'health-topic-verification-timeout' instead. Type: long	false
health-readiness-topic-verification	deprecated - Whether the readiness check should verify that topics exist on the broker. Default to false. Enabling it requires an admin connection. Deprecated: Use 'health-topic-verification-enabled' instead. Type: boolean	false
health-topic-verification-enabled	Whether the startup and readiness check should verify that topics exist on the broker. Default to false. Enabling it requires an admin client connection. Type: boolean	false	`false`
health-topic-verification-timeout	During the startup and readiness health check, the connector connects to the broker and retrieves the list of topics. This attribute specifies the maximum duration (in ms) for the retrieval. If exceeded, the channel is considered not-ready. Type: long	false	`2000`
kafka-configuration	Identifier of a CDI bean that provides the default Kafka consumer/producer configuration for this channel. The channel configuration can still override any attribute. The bean must have a type of Map<String, Object> and must use the @io.smallrye.common.annotation.Identifier qualifier to set the identifier. Type: string	false
key	A key to used when writing the record Type: string	false
key-serialization-failure-handler	The name set in `@Identifier` of a bean that implements `io.smallrye.reactive.messaging.kafka.SerializationFailureHandler`. If set, serialization failure happening when serializing keys are delegated to this handler which may provide a fallback value. Type: string	false
key.serializer	The serializer classname used to serialize the record’s key Type: string	false	`org.apache.kafka.common.serialization.StringSerializer`
max-inflight-messages	The maximum number of messages to be written to Kafka concurrently. It limits the number of messages waiting to be written and acknowledged by the broker. You can set this attribute to `0` remove the limit Type: long	false	`1024`
merge	Whether the connector should allow multiple upstreams Type: boolean	false	`false`
partition	The target partition id. -1 to let the client determine the partition Type: int	false	`-1`
propagate-record-key	Propagate incoming record key to the outgoing record Type: boolean	false	`false`
retries	If set to a positive number, the connector will try to resend any record that was not delivered successfully (with a potentially transient error) until the number of retries is reached. If set to 0, retries are disabled. If not set, the connector tries to resend any record that failed to be delivered (because of a potentially transient error) during an amount of time configured by `delivery.timeout.ms`. Type: long	false	`2147483647`
topic	The consumed / populated Kafka topic. If neither this property nor the `topics` properties are set, the channel name is used Type: string	false
tracing-enabled	Whether tracing is enabled (default) or disabled Type: boolean	false	`true`
value-serialization-failure-handler	The name set in `@Identifier` of a bean that implements `io.smallrye.reactive.messaging.kafka.SerializationFailureHandler`. If set, serialization failure happening when serializing values are delegated to this handler which may provide a fallback value. Type: string	false
value.serializer	The serializer classname used to serialize the payload Type: string	true
waitForWriteCompletion	Whether the client waits for Kafka to acknowledge the written record before acknowledging the message Type: boolean	false	`true`

You can also pass any property supported by the underlying Kafka producer.

For example, to configure the batch.size property, use:

mp.messaging.outgoing.[channel].batch.size=32768

Some producer client properties are configured to sensible default values:

If not set, reconnect.backoff.max.ms is set to 10000 to avoid high load on disconnection.

If not set, key.serializer is set to org.apache.kafka.common.serialization.StringSerializer.

If not set, producer client.id is generated as kafka-producer-[channel].

Retrieving Kafka default configuration

If your application/runtime exposes as a CDI bean of type Map<String, Object with the identifier default-kafka-broker, this configuration is used to establish the connection with the Kafka broker.

For example, you can imagine exposing this map as follows:

@Produces
@ApplicationScoped
@Identifier("default-kafka-broker")
public Map<String, Object> createKafkaRuntimeConfig() {
    Map<String, Object> properties = new HashMap<>();

    StreamSupport
        .stream(config.getPropertyNames().spliterator(), false)
        .map(String::toLowerCase)
        .filter(name -> name.startsWith("kafka"))
        .distinct()
        .sorted()
        .forEach(name -> {
            final String key = name.substring("kafka".length() + 1).toLowerCase().replaceAll("[^a-z0-9.]", ".");
            final String value = config.getOptionalValue(name, String.class).orElse("");
            properties.put(key, value);
        });

    return properties;
}

This previous example would extract all the configuration keys from MicroProfile Config starting with kafka.

Quarkus

Starting with Quarkus 1.5, a map corresponding to the previous example is automatically provided.

In addition to this default configuration, you can configure the name of the Map producer using the kafka-configuration attribute:

mp.messaging.incoming.my-channel.connector=smallrye-kafka
mp.messaging.incoming.my-channel.kafka-configuration=my-configuration

In this case, the connector looks for the Map associated with the my-configuration name. If kafka-configuration is not set, an optional lookup for a Map exposed with the channel name (my-channel in the previous example) is done.

If kafka-configuration is set and no Map can be found, the deployment fails.

Attribute values are resolved as follows:

if the attribute is set directly on the channel configuration (mp.messaging.incoming.my-channel.attribute=value), this value is used
if the attribute is not set on the channel, the connector looks for a Map with the channel name or the configured kafka-configuration (if set) and the value is retrieved from that Map
If the resolved Map does not contain the value the default Map is used (exposed with the default-kafka-broker name)

Using Apache Avro serializer/deserializer

If you are using Apache Avro serializer/deserializer, please note the following configuration properties.

For Confluent Schema Registry

Confluent Avro library is io.confluent:kafka-avro-serializer. Note that this library is not available in Maven Central, you need to use the Confluent Maven repository.

Consumer

Property	Recommended value
value.deserializer	io.confluent.kafka.serializers.KafkaAvroDeserializer
schema.registry.url	http://<your_host>:<your_port>/
specific.avro.reader	true

Property

Recommended value

value.deserializer

io.confluent.kafka.serializers.KafkaAvroDeserializer

schema.registry.url

http://<your_host>:<your_port>/

specific.avro.reader

true

Example:

mp.messaging.incoming.[channel].value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
mp.messaging.incoming.[channel].schema.registry.url=http://<your_host>:<your_port>/
mp.messaging.incoming.[channel].specific.avro.reader=true

Producer

Property	Recommended value
value.serializer	io.confluent.kafka.serializers.KafkaAvroSerializer
schema.registry.url	http://<your_host>:<your_port>/

Property

Recommended value

value.serializer

io.confluent.kafka.serializers.KafkaAvroSerializer

schema.registry.url

http://<your_host>:<your_port>/

Example:

mp.messaging.outgoing.[channel].value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
mp.messaging.outgoing.[channel].schema.registry.url=http://<your_host>:<your_port>/

For Apicurio Registry 1.x

Apicurio Registry 1.x Avro library is io.apicurio:apicurio-registry-utils-serde.

The configuration properties listed here are meant to be used with the Apicurio Registry 1.x client library and Apicurio Registry 1.x server.

Consumer

Property	Recommended value
value.deserializer	io.apicurio.registry.utils.serde.AvroKafkaDeserializer
apicurio.registry.url	http://<your_host>:<your_port>/api
apicurio.registry.avro-datum-provider	io.apicurio.registry.utils.serde.avro.DefaultAvroDatumProvider
apicurio.registry.use-specific-avro-reader	true

Property

Recommended value

value.deserializer

io.apicurio.registry.utils.serde.AvroKafkaDeserializer

apicurio.registry.url

http://<your_host>:<your_port>/api

apicurio.registry.avro-datum-provider

io.apicurio.registry.utils.serde.avro.DefaultAvroDatumProvider

apicurio.registry.use-specific-avro-reader

true

Example:

mp.messaging.incoming.[channel].value.deserializer=io.apicurio.registry.utils.serde.AvroKafkaDeserializer
mp.messaging.incoming.[channel].apicurio.registry.url=http://<your_host>:<your_port>/api
mp.messaging.incoming.[channel].apicurio.registry.avro-datum-provider=io.apicurio.registry.utils.serde.avro.DefaultAvroDatumProvider
mp.messaging.incoming.[channel].apicurio.registry.use-specific-avro-reader=true

Producer

Property	Recommended value
value.serializer	io.apicurio.registry.utils.serde.AvroKafkaSerializer
apicurio.registry.url	http://<your_host>:<your_port>/api

Property

Recommended value

value.serializer

io.apicurio.registry.utils.serde.AvroKafkaSerializer

apicurio.registry.url

http://<your_host>:<your_port>/api

To automatically register schemas with the registry, add:

Property	Value
apicurio.registry.global-id	io.apicurio.registry.utils.serde.strategy.GetOrCreateIdStrategy

Property

Value

apicurio.registry.global-id

io.apicurio.registry.utils.serde.strategy.GetOrCreateIdStrategy

Example:

mp.messaging.outgoing.[channel].value.serializer=io.apicurio.registry.utils.serde.AvroKafkaSerializer
mp.messaging.outgoing.[channel].apicurio.registry.url=http://<your_host>:<your_port>/api
mp.messaging.outgoing.[channel].apicurio.registry.global-id=io.apicurio.registry.utils.serde.strategy.GetOrCreateIdStrategy

For Apicurio Registry 2.x

Apicurio Registry 2.x Avro library is io.apicurio:apicurio-registry-serdes-avro-serde.

The configuration properties listed here are meant to be used with the Apicurio Registry 2.x client library and Apicurio Registry 2.x server.

Consumer

Property	Recommended value
value.deserializer	io.apicurio.registry.serde.avro.AvroKafkaDeserializer
apicurio.registry.url	http://<your_host>:<your_port>/apis/registry/v2
apicurio.registry.use-specific-avro-reader	true

Property

Recommended value

value.deserializer

io.apicurio.registry.serde.avro.AvroKafkaDeserializer

apicurio.registry.url

http://<your_host>:<your_port>/apis/registry/v2

apicurio.registry.use-specific-avro-reader

true

Example:

mp.messaging.incoming.[channel].value.deserializer=io.apicurio.registry.serde.avro.AvroKafkaDeserializer
mp.messaging.incoming.[channel].apicurio.registry.url=http://<your_host>:<your_port>/apis/registry/v2
mp.messaging.incoming.[channel].apicurio.registry.use-specific-avro-reader=true

Producer

Property	Recommended value
value.serializer	io.apicurio.registry.serde.avro.AvroKafkaSerializer
apicurio.registry.url	http://<your_host>:<your_port>/apis/registry/v2

Property

Recommended value

value.serializer

io.apicurio.registry.serde.avro.AvroKafkaSerializer

apicurio.registry.url

http://<your_host>:<your_port>/apis/registry/v2

To automatically register schemas with the registry, add:

Property	Value
apicurio.registry.auto-register	true

Property

Value

apicurio.registry.auto-register

true

Example:

mp.messaging.outgoing.[channel].value.serializer=io.apicurio.registry.serde.avro.AvroKafkaSerializer
mp.messaging.outgoing.[channel].apicurio.registry.url=http://<your_host>:<your_port>/apis/registry/v2
mp.messaging.outgoing.[channel].apicurio.registry.auto-register=true

Unresolved include directive in modules/kafka/pages/kafka.adoc - include::kafka-protobuf-configuration.adoc[]

Health reporting

The Kafka connector reports the readiness and liveness of each channel managed by the connector.

To disable health reporting, set the health-enabled attribute for the channel to false.

Readiness

On the inbound side, two strategies are available to check the readiness of the application. The default strategy verifies that we have at least one active connection with the broker. This strategy is lightweight.

You can also enable another strategy by setting the health-readiness-topic-verification attribute to true. In this case, the check verifies that:

the broker is available
the Kafka topic is created (available in the broker).
no failures have been caught

With this second strategy, if you consume multiple topics using the topics attribute, the readiness check verifies that all the consumed topics are available. If you use a pattern (using the pattern attribute), the readiness check verifies that at least one existing topic matches the pattern.

On the outbound side (writing records to Kafka), two strategies are also offered. The default strategy just verifies that the producer has at least one active connection with the broker.

You can also enable another strategy by setting the health-readiness-topic-verification attribute to true. In this case, teh check verifies that

the broker is available
the Kafka topic is created (available in the broker).

With this second strategy, the readiness check uses a Kafka Admin Client to retrieve the existing topics. Retrieving the topics can be a lengthy operation. You can configure a timeout using the health-readiness-timeout attribute. The default timeout is set to 2 seconds.

Also, you can disable the readiness checks altogether by setting health-readiness-enabled to false.

Liveness

On the inbound side (receiving records from Kafka), the liveness check verifies that:

no failures have been caught
the client is connected to the broker

On the outbound side (writing records to Kafka), the liveness check verifies that:

no failures have been caught

Note that a message processing failures nacks the message which is then handled by the failure-strategy. It the responsibility of the failure-strategy to report the failure and influence the outcome of the liveness checks. The fail failure strategy reports the failure and so the liveness check will report the failure.

KafkaClientService

For advanced use cases, SmallRye Reactive Messaging provides a bean of type KafkaClientService that you can inject:

@Inject
KafkaClientService kafka;

From there, you can obtain an io.smallrye.reactive.messaging.kafka.KafkaProducer and an io.smallrye.reactive.messaging.kafka.KafkaConsumer.

KafkaProducer and KafkaConsumer expose a non-blocking API on top of the Kafka client API. They also mediate access to the threads that SmallRye Reactive Messaging uses to run all Kafka operations: the polling thread, used for consuming records from Kafka topics, and the sending thread, used for producing records to Kafka topics. (Just to be clear: each channel has its own polling thread and sending thread.)

The reason why SmallRye Reactive Messaging uses a special thread to run the poll loop should be obvious: the Consumer API is blocking. The Producer API, on the other hand, is documented to be non-blocking. However, in present versions, Kafka doesn’t guarantee that in all cases; see KAFKA-3539 for more details. That is why SmallRye Reactive Messaging uses a dedicated thread to run the send operations as well.

Sometimes, SmallRye Reactive Messaging provides direct access to the Kafka Producer or Consumer. For example, a KafkaConsumerRebalanceListener methods are always invoked on the polling thread, so they give you direct access to Consumer. In such case, you should use the Producer/Consumer API directly, instead of the KafkaProducer/KafkaConsumer API.

Kerberos authentication

When using Kerberos authentication, you need to configure the connector with:

the security protocol set to SASL_PLAINTEXT
the SASL mechanism set to GSSAPI
the Jaas config configured with Krb5LoginModule
the Kerberos service name

The following snippet provides an example:

kafka.bootstrap.servers=ip-192-168-0-207.us-east-2.compute.internal:9094
kafka.sasl.mechanism=GSSAPI
kafka.security.protocol=SASL_PLAINTEXT
kafka.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required doNotPrompt=true refreshKrb5Config=true useKeyTab=true storeKey=true keyTab="file:/opt/kafka/krb5/kafka-producer.keytab" principal="kafka-producer/ip-192-168-0-207.us-east-2.compute.internal@INTERNAL";
kafka.sasl.kerberos.service.name=kafka

Apache Kafka

Introduction

Using the Kafka Connector

Receiving Kafka Records

Example

Deserialization

Inbound Metadata

Acknowledgement

Failure Management

Retrying processing

Handling deserialization failures

Receiving Cloud Events

Binary Cloud Events

Structured Cloud Events

Consumer Rebalance Listener

Example

Receiving Kafka Records in Batches

Configuration Reference

Example

Serialization

Sending key/value pairs

Outbound Metadata

Propagating Record Key

Dynamic topic names

Acknowledgement

Back-pressure and inflight records

Handling serialization failures

Sending Cloud Events

Using ProducerRecord

Configuration Reference

Retrieving Kafka default configuration

Using Apache Avro serializer/deserializer

For Confluent Schema Registry

Consumer

Producer

For Apicurio Registry 1.x

Consumer

Producer

For Apicurio Registry 2.x

Consumer

Producer

Health reporting

Readiness

Liveness

KafkaClientService

Kerberos authentication

Using `ProducerRecord`