* GH-757: Upgrade to Apache Kafka 2.0.0

Resolves #757

Kafka 2.0.0 Compatibility

Tested against local build of the kafka trunk branch.

- `poll(long)` is deprecated.
- `poll(Duration)` does not block until the partitions are assigned.
- Several KStream changes.

- Some tests are still failing sporadically

Increase default poll timeout; ensure unique group.id for all tests.

Test against a build from the 2.0.0-rc1 tag instead of trunk.

Rebase; Polishing; test against official 2.0.0 build

* Rebase, polishing.

Author: garyrussell

build.gradle

@@ -76,7 +76,7 @@ subprojects { subproject ->
 		junit4Version = '4.12'
 		junitJupiterVersion = '5.1.1'
 		junitPlatformVersion = '1.1.1'
-		kafkaVersion = '1.1.1'
+		kafkaVersion = '2.0.0'
 		log4jVersion = '2.11.0'
 		mockitoVersion = '2.18.0'
 		scalaVersion = '2.11'

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -18,6 +18,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -28,8 +29,7 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 import org.I0Itec.zkclient.ZkClient;
@@ -405,7 +405,7 @@ public void consumeFromEmbeddedTopics(Consumer<?, ?> consumer, String... topics)
 		assertThat(this.topics)
 				.as("topic(s):'" + diff + "' are not in embedded topic list")
 				.containsAll(new HashSet<>(Arrays.asList(topics)));
-		final CountDownLatch consumerLatch = new CountDownLatch(1);
+		final AtomicBoolean assigned = new AtomicBoolean();
 		consumer.subscribe(Arrays.asList(topics), new ConsumerRebalanceListener() {
 
 			@Override
@@ -414,26 +414,31 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 
 			@Override
 			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
-				consumerLatch.countDown();
+				assigned.set(true);
 				if (logger.isDebugEnabled()) {
 					logger.debug("partitions assigned: " + partitions);
 				}
 			}
 
 		});
-		ConsumerRecords<?, ?> records = consumer.poll(0); // force assignment
-		if (records.count() > 0) {
+		ConsumerRecords<?, ?> records = null;
+		int n = 0;
+		while (!assigned.get() && n++ < 600) {
+			records = consumer.poll(Duration.ofMillis(100)); // force assignment
+		}
+		if (records != null && records.count() > 0) {
+			final ConsumerRecords<?, ?> theRecords = records;
 			if (logger.isDebugEnabled()) {
 				logger.debug("Records received on initial poll for assignment; re-seeking to beginning; "
 						+ records.partitions().stream()
-						.flatMap(p -> records.records(p).stream())
+						.flatMap(p -> theRecords.records(p).stream())
 						// map to same format as send metadata toString()
 						.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
 						.collect(Collectors.toList()));
 			}
 			consumer.seekToBeginning(records.partitions());
 		}
-		assertThat(consumerLatch.await(30, TimeUnit.SECONDS))
+		assertThat(assigned.get())
 				.as("Failed to be assigned partitions from the embedded topics")
 				.isTrue();
 		logger.debug("Subscription Initiated");

spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java

@@ -97,6 +97,7 @@ public KafkaEmbedded(int count, boolean controlledShutdown, int partitions, Stri
 	 * @return this for chaining configuration
 	 * @see KafkaConfig
 	 */
+	@Override
 	public KafkaEmbedded brokerProperties(Map<String, String> brokerProperties) {
 		super.brokerProperties(brokerProperties);
 		return this;
@@ -109,6 +110,7 @@ public KafkaEmbedded brokerProperties(Map<String, String> brokerProperties) {
 	 * @return the {@link KafkaEmbedded}.
 	 * @since 2.1.4
 	 */
+	@Override
 	public KafkaEmbedded brokerProperty(String property, Object value) {
 		super.brokerProperty(property, value);
 		return this;

spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java

@@ -18,6 +18,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -191,7 +192,7 @@ public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer) {
 	 */
 	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, long timeout) {
 		logger.debug("Polling...");
-		ConsumerRecords<K, V> received = consumer.poll(timeout);
+		ConsumerRecords<K, V> received = consumer.poll(Duration.ofMillis(timeout));
 		if (logger.isDebugEnabled()) {
 			logger.debug("Received: " + received.count() + ", "
 					+ received.partitions().stream()

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
 
 package org.springframework.kafka.annotation;
 
+import java.util.Properties;
+
 import org.apache.kafka.streams.StreamsConfig;
 
 import org.springframework.beans.factory.ObjectProvider;
@@ -35,6 +37,7 @@
  * annotation. See {@link EnableKafkaStreams} Javadoc for complete usage.
  *
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 1.1.4
  */
@@ -54,16 +57,16 @@ public class KafkaStreamsDefaultConfiguration {
 
 	@Bean(name = DEFAULT_STREAMS_BUILDER_BEAN_NAME)
 	public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder(
-			@Qualifier(DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider<StreamsConfig> streamsConfigProvider) {
-		StreamsConfig streamsConfig = streamsConfigProvider.getIfAvailable();
+			@Qualifier(DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider<Properties> streamsConfigProvider) {
+		Properties streamsConfig = streamsConfigProvider.getIfAvailable();
 		if (streamsConfig != null) {
 			return new StreamsBuilderFactoryBean(streamsConfig);
 		}
 		else {
 			throw new UnsatisfiedDependencyException(KafkaStreamsDefaultConfiguration.class.getName(),
 					DEFAULT_STREAMS_BUILDER_BEAN_NAME, "streamsConfig", "There is no '" +
-					DEFAULT_STREAMS_CONFIG_BEAN_NAME + "' StreamsConfig bean in the application context.\n" +
-					"Consider to declare one or don't use @EnableKafkaStreams.");
+					DEFAULT_STREAMS_CONFIG_BEAN_NAME + "' Properties bean in the application context.\n" +
+					"Consider declaring one or don't use @EnableKafkaStreams.");
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java

@@ -17,6 +17,7 @@
 package org.springframework.kafka.core;
 
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.streams.KafkaClientSupplier;
@@ -56,6 +57,8 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private StreamsConfig streamsConfig;
 
+	private Properties properties;
+
 	private final CleanupConfig cleanupConfig;
 
 	private KafkaStreamsCustomizer kafkaStreamsCustomizer;
@@ -77,9 +80,10 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 	private volatile boolean running;
 
 	/**
-	 * Default constructor that creates the factory without a {@link StreamsConfig}.
-	 * It is the factory user's responsibility to properly set {@link StreamsConfig}
-	 * using {@link StreamsBuilderFactoryBean#setStreamsConfig(StreamsConfig)}
+	 * Default constructor that creates the factory without configuration
+	 * {@link Properties}. It is the factory user's responsibility to properly set
+	 * {@link Properties} using
+	 * {@link StreamsBuilderFactoryBean#setStreamsConfiguration(Properties)}.
 	 * @since 2.1.3.
 	 */
 	public StreamsBuilderFactoryBean() {
@@ -100,29 +104,58 @@ public StreamsBuilderFactoryBean(StreamsConfig streamsConfig) {
 	 * @param streamsConfig the streams configuration.
 	 * @param cleanupConfig the cleanup configuration.
 	 * @since 2.1.2.
+	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(Properties, CleanupConfig)}
 	 */
+	@Deprecated
 	public StreamsBuilderFactoryBean(StreamsConfig streamsConfig, CleanupConfig cleanupConfig) {
 		Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
 		Assert.notNull(cleanupConfig, "'cleanupConfig' must not be null");
 		this.streamsConfig = streamsConfig;
 		this.cleanupConfig = cleanupConfig;
 	}
 
+	/**
+	 * Construct an instance with the supplied streams configuration and
+	 * clean up configuration.
+	 * @param streamsConfig the streams configuration.
+	 * @param cleanupConfig the cleanup configuration.
+	 * @since 2.2
+	 */
+	public StreamsBuilderFactoryBean(Properties streamsConfig, CleanupConfig cleanupConfig) {
+		Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
+		Assert.notNull(cleanupConfig, "'cleanupConfig' must not be null");
+		this.properties = streamsConfig;
+		this.cleanupConfig = cleanupConfig;
+	}
+
 	/**
 	 * Construct an instance with the supplied streams configuration.
 	 * @param streamsConfig the streams configuration.
+	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(Properties)}.
 	 */
+	@Deprecated
 	public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig) {
 		this(streamsConfig, new CleanupConfig());
 	}
 
+	/**
+	 * Construct an instance with the supplied streams configuration.
+	 * @param streamsConfig the streams configuration.
+	 * @since 2.2
+	 */
+	public StreamsBuilderFactoryBean(Properties streamsConfig) {
+		this(streamsConfig, new CleanupConfig());
+	}
+
 	/**
 	 * Construct an instance with the supplied streams configuration and
 	 * clean up configuration.
 	 * @param streamsConfig the streams configuration.
 	 * @param cleanupConfig the cleanup configuration.
 	 * @since 2.1.2.
+	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(Properties, CleanupConfig)}.
 	 */
+	@Deprecated
 	public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig, CleanupConfig cleanupConfig) {
 		Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
 		Assert.notNull(cleanupConfig, "'cleanupConfig' must not be null");
@@ -137,6 +170,7 @@ public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig, CleanupConfi
 	 */
 	public void setStreamsConfig(StreamsConfig streamsConfig) {
 		Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
+		Assert.isNull(this.properties, "Cannot have both streamsConfig and streams configuration properties");
 		this.streamsConfig = streamsConfig;
 	}
 
@@ -145,6 +179,22 @@ public StreamsConfig getStreamsConfig() {
 		return this.streamsConfig;
 	}
 
+	/**
+	 * Set {@link StreamsConfig} on this factory.
+	 * @param streamsConfig the streams configuration.
+	 * @since 2.2
+	 */
+	public void setStreamsConfiguration(Properties streamsConfig) {
+		Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
+		Assert.isNull(this.streamsConfig, "Cannot have both streamsConfig and streams configuration properties");
+		this.properties = streamsConfig;
+	}
+
+	@Nullable
+	public Properties getStreamsConfiguration() {
+		return this.properties;
+	}
+
 	public void setClientSupplier(KafkaClientSupplier clientSupplier) {
 		Assert.notNull(clientSupplier, "'clientSupplier' must not be null");
 		this.clientSupplier = clientSupplier; // NOSONAR (sync)
@@ -191,7 +241,8 @@ public Class<?> getObjectType() {
 	@Override
 	protected StreamsBuilder createInstance() throws Exception {
 		if (this.autoStartup) {
-			Assert.notNull(this.streamsConfig, "'streamsConfig' must not be null");
+			Assert.state(this.streamsConfig != null || this.properties != null,
+					"'streamsConfig' or streams configuration properties must not be null");
 		}
 		return new StreamsBuilder();
 	}
@@ -217,16 +268,23 @@ public void stop(Runnable callback) {
 		}
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
 	public synchronized void start() {
 		if (!this.running) {
 			try {
-				Assert.notNull(this.streamsConfig, "'streamsConfig' must not be null");
+				Assert.state(this.streamsConfig != null || this.properties != null,
+						"'streamsConfig' or streams configuration properties must not be null");
 				Topology topology = getObject().build();
 				if (logger.isDebugEnabled()) {
 					logger.debug(topology.describe());
 				}
-				this.kafkaStreams = new KafkaStreams(topology, this.streamsConfig, this.clientSupplier);
+				if (this.properties != null) {
+					this.kafkaStreams = new KafkaStreams(topology, this.properties, this.clientSupplier);
+				}
+				else {
+					this.kafkaStreams = new KafkaStreams(topology, this.streamsConfig, this.clientSupplier);
+				}
 				this.kafkaStreams.setStateListener(this.stateListener);
 				this.kafkaStreams.setGlobalStateRestoreListener(this.stateRestoreListener);
 				this.kafkaStreams.setUncaughtExceptionHandler(this.uncaughtExceptionHandler);

spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java

@@ -95,12 +95,12 @@ public enum AckMode {
 	/**
 	 * The default {@link #setPollTimeout(long) pollTimeout} (ms).
 	 */
-	public static final long DEFAULT_POLL_TIMEOUT = 1000L;
+	public static final long DEFAULT_POLL_TIMEOUT = 5_000L;
 
 	/**
 	 * The default {@link #setShutdownTimeout(long) shutDownTimeout} (ms).
 	 */
-	public static final int DEFAULT_SHUTDOWN_TIMEOUT = 10000;
+	public static final long DEFAULT_SHUTDOWN_TIMEOUT = 10_000L;
 
 	/**
 	 * The default {@link #setMonitorInterval(int) monitorInterval} (s).

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -17,6 +17,7 @@
 package org.springframework.kafka.listener;
 
 import java.lang.reflect.Type;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -704,7 +705,8 @@ public void run() {
 						}
 						publishConsumerPausedEvent(this.consumer.assignment());
 					}
-					ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.getPollTimeout());
+					ConsumerRecords<K, V> records = this.consumer
+							.poll(Duration.ofMillis(this.containerProperties.getPollTimeout()));
 					this.lastPoll = System.currentTimeMillis();
 					if (this.consumerPaused && !isPaused()) {
 						if (this.logger.isDebugEnabled()) {