GH-330: Kafka Topic Provisioning Improvements

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/330

- allow override of binder-wide `replicationFactor` for each binding
- allow specific partition/replica configuration
- allow setting `NewTopic.configs()` properties, similar to the consumer and producer
- use a new `AdminClient` for provisioning (and `close()` it) instead of keeping a long-lived connection open.

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaAdminProperties.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.properties;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Properties for configuring topics.
+ *
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class KafkaAdminProperties {
+
+	private Short replicationFactor;
+
+	private Map<Integer, List<Integer>> replicasAssignments = new HashMap<>();
+
+	private Map<String, String> configuration = new HashMap<>();
+
+	public Short getReplicationFactor() {
+		return this.replicationFactor;
+	}
+
+	public void setReplicationFactor(Short replicationFactor) {
+		this.replicationFactor = replicationFactor;
+	}
+
+	public Map<Integer, List<Integer>> getReplicasAssignments() {
+		return this.replicasAssignments;
+	}
+
+	public void setReplicasAssignments(Map<Integer, List<Integer>> replicasAssignments) {
+		this.replicasAssignments = replicasAssignments;
+	}
+
+	public Map<String, String> getConfiguration() {
+		return this.configuration;
+	}
+
+	public void setConfiguration(Map<String, String> configuration) {
+		this.configuration = configuration;
+	}
+
+}

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java

@@ -87,7 +87,7 @@ public class KafkaBinderConfigurationProperties {
 
 	private String requiredAcks = "1";
 
-	private int replicationFactor = 1;
+	private short replicationFactor = 1;
 
 	private int fetchSize = 1024 * 1024;
 
@@ -345,11 +345,11 @@ public class KafkaBinderConfigurationProperties {
 		this.requiredAcks = requiredAcks;
 	}
 
-	public int getReplicationFactor() {
+	public short getReplicationFactor() {
 		return this.replicationFactor;
 	}
 
-	public void setReplicationFactor(int replicationFactor) {
+	public void setReplicationFactor(short replicationFactor) {
 		this.replicationFactor = replicationFactor;
 	}
 

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -81,6 +81,8 @@ public class KafkaConsumerProperties {
 
 	private Map<String, String> configuration = new HashMap<>();
 
+	private KafkaAdminProperties admin = new KafkaAdminProperties();
+
 	public boolean isAutoCommitOffset() {
 		return this.autoCommitOffset;
 	}
@@ -204,4 +206,12 @@ public class KafkaConsumerProperties {
 		this.idleEventInterval = idleEventInterval;
 	}
 
+	public KafkaAdminProperties getAdmin() {
+		return this.admin;
+	}
+
+	public void setAdmin(KafkaAdminProperties admin) {
+		this.admin = admin;
+	}
+
 }

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java

@@ -44,6 +44,8 @@ public class KafkaProducerProperties {
 
 	private Map<String, String> configuration = new HashMap<>();
 
+	private KafkaAdminProperties admin = new KafkaAdminProperties();
+
 	public int getBufferSize() {
 		return this.bufferSize;
 	}
@@ -101,6 +103,15 @@ public class KafkaProducerProperties {
 		this.configuration = configuration;
 	}
 
+	public KafkaAdminProperties getAdmin() {
+		return this.admin;
+	}
+
+	public void setAdmin(KafkaAdminProperties admin) {
+		this.admin = admin;
+	}
+
+
 	public enum CompressionType {
 		none,
 		gzip,

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.provisioning;
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Callable;
@@ -44,6 +45,7 @@ import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.cloud.stream.binder.BinderException;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaAdminProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
@@ -77,19 +79,18 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 
 	private final KafkaBinderConfigurationProperties configurationProperties;
 
-	private final AdminClient adminClient;
+	private final int operationTimeout = DEFAULT_OPERATION_TIMEOUT;
+
+	private final Map<String, Object> adminClientProperties;
 
 	private RetryOperations metadataRetryOperations;
 
-	private final int operationTimeout = DEFAULT_OPERATION_TIMEOUT;
-
 	public KafkaTopicProvisioner(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties,
 								KafkaProperties kafkaProperties) {
 		Assert.isTrue(kafkaProperties != null, "KafkaProperties cannot be null");
-		Map<String, Object> adminClientProperties = kafkaProperties.buildAdminProperties();
+		this.adminClientProperties = kafkaProperties.buildAdminProperties();
 		this.configurationProperties = kafkaBinderConfigurationProperties;
 		normalalizeBootPropsWithBinder(adminClientProperties, kafkaProperties, kafkaBinderConfigurationProperties);
-		this.adminClient = AdminClient.create(adminClientProperties);
 	}
 
 	/**
@@ -118,33 +119,39 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	}
 
 	@Override
-	public ProducerDestination provisionProducerDestination(final String name, ExtendedProducerProperties<KafkaProducerProperties> properties) {
+	public ProducerDestination provisionProducerDestination(final String name,
+			ExtendedProducerProperties<KafkaProducerProperties> properties) {
+
 		if (this.logger.isInfoEnabled()) {
 			this.logger.info("Using kafka topic for outbound: " + name);
 		}
 		KafkaTopicUtils.validateTopicName(name);
-		createTopic(name, properties.getPartitionCount(), false);
-		if (this.configurationProperties.isAutoCreateTopics() && adminClient != null) {
-			DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(name));
-			KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult.all();
+		try (AdminClient adminClient = AdminClient.create(this.adminClientProperties)) {
+			createTopic(adminClient, name, properties.getPartitionCount(), false, properties.getExtension().getAdmin());
+			if (this.configurationProperties.isAutoCreateTopics()) {
+				DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(name));
+				KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult.all();
 
-			try {
-				Map<String, TopicDescription> topicDescriptions = all.get(operationTimeout, TimeUnit.SECONDS);
-				TopicDescription topicDescription = topicDescriptions.get(name);
-				int partitions = topicDescription.partitions().size();
-				return new KafkaProducerDestination(name, partitions);
+				try {
+					Map<String, TopicDescription> topicDescriptions = all.get(this.operationTimeout, TimeUnit.SECONDS);
+					TopicDescription topicDescription = topicDescriptions.get(name);
+					int partitions = topicDescription.partitions().size();
+					return new KafkaProducerDestination(name, partitions);
+				}
+				catch (Exception e) {
+					throw new ProvisioningException("Problems encountered with partitions finding", e);
+				}
 			}
-			catch (Exception e) {
-				throw new ProvisioningException("Problems encountered with partitions finding", e);
+			else {
+				return new KafkaProducerDestination(name);
 			}
 		}
-		else {
-			return new KafkaProducerDestination(name);
-		}
 	}
 
 	@Override
-	public ConsumerDestination provisionConsumerDestination(final String name, final String group, ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
+	public ConsumerDestination provisionConsumerDestination(final String name, final String group,
+			ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
+
 		KafkaTopicUtils.validateTopicName(name);
 		boolean anonymous = !StringUtils.hasText(group);
 		Assert.isTrue(!anonymous || !properties.getExtension().isEnableDlq(),
@@ -153,27 +160,35 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			throw new IllegalArgumentException("Instance count cannot be zero");
 		}
 		int partitionCount = properties.getInstanceCount() * properties.getConcurrency();
-		createTopic(name, partitionCount, properties.getExtension().isAutoRebalanceEnabled());
-		if (this.configurationProperties.isAutoCreateTopics() && adminClient != null) {
-			DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(name));
-			KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult.all();
-			try {
-				Map<String, TopicDescription> topicDescriptions = all.get(operationTimeout, TimeUnit.SECONDS);
-				TopicDescription topicDescription = topicDescriptions.get(name);
-				int partitions = topicDescription.partitions().size();
-				ConsumerDestination dlqTopic = createDlqIfNeedBe(name, group, properties, anonymous, partitions);
-				if (dlqTopic != null) {
-					return dlqTopic;
+		try (AdminClient adminClient = createAdminClient()) {
+			createTopic(adminClient, name, partitionCount, properties.getExtension().isAutoRebalanceEnabled(),
+					properties.getExtension().getAdmin());
+			if (this.configurationProperties.isAutoCreateTopics()) {
+				DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(name));
+				KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult.all();
+				try {
+					Map<String, TopicDescription> topicDescriptions = all.get(operationTimeout, TimeUnit.SECONDS);
+					TopicDescription topicDescription = topicDescriptions.get(name);
+					int partitions = topicDescription.partitions().size();
+					ConsumerDestination dlqTopic = createDlqIfNeedBe(adminClient, name, group, properties, anonymous,
+							partitions);
+					if (dlqTopic != null) {
+						return dlqTopic;
+					}
+					return new KafkaConsumerDestination(name, partitions);
+				}
+				catch (Exception e) {
+					throw new ProvisioningException("provisioning exception", e);
 				}
-				return new KafkaConsumerDestination(name, partitions);
-			}
-			catch (Exception e) {
-				throw new ProvisioningException("provisioning exception", e);
 			}
 		}
 		return new KafkaConsumerDestination(name);
 	}
 
+	AdminClient createAdminClient() {
+		return AdminClient.create(this.adminClientProperties);
+	}
+
 	/**
 	 * In general, binder properties supersede boot kafka properties.
 	 * The one exception is the bootstrap servers. In that case, we should only override
@@ -209,14 +224,15 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		});
 	}
 
-	private ConsumerDestination createDlqIfNeedBe(String name, String group,
+	private ConsumerDestination createDlqIfNeedBe(AdminClient adminClient, String name, String group,
 												ExtendedConsumerProperties<KafkaConsumerProperties> properties,
 												boolean anonymous, int partitions) {
 		if (properties.getExtension().isEnableDlq() && !anonymous) {
 			String dlqTopic = StringUtils.hasText(properties.getExtension().getDlqName()) ?
 					properties.getExtension().getDlqName() : "error." + name + "." + group;
 			try {
-				createTopicAndPartitions(dlqTopic, partitions, properties.getExtension().isAutoRebalanceEnabled());
+				createTopicAndPartitions(adminClient, dlqTopic, partitions,
+						properties.getExtension().isAutoRebalanceEnabled(), properties.getExtension().getAdmin());
 			}
 			catch (Throwable throwable) {
 				if (throwable instanceof Error) {
@@ -231,9 +247,10 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		return null;
 	}
 
-	private void createTopic(String name, int partitionCount, boolean tolerateLowerPartitionsOnBroker) {
+	private void createTopic(AdminClient adminClient, String name, int partitionCount, boolean tolerateLowerPartitionsOnBroker,
+			KafkaAdminProperties properties) {
 		try {
-			createTopicIfNecessary(name, partitionCount, tolerateLowerPartitionsOnBroker);
+			createTopicIfNecessary(adminClient, name, partitionCount, tolerateLowerPartitionsOnBroker, properties);
 		}
 		catch (Throwable throwable) {
 			if (throwable instanceof Error) {
@@ -245,16 +262,14 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		}
 	}
 
-	private void createTopicIfNecessary(final String topicName, final int partitionCount,
-										boolean tolerateLowerPartitionsOnBroker) throws Throwable {
-		if (this.configurationProperties.isAutoCreateTopics() && adminClient != null) {
-			createTopicAndPartitions(topicName, partitionCount, tolerateLowerPartitionsOnBroker);
+	private void createTopicIfNecessary(AdminClient adminClient, final String topicName, final int partitionCount,
+			boolean tolerateLowerPartitionsOnBroker, KafkaAdminProperties properties) throws Throwable {
+
+		if (this.configurationProperties.isAutoCreateTopics()) {
+			createTopicAndPartitions(adminClient, topicName, partitionCount, tolerateLowerPartitionsOnBroker,
+					properties);
 		}
-		else if (this.configurationProperties.isAutoCreateTopics() && adminClient == null) {
-			this.logger.warn("Auto creation of topics is enabled, but Kafka AdminUtils class is not present on the classpath. " +
-					"No topic will be created by the binder");
-		}
-		else if (!this.configurationProperties.isAutoCreateTopics()) {
+		else {
 			this.logger.info("Auto creation of topics is disabled.");
 		}
 	}
@@ -262,9 +277,12 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	/**
 	 * Creates a Kafka topic if needed, or try to increase its partition count to the
 	 * desired number.
+	 * @param adminClient
+	 * @param adminProperties
 	 */
-	private void createTopicAndPartitions(final String topicName, final int partitionCount,
-										boolean tolerateLowerPartitionsOnBroker) throws Throwable {
+	private void createTopicAndPartitions(AdminClient adminClient, final String topicName, final int partitionCount,
+			boolean tolerateLowerPartitionsOnBroker, KafkaAdminProperties adminProperties) throws Throwable {
+
 		ListTopicsResult listTopicsResult = adminClient.listTopics();
 		KafkaFuture<Set<String>> namesFutures = listTopicsResult.names();
 
@@ -298,14 +316,26 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 				}
 			}
 		}
-		else if (!names.contains(topicName)) {
+		else {
 			// always consider minPartitionCount for topic creation
 			final int effectivePartitionCount = Math.max(this.configurationProperties.getMinPartitionCount(),
 					partitionCount);
 			this.metadataRetryOperations.execute(context -> {
 
-				NewTopic newTopic = new NewTopic(topicName, effectivePartitionCount,
-						(short) configurationProperties.getReplicationFactor());
+				NewTopic newTopic;
+				Map<Integer, List<Integer>> replicasAssignments = adminProperties.getReplicasAssignments();
+				if (replicasAssignments != null &&  replicasAssignments.size() > 0) {
+					newTopic = new NewTopic(topicName, adminProperties.getReplicasAssignments());
+				}
+				else {
+					newTopic = new NewTopic(topicName, effectivePartitionCount,
+							adminProperties.getReplicationFactor() != null
+									? adminProperties.getReplicationFactor()
+									: configurationProperties.getReplicationFactor());
+				}
+				if (adminProperties.getConfiguration().size() > 0) {
+					newTopic.configs(adminProperties.getConfiguration());
+				}
 				CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(newTopic));
 				try {
 					createTopicsResult.all().get(operationTimeout, TimeUnit.SECONDS);
@@ -318,6 +348,10 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 								logger.warn("Attempt to create topic: " + topicName + ". Topic already exists.");
 							}
 						}
+						else {
+							logger.error("Failed to create topics", e.getCause());
+							throw e.getCause();
+						}
 					}
 					else {
 						logger.error("Failed to create topics", e.getCause());

spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisionerTests.java

@@ -55,10 +55,11 @@ public class KafkaTopicProvisionerTests {
 		binderConfig.getConfiguration().put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, ts.getFile().getAbsolutePath());
 		binderConfig.setBrokers("localhost:9092");
 		KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderConfig, bootConfig);
-		AdminClient adminClient = KafkaTestUtils.getPropertyValue(provisioner, "adminClient", AdminClient.class);
+		AdminClient adminClient = provisioner.createAdminClient();
 		assertThat(KafkaTestUtils.getPropertyValue(adminClient, "client.selector.channelBuilder")).isInstanceOf(SslChannelBuilder.class);
 		Map configs = KafkaTestUtils.getPropertyValue(adminClient, "client.selector.channelBuilder.configs", Map.class);
 		assertThat(((List) configs.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)).get(0)).isEqualTo("localhost:1234");
+		adminClient.close();
 	}
 
 	@SuppressWarnings("rawtypes")
@@ -73,13 +74,13 @@ public class KafkaTopicProvisionerTests {
 		binderConfig.getConfiguration().put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, ts.getFile().getAbsolutePath());
 		binderConfig.setBrokers("localhost:1234");
 		KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderConfig, bootConfig);
-		AdminClient adminClient = KafkaTestUtils.getPropertyValue(provisioner, "adminClient", AdminClient.class);
+		AdminClient adminClient = provisioner.createAdminClient();
 		assertThat(KafkaTestUtils.getPropertyValue(adminClient, "client.selector.channelBuilder")).isInstanceOf(SslChannelBuilder.class);
 		Map configs = KafkaTestUtils.getPropertyValue(adminClient, "client.selector.channelBuilder.configs", Map.class);
 		assertThat(((List) configs.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)).get(0)).isEqualTo("localhost:1234");
+		adminClient.close();
 	}
 
-	@SuppressWarnings("rawtypes")
 	@Test
 	public void brokersInvalid() throws Exception {
 		KafkaProperties bootConfig = new KafkaProperties();

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import java.util.Arrays;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaAdminProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.cloud.stream.config.BinderFactoryConfiguration;
+import org.springframework.cloud.stream.config.BindingServiceConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = { AdminConfigTests.KafkaBinderConfig.class,
+		KafkaBinderConfiguration.class,
+		BinderFactoryConfiguration.class,
+		BindingServiceConfiguration.class })
+@TestPropertySource(properties = {
+		"spring.cloud.stream.kafka.bindings.input.consumer.admin.replication-factor=2",
+		"spring.cloud.stream.kafka.bindings.input.consumer.admin.replicas-assignments.0=0,1",
+		"spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0" })
+public class AdminConfigTests {
+
+	@Autowired
+	private KafkaMessageChannelBinder binder;
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testProps() {
+		KafkaConsumerProperties consumerProps = this.binder.getExtendedConsumerProperties("input");
+		KafkaAdminProperties admin = consumerProps.getAdmin();
+		assertThat(admin.getReplicationFactor()).isEqualTo((short) 2);
+		assertThat(admin.getReplicasAssignments().get(0)).isEqualTo(Arrays.asList(0, 1));
+		assertThat(admin.getConfiguration().get("message.format.version")).isEqualTo("0.9.0.0");
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class KafkaBinderConfig {
+
+		@Bean
+		public KafkaProperties kafkaProperties() {
+			return new KafkaProperties();
+		}
+
+	}
+
+}