For non-reactive applications, Kafka’s Producer/Consumer API provides a low latency interface to publish messages to Kafka and consume messages from Kafka.

Applications using Kafka as a message bus using this API may consider switching to Reactor Kafka if the application is implemented in a functional style.

Kafka Connect provides a simple interface to migrate messages from an external data system (e.g. a database) to one or more Kafka topics. Using existing connectors, this migration can be performed without writing any new code.

Applications using the connector API may consider using Reactor Kafka if a reactive API is available for the external data system and transformations are required for the data. When transformations involve other I/O (e.g. to obtain additional information from another database), a reactive pipeline benefits from end-to-end non-blocking back-pressure provided by Reactor. Messages from/to different Kafka partitions can be processed in parallel, improving throughput by avoiding blocking for I/O. The pull model in Reactor controls the pace of messages flowing through the pipeline, enabling efficient use of threads and memory without the need for overflow handling in the application.

Kafka Streams provides lightweight APIs to build stream processing applications that process data stored in Kafka using standard streaming concepts and transformation primitives. Using a simple threading model, the streams API avoids the need for back-pressure. This model works well in cases where transformations do not involve external interactions.

Reactor Kafka is useful for streams applications which process data from Kafka and use external interactions (e.g. get additional data for records from a database) for transformations. In this case, Reactor can provide end-to-end non-blocking back-pressure combined with better utilization of resources if all external interactions use the reactive model.

If you haven’t yet downloaded Kafka, download Kafka version 2.0.0 or higher.

Unzip the release and set KAFKA_DIR to the installation directory. For example,

Start a single-node Zookeeper instance using the Zookeeper installation included in the Kafka download:

Start a single-node Kafka instance:

Create a Kafka topic:

Check that Kafka topic was created successfully:

Download and build Reactor Kafka from github.com/reactor/reactor-kafka/.

Set CLASSPATH for running reactor-kafka samples. CLASSPATH can be obtained using the classpath task of the samples sub-project.

To build your own application using the Reactor Kafka API, you need to include a dependency to Reactor Kafka.

For gradle:

For maven:

SenderOptions#stopOnError() specifies whether each send sequence should fail immediately if one record could not be delivered to Kafka after the configured number of retries or wait until all records have been processed. This can be used along with ProducerConfig#ACKS_CONFIG and ProducerConfig#RETRIES_CONFIG to configure the required quality of service.

If stopOnError is false, a success or error response is returned for each outgoing record. For error responses, the exception from Kafka indicating the reason for send failure is set on SenderResult and can be retrieved using SenderResult#exception(). The Flux fails with an error after attempting to send all records published on outboundRecords. If outboundRecords is a non-terminating Flux, send continues to send records published on this Flux until the result Flux is explicitly cancelled by the user.

If stopOnError is true, a response is returned for the first failed send and the result Flux is terminated immediately with an error. Since multiple outbound messages may be in-flight at any time, it is possible that some messages are delivered successfully to Kafka after the first failure is detected. SenderOptions#maxInFlight() option may be configured to limit the number of messages in-flight at any time.

If individual results are not required for each send request, ProducerRecord can be sent to Kafka without providing correlation metadata using the KafkaOutbound interface. KafkaOutbound is a fluent interface that enables sends to be chained together.

The send sequence is initiated by subscribing to the Mono obtained from KafkaOutbound#then(). The returned Mono completes successfully if all the outbound records are delivered successfully. The Mono terminates on the first send failure. If outboundRecords is a non-terminating Flux, records continue to be sent to Kafka unless a send fails or the returned Mono is cancelled.

Multiple sends can be chained together using a sequence of sends on KafkaOutbound. When the Mono returned from KafkaOutbound#then() is subscribed to, the sends are invoked in sequence in the declaration order. The sequence is cancelled if any of the sends fail after the configured number of retries.

Note that in all cases the retries configured for the KafkaProducer are attempted and failures returned by the reactive KafkaSender indicate a failure to send after the configured number of retry attempts. Retries can result in messages being delivered out of order. The producer property ProducerConfig#MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION may be set to one to avoid re-ordering.

KafkaProducer uses a separate network thread for sending requests and processing responses. To ensure that the producer network thread is never blocked by applications while processing results, KafkaSender delivers responses to applications on a separate scheduler. By default, this is a single threaded pooled scheduler that is freed when no longer required. The scheduler can be overridden if required, for instance, to use a parallel scheduler when the Kafka sends are part of a larger pipeline. This is done on the SenderOptions instance before the KafkaSender instance is created using:

The number of in-flight sends can be controlled using the maxInFlight option. Requests for more elements from upstream are limited by the configured maxInFlight to ensure that the total number of requests at any time for which responses are pending are limited. Along with buffer.memory and max.block.ms options on KafkaProducer, maxInFlight enables control of memory and thread usage when KafkaSender is used in a reactive pipeline. This option can be configured on SenderOptions before the KafkaSender is created. Default value is 256. For small messages, a higher value will improve throughput.

When the KafkaSender is no longer required, the KafkaSender instance can be closed. The underlying KafkaProducer is closed, closing all client connections and freeing all memory used by the producer.

Reactive applications may sometimes require access to the underlying producer instance to perform actions that are not exposed by the KafkaSender interface. For example, an application might need to know the number of partitions in a topic in order to choose the partition to send a record to. Operations that are not provided directly by KafkaSender like send can be run on the underlying KafkaProducer using KafkaSender#doOnProducer.

User provided methods are executed asynchronously. A Mono is returned by doOnProducer which completes with the value returned by the user-provided function.

Since in reactive streams an error represents a terminal signal, any error signal emitted in the inbound Flux will cause the subscription to be cancelled and effectively cause the consumer to shut down. This can be mitigated by using the retry() operator (or retryWhen for finer grained control), which will ensure that a new consumer is created:

Any errors related to the event processing rather than the KafkaConsumer itself should be handled as close to the source as possible and should ideally be prevented from propagating up to the inbound Flux. This is to ensure that the KafkaConsumer doesn’t get restarted unnecessarily due to unrelated application errors.

The example above subscribed to a single Kafka topic. The same API can be used to subscribe to more than one topic by specifying multiple topics in the collection provided to ReceiverOptions#subscription(). Subscription can also be made to a wildcard pattern by specifying a pattern to subscribe to. Group management in KafkaConsumer dynamically updates topic assignment when topics matching the pattern are created or deleted and assigns partitions of matching topics to available consumer instances.

Changes to ReceiverOptions must be made before the receiver instance is created. Altering the subscription deletes any existing subscriptions on the options instance.

Partitions may be manually assigned to the receiver without using Kafka consumer group management.

Existing subscriptions and assignments on the options instance are deleted when a new assignment is specified. Every receiver created from this options instance with manual assignment consumes messages from all the specified partitions.

Commit frequency can be controlled using a combination of commit interval and commit batch size. Commits are performed when either the interval or batch size is reached. One or both of these options may be set on ReceiverOptions before the receiver instance is created. If commit interval is configured, at least one commit is scheduled within that interval if any records were consumed. If commit batch size is configured, a commit is scheduled when the configured number of records are consumed and acknowledged.

Manual acknowledgement of consumed records after processing along with automatic commits based on the configured commit frequency provides at-least-once delivery semantics. Messages are re-delivered if the consuming application crashes after message was dispatched but before it was processed and acknowledged. Only offsets explicitly acknowledged using ReceiverOffset#acknowledge() are committed. Note that acknowledging an offset acknowledges all previous offsets on the same partition. All acknowledged offsets are committed when partitions are revoked during rebalance and when the receive Flux is terminated.

Applications which require fine-grained control over the timing of commit operations can disable periodic commits and explicitly invoke ReceiverOffset#commit() when required to trigger a commit. This commit is asynchronous by default, but the application many invoke Mono#block() on the returned Mono to implement synchronous commits. Applications may batch commits by acknowledging messages as they are consumed and invoking commit() periodically to commit acknowledged offsets.

Note that committing an offset acknowledges and commits all previous offsets on that partition. All acknowledged offsets are committed when partitions are revoked during rebalance and when the receive Flux is terminated.

Starting with version 1.3.12, when a rebalance occurs due to group member changes, the rebalance is delayed until records received from the previous poll have been processed. This is controlled by two ReceiverOptions - maxDelayRebalance (default 60s) and commitIntervalDuringDelay (default 100ms). While the delay is in process, any offsets available for committal will be committed every commitIntervalDuringDelay milliseconds. This allows orderly completion of processing the records that have already been received. maxDelayRebalance should be less than max.poll.interval.ms to avoid a forced rebalance due to a non-responsive consumer.

Starting with version 1.3.8, commits can be performed out of order and the framework will defer the commits as needed, until any "gaps" are filled. This removes the need for applications to keep track of offsets and commit them in the right order. Deferring commits increases the likelihood of duplicate deliveries if the application crashes while deferred commits are present.

To enable this feature, set the maxDeferredCommits property of ReceiverOptions. If the number of deferred offset commits exceeds this value, the consumer is pause() d until the number of deferred commits is reduced by the application acknowledging or commiting some of the "missing" offsets.

The number is an aggregate of deferred commits across all the assigned topics/partitions.

Leaving the property at its default 0 disables the feature and commits are performed whenever called.

KafkaReceiver#receiveAutoAck returns a Flux of batches of records returned by each KafkaConsumer#poll(). The records in each batch are automatically acknowledged when the Flux corresponding to the batch terminates.

The maximum number of records in each batch can be controlled using the KafkaConsumer property MAX_POLL_RECORDS. This is used together with the fetch size and wait times configured on the KafkaConsumer to control the amount of data fetched from Kafka brokers in each poll. Each batch is returned as a Flux that is acknowledged after the Flux terminates. Acknowledged records are committed periodically based on the configured commit interval and batch size. This mode is simple to use since applications do not need to perform any acknowledge or commit actions. It is efficient as well but can not be used for at-least-once delivery of messages.

KafkaReceiver#receiveBatch returns a Flux of batches of records returned by each KafkaConsumer#poll(). The records in each batch should be manually acknowledged or committed.

Same as the KafkaReceiver#receiveAutoAck method, the maximum number of records in each batch can be controlled using the KafkaConsumer property MAX_POLL_RECORDS. This is used together with the fetch size and wait times configured on the KafkaConsumer to control the amount of data fetched from Kafka brokers in each poll. But unlike the KafkaReceiver#receiveAutoAck, each batch is returned as a Flux that should be acknowledged or committed using ReceiverOffset.

As the KafkaReceiver#receive method messages, each message in the batch is represented as a ReceiverRecord which has a committable ReceiverOffset instance.

KafkaReceiver#receiveBatch combines the batch consumption mode of KafkaReceiver#receiveAutoAck with the manual acknowledgement/commit mode of KafkaReceiver#receive. This batching mode is efficient and is easy to use for at-least-once delivery of messages.

Applications which don’t require offset commits to Kafka may disable automatic commits by not acknowledging any records consumed using KafkaReceiver#receive().

Applications may disable automatic commits to avoid re-delivery of records. ConsumerConfig#AUTO_OFFSET_RESET_CONFIG can be configured to "latest" to consume only new records. But this could mean that an unpredictable number of records are not consumed if an application fails and restarts.

KafkaReceiver#receiveAtmostOnce can be used to consume records with at-most-once semantics with a configurable number of records-per-partition that may be lost if the application fails or crashes. Offsets are committed synchronously before the corresponding record is dispatched. Records are guaranteed not to be re-delivered even if the consuming application fails, but some records may not be processed if an application fails after the commit before the records could be processed.

This mode is expensive since each record is committed individually and records are not delivered until the commit operation succeeds. ReceiverOptions#atmostOnceCommitCommitAheadSize may be configured to reduce the cost of commits and avoid blocking before dispatch if the offset of the record has already been committed. By default, commit-ahead is disabled and at-most one record is lost per-partition if an application crashes. If commit-ahead is configured, the maximum number of records that may be lost per-partition is ReceiverOptions#atmostOnceCommitCommitAheadSize + 1.

Applications can enable assignment and revocation listeners to perform any actions when partitions are assigned or revoked from a consumer.

When group management is used, assignment listeners are invoked whenever partitions are assigned to the consumer after a rebalance operation. When manual assignment is used, assignment listeners are invoked when the consumer is started. Assignment listeners can be used to seek to particular offsets in the assigned partitions so that messages are consumed from the specified offset. When a user pauses topics/partitions before rebalancing, the behavior depends on the value of pauseAllAfterRebalance. If it is set to false, the paused topics/partitions will remain paused after the rebalance. However, if it is set to true, all assigned topics/partitions will be paused after the rebalance.

When group management is used, revocation listeners are invoked whenever partitions are revoked from a consumer after a rebalance operation. When manual assignment is used, revocation listeners are invoked before the consumer is closed. Revocation listeners can be used to commit processed offsets when manual commits are used. Acknowledged offsets are automatically committed on revocation if automatic commits are enabled.

By default, receivers start consuming records from the last committed offset of each assigned partition. If a committed offset is not available, the offset reset strategy ConsumerConfig#AUTO_OFFSET_RESET_CONFIG configured for the KafkaConsumer is used to set the start offset to the earliest or latest offset on the partition. Applications can override offsets by seeking to new offsets in an assignment listener. Methods are provided on ReceiverPartition to seek to the earliest, latest, a specific offset in the partition, or to a record with a timestamp later than a point in time.

For example, the following code block starts consuming messages from the latest offset.

Other methods are available on ReceiverPartition to determine the current position, the beginning offset, and ending offset, at the time the partition is assigned.

Each KafkaReceiver instance is associated with a KafkaConsumer that is created when the inbound Flux returned by one of the receive methods in KafkaReceiver is subscribed to. The consumer is kept alive until the Flux completes. When the Flux completes, all acknowledged offsets are committed and the underlying consumer is closed.

Only one receive operation may be active in a KafkaReceiver at any one time. Any of the receive methods can be invoked after the receive Flux corresponding to the last receive is terminated.