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

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.

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<String, Object> 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 {