Bean name conflicts (Kafka Streams binder)

When two processors with same name are present in the same application,
there is a bean creation conflict. Fixing that issue.

Add test to verify.
Modify existing tests.

Resolves #589

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java

@@ -569,10 +569,11 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator
 						(Class<StreamsBuilderFactoryBean>) streamsBuilder.getClass(),
 						() -> streamsBuilder)
 				.getRawBeanDefinition();
+		final String beanNamePostFix = method.getDeclaringClass().getSimpleName() + "-" + method.getName();
 		((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(
-				"stream-builder-" + method.getName(), streamsBuilderBeanDefinition);
+				"stream-builder-" + beanNamePostFix, streamsBuilderBeanDefinition);
 		StreamsBuilderFactoryBean streamsBuilderX = applicationContext.getBean(
-				"&stream-builder-" + method.getName(), StreamsBuilderFactoryBean.class);
+				"&stream-builder-" + beanNamePostFix, StreamsBuilderFactoryBean.class);
 		this.methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderX);
 	}
 

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java

@@ -155,7 +155,7 @@ public class KafkaStreamsBinderWordCountIntegrationTests {
 			receiveAndValidate(context);
 			// Assertions on StreamBuilderFactoryBean
 			StreamsBuilderFactoryBean streamsBuilderFactoryBean = context
-					.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
+					.getBean("&stream-builder-WordCountProcessorApplication-process", StreamsBuilderFactoryBean.class);
 			KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();
 			ReadOnlyWindowStore<Object, Object> store = kafkaStreams
 					.store("foo-WordCounts", QueryableStoreTypes.windowStore());

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java

@@ -108,7 +108,7 @@ public class KafkastreamsBinderPojoInputStringOutputIntegrationTests {
 			receiveAndValidateFoo(context);
 			// Assertions on StreamBuilderFactoryBean
 			StreamsBuilderFactoryBean streamsBuilderFactoryBean = context
-					.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
+					.getBean("&stream-builder-ProductCountApplication-process", StreamsBuilderFactoryBean.class);
 			CleanupConfig cleanup = TestUtils.getPropertyValue(streamsBuilderFactoryBean,
 					"cleanupConfig", CleanupConfig.class);
 			assertThat(cleanup.cleanupOnStart()).isFalse();

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/MultiProcessorsWithSameNameTests.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2019-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.streams.integration;
+
+import org.apache.kafka.streams.kstream.KStream;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsApplicationSupportProperties;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.kafka.config.StreamsBuilderFactoryBean;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.stereotype.Component;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class MultiProcessorsWithSameNameTests {
+
+	@ClassRule
+	public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true,
+			"counts");
+
+	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule
+			.getEmbeddedKafka();
+
+	@Test
+	public void testBinderStartsSuccessfullyWhenTwoProcessorsWithSameNamesArePresent() {
+		SpringApplication app = new SpringApplication(
+				MultiProcessorsWithSameNameTests.WordCountProcessorApplication.class);
+		app.setWebApplicationType(WebApplicationType.NONE);
+
+		try (ConfigurableApplicationContext context = app.run("--server.port=0",
+				"--spring.jmx.enabled=false",
+				"--spring.cloud.stream.bindings.input.destination=words",
+				"--spring.cloud.stream.bindings.input-2.destination=words",
+				"--spring.cloud.stream.bindings.output.destination=counts",
+				"--spring.cloud.stream.bindings.output.contentType=application/json",
+				"--spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=basic-word-count",
+				"--spring.cloud.stream.kafka.streams.bindings.input-2.consumer.application-id=basic-word-count-1",
+				"--spring.cloud.stream.kafka.streams.binder.brokers="
+						+ embeddedKafka.getBrokersAsString(),
+				"--spring.cloud.stream.kafka.streams.binder.zkNodes="
+						+ embeddedKafka.getZookeeperConnectionString())) {
+			StreamsBuilderFactoryBean streamsBuilderFactoryBean1 = context
+					.getBean("&stream-builder-Foo-process", StreamsBuilderFactoryBean.class);
+			assertThat(streamsBuilderFactoryBean1).isNotNull();
+			StreamsBuilderFactoryBean streamsBuilderFactoryBean2 = context
+					.getBean("&stream-builder-Bar-process", StreamsBuilderFactoryBean.class);
+			assertThat(streamsBuilderFactoryBean2).isNotNull();
+		}
+	}
+
+	@EnableBinding(KafkaStreamsProcessorX.class)
+	@EnableAutoConfiguration
+	@EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class)
+	static class WordCountProcessorApplication {
+
+		@Component
+		static class Foo {
+			@StreamListener
+			public void process(@Input("input-1") KStream<Object, String> input) {
+			}
+		}
+
+		//Second class with a stub processor that has the same name as above ("process")
+		@Component
+		static class Bar {
+			@StreamListener
+			public void process(@Input("input-2") KStream<Object, String> input) {
+			}
+		}
+	}
+
+	interface KafkaStreamsProcessorX {
+
+		@Input("input-1")
+		KStream<?, ?> input1();
+
+		@Input("input-2")
+		KStream<?, ?> input2();
+
+	}
+}