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

Checkstyle fixes

Browse Source 
Reuse checkstyle components from core spring-cloud-stream-tools module.
Addressing checkstyle warnings in kafka streams binder module.

Resolves #489
This commit is contained in:
Soby Chacko
2018-11-09 18:11:51 -05:00
parent 20af89bc00
commit b589f32933
38 changed files with 432 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
pom.xml
View File

@@ -145,6 +145,31 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-tools</artifactId>
<version>${spring-cloud-stream.version}</version>
</dependency>
</dependencies>
<executions>
<execution>
<id>checkstyle-validation</id>
<phase>validate</phase>
<configuration>
<configLocation>checkstyle.xml</configLocation>
<headerLocation>checkstyle-header.txt</headerLocation>
<suppressionsLocation>checkstyle-suppressions.xml</suppressionsLocation>
<encoding>UTF-8</encoding>
<consoleOutput>true</consoleOutput>
<failsOnError>true</failsOnError>
<includeTestSourceDirectory>true</includeTestSourceDirectory>
</configuration>
<goals>
<goal>check</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>

8
spring-cloud-stream-binder-kafka-streams/pom.xml
View File

@@ -102,14 +102,14 @@
<version>${avro.version}</version>
<executions>
<execution>
<phase>generate-sources</phase>
<phase>generate-test-sources</phase>
<goals>
<goal>schema</goal>
<goal>protocol</goal>
<goal>idl-protocol</goal>
</goals>
<configuration>
<sourceDirectory>src/test/resources/avro</sourceDirectory>
<outputDirectory>${project.basedir}/target/generated-test-sources</outputDirectory>
<testOutputDirectory>${project.basedir}/target/generated-test-sources</testOutputDirectory>
<testSourceDirectory>${project.basedir}/src/test/resources/avro</testSourceDirectory>
</configuration>
</execution>
</executions>

8
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java
View File

@@ -56,7 +56,7 @@ public class GlobalKTableBinder extends
private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties();
public GlobalKTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner,
Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
this.binderConfigurationProperties = binderConfigurationProperties;
this.kafkaTopicProvisioner = kafkaTopicProvisioner;
this.kafkaStreamsDlqDispatchers = kafkaStreamsDlqDispatchers;
@@ -67,11 +67,11 @@ public class GlobalKTableBinder extends
protected Binding<GlobalKTable<Object, Object>> doBindConsumer(String name, String group, GlobalKTable<Object, Object> inputTarget,
ExtendedConsumerProperties<KafkaStreamsConsumerProperties> properties) {
if (!StringUtils.hasText(group)) {
group = binderConfigurationProperties.getApplicationId();
group = this.binderConfigurationProperties.getApplicationId();
}
KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(),
kafkaTopicProvisioner,
binderConfigurationProperties, properties, kafkaStreamsDlqDispatchers);
this.kafkaTopicProvisioner,
this.binderConfigurationProperties, properties, this.kafkaStreamsDlqDispatchers);
return new DefaultBinding<>(name, group, inputTarget, null);
}

2
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java
View File

@@ -28,6 +28,8 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
/**
* Configuration for GlobalKTable binder.
*
* @author Soby Chacko
* @since 2.1.0
*/

9
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java
View File

@@ -49,7 +49,7 @@ public class GlobalKTableBoundElementFactory extends AbstractBindingTargetFactor
//Always set multiplex to true in the kafka streams binder
consumerProperties.setMultiplex(true);
GlobalKTableBoundElementFactory.GlobalKTableWrapperHandler wrapper= new GlobalKTableBoundElementFactory.GlobalKTableWrapperHandler();
GlobalKTableBoundElementFactory.GlobalKTableWrapperHandler wrapper = new GlobalKTableBoundElementFactory.GlobalKTableWrapperHandler();
ProxyFactory proxyFactory = new ProxyFactory(GlobalKTableBoundElementFactory.GlobalKTableWrapper.class, GlobalKTable.class);
proxyFactory.addAdvice(wrapper);
@@ -61,6 +61,9 @@ public class GlobalKTableBoundElementFactory extends AbstractBindingTargetFactor
throw new UnsupportedOperationException("Outbound operations are not allowed on target type GlobalKTable");
}
/**
* Wrapper for GlobalKTable proxy.
*/
public interface GlobalKTableWrapper {
void wrap(GlobalKTable<Object, Object> delegate);
}
@@ -78,9 +81,9 @@ public class GlobalKTableBoundElementFactory extends AbstractBindingTargetFactor
@Override
public Object invoke(MethodInvocation methodInvocation) throws Throwable {
if (methodInvocation.getMethod().getDeclaringClass().equals(GlobalKTable.class)) {
Assert.notNull(delegate, "Trying to prepareConsumerBinding " + methodInvocation
Assert.notNull(this.delegate, "Trying to prepareConsumerBinding " + methodInvocation
.getMethod() + " but no delegate has been set.");
return methodInvocation.getMethod().invoke(delegate, methodInvocation.getArguments());
return methodInvocation.getMethod().invoke(this.delegate, methodInvocation.getArguments());
}
else if (methodInvocation.getMethod().getDeclaringClass().equals(GlobalKTableBoundElementFactory.GlobalKTableWrapper.class)) {
return methodInvocation.getMethod().invoke(this, methodInvocation.getArguments());

10
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java
View File

@@ -45,9 +45,10 @@ public class InteractiveQueryService {
private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties;
/**
* Constructor for InteractiveQueryService.
*
* @param kafkaStreamsRegistry holding {@link KafkaStreamsRegistry}
* @param binderConfigurationProperties Kafka Streams binder configuration properties
* @param binderConfigurationProperties kafka Streams binder configuration properties
*/
public InteractiveQueryService(KafkaStreamsRegistry kafkaStreamsRegistry,
KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) {
@@ -60,12 +61,12 @@ public class InteractiveQueryService {
*
* @param storeName name of the queryable store
* @param storeType type of the queryable store
* @param <T> generic queryable store
* @param <T> generic queryable store
* @return queryable store.
*/
public <T> T getQueryableStore(String storeName, QueryableStoreType<T> storeType) {
for (KafkaStreams kafkaStream : this.kafkaStreamsRegistry.getKafkaStreams()) {
try{
try {
T store = kafkaStream.store(storeName, storeType);
if (store != null) {
return store;
@@ -106,6 +107,7 @@ public class InteractiveQueryService {
* Note that the end user applications must provide `applicaiton.server` as a configuration property
* for all the application instances when calling this method. If this is not available, then null maybe returned.
*
* @param <K> generic type for key
* @param store store name
* @param key key to look for
* @param serializer {@link Serializer} for the key
@@ -114,7 +116,7 @@ public class InteractiveQueryService {
public <K> HostInfo getHostInfo(String store, K key, Serializer<K> serializer) {
StreamsMetadata streamsMetadata = this.kafkaStreamsRegistry.getKafkaStreams()
.stream()
.map(k -> Optional.ofNullable(k.metadataForKey(store, key, serializer)))
.map((k) -> Optional.ofNullable(k.metadataForKey(store, key, serializer)))
.filter(Optional::isPresent)
.map(Optional::get)
.findFirst()

15
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java
View File

@@ -52,7 +52,7 @@ class KStreamBinder extends
AbstractBinder<KStream<Object, Object>, ExtendedConsumerProperties<KafkaStreamsConsumerProperties>, ExtendedProducerProperties<KafkaStreamsProducerProperties>>
implements ExtendedPropertiesBinder<KStream<Object, Object>, KafkaStreamsConsumerProperties, KafkaStreamsProducerProperties> {
private final static Log LOG = LogFactory.getLog(KStreamBinder.class);
private static final Log LOG = LogFactory.getLog(KStreamBinder.class);
private final KafkaTopicProvisioner kafkaTopicProvisioner;
@@ -73,7 +73,7 @@ class KStreamBinder extends
KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate,
KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue,
KeyValueSerdeResolver keyValueSerdeResolver,
Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
this.binderConfigurationProperties = binderConfigurationProperties;
this.kafkaTopicProvisioner = kafkaTopicProvisioner;
this.kafkaStreamsMessageConversionDelegate = kafkaStreamsMessageConversionDelegate;
@@ -88,11 +88,11 @@ class KStreamBinder extends
ExtendedConsumerProperties<KafkaStreamsConsumerProperties> properties) {
this.kafkaStreamsBindingInformationCatalogue.registerConsumerProperties(inputTarget, properties.getExtension());
if (!StringUtils.hasText(group)) {
group = binderConfigurationProperties.getApplicationId();
group = this.binderConfigurationProperties.getApplicationId();
}
KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(),
kafkaTopicProvisioner,
binderConfigurationProperties, properties, kafkaStreamsDlqDispatchers);
this.kafkaTopicProvisioner,
this.binderConfigurationProperties, properties, this.kafkaStreamsDlqDispatchers);
return new DefaultBinding<>(name, group, inputTarget, null);
}
@@ -115,9 +115,10 @@ class KStreamBinder extends
Serde<Object> keySerde, Serde<Object> valueSerde) {
if (!isNativeEncoding) {
LOG.info("Native encoding is disabled for " + name + ". Outbound message conversion done by Spring Cloud Stream.");
kafkaStreamsMessageConversionDelegate.serializeOnOutbound(outboundBindTarget)
this.kafkaStreamsMessageConversionDelegate.serializeOnOutbound(outboundBindTarget)
.to(name, Produced.with(keySerde, valueSerde));
} else {
}
else {
LOG.info("Native encoding is enabled for " + name + ". Outbound serialization done at the broker.");
outboundBindTarget.to(name, Produced.with(keySerde, valueSerde));
}

47
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java
View File

@@ -35,6 +35,8 @@ import org.springframework.context.annotation.Import;
import org.springframework.core.type.AnnotationMetadata;
/**
* Configuration for KStream binder.
*
* @author Marius Bogoevici
* @author Gary Russell
* @author Soby Chacko
@@ -43,6 +45,30 @@ import org.springframework.core.type.AnnotationMetadata;
@Import({KafkaAutoConfiguration.class, KStreamBinderConfiguration.KStreamMissingBeansRegistrar.class})
public class KStreamBinderConfiguration {
@Bean
public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties,
KafkaProperties kafkaProperties) {
return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties);
}
@Bean
public KStreamBinder kStreamBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties,
KafkaTopicProvisioner kafkaTopicProvisioner,
KafkaStreamsMessageConversionDelegate KafkaStreamsMessageConversionDelegate,
KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue,
KeyValueSerdeResolver keyValueSerdeResolver,
KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties,
@Qualifier("kafkaStreamsDlqDispatchers") Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner,
KafkaStreamsMessageConversionDelegate, KafkaStreamsBindingInformationCatalogue,
keyValueSerdeResolver, kafkaStreamsDlqDispatchers);
kStreamBinder.setKafkaStreamsExtendedBindingProperties(kafkaStreamsExtendedBindingProperties);
return kStreamBinder;
}
/**
* Registrar for missing beans when there are multiple binders in the application.
*/
static class KStreamMissingBeansRegistrar extends KafkaStreamsBinderUtils.KafkaStreamsMissingBeansRegistrar {
private static final String BEAN_NAME = "outerContext";
@@ -81,25 +107,4 @@ public class KStreamBinderConfiguration {
}
}
@Bean
public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties,
KafkaProperties kafkaProperties) {
return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties);
}
@Bean
public KStreamBinder kStreamBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties,
KafkaTopicProvisioner kafkaTopicProvisioner,
KafkaStreamsMessageConversionDelegate KafkaStreamsMessageConversionDelegate,
KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue,
KeyValueSerdeResolver keyValueSerdeResolver,
KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties,
@Qualifier("kafkaStreamsDlqDispatchers") Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner,
KafkaStreamsMessageConversionDelegate, KafkaStreamsBindingInformationCatalogue,
keyValueSerdeResolver, kafkaStreamsDlqDispatchers);
kStreamBinder.setKafkaStreamsExtendedBindingProperties(kafkaStreamsExtendedBindingProperties);
return kStreamBinder;
}
}

11
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java
View File

@@ -64,18 +64,21 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KStream> {
}
private KStream createProxyForKStream(String name) {
KStreamWrapperHandler wrapper= new KStreamWrapperHandler();
KStreamWrapperHandler wrapper = new KStreamWrapperHandler();
ProxyFactory proxyFactory = new ProxyFactory(KStreamWrapper.class, KStream.class);
proxyFactory.addAdvice(wrapper);
KStream proxy = (KStream) proxyFactory.getProxy();
//Add the binding properties to the catalogue for later retrieval during further binding steps downstream.
BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(name);
this.kafkaStreamsBindingInformationCatalogue.registerBindingProperties(proxy, bindingProperties);
return proxy;
}
/**
* Wrapper object for KStream proxy.
*/
public interface KStreamWrapper {
void wrap(KStream<Object, Object> delegate);
@@ -95,9 +98,9 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KStream> {
@Override
public Object invoke(MethodInvocation methodInvocation) throws Throwable {
if (methodInvocation.getMethod().getDeclaringClass().equals(KStream.class)) {
Assert.notNull(delegate, "Trying to prepareConsumerBinding " + methodInvocation
Assert.notNull(this.delegate, "Trying to prepareConsumerBinding " + methodInvocation
.getMethod() + " but no delegate has been set.");
return methodInvocation.getMethod().invoke(delegate, methodInvocation.getArguments());
return methodInvocation.getMethod().invoke(this.delegate, methodInvocation.getArguments());
}
else if (methodInvocation.getMethod().getDeclaringClass().equals(KStreamWrapper.class)) {
return methodInvocation.getMethod().invoke(this, methodInvocation.getArguments());

6
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.java
View File

@@ -23,10 +23,12 @@ import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
/**
* {@link StreamListenerParameterAdapter} for KStream.
*
* @author Marius Bogoevici
* @author Soby Chacko
*/
class KStreamStreamListenerParameterAdapter implements StreamListenerParameterAdapter<KStream<?,?>, KStream<?, ?>> {
class KStreamStreamListenerParameterAdapter implements StreamListenerParameterAdapter<KStream<?, ?>, KStream<?, ?>> {
private final KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate;
private final KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue;
@@ -53,7 +55,7 @@ class KStreamStreamListenerParameterAdapter implements StreamListenerParameterAd
return bindingTarget;
}
else {
return kafkaStreamsMessageConversionDelegate.deserializeOnInbound(valueClass, bindingTarget);
return this.kafkaStreamsMessageConversionDelegate.deserializeOnInbound(valueClass, bindingTarget);
}
}
}

2
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java
View File

@@ -24,6 +24,8 @@ import org.apache.kafka.streams.kstream.KStream;
import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
/**
* {@link StreamListenerResultAdapter} for KStream.
*
* @author Marius Bogoevici
* @author Soby Chacko
*/

8
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java
View File

@@ -55,7 +55,7 @@ class KTableBinder extends
private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties();
KTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner,
Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
this.binderConfigurationProperties = binderConfigurationProperties;
this.kafkaTopicProvisioner = kafkaTopicProvisioner;
this.kafkaStreamsDlqDispatchers = kafkaStreamsDlqDispatchers;
@@ -66,11 +66,11 @@ class KTableBinder extends
protected Binding<KTable<Object, Object>> doBindConsumer(String name, String group, KTable<Object, Object> inputTarget,
ExtendedConsumerProperties<KafkaStreamsConsumerProperties> properties) {
if (!StringUtils.hasText(group)) {
group = binderConfigurationProperties.getApplicationId();
group = this.binderConfigurationProperties.getApplicationId();
}
KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(),
kafkaTopicProvisioner,
binderConfigurationProperties, properties, kafkaStreamsDlqDispatchers);
this.kafkaTopicProvisioner,
this.binderConfigurationProperties, properties, this.kafkaStreamsDlqDispatchers);
return new DefaultBinding<>(name, group, inputTarget, null);
}

6
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java
View File

@@ -31,6 +31,8 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
/**
* Configuration for KTable binder.
*
* @author Soby Chacko
*/
@SuppressWarnings("ALL")
@@ -41,7 +43,7 @@ public class KTableBinderConfiguration {
@Bean
@ConditionalOnBean(name = "outerContext")
public BeanFactoryPostProcessor outerContextBeanFactoryPostProcessor() {
return beanFactory -> {
return (beanFactory) -> {
ApplicationContext outerContext = (ApplicationContext) beanFactory.getBean("outerContext");
beanFactory.registerSingleton(KafkaStreamsBinderConfigurationProperties.class.getSimpleName(), outerContext
.getBean(KafkaStreamsBinderConfigurationProperties.class));
@@ -59,7 +61,7 @@ public class KTableBinderConfiguration {
@Bean
public KTableBinder kTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties,
KafkaTopicProvisioner kafkaTopicProvisioner,
@Qualifier("kafkaStreamsDlqDispatchers") Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
@Qualifier("kafkaStreamsDlqDispatchers") Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
KTableBinder kStreamBinder = new KTableBinder(binderConfigurationProperties, kafkaTopicProvisioner, kafkaStreamsDlqDispatchers);
return kStreamBinder;
}

9
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java
View File

@@ -48,7 +48,7 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory<KTable> {
//Always set multiplex to true in the kafka streams binder
consumerProperties.setMultiplex(true);
KTableBoundElementFactory.KTableWrapperHandler wrapper= new KTableBoundElementFactory.KTableWrapperHandler();
KTableBoundElementFactory.KTableWrapperHandler wrapper = new KTableBoundElementFactory.KTableWrapperHandler();
ProxyFactory proxyFactory = new ProxyFactory(KTableBoundElementFactory.KTableWrapper.class, KTable.class);
proxyFactory.addAdvice(wrapper);
@@ -61,6 +61,9 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory<KTable> {
throw new UnsupportedOperationException("Outbound operations are not allowed on target type KTable");
}
/**
* Wrapper for KTable proxy.
*/
public interface KTableWrapper {
void wrap(KTable<Object, Object> delegate);
}
@@ -78,9 +81,9 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory<KTable> {
@Override
public Object invoke(MethodInvocation methodInvocation) throws Throwable {
if (methodInvocation.getMethod().getDeclaringClass().equals(KTable.class)) {
Assert.notNull(delegate, "Trying to prepareConsumerBinding " + methodInvocation
Assert.notNull(this.delegate, "Trying to prepareConsumerBinding " + methodInvocation
.getMethod() + " but no delegate has been set.");
return methodInvocation.getMethod().invoke(delegate, methodInvocation.getArguments());
return methodInvocation.getMethod().invoke(this.delegate, methodInvocation.getArguments());
}
else if (methodInvocation.getMethod().getDeclaringClass().equals(KTableBoundElementFactory.KTableWrapper.class)) {
return methodInvocation.getMethod().invoke(this, methodInvocation.getArguments());

2
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java
View File

@@ -25,6 +25,8 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* Application support configuration for Kafka Streams binder.
*
* @author Soby Chacko
*/
@Configuration

10
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java
View File

@@ -50,6 +50,8 @@ import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
* Kafka Streams binder configuration.
*
* @author Marius Bogoevici
* @author Soby Chacko
* @author Gary Russell
@@ -112,10 +114,12 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndContinue) {
properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
LogAndContinueExceptionHandler.class.getName());
} else if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) {
}
else if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) {
properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
LogAndFailExceptionHandler.class.getName());
} else if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) {
}
else if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) {
properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
SendToDlqAndContinue.class.getName());
}
@@ -124,7 +128,7 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
properties.putAll(binderConfigurationProperties.getConfiguration());
}
return properties.entrySet().stream().collect(
Collectors.toMap(e -> String.valueOf(e.getKey()), Map.Entry::getValue));
Collectors.toMap((e) -> String.valueOf(e.getKey()), Map.Entry::getValue));
}
@Bean

13
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java
View File

@@ -33,15 +33,21 @@ import org.springframework.core.type.AnnotationMetadata;
import org.springframework.util.StringUtils;
/**
* Common methods used by various Kafka Streams types across the binders.
*
* @author Soby Chacko
*/
class KafkaStreamsBinderUtils {
final class KafkaStreamsBinderUtils {
private KafkaStreamsBinderUtils() {
}
static void prepareConsumerBinding(String name, String group, ApplicationContext context,
KafkaTopicProvisioner kafkaTopicProvisioner,
KafkaStreamsBinderConfigurationProperties binderConfigurationProperties,
ExtendedConsumerProperties<KafkaStreamsConsumerProperties> properties,
Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers) {
ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties = new ExtendedConsumerProperties<>(
properties.getExtension());
if (binderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) {
@@ -72,6 +78,9 @@ class KafkaStreamsBinderUtils {
}
}
/**
* Helper lass for missing bean registration.
*/
static class KafkaStreamsMissingBeansRegistrar implements ImportBeanDefinitionRegistrar {
private static final String BEAN_NAME = "outerContext";

40
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java
View File

@@ -49,10 +49,10 @@ class KafkaStreamsBindingInformationCatalogue {
* For a given bounded {@link KStream}, retrieve it's corresponding destination
* on the broker.
*
* @param bindingTarget KStream binding target
* @param bindingTarget binding target for KStream
* @return destination topic on Kafka
*/
String getDestination(KStream<?,?> bindingTarget) {
String getDestination(KStream<?, ?> bindingTarget) {
BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget);
return bindingProperties.getDestination();
}
@@ -60,10 +60,10 @@ class KafkaStreamsBindingInformationCatalogue {
/**
* Is native decoding is enabled on this {@link KStream}.
*
* @param bindingTarget KStream binding target
* @param bindingTarget binding target for KStream
* @return true if native decoding is enabled, fasle otherwise.
*/
boolean isUseNativeDecoding(KStream<?,?> bindingTarget) {
boolean isUseNativeDecoding(KStream<?, ?> bindingTarget) {
BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget);
if (bindingProperties.getConsumer() == null) {
bindingProperties.setConsumer(new ConsumerProperties());
@@ -72,48 +72,48 @@ class KafkaStreamsBindingInformationCatalogue {
}
/**
* Is DLQ enabled for this {@link KStream}
* Is DLQ enabled for this {@link KStream}.
*
* @param bindingTarget KStream binding target
* @param bindingTarget binding target for KStream
* @return true if DLQ is enabled, false otherwise.
*/
boolean isDlqEnabled(KStream<?,?> bindingTarget) {
return consumerProperties.get(bindingTarget).isEnableDlq();
boolean isDlqEnabled(KStream<?, ?> bindingTarget) {
return this.consumerProperties.get(bindingTarget).isEnableDlq();
}
/**
* Retrieve the content type associated with a given {@link KStream}
* Retrieve the content type associated with a given {@link KStream}.
*
* @param bindingTarget KStream binding target
* @param bindingTarget binding target for KStream
* @return content Type associated.
*/
String getContentType(KStream<?,?> bindingTarget) {
String getContentType(KStream<?, ?> bindingTarget) {
BindingProperties bindingProperties = this.bindingProperties.get(bindingTarget);
return bindingProperties.getContentType();
}
/**
* Register a cache for bounded KStream -> {@link BindingProperties}
* Register a cache for bounded KStream -> {@link BindingProperties}.
*
* @param bindingTarget KStream binding target
* @param bindingTarget binding target for KStream
* @param bindingProperties {@link BindingProperties} for this KStream
*/
void registerBindingProperties(KStream<?,?> bindingTarget, BindingProperties bindingProperties) {
void registerBindingProperties(KStream<?, ?> bindingTarget, BindingProperties bindingProperties) {
this.bindingProperties.put(bindingTarget, bindingProperties);
}
/**
* Register a cache for bounded KStream -> {@link KafkaStreamsConsumerProperties}
* Register a cache for bounded KStream -> {@link KafkaStreamsConsumerProperties}.
*
* @param bindingTarget KStream binding target
* @param kafkaStreamsConsumerProperties Consumer properties for this KStream
* @param bindingTarget binding target for KStream
* @param kafkaStreamsConsumerProperties consumer properties for this KStream
*/
void registerConsumerProperties(KStream<?,?> bindingTarget, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties) {
void registerConsumerProperties(KStream<?, ?> bindingTarget, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties) {
this.consumerProperties.put(bindingTarget, kafkaStreamsConsumerProperties);
}
/**
* Adds a mapping for KStream -> {@link StreamsBuilderFactoryBean}
* Adds a mapping for KStream -> {@link StreamsBuilderFactoryBean}.
*
* @param streamsBuilderFactoryBean provides the {@link StreamsBuilderFactoryBean} mapped to the KStream
*/
@@ -122,6 +122,6 @@ class KafkaStreamsBindingInformationCatalogue {
}
Set<StreamsBuilderFactoryBean> getStreamsBuilderFactoryBeans() {
return streamsBuilderFactoryBeans;
return this.streamsBuilderFactoryBeans;
}
}

26
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java
View File

@@ -16,11 +16,6 @@
package org.springframework.cloud.stream.binder.kafka.streams;
/**
* @author Soby Chacko
* @author Rafal Zukowski
* @author Gary Russell
*/
import java.util.HashMap;
import java.util.Map;
@@ -42,18 +37,25 @@ import org.springframework.util.ObjectUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
* Send records in error to a DLQ.
*
* @author Soby Chacko
* @author Rafal Zukowski
* @author Gary Russell
*/
class KafkaStreamsDlqDispatch {
private final Log logger = LogFactory.getLog(getClass());
private final KafkaTemplate<byte[],byte[]> kafkaTemplate;
private final KafkaTemplate<byte[], byte[]> kafkaTemplate;
private final String dlqName;
KafkaStreamsDlqDispatch(String dlqName,
KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties,
KafkaConsumerProperties kafkaConsumerProperties) {
ProducerFactory<byte[],byte[]> producerFactory = getProducerFactory(
ProducerFactory<byte[], byte[]> producerFactory = getProducerFactory(
new ExtendedProducerProperties<>(kafkaConsumerProperties.getDlqProducerProperties()),
kafkaBinderConfigurationProperties);
@@ -63,7 +65,7 @@ class KafkaStreamsDlqDispatch {
@SuppressWarnings("unchecked")
public void sendToDlq(byte[] key, byte[] value, int partittion) {
ProducerRecord<byte[],byte[]> producerRecord = new ProducerRecord<>(this.dlqName, partittion,
ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<>(this.dlqName, partittion,
key, value, null);
StringBuilder sb = new StringBuilder().append(" a message with key='")
@@ -72,10 +74,10 @@ class KafkaStreamsDlqDispatch {
.append(toDisplayString(ObjectUtils.nullSafeToString(value)))
.append("'").append(" received from ")
.append(partittion);
ListenableFuture<SendResult<byte[],byte[]>> sentDlq = null;
ListenableFuture<SendResult<byte[], byte[]>> sentDlq = null;
try {
sentDlq = this.kafkaTemplate.send(producerRecord);
sentDlq.addCallback(new ListenableFutureCallback<SendResult<byte[],byte[]>>() {
sentDlq.addCallback(new ListenableFutureCallback<SendResult<byte[], byte[]>>() {
@Override
public void onFailure(Throwable ex) {
@@ -84,7 +86,7 @@ class KafkaStreamsDlqDispatch {
}
@Override
public void onSuccess(SendResult<byte[],byte[]> result) {
public void onSuccess(SendResult<byte[], byte[]> result) {
if (KafkaStreamsDlqDispatch.this.logger.isDebugEnabled()) {
KafkaStreamsDlqDispatch.this.logger.debug(
"Sent to DLQ " + sb.toString());
@@ -100,7 +102,7 @@ class KafkaStreamsDlqDispatch {
}
}
private DefaultKafkaProducerFactory<byte[],byte[]> getProducerFactory(ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
private DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
KafkaBinderConfigurationProperties configurationProperties) {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.RETRIES_CONFIG, 0);

43
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
View File

@@ -48,7 +48,7 @@ import org.springframework.util.StringUtils;
*/
public class KafkaStreamsMessageConversionDelegate {
private final static Log LOG = LogFactory.getLog(KafkaStreamsMessageConversionDelegate.class);
private static final Log LOG = LogFactory.getLog(KafkaStreamsMessageConversionDelegate.class);
private static final ThreadLocal<KeyValue<Object, Object>> keyValueThreadLocal = new ThreadLocal<>();
@@ -77,9 +77,9 @@ public class KafkaStreamsMessageConversionDelegate {
* @return serialized KStream
*/
@SuppressWarnings("rawtypes")
public KStream serializeOnOutbound(KStream<?,?> outboundBindTarget) {
public KStream serializeOnOutbound(KStream<?, ?> outboundBindTarget) {
String contentType = this.kstreamBindingInformationCatalogue.getContentType(outboundBindTarget);
MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
MessageConverter messageConverter = this.compositeMessageConverterFactory.getMessageConverterForAllRegistered();
return outboundBindTarget.mapValues((v) -> {
Message<?> message = v instanceof Message<?> ? (Message<?>) v :
@@ -104,7 +104,7 @@ public class KafkaStreamsMessageConversionDelegate {
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
public KStream deserializeOnInbound(Class<?> valueClass, KStream<?, ?> bindingTarget) {
MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
MessageConverter messageConverter = this.compositeMessageConverterFactory.getMessageConverterForAllRegistered();
final PerRecordContentTypeHolder perRecordContentTypeHolder = new PerRecordContentTypeHolder();
resolvePerRecordContentType(bindingTarget, perRecordContentTypeHolder);
@@ -122,7 +122,8 @@ public class KafkaStreamsMessageConversionDelegate {
Message<?> m1 = null;
if (o2 instanceof Message) {
m1 = perRecordContentTypeHolder.contentType != null
? MessageBuilder.fromMessage((Message<?>) o2).setHeader(MessageHeaders.CONTENT_TYPE, perRecordContentTypeHolder.contentType).build() : (Message<?>)o2;
? MessageBuilder.fromMessage((Message<?>) o2).setHeader(MessageHeaders.CONTENT_TYPE,
perRecordContentTypeHolder.contentType).build() : (Message<?>) o2;
}
else {
m1 = perRecordContentTypeHolder.contentType != null ? MessageBuilder.withPayload(o2)
@@ -158,15 +159,6 @@ public class KafkaStreamsMessageConversionDelegate {
});
}
private static class PerRecordContentTypeHolder {
String contentType;
void setContentType(String contentType) {
this.contentType = contentType;
}
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private void resolvePerRecordContentType(KStream<?, ?> outboundBindTarget, PerRecordContentTypeHolder perRecordContentTypeHolder) {
outboundBindTarget.process(() -> new Processor() {
@@ -180,7 +172,7 @@ public class KafkaStreamsMessageConversionDelegate {
@Override
public void process(Object key, Object value) {
final Headers headers = context.headers();
final Headers headers = this.context.headers();
final Iterable<Header> contentTypes = headers.headers(MessageHeaders.CONTENT_TYPE);
if (contentTypes != null && contentTypes.iterator().hasNext()) {
final String contentType = new String(contentTypes.iterator().next().value());
@@ -220,20 +212,20 @@ public class KafkaStreamsMessageConversionDelegate {
public void process(Object o, Object o2) {
//Only continue if the record was not a tombstone.
if (o2 != null) {
if (kstreamBindingInformationCatalogue.isDlqEnabled(bindingTarget)) {
String destination = context.topic();
if (KafkaStreamsMessageConversionDelegate.this.kstreamBindingInformationCatalogue.isDlqEnabled(bindingTarget)) {
String destination = this.context.topic();
if (o2 instanceof Message) {
Message message = (Message) o2;
sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) message.getPayload(), context.partition());
KafkaStreamsMessageConversionDelegate.this.sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) message.getPayload(), this.context.partition());
}
else {
sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) o2, context.partition());
KafkaStreamsMessageConversionDelegate.this.sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) o2, this.context.partition());
}
}
else if (kstreamBinderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) {
else if (KafkaStreamsMessageConversionDelegate.this.kstreamBinderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) {
throw new IllegalStateException("Inbound deserialization failed. Stopping further processing of records.");
}
else if (kstreamBinderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndContinue) {
else if (KafkaStreamsMessageConversionDelegate.this.kstreamBinderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndContinue) {
//quietly passing through. No action needed, this is similar to log and continue.
LOG.error("Inbound deserialization failed. Skipping this record and continuing.");
}
@@ -246,4 +238,13 @@ public class KafkaStreamsMessageConversionDelegate {
}
});
}
private static class PerRecordContentTypeHolder {
String contentType;
void setContentType(String contentType) {
this.contentType = contentType;
}
}
}

2
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java
View File

@@ -32,7 +32,7 @@ class KafkaStreamsRegistry {
private final Set<KafkaStreams> kafkaStreams = new HashSet<>();
Set<KafkaStreams> getKafkaStreams() {
return kafkaStreams;
return this.kafkaStreams;
}
/**

91
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java
View File

@@ -91,7 +91,7 @@ import org.springframework.util.StringUtils;
*/
class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListenerSetupMethodOrchestrator, ApplicationContextAware {
private final static Log LOG = LogFactory.getLog(KafkaStreamsStreamListenerSetupMethodOrchestrator.class);
private static final Log LOG = LogFactory.getLog(KafkaStreamsStreamListenerSetupMethodOrchestrator.class);
private final StreamListenerParameterAdapter streamListenerParameterAdapter;
@@ -175,7 +175,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
if (result.getClass().isArray()) {
Assert.isTrue(methodAnnotatedOutboundNames.length == ((Object[]) result).length,
"Result does not match with the number of declared outbounds");
} else {
}
else {
Assert.isTrue(methodAnnotatedOutboundNames.length == 1,
"Result does not match with the number of declared outbounds");
}
@@ -184,16 +185,17 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
int i = 0;
for (Object outboundKStream : outboundKStreams) {
Object targetBean = this.applicationContext.getBean(methodAnnotatedOutboundNames[i++]);
for (StreamListenerResultAdapter streamListenerResultAdapter : streamListenerResultAdapters) {
for (StreamListenerResultAdapter streamListenerResultAdapter : this.streamListenerResultAdapters) {
if (streamListenerResultAdapter.supports(outboundKStream.getClass(), targetBean.getClass())) {
streamListenerResultAdapter.adapt(outboundKStream, targetBean);
break;
}
}
}
} else {
}
else {
Object targetBean = this.applicationContext.getBean(methodAnnotatedOutboundNames[0]);
for (StreamListenerResultAdapter streamListenerResultAdapter : streamListenerResultAdapters) {
for (StreamListenerResultAdapter streamListenerResultAdapter : this.streamListenerResultAdapters) {
if (streamListenerResultAdapter.supports(result.getClass(), targetBean.getClass())) {
streamListenerResultAdapter.adapt(result, targetBean);
break;
@@ -202,8 +204,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
}
}
}
catch (Exception e) {
throw new BeanInitializationException("Cannot setup StreamListener for " + method, e);
catch (Exception ex) {
throw new BeanInitializationException("Cannot setup StreamListener for " + method, ex);
}
}
@@ -228,17 +230,17 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
if (targetReferenceValue != null) {
Assert.isInstanceOf(String.class, targetReferenceValue, "Annotation value must be a String");
Object targetBean = applicationContext.getBean((String) targetReferenceValue);
BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(inboundName);
BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(inboundName);
enableNativeDecodingForKTableAlways(parameterType, bindingProperties);
//Retrieve the StreamsConfig created for this method if available.
//Otherwise, create the StreamsBuilderFactory and get the underlying config.
if (!methodStreamsBuilderFactoryBeanMap.containsKey(method)) {
if (!this.methodStreamsBuilderFactoryBeanMap.containsKey(method)) {
buildStreamsBuilderAndRetrieveConfig(method, applicationContext, inboundName);
}
try {
StreamsBuilderFactoryBean streamsBuilderFactoryBean = methodStreamsBuilderFactoryBeanMap.get(method);
StreamsBuilderFactoryBean streamsBuilderFactoryBean = this.methodStreamsBuilderFactoryBeanMap.get(method);
StreamsBuilder streamsBuilder = streamsBuilderFactoryBean.getObject();
KafkaStreamsConsumerProperties extendedConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(inboundName);
KafkaStreamsConsumerProperties extendedConsumerProperties = this.kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(inboundName);
//get state store spec
KafkaStreamsStateStoreProperties spec = buildStateStoreSpec(method);
Serde<?> keySerde = this.keyValueSerdeResolver.getInboundKeySerde(extendedConsumerProperties);
@@ -266,7 +268,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
KStreamBoundElementFactory.KStreamWrapper kStreamWrapper = (KStreamBoundElementFactory.KStreamWrapper) targetBean;
//wrap the proxy created during the initial target type binding with real object (KStream)
kStreamWrapper.wrap((KStream<Object, Object>) stream);
kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean);
this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean);
for (StreamListenerParameterAdapter streamListenerParameterAdapter : streamListenerParameterAdapters) {
if (streamListenerParameterAdapter.supports(stream.getClass(), methodParameter)) {
arguments[parameterIndex] = streamListenerParameterAdapter.adapt(kStreamWrapper, methodParameter);
@@ -281,29 +283,29 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
}
else if (parameterType.isAssignableFrom(KTable.class)) {
String materializedAs = extendedConsumerProperties.getMaterializedAs();
String bindingDestination = bindingServiceProperties.getBindingDestination(inboundName);
String bindingDestination = this.bindingServiceProperties.getBindingDestination(inboundName);
KTable<?, ?> table = getKTable(streamsBuilder, keySerde, valueSerde, materializedAs,
bindingDestination, autoOffsetReset);
KTableBoundElementFactory.KTableWrapper kTableWrapper = (KTableBoundElementFactory.KTableWrapper) targetBean;
//wrap the proxy created during the initial target type binding with real object (KTable)
kTableWrapper.wrap((KTable<Object, Object>) table);
kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean);
this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean);
arguments[parameterIndex] = table;
}
else if (parameterType.isAssignableFrom(GlobalKTable.class)) {
String materializedAs = extendedConsumerProperties.getMaterializedAs();
String bindingDestination = bindingServiceProperties.getBindingDestination(inboundName);
String bindingDestination = this.bindingServiceProperties.getBindingDestination(inboundName);
GlobalKTable<?, ?> table = getGlobalKTable(streamsBuilder, keySerde, valueSerde, materializedAs,
bindingDestination, autoOffsetReset);
GlobalKTableBoundElementFactory.GlobalKTableWrapper globalKTableWrapper = (GlobalKTableBoundElementFactory.GlobalKTableWrapper) targetBean;
//wrap the proxy created during the initial target type binding with real object (KTable)
globalKTableWrapper.wrap((GlobalKTable<Object, Object>) table);
kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean);
this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactory(streamsBuilderFactoryBean);
arguments[parameterIndex] = table;
}
}
catch (Exception e) {
throw new IllegalStateException(e);
catch (Exception ex) {
throw new IllegalStateException(ex);
}
}
else {
@@ -314,7 +316,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
}
private GlobalKTable<?, ?> getGlobalKTable(StreamsBuilder streamsBuilder, Serde<?> keySerde, Serde<?> valueSerde, String materializedAs,
String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) {
String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) {
return materializedAs != null ?
materializedAsGlobalKTable(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde, autoOffsetReset) :
streamsBuilder.globalTable(bindingDestination,
@@ -322,24 +324,24 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
}
private KTable<?, ?> getKTable(StreamsBuilder streamsBuilder, Serde<?> keySerde, Serde<?> valueSerde, String materializedAs,
String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) {
String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) {
return materializedAs != null ?
materializedAs(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde, autoOffsetReset ) :
materializedAs(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde, autoOffsetReset) :
streamsBuilder.table(bindingDestination,
Consumed.with(keySerde, valueSerde).withOffsetResetPolicy(autoOffsetReset));
}
private <K,V> KTable<K,V> materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName, Serde<K> k, Serde<V> v,
Topology.AutoOffsetReset autoOffsetReset) {
return streamsBuilder.table(bindingServiceProperties.getBindingDestination(destination),
Consumed.with(k,v).withOffsetResetPolicy(autoOffsetReset),
private <K, V> KTable<K, V> materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName, Serde<K> k, Serde<V> v,
Topology.AutoOffsetReset autoOffsetReset) {
return streamsBuilder.table(this.bindingServiceProperties.getBindingDestination(destination),
Consumed.with(k, v).withOffsetResetPolicy(autoOffsetReset),
getMaterialized(storeName, k, v));
}
private <K,V> GlobalKTable<K,V> materializedAsGlobalKTable(StreamsBuilder streamsBuilder, String destination, String storeName, Serde<K> k, Serde<V> v,
Topology.AutoOffsetReset autoOffsetReset) {
return streamsBuilder.globalTable(bindingServiceProperties.getBindingDestination(destination),
Consumed.with(k,v).withOffsetResetPolicy(autoOffsetReset),
private <K, V> GlobalKTable<K, V> materializedAsGlobalKTable(StreamsBuilder streamsBuilder, String destination, String storeName, Serde<K> k, Serde<V> v,
Topology.AutoOffsetReset autoOffsetReset) {
return streamsBuilder.globalTable(this.bindingServiceProperties.getBindingDestination(destination),
Consumed.with(k, v).withOffsetResetPolicy(autoOffsetReset),
getMaterialized(storeName, k, v));
}
@@ -375,18 +377,17 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
if (spec.isLoggingDisabled()) {
builder = builder.withLoggingDisabled();
}
return builder;
}catch (Exception e) {
LOG.error("failed to build state store exception : " + e);
throw e;
}
catch (Exception ex) {
LOG.error("failed to build state store exception : " + ex);
throw ex;
}
}
private KStream<?, ?> getkStream(String inboundName, KafkaStreamsStateStoreProperties storeSpec,
BindingProperties bindingProperties,
StreamsBuilder streamsBuilder,
StreamsBuilder streamsBuilder,
Serde<?> keySerde, Serde<?> valueSerde, Topology.AutoOffsetReset autoOffsetReset) {
if (storeSpec != null) {
StoreBuilder storeBuilder = buildStateStore(storeSpec);
@@ -396,21 +397,21 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
}
}
String[] bindingTargets = StringUtils
.commaDelimitedListToStringArray(bindingServiceProperties.getBindingDestination(inboundName));
.commaDelimitedListToStringArray(this.bindingServiceProperties.getBindingDestination(inboundName));
KStream<?, ?> stream =
streamsBuilder.stream(Arrays.asList(bindingTargets),
Consumed.with(keySerde, valueSerde)
.withOffsetResetPolicy(autoOffsetReset));
final boolean nativeDecoding = bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding();
if (nativeDecoding){
final boolean nativeDecoding = this.bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding();
if (nativeDecoding) {
LOG.info("Native decoding is enabled for " + inboundName + ". Inbound deserialization done at the broker.");
}
else {
LOG.info("Native decoding is disabled for " + inboundName + ". Inbound message conversion done by Spring Cloud Stream.");
}
stream = stream.mapValues(value -> {
stream = stream.mapValues((value) -> {
Object returnValue;
String contentType = bindingProperties.getContentType();
if (value != null && !StringUtils.isEmpty(contentType) && !nativeDecoding) {
@@ -442,7 +443,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
Map<String, Object> streamConfigGlobalProperties = applicationContext.getBean("streamConfigGlobalProperties", Map.class);
KafkaStreamsConsumerProperties extendedConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(inboundName);
KafkaStreamsConsumerProperties extendedConsumerProperties = this.kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(inboundName);
streamConfigGlobalProperties.putAll(extendedConsumerProperties.getConfiguration());
String applicationId = extendedConsumerProperties.getApplicationId();
@@ -451,7 +452,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
}
int concurrency = bindingServiceProperties.getConsumerProperties(inboundName).getConcurrency();
int concurrency = this.bindingServiceProperties.getConsumerProperties(inboundName).getConcurrency();
// override concurrency if set at the individual binding level.
if (concurrency > 1) {
streamConfigGlobalProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, concurrency);
@@ -477,7 +478,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
.getRawBeanDefinition();
((BeanDefinitionRegistry) beanFactory).registerBeanDefinition("stream-builder-" + method.getName(), streamsBuilderBeanDefinition);
StreamsBuilderFactoryBean streamsBuilderX = applicationContext.getBean("&stream-builder-" + method.getName(), StreamsBuilderFactoryBean.class);
methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderX);
this.methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderX);
}
@Override
@@ -508,15 +509,15 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
private boolean isDeclarativeOutput(Method m, String targetBeanName) {
boolean declarative;
Class<?> returnType = m.getReturnType();
if (returnType.isArray()){
if (returnType.isArray()) {
Class<?> targetBeanClass = this.applicationContext.getType(targetBeanName);
declarative = this.streamListenerResultAdapters.stream()
.anyMatch(slpa -> slpa.supports(returnType.getComponentType(), targetBeanClass));
.anyMatch((slpa) -> slpa.supports(returnType.getComponentType(), targetBeanClass));
return declarative;
}
Class<?> targetBeanClass = this.applicationContext.getType(targetBeanName);
declarative = this.streamListenerResultAdapters.stream()
.anyMatch(slpa -> slpa.supports(returnType, targetBeanClass));
.anyMatch((slpa) -> slpa.supports(returnType, targetBeanClass));
return declarative;
}

41
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java
View File

@@ -44,25 +44,26 @@ import org.springframework.util.StringUtils;
* If native encoding is disabled, then the binder will do serialization using a contentType. Keys are always serialized
* by the broker.
*
* For state store, use serdes class specified in {@link KafkaStreamsStateStore} to create Serde accordingly.
* For state store, use serdes class specified in
* {@link org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsStateStore} to create Serde accordingly.
*
* @author Soby Chacko
* @author Lei Chen
*/
class KeyValueSerdeResolver {
private final Map<String,Object> streamConfigGlobalProperties;
private final Map<String, Object> streamConfigGlobalProperties;
private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties;
KeyValueSerdeResolver(Map<String,Object> streamConfigGlobalProperties,
KeyValueSerdeResolver(Map<String, Object> streamConfigGlobalProperties,
KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) {
this.streamConfigGlobalProperties = streamConfigGlobalProperties;
this.binderConfigurationProperties = binderConfigurationProperties;
}
/**
* Provide the {@link Serde} for inbound key
* Provide the {@link Serde} for inbound key.
*
* @param extendedConsumerProperties binding level extended {@link KafkaStreamsConsumerProperties}
* @return configurd {@link Serde} for the inbound key.
@@ -74,7 +75,7 @@ class KeyValueSerdeResolver {
}
/**
* Provide the {@link Serde} for inbound value
* Provide the {@link Serde} for inbound value.
*
* @param consumerProperties {@link ConsumerProperties} on binding
* @param extendedConsumerProperties binding level extended {@link KafkaStreamsConsumerProperties}
@@ -92,16 +93,16 @@ class KeyValueSerdeResolver {
else {
valueSerde = Serdes.ByteArray();
}
valueSerde.configure(streamConfigGlobalProperties, false);
valueSerde.configure(this.streamConfigGlobalProperties, false);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
catch (ClassNotFoundException ex) {
throw new IllegalStateException("Serde class not found: ", ex);
}
return valueSerde;
}
/**
* Provide the {@link Serde} for outbound key
* Provide the {@link Serde} for outbound key.
*
* @param properties binding level extended {@link KafkaStreamsProducerProperties}
* @return configurd {@link Serde} for the outbound key.
@@ -111,7 +112,7 @@ class KeyValueSerdeResolver {
}
/**
* Provide the {@link Serde} for outbound value
* Provide the {@link Serde} for outbound value.
*
* @param producerProperties {@link ProducerProperties} on binding
* @param kafkaStreamsProducerProperties binding level extended {@link KafkaStreamsProducerProperties}
@@ -126,16 +127,16 @@ class KeyValueSerdeResolver {
else {
valueSerde = Serdes.ByteArray();
}
valueSerde.configure(streamConfigGlobalProperties, false);
valueSerde.configure(this.streamConfigGlobalProperties, false);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
catch (ClassNotFoundException ex) {
throw new IllegalStateException("Serde class not found: ", ex);
}
return valueSerde;
}
/**
* Provide the {@link Serde} for state store
* Provide the {@link Serde} for state store.
*
* @param keySerdeString serde class used for key
* @return {@link Serde} for the state store key.
@@ -145,7 +146,7 @@ class KeyValueSerdeResolver {
}
/**
* Provide the {@link Serde} for state store value
* Provide the {@link Serde} for state store value.
*
* @param valueSerdeString serde class used for value
* @return {@link Serde} for the state store value.
@@ -154,8 +155,8 @@ class KeyValueSerdeResolver {
try {
return getValueSerde(valueSerdeString);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
catch (ClassNotFoundException ex) {
throw new IllegalStateException("Serde class not found: ", ex);
}
}
@@ -169,11 +170,11 @@ class KeyValueSerdeResolver {
keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("default.key.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("default.key.serde"), Serde.class) : Serdes.ByteArray();
}
keySerde.configure(streamConfigGlobalProperties, true);
keySerde.configure(this.streamConfigGlobalProperties, true);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
catch (ClassNotFoundException ex) {
throw new IllegalStateException("Serde class not found: ", ex);
}
return keySerde;
}

2
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java
View File

@@ -49,7 +49,7 @@ public class QueryableStoreRegistry {
public <T> T getQueryableStoreType(String storeName, QueryableStoreType<T> storeType) {
for (KafkaStreams kafkaStream : this.kafkaStreamsRegistry.getKafkaStreams()) {
try{
try {
T store = kafkaStream.store(storeName, storeType);
if (store != null) {
return store;

26
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java
View File

@@ -35,12 +35,14 @@ 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
* @since 2.0.0
*/
public class SendToDlqAndContinue implements DeserializationExceptionHandler{
public class SendToDlqAndContinue implements DeserializationExceptionHandler {
/**
* Key used for DLQ dispatchers.
*/
public static final String KAFKA_STREAMS_DLQ_DISPATCHERS = "spring.cloud.stream.kafka.streams.dlq.dispatchers";
/**
@@ -57,9 +59,9 @@ public class SendToDlqAndContinue implements DeserializationExceptionHandler{
* @param value to send
* @param partition for the topic where this record should be sent
*/
public void sendToDlq(String topic, byte[] key, byte[] value, int partition){
public void sendToDlq(String topic, byte[] key, byte[] value, int partition) {
KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch = this.dlqDispatchers.get(topic);
kafkaStreamsDlqDispatch.sendToDlq(key,value, partition);
kafkaStreamsDlqDispatch.sendToDlq(key, value, partition);
}
@Override
@@ -75,19 +77,19 @@ public class SendToDlqAndContinue implements DeserializationExceptionHandler{
// 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;
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;
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;
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);
@@ -105,7 +107,7 @@ public class SendToDlqAndContinue implements DeserializationExceptionHandler{
this.dlqDispatchers = (Map<String, KafkaStreamsDlqDispatch>) configs.get(KAFKA_STREAMS_DLQ_DISPATCHERS);
}
void addKStreamDlqDispatch(String topic, KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch){
void addKStreamDlqDispatch(String topic, KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch) {
this.dlqDispatchers.put(topic, kafkaStreamsDlqDispatch);
}

11
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java
View File

@@ -68,11 +68,12 @@ class StreamsBuilderFactoryManager implements SmartLifecycle {
Set<StreamsBuilderFactoryBean> streamsBuilderFactoryBeans = this.kafkaStreamsBindingInformationCatalogue.getStreamsBuilderFactoryBeans();
for (StreamsBuilderFactoryBean streamsBuilderFactoryBean : streamsBuilderFactoryBeans) {
streamsBuilderFactoryBean.start();
kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean.getKafkaStreams());
this.kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean.getKafkaStreams());
}
this.running = true;
} catch (Exception e) {
throw new KafkaException("Could not start stream: ", e);
}
catch (Exception ex) {
throw new KafkaException("Could not start stream: ", ex);
}
}
}
@@ -86,8 +87,8 @@ class StreamsBuilderFactoryManager implements SmartLifecycle {
streamsBuilderFactoryBean.stop();
}
}
catch (Exception e) {
throw new IllegalStateException(e);
catch (Exception ex) {
throw new IllegalStateException(ex);
}
finally {
this.running = false;

18
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsProcessor.java
View File

@@ -34,16 +34,16 @@ import org.springframework.cloud.stream.annotation.Output;
* <pre class="code">
* interface KStreamBranchProcessor {
* &#064;Input("input")
* KStream<?, ?> input();
* KStream&lt;?, ?&gt; input();
*
* &#064;Output("output-1")
* KStream<?, ?> output1();
* KStream&lt;?, ?&gt; output1();
*
* &#064;Output("output-2")
* KStream<?, ?> output2();
* KStream&lt;?, ?&gt; output2();
*
* &#064;Output("output-3")
* KStream<?, ?> output3();
* KStream&lt;?, ?&gt; output3();
*
* ......
*
@@ -53,13 +53,13 @@ import org.springframework.cloud.stream.annotation.Output;
* <pre class="code">
* interface KStreamKtableProcessor {
* &#064;Input("input-1")
* KStream<?, ?> input1();
* KStream&lt;?, ?&gt; input1();
*
* &#064;Input("input-2")
* KTable<?, ?> input2();
* KTable&lt;?, ?&gt; input2();
*
* &#064;Output("output")
* KStream<?, ?> output();
* KStream&lt;?, ?&gt; output();
*
* ......
*
@@ -72,12 +72,16 @@ import org.springframework.cloud.stream.annotation.Output;
public interface KafkaStreamsProcessor {
/**
* Input binding.
*
* @return {@link Input} binding for {@link KStream} type.
*/
@Input("input")
KStream<?, ?> input();
/**
* Output binding.
*
* @return {@link Output} binding for {@link KStream} type.
*/
@Output("output")

27
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsStateStore.java
View File

@@ -24,7 +24,6 @@ import java.lang.annotation.Target;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsStateStoreProperties;
/**
* Interface for Kafka Stream state store.
*
@@ -37,23 +36,23 @@ import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStr
* <pre class="code">
* &#064;StreamListener("input")
* &#064;KafkaStreamsStateStore(name="mystate", type= KafkaStreamsStateStoreProperties.StoreType.WINDOW, size=300000)
* public void process(KStream<Object, Product> input) {
* public void process(KStream&lt;Object, Product&gt; input) {
* ......
* }
*</pre>
* </pre>
*
* With that, you should be able to read/write this state store in your processor/transformer code.
*
* <pre class="code">
* new Processor<Object, Product>() {
* WindowStore<Object, String> state;
* new Processor&lt;Object, Product&gt;() {
* WindowStore&lt;Object, String&gt; state;
* &#064;Override
* public void init(ProcessorContext processorContext) {
* state = (WindowStore)processorContext.getStateStore("mystate");
* ......
* }
* }
*</pre>
* </pre>
*
* @author Lei Chen
*/
@@ -64,41 +63,57 @@ import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStr
public @interface KafkaStreamsStateStore {
/**
* Provides name of the state store.
*
* @return name of state store.
*/
String name() default "";
/**
* State store type.
*
* @return {@link KafkaStreamsStateStoreProperties.StoreType} of state store.
*/
KafkaStreamsStateStoreProperties.StoreType type() default KafkaStreamsStateStoreProperties.StoreType.KEYVALUE;
/**
* Serde used for key.
*
* @return key serde of state store.
*/
String keySerde() default "org.apache.kafka.common.serialization.Serdes$StringSerde";
/**
* Serde used for value.
*
* @return value serde of state store.
*/
String valueSerde() default "org.apache.kafka.common.serialization.Serdes$StringSerde";
/**
* Length in milli-second of Windowed store window.
*
* @return length in milli-second of window(for windowed store).
*/
long lengthMs() default 0;
/**
* Retention period for Windowed store windows.
*
* @return the maximum period of time in milli-second to keep each window in this store(for windowed store).
*/
long retentionMs() default 0;
/**
* Whether catching is enabled or not.
*
* @return whether caching should be enabled on the created store.
*/
boolean cache() default false;
/**
* Whether logging is enabled or not.
*
* @return whether logging should be enabled on the created store.
*/
boolean logging() default true;

9
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java
View File

@@ -32,13 +32,16 @@ public class KafkaStreamsApplicationSupportProperties {
private TimeWindow timeWindow;
public TimeWindow getTimeWindow() {
return timeWindow;
return this.timeWindow;
}
public void setTimeWindow(TimeWindow timeWindow) {
this.timeWindow = timeWindow;
}
/**
* Properties required by time windows.
*/
public static class TimeWindow {
private int length;
@@ -46,7 +49,7 @@ public class KafkaStreamsApplicationSupportProperties {
private int advanceBy;
public int getLength() {
return length;
return this.length;
}
public void setLength(int length) {
@@ -54,7 +57,7 @@ public class KafkaStreamsApplicationSupportProperties {
}
public int getAdvanceBy() {
return advanceBy;
return this.advanceBy;
}
public void setAdvanceBy(int advanceBy) {

18
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java
View File

@@ -20,6 +20,8 @@ import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
/**
* Kafka Streams binder configuration properties.
*
* @author Soby Chacko
* @author Gary Russell
*/
@@ -29,16 +31,28 @@ public class KafkaStreamsBinderConfigurationProperties extends KafkaBinderConfig
super(kafkaProperties);
}
/**
* Enumeration for various Serde errors.
*/
public enum SerdeError {
/**
* Deserialization error handler with log and continue.
*/
logAndContinue,
/**
* Deserialization error handler with log and fail.
*/
logAndFail,
/**
* Deserialization error handler with DLQ send.
*/
sendToDlq
}
private String applicationId;
public String getApplicationId() {
return applicationId;
return this.applicationId;
}
public void setApplicationId(String applicationId) {
@@ -53,7 +67,7 @@ public class KafkaStreamsBinderConfigurationProperties extends KafkaBinderConfig
private KafkaStreamsBinderConfigurationProperties.SerdeError serdeError;
public KafkaStreamsBinderConfigurationProperties.SerdeError getSerdeError() {
return serdeError;
return this.serdeError;
}
public void setSerdeError(KafkaStreamsBinderConfigurationProperties.SerdeError serdeError) {

6
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBindingProperties.java
View File

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams.properties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
/**
* Extended binding properties holder that delegates to Kafka Streams producer and consumer properties.
*
* @author Marius Bogoevici
*/
public class KafkaStreamsBindingProperties implements BinderSpecificPropertiesProvider {
@@ -28,7 +30,7 @@ public class KafkaStreamsBindingProperties implements BinderSpecificPropertiesPr
private KafkaStreamsProducerProperties producer = new KafkaStreamsProducerProperties();
public KafkaStreamsConsumerProperties getConsumer() {
return consumer;
return this.consumer;
}
public void setConsumer(KafkaStreamsConsumerProperties consumer) {
@@ -36,7 +38,7 @@ public class KafkaStreamsBindingProperties implements BinderSpecificPropertiesPr
}
public KafkaStreamsProducerProperties getProducer() {
return producer;
return this.producer;
}
public void setProducer(KafkaStreamsProducerProperties producer) {

12
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java
View File

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams.properties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
/**
* Extended properties for Kafka Streams consumer.
*
* @author Marius Bogoevici
* @author Soby Chacko
*/
@@ -37,12 +39,12 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties {
private String valueSerde;
/**
* Materialized as a KeyValueStore
* Materialized as a KeyValueStore.
*/
private String materializedAs;
public String getApplicationId() {
return applicationId;
return this.applicationId;
}
public void setApplicationId(String applicationId) {
@@ -50,7 +52,7 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties {
}
public String getKeySerde() {
return keySerde;
return this.keySerde;
}
public void setKeySerde(String keySerde) {
@@ -58,7 +60,7 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties {
}
public String getValueSerde() {
return valueSerde;
return this.valueSerde;
}
public void setValueSerde(String valueSerde) {
@@ -66,7 +68,7 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties {
}
public String getMaterializedAs() {
return materializedAs;
return this.materializedAs;
}
public void setMaterializedAs(String materializedAs) {

2
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java
View File

@@ -20,6 +20,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
/**
* Kafka streams specific extended binding properties class that extends from {@link AbstractExtendedBindingProperties}.
*
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/

6
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java
View File

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams.properties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
/**
* Extended properties for Kafka Streams producer.
*
* @author Marius Bogoevici
* @author Soby Chacko
*/
@@ -35,7 +37,7 @@ public class KafkaStreamsProducerProperties extends KafkaProducerProperties {
private String valueSerde;
public String getKeySerde() {
return keySerde;
return this.keySerde;
}
public void setKeySerde(String keySerde) {
@@ -43,7 +45,7 @@ public class KafkaStreamsProducerProperties extends KafkaProducerProperties {
}
public String getValueSerde() {
return valueSerde;
return this.valueSerde;
}
public void setValueSerde(String valueSerde) {

46
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsStateStoreProperties.java
View File

@@ -18,39 +18,49 @@ package org.springframework.cloud.stream.binder.kafka.streams.properties;
/**
* Properties for Kafka Streams state store.
*
* @author Lei Chen
*/
public class KafkaStreamsStateStoreProperties {
/**
* Enumeration for store type.
*/
public enum StoreType {
/**
* Key value store.
*/
KEYVALUE("keyvalue"),
/**
* Window store.
*/
WINDOW("window"),
SESSION("session")
;
/**
* Session store.
*/
SESSION("session");
private final String type;
/**
* @param type
*/
StoreType(final String type) {
this.type = type;
}
@Override
public String toString() {
return type;
return this.type;
}
}
/**
* name for this state store
* Name for this state store.
*/
private String name;
/**
* type for this state store
* Type for this state store.
*/
private StoreType type;
@@ -75,18 +85,18 @@ public class KafkaStreamsStateStoreProperties {
private String valueSerdeString;
/**
* Whether enable cache in this state store.
* Whether caching is enabled on this state store.
*/
private boolean cacheEnabled;
/**
* Whether enable logging in this state store.
* Whether logging is enabled on this state store.
*/
private boolean loggingDisabled;
public String getName() {
return name;
return this.name;
}
public void setName(String name) {
@@ -94,7 +104,7 @@ public class KafkaStreamsStateStoreProperties {
}
public StoreType getType() {
return type;
return this.type;
}
public void setType(StoreType type) {
@@ -102,7 +112,7 @@ public class KafkaStreamsStateStoreProperties {
}
public long getLength() {
return length;
return this.length;
}
public void setLength(long length) {
@@ -110,7 +120,7 @@ public class KafkaStreamsStateStoreProperties {
}
public long getRetention() {
return retention;
return this.retention;
}
public void setRetention(long retention) {
@@ -118,7 +128,7 @@ public class KafkaStreamsStateStoreProperties {
}
public String getKeySerdeString() {
return keySerdeString;
return this.keySerdeString;
}
public void setKeySerdeString(String keySerdeString) {
@@ -126,7 +136,7 @@ public class KafkaStreamsStateStoreProperties {
}
public String getValueSerdeString() {
return valueSerdeString;
return this.valueSerdeString;
}
public void setValueSerdeString(String valueSerdeString) {
@@ -134,7 +144,7 @@ public class KafkaStreamsStateStoreProperties {
}
public boolean isCacheEnabled() {
return cacheEnabled;
return this.cacheEnabled;
}
public void setCacheEnabled(boolean cacheEnabled) {
@@ -142,7 +152,7 @@ public class KafkaStreamsStateStoreProperties {
}
public boolean isLoggingDisabled() {
return loggingDisabled;
return this.loggingDisabled;
}
public void setLoggingDisabled(boolean loggingDisabled) {

20
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java
View File

@@ -55,7 +55,7 @@ import org.springframework.util.MimeTypeUtils;
* to be included in the configuration map with the key "contentType". For example,
*
* <pre class="code">
* Map<String, Object> config = new HashMap<>();
* Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
* config.put("valueClass", Foo.class);
* config.put("contentType", "application/avro");
* </pre>
@@ -68,6 +68,8 @@ import org.springframework.util.MimeTypeUtils;
* An instance of this class is provided as a bean by the binder configuration and typically the applications
* can autowire that bean. This is the expected usage pattern of this class.
*
* @param <T> type of the object to marshall
*
* @author Soby Chacko
* @since 2.1
*/
@@ -110,12 +112,12 @@ public class CompositeNonNativeSerde<T> implements Serde<T> {
}
private static MimeType resolveMimeType(Map<String, ?> configs) {
if (configs.containsKey(MessageHeaders.CONTENT_TYPE)){
String contentType = (String)configs.get(MessageHeaders.CONTENT_TYPE);
if (configs.containsKey(MessageHeaders.CONTENT_TYPE)) {
String contentType = (String) configs.get(MessageHeaders.CONTENT_TYPE);
if (DEFAULT_AVRO_MIME_TYPE.equals(MimeTypeUtils.parseMimeType(contentType))) {
return DEFAULT_AVRO_MIME_TYPE;
}
else if(contentType.contains("avro")) {
else if (contentType.contains("avro")) {
return MimeTypeUtils.parseMimeType("application/avro");
}
else {
@@ -130,7 +132,7 @@ public class CompositeNonNativeSerde<T> implements Serde<T> {
/**
* Custom {@link Deserializer} that uses the {@link CompositeMessageConverterFactory}.
*
* @param <U> Parameterized target type for deserialization
* @param <U> parameterized target type for deserialization
*/
private static class CompositeNonNativeDeserializer<U> implements Deserializer<U> {
@@ -158,7 +160,7 @@ public class CompositeNonNativeSerde<T> implements Serde<T> {
public U deserialize(String topic, byte[] data) {
Message<?> message = MessageBuilder.withPayload(data)
.setHeader(MessageHeaders.CONTENT_TYPE, this.mimeType.toString()).build();
U messageConverted = (U)messageConverter.fromMessage(message, this.valueClass);
U messageConverted = (U) this.messageConverter.fromMessage(message, this.valueClass);
Assert.notNull(messageConverted, "Deserialization failed.");
return messageConverted;
}
@@ -172,7 +174,7 @@ public class CompositeNonNativeSerde<T> implements Serde<T> {
/**
* Custom {@link Serializer} that uses the {@link CompositeMessageConverterFactory}.
*
* @param <V> Parameterized type for serialization
* @param <V> parameterized type for serialization
*/
private static class CompositeNonNativeSerializer<V> implements Serializer<V> {
@@ -194,9 +196,9 @@ public class CompositeNonNativeSerde<T> implements Serde<T> {
Map<String, Object> headers = new HashMap<>(message.getHeaders());
headers.put(MessageHeaders.CONTENT_TYPE, this.mimeType.toString());
MessageHeaders messageHeaders = new MessageHeaders(headers);
final Object payload = messageConverter.toMessage(message.getPayload(),
final Object payload = this.messageConverter.toMessage(message.getPayload(),
messageHeaders).getPayload();
return (byte[])payload;
return (byte[]) payload;
}
@Override

94
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
View File

@@ -73,52 +73,7 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
@Override
public Health health() {
ExecutorService exec = Executors.newSingleThreadExecutor();
Future<Health> future = exec.submit(() -> {
try {
if (this.metadataConsumer == null) {
synchronized (KafkaBinderHealthIndicator.this) {
if (this.metadataConsumer == null) {
this.metadataConsumer = this.consumerFactory.createConsumer();
}
}
}
synchronized (this.metadataConsumer) {
Set<String> downMessages = new HashSet<>();
final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse =
KafkaBinderHealthIndicator.this.binder.getTopicsInUse();
if (topicsInUse.isEmpty()) {
return Health.down()
.withDetail("No topic information available", "Kafka broker is not reachable")
.build();
}
else {
for (String topic : topicsInUse.keySet()) {
KafkaMessageChannelBinder.TopicInformation topicInformation = topicsInUse.get(topic);
if (!topicInformation.isTopicPattern()) {
List<PartitionInfo> partitionInfos = this.metadataConsumer.partitionsFor(topic);
for (PartitionInfo partitionInfo : partitionInfos) {
if (topicInformation.getPartitionInfos()
.contains(partitionInfo) && partitionInfo.leader().id() == -1) {
downMessages.add(partitionInfo.toString());
}
}
}
}
}
if (downMessages.isEmpty()) {
return Health.up().build();
}
else {
return Health.down()
.withDetail("Following partitions in use have no leaders: ", downMessages.toString())
.build();
}
}
}
catch (Exception ex) {
return Health.down(ex).build();
}
});
Future<Health> future = exec.submit(this::buildHealthStatus);
try {
return future.get(this.timeout, TimeUnit.SECONDS);
}
@@ -141,4 +96,51 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
}
}
private Health buildHealthStatus() {
try {
if (this.metadataConsumer == null) {
synchronized (KafkaBinderHealthIndicator.this) {
if (this.metadataConsumer == null) {
this.metadataConsumer = this.consumerFactory.createConsumer();
}
}
}
synchronized (this.metadataConsumer) {
Set<String> downMessages = new HashSet<>();
final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse =
KafkaBinderHealthIndicator.this.binder.getTopicsInUse();
if (topicsInUse.isEmpty()) {
return Health.down()
.withDetail("No topic information available", "Kafka broker is not reachable")
.build();
}
else {
for (String topic : topicsInUse.keySet()) {
KafkaMessageChannelBinder.TopicInformation topicInformation = topicsInUse.get(topic);
if (!topicInformation.isTopicPattern()) {
List<PartitionInfo> partitionInfos = this.metadataConsumer.partitionsFor(topic);
for (PartitionInfo partitionInfo : partitionInfos) {
if (topicInformation.getPartitionInfos()
.contains(partitionInfo) && partitionInfo.leader().id() == -1) {
downMessages.add(partitionInfo.toString());
}
}
}
}
}
if (downMessages.isEmpty()) {
return Health.up().build();
}
else {
return Health.down()
.withDetail("Following partitions in use have no leaders: ", downMessages.toString())
.build();
}
}
}
catch (Exception ex) {
return Health.down(ex).build();
}
}
}

2
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
View File

@@ -21,8 +21,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;