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

GH-899: num.stream.threads not taking effect

Browse Source 
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
This commit is contained in:
Soby Chacko
2020-05-21 16:56:19 -04:00
parent 6578a996cb
commit fa0d6fe60e
4 changed files with 126 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
docs/src/main/asciidoc/kafka-streams.adoc
View File

@@ -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.

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

@@ -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<String, Object> streamConfigGlobalProperties = applicationContext
.getBean("streamConfigGlobalProperties", Map.class);
if (this.concurrencyAtTheGlobal == null) {
this.concurrencyAtTheGlobal = streamConfigGlobalProperties.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG);
}
if (kafkaStreamsBinderConfigurationProperties != null) {
final Map<String, KafkaStreamsBinderConfigurationProperties.Functions> functionConfigMap = kafkaStreamsBinderConfigurationProperties.getFunctions();
if (!CollectionUtils.isEmpty(functionConfigMap)) {
final KafkaStreamsBinderConfigurationProperties.Functions functionConfig = functionConfigMap.get(beanNamePostPrefix);
final Map<String, String> functionSpecificConfig = functionConfig.getConfiguration();
if (!CollectionUtils.isEmpty(functionSpecificConfig)) {
streamConfigGlobalProperties.putAll(functionSpecificConfig);
}
if (functionConfig != null) {
final Map<String, String> 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<String, Object> 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

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

@@ -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<KStream<String, String>> anotherProcess() {
return c -> {
};
}
}
}

11
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java
View File

@@ -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,