From e500138486d98b543fb0af61eb07c5e787bf408d Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 25 Aug 2021 13:57:42 -0400 Subject: [PATCH] Consumer/Producer prefix in Kafka Streams binder (#1131) * 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 --- ...StreamsBinderSupportAutoConfiguration.java | 32 +++++++++++++------ ...kaStreamsBinderWordCountFunctionTests.java | 8 +++-- 2 files changed, 28 insertions(+), 12 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 00765c321..17c4d8723 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -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 mergedConsumerConfig = configProperties.mergedConsumerConfiguration(); - if (!ObjectUtils.isEmpty(mergedConsumerConfig)) { - properties.putAll(mergedConsumerConfig); - } - Map mergedProducerConfig = configProperties.mergedProducerConfiguration(); - if (!ObjectUtils.isEmpty(mergedProducerConfig)) { - properties.putAll(mergedProducerConfig); - } + + Map 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 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 mergedConsProdConfig, String prefix) { + Map 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(); diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java index 35cdc6db0..f2eb680cf 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java @@ -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 streamConfigGlobalProperties = (Map) 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> inputBindings = (Collection>) new DirectFieldAccessor(inputBindingLifecycle)