Apache Kafka Reference Guide

La Mensajería Reactiva invoca su método en un hilo I/O. Consulte la documentación de la Arquitectura Reactiva de Quarkus para más detalles sobre este tema. Pero, a menudo necesita combinar la Mensajería Reactiva con el procesamiento de bloqueo, como las interacciones con la base de datos. Para ello, necesita utilizar la anotación @Blocking indicando que el procesamiento es bloqueante y no debe ejecutarse en el hilo de llamada.

Por ejemplo, el siguiente código ilustra cómo puede almacenar las cargas útiles entrantes en una base de datos utilizando Hibernate con Panache:

The complete example is available in the kafka-panache-quickstart directory.

All messages received by a consumer must be acknowledged. In the absence of acknowledgment, the processing is considered in error. If the consumer method receives a Record or a payload, the message will be acked on method return, also known as Strategy.POST_PROCESSING. If the consumer method returns another reactive stream or CompletionStage, the message will be acked when the downstream message is acked. You can override the default behavior to ack the message on arrival (Strategy.PRE_PROCESSING), or do not ack the message at all (Strategy.NONE) on the consumer method as in the following example:

If the consumer method receives a Message, the acknowledgment strategy is Strategy.MANUAL and the consumer method is in charge of ack/nack the message.

As mentioned above, the method can also override the acknowledgment strategy to PRE_PROCESSING or NONE.

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:

SmallRye Reactive Messaging enables implementing custom commit strategies. See SmallRye Reactive Messaging documentation for more information.

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

The strategy is selected using the failure-strategy attribute.

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

The record written on the dead letter queue contains a set of additional headers about the original record:

SmallRye Reactive Messaging enables implementing custom failure strategies. See SmallRye Reactive Messaging documentation for more information.

In Kafka, a consumer group is a set of consumers which cooperate to consume data from a topic. A topic is divided into a set of partitions. The partitions of a topic are assigned among the consumers in the group, effectively allowing to scale consumption throughput. Note that each partition is assigned to a single consumer from a group. However, a consumer can be assigned multiple partitions if the number of partitions is greater than the number of consumer in the group.

Let’s explore briefly different producer/consumer patterns and how to implement them using Quarkus:

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 specify a compatible container type to receive all the data:

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

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.

SmallRye Reactive Messaging checkpoint commit strategy allows consumer applications to process messages in a stateful manner, while also respecting Kafka consumer scalability. An incoming channel with checkpoint commit strategy persists consumer offsets on an external state store, such as a relational database or a key-value store. As a result of processing consumed records, the consumer application can accumulate an internal state for each topic-partition assigned to the Kafka consumer. This local state will be periodically persisted to the state store and will be associated with the offset of the record that produced it.

This strategy does not commit any offsets to the Kafka broker, so when new partitions get assigned to the consumer, i.e. consumer restarts or consumer group instances scale, the consumer resumes the processing from the latest checkpointed offset with its saved state.

The @Incoming channel consumer code can manipulate the processing state through the CheckpointMetadata API. For example, a consumer calculating the moving average of prices received on a Kafka topic would look the following:

The transform method applies the transformation function to the current state, producing a changed state and registering it locally for checkpointing. By default, the local state is persisted to the state store periodically, period specified by auto.commit.interval.ms, (default: 5000). If persistOnAck flag is given, the latest state is persisted to the state store eagerly on message acknowledgment. The setNext method works similarly directly setting the latest state.

The checkpoint commit strategy tracks when a processing state is last persisted for each topic-partition. If an outstanding state change can not be persisted for checkpoint.unsynced-state-max-age.ms (default: 10000), the channel is marked unhealthy.

Sometimes, you need to have an imperative way of sending messages.

For example, if you need to send a message to a stream when receiving a POST request inside a REST endpoint. In this case, you cannot use @Outgoing because your method has parameters.

For this, you can use an Emitter.

Sending a payload returns a CompletionStage, completed when the message is acked. If the message transmission fails, the CompletionStage is completed exceptionally with the reason of the nack.

With the Emitter API, you can also encapsulate the outgoing payload inside Message<T>. As with the previous examples, Message lets you handle the ack/nack cases differently.

If you prefer using Reactive Stream APIs, you can use MutinyEmitter that will return Uni<Void> from the send method. You can therefore use Mutiny APIs for handling downstream messages and errors.

It is also possible to block on sending the event to the emitter with the sendAndAwait method. It will only return from the method when the event is acked or nacked by the receiver.

More information on how to use Emitter can be found in SmallRye Reactive Messaging – Emitters and Channels

When Kafka broker receives a record, its acknowledgement can take time 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.

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 in-flight 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.

When the Kafka producer receives an error from the server, if it is a transient, recoverable error, the client will retry sending the batch of messages. This behavior is controlled by retries and retry.backoff.ms parameters. In addition to this, SmallRye Reactive Messaging will retry individual messages on recoverable errors, depending on the retries and delivery.timeout.ms parameters.

Note that while having retries in a reliable system is a best practice, the max.in.flight.requests.per.connection parameter defaults to 5, meaning that the order of the messages is not guaranteed. If the message order is a must for your use case, setting max.in.flight.requests.per.connection to 1 will make sure a single batch of messages is sent at a time, in the expense of limiting the throughput of the producer.

For applying retry mechanism on processing errors, see the section on Retrying processing.

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

To use this failure handler, the bean must be exposed with the @Identifier qualifier and the connector configuration must specify the attribute mp.messaging.outgoing.$channel.[key|value]-serialization-failure-handler (for key or value serializers).

The handler is called with details of the serialization, including the action represented as Uni<byte[]>. Note that the method must await on the result and return the serialized byte array.

In some use cases, it is convenient to use the messaging patterns to transfer messages inside the same application. When you don’t connect a channel to a messaging backend like Kafka, everything happens in-memory, and the streams are created by chaining methods together. Each chain is still a reactive stream and enforces the back-pressure protocol.

The framework verifies that the producer/consumer chain is complete, meaning that if the application writes messages into an in-memory channel (using a method with only @Outgoing, or an Emitter), it must also consume the messages from within the application (using a method with only @Incoming or using an unmanaged stream).

By default, a channel can be linked to a single consumer, using @Incoming method or @Channel reactive stream. At application startup, channels are verified to form a chain of consumers and producers with single consumer and producer. You can override this behavior by setting mp.messaging.$channel.broadcast=true on a channel.

In case of in-memory channels, @Broadcast annotation can be used on the @Outgoing method. For example,

Repeating the @Outgoing annotation on outbound or processing methods allows another way of dispatching messages to multiple outgoing channels:

In the previous example generated price will be broadcast to both outbound channels. The following example selectively sends messages to multiple outgoing channels using the Targeted container object, containing key as channel name and value as message payload.

Note that the auto-detection for Kafka serializers doesn’t work for signatures using the Targeted.

For more details on using multiple outgoings, please refer to the SmallRye Reactive Messaging documentation.

Kafka transactions enable atomic writes to multiple Kafka topics and partitions. The Kafka connector provides KafkaTransactions custom emitter for writing Kafka records inside a transaction. It can be injected as a regular emitter @Channel:

The function given to the withTransaction method receives a TransactionalEmitter for producing records, and returns a Uni that provides the result of the transaction.

Kafka transactional producers require configuring acks=all client property, and a unique id for transactional.id, which implies enable.idempotence=true. When Quarkus detects the use of KafkaTransactions for an outgoing channel it configures these properties on the channel, providing a default value of "${quarkus.application.name}-${channelName}" for transactional.id property.

Note that for production use the transactional.id must be unique across all application instances.

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

Enabled with mp.messaging.outgoing.$channel.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.key property is used.

Kafka Transactions allows managing consumer offsets inside a transaction, together with produced messages. This enables coupling a consumer with a transactional producer in a consume-transform-produce pattern, also known as exactly-once processing.

The KafkaTransactions custom emitter provides a way to apply exactly-once processing to an incoming Kafka message inside a transaction.

The following example includes a batch of Kafka records inside a transaction.

When using exactly-once processing, consumed message offset commits are handled by the transaction and therefore the application should not commit offsets through other means. The consumer should have enable.auto.commit=false (the default) and set explicitly commit-strategy=ignore:

Quarkus has built-in support for JSON serialization and deserialization based on Jackson. It will also generate the serializer and deserializer for you, so you do not have to configure anything. When generation is disabled, you can use the provided ObjectMapperSerializer and ObjectMapperDeserializer as explained below.

There is an existing ObjectMapperSerializer that can be used to serialize all data objects via Jackson. You may create an empty subclass if you want to use Serializer/deserializer autodetection.

The corresponding deserializer class needs to be subclassed. So, let’s create a FruitDeserializer that extends the ObjectMapperDeserializer.

Finally, configure your channels to use the Jackson serializer and deserializer.

Now, your Kafka messages will contain a Jackson serialized representation of your Fruit data object. In this case, the deserializer configuration is not necessary as the Serializer/deserializer autodetection is enabled by default.

If you want to deserialize a list of fruits, you need to create a deserializer with a Jackson TypeReference denoted the generic collection used.

First, you need to include the quarkus-jsonb extension.

There is an existing JsonbSerializer that can be used to serialize all data objects via JSON-B. You may create an empty subclass if you want to use Serializer/deserializer autodetection.

The corresponding deserializer class needs to be subclassed. So, let’s create a FruitDeserializer that extends the generic JsonbDeserializer.

Finally, configure your channels to use the JSON-B serializer and deserializer.

Now, your Kafka messages will contain a JSON-B serialized representation of your Fruit data object.

If you want to deserialize a list of fruits, you need to create a deserializer with a Type denoted the generic collection used.

When using the quarkus-kafka-client extension, you can enable readiness health check by setting the quarkus.kafka.health.enabled property to true in your application.properties. This check reports the status of the interaction with a default Kafka broker (configured using kafka.bootstrap.servers). It requires an admin connection with the Kafka broker, and it is disabled by default. If enabled, when you access the /q/health/ready endpoint of your application, you will have information about the connection validation status.

When using Reactive Messaging and the Kafka connector, each configured channel (incoming or outgoing) provides startup, liveness and readiness checks.

For each channel, you can disable the checks using:

Reactive Messaging startup and readiness checks offer two strategies. The default strategy verifies that an active connection is established with the broker. This approach is not intrusive as it’s based on built-in Kafka client metrics.

Using the health-topic-verification-enabled=true attribute, startup probe uses an admin client to check for the list of topics. Whereas the readiness probe for an incoming channel checks that at least one partition is assigned for consumption, and for an outgoing channel checks that the topic used by the producer exist in the broker.

Note that to achieve this, an admin connection is required. You can adjust the timeout for topic verification calls to the broker using the health-topic-verification-timeout configuration.

Per channel metrics can also be gathered and exposed as Micrometer meters. Following metrics can be gathered per channel, identified with the channel tag:

For backwards compatibility reasons channel metrics are not enabled by default and can be enabled with:

It can be useful to test the application without having to start a Kafka broker. To achieve this, you can switch the channels managed by the Kafka connector to in-memory.

Let’s say we want to test the following processor application:

First, add the following test dependency to your application:

Then, create a Quarkus Test Resource as follows:

Create a Quarkus Test using the test resource created above:

If your Kafka consumer is batch based, you will need to send a batch of messages to the channel as by creating them manually.

For instance:

If you are using Servicios de desarrollo para Kafka, a Kafka broker will be started and available throughout the tests, unless it is disabled in %test profile. While it is possible to connect to this broker using Kafka Clients API, Kafka Companion Library proposes an easier way of interacting with a Kafka broker and, creating consumer, producer and admin actions inside tests.

For using KafkaCompanion API in tests, start by adding the following dependency:

which provides io.quarkus.test.kafka.KafkaCompanionResource - an implementation of io.quarkus.test.common.QuarkusTestResourceLifecycleManager.

Then use @QuarkusTestResource to configure the Kafka Companion in tests, for example:

Los servicios de desarrollo para Kafka se activan automáticamente a menos que:

Dev Services for Kafka relies on Docker to start the broker. If your environment does not support Docker, you will need to start the broker manually, or connect to an already running broker. You can configure the broker address using kafka.bootstrap.servers.

Most of the time you need to share the broker between applications. Dev Services for Kafka implements a service discovery mechanism for your multiple Quarkus applications running in dev mode to share a single broker.

If you need multiple (shared) brokers, you can configure the quarkus.kafka.devservices.service-name attribute and indicate the broker name. It looks for a container with the same value, or starts a new one if none can be found. The default service name is kafka.

Sharing is enabled by default in dev mode, but disabled in test mode. You can disable the sharing with quarkus.kafka.devservices.shared=false.

By default, Dev Services for Kafka picks a random port and configures the application. You can set the port by configuring the quarkus.kafka.devservices.port property.

Tenga en cuenta que la dirección anunciada de Kafka se configura automáticamente con el puerto elegido.

Dev Services for Kafka supports Redpanda, kafka-native and Strimzi (in Kraft mode) images.

Redpanda is a Kafka compatible event streaming platform. Because it provides a fast startup times, Dev Services defaults to Redpanda images from vectorized/redpanda. You can select any version from https://hub.docker.com/r/vectorized/redpanda.

kafka-native provides images of standard Apache Kafka distribution compiled to native binary using Quarkus and GraalVM. While still being experimental, it provides very fast startup times with small footprint.

Image type can be configured using

Strimzi provides container images and Operators for running Apache Kafka on Kubernetes. While Strimzi is optimized for Kubernetes, the images work perfectly in classic container environments. Strimzi container images run "genuine" Kafka broker on JVM, which is slower to start.

Para Strimzi, puede seleccionar cualquier imagen con una versión de Kafka que tenga soporte para Kraft (2.8.1 y superior) desde https://quay.io/repository/strimzi-test-container/test-container?tab=tags

You can configure the Dev Services for Kafka to create topics once the broker is started. Topics are created with given number of partitions and 1 replica.

El siguiente ejemplo crea un tema llamado test con 3 particiones, y un segundo tema llamado messages con 2 particiones.

If a topic already exists with the given name, the creation is skipped, without trying to re-partition the existing topic to a different number of partitions.

Puede configurar el tiempo de espera para las llamadas del cliente Kafka admin utilizadas en la creación de temas utilizando quarkus.kafka.devservices.topic-partitions-timeout, por defecto es de 2 segundos.

By default, the Redpanda broker is configured to enable transactions and idempotence features. You can disable those using:

The following attributes are configured using:

Some properties have aliases which can be configured globally:

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

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

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.

The following attributes are configured using:

Some properties have aliases which can be configured globally:

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

For example, to configure the max.block.ms property, use:

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 [client-id-prefix][channel-name].

Quarkus exposes all Kafka related application properties, prefixed with kafka. or KAFKA_ inside a configuration map with default-kafka-broker name. This configuration is used to establish the connection with the Kafka broker.

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

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.

Attribute values are resolved as follows:

Puede configurar los canales utilizando un perfil específico. Así, los canales sólo se configuran (y se añaden a la aplicación) cuando el perfil especificado está activado.

Para conseguirlo, necesitas:

Tenga en cuenta que la mensajería reactiva verifica que el grafo esté completo. Por lo tanto, cuando utilice una configuración condicional de este tipo, asegúrese de que la aplicación funciona con y sin el perfil activado.

Tenga en cuenta que este enfoque también se puede utilizar para cambiar la configuración del canal en función de un perfil.

To send messages to Kafka from an HTTP endpoint, inject an Emitter (or a MutinyEmitter) in your endpoint:

The endpoint sends the passed payload (from a POST HTTP request) to the emitter. The emitter’s channel is mapped to a Kafka topic in the application.properties file:

The endpoint returns a CompletionStage indicating the asynchronous nature of the method. The emitter.send method returns a CompletionStage<Void> . The returned future is completed when the message has been written to Kafka. If the writing fails, the returned CompletionStage is completed exceptionally.

If the endpoint does not return a CompletionStage, the HTTP response may be written before the message is sent to Kafka, and so failures won’t be reported to the user.

If you need to send a Kafka record, use:

To persist objects received from Kafka into a database, you can use Hibernate with Panache.

Let’s imagine you receive Fruit objects. For simplicity purposes, our Fruit class is pretty simple:

To consume Fruit instances stored on a Kafka topic, and persist them into a database, you can use the following approach:

As mentioned in <4>, you need a deserializer that can create a Fruit from the record. This can be done using a Jackson deserializer:

The associated configuration would be:

Check Serializing via Jackson for more detail about the usage of Jackson with Kafka. You can also use Avro.

To persist objects received from Kafka into a database, you can use Hibernate Reactive with Panache.

Let’s imagine you receive Fruit objects. For simplicity purposes, our Fruit class is pretty simple:

To consume Fruit instances stored on a Kafka topic, and persist them into a database, you can use the following approach:

Unlike with classic Hibernate, you can’t use @Transactional. Instead, we use session.withTransaction and persist our entity. The map is used to return a Uni<Void> and not a Uni<Fruit>.

You need a deserializer that can create a Fruit from the record. This can be done using a Jackson deserializer:

The associated configuration would be:

Check Serializing via Jackson for more detail about the usage of Jackson with Kafka. You can also use Avro.

Let’s imagine the following process:

Because we write to a database, we must run this method in a transaction. Yet, sending the entity to Kafka happens asynchronously. The operation returns a CompletionStage (or a Uni if you use a MutinyEmitter) reporting when the operation completes. We must be sure that the transaction is still running until the object is written. Otherwise, you may access the object outside the transaction, which is not allowed.

To implement this process, you need the following approach:

To send to Kafka entities managed by Hibernate Reactive, we recommend using:

The following example demonstrates how to receive a payload, store it in the database using Hibernate Reactive with Panache, and send the persisted entity to Kafka:

Streaming a Kafka topic as server-sent events (SSE) is straightforward:

The following code provides an example:

Some environment cuts the SSE connection when there is not enough activity. The workaround consists of sending ping messages (or empty objects) periodically.

The workaround is a bit more complex as besides sending the fruits coming from Kafka, we need to send pings periodically. To achieve this we merge the stream coming from Kafka and a periodic stream emitting {} every 10 seconds.

By chaining a Kafka transaction with a Hibernate Reactive transaction you can send records to a Kafka transaction, perform database updates and commit the Kafka transaction only if the database transaction is successful.

The following example demonstrates:

In the previous example the database transaction (inner) will commit followed by the Kafka transaction (outer). If you wish to commit the Kafka transaction first and the database transaction second, you need to nest them in the reverse order.

The next example demonstrates that using the Hibernate Reactive API (without Panache):

Azure Event Hub provides an endpoint compatible with Apache Kafka.

To connect to Azure Event Hub, using the Kafka protocol with TLS, you need the following configuration:

Replace <YOUR.EVENTHUBS.CONNECTION.STRING> with the connection string for your Event Hubs namespace. For instructions on getting the connection string, see Get an Event Hubs connection string. The result would be something like:

This configuration can be global (as above), or set in the channel configuration:

Red Hat OpenShift Streams for Apache Kafka provides managed Kafka brokers. First, follow the instructions from Getting started with the rhoas CLI for Red Hat OpenShift Streams for Apache Kafka to create your Kafka broker instance. Make sure you copied the client id and client secret associated with the ServiceAccount you created.

Then, you can configure the Quarkus application to connect to the broker as follows:

When using Kubernetes, it is recommended to set the client id and secret in a Kubernetes secret:

To allow your Quarkus application to use that secret, add the following line to the application.properties file: