reactor.kafka.receiver

Class ReceiverOptions<K,V>

java.lang.Object

reactor.kafka.receiver.ReceiverOptions<K,V>

public class ReceiverOptions<K,V>
extends Object

Configuration properties for Reactive Kafka Receiver and its underlying KafkaConsumer.

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<TopicPartition>	assignment() Returns the collection of partitions to be assigned if this instance is configured for manual partition assignment.
ReceiverOptions<K,V>	assignment(Collection<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.
ReceiverOptions<K,V>	clearAssignListeners() Removes all partition assignment listeners.
ReceiverOptions<K,V>	clearRevokeListeners() Removes all partition revocation listeners.
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.
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.
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.
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.
Consumer<Consumer<K,V>>	subscriber(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.
ReceiverOptions<K,V>	toImmutable() Returns a new immutable instance with the configuration properties of this instance.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

create

public static <K,V> ReceiverOptions<K,V> create()

Creates an options instance with default properties.

Returns:

new instance of receiver options

create

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

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

Returns:

new instance of receiver options

create

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

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

Returns:

new instance of receiver options

consumerProperties

public Map<String,Object> consumerProperties()

Returns the configuration properties of the underlying KafkaConsumer.

Returns:

options to configure for Kafka consumer.

consumerProperty

public Object consumerProperty(String name)

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

Returns:

Kafka consumer configuration option value

consumerProperty

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

Sets KafkaConsumer configuration property to the specified value.

Returns:

options instance with updated Kafka consumer property

pollTimeout

public Duration pollTimeout()

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

Returns:

poll timeout duration

pollTimeout

public ReceiverOptions<K,V> pollTimeout(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 Receiver.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

public Duration closeTimeout()

Returns timeout for graceful shutdown of KafkaConsumer.

Returns:

close timeout duration

closeTimeout

public ReceiverOptions<K,V> closeTimeout(Duration timeout)

Sets timeout for graceful shutdown of KafkaConsumer.

Returns:

options instance with new close timeout

addAssignListener

public ReceiverOptions<K,V> addAssignListener(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

public ReceiverOptions<K,V> addRevokeListener(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

public ReceiverOptions<K,V> clearAssignListeners()

Removes all partition assignment listeners.

Returns:

options instance without any partition assignment listeners

clearRevokeListeners

public ReceiverOptions<K,V> clearRevokeListeners()

Removes all partition revocation listeners.

Returns:

options instance without any partition revocation listeners

assignListeners

public List<Consumer<Collection<ReceiverPartition>>> assignListeners()

Returns list of configured partition assignment listeners.

Returns:

list of assignment listeners

revokeListeners

public List<Consumer<Collection<ReceiverPartition>>> revokeListeners()

Returns list of configured partition revocation listeners.

Returns:

list of revocation listeners

subscription

public 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 Receiver using this options instance is subscribed to. Any existing subscriptions or assignments on this option are deleted.

Returns:

options instance with new subscription

subscription

public 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 Receiver 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

assignment

public ReceiverOptions<K,V> assignment(Collection<TopicPartition> partitions)

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

Returns:

options instance with new partition assignment

assignment

public Collection<TopicPartition> assignment()

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

Returns:

partitions to be assigned

subscriber

public Consumer<Consumer<K,V>> subscriber(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

groupId

public String groupId()

Returns the configured Kafka consumer group id.

Returns:

group id

heartbeatInterval

public Duration heartbeatInterval()

Returns the configured heartbeat interval for Kafka consumer.

Returns:

heartbeat interval duration

commitInterval

public Duration commitInterval()

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

Returns:

commit interval duration

commitInterval

public 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

public int commitBatchSize()

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

Returns:

commit batch size

commitBatchSize

public 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

public 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

atmostOnceCommitAheadSize

public 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 Receiver.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

public 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

maxCommitAttempts

public 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

toImmutable

public ReceiverOptions<K,V> toImmutable()

Returns a new immutable instance with the configuration properties of this instance.

Returns:

new immutable options instance