reactor.kafka.receiver

Interface ReceiverOptions<K,V>

public interface ReceiverOptions<K,V>

Nested Class Summary

Nested Classes

Modifier and Type	Interface and Description
static interface	ReceiverOptions.ConsumerListener Called whenever a consumer is added or removed.

Method Summary

Modifier and Type	Method and Description
ReceiverOptions<K,V>	addAssignListener(Consumer<Collection<ReceiverPartition>> onAssign) Adds a listener for partition assignments.
ReceiverOptions<K,V>	addRevokeListener(Consumer<Collection<ReceiverPartition>> onRevoke) Adds a listener for partition revocations.
List<Consumer<Collection<ReceiverPartition>>>	assignListeners() Returns list of configured partition assignment listeners.
Collection<org.apache.kafka.common.TopicPartition>	assignment() Returns the collection of partitions to be assigned if this instance is configured for manual partition assignment.
ReceiverOptions<K,V>	assignment(Collection<org.apache.kafka.common.TopicPartition> partitions) Sets subscription using manual assignment to the specified partitions.
int	atmostOnceCommitAheadSize() Returns the maximum difference between the offset committed for at-most-once delivery and the offset of the last record dispatched.
ReceiverOptions<K,V>	atmostOnceCommitAheadSize(int commitAheadSize) Configures commit ahead size per partition for at-most-once delivery.
default String	bootstrapServers() Return the bootstrap servers from the provided ConsumerConfig.
ReceiverOptions<K,V>	clearAssignListeners() Removes all partition assignment listeners.
ReceiverOptions<K,V>	clearRevokeListeners() Removes all partition revocation listeners.
default String	clientId() Return the client id provided by the ConsumerConfig.
Duration	closeTimeout() Returns timeout for graceful shutdown of KafkaConsumer.
ReceiverOptions<K,V>	closeTimeout(Duration timeout) Sets timeout for graceful shutdown of KafkaConsumer.
int	commitBatchSize() Returns the configured commit batch size for automatic commits of acknowledged records.
ReceiverOptions<K,V>	commitBatchSize(int commitBatchSize) Configures commit batch size for automatic commits.
Duration	commitInterval() Returns the configured commit interval for automatic commits of acknowledged records.
ReceiverOptions<K,V>	commitInterval(Duration commitInterval) Configures commit interval for automatic commits.
default long	commitIntervalDuringDelay() Get how often to commit offsets, in milliseconds, while a rebalance is being delayed.
default ReceiverOptions<K,V>	commitIntervalDuringDelay(long interval) Set how often to commit offsets, in milliseconds, while a rebalance is being delayed.
Duration	commitRetryInterval() Returns the configured retry commit interval for commits that fail with non-fatal RetriableCommitFailedExceptions.
ReceiverOptions<K,V>	commitRetryInterval(Duration commitRetryInterval) Configures the retry commit interval for commits that fail with non-fatal RetriableCommitFailedException.
default ReceiverOptions.ConsumerListener	consumerListener() Returns the function that will be applied after a consumer is created but before it is subscribed.
default ReceiverOptions<K,V>	consumerListener(ReceiverOptions.ConsumerListener listener) Set a function that will be applied after a consumer is created but before it is subscribed.
Map<String,Object>	consumerProperties() Returns the configuration properties of the underlying KafkaConsumer.
Object	consumerProperty(String name) Returns the KafkaConsumer configuration property value for the specified option name.
ReceiverOptions<K,V>	consumerProperty(String name, Object newValue) Sets KafkaConsumer configuration property to the specified value.
static <K,V> ReceiverOptions<K,V>	create() Creates an options instance with default properties.
static <K,V> ReceiverOptions<K,V>	create(Map<String,Object> configProperties) Creates an options instance with the specified config overrides for KafkaConsumer.
static <K,V> ReceiverOptions<K,V>	create(Properties configProperties) Creates an options instance with the specified config overrides for KafkaConsumer.
String	groupId() Returns the configured Kafka consumer group id.
Duration	heartbeatInterval() Returns the configured heartbeat interval for Kafka consumer.
org.apache.kafka.common.serialization.Deserializer<K>	keyDeserializer() Returns optionally a deserializer witch is used by KafkaConsumer for key deserialization.
int	maxCommitAttempts() Returns the maximum number of consecutive non-fatal commit failures that are tolerated.
ReceiverOptions<K,V>	maxCommitAttempts(int maxAttempts) Configures the maximum number of consecutive non-fatal RetriableCommitFailedException commit failures that are tolerated.
default int	maxDeferredCommits() When greater than 0, enables out of order commit sequencing.
default ReceiverOptions<K,V>	maxDeferredCommits(int maxDeferred) Set to greater than 0 to enable out of order commit sequencing.
default Duration	maxDelayRebalance() Get the maximum amount of time to delay a rebalance until existing records in the pipeline have been processed.
default ReceiverOptions<K,V>	maxDelayRebalance(Duration maxDelay) Set the maximum amount of time to delay a rebalance until existing records in the pipeline have been processed.
KafkaReceiverObservationConvention	observationConvention() Return a KafkaReceiverObservationConvention to support a publishing operation observation.
io.micrometer.observation.ObservationRegistry	observationRegistry() Return an ObservationRegistry to observe Kafka record consuming operation.
default boolean	pauseAllAfterRebalance() When true, pause all partitions on assignment after rebalance, if any partitions were paused by User before the rebalance.
default ReceiverOptions<K,V>	pauseAllAfterRebalance(boolean pauseAll) When true, pause all partitions on assignment after rebalance, if any partitions were paused by User before the rebalance.
Duration	pollTimeout() Returns the timeout for each KafkaConsumer.poll(long) operation.
ReceiverOptions<K,V>	pollTimeout(Duration timeout) Sets the timeout for each KafkaConsumer.poll(long) operation.
List<Consumer<Collection<ReceiverPartition>>>	revokeListeners() Returns list of configured partition revocation listeners.
Supplier<Scheduler>	schedulerSupplier() Returns the Supplier for a Scheduler that Records will be published on
ReceiverOptions<K,V>	schedulerSupplier(Supplier<Scheduler> schedulerSupplier) Configures the Supplier for a Scheduler on which Records will be published
default Consumer<org.apache.kafka.clients.consumer.Consumer<K,V>>	subscriber(org.apache.kafka.clients.consumer.ConsumerRebalanceListener listener) Returns the KafkaConsumer.subscribe(Collection, ConsumerRebalanceListener), KafkaConsumer.subscribe(Pattern, ConsumerRebalanceListener) or KafkaConsumer.assign(Collection) operation corresponding to the subscription or assignment options configured for this instance.
ReceiverOptions<K,V>	subscription(Collection<String> topics) Sets subscription using group management to the specified collection of topics.
ReceiverOptions<K,V>	subscription(Pattern pattern) Sets subscription using group management to the specified pattern.
Pattern	subscriptionPattern() Returns the Pattern by which the topic should be selected
Collection<String>	subscriptionTopics() Returns the collection of Topics to be subscribed
org.apache.kafka.common.serialization.Deserializer<V>	valueDeserializer() Returns optionally a deserializer witch is used by KafkaConsumer for value deserialization.
ReceiverOptions<K,V>	withKeyDeserializer(org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer) Set a concrete deserializer instant to be used by the KafkaConsumer for keys.
default ReceiverOptions<K,V>	withObservation(io.micrometer.observation.ObservationRegistry observationRegistry) Configure an ObservationRegistry to observe Kafka record consuming operation.
ReceiverOptions<K,V>	withObservation(io.micrometer.observation.ObservationRegistry observationRegistry, KafkaReceiverObservationConvention observationConvention) Configure an ObservationRegistry to observe Kafka record receiving operation.
ReceiverOptions<K,V>	withValueDeserializer(org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer) Set a concrete deserializer instant to be used by the KafkaConsumer for values.

Method Detail

create

@NonNull
static <K,V> ReceiverOptions<K,V> create()

Creates an options instance with default properties.

Returns:

new instance of receiver options

create

@NonNull
static <K,V> ReceiverOptions<K,V> create(@NonNull
                                                  Map<String,Object> configProperties)

Creates an options instance with the specified config overrides for KafkaConsumer.

Returns:

new instance of receiver options

create

@NonNull
static <K,V> ReceiverOptions<K,V> create(@NonNull
                                                  Properties configProperties)

Creates an options instance with the specified config overrides for KafkaConsumer.

Returns:

new instance of receiver options

consumerProperty

@NonNull
ReceiverOptions<K,V> consumerProperty(@NonNull
                                               String name,
                                               @NonNull
                                               Object newValue)

Sets KafkaConsumer configuration property to the specified value.

Returns:

options instance with updated Kafka consumer property

withKeyDeserializer

@NonNull
ReceiverOptions<K,V> withKeyDeserializer(@NonNull
                                                  org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer)

Set a concrete deserializer instant to be used by the KafkaConsumer for keys. Overrides any setting of the ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG property.

Parameters:

keyDeserializer - key deserializer to use in the consumer

Returns:

options instance with new key deserializer

withValueDeserializer

@NonNull
ReceiverOptions<K,V> withValueDeserializer(@NonNull
                                                    org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer)

Set a concrete deserializer instant to be used by the KafkaConsumer for values. Overrides any setting of the ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG property.

Parameters:

valueDeserializer - value deserializer to use in the consumer

Returns:

options instance with new value deserializer

pollTimeout

@NonNull
ReceiverOptions<K,V> pollTimeout(@NonNull
                                          Duration timeout)

Sets the timeout for each KafkaConsumer.poll(long) operation. Since the underlying Kafka consumer is not thread-safe, long poll intervals may delay commits and other operations invoked using KafkaReceiver.doOnConsumer(java.util.function.Function). Very short timeouts may reduce batching and increase load on the broker,

Returns:

options instance with new poll timeout

closeTimeout

@NonNull
ReceiverOptions<K,V> closeTimeout(@NonNull
                                           Duration timeout)

Sets timeout for graceful shutdown of KafkaConsumer.

Returns:

options instance with new close timeout

addAssignListener

@NonNull
ReceiverOptions<K,V> addAssignListener(@NonNull
                                                Consumer<Collection<ReceiverPartition>> onAssign)

Adds a listener for partition assignments. Applications can use this listener to seek to different offsets of the assigned partitions using any of the seek methods in ReceiverPartition. When group management is used, assign listeners are invoked after every rebalance operation. With manual partition assignment using assignment(), assign listeners are invoked once when the receive Flux is subscribed to.

Returns:

options instance with new partition assignment listener

addRevokeListener

@NonNull
ReceiverOptions<K,V> addRevokeListener(@NonNull
                                                Consumer<Collection<ReceiverPartition>> onRevoke)

Adds a listener for partition revocations. Applications can use this listener to commit offsets if required. Acknowledged offsets are committed automatically on revocation. When group management is used, revoke listeners are invoked before every rebalance operation. With manual partition assignment using assignment(), revoke listeners are invoked once when the receive Flux is terminated.

Returns:

options instance with new partition revocation listener

clearAssignListeners

@NonNull
ReceiverOptions<K,V> clearAssignListeners()

Removes all partition assignment listeners.

Returns:

options instance without any partition assignment listeners

clearRevokeListeners

@NonNull
ReceiverOptions<K,V> clearRevokeListeners()

Removes all partition revocation listeners.

Returns:

options instance without any partition revocation listeners

assignment

@NonNull
ReceiverOptions<K,V> assignment(Collection<org.apache.kafka.common.TopicPartition> partitions)

Sets subscription using manual assignment to the specified partitions. This assignment is enabled when the receive Flux of a KafkaReceiver using this options instance is subscribed to. Any existing subscriptions or assignments on this option are deleted.

Returns:

options instance with new partition assignment

subscription

@NonNull
ReceiverOptions<K,V> subscription(Collection<String> topics)

Sets subscription using group management to the specified collection of topics. This subscription is enabled when the receive Flux of a KafkaReceiver using this options instance is subscribed to. Any existing subscriptions or assignments on this option are deleted.

Returns:

options instance with new subscription

subscription

@NonNull
ReceiverOptions<K,V> subscription(Pattern pattern)

Sets subscription using group management to the specified pattern. This subscription is enabled when the receive Flux of a KafkaReceiver using this options instance is subscribed to. Any existing subscriptions or assignments on this option are deleted. Topics are dynamically assigned or removed when topics matching the pattern are created or deleted.

Returns:

options instance with new subscription

commitInterval

@NonNull
ReceiverOptions<K,V> commitInterval(Duration commitInterval)

Configures commit interval for automatic commits. At least one commit operation is attempted within this interval if records are consumed and acknowledged.

If commitInterval is zero, periodic commits based on time intervals are disabled. If commit batch size is configured, offsets are committed when the number of acknowledged offsets reaches the batch size. If commit batch size is also zero, it is the responsibility of the application to explicitly commit records using ReceiverOffset.commit() if required.

If commit interval and commit batch size are configured, a commit operation is scheduled when either the interval or batch size is reached.

Returns:

options instance with new commit interval

commitBatchSize

@NonNull
ReceiverOptions<K,V> commitBatchSize(int commitBatchSize)

Configures commit batch size for automatic commits. At least one commit operation is attempted when the number of acknowledged uncommitted offsets reaches this batch size.

If commitBatchSize is 0, commits are only performed based on commit interval. If commit interval is null, no automatic commits are performed and it is the responsibility of the application to commit offsets explicitly using ReceiverOffset.commit() if required.

If commit batch size and commit interval are configured, a commit operation is scheduled when either the batch size or interval is reached.

Returns:

options instance with new commit batch size

atmostOnceCommitAheadSize

@NonNull
ReceiverOptions<K,V> atmostOnceCommitAheadSize(int commitAheadSize)

Configures commit ahead size per partition for at-most-once delivery. Before dispatching each record, an offset ahead by this size may be committed. The maximum number of records that may be lost if the application fails is commitAheadSize + 1. A high commit ahead size reduces the cost of commits in at-most-once delivery by reducing the number of commits and avoiding blocking before dispatch if the offset corresponding to the record was already committed.

If commitAheadSize is zero (default), offsets are committed synchronously before each record is dispatched for KafkaReceiver.receiveAtmostOnce(). Otherwise, commits are performed ahead of dispatch and record dispatch is blocked only if commits haven't completed.

Returns:

options instance with new commit ahead size

maxCommitAttempts

@NonNull
ReceiverOptions<K,V> maxCommitAttempts(int maxAttempts)

Configures the maximum number of consecutive non-fatal RetriableCommitFailedException commit failures that are tolerated. For manual commits, failure in commit after the configured number of attempts fails the commit operation. For auto commits, the receive Flux is terminated if the commit does not succeed after these attempts.

Returns:

options instance with updated number of commit attempts

See Also:

commitRetryInterval()

commitRetryInterval

@NonNull
ReceiverOptions<K,V> commitRetryInterval(Duration commitRetryInterval)

Configures the retry commit interval for commits that fail with non-fatal RetriableCommitFailedException.

Returns:

options instance with new commit retry interval

Since:

1.3.11

maxDeferredCommits

default ReceiverOptions<K,V> maxDeferredCommits(int maxDeferred)

Set to greater than 0 to enable out of order commit sequencing. If the number of deferred commits exceeds this value, the consumer is paused until the deferred commits are reduced.

Returns:

options instance with updated number of max deferred commits.

Since:

1.3.8

maxDelayRebalance

default ReceiverOptions<K,V> maxDelayRebalance(Duration maxDelay)

Set the maximum amount of time to delay a rebalance until existing records in the pipeline have been processed. Default 60s. Must be somewhat less than max.poll.interval.ms.

Parameters:

maxDelay - the max delay.

Returns:

options updated with the max delay.

Since:

1.3.12

See Also:

commitIntervalDuringDelay(long)

pauseAllAfterRebalance

default ReceiverOptions<K,V> pauseAllAfterRebalance(boolean pauseAll)

When true, pause all partitions on assignment after rebalance, if any partitions were paused by User before the rebalance. Default false

Parameters:

pauseAll -

Returns:

commitIntervalDuringDelay

default ReceiverOptions<K,V> commitIntervalDuringDelay(long interval)

Set how often to commit offsets, in milliseconds, while a rebalance is being delayed. Default 100ms.

Parameters:

interval - the interval.

Returns:

options updated with the interval

Since:

1.3.12

See Also:

maxDelayRebalance(Duration)

schedulerSupplier

@NonNull
ReceiverOptions<K,V> schedulerSupplier(Supplier<Scheduler> schedulerSupplier)

Configures the Supplier for a Scheduler on which Records will be published

Returns:

options instance with updated publishing Scheduler Supplier

consumerListener

default ReceiverOptions<K,V> consumerListener(@Nullable
                                              ReceiverOptions.ConsumerListener listener)

Set a function that will be applied after a consumer is created but before it is subscribed.

Returns:

options instance with the updated function.

Since:

1.3.17

withObservation

@NonNull
default ReceiverOptions<K,V> withObservation(@NonNull
                                                      io.micrometer.observation.ObservationRegistry observationRegistry)

Configure an ObservationRegistry to observe Kafka record consuming operation.

Parameters:

observationRegistry - ObservationRegistry to use.

Returns:

receiver options with updated observationRegistry

Since:

1.3.21

withObservation

@NonNull
ReceiverOptions<K,V> withObservation(@NonNull
                                              io.micrometer.observation.ObservationRegistry observationRegistry,
                                              @Nullable
                                              KafkaReceiverObservationConvention observationConvention)

Configure an ObservationRegistry to observe Kafka record receiving operation. This functionality makes sense only in simple use-cases where it needs to be closed gaps in tracing on this consumer side: an observation is opened and closed immediately to attach respective consumer span to the trace. For more complex (e.g. tracing continuation) the KafkaReceiverObservation API should be used: the tracing and parent span information is extracted from the consumer record.

Parameters:

observationRegistry - ObservationRegistry to use.

observationConvention - the KafkaReceiverObservationConvention to use.

Returns:

receiver options with updated observationRegistry

Since:

1.3.21

consumerProperties

@NonNull
Map<String,Object> consumerProperties()

Returns the configuration properties of the underlying KafkaConsumer.

Returns:

options to configure for Kafka consumer.

consumerProperty

@Nullable
Object consumerProperty(@NonNull
                                  String name)

Returns the KafkaConsumer configuration property value for the specified option name.

Returns:

Kafka consumer configuration option value

keyDeserializer

@Nullable
org.apache.kafka.common.serialization.Deserializer<K> keyDeserializer()

Returns optionally a deserializer witch is used by KafkaConsumer for key deserialization.

Returns:

configured key deserializer instant

valueDeserializer

@Nullable
org.apache.kafka.common.serialization.Deserializer<V> valueDeserializer()

Returns optionally a deserializer witch is used by KafkaConsumer for value deserialization.

Returns:

configured value deserializer instant

pollTimeout

@NonNull
Duration pollTimeout()

Returns the timeout for each KafkaConsumer.poll(long) operation.

Returns:

poll timeout duration

closeTimeout

@NonNull
Duration closeTimeout()

Returns timeout for graceful shutdown of KafkaConsumer.

Returns:

close timeout duration

assignListeners

@NonNull
List<Consumer<Collection<ReceiverPartition>>> assignListeners()

Returns list of configured partition assignment listeners.

Returns:

list of assignment listeners

revokeListeners

@NonNull
List<Consumer<Collection<ReceiverPartition>>> revokeListeners()

Returns list of configured partition revocation listeners.

Returns:

list of revocation listeners

assignment

@Nullable
Collection<org.apache.kafka.common.TopicPartition> assignment()

Returns the collection of partitions to be assigned if this instance is configured for manual partition assignment.

Returns:

partitions to be assigned

subscriptionTopics

@Nullable
Collection<String> subscriptionTopics()

Returns the collection of Topics to be subscribed

Returns:

topics to be assigned

subscriptionPattern

@Nullable
Pattern subscriptionPattern()

Returns the Pattern by which the topic should be selected

Returns:

pattern of topics selection

groupId

@Nullable
String groupId()

Returns the configured Kafka consumer group id.

Returns:

group id

heartbeatInterval

@NonNull
Duration heartbeatInterval()

Returns the configured heartbeat interval for Kafka consumer.

Returns:

heartbeat interval duration

commitInterval

@NonNull
Duration commitInterval()

Returns the configured commit interval for automatic commits of acknowledged records.

Returns:

commit interval duration

commitBatchSize

@NonNull
int commitBatchSize()

Returns the configured commit batch size for automatic commits of acknowledged records.

Returns:

commit batch size

atmostOnceCommitAheadSize

@NonNull
int atmostOnceCommitAheadSize()

Returns the maximum difference between the offset committed for at-most-once delivery and the offset of the last record dispatched. The maximum number of records that may be lost per-partition if the application fails is commitAheadSize + 1

Returns:

commit ahead size for at-most-once delivery

maxCommitAttempts

@NonNull
int maxCommitAttempts()

Returns the maximum number of consecutive non-fatal commit failures that are tolerated. For manual commits, failure in commit after the configured number of attempts fails the commit operation. For auto commits, the receive Flux is terminated.

Returns:

maximum number of commit attempts

commitRetryInterval

@NonNull
Duration commitRetryInterval()

Returns the configured retry commit interval for commits that fail with non-fatal RetriableCommitFailedExceptions.

Returns:

commit interval duration

Since:

1.3.11

See Also:

maxCommitAttempts()

maxDeferredCommits

default int maxDeferredCommits()

When greater than 0, enables out of order commit sequencing. If the number of deferred commits exceeds this value, the consumer is paused until the deferred commits are reduced.

Returns:

the maximum deferred commits.

Since:

1.3.8

maxDelayRebalance

default Duration maxDelayRebalance()

Get the maximum amount of time to delay a rebalance until existing records in the pipeline have been processed. Default 60s.

Returns:

options updated with the max delay.

Since:

1.3.12

See Also:

commitIntervalDuringDelay()

pauseAllAfterRebalance

default boolean pauseAllAfterRebalance()

When true, pause all partitions on assignment after rebalance, if any partitions were paused by User before the rebalance. Default false

Returns:

commitIntervalDuringDelay

default long commitIntervalDuringDelay()

Get how often to commit offsets, in milliseconds, while a rebalance is being delayed. Default 100ms.

Returns:

options updated with the interval

Since:

1.3.12

See Also:

maxDelayRebalance()

schedulerSupplier

@NonNull
Supplier<Scheduler> schedulerSupplier()

Returns the Supplier for a Scheduler that Records will be published on

Returns:

Scheduler Supplier to use for publishing

consumerListener

@Nullable
default ReceiverOptions.ConsumerListener consumerListener()

Returns the function that will be applied after a consumer is created but before it is subscribed.

Returns:

the function.

Since:

1.3.17

observationRegistry

@NonNull
io.micrometer.observation.ObservationRegistry observationRegistry()

Return an ObservationRegistry to observe Kafka record consuming operation.

Returns:

the ObservationRegistry.

Since:

1.3.21

observationConvention

@Nullable
KafkaReceiverObservationConvention observationConvention()

Return a KafkaReceiverObservationConvention to support a publishing operation observation.

Returns:

the KafkaReceiverObservationConvention.

Since:

1.3.21

clientId

@Nullable
default String clientId()

Return the client id provided by the ConsumerConfig.

Returns:

the client id

bootstrapServers

@NonNull
default String bootstrapServers()

Return the bootstrap servers from the provided ConsumerConfig.

Returns:

the bootstrap servers list.

subscriber

@NonNull
default Consumer<org.apache.kafka.clients.consumer.Consumer<K,V>> subscriber(@NonNull
                                                                                      org.apache.kafka.clients.consumer.ConsumerRebalanceListener listener)

Returns the KafkaConsumer.subscribe(Collection, ConsumerRebalanceListener), KafkaConsumer.subscribe(Pattern, ConsumerRebalanceListener) or KafkaConsumer.assign(Collection) operation corresponding to the subscription or assignment options configured for this instance.

Returns:

subscribe or assign operation with rebalance listeners corresponding to this options instance