Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2891: Fix isAckAfterHandle with No group.id

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/2891

When using manual partition assignment and `AckMode.MANUAL`, it is possible
to have a `null` `group.id`, whereby Kafka does not maintain any committed offsets.

In this case, we should not attempt to commit the offset after recovery, even if
the error handler `ackAfterHandle` property is true.

**cherry-pick to 3.0.x, 2.9.x**

(cherry picked from commit 9f1050a849)

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
This commit is contained in:
Gary Russell
2023-11-16 14:55:17 -05:00
committed by Artem Bilan
parent d2e213a1d3
commit e62ca5c080
1 changed files with 6 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2233,7 +2233,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
try {
this.batchFailed = true;
invokeBatchErrorHandler(records, recordList, e);
commitOffsetsIfNeeded(records);
commitOffsetsIfNeededAfterHandlingError(records);
}
catch (KafkaException ke) {
ke.selfLog(ERROR_HANDLER_THREW_AN_EXCEPTION, this.logger);
@@ -2254,8 +2254,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
return null;
}
private void commitOffsetsIfNeeded(final ConsumerRecords<K, V> records) {
if ((!this.autoCommit && this.commonErrorHandler.isAckAfterHandle())
private void commitOffsetsIfNeededAfterHandlingError(final ConsumerRecords<K, V> records) {
if ((!this.autoCommit && this.commonErrorHandler.isAckAfterHandle() && this.consumerGroupId != null)
|| this.producer != null) {
if (this.remainingRecords != null) {
ConsumerRecord<K, V> firstUncommitted = this.remainingRecords.iterator().next();
@@ -2720,7 +2720,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
try {
invokeErrorHandler(record, iterator, e);
commitOffsetsIfNeeded(record);
commitOffsetsIfNeededAfterHandlingError(record);
}
catch (KafkaException ke) {
ke.selfLog(ERROR_HANDLER_THREW_AN_EXCEPTION, this.logger);
@@ -2738,8 +2738,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
return null;
}
private void commitOffsetsIfNeeded(final ConsumerRecord<K, V> cRecord) {
if ((!this.autoCommit && this.commonErrorHandler.isAckAfterHandle())
private void commitOffsetsIfNeededAfterHandlingError(final ConsumerRecord<K, V> cRecord) {
if ((!this.autoCommit && this.commonErrorHandler.isAckAfterHandle() && this.consumerGroupId != null)
|| this.producer != null) {
if (this.isManualAck) {
this.commitRecovered = true;