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

Fix how stream function properties for bindings are used

Browse Source
This commit is contained in:
Oleg Zhurakousky
2019-10-08 06:04:54 -05:00
parent cc8c645c5a
commit c7fa1ce275
4 changed files with 14 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -260,7 +260,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
}
private List<String> getOutputBindings(String functionName, int outputs) {
List<String> outputBindings = this.streamFunctionProperties.getOutputBindings().get(functionName);
List<String> outputBindings = this.streamFunctionProperties.getOutputBindings(functionName);
List<String> outputBindingNames = new ArrayList<>();
if (!CollectionUtils.isEmpty(outputBindings)) {
outputBindingNames.addAll(outputBindings);

4
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java
View File

@@ -121,7 +121,7 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto
// if the type is array, we need to do a late binding as we don't know the number of
// output bindings at this point in the flow.
List<String> outputBindings = streamFunctionProperties.getOutputBindings().get(this.functionName);
List<String> outputBindings = streamFunctionProperties.getOutputBindings(this.functionName);
String outputBinding = null;
if (!CollectionUtils.isEmpty(outputBindings)) {
@@ -168,7 +168,7 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto
*/
private List<String> buildInputBindings() {
List<String> inputs = new ArrayList<>();
List<String> inputBindings = streamFunctionProperties.getInputBindings().get(this.functionName);
List<String> inputBindings = streamFunctionProperties.getInputBindings(this.functionName);
if (!CollectionUtils.isEmpty(inputBindings)) {
inputs.addAll(inputBindings);
return inputs;

8
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java
View File

@@ -79,13 +79,15 @@ public class KafkaStreamsBinderWordCountBranchesFunctionTests {
app.setWebApplicationType(WebApplicationType.NONE);
ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.function.inputBindings.process=input",
"--spring.cloud.stream.function.outputBindings.process=output1,output2,output3",
"--spring.cloud.stream.function.bindings.process-in-0=input",
"--spring.cloud.stream.bindings.input.destination=words",
"--spring.cloud.stream.function.bindings.process-out-0=output1",
"--spring.cloud.stream.bindings.output1.destination=counts",
"--spring.cloud.stream.function.bindings.process-out-1=output2",
"--spring.cloud.stream.bindings.output2.destination=foo",
"--spring.cloud.stream.function.bindings.process-out-2=output3",
"--spring.cloud.stream.bindings.output3.destination=bar",
"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",

8
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java
View File

@@ -67,12 +67,16 @@ public class StreamToGlobalKTableFunctionTests {
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.function.inputBindings.process=order,customer,product",
"--spring.cloud.stream.function.outputBindings.process=enriched-order",
"--spring.cloud.stream.function.definition=process",
"--spring.cloud.stream.function.bindings.process-in-0=order",
"--spring.cloud.stream.function.bindings.process-in-1=customer",
"--spring.cloud.stream.function.bindings.process-in-2=product",
"--spring.cloud.stream.function.bindings.process-out-0=enriched-order",
"--spring.cloud.stream.bindings.order.destination=orders",
"--spring.cloud.stream.bindings.customer.destination=customers",
"--spring.cloud.stream.bindings.product.destination=products",
"--spring.cloud.stream.bindings.enriched-order.destination=enriched-order",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +