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.

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<K, V> implements ConsumerFactory<K, V>
 
 	@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;
 	}

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<String, Object> props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka);
+		final Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka);
 		TopicPartition topic1Partition0 = new TopicPartition(topic3, 0);
-		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+
+		final CountDownLatch initialConsumersLatch = new CountDownLatch(2);
+
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
+
+			@Override
+			public Consumer<Integer, String> createConsumer() {
+				return new KafkaConsumer<Integer, String>(props) {
+
+					@Override
+					public ConsumerRecords<Integer, String> poll(long timeout) {
+						try {
+							return super.poll(timeout);
+						}
+						finally {
+							initialConsumersLatch.countDown();
+						}
+					}
+
+				};
+			}
+
+		};
+
 		ConcurrentMessageListenerContainer<Integer, String> 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<Integer, String> 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<String, Object> senderProps = KafkaTestUtils.senderProps(embeddedKafka);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
 		KafkaTemplate<Integer, String> 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();
 

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