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

Concurrency property issues in KStream binder

Browse Source 
In Kafka Streams binder, when using a function with camelcase names,
it causes issues for parsing binding level concurrency properties.

Fixing this issue.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2316
This commit is contained in:
Soby Chacko
2022-03-31 16:38:05 -04:00
parent 85ffd4d92d
commit b4c6ec8332
2 changed files with 25 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java
View File

@@ -371,10 +371,10 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
public Object onSuccess(ConfigurationPropertyName name, Bindable<?> target,
BindContext context, Object result) {
if (!concurrencyExplicitlyProvided[0]) {
concurrencyExplicitlyProvided[0] = name.getLastElement(ConfigurationPropertyName.Form.UNIFORM)
.equals("concurrency") &&
ConfigurationPropertyName.of("spring.cloud.stream.bindings." + inboundName + ".consumer").isAncestorOf(name);
concurrencyExplicitlyProvided[0] = name.getLastElement(ConfigurationPropertyName.Form.UNIFORM).equals("concurrency") &&
// name is normalized to contain only uniform elements and thus safe to call toLowerCase here.
ConfigurationPropertyName.of("spring.cloud.stream.bindings." + inboundName.toLowerCase() + ".consumer")
.isAncestorOf(name);
}
return result;
}

42
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/MultipleFunctionsInSameAppTests.java
View File

@@ -86,26 +86,26 @@ public class MultipleFunctionsInSameAppTests {
try (ConfigurableApplicationContext context = app.run(
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.function.definition=process;analyze;anotherProcess;yetAnotherProcess",
"--spring.cloud.stream.bindings.process-in-0.destination=purchases",
"--spring.cloud.stream.bindings.process-out-0.destination=coffee",
"--spring.cloud.stream.bindings.process-out-1.destination=electronics",
"--spring.cloud.stream.function.definition=processItem;analyze;anotherProcess;yetAnotherProcess",
"--spring.cloud.stream.bindings.processItem-in-0.destination=purchases",
"--spring.cloud.stream.bindings.processItem-out-0.destination=coffee",
"--spring.cloud.stream.bindings.processItem-out-1.destination=electronics",
"--spring.cloud.stream.bindings.analyze-in-0.destination=coffee",
"--spring.cloud.stream.bindings.analyze-in-1.destination=electronics",
"--spring.cloud.stream.kafka.streams.binder.functions.analyze.applicationId=analyze-id-0",
"--spring.cloud.stream.kafka.streams.binder.functions.process.applicationId=process-id-0",
"--spring.cloud.stream.kafka.streams.binder.functions.processItem.applicationId=processItem-id-0",
"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
"--spring.cloud.stream.bindings.process-in-0.consumer.concurrency=2",
"--spring.cloud.stream.bindings.processItem-in-0.consumer.concurrency=2",
"--spring.cloud.stream.bindings.analyze-in-0.consumer.concurrency=1",
"--spring.cloud.stream.kafka.streams.binder.configuration.num.stream.threads=3",
"--spring.cloud.stream.kafka.streams.binder.functions.process.configuration.client.id=process-client",
"--spring.cloud.stream.kafka.streams.binder.functions.processItem.configuration.client.id=processItem-client",
"--spring.cloud.stream.kafka.streams.binder.functions.analyze.configuration.client.id=analyze-client",
"--spring.cloud.stream.kafka.streams.binder.functions.anotherProcess.configuration.client.id=anotherProcess-client",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
receiveAndValidate("purchases", "coffee", "electronics");
StreamsBuilderFactoryBean processStreamsBuilderFactoryBean = context
.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
.getBean("&stream-builder-processItem", StreamsBuilderFactoryBean.class);
StreamsBuilderFactoryBean analyzeStreamsBuilderFactoryBean = context
.getBean("&stream-builder-analyze", StreamsBuilderFactoryBean.class);
@@ -117,7 +117,7 @@ public class MultipleFunctionsInSameAppTests {
final Properties analyzeStreamsConfiguration = analyzeStreamsBuilderFactoryBean.getStreamsConfiguration();
final Properties anotherProcessStreamsConfiguration = anotherProcessStreamsBuilderFactoryBean.getStreamsConfiguration();
assertThat(processStreamsConfiguration.getProperty("client.id")).isEqualTo("process-client");
assertThat(processStreamsConfiguration.getProperty("client.id")).isEqualTo("processItem-client");
assertThat(analyzeStreamsConfiguration.getProperty("client.id")).isEqualTo("analyze-client");
Integer concurrency = (Integer) processStreamsConfiguration.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG);
@@ -144,14 +144,14 @@ public class MultipleFunctionsInSameAppTests {
try (ConfigurableApplicationContext context = app.run(
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.function.definition=process;analyze",
"--spring.cloud.stream.bindings.process-in-0.destination=purchases",
"--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.startOffset=latest",
"--spring.cloud.stream.bindings.process-in-0.binder=kafka1",
"--spring.cloud.stream.bindings.process-out-0.destination=coffee",
"--spring.cloud.stream.bindings.process-out-0.binder=kafka1",
"--spring.cloud.stream.bindings.process-out-1.destination=electronics",
"--spring.cloud.stream.bindings.process-out-1.binder=kafka1",
"--spring.cloud.stream.function.definition=processItem;analyze",
"--spring.cloud.stream.bindings.processItem-in-0.destination=purchases",
"--spring.cloud.stream.kafka.streams.bindings.processItem-in-0.consumer.startOffset=latest",
"--spring.cloud.stream.bindings.processItem-in-0.binder=kafka1",
"--spring.cloud.stream.bindings.processItem-out-0.destination=coffee",
"--spring.cloud.stream.bindings.processItem-out-0.binder=kafka1",
"--spring.cloud.stream.bindings.processItem-out-1.destination=electronics",
"--spring.cloud.stream.bindings.processItem-out-1.binder=kafka1",
"--spring.cloud.stream.bindings.analyze-in-0.destination=coffee",
"--spring.cloud.stream.bindings.analyze-in-0.binder=kafka2",
"--spring.cloud.stream.bindings.analyze-in-1.destination=electronics",
@@ -160,7 +160,7 @@ public class MultipleFunctionsInSameAppTests {
"--spring.cloud.stream.binders.kafka1.type=kstream",
"--spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(),
"--spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.applicationId=my-app-1",
"--spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.configuration.client.id=process-client",
"--spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.configuration.client.id=processItem-client",
"--spring.cloud.stream.binders.kafka2.type=kstream",
"--spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(),
"--spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.applicationId=my-app-2",
@@ -170,7 +170,7 @@ public class MultipleFunctionsInSameAppTests {
receiveAndValidate("purchases", "coffee", "electronics");
StreamsBuilderFactoryBean processStreamsBuilderFactoryBean = context
.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
.getBean("&stream-builder-processItem", StreamsBuilderFactoryBean.class);
StreamsBuilderFactoryBean analyzeStreamsBuilderFactoryBean = context
.getBean("&stream-builder-analyze", StreamsBuilderFactoryBean.class);
@@ -180,7 +180,7 @@ public class MultipleFunctionsInSameAppTests {
assertThat(processStreamsConfiguration.getProperty("application.id")).isEqualTo("my-app-1");
assertThat(analyzeStreamsConfiguration.getProperty("application.id")).isEqualTo("my-app-2");
assertThat(processStreamsConfiguration.getProperty("client.id")).isEqualTo("process-client");
assertThat(processStreamsConfiguration.getProperty("client.id")).isEqualTo("processItem-client");
assertThat(analyzeStreamsConfiguration.getProperty("client.id")).isEqualTo("analyze-client");
Integer concurrency = (Integer) analyzeStreamsConfiguration.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG);
@@ -216,7 +216,7 @@ public class MultipleFunctionsInSameAppTests {
public static class MultipleFunctionsInSameApp {
@Bean
public Function<KStream<String, String>, KStream<String, String>[]> process() {
public Function<KStream<String, String>, KStream<String, String>[]> processItem() {
return input -> input.branch(
(s, p) -> p.equalsIgnoreCase("coffee"),
(s, p) -> p.equalsIgnoreCase("electronics"));