Provide synchronous approach to partially commit batch

[daniel-shuy]

Affects Version(s): 2.7.3

When using BatchAcknowledgingMessageListener to manually control commits, Acknowledgment#acknowledge() (commit entire batch) will be synchronous/asynchronous depending on whether syncCommits is set to true/false (assuming it is called from the consumer thread and ackMode is AckMode.MANUAL_IMMEDIATE).

But when using Acknowledgment#nack(int, long) to partial commit a batch, the commit is always done asynchronously, regardless of whether syncCommits is true/false. This has 2 downsides:

• Unless an arbitrarily long sleep time is set, the next poll() might be called before the commits complete, resulting in already successfully processed records being polled again
• If the Kafka commit fails, it will not give the caller a chance to handle it (eg. rollback database transaction)

While neither of these issues are dealbreaking (consumers should be idempotent), it would be nice to have a way to partially commit synchronously.

[daniel-shuy]

@garyrussell thanks for taking this up!

To avoid breaking backwards compatibility, I'm guessing its probably better to create a new method instead of modifying the behavior of Acknowledgement#nack(int, long) to maintain backwards compatibility (since the sleep parameter will be redundant if it is synchronous anyways), eg.

public interface Acknowledge {
  // ...

  /**
   * Acknowledge the record at an index in a batch - commit the offset(s) of records before and at the index.
   * @param index the index of the record in the batch to acknowledge.
   **/
  void acknowledge(int index);

  // ...
}

[garyrussell]

I am not sure what you mean. The sleep parameter is not redundant; it was added because, for transient errors, retrying immediately doesn't usually make sense.

[daniel-shuy]

Oh, I meant that it will be redundant if Acknowledgement#nack(int, long) was changed to synchronous (since the retry will be guaranteed to happen after the commit returns), therefore it would be better to introduce a new method to partial commit synchronously than to modify the behavior of Acknowledgement#nack(int, long)

[garyrussell]

I just got around to looking at the code.

But when using Acknowledgment#nack(int, long) to partial commit a batch, the commit is always done asynchronously, regardless of whether syncCommits is true/false. This has 2 downsides:

What makes you think that?

All that method does is set the nackIndex and nackSleep fields.

When the listener exits we decide which records to commit (add to acks) and which to re-seek; here

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

Lines 2046 to 2057 in 7aadccb

	if (this.nackSleep >= 0) {
	int index = 0;
	toSeek = new ArrayList<>();
	for (ConsumerRecord<K, V> record : records) {
	if (index++ >= this.nackIndex) {
	toSeek.add(record);
	}
	else {
	this.acks.put(record);
	}
	}
	}

then

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

Lines 2068 to 2074 in 7aadccb

	if (toSeek != null) {
	if (!this.autoCommit) {
	processCommits();
	}
	SeekUtils.doSeeks(toSeek, this.consumer, null, true, (rec, ex) -> false, this.logger); // NOSONAR
	nackSleepAndReset();
	}

processCommits calls commitIfNecessary, which commits sync or async depending on the property.

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

Lines 2811 to 2816 in 7aadccb

	if (this.syncCommits) {
	commitSync(commits);
	}
	else {
	commitAsync(commits, 0);
	}

In fact, we have a test case that verifies the partial batch is commited sync, by default...

spring-kafka/spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackBatchTests.java

Lines 202 to 205 in 29e93b5

	willAnswer(i -> {
	this.commitLatch.countDown();
	return null;
	}).given(consumer).commitSync(anyMap(), any());

I changed it to MANUAL_IMMEDIATE and it still works (MANUAL and MANUAL_IMMEDIATE are effectively the same for batch listeners),

[garyrussell]

Furthermore; committed offsets have nothing to do with what poll() returns - it's position() that determines the next records retrieved on the poll; seek changes the position().

[garyrussell]

MANUAL and MANUAL_IMMEDIATE are effectively the same for batch listeners.

They are effectively the same, but MANUAL is more efficient. MANUAL_IMMEDIATE commits them one-at-atime, whereas MANUAL commits them as a batch.

We should consider Coercing MANUAL_IMMEDIATE to MANUAL for batch listeners.