From d80ade57f7eca6c196a4fb023640aff5ef8c4c5f Mon Sep 17 00:00:00 2001
From: Bastien Bouclet <bastien-externe.bouclet@enedis.fr>
Date: Tue, 4 Jun 2019 10:41:49 +0200
Subject: [PATCH] GH-1111 Propagate the CMLC paused state on start

The paused state from ConcurrentMessageListenerContainer was not
propagated to the newly instanciated KafkaMessageListenerContainers
when calling start().
---
 .../ConcurrentMessageListenerContainer.java   |  3 ++
 ...ncurrentMessageListenerContainerTests.java | 42 ++++++++++++++++++-
 2 files changed, 44 insertions(+), 1 deletion(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
index f7562f8f..5eb94394 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
@@ -167,6 +167,9 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 					});
 					publishContainerStoppedEvent();
 				});
+				if (isPaused()) {
+					container.pause();
+				}
 				container.start();
 				this.containers.add(container);
 			}
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
index 96241f15..84ae48ae 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
@@ -95,9 +95,11 @@ public class ConcurrentMessageListenerContainerTests {
 
 	private static String topic11 = "testTopic11";
 
+	private static String topic12 = "testTopic12";
+
 	@ClassRule
 	public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, topic1, topic2, topic4, topic5,
-			topic6, topic7, topic8, topic9, topic10, topic11);
+			topic6, topic7, topic8, topic9, topic10, topic11, topic12);
 
 	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
 
@@ -410,6 +412,44 @@ public class ConcurrentMessageListenerContainerTests {
 		this.logger.info("Stop MANUAL_IMMEDIATE with Existing");
 	}
 
+	@Test
+	public void testPausedStart() throws Exception {
+		this.logger.info("Start paused start");
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test12", "false", embeddedKafka);
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties(topic12);
+
+		final CountDownLatch latch = new CountDownLatch(2);
+		containerProps.setMessageListener((MessageListener<Integer, String>) message -> {
+			ConcurrentMessageListenerContainerTests.this.logger.info("paused start: " + message);
+			latch.countDown();
+		});
+
+		ConcurrentMessageListenerContainer<Integer, String> container =
+				new ConcurrentMessageListenerContainer<>(cf, containerProps);
+		container.setConcurrency(2);
+		container.setBeanName("testBatch");
+		container.pause();
+		container.start();
+
+		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
+
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(topic12);
+		template.sendDefault(0, "foo");
+		template.sendDefault(2, "bar");
+		template.flush();
+		assertThat(latch.await(100, TimeUnit.MILLISECONDS)).isFalse();
+
+		container.resume();
+
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		this.logger.info("Stop paused start");
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testConcurrencyWithPartitions() {