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

GH-199: Do not wakeup consumer after queueing ack

Browse Source 
Fixes GH-199 (https://github.com/spring-projects/spring-kafka/issues/199)

Waking up the consumer too often will trigger a rebalance and messages are not actually committed

(cherry picked from commit c245cbb)
This commit is contained in:
Martin Dam
2016-11-02 21:28:31 -04:00
committed by Artem Bilan
parent eb55a3bf69
commit 15102ddb5e
2 changed files with 1 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -603,9 +603,6 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
if (!this.isAnyManualAck && !this.autoCommit) {
this.acks.add(record);
}
if (this.isRecordAck) {
this.consumer.wakeup();
}
}
catch (Exception e) {
if (this.containerProperties.isAckOnError() && !this.autoCommit) {
@@ -619,9 +616,6 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
}
}
}
if (this.isManualAck || this.isBatchAck) {
this.consumer.wakeup();
}
}
private void processCommits() {
@@ -792,9 +786,6 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
ListenerConsumer.this.logger.debug("Interrupt ignored");
}
}
if (!ListenerConsumer.this.isManualImmediateAck && this.active) {
ListenerConsumer.this.consumer.wakeup();
}
}
}
finally {

2
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
View File

@@ -525,7 +525,7 @@ public class KafkaMessageListenerContainerTests {
});
containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.BATCH);
containerProps.setPollTimeout(10000);
containerProps.setPollTimeout(100);
containerProps.setAckOnError(false);
KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,