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);