From 64ca773a4f2dcbf6fc2fef667fe841ed200ecb5b Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 9 Aug 2019 13:38:10 -0400 Subject: [PATCH] Kafka Streams application id changes Generate a random application id for Kafka Streams binder if the user doesn't set one for the application. This is useful for development purposes, as it avoids the creation of an explicit application id. For production workloads, it is highly recommended to explicitly provide and application id. The gnerated application id follows a patter where it uses the function bean name followed by a random UUID string which is followed by the literal appplicaitonId. In the case of StreamListener, instead of function bean name, it uses the containing class + StreamListener method name. If the binder generates the application id, that information will be logged on the console at startup. Resolves #718 Resolves #719 --- .../AbstractKafkaStreamsBinderProcessor.java | 15 ++++++++++--- .../binder/kafka/streams/KStreamBinder.java | 4 ++++ .../KafkaStreamsFunctionProcessor.java | 5 +++-- ...StreamListenerSetupMethodOrchestrator.java | 3 +++ .../kafka/streams/KeyValueSerdeResolver.java | 4 ++++ ...kaStreamsBinderWordCountFunctionTests.java | 22 +++++++++++++++++++ 6 files changed, 48 insertions(+), 5 deletions(-) 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 873d354ff..c5a3ce4ef 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 @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.Map; import java.util.Properties; +import java.util.UUID; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -197,13 +198,21 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application streamConfigGlobalProperties .putAll(extendedConsumerProperties.getConfiguration()); - String applicationId = extendedConsumerProperties.getApplicationId(); + String bindingLevelApplicationId = extendedConsumerProperties.getApplicationId(); // override application.id if set at the individual binding level. - if (StringUtils.hasText(applicationId)) { + if (StringUtils.hasText(bindingLevelApplicationId)) { streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, - applicationId); + bindingLevelApplicationId); } + //If the application id is not set by any mechanism, then generate it. + streamConfigGlobalProperties.computeIfAbsent(StreamsConfig.APPLICATION_ID_CONFIG, + k -> { + String generatedApplicationID = beanNamePostPrefix + "-" + UUID.randomUUID().toString() + "-applicationId"; + LOG.info("Generated Kafka Streams Application ID: " + generatedApplicationID); + return generatedApplicationID; + }); + int concurrency = this.bindingServiceProperties.getConsumerProperties(inboundName) .getConcurrency(); // override concurrency if set at the individual binding level. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index caa661567..47a6c105a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java @@ -123,6 +123,8 @@ class KStreamBinder extends extendedProducerProperties); Serde keySerde = this.keyValueSerdeResolver .getOuboundKeySerde(properties.getExtension(), kafkaStreamsBindingInformationCatalogue.getOutboundKStreamResolvable()); + LOG.info("Key Serde used for (outbound) " + name + ": " + keySerde.getClass().getName()); + Serde valueSerde; if (properties.isUseNativeEncoding()) { valueSerde = this.keyValueSerdeResolver.getOutboundValueSerde(properties, @@ -131,6 +133,8 @@ class KStreamBinder extends else { valueSerde = Serdes.ByteArray(); } + LOG.info("Key Serde used for (outbound) " + name + ": " + valueSerde.getClass().getName()); + to(properties.isUseNativeEncoding(), name, outboundBindTarget, (Serde) keySerde, (Serde) valueSerde); return new DefaultBinding<>(name, null, outboundBindTarget, null); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java index 284569738..1ebbf075d 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java @@ -291,7 +291,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro //Retrieve the StreamsConfig created for this method if available. //Otherwise, create the StreamsBuilderFactory and get the underlying config. if (!this.methodStreamsBuilderFactoryBeanMap.containsKey(functionName)) { - StreamsBuilderFactoryBean streamsBuilderFactoryBean = buildStreamsBuilderAndRetrieveConfig("stream-builder-" + functionName, applicationContext, input); + StreamsBuilderFactoryBean streamsBuilderFactoryBean = buildStreamsBuilderAndRetrieveConfig(functionName, applicationContext, input); this.methodStreamsBuilderFactoryBeanMap.put(functionName, streamsBuilderFactoryBean); } try { @@ -303,9 +303,10 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro //get state store spec Serde keySerde = this.keyValueSerdeResolver.getInboundKeySerde(extendedConsumerProperties, stringResolvableTypeMap.get(input)); + LOG.info("Key Serde used for " + input + ": " + keySerde.getClass().getName()); Serde valueSerde = bindingServiceProperties.getConsumerProperties(input).isUseNativeDecoding() ? getValueSerde(input, extendedConsumerProperties, stringResolvableTypeMap.get(input)) : Serdes.ByteArray(); - + LOG.info("Value Serde used for " + input + ": " + valueSerde.getClass().getName()); final Topology.AutoOffsetReset autoOffsetReset = getAutoOffsetReset(input, extendedConsumerProperties); if (parameterType.isAssignableFrom(KStream.class)) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index b9dcf78be..8e8461ad0 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -260,8 +260,11 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr Serde keySerde = this.keyValueSerdeResolver .getInboundKeySerde(extendedConsumerProperties, ResolvableType.forMethodParameter(methodParameter)); + LOG.info("Key Serde used for " + targetReferenceValue + ": " + keySerde.getClass().getName()); + Serde valueSerde = bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding() ? getValueSerde(inboundName, extendedConsumerProperties, ResolvableType.forMethodParameter(methodParameter)) : Serdes.ByteArray(); + LOG.info("Value Serde used for " + targetReferenceValue + ": " + valueSerde.getClass().getName()); Topology.AutoOffsetReset autoOffsetReset = getAutoOffsetReset(inboundName, extendedConsumerProperties); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java index cada2d380..f3532080b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java @@ -18,6 +18,8 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.Map; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.Utils; @@ -60,6 +62,8 @@ import org.springframework.util.StringUtils; */ public class KeyValueSerdeResolver { + private static final Log LOG = LogFactory.getLog(KeyValueSerdeResolver.class); + private final Map streamConfigGlobalProperties; private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties; diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java index e22c0aa2b..41145bbcc 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java @@ -99,6 +99,28 @@ public class KafkaStreamsBinderWordCountFunctionTests { } } + @Test + public void testKstreamWordCountFunctionWithGeneratedApplicationId() throws Exception { + SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--spring.cloud.stream.function.inputBindings.process=input", + "--spring.cloud.stream.function.outputBindings.process=output", + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.destination=words", + "--spring.cloud.stream.bindings.output.destination=counts", + "--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", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + receiveAndValidate(context); + } + } + private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps);