diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index e331e7e9a..924afa4fd 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/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) 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); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java index 0d94ec0e7..69b67c5eb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java +++ b/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 store = kafkaStreams .store("foo-WordCounts", QueryableStoreTypes.windowStore()); diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java index b68749a7d..d4a1e88c9 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java +++ b/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(); diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/MultiProcessorsWithSameNameTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/MultiProcessorsWithSameNameTests.java new file mode 100644 index 000000000..51ab319d6 --- /dev/null +++ b/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 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 input) { + } + } + } + + interface KafkaStreamsProcessorX { + + @Input("input-1") + KStream input1(); + + @Input("input-2") + KStream input2(); + + } +}