SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-951 update TopicConfig when diffrent from topic properties

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/951
This commit is contained in:
Heiko Does
2020-08-14 18:54:16 +02:00
committed by Soby Chacko
parent 1e9aa60c4e
commit 859952b32a
2 changed files with 166 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<String> 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<Map<ConfigResource, Config>> topicConfigurationFuture = describeConfigsResult.all();
Map<ConfigResource, Config> topicConfigMap = topicConfigurationFuture
.get(this.operationTimeout, TimeUnit.SECONDS);
Config config = topicConfigMap.get(topicConfigResource);
final List<AlterConfigOp> 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<ConfigResource, Collection<AlterConfigOp>> 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<Map<String, TopicDescription>> 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();
}

123
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderTopicPropertiesUpdateTest.java Normal file
View File

@@ -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<String, Object> 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<Map<ConfigResource, Config>> kafkaFuture = describeConfigsResult.all();
Map<ConfigResource, Config> 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();
}
}