GH-389: Topic props: Use .topic instead of .admin

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

GH-389: Deprecated [producer/consumer].admin in favor of topic property

GH-389: Fixed comments as suggest by Gary

Polishing - 2 more deprecation warning suppressions

README.adoc

@@ -162,22 +162,13 @@ The following properties are available for Kafka consumers only and
 must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.consumer.`.
 
 admin.configuration::
-A `Map` of Kafka topic properties used when provisioning topics -- for example, `spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0`
-+
-Default: none.
+Since version 2.1.1, this property is deprecated in favor of `topic.properties`, and support for it will be removed in a future version.
 
 admin.replicas-assignment::
-A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
-Used when provisioning new topics.
-See the `NewTopic` Javadocs in the `kafka-clients` jar.
-+
-Default: none.
+Since version 2.1.1, this property is deprecated in favor of `topic.replicas-assignment`, and support for it will be removed in a future version.
 
 admin.replication-factor::
-The replication factor to use when provisioning topics. Overrides the binder-wide setting.
-Ignored if `replicas-assignments` is present.
-+
-Default: none (the binder-wide default of 1 is used).
+Since version 2.1.1, this property is deprecated in favor of `topic.replication-factor`, and support for it will be removed in a future version.
 
 autoRebalanceEnabled::
 When `true`, topic partitions is automatically rebalanced between the members of a consumer group.
@@ -270,6 +261,21 @@ Note, the time taken to detect new topics that match the pattern is controlled b
 This can be configured using the `configuration` property above.
 +
 Default: `false`
+topic.properties::
+A `Map` of Kafka topic properties used when provisioning new topics -- for example, `spring.cloud.stream.kafka.bindings.input.consumer.topic.properties.message.format.version=0.9.0.0`
++
+Default: none.
+topic.replicas-assignment::
+A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
+Used when provisioning new topics.
+See the `NewTopic` Javadocs in the `kafka-clients` jar.
++
+Default: none.
+topic.replication-factor::
+The replication factor to use when provisioning topics. Overrides the binder-wide setting.
+Ignored if `replicas-assignments` is present.
++
+Default: none (the binder-wide default of 1 is used).
 
 [[kafka-producer-properties]]
 ==== Kafka Producer Properties
@@ -278,22 +284,13 @@ The following properties are available for Kafka producers only and
 must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.producer.`.
 
 admin.configuration::
-A `Map` of Kafka topic properties used when provisioning new topics -- for example, `spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0`
-+
-Default: none.
+Since version 2.1.1, this property is deprecated in favor of `topic.properties`, and support for it will be removed in a future version.
 
 admin.replicas-assignment::
-A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
-Used when provisioning new topics.
-See `NewTopic` javadocs in the `kafka-clients` jar.
-+
-Default: none.
+Since version 2.1.1, this property is deprecated in favor of `topic.replicas-assignment`, and support for it will be removed in a future version.
 
 admin.replication-factor::
-The replication factor to use when provisioning new topics. Overrides the binder-wide setting.
-Ignored if `replicas-assignments` is present.
-+
-Default: none (the binder-wide default of 1 is used).
+Since version 2.1.1, this property is deprecated in favor of `topic.replication-factor`, and support for it will be removed in a future version.
 
 bufferSize::
 Upper limit, in bytes, of how much data the Kafka producer attempts to batch before sending.
@@ -326,6 +323,20 @@ configuration::
 Map with a key/value pair containing generic Kafka producer properties.
 +
 Default: Empty map.
+topic.properties::
+A `Map` of Kafka topic properties used when provisioning new topics -- for example, `spring.cloud.stream.kafka.bindings.output.producer.topic.properties.message.format.version=0.9.0.0`
++
+topic.replicas-assignment::
+A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
+Used when provisioning new topics.
+See the `NewTopic` Javadocs in the `kafka-clients` jar.
++
+Default: none.
+topic.replication-factor::
+The replication factor to use when provisioning topics. Overrides the binder-wide setting.
+Ignored if `replicas-assignments` is present.
++
+Default: none (the binder-wide default of 1 is used).
 
 NOTE: The Kafka binder uses the `partitionCount` setting of the producer as a hint to create a topic with the given partition count (in conjunction with the `minPartitionCount`, the maximum of the two being the value being used).
 Exercise caution when configuring both `minPartitionCount` for a binder and `partitionCount` for an application, as the larger value is used.

docs/src/main/asciidoc/overview.adoc

@@ -146,22 +146,13 @@ The following properties are available for Kafka consumers only and
 must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.consumer.`.
 
 admin.configuration::
-A `Map` of Kafka topic properties used when provisioning topics -- for example, `spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0`
-+
-Default: none.
+Since version 2.1.1, this property is deprecated in favor of `topic.properties`, and support for it will be removed in a future version.
 
 admin.replicas-assignment::
-A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
-Used when provisioning new topics.
-See the `NewTopic` Javadocs in the `kafka-clients` jar.
-+
-Default: none.
+Since version 2.1.1, this property is deprecated in favor of `topic.replicas-assignment`, and support for it will be removed in a future version.
 
 admin.replication-factor::
-The replication factor to use when provisioning topics. Overrides the binder-wide setting.
-Ignored if `replicas-assignments` is present.
-+
-Default: none (the binder-wide default of 1 is used).
+Since version 2.1.1, this property is deprecated in favor of `topic.replication-factor`, and support for it will be removed in a future version.
 
 autoRebalanceEnabled::
 When `true`, topic partitions is automatically rebalanced between the members of a consumer group.
@@ -254,6 +245,21 @@ Note, the time taken to detect new topics that match the pattern is controlled b
 This can be configured using the `configuration` property above.
 +
 Default: `false`
+topic.properties::
+A `Map` of Kafka topic properties used when provisioning new topics -- for example, `spring.cloud.stream.kafka.bindings.input.consumer.topic.properties.message.format.version=0.9.0.0`
++
+Default: none.
+topic.replicas-assignment::
+A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
+Used when provisioning new topics.
+See the `NewTopic` Javadocs in the `kafka-clients` jar.
++
+Default: none.
+topic.replication-factor::
+The replication factor to use when provisioning topics. Overrides the binder-wide setting.
+Ignored if `replicas-assignments` is present.
++
+Default: none (the binder-wide default of 1 is used).
 
 [[kafka-producer-properties]]
 ==== Kafka Producer Properties
@@ -262,22 +268,13 @@ The following properties are available for Kafka producers only and
 must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.producer.`.
 
 admin.configuration::
-A `Map` of Kafka topic properties used when provisioning new topics -- for example, `spring.cloud.stream.kafka.bindings.input.consumer.admin.configuration.message.format.version=0.9.0.0`
-+
-Default: none.
+Since version 2.1.1, this property is deprecated in favor of `topic.properties`, and support for it will be removed in a future version.
 
 admin.replicas-assignment::
-A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
-Used when provisioning new topics.
-See `NewTopic` javadocs in the `kafka-clients` jar.
-+
-Default: none.
+Since version 2.1.1, this property is deprecated in favor of `topic.replicas-assignment`, and support for it will be removed in a future version.
 
 admin.replication-factor::
-The replication factor to use when provisioning new topics. Overrides the binder-wide setting.
-Ignored if `replicas-assignments` is present.
-+
-Default: none (the binder-wide default of 1 is used).
+Since version 2.1.1, this property is deprecated in favor of `topic.replication-factor`, and support for it will be removed in a future version.
 
 bufferSize::
 Upper limit, in bytes, of how much data the Kafka producer attempts to batch before sending.
@@ -310,6 +307,21 @@ configuration::
 Map with a key/value pair containing generic Kafka producer properties.
 +
 Default: Empty map.
+topic.properties::
+A `Map` of Kafka topic properties used when provisioning new topics -- for example, `spring.cloud.stream.kafka.bindings.output.producer.topic.properties.message.format.version=0.9.0.0`
++
+topic.replicas-assignment::
+A Map<Integer, List<Integer>> of replica assignments, with the key being the partition and the value being the assignments.
+Used when provisioning new topics.
+See the `NewTopic` Javadocs in the `kafka-clients` jar.
++
+Default: none.
+topic.replication-factor::
+The replication factor to use when provisioning topics. Overrides the binder-wide setting.
+Ignored if `replicas-assignments` is present.
++
+Default: none (the binder-wide default of 1 is used).
+
 
 NOTE: The Kafka binder uses the `partitionCount` setting of the producer as a hint to create a topic with the given partition count (in conjunction with the `minPartitionCount`, the maximum of the two being the value being used).
 Exercise caution when configuring both `minPartitionCount` for a binder and `partitionCount` for an application, as the larger value is used.

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

@@ -16,8 +16,6 @@
 
 package org.springframework.cloud.stream.binder.kafka.properties;
 
-import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 
 /**
@@ -26,37 +24,17 @@ import java.util.Map;
  * @author Gary Russell
  * @since 2.0
  *
+ * @deprecated in favor of {@link KafkaTopicProperties}
  */
-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;
-	}
+@Deprecated
+public class KafkaAdminProperties extends KafkaTopicProperties {
 
 	public Map<String, String> getConfiguration() {
-		return this.configuration;
+		return getProperties();
 	}
 
 	public void setConfiguration(Map<String, String> configuration) {
-		this.configuration = configuration;
+		setProperties(configuration);
 	}
 
 }

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

@@ -49,6 +49,7 @@ import org.springframework.util.StringUtils;
  * @author Soby Chacko
  * @author Gary Russell
  * @author Rafal Zukowski
+ * @author Aldo Sinanaj
  */
 @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder")
 public class KafkaBinderConfigurationProperties {
@@ -766,14 +767,24 @@ public class KafkaBinderConfigurationProperties {
 			this.kafkaProducerProperties.setConfiguration(configuration);
 		}
 
+		@SuppressWarnings("deprecation")
 		public KafkaAdminProperties getAdmin() {
 			return this.kafkaProducerProperties.getAdmin();
 		}
 
+		@SuppressWarnings("deprecation")
 		public void setAdmin(KafkaAdminProperties admin) {
 			this.kafkaProducerProperties.setAdmin(admin);
 		}
 
+		public KafkaTopicProperties getTopic() {
+			return this.kafkaProducerProperties.getTopic();
+		}
+
+		public void setTopic(KafkaTopicProperties topic) {
+			this.kafkaProducerProperties.setTopic(topic);
+		}
+
 		public KafkaProducerProperties getExtension() {
 			return this.kafkaProducerProperties;
 		}

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.properties;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.springframework.boot.context.properties.DeprecatedConfigurationProperty;
 
 /**
  * Extended consumer properties for Kafka binder.
@@ -27,6 +28,7 @@ import java.util.Map;
  * @author Ilayaperumal Gopinathan
  * @author Soby Chacko
  * @author Gary Russell
+ * @author Aldo Sinanaj
  *
  * <p>
  * Thanks to Laszlo Szabo for providing the initial patch for generic property support.
@@ -112,7 +114,7 @@ public class KafkaConsumerProperties {
 
 	private Map<String, String> configuration = new HashMap<>();
 
-	private KafkaAdminProperties admin = new KafkaAdminProperties();
+	private KafkaTopicProperties topic = new KafkaTopicProperties();
 
 	public boolean isAckEachRecord() {
 		return this.ackEachRecord;
@@ -253,12 +255,35 @@ public class KafkaConsumerProperties {
 		this.destinationIsPattern = destinationIsPattern;
 	}
 
+	/**
+	 * No longer used; get properties such as this via {@link #getTopic()}.
+	 * @return Kafka admin properties
+	 * @deprecated No longer used
+	 */
+	@Deprecated
+	@DeprecatedConfigurationProperty(reason = "Not used since 2.1.1, set properties such as this via 'topic'")
+	@SuppressWarnings("deprecation")
 	public KafkaAdminProperties getAdmin() {
-		return this.admin;
+		// Temporary workaround to copy the topic properties to the admin one.
+		final KafkaAdminProperties kafkaAdminProperties = new KafkaAdminProperties();
+		kafkaAdminProperties.setReplicationFactor(this.topic.getReplicationFactor());
+		kafkaAdminProperties.setReplicasAssignments(this.topic.getReplicasAssignments());
+		kafkaAdminProperties.setConfiguration(this.topic.getProperties());
+		return kafkaAdminProperties;
 	}
 
+	@Deprecated
+	@SuppressWarnings("deprecation")
 	public void setAdmin(KafkaAdminProperties admin) {
-		this.admin = admin;
+		this.topic = admin;
+	}
+
+	public KafkaTopicProperties getTopic() {
+		return this.topic;
+	}
+
+	public void setTopic(KafkaTopicProperties topic) {
+		this.topic = topic;
 	}
 
 }

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

@@ -21,6 +21,7 @@ import java.util.Map;
 
 import javax.validation.constraints.NotNull;
 
+import org.springframework.boot.context.properties.DeprecatedConfigurationProperty;
 import org.springframework.expression.Expression;
 
 /**
@@ -29,6 +30,7 @@ import org.springframework.expression.Expression;
  * @author Marius Bogoevici
  * @author Henryk Konsek
  * @author Gary Russell
+ * @author Aldo Sinanaj
  */
 public class KafkaProducerProperties {
 
@@ -46,7 +48,7 @@ public class KafkaProducerProperties {
 
 	private Map<String, String> configuration = new HashMap<>();
 
-	private KafkaAdminProperties admin = new KafkaAdminProperties();
+	private KafkaTopicProperties topic = new KafkaTopicProperties();
 
 	public int getBufferSize() {
 		return this.bufferSize;
@@ -105,12 +107,35 @@ public class KafkaProducerProperties {
 		this.configuration = configuration;
 	}
 
+	/**
+	 * No longer used; get properties such as this via {@link #getTopic()}.
+	 * @return Kafka admin properties
+	 * @deprecated No longer used
+	 */
+	@Deprecated
+	@DeprecatedConfigurationProperty(reason = "Not used since 2.1.1, set properties such as this via 'topic'")
+	@SuppressWarnings("deprecation")
 	public KafkaAdminProperties getAdmin() {
-		return this.admin;
+		// Temporary workaround to copy the topic properties to the admin one.
+		final KafkaAdminProperties kafkaAdminProperties = new KafkaAdminProperties();
+		kafkaAdminProperties.setReplicationFactor(this.topic.getReplicationFactor());
+		kafkaAdminProperties.setReplicasAssignments(this.topic.getReplicasAssignments());
+		kafkaAdminProperties.setConfiguration(this.topic.getProperties());
+		return kafkaAdminProperties;
 	}
 
+	@Deprecated
+	@SuppressWarnings("deprecation")
 	public void setAdmin(KafkaAdminProperties admin) {
-		this.admin = admin;
+		this.topic = admin;
+	}
+
+	public KafkaTopicProperties getTopic() {
+		return this.topic;
+	}
+
+	public void setTopic(KafkaTopicProperties topic) {
+		this.topic = topic;
 	}
 
 	/**

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2019 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 Aldo Sinanaj
+ * @since 2.2
+ *
+ */
+public class KafkaTopicProperties {
+
+	private Short replicationFactor;
+
+	private Map<Integer, List<Integer>> replicasAssignments = new HashMap<>();
+
+	private Map<String, String> properties = new HashMap<>();
+
+	public Short getReplicationFactor() {
+		return replicationFactor;
+	}
+
+	public void setReplicationFactor(Short replicationFactor) {
+		this.replicationFactor = replicationFactor;
+	}
+
+	public Map<Integer, List<Integer>> getReplicasAssignments() {
+		return replicasAssignments;
+	}
+
+	public void setReplicasAssignments(Map<Integer, List<Integer>> replicasAssignments) {
+		this.replicasAssignments = replicasAssignments;
+	}
+
+	public Map<String, String> getProperties() {
+		return properties;
+	}
+
+	public void setProperties(Map<String, String> properties) {
+		this.properties = properties;
+	}
+
+}

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

@@ -47,10 +47,10 @@ 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;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaTopicProperties;
 import org.springframework.cloud.stream.binder.kafka.utils.KafkaTopicUtils;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
@@ -73,6 +73,7 @@ import org.springframework.util.StringUtils;
  * @author Ilayaperumal Gopinathan
  * @author Simon Flandergan
  * @author Oleg Zhurakousky
+ * @author Aldo Sinanaj
  */
 public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsumerProperties<KafkaConsumerProperties>,
 		ExtendedProducerProperties<KafkaProducerProperties>>, InitializingBean {
@@ -133,7 +134,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		}
 		KafkaTopicUtils.validateTopicName(name);
 		try (AdminClient adminClient = AdminClient.create(this.adminClientProperties)) {
-			createTopic(adminClient, name, properties.getPartitionCount(), false, properties.getExtension().getAdmin());
+			createTopic(adminClient, name, properties.getPartitionCount(), false, properties.getExtension().getTopic());
 			int partitions = 0;
 			if (this.configurationProperties.isAutoCreateTopics()) {
 				DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(name));
@@ -191,7 +192,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		ConsumerDestination consumerDestination = new KafkaConsumerDestination(name);
 		try (AdminClient adminClient = createAdminClient()) {
 			createTopic(adminClient, name, partitionCount, properties.getExtension().isAutoRebalanceEnabled(),
-					properties.getExtension().getAdmin());
+					properties.getExtension().getTopic());
 			if (this.configurationProperties.isAutoCreateTopics()) {
 				DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(name));
 				KafkaFuture<Map<String, TopicDescription>> all = describeTopicsResult.all();
@@ -259,7 +260,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 					properties.getExtension().getDlqName() : "error." + name + "." + group;
 			try {
 				createTopicAndPartitions(adminClient, dlqTopic, partitions,
-						properties.getExtension().isAutoRebalanceEnabled(), properties.getExtension().getAdmin());
+						properties.getExtension().isAutoRebalanceEnabled(), properties.getExtension().getTopic());
 			}
 			catch (Throwable throwable) {
 				if (throwable instanceof Error) {
@@ -275,7 +276,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	}
 
 	private void createTopic(AdminClient adminClient, String name, int partitionCount, boolean tolerateLowerPartitionsOnBroker,
-			KafkaAdminProperties properties) {
+			KafkaTopicProperties properties) {
 		try {
 			createTopicIfNecessary(adminClient, name, partitionCount, tolerateLowerPartitionsOnBroker, properties);
 		}
@@ -292,7 +293,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	}
 
 	private void createTopicIfNecessary(AdminClient adminClient, final String topicName, final int partitionCount,
-			boolean tolerateLowerPartitionsOnBroker, KafkaAdminProperties properties) throws Throwable {
+			boolean tolerateLowerPartitionsOnBroker, KafkaTopicProperties properties) throws Throwable {
 
 		if (this.configurationProperties.isAutoCreateTopics()) {
 			createTopicAndPartitions(adminClient, topicName, partitionCount, tolerateLowerPartitionsOnBroker,
@@ -310,11 +311,11 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	 * @param topicName topic name
 	 * @param partitionCount partition count
 	 * @param tolerateLowerPartitionsOnBroker whether lower partitions count on broker is tolerated ot not
-	 * @param adminProperties kafka admin properties
+	 * @param topicProperties kafka topic properties
 	 * @throws Throwable from topic creation
 	 */
 	private void createTopicAndPartitions(AdminClient adminClient, final String topicName, final int partitionCount,
-			boolean tolerateLowerPartitionsOnBroker, KafkaAdminProperties adminProperties) throws Throwable {
+			boolean tolerateLowerPartitionsOnBroker, KafkaTopicProperties topicProperties) throws Throwable {
 
 		ListTopicsResult listTopicsResult = adminClient.listTopics();
 		KafkaFuture<Set<String>> namesFutures = listTopicsResult.names();
@@ -356,18 +357,18 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			this.metadataRetryOperations.execute((context) -> {
 
 				NewTopic newTopic;
-				Map<Integer, List<Integer>> replicasAssignments = adminProperties.getReplicasAssignments();
+				Map<Integer, List<Integer>> replicasAssignments = topicProperties.getReplicasAssignments();
 				if (replicasAssignments != null &&  replicasAssignments.size() > 0) {
-					newTopic = new NewTopic(topicName, adminProperties.getReplicasAssignments());
+					newTopic = new NewTopic(topicName, topicProperties.getReplicasAssignments());
 				}
 				else {
 					newTopic = new NewTopic(topicName, effectivePartitionCount,
-							adminProperties.getReplicationFactor() != null
-									? adminProperties.getReplicationFactor()
+							topicProperties.getReplicationFactor() != null
+									? topicProperties.getReplicationFactor()
 									: this.configurationProperties.getReplicationFactor());
 				}
-				if (adminProperties.getConfiguration().size() > 0) {
-					newTopic.configs(adminProperties.getConfiguration());
+				if (topicProperties.getProperties().size() > 0) {
+					newTopic.configs(topicProperties.getProperties());
 				}
 				CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singletonList(newTopic));
 				try {

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

@@ -24,8 +24,9 @@ import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 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.binder.kafka.properties.KafkaProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaTopicProperties;
 import org.springframework.cloud.stream.config.BindingServiceConfiguration;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.test.context.TestPropertySource;
@@ -35,6 +36,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author Gary Russell
+ * @author Aldo Sinanaj
  * @since 2.0
  *
  */
@@ -45,6 +47,12 @@ import static org.assertj.core.api.Assertions.assertThat;
 		"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",
+		"spring.cloud.stream.kafka.bindings.secondInput.consumer.topic.replication-factor=3",
+		"spring.cloud.stream.kafka.bindings.secondInput.consumer.topic.replicas-assignments.0=0,1",
+		"spring.cloud.stream.kafka.bindings.secondInput.consumer.topic.properties.message.format.version=0.9.1.0",
+		"spring.cloud.stream.kafka.bindings.output.producer.topic.replication-factor=2",
+		"spring.cloud.stream.kafka.bindings.output.producer.topic.replicas-assignments.0=0,1",
+		"spring.cloud.stream.kafka.bindings.output.producer.topic.properties.message.format.version=0.9.0.0",
 		"spring.main.allow-bean-definition-overriding=true"})
 @EnableIntegration
 public class AdminConfigTests {
@@ -53,11 +61,32 @@ public class AdminConfigTests {
 	private KafkaMessageChannelBinder binder;
 
 	@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");
+	public void testDeprecatedAdminConfigurationToMapTopicProperties() {
+		final KafkaConsumerProperties consumerProps = this.binder.getExtendedConsumerProperties("input");
+		final KafkaTopicProperties kafkaTopicProperties = consumerProps.getTopic();
+
+		assertThat(kafkaTopicProperties.getReplicationFactor()).isEqualTo((short) 2);
+		assertThat(kafkaTopicProperties.getReplicasAssignments().get(0)).isEqualTo(Arrays.asList(0, 1));
+		assertThat(kafkaTopicProperties.getProperties().get("message.format.version")).isEqualTo("0.9.0.0");
+	}
+
+	@Test
+	public void testConsumerTopicProperties() {
+		final KafkaConsumerProperties consumerProperties = this.binder.getExtendedConsumerProperties("secondInput");
+		final KafkaTopicProperties kafkaTopicProperties = consumerProperties.getTopic();
+
+		assertThat(kafkaTopicProperties.getReplicationFactor()).isEqualTo((short) 3);
+		assertThat(kafkaTopicProperties.getReplicasAssignments().get(0)).isEqualTo(Arrays.asList(0, 1));
+		assertThat(kafkaTopicProperties.getProperties().get("message.format.version")).isEqualTo("0.9.1.0");
+	}
+
+	@Test
+	public void testProducerTopicProperties() {
+		final KafkaProducerProperties producerProperties = this.binder.getExtendedProducerProperties("output");
+		final KafkaTopicProperties kafkaTopicProperties = producerProperties.getTopic();
+
+		assertThat(kafkaTopicProperties.getReplicationFactor()).isEqualTo((short) 2);
+		assertThat(kafkaTopicProperties.getReplicasAssignments().get(0)).isEqualTo(Arrays.asList(0, 1));
+		assertThat(kafkaTopicProperties.getProperties().get("message.format.version")).isEqualTo("0.9.0.0");
 	}
 }