From 41d9136812a9db4b4924d51ce8bc5fcfd47826d0 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 14 Sep 2018 10:51:15 -0400 Subject: [PATCH] GH-384: Binding support for GlobalKTable Resolves spring-cloud/spring-cloud-stream-binder-kafka#384 * New binding targets and binder implementation for GlobalKTable within kafka-streams binder * Refactoring existing structure to accommodate the new binder * Adding integration test to verify the GlobalKTable behavior Resolves #384 * Addressing PR review comments * Update spring-kafka to 2.2.0.M3 Addressing PR review comments Polishing * Addressing PR review comments * Addressing PR review comments --- pom.xml | 2 +- .../kafka/streams/GlobalKTableBinder.java | 93 +++++ .../GlobalKTableBinderConfiguration.java | 82 +++++ .../GlobalKTableBoundElementFactory.java | 93 +++++ .../binder/kafka/streams/KStreamBinder.java | 46 +-- .../streams/KStreamBoundElementFactory.java | 2 +- .../binder/kafka/streams/KTableBinder.java | 44 +-- .../streams/KTableBinderConfiguration.java | 1 + .../streams/KTableBoundElementFactory.java | 4 +- ...StreamsBinderSupportAutoConfiguration.java | 5 + ...fkaStreamsBindingInformationCatalogue.java | 2 +- .../KafkaStreamsConsumerBindingUtils.java | 74 ++++ ...StreamListenerSetupMethodOrchestrator.java | 65 ++-- .../streams/StreamsBuilderFactoryManager.java | 2 +- .../main/resources/META-INF/spring.binders | 2 + ...treamsBinderWordCountIntegrationTests.java | 2 +- ...PojoInputStringOutputIntegrationTests.java | 2 +- ...eamToGlobalKTableJoinIntegrationTests.java | 321 ++++++++++++++++++ .../StreamToTableJoinIntegrationTests.java | 19 +- 19 files changed, 745 insertions(+), 116 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java create mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java create mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java create mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsConsumerBindingUtils.java create mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java diff --git a/pom.xml b/pom.xml index 0508dd1ac..fa5f46762 100644 --- a/pom.xml +++ b/pom.xml @@ -12,7 +12,7 @@ 1.8 - 2.2.0.BUILD-SNAPSHOT + 2.2.0.M3 3.1.0.BUILD-SNAPSHOT 2.0.0 2.1.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java new file mode 100644 index 000000000..39f49419f --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java @@ -0,0 +1,93 @@ +/* + * 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.kafka.streams; + +import org.apache.kafka.streams.kstream.GlobalKTable; + +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.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties; +import org.springframework.util.StringUtils; + +/** + * An {@link AbstractBinder} implementation for {@link GlobalKTable}. + * + * Provides only consumer binding for the bound {@link GlobalKTable}. + * Output bindings are not allowed on this binder. + * + * @author Soby Chacko + * @since 2.1.0 + */ +public class GlobalKTableBinder extends + AbstractBinder, ExtendedConsumerProperties, ExtendedProducerProperties> + implements ExtendedPropertiesBinder, KafkaStreamsConsumerProperties, KafkaStreamsProducerProperties> { + + private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties; + + private final KafkaTopicProvisioner kafkaTopicProvisioner; + + private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; + + private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties(); + + public GlobalKTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) { + this.binderConfigurationProperties = binderConfigurationProperties; + this.kafkaTopicProvisioner = kafkaTopicProvisioner; + this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; + } + + @Override + @SuppressWarnings("unchecked") + protected Binding> doBindConsumer(String name, String group, GlobalKTable inputTarget, + ExtendedConsumerProperties properties) { + if (!StringUtils.hasText(group)) { + group = binderConfigurationProperties.getApplicationId(); + } + KafkaStreamsConsumerBindingUtils.prepareConsumerBinding(name, group, inputTarget, + getApplicationContext(), + kafkaTopicProvisioner, + kafkaStreamsBindingInformationCatalogue, + binderConfigurationProperties, properties); + return new DefaultBinding<>(name, group, inputTarget, null); + } + + @Override + protected Binding> doBindProducer(String name, GlobalKTable outboundBindTarget, + ExtendedProducerProperties properties) { + throw new UnsupportedOperationException("No producer level binding is allowed for GlobalKTable"); + } + + @Override + public KafkaStreamsConsumerProperties getExtendedConsumerProperties(String channelName) { + return this.kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(channelName); + } + + @Override + public KafkaStreamsProducerProperties getExtendedProducerProperties(String channelName) { + throw new UnsupportedOperationException("No producer binding is allowed and therefore no properties"); + } + +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java new file mode 100644 index 000000000..5ec5eef73 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java @@ -0,0 +1,82 @@ +/* + * 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.kafka.streams; + +import org.springframework.beans.factory.config.MethodInvokingFactoryBean; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +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.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; +import org.springframework.core.type.AnnotationMetadata; + +/** + * @author Soby Chacko + * @since 2.1.0 + */ +@Configuration +@Import(GlobalKTableBinderConfiguration.Registrar.class) +public class GlobalKTableBinderConfiguration { + + static class Registrar implements ImportBeanDefinitionRegistrar { + + private static final String BEAN_NAME = "outerContext"; + + @Override + public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, + BeanDefinitionRegistry registry) { + if (registry.containsBeanDefinition(BEAN_NAME)) { + + AbstractBeanDefinition configBean = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingFactoryBean.class) + .addPropertyReference("targetObject", BEAN_NAME) + .addPropertyValue("targetMethod", "getBean") + .addPropertyValue("arguments", KafkaStreamsBinderConfigurationProperties.class) + .getBeanDefinition(); + + registry.registerBeanDefinition(KafkaStreamsBinderConfigurationProperties.class.getSimpleName(), configBean); + + AbstractBeanDefinition catalogueBean = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingFactoryBean.class) + .addPropertyReference("targetObject", BEAN_NAME) + .addPropertyValue("targetMethod", "getBean") + .addPropertyValue("arguments", KafkaStreamsBindingInformationCatalogue.class) + .getBeanDefinition(); + + registry.registerBeanDefinition(KafkaStreamsBindingInformationCatalogue.class.getSimpleName(), catalogueBean); + } + } + } + + @Bean + public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties, + KafkaProperties kafkaProperties) { + return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties); + } + + @Bean + public GlobalKTableBinder GlobalKTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, + KafkaTopicProvisioner kafkaTopicProvisioner, + KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue) { + return new GlobalKTableBinder(binderConfigurationProperties, kafkaTopicProvisioner, + KafkaStreamsBindingInformationCatalogue); + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java new file mode 100644 index 000000000..9578c0e45 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java @@ -0,0 +1,93 @@ +/* + * 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.kafka.streams; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.apache.kafka.streams.kstream.GlobalKTable; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.util.Assert; + +/** + * {@link org.springframework.cloud.stream.binding.BindingTargetFactory} for {@link GlobalKTable} + * + * Input bindings are only created as output bindings on GlobalKTable are not allowed. + * + * @author Soby Chacko + * @since 2.1.0 + */ +public class GlobalKTableBoundElementFactory extends AbstractBindingTargetFactory { + + private final BindingServiceProperties bindingServiceProperties; + + GlobalKTableBoundElementFactory(BindingServiceProperties bindingServiceProperties) { + super(GlobalKTable.class); + this.bindingServiceProperties = bindingServiceProperties; + } + + @Override + public GlobalKTable createInput(String name) { + ConsumerProperties consumerProperties = this.bindingServiceProperties.getConsumerProperties(name); + //Always set multiplex to true in the kafka streams binder + consumerProperties.setMultiplex(true); + + GlobalKTableBoundElementFactory.GlobalKTableWrapperHandler wrapper= new GlobalKTableBoundElementFactory.GlobalKTableWrapperHandler(); + ProxyFactory proxyFactory = new ProxyFactory(GlobalKTableBoundElementFactory.GlobalKTableWrapper.class, GlobalKTable.class); + proxyFactory.addAdvice(wrapper); + + return (GlobalKTable) proxyFactory.getProxy(); + } + + @Override + public GlobalKTable createOutput(String name) { + throw new UnsupportedOperationException("Outbound operations are not allowed on target type GlobalKTable"); + } + + public interface GlobalKTableWrapper { + void wrap(GlobalKTable delegate); + } + + private static class GlobalKTableWrapperHandler implements GlobalKTableBoundElementFactory.GlobalKTableWrapper, MethodInterceptor { + + private GlobalKTable delegate; + + public void wrap(GlobalKTable 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(GlobalKTable.class)) { + Assert.notNull(delegate, "Trying to prepareConsumerBinding " + methodInvocation + .getMethod() + " but no delegate has been set."); + return methodInvocation.getMethod().invoke(delegate, methodInvocation.getArguments()); + } + else if (methodInvocation.getMethod().getDeclaringClass().equals(GlobalKTableBoundElementFactory.GlobalKTableWrapper.class)) { + return methodInvocation.getMethod().invoke(this, methodInvocation.getArguments()); + } + else { + throw new IllegalStateException("Only GlobalKTable method invocations are permitted"); + } + } + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index 6d458af72..99a5849e7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java @@ -19,8 +19,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; 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.kstream.KStream; import org.apache.kafka.streams.kstream.Produced; @@ -30,7 +28,6 @@ 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.properties.KafkaProducerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; @@ -62,7 +59,7 @@ class KStreamBinder extends private final KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate; - private final KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue; + private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; private final KeyValueSerdeResolver keyValueSerdeResolver; @@ -74,7 +71,7 @@ class KStreamBinder extends this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; this.kafkaStreamsMessageConversionDelegate = kafkaStreamsMessageConversionDelegate; - this.KafkaStreamsBindingInformationCatalogue = KafkaStreamsBindingInformationCatalogue; + this.kafkaStreamsBindingInformationCatalogue = KafkaStreamsBindingInformationCatalogue; this.keyValueSerdeResolver = keyValueSerdeResolver; } @@ -83,42 +80,15 @@ class KStreamBinder extends protected Binding> doBindConsumer(String name, String group, KStream inputTarget, ExtendedConsumerProperties properties) { - this.KafkaStreamsBindingInformationCatalogue.registerConsumerProperties(inputTarget, properties.getExtension()); - ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( - properties.getExtension()); - if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) { - extendedConsumerProperties.getExtension().setEnableDlq(true); - } + this.kafkaStreamsBindingInformationCatalogue.registerConsumerProperties(inputTarget, properties.getExtension()); if (!StringUtils.hasText(group)) { group = binderConfigurationProperties.getApplicationId(); } - - String[] inputTopics = StringUtils.commaDelimitedListToStringArray(name); - for (String inputTopic : inputTopics) { - this.kafkaTopicProvisioner.provisionConsumerDestination(inputTopic, group, extendedConsumerProperties); - } - - if (extendedConsumerProperties.getExtension().isEnableDlq()) { - StreamsConfig streamsConfig = this.KafkaStreamsBindingInformationCatalogue.getStreamsConfig(inputTarget); - - KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch = !StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName()) ? - new KafkaStreamsDlqDispatch(extendedConsumerProperties.getExtension().getDlqName(), binderConfigurationProperties, - extendedConsumerProperties.getExtension()) : null; - for (String inputTopic : inputTopics) { - if (StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName())) { - String dlqName = "error." + inputTopic + "." + group; - kafkaStreamsDlqDispatch = new KafkaStreamsDlqDispatch(dlqName, binderConfigurationProperties, - extendedConsumerProperties.getExtension()); - } - SendToDlqAndContinue sendToDlqAndContinue = this.getApplicationContext().getBean(SendToDlqAndContinue.class); - sendToDlqAndContinue.addKStreamDlqDispatch(inputTopic, kafkaStreamsDlqDispatch); - - DeserializationExceptionHandler deserializationExceptionHandler = streamsConfig.defaultDeserializationExceptionHandler(); - if (deserializationExceptionHandler instanceof SendToDlqAndContinue) { - ((SendToDlqAndContinue) deserializationExceptionHandler).addKStreamDlqDispatch(inputTopic, kafkaStreamsDlqDispatch); - } - } - } + KafkaStreamsConsumerBindingUtils.prepareConsumerBinding(name, group, inputTarget, + getApplicationContext(), + kafkaTopicProvisioner, + kafkaStreamsBindingInformationCatalogue, + binderConfigurationProperties, properties); return new DefaultBinding<>(name, group, inputTarget, null); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java index a45e58a45..b09e48d2b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java @@ -95,7 +95,7 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory { @Override public Object invoke(MethodInvocation methodInvocation) throws Throwable { if (methodInvocation.getMethod().getDeclaringClass().equals(KStream.class)) { - Assert.notNull(delegate, "Trying to invoke " + methodInvocation + Assert.notNull(delegate, "Trying to prepareConsumerBinding " + methodInvocation .getMethod() + " but no delegate has been set."); return methodInvocation.getMethod().invoke(delegate, methodInvocation.getArguments()); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java index 6c68358d9..1f72846e5 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java @@ -16,8 +16,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; -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; @@ -26,7 +24,6 @@ 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.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; @@ -50,7 +47,7 @@ class KTableBinder extends private final KafkaTopicProvisioner kafkaTopicProvisioner; - private final KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue; + private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties(); @@ -58,48 +55,21 @@ class KTableBinder extends KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; - this.KafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; + this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; } @Override @SuppressWarnings("unchecked") protected Binding> doBindConsumer(String name, String group, KTable inputTarget, ExtendedConsumerProperties properties) { - ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( - properties.getExtension()); - if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) { - extendedConsumerProperties.getExtension().setEnableDlq(true); - } if (!StringUtils.hasText(group)) { group = binderConfigurationProperties.getApplicationId(); } - - String[] inputTopics = StringUtils.commaDelimitedListToStringArray(name); - for (String inputTopic : inputTopics) { - this.kafkaTopicProvisioner.provisionConsumerDestination(inputTopic, group, extendedConsumerProperties); - } - - if (extendedConsumerProperties.getExtension().isEnableDlq()) { - StreamsConfig streamsConfig = this.KafkaStreamsBindingInformationCatalogue.getStreamsConfig(inputTarget); - - KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch = !StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName()) ? - new KafkaStreamsDlqDispatch(extendedConsumerProperties.getExtension().getDlqName(), binderConfigurationProperties, - extendedConsumerProperties.getExtension()) : null; - for (String inputTopic : inputTopics) { - if (StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName())) { - String dlqName = "error." + inputTopic + "." + group; - kafkaStreamsDlqDispatch = new KafkaStreamsDlqDispatch(dlqName, binderConfigurationProperties, - extendedConsumerProperties.getExtension()); - } - SendToDlqAndContinue sendToDlqAndContinue = this.getApplicationContext().getBean(SendToDlqAndContinue.class); - sendToDlqAndContinue.addKStreamDlqDispatch(inputTopic, kafkaStreamsDlqDispatch); - - DeserializationExceptionHandler deserializationExceptionHandler = streamsConfig.defaultDeserializationExceptionHandler(); - if (deserializationExceptionHandler instanceof SendToDlqAndContinue) { - ((SendToDlqAndContinue) deserializationExceptionHandler).addKStreamDlqDispatch(inputTopic, kafkaStreamsDlqDispatch); - } - } - } + KafkaStreamsConsumerBindingUtils.prepareConsumerBinding(name, group, inputTarget, + getApplicationContext(), + kafkaTopicProvisioner, + kafkaStreamsBindingInformationCatalogue, + binderConfigurationProperties, properties); return new DefaultBinding<>(name, group, inputTarget, null); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java index e97bdacd7..ca8b7d96a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java @@ -29,6 +29,7 @@ import org.springframework.context.annotation.Configuration; /** * @author Soby Chacko */ +@SuppressWarnings("ALL") @Configuration public class KTableBinderConfiguration { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java index e1d64fe75..3a306b82f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java @@ -78,7 +78,7 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory { @Override public Object invoke(MethodInvocation methodInvocation) throws Throwable { if (methodInvocation.getMethod().getDeclaringClass().equals(KTable.class)) { - Assert.notNull(delegate, "Trying to invoke " + methodInvocation + Assert.notNull(delegate, "Trying to prepareConsumerBinding " + methodInvocation .getMethod() + " but no delegate has been set."); return methodInvocation.getMethod().invoke(delegate, methodInvocation.getArguments()); } @@ -86,7 +86,7 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory { return methodInvocation.getMethod().invoke(this, methodInvocation.getArguments()); } else { - throw new IllegalStateException("Only KStream method invocations are permitted"); + throw new IllegalStateException("Only KTable method invocations are permitted"); } } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index d938404db..4bb08df47 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -133,6 +133,11 @@ public class KafkaStreamsBinderSupportAutoConfiguration { return new KTableBoundElementFactory(bindingServiceProperties); } + @Bean + public GlobalKTableBoundElementFactory globalKTableBoundElementFactory(BindingServiceProperties bindingServiceProperties) { + return new GlobalKTableBoundElementFactory(bindingServiceProperties); + } + @Bean public SendToDlqAndContinue sendToDlqAndContinue() { return new SendToDlqAndContinue(); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java index cf0c0eba4..827370beb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java @@ -27,7 +27,7 @@ import org.apache.kafka.streams.kstream.KStream; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.config.BindingProperties; -import org.springframework.kafka.config.StreamsBuilderFactoryBean; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; /** * A catalogue that provides binding information for Kafka Streams target types such as KStream. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsConsumerBindingUtils.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsConsumerBindingUtils.java new file mode 100644 index 000000000..1e5f991d8 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsConsumerBindingUtils.java @@ -0,0 +1,74 @@ +/* + * 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.kafka.streams; + +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.errors.DeserializationExceptionHandler; + +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.util.StringUtils; + +/** + * @author Soby Chacko + */ +class KafkaStreamsConsumerBindingUtils { + + static void prepareConsumerBinding(String name, String group, Object inputTarget, + ApplicationContext context, + KafkaTopicProvisioner kafkaTopicProvisioner, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, + ExtendedConsumerProperties properties) { + ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( + properties.getExtension()); + if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) { + extendedConsumerProperties.getExtension().setEnableDlq(true); + } + + String[] inputTopics = StringUtils.commaDelimitedListToStringArray(name); + for (String inputTopic : inputTopics) { + kafkaTopicProvisioner.provisionConsumerDestination(inputTopic, group, extendedConsumerProperties); + } + + if (extendedConsumerProperties.getExtension().isEnableDlq()) { + StreamsConfig streamsConfig = kafkaStreamsBindingInformationCatalogue.getStreamsConfig(inputTarget); + + KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch = !StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName()) ? + new KafkaStreamsDlqDispatch(extendedConsumerProperties.getExtension().getDlqName(), binderConfigurationProperties, + extendedConsumerProperties.getExtension()) : null; + for (String inputTopic : inputTopics) { + if (StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName())) { + String dlqName = "error." + inputTopic + "." + group; + kafkaStreamsDlqDispatch = new KafkaStreamsDlqDispatch(dlqName, binderConfigurationProperties, + extendedConsumerProperties.getExtension()); + } + SendToDlqAndContinue sendToDlqAndContinue = context.getBean(SendToDlqAndContinue.class); + sendToDlqAndContinue.addKStreamDlqDispatch(inputTopic, kafkaStreamsDlqDispatch); + + DeserializationExceptionHandler deserializationExceptionHandler = streamsConfig.defaultDeserializationExceptionHandler(); + if (deserializationExceptionHandler instanceof SendToDlqAndContinue) { + ((SendToDlqAndContinue) deserializationExceptionHandler).addKStreamDlqDispatch(inputTopic, kafkaStreamsDlqDispatch); + } + } + } + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index 10dbec299..c46d08c5b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -30,6 +30,7 @@ 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.Consumed; +import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Materialized; @@ -62,8 +63,8 @@ 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.config.StreamsBuilderFactoryBean; import org.springframework.kafka.core.CleanupConfig; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.support.MessageBuilder; @@ -79,7 +80,7 @@ import org.springframework.util.StringUtils; * The orchestration primarily focus on the following areas: * * 1. Allow multiple KStream output bindings (KStream branching) by allowing more than one output values on {@link SendTo} - * 2. Allow multiple inbound bindings for multiple KStream and or KTable types. + * 2. Allow multiple inbound bindings for multiple KStream and or KTable/GlobalKTable types. * 3. Each StreamListener method that it orchestrates gets its own {@link StreamsBuilderFactoryBean} and {@link StreamsConfig} * * @author Soby Chacko @@ -111,13 +112,13 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene private ConfigurableApplicationContext applicationContext; KafkaStreamsStreamListenerSetupMethodOrchestrator(BindingServiceProperties bindingServiceProperties, - KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, - KeyValueSerdeResolver keyValueSerdeResolver, - KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, - StreamListenerParameterAdapter streamListenerParameterAdapter, - Collection streamListenerResultAdapters, - KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, - CleanupConfig cleanupConfig) { + KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, + KeyValueSerdeResolver keyValueSerdeResolver, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, + StreamListenerParameterAdapter streamListenerParameterAdapter, + Collection streamListenerResultAdapters, + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, + CleanupConfig cleanupConfig) { this.bindingServiceProperties = bindingServiceProperties; this.kafkaStreamsExtendedBindingProperties = kafkaStreamsExtendedBindingProperties; this.keyValueSerdeResolver = keyValueSerdeResolver; @@ -148,7 +149,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene 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)) { + if (parameterType.equals(KStream.class) || parameterType.equals(KTable.class) + || parameterType.equals(GlobalKTable.class)) { supports = true; } } @@ -281,6 +283,22 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene } arguments[parameterIndex] = table; } + else if (parameterType.isAssignableFrom(GlobalKTable.class)) { + String materializedAs = extendedConsumerProperties.getMaterializedAs(); + String bindingDestination = bindingServiceProperties.getBindingDestination(inboundName); + GlobalKTable table = materializedAs != null ? + materializedAsGlobalKTable(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde ) : + streamsBuilder.globalTable(bindingDestination, + Consumed.with(keySerde, valueSerde)); + GlobalKTableBoundElementFactory.GlobalKTableWrapper globalKTableWrapper = (GlobalKTableBoundElementFactory.GlobalKTableWrapper) targetBean; + //wrap the proxy created during the initial target type binding with real object (KTable) + globalKTableWrapper.wrap((GlobalKTable) table); + kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean); + if (streamsConfig != null){ + kafkaStreamsBindingInformationCatalogue.addStreamsConfigs(globalKTableWrapper, streamsConfig); + } + arguments[parameterIndex] = table; + } } catch (Exception e) { throw new IllegalStateException(e); @@ -294,11 +312,19 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene } 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)); + getMaterialized(storeName, k, v)); + } + + private GlobalKTable materializedAsGlobalKTable(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v) { + return streamsBuilder.globalTable(bindingServiceProperties.getBindingDestination(destination), + getMaterialized(storeName, k, v)); + } + + private Materialized> getMaterialized(String storeName, Serde k, Serde v) { + return Materialized.>as(storeName) + .withKeySerde(k) + .withValueSerde(v); } private StoreBuilder buildStateStore(KafkaStreamsStateStoreProperties spec) { @@ -336,7 +362,6 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene } } - private KStream getkStream(String inboundName, KafkaStreamsStateStoreProperties storeSpec, BindingProperties bindingProperties, StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde) { @@ -362,9 +387,9 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene stream = stream.mapValues(value -> { Object returnValue; String contentType = bindingProperties.getContentType(); - if (value != null && !StringUtils.isEmpty(contentType) && !nativeDecoding) { + if (!StringUtils.isEmpty(contentType) && !nativeDecoding) { returnValue = MessageBuilder.withPayload(value) - .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); + .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); } else { returnValue = value; @@ -375,11 +400,11 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene } private void enableNativeDecodingForKTableAlways(Class parameterType, BindingProperties bindingProperties) { - if (parameterType.isAssignableFrom(KTable.class)) { + if (parameterType.isAssignableFrom(KTable.class) || parameterType.isAssignableFrom(GlobalKTable.class)) { if (bindingProperties.getConsumer() == null) { bindingProperties.setConsumer(new ConsumerProperties()); } - //No framework level message conversion provided for KTable, its done by the broker. + //No framework level message conversion provided for KTable/GlobalKTable, its done by the broker. bindingProperties.getConsumer().setUseNativeDecoding(true); } } @@ -420,7 +445,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene streamsBuilder.setAutoStartup(false); BeanDefinition streamsBuilderBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition((Class) streamsBuilder.getClass(), () -> streamsBuilder) - .getRawBeanDefinition(); + .getRawBeanDefinition(); ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("stream-builder-" + method.getName(), streamsBuilderBeanDefinition); StreamsBuilderFactoryBean streamsBuilderX = applicationContext.getBean("&stream-builder-" + method.getName(), StreamsBuilderFactoryBean.class); BeanDefinition streamsConfigBeanDefinition = diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java index d98249ef5..fe765cbb3 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java @@ -20,7 +20,7 @@ import java.util.Set; import org.springframework.context.SmartLifecycle; import org.springframework.kafka.KafkaException; -import org.springframework.kafka.config.StreamsBuilderFactoryBean; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; /** * Iterate through all {@link StreamsBuilderFactoryBean} in the application context diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.binders b/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.binders index 9d7d8c882..dde457315 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.binders +++ b/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.binders @@ -2,5 +2,7 @@ kstream:\ org.springframework.cloud.stream.binder.kafka.streams.KStreamBinderConfiguration ktable:\ org.springframework.cloud.stream.binder.kafka.streams.KTableBinderConfiguration +globalktable:\ +org.springframework.cloud.stream.binder.kafka.streams.GlobalKTableBinderConfiguration diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java index bfd514bf5..5d3ed84c9 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java @@ -52,11 +52,11 @@ import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStr import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.integration.test.util.TestUtils; -import org.springframework.kafka.config.StreamsBuilderFactoryBean; import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java index e6be7bfae..629641ba2 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java @@ -39,11 +39,11 @@ import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.integration.test.util.TestUtils; -import org.springframework.kafka.config.StreamsBuilderFactoryBean; import org.springframework.kafka.core.CleanupConfig; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.StreamsBuilderFactoryBean; import org.springframework.kafka.support.serializer.JsonSerde; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java new file mode 100644 index 000000000..bef59adf3 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java @@ -0,0 +1,321 @@ +/* + * 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.kafka.streams.integration; + +import java.util.ArrayList; +import java.util.Comparator; +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.streams.KeyValue; +import org.apache.kafka.streams.kstream.GlobalKTable; +import org.apache.kafka.streams.kstream.KStream; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +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.kafka.streams.annotations.KafkaStreamsProcessor; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsApplicationSupportProperties; +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.support.serializer.JsonDeserializer; +import org.springframework.kafka.support.serializer.JsonSerde; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +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 StreamToGlobalKTableJoinIntegrationTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, "enriched-order"); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + + private static Consumer consumer; + + interface CustomGlobalKTableProcessor extends KafkaStreamsProcessor { + + @Input("inputX") + GlobalKTable inputX(); + + @Input("inputY") + GlobalKTable inputY(); + } + + @EnableBinding(CustomGlobalKTableProcessor.class) + @EnableAutoConfiguration + @EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class) + public static class OrderEnricherApplication { + + @StreamListener + @SendTo("output") + public KStream process(@Input("input") KStream ordersStream, + @Input("inputX") GlobalKTable customers, + @Input("inputY") GlobalKTable products) { + + KStream customerOrdersStream = ordersStream.join(customers, + (orderId, order) -> order.getCustomerId(), + (order, customer) -> new CustomerOrder(customer, order)); + + return customerOrdersStream.join(products, + (orderId, customerOrder) -> customerOrder + .productId(), + (customerOrder, product) -> { + EnrichedOrder enrichedOrder = new EnrichedOrder(); + enrichedOrder.setProduct(product); + enrichedOrder.setCustomer(customerOrder.customer); + enrichedOrder.setOrder(customerOrder.order); + return enrichedOrder; + }); + } + } + + @Test + public void testStreamToGlobalKTable() throws Exception { + SpringApplication app = new SpringApplication(StreamToGlobalKTableJoinIntegrationTests.OrderEnricherApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.destination=orders", + "--spring.cloud.stream.bindings.inputX.destination=customers", + "--spring.cloud.stream.bindings.inputY.destination=products", + "--spring.cloud.stream.bindings.output.destination=enriched-order", + "--spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.inputX.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.inputY.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$OrderSerde", + "--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$CustomerSerde", + "--spring.cloud.stream.kafka.streams.bindings.inputY.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.kafka.streams.bindings.inputY.consumer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$ProductSerde", + "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$EnrichedOrderSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=10000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) { + Map senderPropsCustomer = KafkaTestUtils.producerProps(embeddedKafka); + senderPropsCustomer.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class); + CustomerSerde customerSerde = new CustomerSerde(); + senderPropsCustomer.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, customerSerde.serializer().getClass()); + + DefaultKafkaProducerFactory pfCustomer = new DefaultKafkaProducerFactory<>(senderPropsCustomer); + KafkaTemplate template = new KafkaTemplate<>(pfCustomer, true); + template.setDefaultTopic("customers"); + for (long i = 0; i < 5; i++) { + final Customer customer = new Customer(); + customer.setName("customer-" + i); + template.sendDefault(i, customer); + } + + Map senderPropsProduct = KafkaTestUtils.producerProps(embeddedKafka); + senderPropsProduct.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class); + ProductSerde productSerde = new ProductSerde(); + senderPropsProduct.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, productSerde.serializer().getClass()); + + DefaultKafkaProducerFactory pfProduct = new DefaultKafkaProducerFactory<>(senderPropsProduct); + KafkaTemplate productTemplate = new KafkaTemplate<>(pfProduct, true); + productTemplate.setDefaultTopic("products"); + + for (long i = 0; i < 5; i++) { + final Product product = new Product(); + product.setName("product-" + i); + productTemplate.sendDefault(i, product); + } + + Map senderPropsOrder = KafkaTestUtils.producerProps(embeddedKafka); + senderPropsOrder.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class); + OrderSerde orderSerde = new OrderSerde(); + senderPropsOrder.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, orderSerde.serializer().getClass()); + + DefaultKafkaProducerFactory pfOrder = new DefaultKafkaProducerFactory<>(senderPropsOrder); + KafkaTemplate orderTemplate = new KafkaTemplate<>(pfOrder, true); + orderTemplate.setDefaultTopic("orders"); + + for (long i = 0; i < 5; i++) { + final Order order = new Order(); + order.setCustomerId(i); + order.setProductId(i); + orderTemplate.sendDefault(i, order); + } + + Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class); + EnrichedOrderSerde enrichedOrderSerde = new EnrichedOrderSerde(); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, enrichedOrderSerde.deserializer().getClass()); + consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests.EnrichedOrder"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "enriched-order"); + + int count = 0; + long start = System.currentTimeMillis(); + List> enrichedOrders = new ArrayList<>(); + do { + ConsumerRecords records = KafkaTestUtils.getRecords(consumer); + count = count + records.count(); + for (ConsumerRecord record : records) { + enrichedOrders.add(new KeyValue<>(record.key(), record.value())); + } + } while (count < 5 && (System.currentTimeMillis() - start) < 30000); + + assertThat(count == 5).isTrue(); + assertThat(enrichedOrders.size() == 5).isTrue(); + + enrichedOrders.sort(Comparator.comparing(o -> o.key)); + + for (int i = 0; i < 5; i++) { + KeyValue enrichedOrderKeyValue = enrichedOrders.get(i); + assertThat(enrichedOrderKeyValue.key == i).isTrue(); + EnrichedOrder enrichedOrder = enrichedOrderKeyValue.value; + assertThat(enrichedOrder.getOrder().customerId == i).isTrue(); + assertThat(enrichedOrder.getOrder().productId == i).isTrue(); + assertThat(enrichedOrder.getCustomer().name.equals("customer-" + i)).isTrue(); + assertThat(enrichedOrder.getProduct().name.equals("product-" + i)).isTrue(); + } + pfCustomer.destroy(); + pfProduct.destroy(); + pfOrder.destroy(); + consumer.close(); + } + + } + + static class Order { + + long customerId; + long productId; + + public long getCustomerId() { + return customerId; + } + + public void setCustomerId(long customerId) { + this.customerId = customerId; + } + + public long getProductId() { + return productId; + } + + public void setProductId(long productId) { + this.productId = productId; + } + } + + static class Customer { + + String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + + static class Product { + + String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + + static class EnrichedOrder { + + Product product; + Customer customer; + Order order; + + public Product getProduct() { + return product; + } + + public void setProduct(Product product) { + this.product = product; + } + + public Customer getCustomer() { + return customer; + } + + public void setCustomer(Customer customer) { + this.customer = customer; + } + + public Order getOrder() { + return order; + } + + public void setOrder(Order order) { + this.order = order; + } + } + + private static class CustomerOrder { + private final Customer customer; + private final Order order; + + CustomerOrder(final Customer customer, final Order order) { + this.customer = customer; + this.order = order; + } + + long productId() { + return order.getProductId(); + } + } + + public static class OrderSerde extends JsonSerde {} + public static class CustomerSerde extends JsonSerde {} + public static class ProductSerde extends JsonSerde {} + public static class EnrichedOrderSerde extends JsonSerde {} +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java index 106234873..af975c2c8 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java @@ -116,11 +116,11 @@ public class StreamToTableJoinIntegrationTests { } @Test - public void testStreamToTable() throws Exception { + public void testStreamToTable() { SpringApplication app = new SpringApplication(CountClicksPerRegionApplication.class); app.setWebApplicationType(WebApplicationType.NONE); - ConfigurableApplicationContext context = app.run("--server.port=0", + try (ConfigurableApplicationContext ignored = 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", @@ -141,8 +141,7 @@ public class StreamToTableJoinIntegrationTests { "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", "--spring.cloud.stream.bindings.inputX.consumer.headerMode=raw", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), - "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); - try { + "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) { // Input 1: Clicks per user (multiple records allowed per user). List> userClicks = Arrays.asList( new KeyValue<>("alice", 13L), @@ -163,7 +162,7 @@ public class StreamToTableJoinIntegrationTests { KafkaTemplate template = new KafkaTemplate<>(pf, true); template.setDefaultTopic("user-clicks"); - for (KeyValue keyValue : userClicks) { + for (KeyValue keyValue : userClicks) { template.sendDefault(keyValue.key, keyValue.value); } @@ -186,7 +185,7 @@ public class StreamToTableJoinIntegrationTests { KafkaTemplate template1 = new KafkaTemplate<>(pf1, true); template1.setDefaultTopic("user-regions"); - for (KeyValue keyValue : userRegions) { + for (KeyValue keyValue : userRegions) { template1.sendDefault(keyValue.key, keyValue.value); } @@ -206,17 +205,11 @@ public class StreamToTableJoinIntegrationTests { for (ConsumerRecord record : records) { actualClicksPerRegion.add(new KeyValue<>(record.key(), record.value())); } - } while (count < expectedClicksPerRegion.size() && (System.currentTimeMillis() - start) < 30000 ); + } 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(); - } } /**