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

GH-1111 Propagate the CMLC paused state on start

Browse Source 
The paused state from ConcurrentMessageListenerContainer was not
propagated to the newly instanciated KafkaMessageListenerContainers
when calling start().
This commit is contained in:
Bastien Bouclet
2019-06-04 10:41:49 +02:00
committed by Gary Russell
parent 9d8adf2481
commit d80ade57f7
2 changed files with 44 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
View File

@@ -167,6 +167,9 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
});
publishContainerStoppedEvent();
});
if (isPaused()) {
container.pause();
}
container.start();
this.containers.add(container);
}

42
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
View File

@@ -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() {