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 d7c5273c7..9da4bece0 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 @@ -25,14 +25,20 @@ import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.AlterConfigOp; +import org.apache.kafka.clients.admin.AlterConfigsResult; +import org.apache.kafka.clients.admin.Config; +import org.apache.kafka.clients.admin.ConfigEntry; import org.apache.kafka.clients.admin.CreatePartitionsResult; import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.DescribeConfigsResult; import org.apache.kafka.clients.admin.DescribeTopicsResult; import org.apache.kafka.clients.admin.ListTopicsResult; import org.apache.kafka.clients.admin.NewPartitions; @@ -40,6 +46,7 @@ import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.admin.TopicDescription; import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.errors.TopicExistsException; import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; @@ -262,7 +269,7 @@ public class KafkaTopicProvisioner implements if (ObjectUtils .isEmpty(adminProps.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG)) || !kafkaConnectionString - .equals(binderProps.getDefaultKafkaConnectionString())) { + .equals(binderProps.getDefaultKafkaConnectionString())) { adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConnectionString); } @@ -367,13 +374,42 @@ public class KafkaTopicProvisioner implements Set names = namesFutures.get(this.operationTimeout, TimeUnit.SECONDS); if (names.contains(topicName)) { + //check if topic.properties are different from Topic Configuration in Kafka + if (this.configurationProperties.isAutoCreateTopics()) { + ConfigResource topicConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName); + DescribeConfigsResult describeConfigsResult = adminClient + .describeConfigs(Collections.singletonList(topicConfigResource)); + KafkaFuture> topicConfigurationFuture = describeConfigsResult.all(); + Map topicConfigMap = topicConfigurationFuture + .get(this.operationTimeout, TimeUnit.SECONDS); + Config config = topicConfigMap.get(topicConfigResource); + final List updatedConfigEntries = topicProperties.getProperties().entrySet().stream() + .filter(propertiesEntry -> { + // Property is new and should be added + if (config.get(propertiesEntry.getKey()) == null) { + return true; + } + else { + // Property changed and should be updated + return !config.get(propertiesEntry.getKey()).value().equals(propertiesEntry.getValue()); + } + + }).map(propertyEntry -> new ConfigEntry(propertyEntry.getKey(), propertyEntry.getValue())) + .map(configEntry -> new AlterConfigOp(configEntry, AlterConfigOp.OpType.SET)) + .collect(Collectors + .toList()); + Map> alterConfigForTopics = new HashMap<>(); + alterConfigForTopics.put(topicConfigResource, updatedConfigEntries); + AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(alterConfigForTopics); + alterConfigsResult.all().get(this.operationTimeout, TimeUnit.SECONDS); + } // only consider minPartitionCount for resizing if autoAddPartitions is true int effectivePartitionCount = this.configurationProperties .isAutoAddPartitions() - ? Math.max( - this.configurationProperties.getMinPartitionCount(), - partitionCount) - : partitionCount; + ? Math.max( + this.configurationProperties.getMinPartitionCount(), + partitionCount) + : partitionCount; DescribeTopicsResult describeTopicsResult = adminClient .describeTopics(Collections.singletonList(topicName)); KafkaFuture> topicDescriptionsFuture = describeTopicsResult @@ -426,7 +462,7 @@ public class KafkaTopicProvisioner implements topicProperties.getReplicationFactor() != null ? topicProperties.getReplicationFactor() : this.configurationProperties - .getReplicationFactor()); + .getReplicationFactor()); } if (topicProperties.getProperties().size() > 0) { newTopic.configs(topicProperties.getProperties()); @@ -494,7 +530,7 @@ public class KafkaTopicProvisioner implements try (AdminClient adminClient = AdminClient .create(this.adminClientProperties)) { final DescribeTopicsResult describeTopicsResult = adminClient - .describeTopics(Collections.singletonList(topicName)); + .describeTopics(Collections.singletonList(topicName)); describeTopicsResult.all().get(); } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderTopicPropertiesUpdateTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderTopicPropertiesUpdateTest.java new file mode 100644 index 000000000..6f264562d --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderTopicPropertiesUpdateTest.java @@ -0,0 +1,123 @@ +/* + * Copyright 2018-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 + * + * https://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.integration; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.Config; +import org.apache.kafka.clients.admin.DescribeConfigsResult; +import org.apache.kafka.common.KafkaFuture; +import org.apache.kafka.common.config.ConfigResource; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Heiko Does + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { + "spring.cloud.stream.kafka.bindings.standard-out.producer.topic.properties.retention.ms=9001", + "spring.cloud.stream.kafka.default.producer.topic.properties.retention.ms=-1", + "spring.cloud.stream.kafka.bindings.standard-in.consumer.topic.properties.retention.ms=9001", + "spring.cloud.stream.kafka.default.consumer.topic.properties.retention.ms=-1" +}) +@DirtiesContext +public class KafkaBinderTopicPropertiesUpdateTest { + + private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; + + @ClassRule + public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true, "standard-in", "standard-out"); + + @BeforeClass + public static void setup() { + System.setProperty(KAFKA_BROKERS_PROPERTY, + kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); + } + + @AfterClass + public static void clean() { + System.clearProperty(KAFKA_BROKERS_PROPERTY); + } + + @Autowired + private ConfigurableApplicationContext context; + + @Test + public void testKafkaBinderUpdateTopicConfiguration() throws Exception { + Map adminClientConfig = new HashMap<>(); + adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); + AdminClient adminClient = AdminClient.create(adminClientConfig); + ConfigResource standardInConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, "standard-in"); + ConfigResource standardOutConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, "standard-out"); + DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays + .asList(standardInConfigResource, standardOutConfigResource)); + KafkaFuture> kafkaFuture = describeConfigsResult.all(); + Map configResourceConfigMap = kafkaFuture.get(3, TimeUnit.SECONDS); + Config standardInTopicConfig = configResourceConfigMap.get(standardInConfigResource); + assertThat(standardInTopicConfig.get("retention.ms").value()).isEqualTo("9001"); + + Config standardOutTopicConfig = configResourceConfigMap.get(standardOutConfigResource); + assertThat(standardOutTopicConfig.get("retention.ms").value()).isEqualTo("9001"); + } + + @EnableBinding(CustomBindingForTopicPropertiesUpdateTesting.class) + @EnableAutoConfiguration + public static class KafkaMetricsTestConfig { + + @StreamListener("standard-in") + @SendTo("standard-out") + public String process(String payload) { + return payload; + } + } + + interface CustomBindingForTopicPropertiesUpdateTesting { + + @Input("standard-in") + SubscribableChannel standardIn(); + + @Output("standard-out") + MessageChannel standardOut(); + } + +}