diff --git a/pom.xml b/pom.xml index 1daecaac4..0ba422c88 100644 --- a/pom.xml +++ b/pom.xml @@ -24,6 +24,7 @@ spring-cloud-stream-binder-kafka-0.10.1-test spring-cloud-stream-binder-kafka-0.10.2-test spring-cloud-stream-binder-kafka-core + spring-cloud-stream-binder-kstream @@ -100,6 +101,17 @@ test ${kafka.version} + + org.apache.kafka + kafka-streams + ${kafka.version} + + + org.slf4j + slf4j-log4j12 + + + diff --git a/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml b/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml index 7eacccc27..398033726 100644 --- a/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml +++ b/spring-cloud-stream-binder-kafka/src/test/resources/logback.xml @@ -11,4 +11,4 @@ - \ No newline at end of file + diff --git a/spring-cloud-stream-binder-kstream/pom.xml b/spring-cloud-stream-binder-kstream/pom.xml new file mode 100644 index 000000000..1951b05e0 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/pom.xml @@ -0,0 +1,69 @@ + + + 4.0.0 + + spring-cloud-stream-binder-kstream + jar + spring-cloud-stream-binder-kstream + Kafka Streams Binder Implementation + + + org.springframework.cloud + spring-cloud-stream-binder-kafka-parent + 1.3.0.BUILD-SNAPSHOT + + + + + org.springframework.cloud + spring-cloud-stream-binder-kafka-core + + + org.springframework.boot + spring-boot-configuration-processor + true + + + org.springframework.cloud + spring-cloud-stream-codec + + + org.springframework.boot + spring-boot-autoconfigure + true + + + org.apache.kafka + kafka_2.11 + + + org.apache.kafka + kafka-streams + + + org.springframework.kafka + spring-kafka + + + org.springframework.boot + spring-boot-test + test + + + org.springframework.kafka + spring-kafka-test + test + + + org.apache.kafka + kafka_2.11 + test + test + + + org.springframework.cloud + spring-cloud-stream-binder-test + test + + + 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 new file mode 100644 index 000000000..272d139f8 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java @@ -0,0 +1,189 @@ +/* + * Copyright 2017 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.common.Configurable; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; + +import org.springframework.cloud.stream.binder.AbstractBinder; +import org.springframework.cloud.stream.binder.BinderHeaders; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.DefaultBinding; +import org.springframework.cloud.stream.binder.EmbeddedHeaderUtils; +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.HeaderMode; +import org.springframework.cloud.stream.binder.MessageValues; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; +import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +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.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.util.MimeType; +import org.springframework.util.StringUtils; + +/** + * @author Marius Bogoevici + */ +public class KStreamBinder extends + AbstractBinder, ExtendedConsumerProperties, ExtendedProducerProperties> + implements ExtendedPropertiesBinder, KStreamConsumerProperties, KStreamProducerProperties> { + + private String[] headers; + + private final KafkaTopicProvisioner kafkaTopicProvisioner; + + private final KStreamExtendedBindingProperties kStreamExtendedBindingProperties; + + private final StreamsConfig streamsConfig; + + public KStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, + KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) { + this.headers = EmbeddedHeaderUtils.headersToEmbed(binderConfigurationProperties.getHeaders()); + this.kafkaTopicProvisioner = kafkaTopicProvisioner; + this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties; + this.streamsConfig = streamsConfig; + } + + @Override + protected Binding> doBindConsumer(String name, String group, + KStream inputTarget, ExtendedConsumerProperties properties) { + + ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties( + new KafkaConsumerProperties()); + this.kafkaTopicProvisioner.provisionConsumerDestination(name, group, extendedConsumerProperties); + return new DefaultBinding<>(name, group, inputTarget, null); + } + + @Override + @SuppressWarnings("unchecked") + protected Binding> doBindProducer(String name, KStream outboundBindTarget, + ExtendedProducerProperties properties) { + ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties( + new KafkaProducerProperties()); + this.kafkaTopicProvisioner.provisionProducerDestination(name , extendedProducerProperties); + if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) { + outboundBindTarget = outboundBindTarget.map(new KeyValueMapper>() { + @Override + public KeyValue apply(Object k, Object v) { + if (v instanceof Message) { + try { + return new KeyValue<>(k, (Object)KStreamBinder.this.serializeAndEmbedHeadersIfApplicable((Message) v)); + } + catch (Exception e) { + throw new IllegalArgumentException(e); + } + } + else { + throw new IllegalArgumentException("Wrong type of message " + v); + } + } + }); + } + else { + if (!properties.isUseNativeEncoding()) { + outboundBindTarget = outboundBindTarget + .map(new KeyValueMapper>() { + @Override + public KeyValue apply(Object k, Object v) { + return KeyValue.pair(k, (Object)KStreamBinder.this.serializePayloadIfNecessary((Message) v)); + } + }); + } + else { + outboundBindTarget = outboundBindTarget + .map(new KeyValueMapper>() { + @Override + public KeyValue apply(Object k, Object v) { + return KeyValue.pair(k, ((Message) v).getPayload()); + } + }); + } + } + if (!properties.isUseNativeEncoding() || StringUtils.hasText(properties.getExtension().getKeySerde()) || StringUtils.hasText(properties.getExtension().getValueSerde())) { + Serde keySerde = Serdes.ByteArray(); + Serde valueSerde = Serdes.ByteArray(); + try { + if (StringUtils.hasText(properties.getExtension().getKeySerde())) { + keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class); + if (keySerde instanceof Configurable) { + ((Configurable) keySerde).configure(streamsConfig.originals()); + } + } + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("Serde class not found: ", e); + } + try { + if (StringUtils.hasText(properties.getExtension().getValueSerde())) { + valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class); + if (valueSerde instanceof Configurable) { + ((Configurable) valueSerde).configure(streamsConfig.originals()); + } + } + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("Serde class not found: ", e); + } + outboundBindTarget.to((Serde) keySerde, (Serde) valueSerde, name); + } + else { + outboundBindTarget.to(name); + } + return new DefaultBinding<>(name, null, outboundBindTarget, null); + } + + private byte[] serializeAndEmbedHeadersIfApplicable(Message message) throws Exception { + MessageValues transformed = serializePayloadIfNecessary(message); + byte[] payload; + + Object contentType = transformed.get(MessageHeaders.CONTENT_TYPE); + // transform content type headers to String, so that they can be properly embedded + // in JSON + if (contentType instanceof MimeType) { + transformed.put(MessageHeaders.CONTENT_TYPE, contentType.toString()); + } + Object originalContentType = transformed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE); + if (originalContentType instanceof MimeType) { + transformed.put(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, originalContentType.toString()); + } + payload = EmbeddedHeaderUtils.embedHeaders(transformed, headers); + return payload; + } + + @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/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java new file mode 100644 index 000000000..8c4a58459 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java @@ -0,0 +1,167 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KStreamBuilder; +import org.apache.kafka.streams.kstream.KeyValueMapper; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.EmbeddedHeaderUtils; +import org.springframework.cloud.stream.binder.HeaderMode; +import org.springframework.cloud.stream.binder.MessageSerializationUtils; +import org.springframework.cloud.stream.binder.MessageValues; +import org.springframework.cloud.stream.binder.StringConvertingContentTypeResolver; +import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.integration.codec.Codec; +import org.springframework.integration.support.MutableMessageHeaders; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.Assert; +import org.springframework.util.MimeType; +import org.springframework.util.StringUtils; + +/** + * @author Marius Bogoevici + */ +public class KStreamBoundElementFactory extends AbstractBindingTargetFactory { + + private final KStreamBuilder kStreamBuilder; + + private final BindingServiceProperties bindingServiceProperties; + + private volatile Codec codec; + + private final StringConvertingContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver(); + + private volatile Map> payloadTypeCache = new ConcurrentHashMap<>(); + + private CompositeMessageConverterFactory compositeMessageConverterFactory; + + public KStreamBoundElementFactory(KStreamBuilder streamBuilder, BindingServiceProperties bindingServiceProperties, + Codec codec, CompositeMessageConverterFactory compositeMessageConverterFactory) { + super(KStream.class); + this.bindingServiceProperties = bindingServiceProperties; + this.kStreamBuilder = streamBuilder; + this.codec = codec; + this.compositeMessageConverterFactory = compositeMessageConverterFactory; + } + + @Override + public KStream createInput(String name) { + KStream stream = kStreamBuilder.stream(bindingServiceProperties.getBindingDestination(name)); + ConsumerProperties properties = bindingServiceProperties.getConsumerProperties(name); + if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) { + + stream = stream.map(new KeyValueMapper>() { + @Override + public KeyValue apply(Object key, Object value) { + if (!(value instanceof byte[])) { + return new KeyValue<>(key, value); + } + try { + MessageValues messageValues = EmbeddedHeaderUtils + .extractHeaders(MessageBuilder.withPayload((byte[]) value).build(), true); + messageValues = deserializePayloadIfNecessary(messageValues); + return new KeyValue(null, messageValues.toMessage()); + } + catch (Exception e) { + throw new IllegalArgumentException(e); + } + } + }); + } + return stream; + } + + @Override + @SuppressWarnings("unchecked") + public KStream createOutput(final String name) { + BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name); + String contentType = bindingProperties.getContentType(); + MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory + .getMessageConverterForType(MimeType.valueOf(contentType)) : null; + KStreamWrapperHandler handler = new KStreamWrapperHandler(messageConverter); + ProxyFactory proxyFactory = new ProxyFactory(KStreamWrapper.class, KStream.class); + proxyFactory.addAdvice(handler); + return (KStream) proxyFactory.getProxy(); + } + + private MessageValues deserializePayloadIfNecessary(MessageValues messageValues) { + return MessageSerializationUtils.deserializePayload(messageValues, this.contentTypeResolver, this.codec); + } + + interface KStreamWrapper { + + void wrap(KStream delegate); + } + + static class KStreamWrapperHandler implements KStreamWrapper, MethodInterceptor { + + private KStream delegate; + + private final MessageConverter messageConverter; + + public KStreamWrapperHandler(MessageConverter messageConverter) { + this.messageConverter = messageConverter; + } + + public void wrap(KStream delegate) { + Assert.notNull(delegate, "delegate cannot be null"); + Assert.isNull(this.delegate, "delegate already set to " + this.delegate); + if (messageConverter != null) { + KeyValueMapper> keyValueMapper = new KeyValueMapper>() { + @Override + public KeyValue apply(Object k, Object v) { + Message message = (Message) v; + return new KeyValue(k, + messageConverter.toMessage(message.getPayload(), + new MutableMessageHeaders(((Message) v).getHeaders()))); + } + }; + delegate = delegate.map(keyValueMapper); + } + this.delegate = delegate; + } + + @Override + public Object invoke(MethodInvocation methodInvocation) throws Throwable { + if (methodInvocation.getMethod().getDeclaringClass().equals(KStream.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(KStreamWrapper.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/KStreamListenerParameterAdapter.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java new file mode 100644 index 000000000..01d00b5a6 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java @@ -0,0 +1,75 @@ +/* + * Copyright 2017 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.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; + +import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter; +import org.springframework.core.MethodParameter; +import org.springframework.core.ResolvableType; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.support.MessageBuilder; + +/** + * @author Marius Bogoevici + * @author Soby Chacko + */ +public class KStreamListenerParameterAdapter implements StreamListenerParameterAdapter, KStream> { + + private final MessageConverter messageConverter; + + public KStreamListenerParameterAdapter(MessageConverter messageConverter) { + this.messageConverter = messageConverter; + } + + @Override + public boolean supports(Class bindingTargetType, MethodParameter methodParameter) { + return KStream.class.isAssignableFrom(bindingTargetType) + && KStream.class.isAssignableFrom(methodParameter.getParameterType()); + } + + @Override + @SuppressWarnings("unchecked") + public KStream adapt(KStream bindingTarget, MethodParameter parameter) { + ResolvableType resolvableType = ResolvableType.forMethodParameter(parameter); + final Class valueClass = (resolvableType.getGeneric(1).getRawClass() != null) + ? (resolvableType.getGeneric(1).getRawClass()) : Object.class; + + return bindingTarget.map(new KeyValueMapper() { + @Override + public Object apply(Object o, Object o2) { + if (valueClass.isAssignableFrom(o2.getClass())) { + return new KeyValue<>(o, o2); + } + else if (o2 instanceof Message) { + return new KeyValue<>(o, messageConverter.fromMessage((Message) o2, valueClass)); + } + else if(o2 instanceof String || o2 instanceof byte[]) { + Message message = MessageBuilder.withPayload(o2).build(); + return new KeyValue<>(o, messageConverter.fromMessage(message, valueClass)); + } + else { + return new KeyValue<>(o, o2); + } + } + }); + } + +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamStreamListenerResultAdapter.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamStreamListenerResultAdapter.java new file mode 100644 index 000000000..5e4af0a8d --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamStreamListenerResultAdapter.java @@ -0,0 +1,65 @@ +/* + * Copyright 2017 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.io.Closeable; +import java.io.IOException; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; + +import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; + +/** + * @author Marius Bogoevici + */ +public class KStreamStreamListenerResultAdapter implements StreamListenerResultAdapter { + + @Override + public boolean supports(Class resultType, Class boundElement) { + return KStream.class.isAssignableFrom(resultType) && KStream.class.isAssignableFrom(boundElement); + } + + @Override + @SuppressWarnings("unchecked") + public Closeable adapt(KStream streamListenerResult, KStreamBoundElementFactory.KStreamWrapper boundElement) { + boundElement.wrap(streamListenerResult.map(new KeyValueMapper() { + @Override + public Object apply(Object k, Object v) { + if (v instanceof Message) { + return new KeyValue<>(k, v); + } + else { + return new KeyValue<>(k, MessageBuilder.withPayload(v).build()); + } + } + })); + return new NoOpCloseable(); + } + + private static final class NoOpCloseable implements Closeable { + + @Override + public void close() throws IOException { + + } + + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/annotations/KStreamProcessor.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/annotations/KStreamProcessor.java new file mode 100644 index 000000000..3a9434039 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/annotations/KStreamProcessor.java @@ -0,0 +1,34 @@ +/* + * Copyright 2017 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.annotations; + +import org.apache.kafka.streams.kstream.KStream; + +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Output; + +/** + * @author Marius Bogoevici + */ +public interface KStreamProcessor { + + @Input("input") + KStream input(); + + @Output("output") + KStream output(); +} 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 new file mode 100644 index 000000000..c4f628480 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java @@ -0,0 +1,96 @@ +/* + * Copyright 2017 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.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.common.utils.AppInfoParser; +import org.apache.kafka.streams.StreamsConfig; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binder.kstream.KStreamBinder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Condition; +import org.springframework.context.annotation.ConditionContext; +import org.springframework.context.annotation.Conditional; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.type.AnnotatedTypeMetadata; + +/** + * @author Marius Bogoevici + */ +@Configuration +@EnableConfigurationProperties(KStreamExtendedBindingProperties.class) +public class KStreamBinderConfiguration { + + @Autowired(required = false) + private AdminUtilsOperation adminUtilsOperation; + + private static final Log logger = LogFactory.getLog(KStreamBinderConfiguration.class); + + @Bean + public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties) { + return new KafkaTopicProvisioner(binderConfigurationProperties, adminUtilsOperation); + } + + @Bean + public KStreamBinder kStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, + KafkaTopicProvisioner kafkaTopicProvisioner, + KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) { + return new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, kStreamExtendedBindingProperties, + streamsConfig); + } + + @Bean(name = "adminUtilsOperation") + @Conditional(Kafka09Present.class) + @ConditionalOnClass(name = "kafka.admin.AdminUtils") + public AdminUtilsOperation kafka09AdminUtilsOperation() { + logger.info("AdminUtils selected: Kafka 0.9 AdminUtils"); + return new Kafka09AdminUtilsOperation(); + } + + @Bean(name = "adminUtilsOperation") + @Conditional(Kafka10Present.class) + @ConditionalOnClass(name = "kafka.admin.AdminUtils") + public AdminUtilsOperation kafka10AdminUtilsOperation() { + logger.info("AdminUtils selected: Kafka 0.10 AdminUtils"); + return new Kafka10AdminUtilsOperation(); + } + + static class Kafka10Present implements Condition { + + @Override + public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) { + return AppInfoParser.getVersion().startsWith("0.10"); + } + } + + static class Kafka09Present implements Condition { + + @Override + public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) { + return AppInfoParser.getVersion().startsWith("0.9"); + } + } +} 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 new file mode 100644 index 000000000..ee92fb200 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java @@ -0,0 +1,103 @@ +/* + * Copyright 2017 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 java.util.Properties; + +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.kstream.KStreamBuilder; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.UnsatisfiedDependencyException; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kstream.KStreamBoundElementFactory; +import org.springframework.cloud.stream.binder.kstream.KStreamListenerParameterAdapter; +import org.springframework.cloud.stream.binder.kstream.KStreamStreamListenerResultAdapter; +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.codec.Codec; +import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration; +import org.springframework.kafka.core.KStreamBuilderFactoryBean; +import org.springframework.util.ObjectUtils; + +/** + * @author Marius Bogoevici + */ +public class KStreamBinderSupportAutoConfiguration { + + @Bean + @ConfigurationProperties(prefix = "spring.cloud.stream.kstream.binder") + public KafkaBinderConfigurationProperties binderConfigurationProperties() { + return new KafkaBinderConfigurationProperties(); + } + + @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_KSTREAM_BUILDER_BEAN_NAME) + public KStreamBuilderFactoryBean defaultKStreamBuilder( + @Qualifier(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider streamsConfigProvider) { + StreamsConfig streamsConfig = streamsConfigProvider.getIfAvailable(); + if (streamsConfig != null) { + KStreamBuilderFactoryBean kStreamBuilderFactoryBean = new KStreamBuilderFactoryBean(streamsConfig); + kStreamBuilderFactoryBean.setPhase(Integer.MAX_VALUE - 500); + return kStreamBuilderFactoryBean; + } + else { + throw new UnsatisfiedDependencyException(KafkaStreamsDefaultConfiguration.class.getName(), + KafkaStreamsDefaultConfiguration.DEFAULT_KSTREAM_BUILDER_BEAN_NAME, "streamsConfig", + "There is no '" + KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME + + "' StreamsConfig bean in the application context.\n"); + } + } + + @Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) + public StreamsConfig streamsConfig(KafkaBinderConfigurationProperties binderConfigurationProperties) { + Properties props = new Properties(); + props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, binderConfigurationProperties.getKafkaConnectionString()); + props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName()); + props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName()); + props.put(StreamsConfig.APPLICATION_ID_CONFIG, "default"); + props.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, binderConfigurationProperties.getZkConnectionString()); + if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConfiguration())) { + props.putAll(binderConfigurationProperties.getConfiguration()); + } + return new StreamsConfig(props); + } + + @Bean + public KStreamStreamListenerResultAdapter kStreamStreamListenerResultAdapter() { + return new KStreamStreamListenerResultAdapter(); + } + + @Bean + public KStreamListenerParameterAdapter kStreamListenerParameterAdapter( + CompositeMessageConverterFactory compositeMessageConverterFactory) { + return new KStreamListenerParameterAdapter( + compositeMessageConverterFactory.getMessageConverterForAllRegistered()); + } + + @Bean + public KStreamBoundElementFactory kStreamBindableTargetFactory(KStreamBuilder kStreamBuilder, + BindingServiceProperties bindingServiceProperties, Codec codec, + CompositeMessageConverterFactory compositeMessageConverterFactory) { + return new KStreamBoundElementFactory(kStreamBuilder, bindingServiceProperties, codec, + compositeMessageConverterFactory); + } + +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBindingProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBindingProperties.java new file mode 100644 index 000000000..34f092286 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBindingProperties.java @@ -0,0 +1,43 @@ +/* + * Copyright 2017 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; + +/** + * @author Marius Bogoevici + */ +public class KStreamBindingProperties { + + private KStreamConsumerProperties consumer = new KStreamConsumerProperties(); + + private KStreamProducerProperties producer = new KStreamProducerProperties(); + + public KStreamConsumerProperties getConsumer() { + return consumer; + } + + public void setConsumer(KStreamConsumerProperties consumer) { + this.consumer = consumer; + } + + public KStreamProducerProperties getProducer() { + return producer; + } + + public void setProducer(KStreamProducerProperties producer) { + this.producer = producer; + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamCommonProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamCommonProperties.java new file mode 100644 index 000000000..973587d15 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamCommonProperties.java @@ -0,0 +1,43 @@ +/* + * Copyright 2017 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; + +/** + * @author Soby Chacko + */ +public class KStreamCommonProperties { + + private String keySerde; + + private String valueSerde; + + public String getKeySerde() { + return keySerde; + } + + public void setKeySerde(String keySerde) { + this.keySerde = keySerde; + } + + public String getValueSerde() { + return valueSerde; + } + + public void setValueSerde(String valueSerde) { + this.valueSerde = valueSerde; + } +} diff --git a/spring-cloud-stream-binder-kstream/src/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 new file mode 100644 index 000000000..263d2556b --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamConsumerProperties.java @@ -0,0 +1,24 @@ +/* + * Copyright 2017 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; + +/** + * @author Marius Bogoevici + */ +public class KStreamConsumerProperties extends KStreamCommonProperties { + +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamExtendedBindingProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamExtendedBindingProperties.java new file mode 100644 index 000000000..52c3152be --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamExtendedBindingProperties.java @@ -0,0 +1,61 @@ +/* + * Copyright 2017 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 java.util.HashMap; +import java.util.Map; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.stream.binder.ExtendedBindingProperties; + +/** + * @author Marius Bogoevici + */ +@ConfigurationProperties("spring.cloud.stream.kstream") +public class KStreamExtendedBindingProperties + implements ExtendedBindingProperties { + + private Map bindings = new HashMap<>(); + + public Map getBindings() { + return this.bindings; + } + + public void setBindings(Map bindings) { + this.bindings = bindings; + } + + @Override + public KStreamConsumerProperties getExtendedConsumerProperties(String binding) { + if (this.bindings.containsKey(binding) && this.bindings.get(binding).getConsumer() != null) { + return this.bindings.get(binding).getConsumer(); + } + else { + return new KStreamConsumerProperties(); + } + } + + @Override + public KStreamProducerProperties getExtendedProducerProperties(String binding) { + if (this.bindings.containsKey(binding) && this.bindings.get(binding).getProducer() != null) { + return this.bindings.get(binding).getProducer(); + } + else { + return new KStreamProducerProperties(); + } + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java new file mode 100644 index 000000000..a40e2c217 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java @@ -0,0 +1,24 @@ +/* + * Copyright 2017 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; + +/** + * @author Marius Bogoevici + */ +public class KStreamProducerProperties extends KStreamCommonProperties { + +} 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 new file mode 100644 index 000000000..0195e2832 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.binders @@ -0,0 +1,4 @@ +kstream:\ +org.springframework.cloud.stream.binder.kstream.config.KStreamBinderConfiguration + + diff --git a/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..7246dbc7c --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/resources/META-INF/spring.factories @@ -0,0 +1,4 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + org.springframework.cloud.stream.binder.kstream.config.KStreamBinderSupportAutoConfiguration + + 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 new file mode 100644 index 000000000..7b857b835 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java @@ -0,0 +1,158 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.serialization.LongDeserializer; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; +import org.apache.kafka.streams.kstream.Predicate; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.apache.kafka.streams.kstream.Windowed; +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.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +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.JsonSerde; +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 KStreamBinderPojoInputAndPrimitiveTypeOutputTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts-id"); + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName()); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception { + SpringApplication app = new SpringApplication(ProductCountApplication.class); + app.setWebEnvironment(false); + ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.cloud.stream.bindings.input.destination=foos", + "--spring.cloud.stream.bindings.output.destination=counts-id", + "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.bindings.output.producer.headerMode=raw", + "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "--spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde", + "--spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + receiveAndValidateFoo(context); + context.close(); + } + + private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception{ + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foos"); + template.sendDefault("{\"id\":\"123\"}"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id"); + + assertThat(cr.key().equals(123)); + assertThat(cr.value().equals(1L)); + } + + @EnableBinding(KStreamProcessor.class) + @EnableAutoConfiguration + public static class ProductCountApplication { + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + return input + .filter(new Predicate() { + + @Override + public boolean test(Object key, Product product) { + return product.getId() == 123; + } + }) + .map(new KeyValueMapper>() { + + @Override + public KeyValue apply(Object key, Product value) { + return new KeyValue<>(value, value); + } + }) + .groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class)) + .count(TimeWindows.of(5000), "id-count-store") + .toStream() + .map(new KeyValueMapper, Long, KeyValue>() { + + @Override + public KeyValue apply(Windowed key, Long value) { + return new KeyValue<>(key.key().id, value); + } + }); + } + } + + static class Product { + + Integer id; + + public Integer getId() { + return id; + } + + public void setId(Integer id) { + this.id = id; + } + } +} diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java new file mode 100644 index 000000000..a60a956bc --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java @@ -0,0 +1,253 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Arrays; +import java.util.Date; +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.common.serialization.Serdes; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.apache.kafka.streams.kstream.ValueMapper; +import org.apache.kafka.streams.kstream.Windowed; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KStreamBuilderFactoryBean; +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 Marius Bogoevici + * @author Soby Chacko + */ +public class KStreamBinderWordCountIntegrationTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts"); + + 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"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts"); + } + + @AfterClass + public static void tearDown() { + 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.cloud.stream.bindings.input.destination=words", + "--spring.cloud.stream.bindings.output.destination=counts", + "--spring.cloud.stream.bindings.output.contentType=application/json", + "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.bindings.output.producer.headerMode=raw", + "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + receiveAndValidate(context); + 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("foobar"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts"); + assertThat(cr.value().contains("\"word\":\"foobar\",\"count\":1")).isTrue(); + } + + @EnableBinding(KStreamProcessor.class) + @EnableAutoConfiguration + @EnableConfigurationProperties(WordCountProcessorProperties.class) + public static class WordCountProcessorApplication { + + @Autowired + private WordCountProcessorProperties processorProperties; + + @Autowired + private KStreamBuilderFactoryBean kafkaStreams; + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + + return input + .flatMapValues(new ValueMapper>() { + + @Override + public List apply(String value) { + return Arrays.asList(value.toLowerCase().split("\\W+")); + } + }) + .map(new KeyValueMapper>() { + + @Override + public KeyValue apply(Object key, String value) { + return new KeyValue<>(value, value); + } + }) + .groupByKey(Serdes.String(), Serdes.String()) + .count(configuredTimeWindow(), processorProperties.getStoreName()) + .toStream() + .map(new KeyValueMapper, Long, KeyValue>() { + + @Override + public KeyValue apply(Windowed key, Long value) { + return new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))); + } + }); + } + + /** + * Constructs a {@link TimeWindows} property. + * + * @return + */ + private TimeWindows configuredTimeWindow() { + return processorProperties.getAdvanceBy() > 0 + ? TimeWindows.of(processorProperties.getWindowLength()).advanceBy(processorProperties.getAdvanceBy()) + : TimeWindows.of(processorProperties.getWindowLength()); + } + } + + @ConfigurationProperties(prefix = "kstream.word.count") + static class WordCountProcessorProperties { + + private int windowLength = 5000; + + private int advanceBy = 0; + + private String storeName = "WordCounts"; + + int getWindowLength() { + return windowLength; + } + + public void setWindowLength(int windowLength) { + this.windowLength = windowLength; + } + + int getAdvanceBy() { + return advanceBy; + } + + public void setAdvanceBy(int advanceBy) { + this.advanceBy = advanceBy; + } + + String getStoreName() { + return storeName; + } + + public void setStoreName(String storeName) { + this.storeName = storeName; + } + } + + static class WordCount { + + private String word; + + private long count; + + private Date start; + + private Date end; + + WordCount(String word, long count, Date start, Date end) { + this.word = word; + this.count = count; + this.start = start; + this.end = end; + } + + public String getWord() { + return word; + } + + public void setWord(String word) { + this.word = word; + } + + public long getCount() { + return count; + } + + public void setCount(long count) { + this.count = count; + } + + public Date getStart() { + return start; + } + + public void setStart(Date start) { + this.start = start; + } + + public Date getEnd() { + return end; + } + + public void setEnd(Date end) { + this.end = end; + } + } + +} 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 new file mode 100644 index 000000000..8897e29a5 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java @@ -0,0 +1,184 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; +import org.apache.kafka.streams.kstream.Predicate; +import org.apache.kafka.streams.state.QueryableStoreTypes; +import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KStreamBuilderFactoryBean; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.serializer.JsonSerde; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.handler.annotation.SendTo; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Soby Chacko + */ +public class KStreamInteractiveQueryIntegrationTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts-id"); + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception { + SpringApplication app = new SpringApplication(ProductCountApplication.class); + app.setWebEnvironment(false); + ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.cloud.stream.bindings.input.destination=foos", + "--spring.cloud.stream.bindings.output.destination=counts-id", + "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.bindings.output.producer.headerMode=raw", + "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + receiveAndValidateFoo(context); + context.close(); + } + + private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception{ + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foos"); + template.sendDefault("{\"id\":\"123\"}"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id"); + assertThat(cr.value().contains("Count for product with ID 123: 1")).isTrue(); + + ProductCountApplication.Foo foo = context.getBean(ProductCountApplication.Foo.class); + assertThat(foo.getProductStock(123).equals(1L)); + } + + @EnableBinding(KStreamProcessor.class) + @EnableAutoConfiguration + public static class ProductCountApplication { + + @Autowired + private KStreamBuilderFactoryBean kStreamBuilderFactoryBean; + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + + return input + .filter(new Predicate() { + + @Override + public boolean test(Object key, Product product) { + return product.getId() == 123; + } + }) + .map(new KeyValueMapper>() { + + @Override + public KeyValue apply(Object key, Product value) { + return new KeyValue<>(value.id, value); + } + }) + .groupByKey(new Serdes.IntegerSerde(), new JsonSerde<>(Product.class)) + .count("prod-id-count-store") + .toStream() + .map(new KeyValueMapper>() { + + @Override + public KeyValue apply(Integer key, Long value) { + return new KeyValue<>(null, "Count for product with ID 123: " + value); + } + }); + } + + @Bean + public Foo foo(KStreamBuilderFactoryBean kStreamBuilderFactoryBean) { + return new Foo(kStreamBuilderFactoryBean); + } + + + static class Foo { + KStreamBuilderFactoryBean kStreamBuilderFactoryBean; + + Foo(KStreamBuilderFactoryBean kStreamBuilderFactoryBean) { + this.kStreamBuilderFactoryBean = kStreamBuilderFactoryBean; + } + + public Long getProductStock(Integer id) { + KafkaStreams streams = kStreamBuilderFactoryBean.getKafkaStreams(); + ReadOnlyKeyValueStore keyValueStore = + streams.store("prod-id-count-store", QueryableStoreTypes.keyValueStore()); + return (Long)keyValueStore.get(id); + } + } + + } + + static class Product { + + Integer id; + + public Integer getId() { + return id; + } + + public void setId(Integer id) { + this.id = id; + } + } +} diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java new file mode 100644 index 000000000..95da3b434 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java @@ -0,0 +1,153 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kstream; + +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; +import org.apache.kafka.streams.kstream.Predicate; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.apache.kafka.streams.kstream.Windowed; +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.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +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.JsonSerde; +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 Marius Bogoevici + * @author Soby Chacko + */ +public class KstreamBinderPojoInputStringOutputIntegrationTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts-id"); + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception { + SpringApplication app = new SpringApplication(ProductCountApplication.class); + app.setWebEnvironment(false); + ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.cloud.stream.bindings.input.destination=foos", + "--spring.cloud.stream.bindings.output.destination=counts-id", + "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.bindings.output.producer.headerMode=raw", + "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + receiveAndValidateFoo(context); + context.close(); + } + + private void receiveAndValidateFoo(ConfigurableApplicationContext context) throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foos"); + template.sendDefault("{\"id\":\"123\"}"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id"); + assertThat(cr.value().contains("Count for product with ID 123: 1")).isTrue(); + } + + @EnableBinding(KStreamProcessor.class) + @EnableAutoConfiguration + public static class ProductCountApplication { + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + + return input + .filter(new Predicate() { + + @Override + public boolean test(Object key, Product product) { + return product.getId() == 123; + } + }) + .map(new KeyValueMapper>() { + + @Override + public KeyValue apply(Object key, Product value) { + return new KeyValue<>(value, value); + } + }) + .groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class)) + .count(TimeWindows.of(5000), "id-count-store") + .toStream() + .map(new KeyValueMapper, Long, KeyValue>() { + + @Override + public KeyValue apply(Windowed key, Long value) { + return new KeyValue<>(null, "Count for product with ID 123: " + value); + } + }); + } + } + + static class Product { + + Integer id; + + public Integer getId() { + return id; + } + + public void setId(Integer id) { + this.id = id; + } + } +} diff --git a/spring-cloud-stream-binder-kstream/src/test/resources/logback.xml b/spring-cloud-stream-binder-kstream/src/test/resources/logback.xml new file mode 100644 index 000000000..98e9018df --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/resources/logback.xml @@ -0,0 +1,15 @@ + + + + %d{ISO8601} %5p %t %c{2}:%L - %m%n + + + + + + + + + + +