SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow binding names to be reused in Kafka Streams.

Browse Source 
Allow same binding names to be reused from multiple StreamListener methods in Kafka Streams binder.

Resolves #760
This commit is contained in:
Soby Chacko
2019-10-01 13:58:14 -04:00
parent ac75f8fecf
commit a02308a5a3
4 changed files with 14 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java
View File

@@ -90,8 +90,9 @@ public class GlobalKTableBoundElementFactory
public void wrap(GlobalKTable<Object, Object> delegate) {
Assert.notNull(delegate, "delegate cannot be null");
Assert.isNull(this.delegate, "delegate already set to " + this.delegate);
this.delegate = delegate;
if (this.delegate == null) {
this.delegate = delegate;
}
}
@Override

5
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java
View File

@@ -109,8 +109,9 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KStream> {
public void wrap(KStream<Object, Object> delegate) {
Assert.notNull(delegate, "delegate cannot be null");
Assert.isNull(this.delegate, "delegate already set to " + this.delegate);
this.delegate = delegate;
if (this.delegate == null) {
this.delegate = delegate;
}
}
@Override

5
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java
View File

@@ -86,8 +86,9 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory<KTable> {
public void wrap(KTable<Object, Object> delegate) {
Assert.notNull(delegate, "delegate cannot be null");
Assert.isNull(this.delegate, "delegate already set to " + this.delegate);
this.delegate = delegate;
if (this.delegate == null) {
this.delegate = delegate;
}
}
@Override

15
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/MultiProcessorsWithSameNameTests.java → spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/MultiProcessorsWithSameNameAndBindingTests.java
View File

@@ -34,7 +34,7 @@ import org.springframework.stereotype.Component;
import static org.assertj.core.api.Assertions.assertThat;
public class MultiProcessorsWithSameNameTests {
public class MultiProcessorsWithSameNameAndBindingTests {
@ClassRule
public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true,
@@ -44,19 +44,17 @@ public class MultiProcessorsWithSameNameTests {
.getEmbeddedKafka();
@Test
public void testBinderStartsSuccessfullyWhenTwoProcessorsWithSameNamesArePresent() {
public void testBinderStartsSuccessfullyWhenTwoProcessorsWithSameNamesAndBindingsPresent() {
SpringApplication app = new SpringApplication(
MultiProcessorsWithSameNameTests.WordCountProcessorApplication.class);
MultiProcessorsWithSameNameAndBindingTests.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.input-1.destination=words",
"--spring.cloud.stream.bindings.output.destination=counts",
"--spring.cloud.stream.bindings.output.contentType=application/json",
"--spring.cloud.stream.kafka.streams.bindings.input-1.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())) {
StreamsBuilderFactoryBean streamsBuilderFactoryBean1 = context
@@ -83,7 +81,7 @@ public class MultiProcessorsWithSameNameTests {
@Component
static class Bar {
@StreamListener
public void process(@Input("input-2") KStream<Object, String> input) {
public void process(@Input("input-1") KStream<Object, String> input) {
}
}
}
@@ -93,8 +91,5 @@ public class MultiProcessorsWithSameNameTests {
@Input("input-1")
KStream<?, ?> input1();
@Input("input-2")
KStream<?, ?> input2();
}
}