From 9fd16416d68cbcbdfe5b367a223a8b6dc86fd294 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 24 Sep 2021 09:58:20 -0400 Subject: [PATCH] GH-1149: Kafka Streams global config issues When there are multiple functions, streamConfigGlobalProperties are overriden for subsequent functions, after a binding specific config takes effect in a function. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1149 --- .../AbstractKafkaStreamsBinderProcessor.java | 6 +-- .../KafkaStreamsBinderBootstrapTest.java | 42 +++++++++++++++++++ 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index b5f461b70..21480f0f1 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.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. @@ -271,9 +271,9 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application Assert.state(!bindingConfig.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + "use multiple binders instead"); - streamConfigGlobalProperties.putAll(bindingConfig); + // We will only add the per binding configuration to the current streamConfiguration and not the global one. streamConfiguration - .putAll(extendedConsumerProperties.getConfiguration()); + .putAll(bindingConfig); String bindingLevelApplicationId = extendedConsumerProperties.getApplicationId(); // override application.id if set at the individual binding level. diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java index e63db710f..ee572f870 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.binder.kafka.streams.bootstrap; +import java.util.Map; +import java.util.Properties; import java.util.function.Consumer; import org.apache.kafka.common.security.JaasUtils; @@ -31,8 +33,11 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.kafka.config.StreamsBuilderFactoryBean; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import static org.assertj.core.api.AssertionsForClassTypes.assertThat; + /** * @author Soby Chacko */ @@ -96,6 +101,43 @@ public class KafkaStreamsBinderBootstrapTest { applicationContext.close(); } + @Test + @SuppressWarnings("unchecked") + public void testStreamConfigGlobalProperties_GH1149() { + ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder( + SimpleKafkaStreamsApplication.class).web(WebApplicationType.NONE).run( + "--spring.cloud.function.definition=input1;input2;input3", + "--spring.cloud.stream.kafka.streams.bindings.input1-in-0.consumer.application-id" + + "=testKafkaStreamsBinderWithStandardConfigurationCanStart", + "--spring.cloud.stream.kafka.streams.bindings.input2-in-0.consumer.application-id" + + "=testKafkaStreamsBinderWithStandardConfigurationCanStart-foo", + "--spring.cloud.stream.kafka.streams.bindings.input2-in-0.consumer.configuration.spring.json.value.type.method=com.test.MyClass", + "--spring.cloud.stream.kafka.streams.bindings.input3-in-0.consumer.application-id" + + "=testKafkaStreamsBinderWithStandardConfigurationCanStart-foobar", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); + + Map streamConfigGlobalProperties = applicationContext + .getBean("streamConfigGlobalProperties", Map.class); + // Make sure that global stream configs do not contain individual binding config set on second function. + assertThat(streamConfigGlobalProperties.containsKey("spring.json.value.type.method")).isFalse(); + + // Make sure that only input2 function gets the specific binding property set on it. + final StreamsBuilderFactoryBean input1SBFB = applicationContext.getBean("&stream-builder-input1", StreamsBuilderFactoryBean.class); + final Properties streamsConfiguration1 = input1SBFB.getStreamsConfiguration(); + assertThat(streamsConfiguration1.containsKey("spring.json.value.type.method")).isFalse(); + + final StreamsBuilderFactoryBean input2SBFB = applicationContext.getBean("&stream-builder-input2", StreamsBuilderFactoryBean.class); + final Properties streamsConfiguration2 = input2SBFB.getStreamsConfiguration(); + assertThat(streamsConfiguration2.containsKey("spring.json.value.type.method")).isTrue(); + + final StreamsBuilderFactoryBean input3SBFB = applicationContext.getBean("&stream-builder-input3", StreamsBuilderFactoryBean.class); + final Properties streamsConfiguration3 = input3SBFB.getStreamsConfiguration(); + assertThat(streamsConfiguration3.containsKey("spring.json.value.type.method")).isFalse(); + + applicationContext.close(); + } + @SpringBootApplication static class SimpleKafkaStreamsApplication {