diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java index 5a0c47152..b7d87563c 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-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,17 +16,14 @@ package org.springframework.cloud.stream.binder.kstream; -import java.util.Map; - +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.errors.DeserializationExceptionHandler; -import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler; -import org.apache.kafka.streams.errors.LogAndFailExceptionHandler; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Produced; -import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.binder.AbstractBinder; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.DefaultBinding; @@ -40,7 +37,6 @@ import org.springframework.cloud.stream.binder.kstream.config.KStreamBinderConfi import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties; import org.springframework.cloud.stream.binder.kstream.config.KStreamExtendedBindingProperties; import org.springframework.cloud.stream.binder.kstream.config.KStreamProducerProperties; -import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.util.StringUtils; /** @@ -51,6 +47,8 @@ public class KStreamBinder extends AbstractBinder, ExtendedConsumerProperties, ExtendedProducerProperties> implements ExtendedPropertiesBinder, KStreamConsumerProperties, KStreamProducerProperties> { + private final static Log LOG = LogFactory.getLog(KStreamBinder.class); + private final KafkaTopicProvisioner kafkaTopicProvisioner; private KStreamExtendedBindingProperties kStreamExtendedBindingProperties = new KStreamExtendedBindingProperties(); @@ -63,20 +61,16 @@ public class KStreamBinder extends private final KeyValueSerdeResolver keyValueSerdeResolver; - private final QueryableStoreRegistry queryableStoreRegistry; - public KStreamBinder(KStreamBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, KStreamBoundMessageConversionDelegate kStreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue, - KeyValueSerdeResolver keyValueSerdeResolver, - QueryableStoreRegistry queryableStoreRegistry) { + KeyValueSerdeResolver keyValueSerdeResolver) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; this.kStreamBoundMessageConversionDelegate = kStreamBoundMessageConversionDelegate; this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue; this.keyValueSerdeResolver = keyValueSerdeResolver; - this.queryableStoreRegistry = queryableStoreRegistry; } @Override @@ -84,67 +78,17 @@ public class KStreamBinder extends protected Binding> doBindConsumer(String name, String group, KStream inputTarget, ExtendedConsumerProperties properties) { - this.KStreamBindingInformationCatalogue.registerConsumerProperties(inputTarget, properties.getExtension()); ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( properties.getExtension()); - if (properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.sendToDlq) { + if (binderConfigurationProperties.getSerdeError() == KStreamBinderConfigurationProperties.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()); - + StreamsConfig streamsConfig = this.KStreamBindingInformationCatalogue.getStreamsConfig(inputTarget); if (extendedConsumerProperties.getExtension().isEnableDlq()) { String dlqName = StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName()) ? "error." + name + "." + group : extendedConsumerProperties.getExtension().getDlqName(); @@ -168,12 +112,9 @@ public class KStreamBinder extends ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>( new KafkaProducerProperties()); this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties); - 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); } @@ -181,10 +122,12 @@ public class KStreamBinder extends private void to(boolean isNativeEncoding, String name, KStream outboundBindTarget, Serde keySerde, Serde valueSerde) { if (!isNativeEncoding) { + LOG.info("Native encoding is disabled for " + name + ". Outbound message conversion done by Spring Cloud Stream."); kStreamBoundMessageConversionDelegate.serializeOnOutbound(outboundBindTarget) .to(name, Produced.with(keySerde, valueSerde)); } else { + LOG.info("Native encoding is enabled for " + name + ". Outbound serialization done at the broker."); outboundBindTarget.to(name, Produced.with(keySerde, valueSerde)); } } 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 index 277b86bd8..1e37e7ff1 100644 --- 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 @@ -16,13 +16,19 @@ package org.springframework.cloud.stream.binder.kstream; +import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.kstream.KStream; +import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties; import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; /** * A catalogue containing all the inbound and outboud KStreams. @@ -40,6 +46,9 @@ public class KStreamBindingInformationCatalogue { private final Map, BindingProperties> bindingProperties = new ConcurrentHashMap<>(); private final Map, KStreamConsumerProperties> consumerProperties = new ConcurrentHashMap<>(); + private final Map streamsConfigs = new HashMap<>(); + private final Set streamsBuilderFactoryBeans = new HashSet<>(); + /** * For a given bounded {@link KStream}, retrieve it's corresponding destination * on the broker. @@ -60,6 +69,9 @@ public class KStreamBindingInformationCatalogue { */ public boolean isUseNativeDecoding(KStream bindingTarget) { BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget); + if (bindingProperties.getConsumer() == null) { + bindingProperties.setConsumer(new ConsumerProperties()); + } return bindingProperties.getConsumer().isUseNativeDecoding(); } @@ -69,7 +81,7 @@ public class KStreamBindingInformationCatalogue { * @param bindingTarget KStream binding target * @return true if DLQ is enabled, false otherwise. */ - public boolean isEnableDlq(KStream bindingTarget) { + public boolean isDlqEnabled(KStream bindingTarget) { return consumerProperties.get(bindingTarget).isEnableDlq(); } @@ -85,13 +97,13 @@ public class KStreamBindingInformationCatalogue { } /** - * Retrieve any configured Serde error handling strategies for this {@link KStream} + * Retrieve and return the registered {@link StreamsBuilderFactoryBean} for the given KStream * * @param bindingTarget KStream binding target - * @return configured Serde error handling strategy + * @return corresponding {@link StreamsBuilderFactoryBean} */ - public KStreamConsumerProperties.SerdeError getSerdeError(KStream bindingTarget) { - return consumerProperties.get(bindingTarget).getSerdeError(); + public StreamsConfig getStreamsConfig(Object bindingTarget) { + return streamsConfigs.get(bindingTarget); } /** @@ -114,4 +126,20 @@ public class KStreamBindingInformationCatalogue { this.consumerProperties.put(bindingTarget, kStreamConsumerProperties); } + /** + * Adds a mapping for KStream -> {@link StreamsBuilderFactoryBean} + * + * @param streamsBuilderFactoryBean provides the {@link StreamsBuilderFactoryBean} mapped to the KStream + */ + public void addStreamBuilderFactory(StreamsBuilderFactoryBean streamsBuilderFactoryBean) { + this.streamsBuilderFactoryBeans.add(streamsBuilderFactoryBean); + } + + public void addStreamsConfigs(Object bindingTarget, StreamsConfig streamsConfig) { + this.streamsConfigs.put(bindingTarget, streamsConfig); + } + + public Set getStreamsBuilderFactoryBeans() { + return streamsBuilderFactoryBeans; + } } 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 a17166872..ac96698d9 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,114 +18,43 @@ 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; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; /** * @author Marius Bogoevici * @author Soby Chacko */ -public class KStreamBoundElementFactory extends AbstractBindingTargetFactory implements ApplicationContextAware { +public class KStreamBoundElementFactory extends AbstractBindingTargetFactory { private final 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) { + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { super(KStream.class); this.bindingServiceProperties = bindingServiceProperties; 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) { - - 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; - String contentType = bindingProperties.getContentType(); - if (!StringUtils.isEmpty(contentType) && !bindingProperties.getConsumer().isUseNativeDecoding()) { - Message message = MessageBuilder.withPayload(value) - .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); - keyValue = new KeyValue<>(key, message); - } - else { - keyValue = new KeyValue<>(key, value); - } - return keyValue; - }); - this.KStreamBindingInformationCatalogue.registerBindingProperties(stream, bindingProperties); - return stream; + return createProxyForKStream(name); } @Override @SuppressWarnings("unchecked") public KStream createOutput(final String name) { + return createProxyForKStream(name); + } + + private KStream createProxyForKStream(String name) { KStreamWrapperHandler wrapper= new KStreamWrapperHandler(); ProxyFactory proxyFactory = new ProxyFactory(KStreamWrapper.class, KStream.class); proxyFactory.addAdvice(wrapper); @@ -140,6 +69,7 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory delegate); + } private static class KStreamWrapperHandler implements KStreamWrapper, MethodInterceptor { diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundMessageConversionDelegate.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundMessageConversionDelegate.java index 05e8601bd..bca7ea66e 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundMessageConversionDelegate.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundMessageConversionDelegate.java @@ -24,7 +24,7 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.processor.Processor; import org.apache.kafka.streams.processor.ProcessorContext; -import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties; +import org.springframework.cloud.stream.binder.kstream.config.KStreamBinderConfigurationProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; @@ -52,14 +52,18 @@ public class KStreamBoundMessageConversionDelegate { private final SendToDlqAndContinue sendToDlqAndContinue; - private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue; + private final KStreamBindingInformationCatalogue kstreamBindingInformationCatalogue; + + private final KStreamBinderConfigurationProperties kstreamBinderConfigurationProperties; public KStreamBoundMessageConversionDelegate(CompositeMessageConverterFactory compositeMessageConverterFactory, SendToDlqAndContinue sendToDlqAndContinue, - KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { + KStreamBindingInformationCatalogue kstreamBindingInformationCatalogue, + KStreamBinderConfigurationProperties kstreamBinderConfigurationProperties) { this.compositeMessageConverterFactory = compositeMessageConverterFactory; this.sendToDlqAndContinue = sendToDlqAndContinue; - this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue; + this.kstreamBindingInformationCatalogue = kstreamBindingInformationCatalogue; + this.kstreamBinderConfigurationProperties = kstreamBinderConfigurationProperties; } /** @@ -69,7 +73,7 @@ public class KStreamBoundMessageConversionDelegate { * @return serialized KStream */ public KStream serializeOnOutbound(KStream outboundBindTarget) { - String contentType = this.KStreamBindingInformationCatalogue.getContentType(outboundBindTarget); + String contentType = this.kstreamBindingInformationCatalogue.getContentType(outboundBindTarget); MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory .getMessageConverterForType(MimeType.valueOf(contentType)) : null; @@ -125,6 +129,7 @@ public class KStreamBoundMessageConversionDelegate { isValidRecord = true; } catch (Exception ignored) { + System.out.println(); //pass through } return isValidRecord; @@ -160,8 +165,8 @@ public class KStreamBoundMessageConversionDelegate { @Override public void process(Object o, Object o2) { - if (KStreamBindingInformationCatalogue.isEnableDlq(bindingTarget)) { - String destination = KStreamBindingInformationCatalogue.getDestination(bindingTarget); + if (kstreamBindingInformationCatalogue.isDlqEnabled(bindingTarget)) { + String destination = kstreamBindingInformationCatalogue.getDestination(bindingTarget); if (o2 instanceof Message) { Message message = (Message) o2; sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) message.getPayload(), context.partition()); @@ -170,10 +175,10 @@ public class KStreamBoundMessageConversionDelegate { sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) o2, context.partition()); } } - else if (KStreamBindingInformationCatalogue.getSerdeError(bindingTarget) == KStreamConsumerProperties.SerdeError.logAndFail) { + else if (kstreamBinderConfigurationProperties.getSerdeError() == KStreamBinderConfigurationProperties.SerdeError.logAndFail) { throw new IllegalStateException("Inbound deserialization failed."); } - else if (KStreamBindingInformationCatalogue.getSerdeError(bindingTarget) == KStreamConsumerProperties.SerdeError.logAndContinue) { + else if (kstreamBinderConfigurationProperties.getSerdeError() == KStreamBinderConfigurationProperties.SerdeError.logAndContinue) { //quietly pass through. No action needed, this is similar to log and continue. } } 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 a131c4523..8f1cf1b18 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 @@ -18,22 +18,52 @@ package org.springframework.cloud.stream.binder.kstream; import java.lang.reflect.Method; import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.Consumed; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.errors.DeserializationExceptionHandler; import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.state.KeyValueStore; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.ConsumerProperties; +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.KStreamExtendedBindingProperties; import org.springframework.cloud.stream.binding.StreamListenerErrorMessages; import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter; import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; import org.springframework.cloud.stream.binding.StreamListenerSetupMethodOrchestrator; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.MethodParameter; import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -48,20 +78,39 @@ import org.springframework.util.StringUtils; */ public class KStreamListenerSetupMethodOrchestrator implements StreamListenerSetupMethodOrchestrator, ApplicationContextAware { + private final static Log LOG = LogFactory.getLog(KStreamListenerSetupMethodOrchestrator.class); + + private final StreamListenerParameterAdapter streamListenerParameterAdapter; + private final Collection streamListenerResultAdapters; + private final BindingServiceProperties bindingServiceProperties; + private final KStreamExtendedBindingProperties kStreamExtendedBindingProperties; + private final KeyValueSerdeResolver keyValueSerdeResolver; + private final KStreamBindingInformationCatalogue kStreamBindingInformationCatalogue; + private final Map methodStreamsBuilderFactoryBeanMap = new HashMap<>(); + private final KStreamBinderConfigurationProperties binderConfigurationProperties; + private ConfigurableApplicationContext applicationContext; - private StreamListenerParameterAdapter streamListenerParameterAdapter; - private Collection streamListenerResultAdapters; - - public KStreamListenerSetupMethodOrchestrator(StreamListenerParameterAdapter streamListenerParameterAdapter, - Collection streamListenerResultAdapters) { + public KStreamListenerSetupMethodOrchestrator(BindingServiceProperties bindingServiceProperties, + KStreamExtendedBindingProperties kStreamExtendedBindingProperties, + KeyValueSerdeResolver keyValueSerdeResolver, + KStreamBindingInformationCatalogue kStreamBindingInformationCatalogue, + StreamListenerParameterAdapter streamListenerParameterAdapter, + Collection streamListenerResultAdapters, + KStreamBinderConfigurationProperties binderConfigurationProperties) { + this.bindingServiceProperties = bindingServiceProperties; + this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties; + this.keyValueSerdeResolver = keyValueSerdeResolver; + this.kStreamBindingInformationCatalogue = kStreamBindingInformationCatalogue; this.streamListenerParameterAdapter = streamListenerParameterAdapter; this.streamListenerResultAdapters = streamListenerResultAdapters; + this.binderConfigurationProperties = binderConfigurationProperties; } @Override public boolean supports(Method method) { - return methodParameterSuppports(method) && methodReturnTypeSuppports(method); + return methodParameterSuppports(method) && + (methodReturnTypeSuppports(method) || Void.TYPE.equals(method.getReturnType())); } private boolean methodReturnTypeSuppports(Method method) { @@ -74,9 +123,15 @@ public class KStreamListenerSetupMethodOrchestrator implements StreamListenerSet } private boolean methodParameterSuppports(Method method) { - MethodParameter methodParameter = MethodParameter.forExecutable(method, 0); - Class parameterType = methodParameter.getParameterType(); - return parameterType.equals(KStream.class); + boolean supports = false; + for (int i = 0; i < method.getParameterCount(); i++) { + MethodParameter methodParameter = MethodParameter.forExecutable(method, i); + Class parameterType = methodParameter.getParameterType(); + if (parameterType.equals(KStream.class) || parameterType.equals(KTable.class)) { + supports = true; + } + } + return supports; } @Override @@ -84,39 +139,43 @@ public class KStreamListenerSetupMethodOrchestrator implements StreamListenerSet public void orchestrateStreamListenerSetupMethod(StreamListener streamListener, Method method, Object bean) { String[] methodAnnotatedOutboundNames = getOutboundBindingTargetNames(method); validateStreamListenerMethod(streamListener, method, methodAnnotatedOutboundNames); - String methodAnnotatedInboundName = streamListener.value(); Object[] adaptedInboundArguments = adaptAndRetrieveInboundArguments(method, methodAnnotatedInboundName, this.applicationContext, this.streamListenerParameterAdapter); - try { - Object result = method.invoke(bean, adaptedInboundArguments); - - if (result.getClass().isArray()) { - Assert.isTrue(methodAnnotatedOutboundNames.length == ((Object[]) result).length, "Big error"); - } else { - Assert.isTrue(methodAnnotatedOutboundNames.length == 1, "Big error"); - } - if (result.getClass().isArray()) { - Object[] outboundKStreams = (Object[]) result; - int i = 0; - for (Object outboundKStream : outboundKStreams) { - Object targetBean = this.applicationContext.getBean(methodAnnotatedOutboundNames[i++]); - for (StreamListenerResultAdapter streamListenerResultAdapter : streamListenerResultAdapters) { - if (streamListenerResultAdapter.supports(outboundKStream.getClass(), targetBean.getClass())) { - streamListenerResultAdapter.adapt(outboundKStream, targetBean); - break; - } - } - } + if (Void.TYPE.equals(method.getReturnType())) { + method.invoke(bean, adaptedInboundArguments); } else { - Object targetBean = this.applicationContext.getBean(methodAnnotatedOutboundNames[0]); - for (StreamListenerResultAdapter streamListenerResultAdapter : streamListenerResultAdapters) { - if (streamListenerResultAdapter.supports(result.getClass(), targetBean.getClass())) { - streamListenerResultAdapter.adapt(result, targetBean); - break; + Object result = method.invoke(bean, adaptedInboundArguments); + + if (result.getClass().isArray()) { + Assert.isTrue(methodAnnotatedOutboundNames.length == ((Object[]) result).length, + "Result does not match with the number of declared outbounds"); + } else { + Assert.isTrue(methodAnnotatedOutboundNames.length == 1, + "Result does not match with the number of declared outbounds"); + } + if (result.getClass().isArray()) { + Object[] outboundKStreams = (Object[]) result; + int i = 0; + for (Object outboundKStream : outboundKStreams) { + Object targetBean = this.applicationContext.getBean(methodAnnotatedOutboundNames[i++]); + for (StreamListenerResultAdapter streamListenerResultAdapter : streamListenerResultAdapters) { + if (streamListenerResultAdapter.supports(outboundKStream.getClass(), targetBean.getClass())) { + streamListenerResultAdapter.adapt(outboundKStream, targetBean); + break; + } + } + } + } else { + Object targetBean = this.applicationContext.getBean(methodAnnotatedOutboundNames[0]); + for (StreamListenerResultAdapter streamListenerResultAdapter : streamListenerResultAdapters) { + if (streamListenerResultAdapter.supports(result.getClass(), targetBean.getClass())) { + streamListenerResultAdapter.adapt(result, targetBean); + break; + } } } } @@ -126,6 +185,176 @@ public class KStreamListenerSetupMethodOrchestrator implements StreamListenerSet } } + @Override + @SuppressWarnings({"unchecked"}) + public Object[] adaptAndRetrieveInboundArguments(Method method, String inboundName, + ApplicationContext applicationContext, + StreamListenerParameterAdapter... streamListenerParameterAdapters) { + Object[] arguments = new Object[method.getParameterTypes().length]; + for (int parameterIndex = 0; parameterIndex < arguments.length; parameterIndex++) { + MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); + Class parameterType = methodParameter.getParameterType(); + Object targetReferenceValue = null; + if (methodParameter.hasParameterAnnotation(Input.class)) { + targetReferenceValue = AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Input.class)); + Input methodAnnotation = methodParameter.getParameterAnnotation(Input.class); + inboundName = methodAnnotation.value(); + } + else if (arguments.length == 1 && StringUtils.hasText(inboundName)) { + targetReferenceValue = inboundName; + } + if (targetReferenceValue != null) { + Assert.isInstanceOf(String.class, targetReferenceValue, "Annotation value must be a String"); + Object targetBean = applicationContext.getBean((String) targetReferenceValue); + BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(inboundName); + enableNativeDecodingForKTableAlways(parameterType, bindingProperties); + StreamsConfig streamsConfig = null; + if (!methodStreamsBuilderFactoryBeanMap.containsKey(method)) { + streamsConfig = buildStreamsBuilderAndRetrieveConfig(method, applicationContext, bindingProperties); + } + try { + StreamsBuilderFactoryBean streamsBuilderFactoryBean = methodStreamsBuilderFactoryBeanMap.get(method); + StreamsBuilder streamsBuilder = streamsBuilderFactoryBean.getObject(); + KStreamConsumerProperties extendedConsumerProperties = kStreamExtendedBindingProperties.getExtendedConsumerProperties(inboundName); + Serde keySerde = this.keyValueSerdeResolver.getInboundKeySerde(extendedConsumerProperties); + Serde valueSerde = this.keyValueSerdeResolver.getInboundValueSerde(bindingProperties.getConsumer(), extendedConsumerProperties); + if (parameterType.isAssignableFrom(KStream.class)) { + KStream stream = getkStream(inboundName, bindingProperties, streamsBuilder, keySerde, valueSerde); + KStreamBoundElementFactory.KStreamWrapper kStreamWrapper = (KStreamBoundElementFactory.KStreamWrapper) targetBean; + kStreamWrapper.wrap((KStream) stream); + kStreamBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean); + if (streamsConfig != null){ + kStreamBindingInformationCatalogue.addStreamsConfigs(kStreamWrapper, streamsConfig); + } + // Iterate existing parameter adapters first + for (StreamListenerParameterAdapter streamListenerParameterAdapter : streamListenerParameterAdapters) { + if (streamListenerParameterAdapter.supports(stream.getClass(), methodParameter)) { + arguments[parameterIndex] = streamListenerParameterAdapter.adapt(kStreamWrapper, methodParameter); + break; + } + } + if (arguments[parameterIndex] == null && parameterType.isAssignableFrom(stream.getClass())) { + arguments[parameterIndex] = stream; + } + Assert.notNull(arguments[parameterIndex], "Cannot convert argument " + parameterIndex + " of " + method + + "from " + stream.getClass() + " to " + parameterType); + } + else if (parameterType.isAssignableFrom(KTable.class)) { + String materializedAs = extendedConsumerProperties.getMaterializedAs(); + String bindingDestination = bindingServiceProperties.getBindingDestination(inboundName); + KTable table = materializedAs != null ? + materializedAs(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde ) : + streamsBuilder.table(bindingDestination, + Consumed.with(keySerde, valueSerde)); + KTableBoundElementFactory.KTableWrapper kTableWrapper = (KTableBoundElementFactory.KTableWrapper) targetBean; + kTableWrapper.wrap((KTable) table); + kStreamBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean); + if (streamsConfig != null){ + kStreamBindingInformationCatalogue.addStreamsConfigs(kTableWrapper, streamsConfig); + } + arguments[parameterIndex] = table; + } + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + else { + throw new IllegalStateException(StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS); + } + } + return arguments; + } + + private KTable materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v) { + return streamsBuilder.table(bindingServiceProperties.getBindingDestination(destination), + Materialized.>as(storeName) + .withKeySerde(k) + .withValueSerde(v)); + } + + private KStream getkStream(String inboundName, BindingProperties bindingProperties, StreamsBuilder streamsBuilder, + Serde keySerde, Serde valueSerde) { + KStream stream = streamsBuilder.stream(bindingServiceProperties.getBindingDestination(inboundName), + Consumed.with(keySerde, valueSerde)); + if (bindingProperties.getConsumer().isUseNativeDecoding()){ + LOG.info("Native decoding is enabled for " + inboundName + ". Inbound deserialization done at the broker."); + } + else { + LOG.info("Native decoding is disabled for " + inboundName + ". Inbound message conversion done by Spring Cloud Stream."); + } + stream = stream.map((key, value) -> { + KeyValue keyValue; + String contentType = bindingProperties.getContentType(); + if (!StringUtils.isEmpty(contentType) && !bindingProperties.getConsumer().isUseNativeDecoding()) { + Message message = MessageBuilder.withPayload(value) + .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); + keyValue = new KeyValue<>(key, message); + } + else { + keyValue = new KeyValue<>(key, value); + } + return keyValue; + }); + return stream; + } + + private void enableNativeDecodingForKTableAlways(Class parameterType, BindingProperties bindingProperties) { + if (parameterType.isAssignableFrom(KTable.class)) { + if (bindingProperties.getConsumer() == null) { + bindingProperties.setConsumer(new ConsumerProperties()); + } + //No framework level message conversion provided for KTable, its done by the broker. + bindingProperties.getConsumer().setUseNativeDecoding(true); + } + } + + @SuppressWarnings({"unchecked"}) + private StreamsConfig buildStreamsBuilderAndRetrieveConfig(Method method, ApplicationContext applicationContext, + BindingProperties bindingProperties) { + ConfigurableListableBeanFactory beanFactory = this.applicationContext.getBeanFactory(); + StreamsBuilderFactoryBean streamsBuilder = new StreamsBuilderFactoryBean(); + streamsBuilder.setAutoStartup(false); + String uuid = UUID.randomUUID().toString(); + BeanDefinition streamsBuilderBeanDefinition = + BeanDefinitionBuilder.genericBeanDefinition((Class) streamsBuilder.getClass(), () -> streamsBuilder) + .getRawBeanDefinition(); + ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("stream-builder-" + uuid, streamsBuilderBeanDefinition); + StreamsBuilderFactoryBean streamsBuilderX = applicationContext.getBean("&stream-builder-" + uuid, StreamsBuilderFactoryBean.class); + String group = bindingProperties.getGroup(); + if (!StringUtils.hasText(group)) { + group = binderConfigurationProperties.getApplicationId(); + } + Map streamConfigGlobalProperties = applicationContext.getBean("streamConfigGlobalProperties", Map.class); + streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, group); + StreamsConfig streamsConfig = new StreamsConfig(streamConfigGlobalProperties) { + DeserializationExceptionHandler deserializationExceptionHandler; + @Override + @SuppressWarnings("unchecked") + public T getConfiguredInstance(String key, Class clazz) { + if (key.equals(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG)){ + if (deserializationExceptionHandler != null){ + return (T)deserializationExceptionHandler; + } + else { + T t = super.getConfiguredInstance(key, clazz); + deserializationExceptionHandler = (DeserializationExceptionHandler)t; + return t; + } + } + return super.getConfiguredInstance(key, clazz); + } + }; + BeanDefinition streamsConfigBeanDefinition = + BeanDefinitionBuilder.genericBeanDefinition((Class) streamsConfig.getClass(), () -> streamsConfig) + .getRawBeanDefinition(); + ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("streamsConfig-" + uuid, streamsConfigBeanDefinition); + + streamsBuilder.setStreamsConfig(streamsConfig); + methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderX); + return streamsConfig; + } + @Override public final void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; @@ -133,9 +362,11 @@ public class KStreamListenerSetupMethodOrchestrator implements StreamListenerSet private void validateStreamListenerMethod(StreamListener streamListener, Method method, String[] methodAnnotatedOutboundNames) { String methodAnnotatedInboundName = streamListener.value(); - for (String s : methodAnnotatedOutboundNames) { - if (StringUtils.hasText(s)) { - Assert.isTrue(isDeclarativeOutput(method, s), "Method must be declarative"); + if (methodAnnotatedOutboundNames != null) { + for (String s : methodAnnotatedOutboundNames) { + if (StringUtils.hasText(s)) { + Assert.isTrue(isDeclarativeOutput(method, s), "Method must be declarative"); + } } } if (StringUtils.hasText(methodAnnotatedInboundName)) { diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KTableBinder.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KTableBinder.java new file mode 100644 index 000000000..534aaa502 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KTableBinder.java @@ -0,0 +1,108 @@ +/* + * 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 org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.errors.DeserializationExceptionHandler; +import org.apache.kafka.streams.kstream.KTable; + +import org.springframework.cloud.stream.binder.AbstractBinder; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.DefaultBinding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +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.KStreamExtendedBindingProperties; +import org.springframework.cloud.stream.binder.kstream.config.KStreamProducerProperties; +import org.springframework.util.StringUtils; + +/** + * + * @since 2.0.0 + * + * @author Soby Chacko + */ +public class KTableBinder extends + AbstractBinder, ExtendedConsumerProperties, ExtendedProducerProperties> + implements ExtendedPropertiesBinder, KStreamConsumerProperties, KStreamProducerProperties> { + + private final KStreamBinderConfigurationProperties binderConfigurationProperties; + + private final KafkaTopicProvisioner kafkaTopicProvisioner; + + private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue; + + private KStreamExtendedBindingProperties kStreamExtendedBindingProperties = new KStreamExtendedBindingProperties(); + + public KTableBinder(KStreamBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, + KStreamBindingInformationCatalogue kStreamBindingInformationCatalogue) { + this.binderConfigurationProperties = binderConfigurationProperties; + this.kafkaTopicProvisioner = kafkaTopicProvisioner; + KStreamBindingInformationCatalogue = kStreamBindingInformationCatalogue; + } + + @Override + @SuppressWarnings("unchecked") + protected Binding> doBindConsumer(String name, String group, KTable inputTarget, + ExtendedConsumerProperties properties) { + ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( + properties.getExtension()); + if (binderConfigurationProperties.getSerdeError() == KStreamBinderConfigurationProperties.SerdeError.sendToDlq) { + extendedConsumerProperties.getExtension().setEnableDlq(true); + } + if (!StringUtils.hasText(group)) { + group = binderConfigurationProperties.getApplicationId(); + } + this.kafkaTopicProvisioner.provisionConsumerDestination(name, group, extendedConsumerProperties); + + 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); + + StreamsConfig streamsConfig = this.KStreamBindingInformationCatalogue.getStreamsConfig(inputTarget); + DeserializationExceptionHandler deserializationExceptionHandler = streamsConfig.defaultDeserializationExceptionHandler(); + if(deserializationExceptionHandler instanceof SendToDlqAndContinue) { + ((SendToDlqAndContinue)deserializationExceptionHandler).addKStreamDlqDispatch(name, kStreamDlqDispatch); + } + } + return new DefaultBinding<>(name, group, inputTarget, null); + } + + @Override + protected Binding> doBindProducer(String name, KTable outboundBindTarget, + ExtendedProducerProperties properties) { + throw new UnsupportedOperationException("No producer level binding is allowed for KTable"); + } + + @Override + public KStreamConsumerProperties getExtendedConsumerProperties(String channelName) { + return this.kStreamExtendedBindingProperties.getExtendedConsumerProperties(channelName); + } + + @Override + public KStreamProducerProperties getExtendedProducerProperties(String channelName) { + return this.kStreamExtendedBindingProperties.getExtendedProducerProperties(channelName); + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KTableBoundElementFactory.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KTableBoundElementFactory.java new file mode 100644 index 000000000..b2fa09993 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KTableBoundElementFactory.java @@ -0,0 +1,83 @@ +/* + * 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 org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.apache.kafka.streams.kstream.KTable; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; +import org.springframework.util.Assert; + +/** + * {@link org.springframework.cloud.stream.binding.BindingTargetFactory} for {@link KTable} + * + * @since 2.0.0 + * @author Soby Chacko + */ +public class KTableBoundElementFactory extends AbstractBindingTargetFactory { + + public KTableBoundElementFactory() { + super(KTable.class); + } + + @Override + public KTable createInput(String name) { + KTableBoundElementFactory.KTableWrapperHandler wrapper= new KTableBoundElementFactory.KTableWrapperHandler(); + ProxyFactory proxyFactory = new ProxyFactory(KTableBoundElementFactory.KTableWrapper.class, KTable.class); + proxyFactory.addAdvice(wrapper); + + return (KTable) proxyFactory.getProxy(); + } + + @Override + @SuppressWarnings("unchecked") + public KTable createOutput(final String name) { + throw new UnsupportedOperationException("Outbound operations are not allowed on target type KTable"); + } + + public interface KTableWrapper { + void wrap(KTable delegate); + } + + private static class KTableWrapperHandler implements KTableBoundElementFactory.KTableWrapper, MethodInterceptor { + + private KTable delegate; + + public void wrap(KTable delegate) { + Assert.notNull(delegate, "delegate cannot be null"); + Assert.isNull(this.delegate, "delegate already set to " + this.delegate); + this.delegate = delegate; + } + + @Override + public Object invoke(MethodInvocation methodInvocation) throws Throwable { + if (methodInvocation.getMethod().getDeclaringClass().equals(KTable.class)) { + Assert.notNull(delegate, "Trying to invoke " + methodInvocation + .getMethod() + " but no delegate has been set."); + return methodInvocation.getMethod().invoke(delegate, methodInvocation.getArguments()); + } + else if (methodInvocation.getMethod().getDeclaringClass().equals(KTableBoundElementFactory.KTableWrapper.class)) { + return methodInvocation.getMethod().invoke(this, methodInvocation.getArguments()); + } + else { + throw new IllegalStateException("Only KStream method invocations are permitted"); + } + } + } +} 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 index 107a40162..ff9a198a0 100644 --- 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 @@ -139,8 +139,8 @@ public class KeyValueSerdeResolver { 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 = this.binderConfigurationProperties.getConfiguration().containsKey("default.key.serde") ? + Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("default.key.serde"), Serde.class) : Serdes.ByteArray(); } keySerde.configure(streamConfigGlobalProperties, true); @@ -157,8 +157,8 @@ public class KeyValueSerdeResolver { 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(); + valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("default.value.serde") ? + Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("default.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/StreamsBuildersLifecycle.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/StreamsBuildersLifecycle.java new file mode 100644 index 000000000..64d118940 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/StreamsBuildersLifecycle.java @@ -0,0 +1,97 @@ +/* + * 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.Set; + +import org.springframework.context.SmartLifecycle; +import org.springframework.kafka.KafkaException; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; + +/** + * @author Soby Chacko + */ +public class StreamsBuildersLifecycle implements SmartLifecycle { + + private final KStreamBindingInformationCatalogue kStreamBindingInformationCatalogue; + private final QueryableStoreRegistry queryableStoreRegistry; + + private volatile boolean running; + + public StreamsBuildersLifecycle(KStreamBindingInformationCatalogue kStreamBindingInformationCatalogue, QueryableStoreRegistry queryableStoreRegistry) { + this.kStreamBindingInformationCatalogue = kStreamBindingInformationCatalogue; + this.queryableStoreRegistry = queryableStoreRegistry; + } + + @Override + public boolean isAutoStartup() { + return true; + } + + @Override + public void stop(Runnable callback) { + stop(); + if (callback != null) { + callback.run(); + } + } + + @Override + public synchronized void start() { + if (!this.running) { + try { + Set streamsBuilderFactoryBeans = this.kStreamBindingInformationCatalogue.getStreamsBuilderFactoryBeans(); + for (StreamsBuilderFactoryBean streamsBuilderFactoryBean : streamsBuilderFactoryBeans) { + streamsBuilderFactoryBean.start(); + queryableStoreRegistry.registerKafkaStreams(streamsBuilderFactoryBean.getKafkaStreams()); + } + this.running = true; + } catch (Exception e) { + throw new KafkaException("Could not start stream: ", e); + } + } + } + + @Override + public synchronized void stop() { + if (this.running) { + try { + Set streamsBuilderFactoryBeans = this.kStreamBindingInformationCatalogue.getStreamsBuilderFactoryBeans(); + for (StreamsBuilderFactoryBean streamsBuilderFactoryBean : streamsBuilderFactoryBeans) { + streamsBuilderFactoryBean.stop(); + } + } + catch (Exception e) { + throw new IllegalStateException(e); + } + finally { + this.running = false; + } + } + } + + @Override + public synchronized boolean isRunning() { + return this.running; + } + + @Override + public int getPhase() { + return Integer.MAX_VALUE - 100; + } + +} 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 528cd6fc9..a8689d341 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 @@ -27,7 +27,6 @@ import org.springframework.cloud.stream.binder.kstream.KStreamBinder; 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; @@ -57,10 +56,10 @@ public class KStreamBinderConfiguration { KafkaTopicProvisioner kafkaTopicProvisioner, KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue, - KeyValueSerdeResolver keyValueSerdeResolver, QueryableStoreRegistry queryableStoreRegistry) { + KeyValueSerdeResolver keyValueSerdeResolver) { KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, KStreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue, - keyValueSerdeResolver, queryableStoreRegistry); + keyValueSerdeResolver); kStreamBinder.setkStreamExtendedBindingProperties(kStreamExtendedBindingProperties); return kStreamBinder; } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfigurationProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfigurationProperties.java index 8641b8cfb..010e23338 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfigurationProperties.java @@ -23,6 +23,12 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfi */ public class KStreamBinderConfigurationProperties extends KafkaBinderConfigurationProperties { + public enum SerdeError { + logAndContinue, + logAndFail, + sendToDlq + } + private String applicationId = "default"; public String getApplicationId() { @@ -33,4 +39,19 @@ public class KStreamBinderConfigurationProperties extends KafkaBinderConfigurati this.applicationId = applicationId; } + /** + * {@link org.apache.kafka.streams.errors.DeserializationExceptionHandler} to use + * when there is a Serde error. {@link KStreamBinderConfigurationProperties.SerdeError} + * values are used to provide the exception handler on consumer binding. + */ + private KStreamBinderConfigurationProperties.SerdeError serdeError; + + public KStreamBinderConfigurationProperties.SerdeError getSerdeError() { + return serdeError; + } + + public void setSerdeError(KStreamBinderConfigurationProperties.SerdeError serdeError) { + this.serdeError = serdeError; + } + } 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 ec430f1c1..58d02e8e5 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 @@ -22,6 +22,8 @@ import java.util.Map; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler; +import org.apache.kafka.streams.errors.LogAndFailExceptionHandler; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.context.properties.ConfigurationProperties; @@ -32,9 +34,11 @@ import org.springframework.cloud.stream.binder.kstream.KStreamBoundMessageConver 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.KTableBoundElementFactory; 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.binder.kstream.StreamsBuildersLifecycle; import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; @@ -62,6 +66,19 @@ public class KStreamBinderSupportAutoConfiguration { props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName()); props.put(StreamsConfig.APPLICATION_ID_CONFIG, binderConfigurationProperties.getApplicationId()); + if(binderConfigurationProperties.getSerdeError() == KStreamBinderConfigurationProperties.SerdeError.logAndContinue) { + props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + LogAndContinueExceptionHandler.class); + } + else if(binderConfigurationProperties.getSerdeError() == KStreamBinderConfigurationProperties.SerdeError.logAndFail) { + props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + LogAndFailExceptionHandler.class); + } + else if (binderConfigurationProperties.getSerdeError() == KStreamBinderConfigurationProperties.SerdeError.sendToDlq) { + props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + SendToDlqAndContinue.class); + } + if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConfiguration())) { props.putAll(binderConfigurationProperties.getConfiguration()); } @@ -82,28 +99,37 @@ public class KStreamBinderSupportAutoConfiguration { @Bean public KStreamListenerSetupMethodOrchestrator kStreamListenerSetupMethodOrchestrator( + BindingServiceProperties bindingServiceProperties, + KStreamExtendedBindingProperties kStreamExtendedBindingProperties, + KeyValueSerdeResolver keyValueSerdeResolver, + KStreamBindingInformationCatalogue kStreamBindingInformationCatalogue, KStreamListenerParameterAdapter kafkaStreamListenerParameterAdapter, - Collection streamListenerResultAdapters){ - return new KStreamListenerSetupMethodOrchestrator(kafkaStreamListenerParameterAdapter, streamListenerResultAdapters); + Collection streamListenerResultAdapters, + KStreamBinderConfigurationProperties binderConfigurationProperties) { + return new KStreamListenerSetupMethodOrchestrator(bindingServiceProperties, + kStreamExtendedBindingProperties, keyValueSerdeResolver, kStreamBindingInformationCatalogue, + kafkaStreamListenerParameterAdapter, streamListenerResultAdapters, binderConfigurationProperties); } @Bean public KStreamBoundMessageConversionDelegate messageConversionDelegate(CompositeMessageConverterFactory compositeMessageConverterFactory, SendToDlqAndContinue sendToDlqAndContinue, - KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue, + KStreamBinderConfigurationProperties binderConfigurationProperties) { return new KStreamBoundMessageConversionDelegate(compositeMessageConverterFactory, sendToDlqAndContinue, - KStreamBindingInformationCatalogue); + KStreamBindingInformationCatalogue, binderConfigurationProperties); } @Bean public KStreamBoundElementFactory kafkaStreamBindableTargetFactory(BindingServiceProperties bindingServiceProperties, - KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue, - KeyValueSerdeResolver keyValueSerdeResolver, - KStreamExtendedBindingProperties kStreamExtendedBindingProperties) { - KStreamBoundElementFactory kStreamBoundElementFactory = new KStreamBoundElementFactory(bindingServiceProperties, - KStreamBindingInformationCatalogue, keyValueSerdeResolver); - kStreamBoundElementFactory.setkStreamExtendedBindingProperties(kStreamExtendedBindingProperties); - return kStreamBoundElementFactory; + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { + return new KStreamBoundElementFactory(bindingServiceProperties, + KStreamBindingInformationCatalogue); + } + + @Bean + public KTableBoundElementFactory kTableBoundElementFactory() { + return new KTableBoundElementFactory(); } @Bean @@ -117,9 +143,10 @@ public class KStreamBinderSupportAutoConfiguration { } @Bean - public KeyValueSerdeResolver keyValueSerdeResolver(@Qualifier("streamConfigGlobalProperties") Map streamConfigGlobalProperties, + @SuppressWarnings("unchecked") + public KeyValueSerdeResolver keyValueSerdeResolver(@Qualifier("streamConfigGlobalProperties") Object streamConfigGlobalProperties, KStreamBinderConfigurationProperties kStreamBinderConfigurationProperties) { - return new KeyValueSerdeResolver(streamConfigGlobalProperties, kStreamBinderConfigurationProperties); + return new KeyValueSerdeResolver((Map)streamConfigGlobalProperties, kStreamBinderConfigurationProperties); } @Bean @@ -127,4 +154,10 @@ public class KStreamBinderSupportAutoConfiguration { return new QueryableStoreRegistry(); } + @Bean + public StreamsBuildersLifecycle streamsBuildersLifecycle(KStreamBindingInformationCatalogue kStreamBindingInformationCatalogue, + QueryableStoreRegistry queryableStoreRegistry){ + return new StreamsBuildersLifecycle(kStreamBindingInformationCatalogue, 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 1b595f67f..7e9008a01 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 @@ -24,12 +24,6 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerPro */ public class KStreamConsumerProperties extends KafkaConsumerProperties { - public enum SerdeError { - logAndContinue, - logAndFail, - sendToDlq - } - /** * Key serde specified per binding. */ @@ -41,11 +35,9 @@ public class KStreamConsumerProperties extends KafkaConsumerProperties { 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. + * Materialized as a KeyValueStore */ - private SerdeError serdeError; + private String materializedAs; public String getKeySerde() { return keySerde; @@ -63,11 +55,12 @@ public class KStreamConsumerProperties extends KafkaConsumerProperties { this.valueSerde = valueSerde; } - public SerdeError getSerdeError() { - return serdeError; + public String getMaterializedAs() { + return materializedAs; } - public void setSerdeError(SerdeError serdeError) { - this.serdeError = serdeError; + public void setMaterializedAs(String materializedAs) { + this.materializedAs = materializedAs; } + } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KTableBinderConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KTableBinderConfiguration.java new file mode 100644 index 000000000..306ff28b5 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KTableBinderConfiguration.java @@ -0,0 +1,51 @@ +/* + * 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.config; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +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.KStreamBindingInformationCatalogue; +import org.springframework.cloud.stream.binder.kstream.KTableBinder; +import org.springframework.context.annotation.Bean; + +/** + * @author Soby Chacko + */ +public class KTableBinderConfiguration { + + @Autowired + private KafkaProperties kafkaProperties; + + @Autowired + private KStreamExtendedBindingProperties kStreamExtendedBindingProperties; + + @Bean + public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties) { + return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties); + } + + @Bean + public KTableBinder kTableBinder(KStreamBinderConfigurationProperties binderConfigurationProperties, + KafkaTopicProvisioner kafkaTopicProvisioner, + KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) { + KTableBinder kStreamBinder = new KTableBinder(binderConfigurationProperties, kafkaTopicProvisioner, + KStreamBindingInformationCatalogue); + return kStreamBinder; + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.binders b/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.binders index 0195e2832..15c31791f 100644 --- a/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.binders +++ b/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.binders @@ -1,4 +1,6 @@ kstream:\ org.springframework.cloud.stream.binder.kstream.config.KStreamBinderConfiguration +ktable:\ +org.springframework.cloud.stream.binder.kstream.config.KTableBinderConfiguration 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 index 182990979..56f226b06 100644 --- 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 @@ -97,8 +97,8 @@ public abstract class DeserializationErrorHandlerByKafkaTests { "spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", "spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", "spring.cloud.stream.bindings.input.group=group", - "spring.cloud.stream.kstream.bindings.input.consumer.serdeError=sendToDlq", - "spring.cloud.stream.kstream.binder.configuration.value.serde=" + + "spring.cloud.stream.kstream.binder.serdeError=sendToDlq", + "spring.cloud.stream.kstream.binder.configuration.default.value.serde=" + "org.apache.kafka.common.serialization.Serdes$IntegerSerde"}, webEnvironment= SpringBootTest.WebEnvironment.NONE ) 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 index e3213aeaa..84b80ab65 100644 --- 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 @@ -92,12 +92,12 @@ public abstract class DeserializtionErrorHandlerByBinderTests { "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.kstream.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "spring.cloud.stream.kstream.binder.configuration.default.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.kstream.binder.serdeError=sendToDlq", "spring.cloud.stream.bindings.input.group=foobar-group"}, webEnvironment= SpringBootTest.WebEnvironment.NONE ) 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 8f5fe4430..8c6feff26 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 @@ -81,8 +81,8 @@ public class KStreamBinderPojoInputAndPrimitiveTypeOutputTests { "--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.kstream.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.default.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", 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 21505ae4e..990899de4 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 @@ -37,6 +37,7 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; 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; @@ -87,8 +88,8 @@ public class KStreamBinderWordCountIntegrationTests { "--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.kstream.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.bindings.output.producer.headerMode=raw", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.kstream.timeWindow.length=5000", @@ -120,10 +121,15 @@ public class KStreamBinderWordCountIntegrationTests { @Autowired private TimeWindows timeWindows; - @StreamListener("input") + @StreamListener @SendTo("output") - public KStream process(KStream input) { + public KStream process(@Input("input") KStream input) { + input.map((k,v) -> { + System.out.println(k); + System.out.println(v); + return new KeyValue<>(k,v); + }); return input .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .map((key, value) -> new KeyValue<>(value, value)) 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 b2e262bac..019d883d0 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 @@ -83,8 +83,8 @@ public class KStreamInteractiveQueryIntegrationTests { "--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.kstream.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.bindings.output.producer.headerMode=raw", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), 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 4af354856..e035f8600 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 @@ -80,8 +80,8 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests { "--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.kstream.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.default.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", diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/StreamToTableJoinIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/StreamToTableJoinIntegrationTests.java new file mode 100644 index 000000000..175555631 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/StreamToTableJoinIntegrationTests.java @@ -0,0 +1,246 @@ +/* + * 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.ArrayList; +import java.util.Arrays; +import java.util.List; +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.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.LongDeserializer; +import org.apache.kafka.common.serialization.LongSerializer; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.Joined; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; +import org.apache.kafka.streams.kstream.Serialized; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +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.ConfigurableApplicationContext; +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 static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Soby Chacko + */ +public class StreamToTableJoinIntegrationTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "output-topic"); + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "output-topic"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @EnableBinding(KStreamProcessorX.class) + @EnableAutoConfiguration + @EnableConfigurationProperties(KStreamApplicationSupportProperties.class) + public static class CountClicksPerRegionApplication { + + @StreamListener + @SendTo("output") + public KStream process(@Input("input") KStream userClicksStream, + @Input("inputX") KTable userRegionsTable) { + + return userClicksStream + .leftJoin(userRegionsTable, (clicks, region) -> new RegionWithClicks(region == null ? "UNKNOWN" : region, clicks), + Joined.with(Serdes.String(), Serdes.Long(), null)) + .map((user, regionWithClicks) -> new KeyValue<>(regionWithClicks.getRegion(), regionWithClicks.getClicks())) + .groupByKey(Serialized.with(Serdes.String(), Serdes.Long())) + .reduce((firstClicks, secondClicks) -> firstClicks + secondClicks) + .toStream(); + } + } + + interface KStreamProcessorX extends KStreamProcessor { + + @Input("inputX") + KTable inputX(); + } + + @Test + public void testStreamToTable() throws Exception { + SpringApplication app = new SpringApplication(CountClicksPerRegionApplication.class); + app.setWebEnvironment(false); + + ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.destination=user-clicks", + "--spring.cloud.stream.bindings.inputX.destination=user-regions", + "--spring.cloud.stream.bindings.output.destination=output-topic", + "--spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.inputX.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "--spring.cloud.stream.kstream.bindings.input.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.bindings.input.consumer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.kstream.bindings.inputX.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.bindings.inputX.consumer.valueSerde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.kstream.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=10000", + "--spring.cloud.stream.bindings.output.producer.headerMode=raw", + "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.bindings.inputX.consumer.headerMode=raw", + "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + try { + // Input 1: Clicks per user (multiple records allowed per user). + List> userClicks = Arrays.asList( + new KeyValue<>("alice", 13L), + new KeyValue<>("bob", 4L), + new KeyValue<>("chao", 25L), + new KeyValue<>("bob", 19L), + new KeyValue<>("dave", 56L), + new KeyValue<>("eve", 78L), + new KeyValue<>("alice", 40L), + new KeyValue<>("fang", 99L) + ); + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, LongSerializer.class); + + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("user-clicks"); + + for (KeyValue keyValue : userClicks) { + template.sendDefault(keyValue.key, keyValue.value); + } + + // Input 2: Region per user (multiple records allowed per user). + List> userRegions = Arrays.asList( + new KeyValue<>("alice", "asia"), /* Alice lived in Asia originally... */ + new KeyValue<>("bob", "americas"), + new KeyValue<>("chao", "asia"), + new KeyValue<>("dave", "europe"), + new KeyValue<>("alice", "europe"), /* ...but moved to Europe some time later. */ + new KeyValue<>("eve", "americas"), + new KeyValue<>("fang", "asia") + ); + + Map senderProps1 = KafkaTestUtils.producerProps(embeddedKafka); + senderProps1.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + senderProps1.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + + DefaultKafkaProducerFactory pf1 = new DefaultKafkaProducerFactory<>(senderProps1); + KafkaTemplate template1 = new KafkaTemplate<>(pf1, true); + template1.setDefaultTopic("user-regions"); + + for (KeyValue keyValue : userRegions) { + template1.sendDefault(keyValue.key, keyValue.value); + } + + List> expectedClicksPerRegion = Arrays.asList( + new KeyValue<>("americas", 101L), + new KeyValue<>("europe", 109L), + new KeyValue<>("asia", 124L) + ); + + //Verify that we receive the expected data + int count = 0; + long start = System.currentTimeMillis(); + List> actualClicksPerRegion = new ArrayList<>(); + do { + ConsumerRecords records = KafkaTestUtils.getRecords(consumer); + count = count + records.count(); + for (ConsumerRecord record : records) { + actualClicksPerRegion.add(new KeyValue<>(record.key(), record.value())); + } + } while (count < expectedClicksPerRegion.size() && (System.currentTimeMillis() - start) < 30000 ); + + assertThat(count == expectedClicksPerRegion.size()).isTrue(); + assertThat(actualClicksPerRegion).hasSameElementsAs(expectedClicksPerRegion); + } + catch (Exception e){ + System.out.println(e); + } + finally { + context.close(); + } + } + + /** + * Tuple for a region and its associated number of clicks. + */ + private static final class RegionWithClicks { + + private final String region; + private final long clicks; + + RegionWithClicks(String region, long clicks) { + if (region == null || region.isEmpty()) { + throw new IllegalArgumentException("region must be set"); + } + if (clicks < 0) { + throw new IllegalArgumentException("clicks must not be negative"); + } + this.region = region; + this.clicks = clicks; + } + + public String getRegion() { + return region; + } + + public long getClicks() { + return clicks; + } + + } +} 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 4988df99a..f94aa39fc 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 @@ -78,57 +78,6 @@ public class WordCountMultipleBranchesIntegrationTests { consumer.close(); } - @Test - public void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { - SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); - 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", - "--spring.cloud.stream.bindings.output2.destination=foo", - "--spring.cloud.stream.bindings.output2.contentType=application/json", - "--spring.cloud.stream.bindings.output3.destination=bar", - "--spring.cloud.stream.bindings.output3.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", - "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), - "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); - try { - receiveAndValidate(context); - } finally { - context.close(); - } - } - - private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception { - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); - KafkaTemplate template = new KafkaTemplate<>(pf, true); - template.setDefaultTopic("words"); - template.sendDefault("english"); - ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts"); - assertThat(cr.value().contains("\"word\":\"english\",\"count\":1")).isTrue(); - - template.sendDefault("french"); - template.sendDefault("french"); - cr = KafkaTestUtils.getSingleRecord(consumer, "foo"); - assertThat(cr.value().contains("\"word\":\"french\",\"count\":2")).isTrue(); - - template.sendDefault("spanish"); - template.sendDefault("spanish"); - template.sendDefault("spanish"); - cr = KafkaTestUtils.getSingleRecord(consumer, "bar"); - assertThat(cr.value().contains("\"word\":\"spanish\",\"count\":3")).isTrue(); - } - @EnableBinding(KStreamProcessorX.class) @EnableAutoConfiguration @EnableConfigurationProperties(KStreamApplicationSupportProperties.class) @@ -172,6 +121,57 @@ public class WordCountMultipleBranchesIntegrationTests { KStream output3(); } + @Test + public void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { + SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); + 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", + "--spring.cloud.stream.bindings.output2.destination=foo", + "--spring.cloud.stream.bindings.output2.contentType=application/json", + "--spring.cloud.stream.bindings.output3.destination=bar", + "--spring.cloud.stream.bindings.output3.contentType=application/json", + "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kstream.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.bindings.output.producer.headerMode=raw", + "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kstream.timeWindow.length=5000", + "--spring.cloud.stream.kstream.timeWindow.advanceBy=0", + "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + try { + receiveAndValidate(context); + } finally { + context.close(); + } + } + + private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words"); + template.sendDefault("english"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts"); + assertThat(cr.value().contains("\"word\":\"english\",\"count\":1")).isTrue(); + + template.sendDefault("french"); + template.sendDefault("french"); + cr = KafkaTestUtils.getSingleRecord(consumer, "foo"); + assertThat(cr.value().contains("\"word\":\"french\",\"count\":2")).isTrue(); + + template.sendDefault("spanish"); + template.sendDefault("spanish"); + template.sendDefault("spanish"); + cr = KafkaTestUtils.getSingleRecord(consumer, "bar"); + assertThat(cr.value().contains("\"word\":\"spanish\",\"count\":3")).isTrue(); + } + static class WordCount { private String word; 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 index 8d7167834..8a5769542 100644 --- 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 @@ -2,8 +2,8 @@ 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.kstream.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde +spring.cloud.stream.kstream.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde spring.cloud.stream.bindings.output.producer.headerMode=raw spring.cloud.stream.bindings.input.consumer.headerMode=raw spring.cloud.stream.kstream.timeWindow.length=5000