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;