diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 5e80a28c..5367794f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -72,6 +72,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * @author Gary Russell * @author Murali Reddy * @author Marius Bogoevici + * @author Martin Dam */ public class KafkaMessageListenerContainer extends AbstractMessageListenerContainer { @@ -428,7 +429,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } this.unsent = checkPause(this.unsent); - if (!this.paused && !this.autoCommit) { + if (!this.autoCommit) { processCommits(); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 4c06b484..e732ea81 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -113,7 +113,7 @@ public class EnableKafkaIntegrationTests { public void testSimple() throws Exception { template.send("annotated1", 0, "foo"); template.flush(); - assertThat(this.listener.latch1.await(20, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.latch1.await(60, TimeUnit.SECONDS)).isTrue(); template.send("annotated2", 0, 123, "foo"); template.flush(); @@ -182,21 +182,21 @@ public class EnableKafkaIntegrationTests { public void testInterface() throws Exception { template.send("annotated7", 0, "foo"); template.flush(); - assertThat(this.ifaceListener.getLatch1().await(20, TimeUnit.SECONDS)).isTrue(); + assertThat(this.ifaceListener.getLatch1().await(60, TimeUnit.SECONDS)).isTrue(); } @Test public void testMulti() throws Exception { template.send("annotated8", 0, "foo"); template.flush(); - assertThat(this.multiListener.latch1.await(20, TimeUnit.SECONDS)).isTrue(); + assertThat(this.multiListener.latch1.await(60, TimeUnit.SECONDS)).isTrue(); } @Test public void testTx() throws Exception { template.send("annotated9", 0, "foo"); template.flush(); - assertThat(this.ifaceListener.getLatch2().await(20, TimeUnit.SECONDS)).isTrue(); + assertThat(this.ifaceListener.getLatch2().await(60, TimeUnit.SECONDS)).isTrue(); } @Test diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 552c5ac6..5483d90c 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -59,7 +59,7 @@ import org.springframework.retry.support.RetryTemplate; * Tests for the listener container. * * @author Gary Russell - * + * @author Martin Dam */ public class KafkaMessageListenerContainerTests { @@ -73,8 +73,10 @@ public class KafkaMessageListenerContainerTests { private static String topic4 = "testTopic4"; + private static String topic5 = "testTopic5"; + @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5); @Rule public TestName testName = new TestName(); @@ -192,6 +194,54 @@ public class KafkaMessageListenerContainerTests { logger.info("Stop " + this.testName.getMethodName() + ackMode); } + @Test + public void testSlowConsumerCommitsAreProcessed() throws Exception { + Map props = KafkaTestUtils.consumerProps("slow", "false", embeddedKafka); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); + ContainerProperties containerProps = new ContainerProperties(topic5); + containerProps.setAckCount(1); + containerProps.setPauseAfter(100); + containerProps.setAckMode(AckMode.MANUAL); + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + final CountDownLatch latch = new CountDownLatch(3); + containerProps.setMessageListener((AcknowledgingMessageListener) (message, ack) -> { + logger.info("slow: " + message); + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + ack.acknowledge(); + latch.countDown(); + }); + container.setBeanName("testSlow"); + container.start(); + ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); + Consumer consumer = spyOnConsumer(container); + + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic5); + template.sendDefault(0, "foo"); + template.sendDefault(2, "bar"); + template.flush(); + Thread.sleep(300); + template.sendDefault(0, "fiz"); + template.sendDefault(2, "buz"); + template.flush(); + + // Verify that commitSync is called when paused + assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); + verify(consumer, atLeastOnce()).pause(any(TopicPartition.class), any(TopicPartition.class)); + verify(consumer, atLeastOnce()).commitSync(any()); + verify(consumer, atLeastOnce()).resume(any(TopicPartition.class), any(TopicPartition.class)); + container.stop(); + } + @Test public void testSlowConsumerWithException() throws Exception { logger.info("Start " + this.testName.getMethodName());