From fa0d6fe60e2123557f41f7363bac6acf5820c9d6 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 21 May 2020 16:56:19 -0400 Subject: [PATCH] GH-899: num.stream.threads not taking effect Binder is overriding the num.stream.threads property specified through the binder configuration. Fixing this issue. Adding tests to verify. Doc changes Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/899 --- docs/src/main/asciidoc/kafka-streams.adoc | 3 + .../AbstractKafkaStreamsBinderProcessor.java | 115 ++++++++++++++---- .../MultipleFunctionsInSameAppTests.java | 32 ++++- ...treamsBinderWordCountIntegrationTests.java | 11 +- 4 files changed, 126 insertions(+), 35 deletions(-) diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index a2500f115..0913adbca 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -1628,3 +1628,6 @@ You can also use the `concurrency` property that core Spring Cloud Stream provid When using this, you need to use it on the consumer. When you have more than one input bindings either in a function or `StreamListener`, set this on the first input binding. For e.g. when setting `spring.cloud.stream.bindings.process-in-0.consumer.concurrency`, it will be translated as `num.stream.threads` by the binder. +If you have multiple processors and one processor defines binding level concurrency, but not the others, those ones with no binding level concurrency will default back to the binder wide property specified through +`spring.cloud.stream.binder.configuration.num.stream.threads`. +If this binder configuration is not available, then the application will use the default set by Kafka Streams. \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index 417259246..94dc34cc2 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -45,6 +45,12 @@ import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.boot.context.properties.bind.BindContext; +import org.springframework.boot.context.properties.bind.BindHandler; +import org.springframework.boot.context.properties.bind.Bindable; +import org.springframework.boot.context.properties.bind.PropertySourcesPlaceholdersResolver; +import org.springframework.boot.context.properties.source.ConfigurationPropertyName; +import org.springframework.boot.context.properties.source.ConfigurationPropertySources; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; @@ -57,6 +63,7 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.ResolvableType; import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.core.env.MutablePropertySources; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.config.StreamsBuilderFactoryBean; import org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer; @@ -89,6 +96,8 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application protected ConfigurableApplicationContext applicationContext; + private Object concurrencyAtTheGlobal; + public AbstractKafkaStreamsBinderProcessor(BindingServiceProperties bindingServiceProperties, KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, @@ -172,19 +181,24 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application Map streamConfigGlobalProperties = applicationContext .getBean("streamConfigGlobalProperties", Map.class); + if (this.concurrencyAtTheGlobal == null) { + this.concurrencyAtTheGlobal = streamConfigGlobalProperties.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + } if (kafkaStreamsBinderConfigurationProperties != null) { final Map functionConfigMap = kafkaStreamsBinderConfigurationProperties.getFunctions(); if (!CollectionUtils.isEmpty(functionConfigMap)) { final KafkaStreamsBinderConfigurationProperties.Functions functionConfig = functionConfigMap.get(beanNamePostPrefix); - final Map functionSpecificConfig = functionConfig.getConfiguration(); - if (!CollectionUtils.isEmpty(functionSpecificConfig)) { - streamConfigGlobalProperties.putAll(functionSpecificConfig); - } + if (functionConfig != null) { + final Map functionSpecificConfig = functionConfig.getConfiguration(); + if (!CollectionUtils.isEmpty(functionSpecificConfig)) { + streamConfigGlobalProperties.putAll(functionSpecificConfig); + } - String applicationId = functionConfig.getApplicationId(); - if (!StringUtils.isEmpty(applicationId)) { - streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); + String applicationId = functionConfig.getApplicationId(); + if (!StringUtils.isEmpty(applicationId)) { + streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); + } } } } @@ -198,9 +212,8 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application if (StringUtils.isEmpty(connectionString)) { connectionString = (String) propertySources.get(bindingProperties.getBinder() + "-kafkaStreamsBinderEnv").getProperty("spring.cloud.stream.kafka.binder.brokers"); } - else { - streamConfigGlobalProperties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, connectionString); - } + + streamConfigGlobalProperties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, connectionString); String binderProvidedApplicationId = multiBinderKafkaStreamsBinderConfigurationProperties.getApplicationId(); if (StringUtils.hasText(binderProvidedApplicationId)) { @@ -234,7 +247,6 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } } - //this is only used primarily for StreamListener based processors. Although in theory, functions can use it, //it is ideal for functions to use the approach used in the above if statement by using a property like //spring.cloud.stream.kafka.streams.binder.functions.process.configuration.num.threads (assuming that process is the function name). @@ -267,18 +279,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application return generatedApplicationID; }); - int concurrency = this.bindingServiceProperties.getConsumerProperties(inboundName) - .getConcurrency(); - // override concurrency if set at the individual binding level. - // Concurrency will be mapped to num.stream.threads. Since this is going into a global config, - // we are explicitly assigning concurrency left at default of 1 to num.stream.threads. Otherwise, - // a potential previous value might still be used in the case of multiple processors or a processor - // with multiple input bindings with various concurrency values. - // See this GH issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/844 - if (concurrency >= 1) { - streamConfigGlobalProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, - concurrency); - } + handleConcurrency(applicationContext, inboundName, streamConfigGlobalProperties); // Override deserialization exception handlers per binding final DeserializationExceptionHandler deserializationExceptionHandler = @@ -320,6 +321,15 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application extendedConsumerProperties.setApplicationId((String) streamConfigGlobalProperties.get(StreamsConfig.APPLICATION_ID_CONFIG)); //Removing the application ID from global properties so that the next function won't re-use it and cause race conditions. streamConfigGlobalProperties.remove(StreamsConfig.APPLICATION_ID_CONFIG); + // If there was a global concurrency set at the binder, restore it for the next processor. + if (concurrencyAtTheGlobal != null) { + streamConfigGlobalProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, this.concurrencyAtTheGlobal); + } + else { + //on the other hand, if the global concurrency was set only by the current processor (thus not really binder-wide), remove it. + //otherwise, in the case of multiple processors, binder assumes that there is a global setting present. + streamConfigGlobalProperties.remove(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + } final StreamsBuilderFactoryBean streamsBuilderFactoryBeanFromContext = applicationContext.getBean( "&stream-builder-" + beanNamePostPrefix, StreamsBuilderFactoryBean.class); @@ -331,6 +341,65 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application return streamsBuilderFactoryBeanFromContext; } + private void handleConcurrency(ApplicationContext applicationContext, String inboundName, + Map streamConfigGlobalProperties) { + Object concurrencyAtTheGlobal = streamConfigGlobalProperties.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + Integer concAtTheGlobal = null; + if (concurrencyAtTheGlobal instanceof String) { + concAtTheGlobal = Integer.valueOf((String) concurrencyAtTheGlobal); + } + else if (concurrencyAtTheGlobal instanceof Integer) { + concAtTheGlobal = (Integer) concurrencyAtTheGlobal; + } + // This rebinding is necessary to capture the concurrency explicitly set by the application. + // This is added to fix this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/899 + org.springframework.boot.context.properties.bind.Binder explicitConcurrencyResolver = + new org.springframework.boot.context.properties.bind.Binder(ConfigurationPropertySources.get(applicationContext.getEnvironment()), + new PropertySourcesPlaceholdersResolver(applicationContext.getEnvironment()), + IntegrationUtils.getConversionService(this.applicationContext.getBeanFactory()), null); + + boolean[] concurrencyExplicitlyProvided = new boolean[] {false}; + BindHandler handler = new BindHandler() { + + @Override + 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); + } + return result; + } + }; + //Re-bind spring.cloud.stream properties to check if the application explicitly provided concurrency. + try { + explicitConcurrencyResolver.bind("spring.cloud.stream", + Bindable.ofInstance(new BindingServiceProperties()), handler); + } + catch (Exception e) { + // Ignore this exception + } + + int concurrency = this.bindingServiceProperties.getConsumerProperties(inboundName) + .getConcurrency(); + // override concurrency if set at the individual binding level. + // Concurrency will be mapped to num.stream.threads. Since this is going into a global config, + // we are explicitly assigning concurrency left at default of 1 to num.stream.threads. Otherwise, + // a potential previous value might still be used in the case of multiple processors or a processor + // with multiple input bindings with various concurrency values. + // See this GH issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/844 + // if neither of the below conditions are met, num.stream.threads will default to 1 by Kafka Streams. + if (concurrency >= 1 && concurrencyExplicitlyProvided[0]) { + streamConfigGlobalProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, + concurrency); + } + else if (concurrencyAtTheGlobal != null) { + streamConfigGlobalProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, concAtTheGlobal); + } + } + protected Serde getValueSerde(String inboundName, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, ResolvableType resolvableType) { if (bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding()) { BindingProperties bindingProperties = this.bindingServiceProperties diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java index 445072bc9..83571385f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java @@ -84,7 +84,7 @@ 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.function.definition=process;analyze;anotherProcess", "--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", @@ -95,8 +95,10 @@ public class MultipleFunctionsInSameAppTests { "--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.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.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"); @@ -106,23 +108,27 @@ public class MultipleFunctionsInSameAppTests { StreamsBuilderFactoryBean analyzeStreamsBuilderFactoryBean = context .getBean("&stream-builder-analyze", StreamsBuilderFactoryBean.class); + StreamsBuilderFactoryBean anotherProcessStreamsBuilderFactoryBean = context + .getBean("&stream-builder-anotherProcess", StreamsBuilderFactoryBean.class); + final Properties processStreamsConfiguration = processStreamsBuilderFactoryBean.getStreamsConfiguration(); final Properties analyzeStreamsConfiguration = analyzeStreamsBuilderFactoryBean.getStreamsConfiguration(); + final Properties anotherProcessStreamsConfiguration = anotherProcessStreamsBuilderFactoryBean.getStreamsConfiguration(); assertThat(processStreamsConfiguration.getProperty("client.id")).isEqualTo("process-client"); assertThat(analyzeStreamsConfiguration.getProperty("client.id")).isEqualTo("analyze-client"); - Integer concurrency = (Integer) processStreamsBuilderFactoryBean.getStreamsConfiguration() - .get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + Integer concurrency = (Integer) processStreamsConfiguration.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); assertThat(concurrency).isEqualTo(2); - concurrency = (Integer) analyzeStreamsBuilderFactoryBean.getStreamsConfiguration() - .get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + concurrency = (Integer) analyzeStreamsConfiguration.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); assertThat(concurrency).isEqualTo(1); + concurrency = (Integer) anotherProcessStreamsConfiguration.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + assertThat(concurrency).isEqualTo(3); } } @Test - public void testMultiFunctionsInSameAppWithMultiBinders() throws InterruptedException { + public void testMultiFunctionsInSameAppWithMultiBinders() throws Exception { SpringApplication app = new SpringApplication(MultipleFunctionsInSameApp.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -141,6 +147,7 @@ public class MultipleFunctionsInSameAppTests { "--spring.cloud.stream.bindings.analyze-in-0.binder=kafka2", "--spring.cloud.stream.bindings.analyze-in-1.destination=electronics", "--spring.cloud.stream.bindings.analyze-in-1.binder=kafka2", + "--spring.cloud.stream.bindings.analyze-in-0.consumer.concurrency=2", "--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", @@ -166,6 +173,12 @@ public class MultipleFunctionsInSameAppTests { assertThat(analyzeStreamsConfiguration.getProperty("application.id")).isEqualTo("my-app-2"); assertThat(processStreamsConfiguration.getProperty("client.id")).isEqualTo("process-client"); assertThat(analyzeStreamsConfiguration.getProperty("client.id")).isEqualTo("analyze-client"); + + Integer concurrency = (Integer) analyzeStreamsConfiguration.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + assertThat(concurrency).isEqualTo(2); + + concurrency = (Integer) processStreamsConfiguration.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); + assertThat(concurrency).isNull(); //thus default to 1 by Kafka Streams. } } @@ -207,5 +220,12 @@ public class MultipleFunctionsInSameAppTests { electronics.foreach((s, p) -> countDownLatch.countDown()); }; } + + @Bean + public java.util.function.Consumer> anotherProcess() { + return c -> { + + }; + } } } 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 7992aab59..ecf010ef9 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 @@ -20,6 +20,7 @@ import java.time.Duration; import java.util.Arrays; import java.util.Date; import java.util.Map; +import java.util.Properties; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; @@ -30,9 +31,9 @@ import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StoreQueryParameters; import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; -import org.apache.kafka.streams.kstream.Serialized; import org.apache.kafka.streams.kstream.TimeWindows; import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.ReadOnlyWindowStore; @@ -152,12 +153,10 @@ public class KafkaStreamsBinderWordCountIntegrationTests { .store(StoreQueryParameters.fromNameAndType("foo-WordCounts", QueryableStoreTypes.windowStore())); assertThat(store).isNotNull(); - Map streamConfigGlobalProperties = context - .getBean("streamConfigGlobalProperties", Map.class); - // Ensure that concurrency settings are mapped to number of stream task // threads in Kafka Streams. - final Integer concurrency = (Integer) streamConfigGlobalProperties + final Properties streamsConfiguration = streamsBuilderFactoryBean.getStreamsConfiguration(); + final Integer concurrency = (Integer) streamsConfiguration .get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); assertThat(concurrency).isEqualTo(2); @@ -220,7 +219,7 @@ public class KafkaStreamsBinderWordCountIntegrationTests { .flatMapValues( value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .map((key, value) -> new KeyValue<>(value, value)) - .groupByKey(Serialized.with(Serdes.String(), Serdes.String())) + .groupByKey(Grouped.with(Serdes.String(), Serdes.String())) .windowedBy(TimeWindows.of(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts")) .toStream() .map((key, value) -> new KeyValue<>(null,