From 090125fa712a45275217ed95458f85032056f06d Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 5 Feb 2018 11:48:13 -0500 Subject: [PATCH] Multiple Input bindings on same StreamListener method - In kafka streams applications, it is essential to have multiple bindings for various target types such as KStream, KTable etc. These changes allow to have more than one type of target type bindings on a single StreamListenerMethod. - Currently support KStream and KTable target types - Refactoring the KStream listener orchestrator strategy - Input bindings are initiated through a proxy and later on wrapped with the real target created from a common Kafka Streams StreamsBuilder - Adding tests to verify multiple input bindings for KStream and KTable Resolves #298 Resolves #303 Polishing Materializing KTables as state stores Void return types on kafka streams StreamListener methods Polishing --- .../stream/binder/kstream/KStreamBinder.java | 77 +---- .../KStreamBindingInformationCatalogue.java | 38 ++- .../kstream/KStreamBoundElementFactory.java | 86 +---- ...KStreamBoundMessageConversionDelegate.java | 23 +- ...StreamListenerSetupMethodOrchestrator.java | 307 +++++++++++++++--- .../stream/binder/kstream/KTableBinder.java | 108 ++++++ .../kstream/KTableBoundElementFactory.java | 83 +++++ .../binder/kstream/KeyValueSerdeResolver.java | 8 +- .../kstream/StreamsBuildersLifecycle.java | 97 ++++++ .../config/KStreamBinderConfiguration.java | 5 +- .../KStreamBinderConfigurationProperties.java | 21 ++ ...KStreamBinderSupportAutoConfiguration.java | 59 +++- .../config/KStreamConsumerProperties.java | 21 +- .../config/KTableBinderConfiguration.java | 51 +++ .../main/resources/META-INF/spring.binders | 2 + ...serializationErrorHandlerByKafkaTests.java | 4 +- ...serializtionErrorHandlerByBinderTests.java | 6 +- ...rPojoInputAndPrimitiveTypeOutputTests.java | 4 +- ...StreamBinderWordCountIntegrationTests.java | 14 +- ...treamInteractiveQueryIntegrationTests.java | 4 +- ...PojoInputStringOutputIntegrationTests.java | 4 +- .../StreamToTableJoinIntegrationTests.java | 246 ++++++++++++++ ...CountMultipleBranchesIntegrationTests.java | 102 +++--- .../binder/kstream/integTest-1.properties | 4 +- 24 files changed, 1075 insertions(+), 299 deletions(-) create mode 100644 spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KTableBinder.java create mode 100644 spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KTableBoundElementFactory.java create mode 100644 spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/StreamsBuildersLifecycle.java create mode 100644 spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KTableBinderConfiguration.java create mode 100644 spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/StreamToTableJoinIntegrationTests.java 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