Package io.smallrye.reactive.messaging.kafka.commit

Class KafkaThrottledLatestProcessedCommit

java.lang.Object

io.smallrye.reactive.messaging.kafka.commit.ContextHolder

io.smallrye.reactive.messaging.kafka.commit.KafkaThrottledLatestProcessedCommit

All Implemented Interfaces:

KafkaCommitHandler

public class KafkaThrottledLatestProcessedCommit
extends ContextHolder
implements KafkaCommitHandler

Will keep track of received messages and commit to the next offset after the latest ACKed message in sequence. Will commit periodically as defined by `auto.commit.interval.ms` (default: 5000)

This strategy mimics the behavior of the kafka consumer when `enable.auto.commit` is `true`.

The connector will be marked as unhealthy in the presence of any received record that has gone too long without being processed as defined by `throttled.unprocessed-record-max-age.ms` (default: 60000). If `throttled.unprocessed-record-max-age.ms` is set to less than or equal to 0 then will not perform any health check (this might lead to running out of memory).

This strategy guarantees at-least-once delivery even if the channel performs asynchronous processing.

To use set `commit-strategy` to `throttled`.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	KafkaThrottledLatestProcessedCommit.TooManyMessagesWithoutAckException

Nested classes/interfaces inherited from interface io.smallrye.reactive.messaging.kafka.commit.KafkaCommitHandler

KafkaCommitHandler.Strategy

Field Summary

Fields inherited from class io.smallrye.reactive.messaging.kafka.commit.ContextHolder

context, vertx

Method Summary

Modifier and Type	Method	Description
static void	clearCache()
static KafkaThrottledLatestProcessedCommit	create(io.vertx.mutiny.core.Vertx vertx, io.vertx.mutiny.kafka.client.consumer.KafkaConsumer<?,?> consumer, String groupId, KafkaConnectorIncomingConfiguration config, KafkaSource<?,?> source)
<K,V> CompletionStage<Void>	handle(IncomingKafkaRecord<K,V> record)	A message has been acknowledged.
void	partitionsAssigned(Set<io.vertx.kafka.client.common.TopicPartition> partitions)	New partitions are assigned.
<K,V> IncomingKafkaRecord<K,V>	received(IncomingKafkaRecord<K,V> record)	Received a new record from Kafka.
void	terminate()

Methods inherited from class io.smallrye.reactive.messaging.kafka.commit.ContextHolder

getContext, runOnContext

Methods inherited from class java.lang.Object

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

Method Detail

clearCache

public static void clearCache()

create

public static KafkaThrottledLatestProcessedCommit create(io.vertx.mutiny.core.Vertx vertx,
                                                         io.vertx.mutiny.kafka.client.consumer.KafkaConsumer<?,?> consumer,
                                                         String groupId,
                                                         KafkaConnectorIncomingConfiguration config,
                                                         KafkaSource<?,?> source)

partitionsAssigned

public void partitionsAssigned(Set<io.vertx.kafka.client.common.TopicPartition> partitions)

New partitions are assigned. This method is called from a Vert.x event loop (the one used by the Kafka client)

Specified by:

partitionsAssigned in interface KafkaCommitHandler

Parameters:

partitions - the partitions

received

public <K,V> IncomingKafkaRecord<K,V> received(IncomingKafkaRecord<K,V> record)

Received a new record from Kafka. This method is called from a Vert.x event loop.

Specified by:

received in interface KafkaCommitHandler

Type Parameters:

K - the key

V - the value

Parameters:

record - the record

Returns:

the record

handle

public <K,V> CompletionStage<Void> handle(IncomingKafkaRecord<K,V> record)

A message has been acknowledged. This method is NOT necessarily called on an event loop.

Specified by:

handle in interface KafkaCommitHandler

Type Parameters:

K - the key

V - the value

Parameters:

record - the record

Returns:

a completion stage indicating when the commit complete

terminate

public void terminate()

Specified by:

terminate in interface KafkaCommitHandler