mindol1004/spring-cloud-stream-binder-kafka
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix concurrency issues with Kafka Streams binder

Browse Source 
See this commit on the master branch for details:
https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/commit/1cc50c1a
This commit is contained in:
Soby Chacko
2020-02-13 15:00:35 -05:00
parent fc1e4602fb
commit 32adf574ac
2 changed files with 4 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -466,8 +466,8 @@ public class KafkaStreamsFunctionProcessor implements ApplicationContextAware {
}
int concurrency = this.bindingServiceProperties.getConsumerProperties(inboundName).getConcurrency();
// override concurrency if set at the individual binding level.
if (concurrency > 1) {
// override concurrency (set potentially on the binding).
if (concurrency >= 1) {
streamConfigGlobalProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, concurrency);
}

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

@@ -539,8 +539,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator
int concurrency = this.bindingServiceProperties.getConsumerProperties(inboundName)
.getConcurrency();
// override concurrency if set at the individual binding level.
if (concurrency > 1) {
// override concurrency (set potentially on the binding).
if (concurrency >= 1) {
streamConfigGlobalProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,
concurrency);
}