From 1d08dac715b72cea40eef88dddf1cfa28c014963 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 7 Mar 2016 18:15:51 -0500 Subject: [PATCH] GH-3: Fix race condition in the partitions test Fixes GH-3 (https://github.com/spring-projects/spring-kafka/issues/3) The default `OffsetResetStrategy` is `LATEST` for the `KafkaConsumer` independently of the `KafkaMessageListenerContainer` option. Therefore the `Thread.sleep(1000)` might not be enough in the test to wait before publishing messages to the topic. With the race condition the listener may be initialized in the latest offset bypassing the required messages to poll. * Add `initialConsumersLatch` to wait until the first `Consumer.poll()` will finish, assuming the offset position has been updated already. * Remove dangerous `Thread.sleep(1000)` * Add `assert` for the `latch2` from the second listener to be sure that everything works as expected. --- .../core/DefaultKafkaConsumerFactory.java | 3 +- ...ncurrentMessageListenerContainerTests.java | 39 +++++++++++++++++-- .../src/test/resources/log4j.properties | 8 ++++ 3 files changed, 45 insertions(+), 5 deletions(-) create mode 100644 spring-kafka/src/test/resources/log4j.properties diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java index e4162f35..3ee8aaca 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java @@ -20,6 +20,7 @@ import java.util.HashMap; import java.util.Map; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; /** @@ -41,7 +42,7 @@ public class DefaultKafkaConsumerFactory implements ConsumerFactory @Override public boolean isAutoCommit() { - Object auto = this.configs.get("enable.auto.commit"); + Object auto = this.configs.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); return auto instanceof Boolean ? (Boolean) auto : auto instanceof String ? Boolean.valueOf((String) auto) : false; } 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 ba95afbf..419f3f39 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 @@ -37,6 +37,7 @@ import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; import org.junit.ClassRule; import org.junit.Test; @@ -57,6 +58,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils; /** * @author Gary Russell + * @author Artem Bilan * */ public class ConcurrentMessageListenerContainerTests { @@ -150,9 +152,32 @@ public class ConcurrentMessageListenerContainerTests { @Test public void testDefinedPartitions() throws Exception { logger.info("Start auto parts"); - Map props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka); + final Map props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka); TopicPartition topic1Partition0 = new TopicPartition(topic3, 0); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); + + final CountDownLatch initialConsumersLatch = new CountDownLatch(2); + + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props) { + + @Override + public Consumer createConsumer() { + return new KafkaConsumer(props) { + + @Override + public ConsumerRecords poll(long timeout) { + try { + return super.poll(timeout); + } + finally { + initialConsumersLatch.countDown(); + } + } + + }; + } + + }; + ConcurrentMessageListenerContainer container1 = new ConcurrentMessageListenerContainer<>(cf, topic1Partition0); final CountDownLatch latch1 = new CountDownLatch(2); @@ -163,10 +188,11 @@ public class ConcurrentMessageListenerContainerTests { logger.info("auto part: " + message); latch1.countDown(); } + }); container1.setBeanName("b1"); container1.start(); - Thread.sleep(1000); + TopicPartition topic1Partition1 = new TopicPartition(topic3, 1); ConcurrentMessageListenerContainer container2 = new ConcurrentMessageListenerContainer<>(cf, topic1Partition1); @@ -178,10 +204,13 @@ public class ConcurrentMessageListenerContainerTests { logger.info("auto part: " + message); latch2.countDown(); } + }); container2.setBeanName("b2"); container2.start(); - Thread.sleep(1000); + + assertTrue(initialConsumersLatch.await(20, TimeUnit.SECONDS)); + Map senderProps = KafkaTestUtils.senderProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); @@ -191,7 +220,9 @@ public class ConcurrentMessageListenerContainerTests { template.convertAndSend(0, "baz"); template.convertAndSend(2, "qux"); template.flush(); + assertTrue(latch1.await(60, TimeUnit.SECONDS)); + assertTrue(latch2.await(60, TimeUnit.SECONDS)); container1.stop(); container2.stop(); diff --git a/spring-kafka/src/test/resources/log4j.properties b/spring-kafka/src/test/resources/log4j.properties new file mode 100644 index 00000000..b4e3588b --- /dev/null +++ b/spring-kafka/src/test/resources/log4j.properties @@ -0,0 +1,8 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n + +log4j.category.org.springframework.kafka=TRACE +log4j.category.org.apache.kafka.clients=WARN