diff --git a/README.adoc b/README.adoc index a12cdf7ad..6f9571a7d 100644 --- a/README.adoc +++ b/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..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> 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> 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..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> 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> 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. diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 816047f37..5eef1861e 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/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..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> 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> 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..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> 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> 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. diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaAdminProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaAdminProperties.java index 5cecb2d1b..0cf2e40e1 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaAdminProperties.java +++ b/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> replicasAssignments = new HashMap<>(); - - private Map configuration = new HashMap<>(); - - public Short getReplicationFactor() { - return this.replicationFactor; - } - - public void setReplicationFactor(Short replicationFactor) { - this.replicationFactor = replicationFactor; - } - - public Map> getReplicasAssignments() { - return this.replicasAssignments; - } - - public void setReplicasAssignments(Map> replicasAssignments) { - this.replicasAssignments = replicasAssignments; - } +@Deprecated +public class KafkaAdminProperties extends KafkaTopicProperties { public Map getConfiguration() { - return this.configuration; + return getProperties(); } public void setConfiguration(Map configuration) { - this.configuration = configuration; + setProperties(configuration); } } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index 28cb3b079..cfe2f974b 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/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; } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java index a9b33e13c..b1f0ff010 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java +++ b/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 * *

* Thanks to Laszlo Szabo for providing the initial patch for generic property support. @@ -112,7 +114,7 @@ public class KafkaConsumerProperties { private Map 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; } } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java index c4444b981..c930f6ced 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java +++ b/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 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; } /** diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaTopicProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaTopicProperties.java new file mode 100644 index 000000000..1124ca797 --- /dev/null +++ b/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> replicasAssignments = new HashMap<>(); + + private Map properties = new HashMap<>(); + + public Short getReplicationFactor() { + return replicationFactor; + } + + public void setReplicationFactor(Short replicationFactor) { + this.replicationFactor = replicationFactor; + } + + public Map> getReplicasAssignments() { + return replicasAssignments; + } + + public void setReplicasAssignments(Map> replicasAssignments) { + this.replicasAssignments = replicasAssignments; + } + + public Map getProperties() { + return properties; + } + + public void setProperties(Map properties) { + this.properties = properties; + } + +} diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index e39a6294f..be2ab30e7 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/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, ExtendedProducerProperties>, InitializingBean { @@ -133,7 +134,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> all = describeTopicsResult.all(); @@ -259,7 +260,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> namesFutures = listTopicsResult.names(); @@ -356,18 +357,18 @@ public class KafkaTopicProvisioner implements ProvisioningProvider { NewTopic newTopic; - Map> replicasAssignments = adminProperties.getReplicasAssignments(); + Map> 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 { diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java index 93bd0f1ba..024cadaf4 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java +++ b/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"); } }