diff --git a/pom.xml b/pom.xml index c49c39003..40b1d1fd9 100644 --- a/pom.xml +++ b/pom.xml @@ -12,8 +12,8 @@ 1.8 - 2.1.2.RELEASE - 3.0.1.RELEASE + 2.1.3.BUILD-SNAPSHOT + 3.0.2.BUILD-SNAPSHOT 1.0.0 2.0.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index 8f3a56a3c..0b54a6a21 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java @@ -387,6 +387,10 @@ public class KafkaTopicProvisioner implements ProvisioningProvider> doBindConsumer(String name, String group, KStream inputTarget, ExtendedConsumerProperties properties) { + this.KStreamBindingInformationCatalogue.registerConsumerProperties(inputTarget, properties.getExtension()); ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( - new KafkaConsumerProperties()); + properties.getExtension()); + if (properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.sendToDlq) { + extendedConsumerProperties.getExtension().setEnableDlq(true); + } + if (!StringUtils.hasText(group)) { + group = binderConfigurationProperties.getApplicationId(); + } this.kafkaTopicProvisioner.provisionConsumerDestination(name, group, extendedConsumerProperties); + + //populate the per binding StreamConfig properties + Map streamConfigGlobalProperties = getApplicationContext().getBean("streamConfigGlobalProperties", Map.class); + + StreamsBuilderFactoryBean streamsBuilder = getApplicationContext().getBean("&stream-builder-" + name, StreamsBuilderFactoryBean.class); + + streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, group); + + if(properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.logAndContinue) { + streamConfigGlobalProperties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + LogAndContinueExceptionHandler.class); + } + else if(properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.logAndFail) { + streamConfigGlobalProperties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + LogAndFailExceptionHandler.class); + } + else if (properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.sendToDlq) { + streamConfigGlobalProperties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + SendToDlqAndContinue.class); + } + + StreamsConfig streamsConfig = new StreamsConfig(streamConfigGlobalProperties) { + + DeserializationExceptionHandler deserializationExceptionHandler; + + @Override + @SuppressWarnings("unchecked") + public T getConfiguredInstance(String key, Class t) { + if (key.equals(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG)){ + if (deserializationExceptionHandler != null){ + return (T)deserializationExceptionHandler; + } + else { + T t1 = super.getConfiguredInstance(key, t); + deserializationExceptionHandler = (DeserializationExceptionHandler)t1; + return t1; + } + } + return super.getConfiguredInstance(key, t); + } + }; + + ConfigurableListableBeanFactory beanFactory = getApplicationContext().getBeanFactory(); + beanFactory.registerSingleton("streamsConfig-" + name, streamsConfig); + beanFactory.initializeBean(streamsConfig, "streamsConfig-" + name); + + streamsBuilder.setStreamsConfig(streamsConfig); + streamsBuilder.start(); + queryableStoreRegistry.registerKafkaStreams(streamsBuilder.getKafkaStreams()); + + if (extendedConsumerProperties.getExtension().isEnableDlq()) { + String dlqName = StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName()) ? + "error." + name + "." + group : extendedConsumerProperties.getExtension().getDlqName(); + KStreamDlqDispatch kStreamDlqDispatch = new KStreamDlqDispatch(dlqName, binderConfigurationProperties, + extendedConsumerProperties.getExtension()); + SendToDlqAndContinue sendToDlqAndContinue = this.getApplicationContext().getBean(SendToDlqAndContinue.class); + sendToDlqAndContinue.addKStreamDlqDispatch(name, kStreamDlqDispatch); + + DeserializationExceptionHandler deserializationExceptionHandler = streamsConfig.defaultDeserializationExceptionHandler(); + if(deserializationExceptionHandler instanceof SendToDlqAndContinue) { + ((SendToDlqAndContinue)deserializationExceptionHandler).addKStreamDlqDispatch(name, kStreamDlqDispatch); + } + } return new DefaultBinding<>(name, group, inputTarget, null); } @@ -89,69 +169,20 @@ public class KStreamBinder extends new KafkaProducerProperties()); this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties); - Serde keySerde = getKeySerde(properties); - Serde valueSerde = getValueSerde(properties); + Serde keySerde = this.keyValueSerdeResolver.getOuboundKeySerde(properties.getExtension()); + Serde valueSerde = this.keyValueSerdeResolver.getOutboundValueSerde(properties, properties.getExtension()); to(properties.isUseNativeEncoding(), name, outboundBindTarget, (Serde) keySerde, (Serde) valueSerde); return new DefaultBinding<>(name, null, outboundBindTarget, null); } - private Serde getKeySerde(ExtendedProducerProperties properties) { - Serde keySerde; - try { - if (StringUtils.hasText(properties.getExtension().getKeySerde())) { - keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class); - if (keySerde instanceof Configurable) { - ((Configurable) keySerde).configure(streamsConfig.originals()); - } - } - else { - keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ? - Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray(); - } - - } - catch (ClassNotFoundException e) { - throw new IllegalStateException("Serde class not found: ", e); - } - return keySerde; - } - - private Serde getValueSerde(ExtendedProducerProperties properties) { - Serde valueSerde; - try { - if (properties.isUseNativeEncoding()) { - if (StringUtils.hasText(properties.getExtension().getValueSerde())) { - valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class); - if (valueSerde instanceof Configurable) { - ((Configurable) valueSerde).configure(streamsConfig.originals()); - } - } - else { - valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ? - Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray(); - } - } - else { - valueSerde = Serdes.ByteArray(); - } - } - catch (ClassNotFoundException e) { - throw new IllegalStateException("Serde class not found: ", e); - } - return valueSerde; - } - @SuppressWarnings("unchecked") private void to(boolean isNativeEncoding, String name, KStream outboundBindTarget, Serde keySerde, Serde valueSerde) { - KeyValueMapper> keyValueMapper = null; if (!isNativeEncoding) { - keyValueMapper = messageConversionDelegate.outboundKeyValueMapper(name); - } - if (!isNativeEncoding) { - outboundBindTarget.map(keyValueMapper).to(name, Produced.with(keySerde, valueSerde)); + kStreamBoundMessageConversionDelegate.serializeOnOutbound(outboundBindTarget) + .to(name, Produced.with(keySerde, valueSerde)); } else { outboundBindTarget.to(name, Produced.with(keySerde, valueSerde)); @@ -167,4 +198,8 @@ public class KStreamBinder extends public KStreamProducerProperties getExtendedProducerProperties(String channelName) { return this.kStreamExtendedBindingProperties.getExtendedProducerProperties(channelName); } + + public void setkStreamExtendedBindingProperties(KStreamExtendedBindingProperties kStreamExtendedBindingProperties) { + this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties; + } } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBindingInformationCatalogue.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBindingInformationCatalogue.java new file mode 100644 index 000000000..277b86bd8 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBindingInformationCatalogue.java @@ -0,0 +1,117 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.kafka.streams.kstream.KStream; + +import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties; +import org.springframework.cloud.stream.config.BindingProperties; + +/** + * A catalogue containing all the inbound and outboud KStreams. + * It registers {@link BindingProperties} and {@link KStreamConsumerProperties} + * for the bounded KStreams. This registry provides services for finding + * specific binding level information for the bounded KStream. This includes + * information such as the configured content type, destination etc. + * + * @since 2.0.0 + * + * @author Soby Chacko + */ +public class KStreamBindingInformationCatalogue { + + private final Map, BindingProperties> bindingProperties = new ConcurrentHashMap<>(); + private final Map, KStreamConsumerProperties> consumerProperties = new ConcurrentHashMap<>(); + + /** + * For a given bounded {@link KStream}, retrieve it's corresponding destination + * on the broker. + * + * @param bindingTarget KStream binding target + * @return destination topic on Kafka + */ + public String getDestination(KStream bindingTarget) { + BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget); + return bindingProperties.getDestination(); + } + + /** + * Is native decoding is enabled on this {@link KStream}. + * + * @param bindingTarget KStream binding target + * @return true if native decoding is enabled, fasle otherwise. + */ + public boolean isUseNativeDecoding(KStream bindingTarget) { + BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget); + return bindingProperties.getConsumer().isUseNativeDecoding(); + } + + /** + * Is DLQ enabled for this {@link KStream} + * + * @param bindingTarget KStream binding target + * @return true if DLQ is enabled, false otherwise. + */ + public boolean isEnableDlq(KStream bindingTarget) { + return consumerProperties.get(bindingTarget).isEnableDlq(); + } + + /** + * Retrieve the content type associated with a given {@link KStream} + * + * @param bindingTarget KStream binding target + * @return content Type associated. + */ + public String getContentType(KStream bindingTarget) { + BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget); + return bindingProperties.getContentType(); + } + + /** + * Retrieve any configured Serde error handling strategies for this {@link KStream} + * + * @param bindingTarget KStream binding target + * @return configured Serde error handling strategy + */ + public KStreamConsumerProperties.SerdeError getSerdeError(KStream bindingTarget) { + return consumerProperties.get(bindingTarget).getSerdeError(); + } + + /** + * Register a cache for bounded KStream -> {@link BindingProperties} + * + * @param bindingTarget KStream binding target + * @param bindingProperties {@link BindingProperties} for this KStream + */ + public void registerBindingProperties(KStream bindingTarget, BindingProperties bindingProperties) { + this.bindingProperties.put(bindingTarget, bindingProperties); + } + + /** + * Register a cache for bounded KStream -> {@link KStreamConsumerProperties} + * + * @param bindingTarget KStream binding target + * @param kStreamConsumerProperties Consumer properties for this KStream + */ + public void registerConsumerProperties(KStream bindingTarget, KStreamConsumerProperties kStreamConsumerProperties) { + this.consumerProperties.put(bindingTarget, kStreamConsumerProperties); + } + +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java index c383c098b..a17166872 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java @@ -18,14 +18,24 @@ package org.springframework.cloud.stream.binder.kstream; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.streams.Consumed; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.KStream; import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties; +import org.springframework.cloud.stream.binder.kstream.config.KStreamExtendedBindingProperties; import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; @@ -36,27 +46,71 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Soby Chacko */ -public class KStreamBoundElementFactory extends AbstractBindingTargetFactory { - - private final StreamsBuilder kStreamBuilder; +public class KStreamBoundElementFactory extends AbstractBindingTargetFactory implements ApplicationContextAware { private final BindingServiceProperties bindingServiceProperties; - public KStreamBoundElementFactory(StreamsBuilder kStreamBuilder, BindingServiceProperties bindingServiceProperties) { + private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue; + + private final KeyValueSerdeResolver keyValueSerdeResolver; + + private volatile AbstractApplicationContext applicationContext; + + private KStreamExtendedBindingProperties kStreamExtendedBindingProperties = new KStreamExtendedBindingProperties(); + + public KStreamBoundElementFactory(BindingServiceProperties bindingServiceProperties, + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue, + KeyValueSerdeResolver keyValueSerdeResolver) { super(KStream.class); this.bindingServiceProperties = bindingServiceProperties; - this.kStreamBuilder = kStreamBuilder; + this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue; + this.keyValueSerdeResolver = keyValueSerdeResolver; + } + + public void setkStreamExtendedBindingProperties(KStreamExtendedBindingProperties kStreamExtendedBindingProperties) { + this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + Assert.isInstanceOf(AbstractApplicationContext.class, applicationContext); + this.applicationContext = (AbstractApplicationContext) applicationContext; } @Override public KStream createInput(String name) { - KStream stream = kStreamBuilder.stream(bindingServiceProperties.getBindingDestination(name)); + + BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name); + String destination = bindingProperties.getDestination(); + if (destination == null) { + destination = name; + } + KStreamConsumerProperties extendedConsumerProperties = kStreamExtendedBindingProperties.getExtendedConsumerProperties(name); + Serde keySerde = this.keyValueSerdeResolver.getInboundKeySerde(extendedConsumerProperties); + + Serde valueSerde = this.keyValueSerdeResolver.getInboundValueSerde(bindingProperties.getConsumer(), + extendedConsumerProperties); + + ConfigurableListableBeanFactory beanFactory = this.applicationContext.getBeanFactory(); + StreamsBuilderFactoryBean streamsBuilder = new StreamsBuilderFactoryBean(); + streamsBuilder.setAutoStartup(false); + beanFactory.registerSingleton("stream-builder-" + destination, streamsBuilder); + beanFactory.initializeBean(streamsBuilder, "stream-builder-" + destination); + + StreamsBuilder streamBuilder = null; + try { + streamBuilder = streamsBuilder.getObject(); + } catch (Exception e) { + //log and bail + } + + KStream stream = streamBuilder.stream(bindingServiceProperties.getBindingDestination(name), + Consumed.with(keySerde, valueSerde)); stream = stream.map((key, value) -> { KeyValue keyValue; - BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name); String contentType = bindingProperties.getContentType(); if (!StringUtils.isEmpty(contentType) && !bindingProperties.getConsumer().isUseNativeDecoding()) { - Message message = MessageBuilder.withPayload(value) + Message message = MessageBuilder.withPayload(value) .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); keyValue = new KeyValue<>(key, message); } @@ -65,6 +119,7 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory> keyValueThreadLocal = new ThreadLocal<>(); + + private final CompositeMessageConverterFactory compositeMessageConverterFactory; + + private final SendToDlqAndContinue sendToDlqAndContinue; + + private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue; + + public KStreamBoundMessageConversionDelegate(CompositeMessageConverterFactory compositeMessageConverterFactory, + SendToDlqAndContinue sendToDlqAndContinue, + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { + this.compositeMessageConverterFactory = compositeMessageConverterFactory; + this.sendToDlqAndContinue = sendToDlqAndContinue; + this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue; + } + + /** + * Serialize {@link KStream} records on outbound based on contentType. + * + * @param outboundBindTarget outbound KStream target + * @return serialized KStream + */ + public KStream serializeOnOutbound(KStream outboundBindTarget) { + String contentType = this.KStreamBindingInformationCatalogue.getContentType(outboundBindTarget); + MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory + .getMessageConverterForType(MimeType.valueOf(contentType)) + : null; + + return outboundBindTarget.map((k, v) -> { + Message message = v instanceof Message ? (Message) v : + MessageBuilder.withPayload(v).build(); + Map headers = new HashMap<>(message.getHeaders()); + if (!StringUtils.isEmpty(contentType)) { + headers.put(MessageHeaders.CONTENT_TYPE, contentType); + } + MessageHeaders messageHeaders = new MessageHeaders(headers); + return new KeyValue<>(k, + messageConverter.toMessage(message.getPayload(), + messageHeaders).getPayload()); + }); + } + + /** + * Deserialize incoming {@link KStream} based on contentType. + * + * @param valueClass on KStream value + * @param bindingTarget inbound KStream target + * @return deserialized KStream + */ + @SuppressWarnings("unchecked") + public KStream deserializeOnInbound(Class valueClass, KStream bindingTarget) { + MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered(); + + KStream[] branch = bindingTarget.branch( + (o, o2) -> { + boolean isValidRecord = false; + + try { + if (valueClass.isAssignableFrom(o2.getClass())) { + keyValueThreadLocal.set(new KeyValue<>(o, o2)); + } + else if (o2 instanceof Message) { + if (valueClass.isAssignableFrom(((Message) o2).getPayload().getClass())) { + keyValueThreadLocal.set(new KeyValue<>(o, ((Message) o2).getPayload())); + } + else { + convertAndSetMessage(o, valueClass, messageConverter, (Message) o2); + } + } + else if (o2 instanceof String || o2 instanceof byte[]) { + Message message = MessageBuilder.withPayload(o2).build(); + convertAndSetMessage(o, valueClass, messageConverter, message); + } + else { + keyValueThreadLocal.set(new KeyValue<>(o, o2)); + } + isValidRecord = true; + } + catch (Exception ignored) { + //pass through + } + return isValidRecord; + }, + (k, v) -> true + ); + processErrorFromDeserialization(bindingTarget, branch[1]); + + return branch[0].map((o, o2) -> { + KeyValue objectObjectKeyValue = keyValueThreadLocal.get(); + keyValueThreadLocal.remove(); + return objectObjectKeyValue; + }); + } + + private void convertAndSetMessage(Object o, Class valueClass, MessageConverter messageConverter, Message msg) { + Object messageConverted = messageConverter.fromMessage(msg, valueClass); + if (messageConverted == null) { + throw new IllegalStateException("Inbound data conversion failed."); + } + keyValueThreadLocal.set(new KeyValue<>(o, messageConverted)); + } + + @SuppressWarnings("unchecked") + private void processErrorFromDeserialization(KStream bindingTarget, KStream branch) { + branch.process(() -> new Processor() { + ProcessorContext context; + + @Override + public void init(ProcessorContext context) { + this.context = context; + } + + @Override + public void process(Object o, Object o2) { + if (KStreamBindingInformationCatalogue.isEnableDlq(bindingTarget)) { + String destination = KStreamBindingInformationCatalogue.getDestination(bindingTarget); + if (o2 instanceof Message) { + Message message = (Message) o2; + sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) message.getPayload(), context.partition()); + } + else { + sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) o2, context.partition()); + } + } + else if (KStreamBindingInformationCatalogue.getSerdeError(bindingTarget) == KStreamConsumerProperties.SerdeError.logAndFail) { + throw new IllegalStateException("Inbound deserialization failed."); + } + else if (KStreamBindingInformationCatalogue.getSerdeError(bindingTarget) == KStreamConsumerProperties.SerdeError.logAndContinue) { + //quietly pass through. No action needed, this is similar to log and continue. + } + } + + @Override + public void punctuate(long timestamp) { + + } + + @Override + public void close() { + + } + }); + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamDlqDispatch.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamDlqDispatch.java new file mode 100644 index 000000000..dda69696c --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamDlqDispatch.java @@ -0,0 +1,144 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +/** + * @author Soby Chacko + */ +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.ByteArraySerializer; + +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.SendResult; +import org.springframework.util.ObjectUtils; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.ListenableFutureCallback; + +class KStreamDlqDispatch { + + private final Log logger = LogFactory.getLog(getClass()); + + private final KafkaTemplate kafkaTemplate; + private final String dlqName; + + KStreamDlqDispatch(String dlqName, + KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties, + KafkaConsumerProperties kafkaConsumerProperties) { + ProducerFactory producerFactory = getProducerFactory(null, + new ExtendedProducerProperties<>(kafkaConsumerProperties.getDlqProducerProperties()), + kafkaBinderConfigurationProperties); + + this.kafkaTemplate = new KafkaTemplate<>(producerFactory); + this.dlqName = dlqName; + } + + @SuppressWarnings("unchecked") + public void sendToDlq(byte[] key, byte[] value, int partittion) { + ProducerRecord producerRecord = new ProducerRecord<>(this.dlqName, partittion, + key, value, null); + + StringBuilder sb = new StringBuilder().append(" a message with key='") + .append(toDisplayString(ObjectUtils.nullSafeToString(key), 50)).append("'") + .append(" and payload='") + .append(toDisplayString(ObjectUtils.nullSafeToString(value), 50)) + .append("'").append(" received from ") + .append(partittion); + ListenableFuture> sentDlq = null; + try { + sentDlq = this.kafkaTemplate.send(producerRecord); + sentDlq.addCallback(new ListenableFutureCallback>() { + + @Override + public void onFailure(Throwable ex) { + KStreamDlqDispatch.this.logger.error( + "Error sending to DLQ " + sb.toString(), ex); + } + + @Override + public void onSuccess(SendResult result) { + if (KStreamDlqDispatch.this.logger.isDebugEnabled()) { + KStreamDlqDispatch.this.logger.debug( + "Sent to DLQ " + sb.toString()); + } + } + }); + } + catch (Exception ex) { + if (sentDlq == null) { + KStreamDlqDispatch.this.logger.error( + "Error sending to DLQ " + sb.toString(), ex); + } + } + } + + private DefaultKafkaProducerFactory getProducerFactory(String transactionIdPrefix, + ExtendedProducerProperties producerProperties, + KafkaBinderConfigurationProperties configurationProperties) { + Map props = new HashMap<>(); + props.put(ProducerConfig.RETRIES_CONFIG, 0); + props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); + props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(configurationProperties.getRequiredAcks())); + if (!ObjectUtils.isEmpty(configurationProperties.getProducerConfiguration())) { + props.putAll(configurationProperties.getProducerConfiguration()); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString()); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { + props.put(ProducerConfig.BATCH_SIZE_CONFIG, + String.valueOf(producerProperties.getExtension().getBufferSize())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { + props.put(ProducerConfig.LINGER_MS_CONFIG, + String.valueOf(producerProperties.getExtension().getBatchTimeout())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { + props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, + producerProperties.getExtension().getCompressionType().toString()); + } + if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) { + props.putAll(producerProperties.getExtension().getConfiguration()); + } + //Always send as byte[] on dlq (the same byte[] that the consumer received) + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(props); + if (transactionIdPrefix != null) { + producerFactory.setTransactionIdPrefix(transactionIdPrefix); + } + return producerFactory; + } + + private String toDisplayString(String original, int maxCharacters) { + if (original.length() <= maxCharacters) { + return original; + } + return original.substring(0, maxCharacters) + "..."; + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java index 5c6707f93..f25deb4cf 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java @@ -16,7 +16,9 @@ package org.springframework.cloud.stream.binder.kstream; +import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter; import org.springframework.core.MethodParameter; @@ -28,10 +30,13 @@ import org.springframework.core.ResolvableType; */ public class KStreamListenerParameterAdapter implements StreamListenerParameterAdapter, KStream> { - private final MessageConversionDelegate messageConversionDelegate; + private final KStreamBoundMessageConversionDelegate kStreamBoundMessageConversionDelegate; + private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue; - public KStreamListenerParameterAdapter(MessageConversionDelegate messageConversionDelegate) { - this.messageConversionDelegate = messageConversionDelegate; + public KStreamListenerParameterAdapter(KStreamBoundMessageConversionDelegate kStreamBoundMessageConversionDelegate, + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { + this.kStreamBoundMessageConversionDelegate = kStreamBoundMessageConversionDelegate; + this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue; } @Override @@ -46,8 +51,11 @@ public class KStreamListenerParameterAdapter implements StreamListenerParameterA ResolvableType resolvableType = ResolvableType.forMethodParameter(parameter); final Class valueClass = (resolvableType.getGeneric(1).getRawClass() != null) ? (resolvableType.getGeneric(1).getRawClass()) : Object.class; - - return bindingTarget.map(messageConversionDelegate.inboundKeyValueMapper(valueClass)); + if (this.KStreamBindingInformationCatalogue.isUseNativeDecoding(bindingTarget)) { + return bindingTarget.map((KeyValueMapper) KeyValue::new); + } + else { + return kStreamBoundMessageConversionDelegate.deserializeOnInbound(valueClass, bindingTarget); + } } - } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerSetupMethodOrchestrator.java index eb2a68388..a131c4523 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerSetupMethodOrchestrator.java @@ -42,6 +42,8 @@ import org.springframework.util.StringUtils; * Kafka Streams specific implementation for {@link StreamListenerSetupMethodOrchestrator} * that overrides the default mechanisms for invoking StreamListener adapters. * + * @since 2.0.0 + * * @author Soby Chacko */ public class KStreamListenerSetupMethodOrchestrator implements StreamListenerSetupMethodOrchestrator, ApplicationContextAware { diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KeyValueSerdeResolver.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KeyValueSerdeResolver.java new file mode 100644 index 000000000..107a40162 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KeyValueSerdeResolver.java @@ -0,0 +1,165 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Map; + +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.utils.Utils; + +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.binder.kstream.config.KStreamBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties; +import org.springframework.cloud.stream.binder.kstream.config.KStreamProducerProperties; +import org.springframework.util.StringUtils; + +/** + * Resolver for key and value Serde. + * + * On the inbound, if native decoding is enabled, then any deserialization on the value is handled by Kafka. + * First, we look for any key/value Serde set on the binding itself, if that is not available then look at the + * common Serde set at the global level. If that fails, it falls back to byte[]. + * If native decoding is disabled, then the binder will do the deserialization on value and ignore any Serde set for value + * and rely on the contentType provided. Keys are always deserialized at the broker. + * + * Same rules apply on the outbound. If native encoding is enabled, then value serialization is done at the broker using + * any binder level Serde for value, if not using common Serde, if not, then byte[]. + * If native encoding is disabled, then the binder will do serialization using a contentType. Keys are always serialized + * by the broker. + * + * @since 2.0.0 + * + * @author Soby Chacko + */ +public class KeyValueSerdeResolver { + + private final Map streamConfigGlobalProperties; + + private final KStreamBinderConfigurationProperties binderConfigurationProperties; + + public KeyValueSerdeResolver(Map streamConfigGlobalProperties, + KStreamBinderConfigurationProperties binderConfigurationProperties) { + this.streamConfigGlobalProperties = streamConfigGlobalProperties; + this.binderConfigurationProperties = binderConfigurationProperties; + } + + /** + * Provide the {@link Serde} for inbound key + * + * @param extendedConsumerProperties binding level extended {@link KStreamConsumerProperties} + * @return configurd {@link Serde} for the inbound key. + */ + public Serde getInboundKeySerde(KStreamConsumerProperties extendedConsumerProperties) { + String keySerdeString = extendedConsumerProperties.getKeySerde(); + + return getKeySerde(keySerdeString); + } + + /** + * Provide the {@link Serde} for inbound value + * + * @param consumerProperties {@link ConsumerProperties} on binding + * @param extendedConsumerProperties binding level extended {@link KStreamConsumerProperties} + * @return configurd {@link Serde} for the inbound value. + */ + public Serde getInboundValueSerde(ConsumerProperties consumerProperties, KStreamConsumerProperties extendedConsumerProperties) { + Serde valueSerde; + + String valueSerdeString = extendedConsumerProperties.getValueSerde(); + try { + if (consumerProperties != null && + consumerProperties.isUseNativeDecoding()) { + valueSerde = getValueSerde(valueSerdeString); + } + else { + valueSerde = Serdes.ByteArray(); + } + valueSerde.configure(streamConfigGlobalProperties, false); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("Serde class not found: ", e); + } + return valueSerde; + } + + /** + * Provide the {@link Serde} for outbound key + * + * @param properties binding level extended {@link KStreamProducerProperties} + * @return configurd {@link Serde} for the outbound key. + */ + public Serde getOuboundKeySerde(KStreamProducerProperties properties) { + return getKeySerde(properties.getKeySerde()); + } + + /** + * Provide the {@link Serde} for outbound value + * + * @param producerProperties {@link ProducerProperties} on binding + * @param kStreamProducerProperties binding level extended {@link KStreamProducerProperties} + * @return configurd {@link Serde} for the outbound value. + */ + public Serde getOutboundValueSerde(ProducerProperties producerProperties, KStreamProducerProperties kStreamProducerProperties) { + Serde valueSerde; + try { + if (producerProperties.isUseNativeEncoding()) { + valueSerde = getValueSerde(kStreamProducerProperties.getValueSerde()); + } + else { + valueSerde = Serdes.ByteArray(); + } + valueSerde.configure(streamConfigGlobalProperties, false); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("Serde class not found: ", e); + } + return valueSerde; + } + + private Serde getKeySerde(String keySerdeString) { + Serde keySerde; + try { + if (StringUtils.hasText(keySerdeString)) { + keySerde = Utils.newInstance(keySerdeString, Serde.class); + } + else { + keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ? + Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray(); + } + keySerde.configure(streamConfigGlobalProperties, true); + + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("Serde class not found: ", e); + } + return keySerde; + } + + private Serde getValueSerde(String valueSerdeString) throws ClassNotFoundException { + Serde valueSerde; + if (StringUtils.hasText(valueSerdeString)) { + valueSerde = Utils.newInstance(valueSerdeString, Serde.class); + } + else { + valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ? + Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray(); + } + return valueSerde; + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/MessageConversionDelegate.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/MessageConversionDelegate.java deleted file mode 100644 index 9eee5e069..000000000 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/MessageConversionDelegate.java +++ /dev/null @@ -1,96 +0,0 @@ -/* - * Copyright 2018 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kstream; - -import java.util.HashMap; -import java.util.Map; - -import org.apache.kafka.streams.KeyValue; -import org.apache.kafka.streams.kstream.KeyValueMapper; - -import org.springframework.cloud.stream.config.BindingProperties; -import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.MessageConverter; -import org.springframework.messaging.support.MessageBuilder; -import org.springframework.util.MimeType; -import org.springframework.util.StringUtils; - -/** - * @author Soby Chacko - */ -public class MessageConversionDelegate { - - private final BindingServiceProperties bindingServiceProperties; - private final CompositeMessageConverterFactory compositeMessageConverterFactory; - - public MessageConversionDelegate(BindingServiceProperties bindingServiceProperties, - CompositeMessageConverterFactory compositeMessageConverterFactory) { - this.bindingServiceProperties = bindingServiceProperties; - this.compositeMessageConverterFactory = compositeMessageConverterFactory; - } - - public KeyValueMapper> outboundKeyValueMapper(String name) { - BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name); - String contentType = bindingProperties.getContentType(); - MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory - .getMessageConverterForType(MimeType.valueOf(contentType)) - : null; - - return (k, v) -> { - Message message = v instanceof Message ? (Message)v : - MessageBuilder.withPayload(v).build(); - Map headers = new HashMap<>(message.getHeaders()); - if (!StringUtils.isEmpty(contentType)) { - headers.put(MessageHeaders.CONTENT_TYPE, contentType); - } - MessageHeaders messageHeaders = new MessageHeaders(headers); - return new KeyValue<>(k, - messageConverter.toMessage(message.getPayload(), - messageHeaders).getPayload()); - }; - } - - @SuppressWarnings("unchecked") - public KeyValueMapper> inboundKeyValueMapper(Class valueClass) { - MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered(); - return (KeyValueMapper) (o, o2) -> { - KeyValue keyValue; - if (valueClass.isAssignableFrom(o2.getClass())) { - keyValue = new KeyValue<>(o, o2); - } - else if (o2 instanceof Message) { - if (valueClass.isAssignableFrom(((Message) o2).getPayload().getClass())) { - keyValue = new KeyValue<>(o, ((Message) o2).getPayload()); - } - else { - keyValue = new KeyValue<>(o, messageConverter.fromMessage((Message) o2, valueClass)); - } - } - else if(o2 instanceof String || o2 instanceof byte[]) { - Message message = MessageBuilder.withPayload(o2).build(); - keyValue = new KeyValue<>(o, messageConverter.fromMessage(message, valueClass)); - } - else { - keyValue = new KeyValue<>(o, o2); - } - return keyValue; - }; - } -} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/QueryableStoreRegistry.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/QueryableStoreRegistry.java new file mode 100644 index 000000000..acb45cef0 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/QueryableStoreRegistry.java @@ -0,0 +1,63 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.HashSet; +import java.util.Set; + +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.state.QueryableStoreType; + +/** + * Registry that contains {@link QueryableStoreType}s those created from + * the user applications. + * + * @since 2.0.0 + * @author Soby Chacko + */ +public class QueryableStoreRegistry { + + private final Set kafkaStreams = new HashSet<>(); + + /** + * Retrieve and return a queryable store by name created in the application. + * + * @param storeName name of the queryable store + * @param storeType type of the queryable store + * @param generic queryable store + * @return queryable store. + */ + public T getQueryableStoreType(String storeName, QueryableStoreType storeType) { + + for (KafkaStreams kafkaStream : kafkaStreams) { + T store = kafkaStream.store(storeName, storeType); + if (store != null) { + return store; + } + } + return null; + } + + /** + * Register the {@link KafkaStreams} object created in the application. + * + * @param kafkaStreams {@link KafkaStreams} object created in the application + */ + public void registerKafkaStreams(KafkaStreams kafkaStreams) { + this.kafkaStreams.add(kafkaStreams); + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/SendToDlqAndContinue.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/SendToDlqAndContinue.java new file mode 100644 index 000000000..2d61536c0 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/SendToDlqAndContinue.java @@ -0,0 +1,95 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.lang.reflect.Field; +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.streams.errors.DeserializationExceptionHandler; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.processor.internals.ProcessorContextImpl; +import org.apache.kafka.streams.processor.internals.StreamTask; + +import org.springframework.util.ReflectionUtils; + +/** + * Custom implementation for {@link DeserializationExceptionHandler} that sends the records + * in error to a DLQ topic, then continue stream processing on new records. + * + * @since 2.0.0 + * + * @author Soby Chacko + */ +public class SendToDlqAndContinue implements DeserializationExceptionHandler{ + + private Map dlqDispatchers = new HashMap<>(); + + public void sendToDlq(String topic, byte[] key, byte[] value, int partittion){ + KStreamDlqDispatch kStreamDlqDispatch = dlqDispatchers.get(topic); + kStreamDlqDispatch.sendToDlq(key,value, partittion); + } + + @Override + public DeserializationHandlerResponse handle(ProcessorContext context, ConsumerRecord record, Exception exception) { + KStreamDlqDispatch kStreamDlqDispatch = dlqDispatchers.get(record.topic()); + kStreamDlqDispatch.sendToDlq(record.key(), record.value(), record.partition()); + context.commit(); + + // The following conditional block should be reconsidered when we have a solution for this SO problem: + // https://stackoverflow.com/questions/48470899/kafka-streams-deserialization-handler + // Currently it seems like when deserialization error happens, there is no commits happening and the + // following code will use reflection to get access to the underlying KafkaConsumer. + // It works with Kafka 1.0.0, but there is no guarantee it will work in future versions of kafka as + // we access private fields by name using reflection, but it is a temporary fix. + if (context instanceof ProcessorContextImpl){ + ProcessorContextImpl processorContextImpl = (ProcessorContextImpl)context; + Field task = ReflectionUtils.findField(ProcessorContextImpl.class, "task"); + ReflectionUtils.makeAccessible(task); + Object taskField = ReflectionUtils.getField(task, processorContextImpl); + + if (taskField.getClass().isAssignableFrom(StreamTask.class)){ + StreamTask streamTask = (StreamTask)taskField; + Field consumer = ReflectionUtils.findField(StreamTask.class, "consumer"); + ReflectionUtils.makeAccessible(consumer); + Object kafkaConsumerField = ReflectionUtils.getField(consumer, streamTask); + if (kafkaConsumerField.getClass().isAssignableFrom(KafkaConsumer.class)){ + KafkaConsumer kafkaConsumer = (KafkaConsumer)kafkaConsumerField; + final Map consumedOffsetsAndMetadata = new HashMap<>(); + TopicPartition tp = new TopicPartition(record.topic(), record.partition()); + OffsetAndMetadata oam = new OffsetAndMetadata(record.offset() + 1); + consumedOffsetsAndMetadata.put(tp, oam); + kafkaConsumer.commitSync(consumedOffsetsAndMetadata); + } + } + } + return DeserializationHandlerResponse.CONTINUE; + } + + @Override + public void configure(Map configs) { + + } + + public void addKStreamDlqDispatch(String topic, KStreamDlqDispatch kStreamDlqDispatch){ + dlqDispatchers.put(topic, kStreamDlqDispatch); + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java index a0ab4c5d6..528cd6fc9 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java @@ -18,15 +18,16 @@ package org.springframework.cloud.stream.binder.kstream.config; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.apache.kafka.streams.StreamsConfig; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; -import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kstream.KStreamBinder; -import org.springframework.cloud.stream.binder.kstream.MessageConversionDelegate; +import org.springframework.cloud.stream.binder.kstream.KStreamBindingInformationCatalogue; +import org.springframework.cloud.stream.binder.kstream.KStreamBoundMessageConversionDelegate; +import org.springframework.cloud.stream.binder.kstream.KeyValueSerdeResolver; +import org.springframework.cloud.stream.binder.kstream.QueryableStoreRegistry; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -36,7 +37,6 @@ import org.springframework.context.annotation.Configuration; * @author Soby Chacko */ @Configuration -@EnableConfigurationProperties(KStreamExtendedBindingProperties.class) public class KStreamBinderConfiguration { private static final Log logger = LogFactory.getLog(KStreamBinderConfiguration.class); @@ -44,18 +44,24 @@ public class KStreamBinderConfiguration { @Autowired private KafkaProperties kafkaProperties; + @Autowired + private KStreamExtendedBindingProperties kStreamExtendedBindingProperties; + @Bean public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties) { return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties); } @Bean - public KStreamBinder kStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, + public KStreamBinder kStreamBinder(KStreamBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, - KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig, - MessageConversionDelegate messageConversionDelegate) { - KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, kStreamExtendedBindingProperties, - streamsConfig, messageConversionDelegate); + KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate, + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue, + KeyValueSerdeResolver keyValueSerdeResolver, QueryableStoreRegistry queryableStoreRegistry) { + KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, + KStreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue, + keyValueSerdeResolver, queryableStoreRegistry); + kStreamBinder.setkStreamExtendedBindingProperties(kStreamExtendedBindingProperties); return kStreamBinder; } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamCommonProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfigurationProperties.java similarity index 60% rename from spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamCommonProperties.java rename to spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfigurationProperties.java index 973587d15..8641b8cfb 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamCommonProperties.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfigurationProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,28 +16,21 @@ package org.springframework.cloud.stream.binder.kstream.config; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; + /** * @author Soby Chacko */ -public class KStreamCommonProperties { +public class KStreamBinderConfigurationProperties extends KafkaBinderConfigurationProperties { - private String keySerde; + private String applicationId = "default"; - private String valueSerde; - - public String getKeySerde() { - return keySerde; + public String getApplicationId() { + return applicationId; } - public void setKeySerde(String keySerde) { - this.keySerde = keySerde; + public void setApplicationId(String applicationId) { + this.applicationId = applicationId; } - public String getValueSerde() { - return valueSerde; - } - - public void setValueSerde(String valueSerde) { - this.valueSerde = valueSerde; - } } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java index 3a4826c50..ec430f1c1 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java @@ -17,69 +17,56 @@ package org.springframework.cloud.stream.binder.kstream.config; import java.util.Collection; -import java.util.Properties; +import java.util.HashMap; +import java.util.Map; import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; -import org.springframework.beans.factory.ObjectProvider; -import org.springframework.beans.factory.UnsatisfiedDependencyException; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.binder.kstream.KStreamBindingInformationCatalogue; import org.springframework.cloud.stream.binder.kstream.KStreamBoundElementFactory; +import org.springframework.cloud.stream.binder.kstream.KStreamBoundMessageConversionDelegate; import org.springframework.cloud.stream.binder.kstream.KStreamListenerParameterAdapter; import org.springframework.cloud.stream.binder.kstream.KStreamListenerSetupMethodOrchestrator; import org.springframework.cloud.stream.binder.kstream.KStreamStreamListenerResultAdapter; -import org.springframework.cloud.stream.binder.kstream.MessageConversionDelegate; +import org.springframework.cloud.stream.binder.kstream.KeyValueSerdeResolver; +import org.springframework.cloud.stream.binder.kstream.QueryableStoreRegistry; +import org.springframework.cloud.stream.binder.kstream.SendToDlqAndContinue; import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.annotation.Bean; -import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration; -import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.util.ObjectUtils; /** * @author Marius Bogoevici * @author Soby Chacko */ +@EnableConfigurationProperties(KStreamExtendedBindingProperties.class) public class KStreamBinderSupportAutoConfiguration { @Bean @ConfigurationProperties(prefix = "spring.cloud.stream.kstream.binder") - public KafkaBinderConfigurationProperties binderConfigurationProperties() { - return new KafkaBinderConfigurationProperties(); + public KStreamBinderConfigurationProperties binderConfigurationProperties() { + return new KStreamBinderConfigurationProperties(); } - @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME) - public StreamsBuilderFactoryBean defaultKafkaStreamBuilder( - @Qualifier(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider streamsConfigProvider) { - StreamsConfig streamsConfig = streamsConfigProvider.getIfAvailable(); - if (streamsConfig != null) { - StreamsBuilderFactoryBean kStreamBuilderFactoryBean = new StreamsBuilderFactoryBean(streamsConfig); - kStreamBuilderFactoryBean.setPhase(Integer.MAX_VALUE - 500); - return kStreamBuilderFactoryBean; - } else { - throw new UnsatisfiedDependencyException(KafkaStreamsDefaultConfiguration.class.getName(), - KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME, "streamsConfig", - "There is no '" + KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME - + "' StreamsConfig bean in the application context.\n"); - } - } - - @Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) - public StreamsConfig streamsConfig(KafkaBinderConfigurationProperties binderConfigurationProperties) { - Properties props = new Properties(); + @Bean("streamConfigGlobalProperties") + public Map streamConfigGlobalProperties(KStreamBinderConfigurationProperties 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, "default"); + props.put(StreamsConfig.APPLICATION_ID_CONFIG, binderConfigurationProperties.getApplicationId()); + if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConfiguration())) { props.putAll(binderConfigurationProperties.getConfiguration()); } - return new StreamsConfig(props); + + return props; } @Bean @@ -89,8 +76,8 @@ public class KStreamBinderSupportAutoConfiguration { @Bean public KStreamListenerParameterAdapter kafkaStreamListenerParameterAdapter( - MessageConversionDelegate messageConversionDelegate) { - return new KStreamListenerParameterAdapter(messageConversionDelegate); + KStreamBoundMessageConversionDelegate kstreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { + return new KStreamListenerParameterAdapter(kstreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue); } @Bean @@ -101,15 +88,43 @@ public class KStreamBinderSupportAutoConfiguration { } @Bean - public MessageConversionDelegate messageConversionDelegate(BindingServiceProperties bindingServiceProperties, - CompositeMessageConverterFactory compositeMessageConverterFactory) { - return new MessageConversionDelegate(bindingServiceProperties, compositeMessageConverterFactory); + public KStreamBoundMessageConversionDelegate messageConversionDelegate(CompositeMessageConverterFactory compositeMessageConverterFactory, + SendToDlqAndContinue sendToDlqAndContinue, + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { + return new KStreamBoundMessageConversionDelegate(compositeMessageConverterFactory, sendToDlqAndContinue, + KStreamBindingInformationCatalogue); } @Bean - public KStreamBoundElementFactory kafkaStreamBindableTargetFactory(StreamsBuilder kStreamBuilder, - BindingServiceProperties bindingServiceProperties) { - return new KStreamBoundElementFactory(kStreamBuilder, bindingServiceProperties); + public KStreamBoundElementFactory kafkaStreamBindableTargetFactory(BindingServiceProperties bindingServiceProperties, + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue, + KeyValueSerdeResolver keyValueSerdeResolver, + KStreamExtendedBindingProperties kStreamExtendedBindingProperties) { + KStreamBoundElementFactory kStreamBoundElementFactory = new KStreamBoundElementFactory(bindingServiceProperties, + KStreamBindingInformationCatalogue, keyValueSerdeResolver); + kStreamBoundElementFactory.setkStreamExtendedBindingProperties(kStreamExtendedBindingProperties); + return kStreamBoundElementFactory; + } + + @Bean + public SendToDlqAndContinue kStreamDlqSender() { + return new SendToDlqAndContinue(); + } + + @Bean + public KStreamBindingInformationCatalogue boundedKStreamRegistryService() { + return new KStreamBindingInformationCatalogue(); + } + + @Bean + public KeyValueSerdeResolver keyValueSerdeResolver(@Qualifier("streamConfigGlobalProperties") Map streamConfigGlobalProperties, + KStreamBinderConfigurationProperties kStreamBinderConfigurationProperties) { + return new KeyValueSerdeResolver(streamConfigGlobalProperties, kStreamBinderConfigurationProperties); + } + + @Bean + public QueryableStoreRegistry queryableStoreTypeRegistry() { + return new QueryableStoreRegistry(); } } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamConsumerProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamConsumerProperties.java index 263d2556b..1b595f67f 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamConsumerProperties.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamConsumerProperties.java @@ -16,9 +16,58 @@ package org.springframework.cloud.stream.binder.kstream.config; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; + /** * @author Marius Bogoevici + * @author Soby Chacko */ -public class KStreamConsumerProperties extends KStreamCommonProperties { +public class KStreamConsumerProperties extends KafkaConsumerProperties { + public enum SerdeError { + logAndContinue, + logAndFail, + sendToDlq + } + + /** + * Key serde specified per binding. + */ + private String keySerde; + + /** + * Value serde specified per binding. + */ + private String valueSerde; + + /** + * {@link org.apache.kafka.streams.errors.DeserializationExceptionHandler} to use + * when there is a Serde error. {@link SerdeError} values are used to provide the + * exception handler on consumer binding. + */ + private SerdeError serdeError; + + public String getKeySerde() { + return keySerde; + } + + public void setKeySerde(String keySerde) { + this.keySerde = keySerde; + } + + public String getValueSerde() { + return valueSerde; + } + + public void setValueSerde(String valueSerde) { + this.valueSerde = valueSerde; + } + + public SerdeError getSerdeError() { + return serdeError; + } + + public void setSerdeError(SerdeError serdeError) { + this.serdeError = serdeError; + } } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java index 4946ff850..63191c1e7 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java @@ -16,10 +16,37 @@ package org.springframework.cloud.stream.binder.kstream.config; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; + /** * @author Marius Bogoevici * @author Soby Chacko */ -public class KStreamProducerProperties extends KStreamCommonProperties { +public class KStreamProducerProperties extends KafkaProducerProperties { + /** + * Key serde specified per binding. + */ + private String keySerde; + + /** + * Value serde specified per binding. + */ + private String valueSerde; + + public String getKeySerde() { + return keySerde; + } + + public void setKeySerde(String keySerde) { + this.keySerde = keySerde; + } + + public String getValueSerde() { + return valueSerde; + } + + public void setValueSerde(String valueSerde) { + this.valueSerde = valueSerde; + } } diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/DeserializationErrorHandlerByKafkaTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/DeserializationErrorHandlerByKafkaTests.java new file mode 100644 index 000000000..182990979 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/DeserializationErrorHandlerByKafkaTests.java @@ -0,0 +1,155 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Arrays; +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +import org.springframework.cloud.stream.binder.kstream.config.KStreamApplicationSupportProperties; +import org.springframework.context.annotation.PropertySource; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** + * @author Soby Chacko + */ +@RunWith(SpringRunner.class) +@ContextConfiguration +@DirtiesContext +public abstract class DeserializationErrorHandlerByKafkaTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts", "error.words.group"); + + @SpyBean + KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate; + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + System.setProperty("spring.cloud.stream.kstream.binder.brokers", embeddedKafka.getBrokersAsString()); + System.setProperty("spring.cloud.stream.kstream.binder.zkNodes", embeddedKafka.getZookeeperConnectionString()); + + System.setProperty("server.port","0"); + System.setProperty("spring.jmx.enabled","false"); + + Map consumerProps = KafkaTestUtils.consumerProps("fooc", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @SpringBootTest(properties = { + "spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", + "spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "spring.cloud.stream.bindings.input.group=group", + "spring.cloud.stream.kstream.bindings.input.consumer.serdeError=sendToDlq", + "spring.cloud.stream.kstream.binder.configuration.value.serde=" + + "org.apache.kafka.common.serialization.Serdes$IntegerSerde"}, + webEnvironment= SpringBootTest.WebEnvironment.NONE + ) + public static class DeserializationByKafkaAndDlqTests extends DeserializationErrorHandlerByKafkaTests { + + @Test + @SuppressWarnings("unchecked") + public void test() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words"); + template.sendDefault("foobar"); + + Map consumerProps = KafkaTestUtils.consumerProps("foobar", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + Consumer consumer1 = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer1, "error.words.group"); + + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer1, "error.words.group"); + assertThat(cr.value().equals("foobar")).isTrue(); + + //Ensuring that the deserialization was indeed done by Kafka natively + verify(KStreamBoundMessageConversionDelegate, never()).deserializeOnInbound(any(Class.class), any(KStream.class)); + verify(KStreamBoundMessageConversionDelegate, never()).serializeOnOutbound(any(KStream.class)); + } + } + + + @EnableBinding(KStreamProcessor.class) + @EnableAutoConfiguration + @PropertySource("classpath:/org/springframework/cloud/stream/binder/kstream/integTest-1.properties") + @EnableConfigurationProperties(KStreamApplicationSupportProperties.class) + static class WordCountProcessorApplication { + + @Autowired + private TimeWindows timeWindows; + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + + return input + .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) + .map((key, value) -> new KeyValue<>(value, value)) + .groupByKey(Serdes.String(), Serdes.String()) + .count(timeWindows, "foo-WordCounts-x") + .toStream() + .map((key, value) -> new KeyValue<>(null, "Count for " + key.key() + " : " + value)); + } + + } +} diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/DeserializtionErrorHandlerByBinderTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/DeserializtionErrorHandlerByBinderTests.java new file mode 100644 index 000000000..e3213aeaa --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/DeserializtionErrorHandlerByBinderTests.java @@ -0,0 +1,157 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.serializer.JsonSerde; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.verify; + +/** + * @author Soby Chacko + */ +@RunWith(SpringRunner.class) +@ContextConfiguration +@DirtiesContext +public abstract class DeserializtionErrorHandlerByBinderTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts-id", "error.foos.foobar-group"); + + @SpyBean + KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate; + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + System.setProperty("spring.cloud.stream.kstream.binder.brokers", embeddedKafka.getBrokersAsString()); + System.setProperty("spring.cloud.stream.kstream.binder.zkNodes", embeddedKafka.getZookeeperConnectionString()); + + System.setProperty("server.port","0"); + System.setProperty("spring.jmx.enabled","false"); + + Map consumerProps = KafkaTestUtils.consumerProps("foob", "false", embeddedKafka); + //consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Deserializer.class.getName()); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @SpringBootTest(properties = { + "spring.cloud.stream.bindings.input.destination=foos", + "spring.cloud.stream.bindings.output.destination=counts-id", + "spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", + "spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "spring.cloud.stream.bindings.output.producer.headerMode=raw", + "spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde", + "spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "spring.cloud.stream.kstream.bindings.input.consumer.serdeError=sendToDlq", + "spring.cloud.stream.bindings.input.group=foobar-group"}, + webEnvironment= SpringBootTest.WebEnvironment.NONE + ) + public static class DeserializationByBinderAndDlqTests extends DeserializtionErrorHandlerByBinderTests { + + @Test + @SuppressWarnings("unchecked") + public void test() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foos"); + template.sendDefault("hello"); + + Map consumerProps = KafkaTestUtils.consumerProps("foobar", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + Consumer consumer1 = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer1, "error.foos.foobar-group"); + + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer1, "error.foos.foobar-group"); + assertThat(cr.value().equals("hello")).isTrue(); + + //Ensuring that the deserialization was indeed done by the binder + verify(KStreamBoundMessageConversionDelegate).deserializeOnInbound(any(Class.class), any(KStream.class)); + } + } + + @EnableBinding(KStreamProcessor.class) + @EnableAutoConfiguration + public static class ProductCountApplication { + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + return input + .filter((key, product) -> product.getId() == 123) + .map((key, value) -> new KeyValue<>(value, value)) + .groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class)) + .count(TimeWindows.of(5000), "id-count-store-x") + .toStream() + .map((key, value) -> new KeyValue<>(key.key().id, value)); + } + } + static class Product { + + Integer id; + + public Integer getId() { + return id; + } + + public void setId(Integer id) { + this.id = id; + } + } +} diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java index ad96cae26..8f5fe4430 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java @@ -77,6 +77,7 @@ public class KStreamBinderPojoInputAndPrimitiveTypeOutputTests { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebEnvironment(false); ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.input.destination=foos", "--spring.cloud.stream.bindings.output.destination=counts-id", "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", @@ -119,7 +120,7 @@ public class KStreamBinderPojoInputAndPrimitiveTypeOutputTests { .filter((key, product) -> product.getId() == 123) .map((key, value) -> new KeyValue<>(value, value)) .groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class)) - .count(TimeWindows.of(5000), "id-count-store") + .count(TimeWindows.of(5000), "id-count-store-x") .toStream() .map((key, value) -> new KeyValue<>(key.key().id, value)); } diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java index 9d75e7f10..21505ae4e 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java @@ -82,6 +82,7 @@ public class KStreamBinderWordCountIntegrationTests { app.setWebEnvironment(false); 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", diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java index ff76fe95e..b2e262bac 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java @@ -22,7 +22,6 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.state.QueryableStoreTypes; @@ -43,7 +42,6 @@ import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.kafka.support.serializer.JsonSerde; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -81,6 +79,7 @@ public class KStreamInteractiveQueryIntegrationTests { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebEnvironment(false); ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.input.destination=foos", "--spring.cloud.stream.bindings.output.destination=counts-id", "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", @@ -115,7 +114,7 @@ public class KStreamInteractiveQueryIntegrationTests { public static class ProductCountApplication { @Autowired - private StreamsBuilderFactoryBean kStreamBuilderFactoryBean; + private QueryableStoreRegistry queryableStoreRegistry; @StreamListener("input") @SendTo("output") @@ -131,21 +130,23 @@ public class KStreamInteractiveQueryIntegrationTests { } @Bean - public Foo foo(StreamsBuilderFactoryBean kStreamBuilderFactoryBean) { - return new Foo(kStreamBuilderFactoryBean); + public Foo foo(QueryableStoreRegistry queryableStoreRegistry) { + return new Foo(queryableStoreRegistry); } static class Foo { - StreamsBuilderFactoryBean kStreamBuilderFactoryBean; + QueryableStoreRegistry queryableStoreRegistry; - Foo(StreamsBuilderFactoryBean kStreamBuilderFactoryBean) { - this.kStreamBuilderFactoryBean = kStreamBuilderFactoryBean; + Foo(QueryableStoreRegistry queryableStoreRegistry) { + this.queryableStoreRegistry = queryableStoreRegistry; } public Long getProductStock(Integer id) { - KafkaStreams streams = kStreamBuilderFactoryBean.getKafkaStreams(); + + ReadOnlyKeyValueStore keyValueStore = - streams.store("prod-id-count-store", QueryableStoreTypes.keyValueStore()); + queryableStoreRegistry.getQueryableStoreType("prod-id-count-store", QueryableStoreTypes.keyValueStore()); + return (Long) keyValueStore.get(id); } } diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamsNativeEncodingDecodingTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamsNativeEncodingDecodingTests.java new file mode 100644 index 000000000..e2496af7c --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamsNativeEncodingDecodingTests.java @@ -0,0 +1,159 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Arrays; +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +import org.springframework.cloud.stream.binder.kstream.config.KStreamApplicationSupportProperties; +import org.springframework.context.annotation.PropertySource; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** + * @author Soby Chacko + */ +@RunWith(SpringRunner.class) +@ContextConfiguration +@DirtiesContext +public abstract class KStreamsNativeEncodingDecodingTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts"); + + @SpyBean + KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate; + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + System.setProperty("spring.cloud.stream.kstream.binder.brokers", embeddedKafka.getBrokersAsString()); + System.setProperty("spring.cloud.stream.kstream.binder.zkNodes", embeddedKafka.getZookeeperConnectionString()); + + System.setProperty("server.port","0"); + System.setProperty("spring.jmx.enabled","false"); + + Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @SpringBootTest(properties = { + "spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", + "spring.cloud.stream.bindings.output.producer.useNativeEncoding=true"}, + webEnvironment= SpringBootTest.WebEnvironment.NONE + ) + public static class NativeEncodingDecodingEnabledTests extends KStreamsNativeEncodingDecodingTests { + + @Test + public void test() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words"); + template.sendDefault("foobar"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts"); + assertThat(cr.value().equals("Count for foobar : 1")).isTrue(); + + verify(KStreamBoundMessageConversionDelegate, never()).serializeOnOutbound(any(KStream.class)); + verify(KStreamBoundMessageConversionDelegate, never()).deserializeOnInbound(any(Class.class), any(KStream.class)); + } + } + + @SpringBootTest(webEnvironment= SpringBootTest.WebEnvironment.NONE) + public static class NativeEncodingDecodingDisabledTests extends KStreamsNativeEncodingDecodingTests { + + @Test + public void test() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words"); + template.sendDefault("foobar"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts"); + assertThat(cr.value().equals("Count for foobar : 1")).isTrue(); + + verify(KStreamBoundMessageConversionDelegate).serializeOnOutbound(any(KStream.class)); + verify(KStreamBoundMessageConversionDelegate).deserializeOnInbound(any(Class.class), any(KStream.class)); + } + } + + @EnableBinding(KStreamProcessor.class) + @EnableAutoConfiguration + @PropertySource("classpath:/org/springframework/cloud/stream/binder/kstream/integTest-1.properties") + @EnableConfigurationProperties(KStreamApplicationSupportProperties.class) + static class WordCountProcessorApplication { + + @Autowired + private TimeWindows timeWindows; + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + + return input + .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) + .map((key, value) -> new KeyValue<>(value, value)) + .groupByKey(Serdes.String(), Serdes.String()) + .count(timeWindows, "foo-WordCounts-x") + .toStream() + .map((key, value) -> new KeyValue<>(null, "Count for " + key.key() + " : " + value)); + } + } + +} diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java index 626747600..4af354856 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java @@ -76,6 +76,7 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebEnvironment(false); ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.input.destination=foos", "--spring.cloud.stream.bindings.output.destination=counts-id", "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/WordCountMultipleBranchesIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/WordCountMultipleBranchesIntegrationTests.java index 397e04bce..4988df99a 100644 --- a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/WordCountMultipleBranchesIntegrationTests.java +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/WordCountMultipleBranchesIntegrationTests.java @@ -66,7 +66,7 @@ public class WordCountMultipleBranchesIntegrationTests { @BeforeClass public static void setUp() throws Exception { - Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps("groupx", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); consumer = cf.createConsumer(); @@ -84,6 +84,7 @@ public class WordCountMultipleBranchesIntegrationTests { app.setWebEnvironment(false); ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.input.destination=words", "--spring.cloud.stream.bindings.output1.destination=counts", "--spring.cloud.stream.bindings.output1.contentType=application/json", diff --git a/spring-cloud-stream-binder-kstream/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties b/spring-cloud-stream-binder-kstream/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties new file mode 100644 index 000000000..8d7167834 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties @@ -0,0 +1,10 @@ +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.kstream.binder.configuration.commit.interval.ms=1000 +spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde +spring.cloud.stream.kstream.binder.configuration.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.kstream.timeWindow.length=5000 +spring.cloud.stream.kstream.timeWindow.advanceBy=0