diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index 63f7ad345..1bcee068f 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -135,6 +135,10 @@ public class KafkaBinderConfigurationProperties { this.kafkaProperties = kafkaProperties; } + public KafkaProperties getKafkaProperties() { + return kafkaProperties; + } + public Transaction getTransaction() { return this.transaction; } diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc index ebb1efb5a..380b6a7e0 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc @@ -115,14 +115,13 @@ serdeError:: + Default: `logAndFail` applicationId:: - Application ID for all the stream configurations in the current application context. - You can override the application id for an individual `StreamListener` method using the `group` property on the binding. - You have to ensure that you are using the same group name for all input bindings in the case of multiple inputs on the same methods. + Convenient way to set the application.id for the Kafka Streams application globally at the binder level. + If the application contains multiple `StreamListener` methods, then application.id should be set at the binding level per input binding. + -Default: `default` +Default: `none` -The following properties are _only_ available for Kafka Streams producers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..producer.` -literal. +The following properties are _only_ available for Kafka Streams producers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..producer.` literal. +For convenience, if there multiple output bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.producer.`. keySerde:: key serde to use @@ -137,9 +136,13 @@ useNativeEncoding:: + Default: `false`. -The following properties are _only_ available for Kafka Streams consumers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..consumer.` -literal. +The following properties are _only_ available for Kafka Streams consumers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings..consumer.`literal. +For convenience, if there multiple input bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.consumer.`. +applicationId:: + Setting application.id per input binding. ++ +Default: `none` keySerde:: key serde to use + diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 4bb08df47..41b232471 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -17,8 +17,9 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.Collection; -import java.util.HashMap; import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsConfig; @@ -40,8 +41,11 @@ import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.annotation.Bean; +import org.springframework.core.env.Environment; +import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.core.CleanupConfig; import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; /** * @author Marius Bogoevici @@ -59,30 +63,66 @@ public class KafkaStreamsBinderSupportAutoConfiguration { return new KafkaStreamsBinderConfigurationProperties(kafkaProperties); } + @Bean + public KafkaStreamsConfiguration kafkaStreamsConfiguration(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, + Environment environment) { + KafkaProperties kafkaProperties = binderConfigurationProperties.getKafkaProperties(); + Map streamsProperties = kafkaProperties.buildStreamsProperties(); + if (kafkaProperties.getStreams().getApplicationId() == null) { + String applicationName = environment.getProperty("spring.application.name"); + if (applicationName != null) { + streamsProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationName); + } + } + return new KafkaStreamsConfiguration(streamsProperties); + } + @Bean("streamConfigGlobalProperties") - public Map streamConfigGlobalProperties(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { - Map props = new HashMap<>(); - props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, binderConfigurationProperties.getKafkaConnectionString()); - props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName()); - props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName()); - props.put(StreamsConfig.APPLICATION_ID_CONFIG, binderConfigurationProperties.getApplicationId()); + public Map streamConfigGlobalProperties(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, + KafkaStreamsConfiguration kafkaStreamsConfiguration) { + + Properties properties = kafkaStreamsConfiguration.asProperties(); + // Override Spring Boot bootstrap server setting if left to default with the value + // configured in the binder + if (ObjectUtils.isEmpty(properties.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG))) { + properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, binderConfigurationProperties.getKafkaConnectionString()); + } + else { + Object bootstrapServerConfig = properties.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG); + if (bootstrapServerConfig instanceof String) { + @SuppressWarnings("unchecked") + String bootStrapServers = (String) properties + .get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG); + if (bootStrapServers.equals("localhost:9092")) { + properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, binderConfigurationProperties.getKafkaConnectionString()); + } + } + } + + String binderProvidedApplicationId = binderConfigurationProperties.getApplicationId(); + if (StringUtils.hasText(binderProvidedApplicationId)) { + properties.put(StreamsConfig.APPLICATION_ID_CONFIG, binderProvidedApplicationId); + } + + properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName()); + properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName()); if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndContinue) { - props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class); } else if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) { - props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class); } else if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) { - props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, SendToDlqAndContinue.class); } if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConfiguration())) { - props.putAll(binderConfigurationProperties.getConfiguration()); + properties.putAll(binderConfigurationProperties.getConfiguration()); } - - return props; + return properties.entrySet().stream().collect( + Collectors.toMap(e -> String.valueOf(e.getKey()), Map.Entry::getValue)); } @Bean 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 c46d08c5b..0abdde3ce 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 @@ -38,14 +38,19 @@ import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; +import org.springframework.beans.BeanUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanInitializationException; 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.Bindable; +import org.springframework.boot.context.properties.bind.PropertySourcesPlaceholdersResolver; +import org.springframework.boot.context.properties.source.ConfigurationPropertySources; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsStateStore; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; @@ -58,11 +63,13 @@ import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; import org.springframework.cloud.stream.binding.StreamListenerSetupMethodOrchestrator; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.cloud.stream.config.MergableProperties; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.MethodParameter; import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.messaging.MessageHeaders; @@ -236,7 +243,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene //Retrieve the StreamsConfig created for this method if available. //Otherwise, create the StreamsBuilderFactory and get the underlying config. if (!methodStreamsBuilderFactoryBeanMap.containsKey(method)) { - streamsConfig = buildStreamsBuilderAndRetrieveConfig(method, applicationContext, bindingProperties); + streamsConfig = buildStreamsBuilderAndRetrieveConfig(method, applicationContext, inboundName); } try { StreamsBuilderFactoryBean streamsBuilderFactoryBean = methodStreamsBuilderFactoryBeanMap.get(method); @@ -387,7 +394,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene stream = stream.mapValues(value -> { Object returnValue; String contentType = bindingProperties.getContentType(); - if (!StringUtils.isEmpty(contentType) && !nativeDecoding) { + if (value != null && !StringUtils.isEmpty(contentType) && !nativeDecoding) { returnValue = MessageBuilder.withPayload(value) .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); } @@ -411,14 +418,22 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene @SuppressWarnings({"unchecked"}) private StreamsConfig buildStreamsBuilderAndRetrieveConfig(Method method, ApplicationContext applicationContext, - BindingProperties bindingProperties) { + String inboundName) { ConfigurableListableBeanFactory beanFactory = this.applicationContext.getBeanFactory(); - String group = bindingProperties.getGroup(); - if (!StringUtils.hasText(group)) { - group = binderConfigurationProperties.getApplicationId(); - } + Map streamConfigGlobalProperties = applicationContext.getBean("streamConfigGlobalProperties", Map.class); - streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, group); + + KafkaStreamsConsumerProperties extendedConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(inboundName); + //Need to apply the default extended properties here as it is not yet done by the BindingService in the binding lifecycle. + handleExtendedDefaultProperties(kafkaStreamsExtendedBindingProperties, + extendedConsumerProperties); + + String applicationId = extendedConsumerProperties.getApplicationId(); + + //override application.id if set at the individual binding level. + if (StringUtils.hasText(applicationId)) { + streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); + } //Custom StreamsConfig implementation that overrides to guarantee that the deserialization handler is cached. StreamsConfig streamsConfig = new StreamsConfig(streamConfigGlobalProperties) { @@ -457,6 +472,28 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene return streamsConfig; } + // This method is mostly copied from core. We should refactor the original method in core so that it is publicly available + // as a utility method. This is currently hidden as a private method in BindingService. + private void handleExtendedDefaultProperties(KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, + MergableProperties extendedProperties) { + String defaultsPrefix = kafkaStreamsExtendedBindingProperties.getDefaultsPrefix(); + + if (defaultsPrefix != null) { + Class extendedPropertiesEntryClass = kafkaStreamsExtendedBindingProperties.getExtendedPropertiesEntryClass(); + if (BinderSpecificPropertiesProvider.class.isAssignableFrom(extendedPropertiesEntryClass)) { + org.springframework.boot.context.properties.bind.Binder extendedPropertiesResolverBinder = + new org.springframework.boot.context.properties.bind.Binder(ConfigurationPropertySources.get(applicationContext.getEnvironment()), + new PropertySourcesPlaceholdersResolver(applicationContext.getEnvironment()), + IntegrationUtils.getConversionService(this.applicationContext.getBeanFactory()), null); + BinderSpecificPropertiesProvider defaultProperties = BeanUtils.instantiateClass(extendedPropertiesEntryClass); + extendedPropertiesResolverBinder.bind(defaultsPrefix, Bindable.ofInstance(defaultProperties)); + + Object binderExtendedProperties = defaultProperties.getConsumer(); + ((MergableProperties)binderExtendedProperties).merge(extendedProperties); + } + } + } + @Override public final void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java index 4c51bcbbb..f057e9b12 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java @@ -35,7 +35,7 @@ public class KafkaStreamsBinderConfigurationProperties extends KafkaBinderConfig sendToDlq } - private String applicationId = "default"; + private String applicationId; public String getApplicationId() { return applicationId; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java index fe36323c1..35393310e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java @@ -24,6 +24,8 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerPro */ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties { + private String applicationId; + /** * Key serde specified per binding. */ @@ -39,6 +41,14 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties { */ private String materializedAs; + public String getApplicationId() { + return applicationId; + } + + public void setApplicationId(String applicationId) { + this.applicationId = applicationId; + } + public String getKeySerde() { return keySerde; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java index e03b1adc0..eca6ba627 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java @@ -30,6 +30,8 @@ import org.springframework.cloud.stream.binder.ExtendedBindingProperties; public class KafkaStreamsExtendedBindingProperties implements ExtendedBindingProperties { + private static final String DEFAULTS_PREFIX = "spring.cloud.stream.kafka.streams.default"; + private Map bindings = new HashMap<>(); public Map getBindings() { @@ -62,7 +64,7 @@ public class KafkaStreamsExtendedBindingProperties @Override public String getDefaultsPrefix() { - return null; + return DEFAULTS_PREFIX; } @Override diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java index 6ca02b768..6d505dde5 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java @@ -102,6 +102,7 @@ public abstract class DeserializationErrorHandlerByKafkaTests { @SpringBootTest(properties = { "spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", "spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=deser-kafka-dlq", "spring.cloud.stream.bindings.input.group=group", "spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq", "spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=" + @@ -138,6 +139,7 @@ public abstract class DeserializationErrorHandlerByKafkaTests { "spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", "spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", "spring.cloud.stream.bindings.input.destination=word1,word2", + "spring.cloud.stream.kafka.streams.default.consumer.applicationId=deser-kafka-dlq-multi-input", "spring.cloud.stream.bindings.input.group=groupx", "spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq", "spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=" + diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java index 079ab78e6..bb5794f7f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java @@ -104,6 +104,7 @@ public abstract class DeserializtionErrorHandlerByBinderTests { "spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde", "spring.cloud.stream.bindings.input.consumer.headerMode=raw", "spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq", + "spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=deserializationByBinderAndDlqTests", "spring.cloud.stream.bindings.input.group=foobar-group"}, webEnvironment= SpringBootTest.WebEnvironment.NONE ) @@ -138,10 +139,9 @@ public abstract class DeserializtionErrorHandlerByBinderTests { "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.bindings.output.producer.headerMode=raw", "spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde", - "spring.cloud.stream.bindings.input.consumer.headerMode=raw", "spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq", + "spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=deserializationByBinderAndDlqTestsWithMultipleInputs", "spring.cloud.stream.bindings.input.group=fooz-group"}, webEnvironment= SpringBootTest.WebEnvironment.NONE ) diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java index 7622155ec..799b9c2d5 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java @@ -101,10 +101,10 @@ public class KafkaStreamsBinderMultipleInputTopicsTest { "--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.bindings.output.producer.headerMode=raw", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.kafka.streams.timeWindow.length=5000", "--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=WordCountProcessorApplication-xyz", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java index d10e0b0e4..44f48ef57 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java @@ -89,9 +89,8 @@ public class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests { "--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.bindings.output.producer.headerMode=raw", "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde", - "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests-xyz", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { 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 5d3ed84c9..b837c339c 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 @@ -93,15 +93,16 @@ public class KafkaStreamsBinderWordCountIntegrationTests { } @Test - public void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { + public void testKstreamWordCountWithApplicationIdSpecifiedAtDefaultConsumer() throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); - ConfigurableApplicationContext context = app.run("--server.port=0", + try (ConfigurableApplicationContext context = app.run("--server.port=0", "--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.input.destination=words", "--spring.cloud.stream.bindings.output.destination=counts", "--spring.cloud.stream.bindings.output.contentType=application/json", + "--spring.cloud.stream.kafka.streams.default.consumer.application-id=basic-word-count", "--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", @@ -110,8 +111,29 @@ public class KafkaStreamsBinderWordCountIntegrationTests { "--spring.cloud.stream.kafka.streams.timeWindow.length=5000", "--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), - "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); - try { + "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) { + receiveAndValidate(context); + } + } + + @Test + public void testKstreamWordCountWithInputBindingLevelApplicationId() throws Exception { + SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.destination=words", + "--spring.cloud.stream.bindings.output.destination=counts", + "--spring.cloud.stream.bindings.output.contentType=application/json", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=basic-word-count", + "--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.timeWindow.length=5000", + "--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) { receiveAndValidate(context); //Assertions on StreamBuilderFactoryBean StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class); @@ -126,9 +148,6 @@ public class KafkaStreamsBinderWordCountIntegrationTests { assertThat(cleanup.cleanupOnStart()).isTrue(); assertThat(cleanup.cleanupOnStop()).isFalse(); } - finally { - context.close(); - } } private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsInteractiveQueryIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsInteractiveQueryIntegrationTests.java index 1d282e757..8270fb31f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsInteractiveQueryIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsInteractiveQueryIntegrationTests.java @@ -93,8 +93,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { "--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.bindings.output.producer.headerMode=raw", - "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=ProductCountApplication-abc", "--spring.cloud.stream.kafka.streams.binder.configuration.application.server=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java index 940f567e8..a8161a859 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingTests.java @@ -100,7 +100,9 @@ public abstract class KafkaStreamsNativeEncodingDecodingTests { @SpringBootTest(properties = { "spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", - "spring.cloud.stream.bindings.output.producer.useNativeEncoding=true"}, + "spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=NativeEncodingDecodingEnabledTests-abc" + }, webEnvironment= SpringBootTest.WebEnvironment.NONE ) public static class NativeEncodingDecodingEnabledTests extends KafkaStreamsNativeEncodingDecodingTests { @@ -120,7 +122,8 @@ public abstract class KafkaStreamsNativeEncodingDecodingTests { } } - @SpringBootTest(webEnvironment= SpringBootTest.WebEnvironment.NONE) + @SpringBootTest(webEnvironment= SpringBootTest.WebEnvironment.NONE, + properties = "spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=NativeEncodingDecodingEnabledTests-xyz") public static class NativeEncodingDecodingDisabledTests extends KafkaStreamsNativeEncodingDecodingTests { @Test diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java index 541cb0b15..de08e4195 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java @@ -64,7 +64,7 @@ public class KafkaStreamsStateStoreIntegrationTests { "--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.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=KafkaStreamsStateStoreIntegrationTests-abc", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java index 629641ba2..d7b049059 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java @@ -91,9 +91,8 @@ public class KafkastreamsBinderPojoInputStringOutputIntegrationTests { "--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.bindings.output.producer.headerMode=raw", "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde", - "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=ProductCountApplication-xyz", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java index bef59adf3..2b2ea1054 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java @@ -130,6 +130,7 @@ public class StreamToGlobalKTableJoinIntegrationTests { "--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.configuration.commit.interval.ms=10000", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=StreamToGlobalKTableJoinIntegrationTests-abc", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) { Map senderPropsCustomer = KafkaTestUtils.producerProps(embeddedKafka); diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java index af975c2c8..36fb755b8 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java @@ -137,9 +137,7 @@ public class StreamToTableJoinIntegrationTests { "--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.configuration.commit.interval.ms=10000", - "--spring.cloud.stream.bindings.output.producer.headerMode=raw", - "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", - "--spring.cloud.stream.bindings.inputX.consumer.headerMode=raw", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=StreamToTableJoinIntegrationTests-abc", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) { // Input 1: Clicks per user (multiple records allowed per user). diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/WordCountMultipleBranchesIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/WordCountMultipleBranchesIntegrationTests.java index 6daa0db68..e862b2129 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/WordCountMultipleBranchesIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/WordCountMultipleBranchesIntegrationTests.java @@ -142,10 +142,9 @@ public class WordCountMultipleBranchesIntegrationTests { "--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.bindings.output.producer.headerMode=raw", - "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.kafka.streams.timeWindow.length=5000", "--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=WordCountMultipleBranchesIntegrationTests-abc", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties b/spring-cloud-stream-binder-kafka-streams/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties index 815a9cab9..a2342cc7f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties +++ b/spring-cloud-stream-binder-kafka-streams/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties @@ -4,7 +4,5 @@ spring.cloud.stream.bindings.output.contentType=application/json 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.bindings.output.producer.headerMode=raw -spring.cloud.stream.bindings.input.consumer.headerMode=raw spring.cloud.stream.kafka.streams.timeWindow.length=5000 spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0