Explaination of ack modes in spring kafka

[saravind-hotstar]

Ack modes:
https://docs.spring.io/spring-kafka/api/org/springframework/kafka/listener/ContainerProperties.AckMode.html#COUNT_TIME
ackCount - Does it mean the records successfully processed after a poll
ackTime - Does it mean after X seconds to commit records which are successfully processed

I have a requirement is to commit offsets either after 100 records are processed successfully or 3 seconds have elapsed, which ack mode is suitable for this?

[garyrussell]

Yes, as long as you set the ackMode to AckMode.COUNT_TIME.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java

Lines 83 to 89 in e5d2334

	/**
	* Commit pending offsets after
	* {@link ContainerProperties#setAckCount(int) ackCount} has been
	* exceeded or after {@link ContainerProperties#setAckTime(long)
	* ackTime} has elapsed.
	*/
	COUNT_TIME,

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Lines 635 to 636 in e5d2334

	private final boolean isCountAck = this.containerProperties.getAckMode().equals(AckMode.COUNT)
	|| this.containerProperties.getAckMode().equals(AckMode.COUNT_TIME);

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Lines 2986 to 2993 in e5d2334

	else if (ackMode.equals(AckMode.COUNT_TIME) && elapsed) {
	this.logger.debug(() -> "Committing in AckMode.COUNT_TIME " +
	"because time elapsed exceeds configured limit of " +
	this.containerProperties.getAckTime());
	commitIfNecessary();
	this.last = now;
	this.count = 0;
	}

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Line 2959 in e5d2334

boolean countExceeded = this.isCountAck && this.count >= this.containerProperties.getAckCount();

The actual commit will take place when the listener returns to the container, so the time is only approximate - if the time has been exceeded after processing the current record.

[saravind-hotstar]

@garyrussell Thanks a lot; Just clarifying again, for committing offsets after 100 records or after X time elapse COUNT_TIME is the ackmode right;

[garyrussell]

Yes, except the time is approximate and only when the listener exits will the time be checked.

[arun-hotstar]

@garyrussell Do correct me if my understanding is wrong.

When we use ackMode.COUNT with ackCount as 2 and maxPollRecords as 5, for example, the consumer will commit for every 5 records (assuming that every poll operation is returning 5 records) and not after processing every 2 records. Essentially the lower bound for the incremental offset committed is always the number of records fetched in the poll step. I could also reproduce the behaviour myself for the exact configuration I mentioned at the beginning.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Line 1365 in e5d2334

protected void pollAndInvoke() {

cc: @saravind-hotstar

[garyrussell]

That is correct; with those ack modes, commits are only done between polls. We could look at reworking that, but the intention was to only commit after a larger number than the max.poll.records.

[debraj-manna]

Let's say I have a setup like below

• max.poll.records = 8
• max.poll.interval.ms = 120000 (2 mins)
• ackMode = AckMode.COUNT
• ackCount = 3
• single record processing time = 30 secs

In this case, processing the entire batch of 8 records will take about 4 mins. Once max.poll.interval.ms expires(after processing of 4 records) consumer is moved out of the group(as expected) and if the application is restarted after it, it is again starting from the batch start. So offset is not being committed after ackCount message is processed. Similar behavior, if I use AckMode.TIME with ackTime set to 90 secs (< batch process duration of 4 mins).

@garyrussell Is this expected behavior? I am using spring-kafka 3.0.13.

[garyrussell]

I am no longer involved with this project, but see #2731 - a change was made in 3.1 to address an issue when the ack count was less than the polled records.

[debraj-manna]

Thanks @garyrussell . The ackCount and ackTime is working fine with spring-kafka 3.1.5 and spring-boot 3.2.6 when they are less than poll records and poll time respectively.