SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Introducing error handling for Kafka Streams binder

Browse Source 
- Add support for KIP-161: streams deserialization exception handlers
 - Provide out of the box LogAndContinue and LogAndFail exception handlers
 - Introduce a new exception handler that sends records in error on
   deserialization to a DLQ
 - Ensure that the exception handlers work in both when native decoding
   is enabled (Kafka is doing deserialization) and when native decoding
   is disabled (Binder is doing deserialization)
 - Enhancements on the logic of native decoding in binder.
 - General refactoring and cleanup
 - Adding more tests

 Resolves #275, #281

Introducing multiple StreamListener methods in Kafka Streams apps.

 - Multiple stream listener methods now have individual backing
   StreamBuilder and configuration.
 - Register StreamsBuilderFactoryBean programmatically

 Resolves #299, #300
This commit is contained in:
Soby Chacko
2018-01-19 21:37:21 -05:00
committed by Oleg Zhurakousky
parent b31f902905
commit f44923e480
27 changed files with 1626 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pom.xml
View File

@@ -12,8 +12,8 @@
</parent>
<properties>
<java.version>1.8</java.version>
<spring-kafka.version>2.1.2.RELEASE</spring-kafka.version>
<spring-integration-kafka.version>3.0.1.RELEASE</spring-integration-kafka.version>
<spring-kafka.version>2.1.3.BUILD-SNAPSHOT</spring-kafka.version>
<spring-integration-kafka.version>3.0.2.BUILD-SNAPSHOT</spring-integration-kafka.version>
<kafka.version>1.0.0</kafka.version>
<spring-cloud-stream.version>2.0.0.BUILD-SNAPSHOT</spring-cloud-stream.version>
</properties>

4
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
View File

@@ -387,6 +387,10 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
return this.consumerDestinationName;
}
public String getDlqName() {
return dlqName;
}
@Override
public String toString() {
return "KafkaConsumerDestination{" +

175
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java
View File

@@ -16,29 +16,31 @@
package org.springframework.cloud.stream.binder.kstream;
import org.apache.kafka.common.Configurable;
import java.util.Map;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
import org.apache.kafka.streams.errors.LogAndFailExceptionHandler;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.Produced;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultBinding;
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.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.KStreamBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties;
import org.springframework.cloud.stream.binder.kstream.config.KStreamExtendedBindingProperties;
import org.springframework.cloud.stream.binder.kstream.config.KStreamProducerProperties;
import org.springframework.kafka.core.StreamsBuilderFactoryBean;
import org.springframework.util.StringUtils;
/**
@@ -51,33 +53,111 @@ public class KStreamBinder extends
private final KafkaTopicProvisioner kafkaTopicProvisioner;
private final KStreamExtendedBindingProperties kStreamExtendedBindingProperties;
private KStreamExtendedBindingProperties kStreamExtendedBindingProperties = new KStreamExtendedBindingProperties();
private final StreamsConfig streamsConfig;
private final KStreamBinderConfigurationProperties binderConfigurationProperties;
private final KafkaBinderConfigurationProperties binderConfigurationProperties;
private final KStreamBoundMessageConversionDelegate kStreamBoundMessageConversionDelegate;
private final MessageConversionDelegate messageConversionDelegate;
private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue;
public KStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties,
private final KeyValueSerdeResolver keyValueSerdeResolver;
private final QueryableStoreRegistry queryableStoreRegistry;
public KStreamBinder(KStreamBinderConfigurationProperties binderConfigurationProperties,
KafkaTopicProvisioner kafkaTopicProvisioner,
KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig,
MessageConversionDelegate messageConversionDelegate) {
KStreamBoundMessageConversionDelegate kStreamBoundMessageConversionDelegate,
KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue,
KeyValueSerdeResolver keyValueSerdeResolver,
QueryableStoreRegistry queryableStoreRegistry) {
this.binderConfigurationProperties = binderConfigurationProperties;
this.kafkaTopicProvisioner = kafkaTopicProvisioner;
this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties;
this.streamsConfig = streamsConfig;
this.messageConversionDelegate = messageConversionDelegate;
this.kStreamBoundMessageConversionDelegate = kStreamBoundMessageConversionDelegate;
this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue;
this.keyValueSerdeResolver = keyValueSerdeResolver;
this.queryableStoreRegistry = queryableStoreRegistry;
}
@Override
@SuppressWarnings("unchecked")
protected Binding<KStream<Object, Object>> doBindConsumer(String name, String group,
KStream<Object, Object> inputTarget,
ExtendedConsumerProperties<KStreamConsumerProperties> properties) {
this.KStreamBindingInformationCatalogue.registerConsumerProperties(inputTarget, properties.getExtension());
ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties = new ExtendedConsumerProperties<>(
new KafkaConsumerProperties());
properties.getExtension());
if (properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.sendToDlq) {
extendedConsumerProperties.getExtension().setEnableDlq(true);
}
if (!StringUtils.hasText(group)) {
group = binderConfigurationProperties.getApplicationId();
}
this.kafkaTopicProvisioner.provisionConsumerDestination(name, group, extendedConsumerProperties);
//populate the per binding StreamConfig properties
Map<String, Object> streamConfigGlobalProperties = getApplicationContext().getBean("streamConfigGlobalProperties", Map.class);
StreamsBuilderFactoryBean streamsBuilder = getApplicationContext().getBean("&stream-builder-" + name, StreamsBuilderFactoryBean.class);
streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, group);
if(properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.logAndContinue) {
streamConfigGlobalProperties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
LogAndContinueExceptionHandler.class);
}
else if(properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.logAndFail) {
streamConfigGlobalProperties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
LogAndFailExceptionHandler.class);
}
else if (properties.getExtension().getSerdeError() == KStreamConsumerProperties.SerdeError.sendToDlq) {
streamConfigGlobalProperties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
SendToDlqAndContinue.class);
}
StreamsConfig streamsConfig = new StreamsConfig(streamConfigGlobalProperties) {
DeserializationExceptionHandler deserializationExceptionHandler;
@Override
@SuppressWarnings("unchecked")
public <T> T getConfiguredInstance(String key, Class<T> t) {
if (key.equals(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG)){
if (deserializationExceptionHandler != null){
return (T)deserializationExceptionHandler;
}
else {
T t1 = super.getConfiguredInstance(key, t);
deserializationExceptionHandler = (DeserializationExceptionHandler)t1;
return t1;
}
}
return super.getConfiguredInstance(key, t);
}
};
ConfigurableListableBeanFactory beanFactory = getApplicationContext().getBeanFactory();
beanFactory.registerSingleton("streamsConfig-" + name, streamsConfig);
beanFactory.initializeBean(streamsConfig, "streamsConfig-" + name);
streamsBuilder.setStreamsConfig(streamsConfig);
streamsBuilder.start();
queryableStoreRegistry.registerKafkaStreams(streamsBuilder.getKafkaStreams());
if (extendedConsumerProperties.getExtension().isEnableDlq()) {
String dlqName = StringUtils.isEmpty(extendedConsumerProperties.getExtension().getDlqName()) ?
"error." + name + "." + group : extendedConsumerProperties.getExtension().getDlqName();
KStreamDlqDispatch kStreamDlqDispatch = new KStreamDlqDispatch(dlqName, binderConfigurationProperties,
extendedConsumerProperties.getExtension());
SendToDlqAndContinue sendToDlqAndContinue = this.getApplicationContext().getBean(SendToDlqAndContinue.class);
sendToDlqAndContinue.addKStreamDlqDispatch(name, kStreamDlqDispatch);
DeserializationExceptionHandler deserializationExceptionHandler = streamsConfig.defaultDeserializationExceptionHandler();
if(deserializationExceptionHandler instanceof SendToDlqAndContinue) {
((SendToDlqAndContinue)deserializationExceptionHandler).addKStreamDlqDispatch(name, kStreamDlqDispatch);
}
}
return new DefaultBinding<>(name, group, inputTarget, null);
}
@@ -89,69 +169,20 @@ public class KStreamBinder extends
new KafkaProducerProperties());
this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties);
Serde<?> keySerde = getKeySerde(properties);
Serde<?> valueSerde = getValueSerde(properties);
Serde<?> keySerde = this.keyValueSerdeResolver.getOuboundKeySerde(properties.getExtension());
Serde<?> valueSerde = this.keyValueSerdeResolver.getOutboundValueSerde(properties, properties.getExtension());
to(properties.isUseNativeEncoding(), name, outboundBindTarget, (Serde<Object>) keySerde, (Serde<Object>) valueSerde);
return new DefaultBinding<>(name, null, outboundBindTarget, null);
}
private Serde<?> getKeySerde(ExtendedProducerProperties<KStreamProducerProperties> properties) {
Serde<?> keySerde;
try {
if (StringUtils.hasText(properties.getExtension().getKeySerde())) {
keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class);
if (keySerde instanceof Configurable) {
((Configurable) keySerde).configure(streamsConfig.originals());
}
}
else {
keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray();
}
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
}
return keySerde;
}
private Serde<?> getValueSerde(ExtendedProducerProperties<KStreamProducerProperties> properties) {
Serde<?> valueSerde;
try {
if (properties.isUseNativeEncoding()) {
if (StringUtils.hasText(properties.getExtension().getValueSerde())) {
valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class);
if (valueSerde instanceof Configurable) {
((Configurable) valueSerde).configure(streamsConfig.originals());
}
}
else {
valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray();
}
}
else {
valueSerde = Serdes.ByteArray();
}
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
}
return valueSerde;
}
@SuppressWarnings("unchecked")
private void to(boolean isNativeEncoding, String name, KStream<Object, Object> outboundBindTarget,
Serde<Object> keySerde, Serde<Object> valueSerde) {
KeyValueMapper<Object, Object, KeyValue<Object, Object>> keyValueMapper = null;
if (!isNativeEncoding) {
keyValueMapper = messageConversionDelegate.outboundKeyValueMapper(name);
}
if (!isNativeEncoding) {
outboundBindTarget.map(keyValueMapper).to(name, Produced.with(keySerde, valueSerde));
kStreamBoundMessageConversionDelegate.serializeOnOutbound(outboundBindTarget)
.to(name, Produced.with(keySerde, valueSerde));
}
else {
outboundBindTarget.to(name, Produced.with(keySerde, valueSerde));
@@ -167,4 +198,8 @@ public class KStreamBinder extends
public KStreamProducerProperties getExtendedProducerProperties(String channelName) {
return this.kStreamExtendedBindingProperties.getExtendedProducerProperties(channelName);
}
public void setkStreamExtendedBindingProperties(KStreamExtendedBindingProperties kStreamExtendedBindingProperties) {
this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties;
}
}

117
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBindingInformationCatalogue.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties;
import org.springframework.cloud.stream.config.BindingProperties;
/**
* A catalogue containing all the inbound and outboud KStreams.
* It registers {@link BindingProperties} and {@link KStreamConsumerProperties}
* for the bounded KStreams. This registry provides services for finding
* specific binding level information for the bounded KStream. This includes
* information such as the configured content type, destination etc.
*
* @since 2.0.0
*
* @author Soby Chacko
*/
public class KStreamBindingInformationCatalogue {
private final Map<KStream<?, ?>, BindingProperties> bindingProperties = new ConcurrentHashMap<>();
private final Map<KStream<?, ?>, KStreamConsumerProperties> consumerProperties = new ConcurrentHashMap<>();
/**
* For a given bounded {@link KStream}, retrieve it's corresponding destination
* on the broker.
*
* @param bindingTarget KStream binding target
* @return destination topic on Kafka
*/
public String getDestination(KStream<?,?> bindingTarget) {
BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget);
return bindingProperties.getDestination();
}
/**
* Is native decoding is enabled on this {@link KStream}.
*
* @param bindingTarget KStream binding target
* @return true if native decoding is enabled, fasle otherwise.
*/
public boolean isUseNativeDecoding(KStream<?,?> bindingTarget) {
BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget);
return bindingProperties.getConsumer().isUseNativeDecoding();
}
/**
* Is DLQ enabled for this {@link KStream}
*
* @param bindingTarget KStream binding target
* @return true if DLQ is enabled, false otherwise.
*/
public boolean isEnableDlq(KStream<?,?> bindingTarget) {
return consumerProperties.get(bindingTarget).isEnableDlq();
}
/**
* Retrieve the content type associated with a given {@link KStream}
*
* @param bindingTarget KStream binding target
* @return content Type associated.
*/
public String getContentType(KStream<?,?> bindingTarget) {
BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget);
return bindingProperties.getContentType();
}
/**
* Retrieve any configured Serde error handling strategies for this {@link KStream}
*
* @param bindingTarget KStream binding target
* @return configured Serde error handling strategy
*/
public KStreamConsumerProperties.SerdeError getSerdeError(KStream<?,?> bindingTarget) {
return consumerProperties.get(bindingTarget).getSerdeError();
}
/**
* Register a cache for bounded KStream -> {@link BindingProperties}
*
* @param bindingTarget KStream binding target
* @param bindingProperties {@link BindingProperties} for this KStream
*/
public void registerBindingProperties(KStream<?,?> bindingTarget, BindingProperties bindingProperties) {
this.bindingProperties.put(bindingTarget, bindingProperties);
}
/**
* Register a cache for bounded KStream -> {@link KStreamConsumerProperties}
*
* @param bindingTarget KStream binding target
* @param kStreamConsumerProperties Consumer properties for this KStream
*/
public void registerConsumerProperties(KStream<?,?> bindingTarget, KStreamConsumerProperties kStreamConsumerProperties) {
this.consumerProperties.put(bindingTarget, kStreamConsumerProperties);
}
}

78
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java
View File

@@ -18,14 +18,24 @@ package org.springframework.cloud.stream.binder.kstream;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.streams.Consumed;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties;
import org.springframework.cloud.stream.binder.kstream.config.KStreamExtendedBindingProperties;
import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.kafka.core.StreamsBuilderFactoryBean;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
@@ -36,27 +46,71 @@ import org.springframework.util.StringUtils;
* @author Marius Bogoevici
* @author Soby Chacko
*/
public class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KStream> {
private final StreamsBuilder kStreamBuilder;
public class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KStream> implements ApplicationContextAware {
private final BindingServiceProperties bindingServiceProperties;
public KStreamBoundElementFactory(StreamsBuilder kStreamBuilder, BindingServiceProperties bindingServiceProperties) {
private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue;
private final KeyValueSerdeResolver keyValueSerdeResolver;
private volatile AbstractApplicationContext applicationContext;
private KStreamExtendedBindingProperties kStreamExtendedBindingProperties = new KStreamExtendedBindingProperties();
public KStreamBoundElementFactory(BindingServiceProperties bindingServiceProperties,
KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue,
KeyValueSerdeResolver keyValueSerdeResolver) {
super(KStream.class);
this.bindingServiceProperties = bindingServiceProperties;
this.kStreamBuilder = kStreamBuilder;
this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue;
this.keyValueSerdeResolver = keyValueSerdeResolver;
}
public void setkStreamExtendedBindingProperties(KStreamExtendedBindingProperties kStreamExtendedBindingProperties) {
this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Assert.isInstanceOf(AbstractApplicationContext.class, applicationContext);
this.applicationContext = (AbstractApplicationContext) applicationContext;
}
@Override
public KStream createInput(String name) {
KStream<Object, Object> stream = kStreamBuilder.stream(bindingServiceProperties.getBindingDestination(name));
BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
String destination = bindingProperties.getDestination();
if (destination == null) {
destination = name;
}
KStreamConsumerProperties extendedConsumerProperties = kStreamExtendedBindingProperties.getExtendedConsumerProperties(name);
Serde<?> keySerde = this.keyValueSerdeResolver.getInboundKeySerde(extendedConsumerProperties);
Serde<?> valueSerde = this.keyValueSerdeResolver.getInboundValueSerde(bindingProperties.getConsumer(),
extendedConsumerProperties);
ConfigurableListableBeanFactory beanFactory = this.applicationContext.getBeanFactory();
StreamsBuilderFactoryBean streamsBuilder = new StreamsBuilderFactoryBean();
streamsBuilder.setAutoStartup(false);
beanFactory.registerSingleton("stream-builder-" + destination, streamsBuilder);
beanFactory.initializeBean(streamsBuilder, "stream-builder-" + destination);
StreamsBuilder streamBuilder = null;
try {
streamBuilder = streamsBuilder.getObject();
} catch (Exception e) {
//log and bail
}
KStream<?,?> stream = streamBuilder.stream(bindingServiceProperties.getBindingDestination(name),
Consumed.with(keySerde, valueSerde));
stream = stream.map((key, value) -> {
KeyValue<Object, Object> keyValue;
BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
String contentType = bindingProperties.getContentType();
if (!StringUtils.isEmpty(contentType) && !bindingProperties.getConsumer().isUseNativeDecoding()) {
Message<Object> message = MessageBuilder.withPayload(value)
Message<?> message = MessageBuilder.withPayload(value)
.setHeader(MessageHeaders.CONTENT_TYPE, contentType).build();
keyValue = new KeyValue<>(key, message);
}
@@ -65,6 +119,7 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KSt
}
return keyValue;
});
this.KStreamBindingInformationCatalogue.registerBindingProperties(stream, bindingProperties);
return stream;
}
@@ -74,7 +129,12 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KSt
KStreamWrapperHandler wrapper= new KStreamWrapperHandler();
ProxyFactory proxyFactory = new ProxyFactory(KStreamWrapper.class, KStream.class);
proxyFactory.addAdvice(wrapper);
return (KStream) proxyFactory.getProxy();
KStream proxy = (KStream) proxyFactory.getProxy();
BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
this.KStreamBindingInformationCatalogue.registerBindingProperties(proxy, bindingProperties);
return proxy;
}
public interface KStreamWrapper {

192
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundMessageConversionDelegate.java Normal file
View File

@@ -0,0 +1,192 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeType;
import org.springframework.util.StringUtils;
/**
* Delegate for handling all framework level message conversions inbound and outbound on {@link KStream}.
* If native encoding is not enabled, then serialization will be performed on outbound messages based
* on a contentType. Similarly, if native decoding is not enabled, deserialization will be performed on
* inbound messages based on a contentType. Based on the contentType, a {@link MessageConverter} will
* be resolved.
*
* @since 2.0.0
*
* @author Soby Chacko
*/
public class KStreamBoundMessageConversionDelegate {
private static final ThreadLocal<KeyValue<Object, Object>> keyValueThreadLocal = new ThreadLocal<>();
private final CompositeMessageConverterFactory compositeMessageConverterFactory;
private final SendToDlqAndContinue sendToDlqAndContinue;
private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue;
public KStreamBoundMessageConversionDelegate(CompositeMessageConverterFactory compositeMessageConverterFactory,
SendToDlqAndContinue sendToDlqAndContinue,
KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) {
this.compositeMessageConverterFactory = compositeMessageConverterFactory;
this.sendToDlqAndContinue = sendToDlqAndContinue;
this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue;
}
/**
* Serialize {@link KStream} records on outbound based on contentType.
*
* @param outboundBindTarget outbound KStream target
* @return serialized KStream
*/
public KStream serializeOnOutbound(KStream<?,?> outboundBindTarget) {
String contentType = this.KStreamBindingInformationCatalogue.getContentType(outboundBindTarget);
MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory
.getMessageConverterForType(MimeType.valueOf(contentType))
: null;
return outboundBindTarget.map((k, v) -> {
Message<?> message = v instanceof Message<?> ? (Message<?>) v :
MessageBuilder.withPayload(v).build();
Map<String, Object> headers = new HashMap<>(message.getHeaders());
if (!StringUtils.isEmpty(contentType)) {
headers.put(MessageHeaders.CONTENT_TYPE, contentType);
}
MessageHeaders messageHeaders = new MessageHeaders(headers);
return new KeyValue<>(k,
messageConverter.toMessage(message.getPayload(),
messageHeaders).getPayload());
});
}
/**
* Deserialize incoming {@link KStream} based on contentType.
*
* @param valueClass on KStream value
* @param bindingTarget inbound KStream target
* @return deserialized KStream
*/
@SuppressWarnings("unchecked")
public KStream deserializeOnInbound(Class<?> valueClass, KStream<?, ?> bindingTarget) {
MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
KStream<?, ?>[] branch = bindingTarget.branch(
(o, o2) -> {
boolean isValidRecord = false;
try {
if (valueClass.isAssignableFrom(o2.getClass())) {
keyValueThreadLocal.set(new KeyValue<>(o, o2));
}
else if (o2 instanceof Message) {
if (valueClass.isAssignableFrom(((Message) o2).getPayload().getClass())) {
keyValueThreadLocal.set(new KeyValue<>(o, ((Message) o2).getPayload()));
}
else {
convertAndSetMessage(o, valueClass, messageConverter, (Message) o2);
}
}
else if (o2 instanceof String || o2 instanceof byte[]) {
Message<?> message = MessageBuilder.withPayload(o2).build();
convertAndSetMessage(o, valueClass, messageConverter, message);
}
else {
keyValueThreadLocal.set(new KeyValue<>(o, o2));
}
isValidRecord = true;
}
catch (Exception ignored) {
//pass through
}
return isValidRecord;
},
(k, v) -> true
);
processErrorFromDeserialization(bindingTarget, branch[1]);
return branch[0].map((o, o2) -> {
KeyValue<Object, Object> objectObjectKeyValue = keyValueThreadLocal.get();
keyValueThreadLocal.remove();
return objectObjectKeyValue;
});
}
private void convertAndSetMessage(Object o, Class<?> valueClass, MessageConverter messageConverter, Message<?> msg) {
Object messageConverted = messageConverter.fromMessage(msg, valueClass);
if (messageConverted == null) {
throw new IllegalStateException("Inbound data conversion failed.");
}
keyValueThreadLocal.set(new KeyValue<>(o, messageConverted));
}
@SuppressWarnings("unchecked")
private void processErrorFromDeserialization(KStream<?, ?> bindingTarget, KStream<?, ?> branch) {
branch.process(() -> new Processor() {
ProcessorContext context;
@Override
public void init(ProcessorContext context) {
this.context = context;
}
@Override
public void process(Object o, Object o2) {
if (KStreamBindingInformationCatalogue.isEnableDlq(bindingTarget)) {
String destination = KStreamBindingInformationCatalogue.getDestination(bindingTarget);
if (o2 instanceof Message) {
Message message = (Message) o2;
sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) message.getPayload(), context.partition());
}
else {
sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) o2, context.partition());
}
}
else if (KStreamBindingInformationCatalogue.getSerdeError(bindingTarget) == KStreamConsumerProperties.SerdeError.logAndFail) {
throw new IllegalStateException("Inbound deserialization failed.");
}
else if (KStreamBindingInformationCatalogue.getSerdeError(bindingTarget) == KStreamConsumerProperties.SerdeError.logAndContinue) {
//quietly pass through. No action needed, this is similar to log and continue.
}
}
@Override
public void punctuate(long timestamp) {
}
@Override
public void close() {
}
});
}
}

144
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamDlqDispatch.java Normal file
View File

@@ -0,0 +1,144 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
/**
* @author Soby Chacko
*/
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.ObjectUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
class KStreamDlqDispatch {
private final Log logger = LogFactory.getLog(getClass());
private final KafkaTemplate<byte[],byte[]> kafkaTemplate;
private final String dlqName;
KStreamDlqDispatch(String dlqName,
KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties,
KafkaConsumerProperties kafkaConsumerProperties) {
ProducerFactory<byte[],byte[]> producerFactory = getProducerFactory(null,
new ExtendedProducerProperties<>(kafkaConsumerProperties.getDlqProducerProperties()),
kafkaBinderConfigurationProperties);
this.kafkaTemplate = new KafkaTemplate<>(producerFactory);
this.dlqName = dlqName;
}
@SuppressWarnings("unchecked")
public void sendToDlq(byte[] key, byte[] value, int partittion) {
ProducerRecord<byte[],byte[]> producerRecord = new ProducerRecord<>(this.dlqName, partittion,
key, value, null);
StringBuilder sb = new StringBuilder().append(" a message with key='")
.append(toDisplayString(ObjectUtils.nullSafeToString(key), 50)).append("'")
.append(" and payload='")
.append(toDisplayString(ObjectUtils.nullSafeToString(value), 50))
.append("'").append(" received from ")
.append(partittion);
ListenableFuture<SendResult<byte[],byte[]>> sentDlq = null;
try {
sentDlq = this.kafkaTemplate.send(producerRecord);
sentDlq.addCallback(new ListenableFutureCallback<SendResult<byte[],byte[]>>() {
@Override
public void onFailure(Throwable ex) {
KStreamDlqDispatch.this.logger.error(
"Error sending to DLQ " + sb.toString(), ex);
}
@Override
public void onSuccess(SendResult<byte[],byte[]> result) {
if (KStreamDlqDispatch.this.logger.isDebugEnabled()) {
KStreamDlqDispatch.this.logger.debug(
"Sent to DLQ " + sb.toString());
}
}
});
}
catch (Exception ex) {
if (sentDlq == null) {
KStreamDlqDispatch.this.logger.error(
"Error sending to DLQ " + sb.toString(), ex);
}
}
}
private DefaultKafkaProducerFactory<byte[],byte[]> getProducerFactory(String transactionIdPrefix,
ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
KafkaBinderConfigurationProperties configurationProperties) {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(configurationProperties.getRequiredAcks()));
if (!ObjectUtils.isEmpty(configurationProperties.getProducerConfiguration())) {
props.putAll(configurationProperties.getProducerConfiguration());
}
if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) {
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString());
}
if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) {
props.put(ProducerConfig.BATCH_SIZE_CONFIG,
String.valueOf(producerProperties.getExtension().getBufferSize()));
}
if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) {
props.put(ProducerConfig.LINGER_MS_CONFIG,
String.valueOf(producerProperties.getExtension().getBatchTimeout()));
}
if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) {
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
producerProperties.getExtension().getCompressionType().toString());
}
if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) {
props.putAll(producerProperties.getExtension().getConfiguration());
}
//Always send as byte[] on dlq (the same byte[] that the consumer received)
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
DefaultKafkaProducerFactory<byte[],byte[]> producerFactory = new DefaultKafkaProducerFactory<>(props);
if (transactionIdPrefix != null) {
producerFactory.setTransactionIdPrefix(transactionIdPrefix);
}
return producerFactory;
}
private String toDisplayString(String original, int maxCharacters) {
if (original.length() <= maxCharacters) {
return original;
}
return original.substring(0, maxCharacters) + "...";
}
}

20
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java
View File

@@ -16,7 +16,9 @@
package org.springframework.cloud.stream.binder.kstream;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
import org.springframework.core.MethodParameter;
@@ -28,10 +30,13 @@ import org.springframework.core.ResolvableType;
*/
public class KStreamListenerParameterAdapter implements StreamListenerParameterAdapter<KStream<?,?>, KStream<?, ?>> {
private final MessageConversionDelegate messageConversionDelegate;
private final KStreamBoundMessageConversionDelegate kStreamBoundMessageConversionDelegate;
private final KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue;
public KStreamListenerParameterAdapter(MessageConversionDelegate messageConversionDelegate) {
this.messageConversionDelegate = messageConversionDelegate;
public KStreamListenerParameterAdapter(KStreamBoundMessageConversionDelegate kStreamBoundMessageConversionDelegate,
KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) {
this.kStreamBoundMessageConversionDelegate = kStreamBoundMessageConversionDelegate;
this.KStreamBindingInformationCatalogue = KStreamBindingInformationCatalogue;
}
@Override
@@ -46,8 +51,11 @@ public class KStreamListenerParameterAdapter implements StreamListenerParameterA
ResolvableType resolvableType = ResolvableType.forMethodParameter(parameter);
final Class<?> valueClass = (resolvableType.getGeneric(1).getRawClass() != null)
? (resolvableType.getGeneric(1).getRawClass()) : Object.class;
return bindingTarget.map(messageConversionDelegate.inboundKeyValueMapper(valueClass));
if (this.KStreamBindingInformationCatalogue.isUseNativeDecoding(bindingTarget)) {
return bindingTarget.map((KeyValueMapper) KeyValue::new);
}
else {
return kStreamBoundMessageConversionDelegate.deserializeOnInbound(valueClass, bindingTarget);
}
}
}

2
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerSetupMethodOrchestrator.java
View File

@@ -42,6 +42,8 @@ import org.springframework.util.StringUtils;
* Kafka Streams specific implementation for {@link StreamListenerSetupMethodOrchestrator}
* that overrides the default mechanisms for invoking StreamListener adapters.
*
* @since 2.0.0
*
* @author Soby Chacko
*/
public class KStreamListenerSetupMethodOrchestrator implements StreamListenerSetupMethodOrchestrator, ApplicationContextAware {

165
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KeyValueSerdeResolver.java Normal file
View File

@@ -0,0 +1,165 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.util.Map;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Utils;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.binder.kstream.config.KStreamBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties;
import org.springframework.cloud.stream.binder.kstream.config.KStreamProducerProperties;
import org.springframework.util.StringUtils;
/**
* Resolver for key and value Serde.
*
* On the inbound, if native decoding is enabled, then any deserialization on the value is handled by Kafka.
* First, we look for any key/value Serde set on the binding itself, if that is not available then look at the
* common Serde set at the global level. If that fails, it falls back to byte[].
* If native decoding is disabled, then the binder will do the deserialization on value and ignore any Serde set for value
* and rely on the contentType provided. Keys are always deserialized at the broker.
*
* Same rules apply on the outbound. If native encoding is enabled, then value serialization is done at the broker using
* any binder level Serde for value, if not using common Serde, if not, then byte[].
* If native encoding is disabled, then the binder will do serialization using a contentType. Keys are always serialized
* by the broker.
*
* @since 2.0.0
*
* @author Soby Chacko
*/
public class KeyValueSerdeResolver {
private final Map<String,Object> streamConfigGlobalProperties;
private final KStreamBinderConfigurationProperties binderConfigurationProperties;
public KeyValueSerdeResolver(Map<String,Object> streamConfigGlobalProperties,
KStreamBinderConfigurationProperties binderConfigurationProperties) {
this.streamConfigGlobalProperties = streamConfigGlobalProperties;
this.binderConfigurationProperties = binderConfigurationProperties;
}
/**
* Provide the {@link Serde} for inbound key
*
* @param extendedConsumerProperties binding level extended {@link KStreamConsumerProperties}
* @return configurd {@link Serde} for the inbound key.
*/
public Serde<?> getInboundKeySerde(KStreamConsumerProperties extendedConsumerProperties) {
String keySerdeString = extendedConsumerProperties.getKeySerde();
return getKeySerde(keySerdeString);
}
/**
* Provide the {@link Serde} for inbound value
*
* @param consumerProperties {@link ConsumerProperties} on binding
* @param extendedConsumerProperties binding level extended {@link KStreamConsumerProperties}
* @return configurd {@link Serde} for the inbound value.
*/
public Serde<?> getInboundValueSerde(ConsumerProperties consumerProperties, KStreamConsumerProperties extendedConsumerProperties) {
Serde<?> valueSerde;
String valueSerdeString = extendedConsumerProperties.getValueSerde();
try {
if (consumerProperties != null &&
consumerProperties.isUseNativeDecoding()) {
valueSerde = getValueSerde(valueSerdeString);
}
else {
valueSerde = Serdes.ByteArray();
}
valueSerde.configure(streamConfigGlobalProperties, false);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
}
return valueSerde;
}
/**
* Provide the {@link Serde} for outbound key
*
* @param properties binding level extended {@link KStreamProducerProperties}
* @return configurd {@link Serde} for the outbound key.
*/
public Serde<?> getOuboundKeySerde(KStreamProducerProperties properties) {
return getKeySerde(properties.getKeySerde());
}
/**
* Provide the {@link Serde} for outbound value
*
* @param producerProperties {@link ProducerProperties} on binding
* @param kStreamProducerProperties binding level extended {@link KStreamProducerProperties}
* @return configurd {@link Serde} for the outbound value.
*/
public Serde<?> getOutboundValueSerde(ProducerProperties producerProperties, KStreamProducerProperties kStreamProducerProperties) {
Serde<?> valueSerde;
try {
if (producerProperties.isUseNativeEncoding()) {
valueSerde = getValueSerde(kStreamProducerProperties.getValueSerde());
}
else {
valueSerde = Serdes.ByteArray();
}
valueSerde.configure(streamConfigGlobalProperties, false);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
}
return valueSerde;
}
private Serde<?> getKeySerde(String keySerdeString) {
Serde<?> keySerde;
try {
if (StringUtils.hasText(keySerdeString)) {
keySerde = Utils.newInstance(keySerdeString, Serde.class);
}
else {
keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray();
}
keySerde.configure(streamConfigGlobalProperties, true);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
}
return keySerde;
}
private Serde<?> getValueSerde(String valueSerdeString) throws ClassNotFoundException {
Serde<?> valueSerde;
if (StringUtils.hasText(valueSerdeString)) {
valueSerde = Utils.newInstance(valueSerdeString, Serde.class);
}
else {
valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray();
}
return valueSerde;
}
}

96
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/MessageConversionDelegate.java
View File

@@ -1,96 +0,0 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeType;
import org.springframework.util.StringUtils;
/**
* @author Soby Chacko
*/
public class MessageConversionDelegate {
private final BindingServiceProperties bindingServiceProperties;
private final CompositeMessageConverterFactory compositeMessageConverterFactory;
public MessageConversionDelegate(BindingServiceProperties bindingServiceProperties,
CompositeMessageConverterFactory compositeMessageConverterFactory) {
this.bindingServiceProperties = bindingServiceProperties;
this.compositeMessageConverterFactory = compositeMessageConverterFactory;
}
public KeyValueMapper<Object, Object, KeyValue<Object, Object>> outboundKeyValueMapper(String name) {
BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
String contentType = bindingProperties.getContentType();
MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory
.getMessageConverterForType(MimeType.valueOf(contentType))
: null;
return (k, v) -> {
Message<?> message = v instanceof Message<?> ? (Message<?>)v :
MessageBuilder.withPayload(v).build();
Map<String, Object> headers = new HashMap<>(message.getHeaders());
if (!StringUtils.isEmpty(contentType)) {
headers.put(MessageHeaders.CONTENT_TYPE, contentType);
}
MessageHeaders messageHeaders = new MessageHeaders(headers);
return new KeyValue<>(k,
messageConverter.toMessage(message.getPayload(),
messageHeaders).getPayload());
};
}
@SuppressWarnings("unchecked")
public KeyValueMapper<Object, Object, KeyValue<Object, Object>> inboundKeyValueMapper(Class<?> valueClass) {
MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
return (KeyValueMapper) (o, o2) -> {
KeyValue<Object, Object> keyValue;
if (valueClass.isAssignableFrom(o2.getClass())) {
keyValue = new KeyValue<>(o, o2);
}
else if (o2 instanceof Message) {
if (valueClass.isAssignableFrom(((Message) o2).getPayload().getClass())) {
keyValue = new KeyValue<>(o, ((Message) o2).getPayload());
}
else {
keyValue = new KeyValue<>(o, messageConverter.fromMessage((Message) o2, valueClass));
}
}
else if(o2 instanceof String || o2 instanceof byte[]) {
Message<Object> message = MessageBuilder.withPayload(o2).build();
keyValue = new KeyValue<>(o, messageConverter.fromMessage(message, valueClass));
}
else {
keyValue = new KeyValue<>(o, o2);
}
return keyValue;
};
}
}

63
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/QueryableStoreRegistry.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.util.HashSet;
import java.util.Set;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.state.QueryableStoreType;
/**
* Registry that contains {@link QueryableStoreType}s those created from
* the user applications.
*
* @since 2.0.0
* @author Soby Chacko
*/
public class QueryableStoreRegistry {
private final Set<KafkaStreams> kafkaStreams = new HashSet<>();
/**
* Retrieve and return a queryable store by name created in the application.
*
* @param storeName name of the queryable store
* @param storeType type of the queryable store
* @param <T> generic queryable store
* @return queryable store.
*/
public <T> T getQueryableStoreType(String storeName, QueryableStoreType<T> storeType) {
for (KafkaStreams kafkaStream : kafkaStreams) {
T store = kafkaStream.store(storeName, storeType);
if (store != null) {
return store;
}
}
return null;
}
/**
* Register the {@link KafkaStreams} object created in the application.
*
* @param kafkaStreams {@link KafkaStreams} object created in the application
*/
public void registerKafkaStreams(KafkaStreams kafkaStreams) {
this.kafkaStreams.add(kafkaStreams);
}
}

95
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/SendToDlqAndContinue.java Normal file
View File

@@ -0,0 +1,95 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.internals.ProcessorContextImpl;
import org.apache.kafka.streams.processor.internals.StreamTask;
import org.springframework.util.ReflectionUtils;
/**
* Custom implementation for {@link DeserializationExceptionHandler} that sends the records
* in error to a DLQ topic, then continue stream processing on new records.
*
* @since 2.0.0
*
* @author Soby Chacko
*/
public class SendToDlqAndContinue implements DeserializationExceptionHandler{
private Map<String, KStreamDlqDispatch> dlqDispatchers = new HashMap<>();
public void sendToDlq(String topic, byte[] key, byte[] value, int partittion){
KStreamDlqDispatch kStreamDlqDispatch = dlqDispatchers.get(topic);
kStreamDlqDispatch.sendToDlq(key,value, partittion);
}
@Override
public DeserializationHandlerResponse handle(ProcessorContext context, ConsumerRecord<byte[], byte[]> record, Exception exception) {
KStreamDlqDispatch kStreamDlqDispatch = dlqDispatchers.get(record.topic());
kStreamDlqDispatch.sendToDlq(record.key(), record.value(), record.partition());
context.commit();
// The following conditional block should be reconsidered when we have a solution for this SO problem:
// https://stackoverflow.com/questions/48470899/kafka-streams-deserialization-handler
// Currently it seems like when deserialization error happens, there is no commits happening and the
// following code will use reflection to get access to the underlying KafkaConsumer.
// It works with Kafka 1.0.0, but there is no guarantee it will work in future versions of kafka as
// we access private fields by name using reflection, but it is a temporary fix.
if (context instanceof ProcessorContextImpl){
ProcessorContextImpl processorContextImpl = (ProcessorContextImpl)context;
Field task = ReflectionUtils.findField(ProcessorContextImpl.class, "task");
ReflectionUtils.makeAccessible(task);
Object taskField = ReflectionUtils.getField(task, processorContextImpl);
if (taskField.getClass().isAssignableFrom(StreamTask.class)){
StreamTask streamTask = (StreamTask)taskField;
Field consumer = ReflectionUtils.findField(StreamTask.class, "consumer");
ReflectionUtils.makeAccessible(consumer);
Object kafkaConsumerField = ReflectionUtils.getField(consumer, streamTask);
if (kafkaConsumerField.getClass().isAssignableFrom(KafkaConsumer.class)){
KafkaConsumer kafkaConsumer = (KafkaConsumer)kafkaConsumerField;
final Map<TopicPartition, OffsetAndMetadata> consumedOffsetsAndMetadata = new HashMap<>();
TopicPartition tp = new TopicPartition(record.topic(), record.partition());
OffsetAndMetadata oam = new OffsetAndMetadata(record.offset() + 1);
consumedOffsetsAndMetadata.put(tp, oam);
kafkaConsumer.commitSync(consumedOffsetsAndMetadata);
}
}
}
return DeserializationHandlerResponse.CONTINUE;
}
@Override
public void configure(Map<String, ?> configs) {
}
public void addKStreamDlqDispatch(String topic, KStreamDlqDispatch kStreamDlqDispatch){
dlqDispatchers.put(topic, kStreamDlqDispatch);
}
}

24
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java
View File

@@ -18,15 +18,16 @@ package org.springframework.cloud.stream.binder.kstream.config;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.streams.StreamsConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
import org.springframework.cloud.stream.binder.kstream.KStreamBinder;
import org.springframework.cloud.stream.binder.kstream.MessageConversionDelegate;
import org.springframework.cloud.stream.binder.kstream.KStreamBindingInformationCatalogue;
import org.springframework.cloud.stream.binder.kstream.KStreamBoundMessageConversionDelegate;
import org.springframework.cloud.stream.binder.kstream.KeyValueSerdeResolver;
import org.springframework.cloud.stream.binder.kstream.QueryableStoreRegistry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -36,7 +37,6 @@ import org.springframework.context.annotation.Configuration;
* @author Soby Chacko
*/
@Configuration
@EnableConfigurationProperties(KStreamExtendedBindingProperties.class)
public class KStreamBinderConfiguration {
private static final Log logger = LogFactory.getLog(KStreamBinderConfiguration.class);
@@ -44,18 +44,24 @@ public class KStreamBinderConfiguration {
@Autowired
private KafkaProperties kafkaProperties;
@Autowired
private KStreamExtendedBindingProperties kStreamExtendedBindingProperties;
@Bean
public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties) {
return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties);
}
@Bean
public KStreamBinder kStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties,
public KStreamBinder kStreamBinder(KStreamBinderConfigurationProperties binderConfigurationProperties,
KafkaTopicProvisioner kafkaTopicProvisioner,
KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig,
MessageConversionDelegate messageConversionDelegate) {
KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, kStreamExtendedBindingProperties,
streamsConfig, messageConversionDelegate);
KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate,
KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue,
KeyValueSerdeResolver keyValueSerdeResolver, QueryableStoreRegistry queryableStoreRegistry) {
KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner,
KStreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue,
keyValueSerdeResolver, queryableStoreRegistry);
kStreamBinder.setkStreamExtendedBindingProperties(kStreamExtendedBindingProperties);
return kStreamBinder;
}

25
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamCommonProperties.java → spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfigurationProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,28 +16,21 @@
package org.springframework.cloud.stream.binder.kstream.config;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
/**
* @author Soby Chacko
*/
public class KStreamCommonProperties {
public class KStreamBinderConfigurationProperties extends KafkaBinderConfigurationProperties {
private String keySerde;
private String applicationId = "default";
private String valueSerde;
public String getKeySerde() {
return keySerde;
public String getApplicationId() {
return applicationId;
}
public void setKeySerde(String keySerde) {
this.keySerde = keySerde;
public void setApplicationId(String applicationId) {
this.applicationId = applicationId;
}
public String getValueSerde() {
return valueSerde;
}
public void setValueSerde(String valueSerde) {
this.valueSerde = valueSerde;
}
}

93
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java
View File

@@ -17,69 +17,56 @@
package org.springframework.cloud.stream.binder.kstream.config;
import java.util.Collection;
import java.util.Properties;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.UnsatisfiedDependencyException;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.kstream.KStreamBindingInformationCatalogue;
import org.springframework.cloud.stream.binder.kstream.KStreamBoundElementFactory;
import org.springframework.cloud.stream.binder.kstream.KStreamBoundMessageConversionDelegate;
import org.springframework.cloud.stream.binder.kstream.KStreamListenerParameterAdapter;
import org.springframework.cloud.stream.binder.kstream.KStreamListenerSetupMethodOrchestrator;
import org.springframework.cloud.stream.binder.kstream.KStreamStreamListenerResultAdapter;
import org.springframework.cloud.stream.binder.kstream.MessageConversionDelegate;
import org.springframework.cloud.stream.binder.kstream.KeyValueSerdeResolver;
import org.springframework.cloud.stream.binder.kstream.QueryableStoreRegistry;
import org.springframework.cloud.stream.binder.kstream.SendToDlqAndContinue;
import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.core.StreamsBuilderFactoryBean;
import org.springframework.util.ObjectUtils;
/**
* @author Marius Bogoevici
* @author Soby Chacko
*/
@EnableConfigurationProperties(KStreamExtendedBindingProperties.class)
public class KStreamBinderSupportAutoConfiguration {
@Bean
@ConfigurationProperties(prefix = "spring.cloud.stream.kstream.binder")
public KafkaBinderConfigurationProperties binderConfigurationProperties() {
return new KafkaBinderConfigurationProperties();
public KStreamBinderConfigurationProperties binderConfigurationProperties() {
return new KStreamBinderConfigurationProperties();
}
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME)
public StreamsBuilderFactoryBean defaultKafkaStreamBuilder(
@Qualifier(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider<StreamsConfig> streamsConfigProvider) {
StreamsConfig streamsConfig = streamsConfigProvider.getIfAvailable();
if (streamsConfig != null) {
StreamsBuilderFactoryBean kStreamBuilderFactoryBean = new StreamsBuilderFactoryBean(streamsConfig);
kStreamBuilderFactoryBean.setPhase(Integer.MAX_VALUE - 500);
return kStreamBuilderFactoryBean;
} else {
throw new UnsatisfiedDependencyException(KafkaStreamsDefaultConfiguration.class.getName(),
KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME, "streamsConfig",
"There is no '" + KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME
+ "' StreamsConfig bean in the application context.\n");
}
}
@Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public StreamsConfig streamsConfig(KafkaBinderConfigurationProperties binderConfigurationProperties) {
Properties props = new Properties();
@Bean("streamConfigGlobalProperties")
public Map<String,Object> streamConfigGlobalProperties(KStreamBinderConfigurationProperties binderConfigurationProperties){
Map<String,Object> props = new HashMap<>();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, binderConfigurationProperties.getKafkaConnectionString());
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArraySerde.class.getName());
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "default");
props.put(StreamsConfig.APPLICATION_ID_CONFIG, binderConfigurationProperties.getApplicationId());
if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConfiguration())) {
props.putAll(binderConfigurationProperties.getConfiguration());
}
return new StreamsConfig(props);
return props;
}
@Bean
@@ -89,8 +76,8 @@ public class KStreamBinderSupportAutoConfiguration {
@Bean
public KStreamListenerParameterAdapter kafkaStreamListenerParameterAdapter(
MessageConversionDelegate messageConversionDelegate) {
return new KStreamListenerParameterAdapter(messageConversionDelegate);
KStreamBoundMessageConversionDelegate kstreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) {
return new KStreamListenerParameterAdapter(kstreamBoundMessageConversionDelegate, KStreamBindingInformationCatalogue);
}
@Bean
@@ -101,15 +88,43 @@ public class KStreamBinderSupportAutoConfiguration {
}
@Bean
public MessageConversionDelegate messageConversionDelegate(BindingServiceProperties bindingServiceProperties,
CompositeMessageConverterFactory compositeMessageConverterFactory) {
return new MessageConversionDelegate(bindingServiceProperties, compositeMessageConverterFactory);
public KStreamBoundMessageConversionDelegate messageConversionDelegate(CompositeMessageConverterFactory compositeMessageConverterFactory,
SendToDlqAndContinue sendToDlqAndContinue,
KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue) {
return new KStreamBoundMessageConversionDelegate(compositeMessageConverterFactory, sendToDlqAndContinue,
KStreamBindingInformationCatalogue);
}
@Bean
public KStreamBoundElementFactory kafkaStreamBindableTargetFactory(StreamsBuilder kStreamBuilder,
BindingServiceProperties bindingServiceProperties) {
return new KStreamBoundElementFactory(kStreamBuilder, bindingServiceProperties);
public KStreamBoundElementFactory kafkaStreamBindableTargetFactory(BindingServiceProperties bindingServiceProperties,
KStreamBindingInformationCatalogue KStreamBindingInformationCatalogue,
KeyValueSerdeResolver keyValueSerdeResolver,
KStreamExtendedBindingProperties kStreamExtendedBindingProperties) {
KStreamBoundElementFactory kStreamBoundElementFactory = new KStreamBoundElementFactory(bindingServiceProperties,
KStreamBindingInformationCatalogue, keyValueSerdeResolver);
kStreamBoundElementFactory.setkStreamExtendedBindingProperties(kStreamExtendedBindingProperties);
return kStreamBoundElementFactory;
}
@Bean
public SendToDlqAndContinue kStreamDlqSender() {
return new SendToDlqAndContinue();
}
@Bean
public KStreamBindingInformationCatalogue boundedKStreamRegistryService() {
return new KStreamBindingInformationCatalogue();
}
@Bean
public KeyValueSerdeResolver keyValueSerdeResolver(@Qualifier("streamConfigGlobalProperties") Map<String,Object> streamConfigGlobalProperties,
KStreamBinderConfigurationProperties kStreamBinderConfigurationProperties) {
return new KeyValueSerdeResolver(streamConfigGlobalProperties, kStreamBinderConfigurationProperties);
}
@Bean
public QueryableStoreRegistry queryableStoreTypeRegistry() {
return new QueryableStoreRegistry();
}
}

51
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamConsumerProperties.java
View File

@@ -16,9 +16,58 @@
package org.springframework.cloud.stream.binder.kstream.config;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
/**
* @author Marius Bogoevici
* @author Soby Chacko
*/
public class KStreamConsumerProperties extends KStreamCommonProperties {
public class KStreamConsumerProperties extends KafkaConsumerProperties {
public enum SerdeError {
logAndContinue,
logAndFail,
sendToDlq
}
/**
* Key serde specified per binding.
*/
private String keySerde;
/**
* Value serde specified per binding.
*/
private String valueSerde;
/**
* {@link org.apache.kafka.streams.errors.DeserializationExceptionHandler} to use
* when there is a Serde error. {@link SerdeError} values are used to provide the
* exception handler on consumer binding.
*/
private SerdeError serdeError;
public String getKeySerde() {
return keySerde;
}
public void setKeySerde(String keySerde) {
this.keySerde = keySerde;
}
public String getValueSerde() {
return valueSerde;
}
public void setValueSerde(String valueSerde) {
this.valueSerde = valueSerde;
}
public SerdeError getSerdeError() {
return serdeError;
}
public void setSerdeError(SerdeError serdeError) {
this.serdeError = serdeError;
}
}

29
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java
View File

@@ -16,10 +16,37 @@
package org.springframework.cloud.stream.binder.kstream.config;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
/**
* @author Marius Bogoevici
* @author Soby Chacko
*/
public class KStreamProducerProperties extends KStreamCommonProperties {
public class KStreamProducerProperties extends KafkaProducerProperties {
/**
* Key serde specified per binding.
*/
private String keySerde;
/**
* Value serde specified per binding.
*/
private String valueSerde;
public String getKeySerde() {
return keySerde;
}
public void setKeySerde(String keySerde) {
this.keySerde = keySerde;
}
public String getValueSerde() {
return valueSerde;
}
public void setValueSerde(String valueSerde) {
this.valueSerde = valueSerde;
}
}

155
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/DeserializationErrorHandlerByKafkaTests.java Normal file
View File

@@ -0,0 +1,155 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.util.Arrays;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor;
import org.springframework.cloud.stream.binder.kstream.config.KStreamApplicationSupportProperties;
import org.springframework.context.annotation.PropertySource;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
/**
* @author Soby Chacko
*/
@RunWith(SpringRunner.class)
@ContextConfiguration
@DirtiesContext
public abstract class DeserializationErrorHandlerByKafkaTests {
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts", "error.words.group");
@SpyBean
KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate;
private static Consumer<String, String> consumer;
@BeforeClass
public static void setUp() throws Exception {
System.setProperty("spring.cloud.stream.kstream.binder.brokers", embeddedKafka.getBrokersAsString());
System.setProperty("spring.cloud.stream.kstream.binder.zkNodes", embeddedKafka.getZookeeperConnectionString());
System.setProperty("server.port","0");
System.setProperty("spring.jmx.enabled","false");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("fooc", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts");
}
@AfterClass
public static void tearDown() {
consumer.close();
}
@SpringBootTest(properties = {
"spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true",
"spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
"spring.cloud.stream.bindings.input.group=group",
"spring.cloud.stream.kstream.bindings.input.consumer.serdeError=sendToDlq",
"spring.cloud.stream.kstream.binder.configuration.value.serde=" +
"org.apache.kafka.common.serialization.Serdes$IntegerSerde"},
webEnvironment= SpringBootTest.WebEnvironment.NONE
)
public static class DeserializationByKafkaAndDlqTests extends DeserializationErrorHandlerByKafkaTests {
@Test
@SuppressWarnings("unchecked")
public void test() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("words");
template.sendDefault("foobar");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("foobar", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<String, String> consumer1 = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer1, "error.words.group");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer1, "error.words.group");
assertThat(cr.value().equals("foobar")).isTrue();
//Ensuring that the deserialization was indeed done by Kafka natively
verify(KStreamBoundMessageConversionDelegate, never()).deserializeOnInbound(any(Class.class), any(KStream.class));
verify(KStreamBoundMessageConversionDelegate, never()).serializeOnOutbound(any(KStream.class));
}
}
@EnableBinding(KStreamProcessor.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/binder/kstream/integTest-1.properties")
@EnableConfigurationProperties(KStreamApplicationSupportProperties.class)
static class WordCountProcessorApplication {
@Autowired
private TimeWindows timeWindows;
@StreamListener("input")
@SendTo("output")
public KStream<?, String> process(KStream<Object, String> input) {
return input
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Serdes.String(), Serdes.String())
.count(timeWindows, "foo-WordCounts-x")
.toStream()
.map((key, value) -> new KeyValue<>(null, "Count for " + key.key() + " : " + value));
}
}
}

157
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/DeserializtionErrorHandlerByBinderTests.java Normal file
View File

@@ -0,0 +1,157 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.verify;
/**
* @author Soby Chacko
*/
@RunWith(SpringRunner.class)
@ContextConfiguration
@DirtiesContext
public abstract class DeserializtionErrorHandlerByBinderTests {
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts-id", "error.foos.foobar-group");
@SpyBean
KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate;
private static Consumer<Integer, String> consumer;
@BeforeClass
public static void setUp() throws Exception {
System.setProperty("spring.cloud.stream.kstream.binder.brokers", embeddedKafka.getBrokersAsString());
System.setProperty("spring.cloud.stream.kstream.binder.zkNodes", embeddedKafka.getZookeeperConnectionString());
System.setProperty("server.port","0");
System.setProperty("spring.jmx.enabled","false");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("foob", "false", embeddedKafka);
//consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Deserializer.class.getName());
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id");
}
@AfterClass
public static void tearDown() {
consumer.close();
}
@SpringBootTest(properties = {
"spring.cloud.stream.bindings.input.destination=foos",
"spring.cloud.stream.bindings.output.destination=counts-id",
"spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000",
"spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
"spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
"spring.cloud.stream.bindings.output.producer.headerMode=raw",
"spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$IntegerSerde",
"spring.cloud.stream.bindings.input.consumer.headerMode=raw",
"spring.cloud.stream.kstream.bindings.input.consumer.serdeError=sendToDlq",
"spring.cloud.stream.bindings.input.group=foobar-group"},
webEnvironment= SpringBootTest.WebEnvironment.NONE
)
public static class DeserializationByBinderAndDlqTests extends DeserializtionErrorHandlerByBinderTests {
@Test
@SuppressWarnings("unchecked")
public void test() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("foos");
template.sendDefault("hello");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("foobar", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<String, String> consumer1 = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer1, "error.foos.foobar-group");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer1, "error.foos.foobar-group");
assertThat(cr.value().equals("hello")).isTrue();
//Ensuring that the deserialization was indeed done by the binder
verify(KStreamBoundMessageConversionDelegate).deserializeOnInbound(any(Class.class), any(KStream.class));
}
}
@EnableBinding(KStreamProcessor.class)
@EnableAutoConfiguration
public static class ProductCountApplication {
@StreamListener("input")
@SendTo("output")
public KStream<Integer, Long> process(KStream<Object, Product> input) {
return input
.filter((key, product) -> product.getId() == 123)
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class))
.count(TimeWindows.of(5000), "id-count-store-x")
.toStream()
.map((key, value) -> new KeyValue<>(key.key().id, value));
}
}
static class Product {
Integer id;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
}
}

3
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderPojoInputAndPrimitiveTypeOutputTests.java
View File

@@ -77,6 +77,7 @@ public class KStreamBinderPojoInputAndPrimitiveTypeOutputTests {
SpringApplication app = new SpringApplication(ProductCountApplication.class);
app.setWebEnvironment(false);
ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.destination=foos",
"--spring.cloud.stream.bindings.output.destination=counts-id",
"--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000",
@@ -119,7 +120,7 @@ public class KStreamBinderPojoInputAndPrimitiveTypeOutputTests {
.filter((key, product) -> product.getId() == 123)
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class))
.count(TimeWindows.of(5000), "id-count-store")
.count(TimeWindows.of(5000), "id-count-store-x")
.toStream()
.map((key, value) -> new KeyValue<>(key.key().id, value));
}

1
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamBinderWordCountIntegrationTests.java
View File

@@ -82,6 +82,7 @@ public class KStreamBinderWordCountIntegrationTests {
app.setWebEnvironment(false);
ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.destination=words",
"--spring.cloud.stream.bindings.output.destination=counts",
"--spring.cloud.stream.bindings.output.contentType=application/json",

21
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamInteractiveQueryIntegrationTests.java
View File

@@ -22,7 +22,6 @@ import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.state.QueryableStoreTypes;
@@ -43,7 +42,6 @@ import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.StreamsBuilderFactoryBean;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -81,6 +79,7 @@ public class KStreamInteractiveQueryIntegrationTests {
SpringApplication app = new SpringApplication(ProductCountApplication.class);
app.setWebEnvironment(false);
ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.destination=foos",
"--spring.cloud.stream.bindings.output.destination=counts-id",
"--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000",
@@ -115,7 +114,7 @@ public class KStreamInteractiveQueryIntegrationTests {
public static class ProductCountApplication {
@Autowired
private StreamsBuilderFactoryBean kStreamBuilderFactoryBean;
private QueryableStoreRegistry queryableStoreRegistry;
@StreamListener("input")
@SendTo("output")
@@ -131,21 +130,23 @@ public class KStreamInteractiveQueryIntegrationTests {
}
@Bean
public Foo foo(StreamsBuilderFactoryBean kStreamBuilderFactoryBean) {
return new Foo(kStreamBuilderFactoryBean);
public Foo foo(QueryableStoreRegistry queryableStoreRegistry) {
return new Foo(queryableStoreRegistry);
}
static class Foo {
StreamsBuilderFactoryBean kStreamBuilderFactoryBean;
QueryableStoreRegistry queryableStoreRegistry;
Foo(StreamsBuilderFactoryBean kStreamBuilderFactoryBean) {
this.kStreamBuilderFactoryBean = kStreamBuilderFactoryBean;
Foo(QueryableStoreRegistry queryableStoreRegistry) {
this.queryableStoreRegistry = queryableStoreRegistry;
}
public Long getProductStock(Integer id) {
KafkaStreams streams = kStreamBuilderFactoryBean.getKafkaStreams();
ReadOnlyKeyValueStore<Object, Object> keyValueStore =
streams.store("prod-id-count-store", QueryableStoreTypes.keyValueStore());
queryableStoreRegistry.getQueryableStoreType("prod-id-count-store", QueryableStoreTypes.keyValueStore());
return (Long) keyValueStore.get(id);
}
}

159
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KStreamsNativeEncodingDecodingTests.java Normal file
View File

@@ -0,0 +1,159 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kstream;
import java.util.Arrays;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor;
import org.springframework.cloud.stream.binder.kstream.config.KStreamApplicationSupportProperties;
import org.springframework.context.annotation.PropertySource;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
/**
* @author Soby Chacko
*/
@RunWith(SpringRunner.class)
@ContextConfiguration
@DirtiesContext
public abstract class KStreamsNativeEncodingDecodingTests {
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts");
@SpyBean
KStreamBoundMessageConversionDelegate KStreamBoundMessageConversionDelegate;
private static Consumer<String, String> consumer;
@BeforeClass
public static void setUp() throws Exception {
System.setProperty("spring.cloud.stream.kstream.binder.brokers", embeddedKafka.getBrokersAsString());
System.setProperty("spring.cloud.stream.kstream.binder.zkNodes", embeddedKafka.getZookeeperConnectionString());
System.setProperty("server.port","0");
System.setProperty("spring.jmx.enabled","false");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts");
}
@AfterClass
public static void tearDown() {
consumer.close();
}
@SpringBootTest(properties = {
"spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true",
"spring.cloud.stream.bindings.output.producer.useNativeEncoding=true"},
webEnvironment= SpringBootTest.WebEnvironment.NONE
)
public static class NativeEncodingDecodingEnabledTests extends KStreamsNativeEncodingDecodingTests {
@Test
public void test() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("words");
template.sendDefault("foobar");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer, "counts");
assertThat(cr.value().equals("Count for foobar : 1")).isTrue();
verify(KStreamBoundMessageConversionDelegate, never()).serializeOnOutbound(any(KStream.class));
verify(KStreamBoundMessageConversionDelegate, never()).deserializeOnInbound(any(Class.class), any(KStream.class));
}
}
@SpringBootTest(webEnvironment= SpringBootTest.WebEnvironment.NONE)
public static class NativeEncodingDecodingDisabledTests extends KStreamsNativeEncodingDecodingTests {
@Test
public void test() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("words");
template.sendDefault("foobar");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer, "counts");
assertThat(cr.value().equals("Count for foobar : 1")).isTrue();
verify(KStreamBoundMessageConversionDelegate).serializeOnOutbound(any(KStream.class));
verify(KStreamBoundMessageConversionDelegate).deserializeOnInbound(any(Class.class), any(KStream.class));
}
}
@EnableBinding(KStreamProcessor.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/binder/kstream/integTest-1.properties")
@EnableConfigurationProperties(KStreamApplicationSupportProperties.class)
static class WordCountProcessorApplication {
@Autowired
private TimeWindows timeWindows;
@StreamListener("input")
@SendTo("output")
public KStream<?, String> process(KStream<Object, String> input) {
return input
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Serdes.String(), Serdes.String())
.count(timeWindows, "foo-WordCounts-x")
.toStream()
.map((key, value) -> new KeyValue<>(null, "Count for " + key.key() + " : " + value));
}
}
}

1
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/KstreamBinderPojoInputStringOutputIntegrationTests.java
View File

@@ -76,6 +76,7 @@ public class KstreamBinderPojoInputStringOutputIntegrationTests {
SpringApplication app = new SpringApplication(ProductCountApplication.class);
app.setWebEnvironment(false);
ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.destination=foos",
"--spring.cloud.stream.bindings.output.destination=counts-id",
"--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000",

3
spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/WordCountMultipleBranchesIntegrationTests.java
View File

@@ -66,7 +66,7 @@ public class WordCountMultipleBranchesIntegrationTests {
@BeforeClass
public static void setUp() throws Exception {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("groupx", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
@@ -84,6 +84,7 @@ public class WordCountMultipleBranchesIntegrationTests {
app.setWebEnvironment(false);
ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.destination=words",
"--spring.cloud.stream.bindings.output1.destination=counts",
"--spring.cloud.stream.bindings.output1.contentType=application/json",

10
spring-cloud-stream-binder-kstream/src/test/resources/org/springframework/cloud/stream/binder/kstream/integTest-1.properties Normal file
View File

@@ -0,0 +1,10 @@
spring.cloud.stream.bindings.input.destination=words
spring.cloud.stream.bindings.output.destination=counts
spring.cloud.stream.bindings.output.contentType=application/json
spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000
spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.bindings.output.producer.headerMode=raw
spring.cloud.stream.bindings.input.consumer.headerMode=raw
spring.cloud.stream.kstream.timeWindow.length=5000
spring.cloud.stream.kstream.timeWindow.advanceBy=0