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

Consumer/Producer prefix in Kafka Streams binder (#1131)

Browse Source 
* Consumer/Producer prefix in Kafka Streams binder

Kafka Streams allows the applications to separate the consumer and producer
properties using consumer/producer prefixes. Add these prefixes automatically
if they are missing from the application.

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

* Addressing PR review comments
This commit is contained in:
Soby Chacko
2021-08-25 13:57:42 -04:00
committed by GitHub
parent 970bac41bb
commit e500138486
2 changed files with 28 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2020 the original author or authors.
* Copyright 2017-2021 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.
@@ -98,6 +98,9 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
private static final String GLOBALKTABLE_BINDER_TYPE = "globalktable";
private static final String CONSUMER_PROPERTIES_PREFIX = "consumer.";
private static final String PRODUCER_PROPERTIES_PREFIX = "producer.";
@Bean
@ConfigurationProperties(prefix = "spring.cloud.stream.kafka.streams.binder")
public KafkaStreamsBinderConfigurationProperties binderConfigurationProperties(
@@ -266,14 +269,15 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
if (!ObjectUtils.isEmpty(configProperties.getConfiguration())) {
properties.putAll(configProperties.getConfiguration());
}
Map<String, Object> mergedConsumerConfig = configProperties.mergedConsumerConfiguration();
if (!ObjectUtils.isEmpty(mergedConsumerConfig)) {
properties.putAll(mergedConsumerConfig);
}
Map<String, Object> mergedProducerConfig = configProperties.mergedProducerConfiguration();
if (!ObjectUtils.isEmpty(mergedProducerConfig)) {
properties.putAll(mergedProducerConfig);
}
Map<String, Object> mergedConsumerConfig = new HashMap<>(configProperties.mergedConsumerConfiguration());
//Adding consumer. prefix if they are missing (in order to differentiate them from other property categories such as stream, producer etc.)
addPrefix(properties, mergedConsumerConfig, CONSUMER_PROPERTIES_PREFIX);
Map<String, Object> mergedProducerConfig = new HashMap<>(configProperties.mergedProducerConfiguration());
//Adding producer. prefix if they are missing (in order to differentiate them from other property categories such as stream, consumer etc.)
addPrefix(properties, mergedProducerConfig, PRODUCER_PROPERTIES_PREFIX);
if (!properties.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG)) {
properties.put(StreamsConfig.REPLICATION_FACTOR_CONFIG,
(int) configProperties.getReplicationFactor());
@@ -282,6 +286,16 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
Collectors.toMap((e) -> String.valueOf(e.getKey()), Map.Entry::getValue));
}
private void addPrefix(Properties properties, Map<String, Object> mergedConsProdConfig, String prefix) {
Map<String, Object> mergedConfigs = new HashMap<>();
for (String key : mergedConsProdConfig.keySet()) {
mergedConfigs.put(key.startsWith(prefix) ? key : prefix + key, mergedConsProdConfig.get(key));
}
if (!ObjectUtils.isEmpty(mergedConfigs)) {
properties.putAll(mergedConfigs);
}
}
@Bean
public KStreamStreamListenerResultAdapter kstreamStreamListenerResultAdapter() {
return new KStreamStreamListenerResultAdapter();

8
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2019 the original author or authors.
* Copyright 2019-2021 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.
@@ -112,6 +112,7 @@ public class KafkaStreamsBinderWordCountFunctionTests {
"--spring.cloud.stream.kafka.streams.binder.application-id=testKstreamWordCountFunction",
"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
"--spring.cloud.stream.kafka.streams.binder.consumerProperties.request.timeout.ms=29000", //for testing ...binder.consumerProperties
"--spring.cloud.stream.kafka.streams.binder.consumerProperties.consumer.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer",
"--spring.cloud.stream.kafka.streams.binder.producerProperties.max.block.ms=90000", //for testing ...binder.producerProperties
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
@@ -144,8 +145,9 @@ public class KafkaStreamsBinderWordCountFunctionTests {
//verify that ...binder.consumerProperties and ...binder.producerProperties work.
Map<String, Object> streamConfigGlobalProperties = (Map<String, Object>) context.getBean("streamConfigGlobalProperties");
assertThat(streamConfigGlobalProperties.get("request.timeout.ms")).isEqualTo("29000");
assertThat(streamConfigGlobalProperties.get("max.block.ms")).isEqualTo("90000");
assertThat(streamConfigGlobalProperties.get("consumer.request.timeout.ms")).isEqualTo("29000");
assertThat(streamConfigGlobalProperties.get("consumer.value.deserializer")).isEqualTo("org.apache.kafka.common.serialization.StringDeserializer");
assertThat(streamConfigGlobalProperties.get("producer.max.block.ms")).isEqualTo("90000");
InputBindingLifecycle inputBindingLifecycle = context.getBean(InputBindingLifecycle.class);
final Collection<Binding<Object>> inputBindings = (Collection<Binding<Object>>) new DirectFieldAccessor(inputBindingLifecycle)