From 70f385553a5c2688c2744fe70546f992c4c30baf Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 21 Sep 2018 10:17:34 -0400 Subject: [PATCH] Application ID resolution for Kafka streams binder (#450) * Application ID resolution for kafka streams binder Avoid the need to rely on group property for application id. First, check binding specific application id, if not look at defaults. If nothing works, fall back to the default application id set by the boot auto configuration. Modify tests. Update docs. Resolves #448 * Addressing PR review comments * Minor polishing * Addressing PR review comments --- .../KafkaBinderConfigurationProperties.java | 4 ++ .../src/main/asciidoc/kafka-streams.adoc | 19 +++--- ...StreamsBinderSupportAutoConfiguration.java | 66 +++++++++++++++---- ...StreamListenerSetupMethodOrchestrator.java | 53 ++++++++++++--- ...aStreamsBinderConfigurationProperties.java | 2 +- .../KafkaStreamsConsumerProperties.java | 10 +++ ...KafkaStreamsExtendedBindingProperties.java | 4 +- ...serializationErrorHandlerByKafkaTests.java | 2 + ...serializtionErrorHandlerByBinderTests.java | 4 +- ...aStreamsBinderMultipleInputTopicsTest.java | 2 +- ...rPojoInputAndPrimitiveTypeOutputTests.java | 3 +- ...treamsBinderWordCountIntegrationTests.java | 33 ++++++++-- ...reamsInteractiveQueryIntegrationTests.java | 3 +- ...fkaStreamsNativeEncodingDecodingTests.java | 7 +- ...afkaStreamsStateStoreIntegrationTests.java | 2 +- ...PojoInputStringOutputIntegrationTests.java | 3 +- ...eamToGlobalKTableJoinIntegrationTests.java | 1 + .../StreamToTableJoinIntegrationTests.java | 4 +- ...CountMultipleBranchesIntegrationTests.java | 3 +- .../binder/kstream/integTest-1.properties | 2 - 20 files changed, 170 insertions(+), 57 deletions(-) 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