From b589f32933c666e370d1f3f3f6a980ce0776241b Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 9 Nov 2018 18:11:51 -0500 Subject: [PATCH] Checkstyle fixes Reuse checkstyle components from core spring-cloud-stream-tools module. Addressing checkstyle warnings in kafka streams binder module. Resolves #489 --- pom.xml | 25 +++++ .../pom.xml | 8 +- .../kafka/streams/GlobalKTableBinder.java | 8 +- .../GlobalKTableBinderConfiguration.java | 2 + .../GlobalKTableBoundElementFactory.java | 9 +- .../streams/InteractiveQueryService.java | 10 +- .../binder/kafka/streams/KStreamBinder.java | 15 +-- .../streams/KStreamBinderConfiguration.java | 47 +++++----- .../streams/KStreamBoundElementFactory.java | 11 ++- ...KStreamStreamListenerParameterAdapter.java | 6 +- .../KStreamStreamListenerResultAdapter.java | 2 + .../binder/kafka/streams/KTableBinder.java | 8 +- .../streams/KTableBinderConfiguration.java | 6 +- .../streams/KTableBoundElementFactory.java | 9 +- ...msApplicationSupportAutoConfiguration.java | 2 + ...StreamsBinderSupportAutoConfiguration.java | 10 +- .../streams/KafkaStreamsBinderUtils.java | 13 ++- ...fkaStreamsBindingInformationCatalogue.java | 40 ++++---- .../streams/KafkaStreamsDlqDispatch.java | 26 ++--- ...KafkaStreamsMessageConversionDelegate.java | 43 ++++----- .../kafka/streams/KafkaStreamsRegistry.java | 2 +- ...StreamListenerSetupMethodOrchestrator.java | 91 +++++++++--------- .../kafka/streams/KeyValueSerdeResolver.java | 41 ++++---- .../kafka/streams/QueryableStoreRegistry.java | 2 +- .../kafka/streams/SendToDlqAndContinue.java | 26 ++--- .../streams/StreamsBuilderFactoryManager.java | 11 ++- .../annotations/KafkaStreamsProcessor.java | 18 ++-- .../annotations/KafkaStreamsStateStore.java | 27 ++++-- ...kaStreamsApplicationSupportProperties.java | 9 +- ...aStreamsBinderConfigurationProperties.java | 18 +++- .../KafkaStreamsBindingProperties.java | 6 +- .../KafkaStreamsConsumerProperties.java | 12 ++- ...KafkaStreamsExtendedBindingProperties.java | 2 + .../KafkaStreamsProducerProperties.java | 6 +- .../KafkaStreamsStateStoreProperties.java | 46 +++++---- .../serde/CompositeNonNativeSerde.java | 20 ++-- .../kafka/KafkaBinderHealthIndicator.java | 94 ++++++++++--------- .../binder/kafka/KafkaBinderMetricsTest.java | 2 +- 38 files changed, 432 insertions(+), 301 deletions(-) diff --git a/pom.xml b/pom.xml index 800c469e3..320947bd8 100644 --- a/pom.xml +++ b/pom.xml @@ -145,6 +145,31 @@ org.apache.maven.plugins maven-checkstyle-plugin + + + org.springframework.cloud + spring-cloud-stream-tools + ${spring-cloud-stream.version} + + + + + checkstyle-validation + validate + + checkstyle.xml + checkstyle-header.txt + checkstyle-suppressions.xml + UTF-8 + true + true + true + + + check + + + diff --git a/spring-cloud-stream-binder-kafka-streams/pom.xml b/spring-cloud-stream-binder-kafka-streams/pom.xml index 3f03769ec..c2be860b0 100644 --- a/spring-cloud-stream-binder-kafka-streams/pom.xml +++ b/spring-cloud-stream-binder-kafka-streams/pom.xml @@ -102,14 +102,14 @@ ${avro.version} - generate-sources + generate-test-sources schema - protocol - idl-protocol - src/test/resources/avro + ${project.basedir}/target/generated-test-sources + ${project.basedir}/target/generated-test-sources + ${project.basedir}/src/test/resources/avro diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java index 1923b8518..ab2a07c3e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java @@ -56,7 +56,7 @@ public class GlobalKTableBinder extends private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties(); public GlobalKTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, - Map kafkaStreamsDlqDispatchers) { + Map kafkaStreamsDlqDispatchers) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; this.kafkaStreamsDlqDispatchers = kafkaStreamsDlqDispatchers; @@ -67,11 +67,11 @@ public class GlobalKTableBinder extends protected Binding> doBindConsumer(String name, String group, GlobalKTable inputTarget, ExtendedConsumerProperties 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); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java index 917899840..e047f7f99 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java @@ -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 */ diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java index 9578c0e45..f68a98da7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java @@ -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 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()); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java index ab06a1900..3be5d0da3 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java @@ -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 generic queryable store + * @param generic queryable store * @return queryable store. */ public T getQueryableStore(String storeName, QueryableStoreType 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 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 HostInfo getHostInfo(String store, K key, Serializer 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() diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index 89ea6c4e4..8327045e1 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java @@ -52,7 +52,7 @@ class KStreamBinder extends AbstractBinder, ExtendedConsumerProperties, ExtendedProducerProperties> implements ExtendedPropertiesBinder, 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 kafkaStreamsDlqDispatchers) { + Map kafkaStreamsDlqDispatchers) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; this.kafkaStreamsMessageConversionDelegate = kafkaStreamsMessageConversionDelegate; @@ -88,11 +88,11 @@ class KStreamBinder extends ExtendedConsumerProperties 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 keySerde, Serde 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)); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java index 2ac8709a6..0c13a0929 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java @@ -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 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 kafkaStreamsDlqDispatchers) { - KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, - KafkaStreamsMessageConversionDelegate, KafkaStreamsBindingInformationCatalogue, - keyValueSerdeResolver, kafkaStreamsDlqDispatchers); - kStreamBinder.setKafkaStreamsExtendedBindingProperties(kafkaStreamsExtendedBindingProperties); - return kStreamBinder; - } - } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java index b09e48d2b..18cfa740c 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java @@ -64,18 +64,21 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory { } 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 delegate); @@ -95,9 +98,9 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory { @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()); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.java index 96567c0c5..3ba721c19 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.java @@ -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> { +class KStreamStreamListenerParameterAdapter implements StreamListenerParameterAdapter, 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); } } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java index 72f0ad005..1fdb98949 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java @@ -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 */ diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java index 7d006e76f..35a2b47b1 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java @@ -55,7 +55,7 @@ class KTableBinder extends private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties(); KTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, - Map kafkaStreamsDlqDispatchers) { + Map kafkaStreamsDlqDispatchers) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; this.kafkaStreamsDlqDispatchers = kafkaStreamsDlqDispatchers; @@ -66,11 +66,11 @@ class KTableBinder extends protected Binding> doBindConsumer(String name, String group, KTable inputTarget, ExtendedConsumerProperties 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); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java index 6425f3c78..9e956ece3 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java @@ -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 kafkaStreamsDlqDispatchers) { + @Qualifier("kafkaStreamsDlqDispatchers") Map kafkaStreamsDlqDispatchers) { KTableBinder kStreamBinder = new KTableBinder(binderConfigurationProperties, kafkaTopicProvisioner, kafkaStreamsDlqDispatchers); return kStreamBinder; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java index 3a306b82f..aa917021f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java @@ -48,7 +48,7 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory { //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 { throw new UnsupportedOperationException("Outbound operations are not allowed on target type KTable"); } + /** + * Wrapper for KTable proxy. + */ public interface KTableWrapper { void wrap(KTable delegate); } @@ -78,9 +81,9 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory { @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()); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java index 0319571da..7d681a8da 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsApplicationSupportAutoConfiguration.java @@ -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 diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index fb25bb8cc..5387302a0 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -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 diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java index bc13d8d43..560e0ef9b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java @@ -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 properties, - Map kafkaStreamsDlqDispatchers) { + Map kafkaStreamsDlqDispatchers) { ExtendedConsumerProperties 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"; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java index d48d70bbf..46cfa918a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java @@ -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 getStreamsBuilderFactoryBeans() { - return streamsBuilderFactoryBeans; + return this.streamsBuilderFactoryBeans; } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java index 94e2a07ba..a625761b9 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java @@ -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 kafkaTemplate; + private final KafkaTemplate kafkaTemplate; private final String dlqName; KafkaStreamsDlqDispatch(String dlqName, KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties, KafkaConsumerProperties kafkaConsumerProperties) { - ProducerFactory producerFactory = getProducerFactory( + ProducerFactory 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 producerRecord = new ProducerRecord<>(this.dlqName, partittion, + ProducerRecord 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> sentDlq = null; + ListenableFuture> sentDlq = null; try { sentDlq = this.kafkaTemplate.send(producerRecord); - sentDlq.addCallback(new ListenableFutureCallback>() { + sentDlq.addCallback(new ListenableFutureCallback>() { @Override public void onFailure(Throwable ex) { @@ -84,7 +86,7 @@ class KafkaStreamsDlqDispatch { } @Override - public void onSuccess(SendResult result) { + public void onSuccess(SendResult result) { if (KafkaStreamsDlqDispatch.this.logger.isDebugEnabled()) { KafkaStreamsDlqDispatch.this.logger.debug( "Sent to DLQ " + sb.toString()); @@ -100,7 +102,7 @@ class KafkaStreamsDlqDispatch { } } - private DefaultKafkaProducerFactory getProducerFactory(ExtendedProducerProperties producerProperties, + private DefaultKafkaProducerFactory getProducerFactory(ExtendedProducerProperties producerProperties, KafkaBinderConfigurationProperties configurationProperties) { Map props = new HashMap<>(); props.put(ProducerConfig.RETRIES_CONFIG, 0); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java index ea4bfcc4e..6b0e5f641 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java @@ -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> 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
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; + } + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java index 818e891ae..4b82ded1c 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java @@ -32,7 +32,7 @@ class KafkaStreamsRegistry { private final Set kafkaStreams = new HashSet<>(); Set getKafkaStreams() { - return kafkaStreams; + return this.kafkaStreams; } /** diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index 2afc35809..10557f754 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -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) 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) 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) 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 KTable materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v, - Topology.AutoOffsetReset autoOffsetReset) { - return streamsBuilder.table(bindingServiceProperties.getBindingDestination(destination), - Consumed.with(k,v).withOffsetResetPolicy(autoOffsetReset), + private KTable materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v, + Topology.AutoOffsetReset autoOffsetReset) { + return streamsBuilder.table(this.bindingServiceProperties.getBindingDestination(destination), + Consumed.with(k, v).withOffsetResetPolicy(autoOffsetReset), getMaterialized(storeName, k, v)); } - private GlobalKTable materializedAsGlobalKTable(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v, - Topology.AutoOffsetReset autoOffsetReset) { - return streamsBuilder.globalTable(bindingServiceProperties.getBindingDestination(destination), - Consumed.with(k,v).withOffsetResetPolicy(autoOffsetReset), + private GlobalKTable materializedAsGlobalKTable(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde 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 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; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java index 4fd5d4cf7..3ce9ac28e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java @@ -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 streamConfigGlobalProperties; + private final Map streamConfigGlobalProperties; private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties; - KeyValueSerdeResolver(Map streamConfigGlobalProperties, + KeyValueSerdeResolver(Map 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; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java index c41e8ff5f..70f09c81e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java @@ -49,7 +49,7 @@ public class QueryableStoreRegistry { public T getQueryableStoreType(String storeName, QueryableStoreType storeType) { for (KafkaStreams kafkaStream : this.kafkaStreamsRegistry.getKafkaStreams()) { - try{ + try { T store = kafkaStream.store(storeName, storeType); if (store != null) { return store; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java index 6d2f6e6de..ed300eafd 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java @@ -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 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) configs.get(KAFKA_STREAMS_DLQ_DISPATCHERS); } - void addKStreamDlqDispatch(String topic, KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch){ + void addKStreamDlqDispatch(String topic, KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch) { this.dlqDispatchers.put(topic, kafkaStreamsDlqDispatch); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java index d98249ef5..f34bfb4cd 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java @@ -68,11 +68,12 @@ class StreamsBuilderFactoryManager implements SmartLifecycle { Set 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; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsProcessor.java index 3b01a2e9c..43b44de4e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsProcessor.java @@ -34,16 +34,16 @@ import org.springframework.cloud.stream.annotation.Output; * 
  *     interface KStreamBranchProcessor {
  *         @Input("input")
- *         KStream input();
+ *         KStream<?, ?> input();
  *
  *         @Output("output-1")
- *         KStream output1();
+ *         KStream<?, ?> output1();
  *
  *         @Output("output-2")
- *         KStream output2();
+ *         KStream<?, ?> output2();
  *
  *         @Output("output-3")
- *         KStream output3();
+ *         KStream<?, ?> output3();
  *
  *         ......
  *
@@ -53,13 +53,13 @@ import org.springframework.cloud.stream.annotation.Output;
  * 
  *     interface KStreamKtableProcessor {
  *         @Input("input-1")
- *         KStream input1();
+ *         KStream<?, ?> input1();
  *
  *         @Input("input-2")
- *         KTable input2();
+ *         KTable<?, ?> input2();
  *
  *         @Output("output")
- *         KStream output();
+ *         KStream<?, ?> 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")
diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsStateStore.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsStateStore.java
index b8997370d..d6897df3e 100644
--- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsStateStore.java
+++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/annotations/KafkaStreamsStateStore.java
@@ -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
  * 
  *     @StreamListener("input")
  *     @KafkaStreamsStateStore(name="mystate", type= KafkaStreamsStateStoreProperties.StoreType.WINDOW, size=300000)
- *	   public void process(KStream input) {
+ *	   public void process(KStream<Object, Product> input) {
  *         ......
  *     }
- *

+ * 

  *
  * With that, you should be able to read/write this state store in your processor/transformer code.
  *
  * 
- * 		new Processor() {
- * 			WindowStore state;
+ * 		new Processor<Object, Product>() {
+ * 			WindowStore<Object, String> state;
  * 			@Override
  *			public void init(ProcessorContext processorContext) {
  *			state = (WindowStore)processorContext.getStateStore("mystate");
  *				......
  *			}
  *		}
- *

+ *
* * @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; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java index 402b27ba3..08296784a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsApplicationSupportProperties.java @@ -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) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java index f057e9b12..9a9a235db 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java @@ -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) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBindingProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBindingProperties.java index c53a3a050..eb921cb73 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBindingProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBindingProperties.java @@ -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) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java index 35393310e..678833c1d 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java @@ -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) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java index 23b98ccbb..967af4184 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java @@ -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 */ diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java index 071ff9267..eb9af0c67 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsProducerProperties.java @@ -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) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsStateStoreProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsStateStoreProperties.java index a2d9f67bf..261ed7bbf 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsStateStoreProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsStateStoreProperties.java @@ -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) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java index 224eadfbe..f3f4807af 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java @@ -55,7 +55,7 @@ import org.springframework.util.MimeTypeUtils; * to be included in the configuration map with the key "contentType". For example, * * 
- * Map config = new HashMap<>();
+ * Map<String, Object> config = new HashMap<>();
  * config.put("valueClass", Foo.class);
  * config.put("contentType", "application/avro");
  *
@@ -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 type of the object to marshall + * * @author Soby Chacko * @since 2.1 */ @@ -110,12 +112,12 @@ public class CompositeNonNativeSerde implements Serde { } private static MimeType resolveMimeType(Map 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 implements Serde { /** * Custom {@link Deserializer} that uses the {@link CompositeMessageConverterFactory}. * - * @param Parameterized target type for deserialization + * @param parameterized target type for deserialization */ private static class CompositeNonNativeDeserializer implements Deserializer { @@ -158,7 +160,7 @@ public class CompositeNonNativeSerde implements Serde { 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 implements Serde { /** * Custom {@link Serializer} that uses the {@link CompositeMessageConverterFactory}. * - * @param Parameterized type for serialization + * @param parameterized type for serialization */ private static class CompositeNonNativeSerializer implements Serializer { @@ -194,9 +196,9 @@ public class CompositeNonNativeSerde implements Serde { Map 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 diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java index cac3f93fa..414c043f7 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java @@ -73,52 +73,7 @@ public class KafkaBinderHealthIndicator implements HealthIndicator { @Override public Health health() { ExecutorService exec = Executors.newSingleThreadExecutor(); - Future future = exec.submit(() -> { - try { - if (this.metadataConsumer == null) { - synchronized (KafkaBinderHealthIndicator.this) { - if (this.metadataConsumer == null) { - this.metadataConsumer = this.consumerFactory.createConsumer(); - } - } - } - synchronized (this.metadataConsumer) { - Set downMessages = new HashSet<>(); - final Map 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 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 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 downMessages = new HashSet<>(); + final Map 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 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(); + } + } + } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java index 6b892b35c..f6991d4a1 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java @@ -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;