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
This commit is contained in:
Martin Dam
2016-11-02 18:28:31 -07:00
committed by Artem Bilan
parent 79fceba98b
commit c245cbbe89
2 changed files with 4 additions and 13 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

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

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

@@ -571,7 +571,7 @@ public class KafkaMessageListenerContainerTests {
}); });
containerProps.setSyncCommits(true); containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.BATCH); containerProps.setAckMode(AckMode.BATCH);
containerProps.setPollTimeout(10000); containerProps.setPollTimeout(100);
containerProps.setAckOnError(false); containerProps.setAckOnError(false);
KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf, KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
@@ -640,7 +640,7 @@ public class KafkaMessageListenerContainerTests {
}); });
containerProps.setSyncCommits(true); containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.BATCH); containerProps.setAckMode(AckMode.BATCH);
containerProps.setPollTimeout(10000); containerProps.setPollTimeout(100);
containerProps.setAckOnError(false); containerProps.setAckOnError(false);
KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf, KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
@@ -714,7 +714,7 @@ public class KafkaMessageListenerContainerTests {
}); });
containerProps.setSyncCommits(true); containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE); containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE);
containerProps.setPollTimeout(10000); containerProps.setPollTimeout(100);
containerProps.setAckOnError(false); containerProps.setAckOnError(false);
KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf, KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
@@ -777,7 +777,7 @@ public class KafkaMessageListenerContainerTests {
}); });
containerProps.setSyncCommits(true); containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.BATCH); containerProps.setAckMode(AckMode.BATCH);
containerProps.setPollTimeout(10000); containerProps.setPollTimeout(100);
containerProps.setAckOnError(true); containerProps.setAckOnError(true);
final CountDownLatch latch = new CountDownLatch(4); final CountDownLatch latch = new CountDownLatch(4);
containerProps.setGenericErrorHandler((BatchErrorHandler) (t, messages) -> { containerProps.setGenericErrorHandler((BatchErrorHandler) (t, messages) -> {