diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index d165d7192..500fb88b0 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java @@ -81,9 +81,9 @@ public class KafkaTopicProvisioner implements // @checkstyle:on InitializingBean { - private static final int DEFAULT_OPERATION_TIMEOUT = 30; + private static final Log logger = LogFactory.getLog(KafkaTopicProvisioner.class); - private final Log logger = LogFactory.getLog(getClass()); + private static final int DEFAULT_OPERATION_TIMEOUT = 30; private final KafkaBinderConfigurationProperties configurationProperties; @@ -242,7 +242,7 @@ public class KafkaTopicProvisioner implements * @param bootProps the boot kafka properties. * @param binderProps the binder kafka properties. */ - private void normalalizeBootPropsWithBinder(Map adminProps, + public static void normalalizeBootPropsWithBinder(Map adminProps, KafkaProperties bootProps, KafkaBinderConfigurationProperties binderProps) { // First deal with the outlier String kafkaConnectionString = binderProps.getKafkaConnectionString(); @@ -263,8 +263,8 @@ public class KafkaTopicProvisioner implements } if (adminConfigNames.contains(key)) { Object replaced = adminProps.put(key, value); - if (replaced != null && this.logger.isDebugEnabled()) { - this.logger.debug("Overrode boot property: [" + key + "], from: [" + if (replaced != null && KafkaTopicProvisioner.logger.isDebugEnabled()) { + KafkaTopicProvisioner.logger.debug("Overrode boot property: [" + key + "], from: [" + replaced + "] to: [" + value + "]"); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java index 19e195984..bfafbabb8 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java @@ -16,11 +16,17 @@ package org.springframework.cloud.stream.binder.kafka.streams; +import java.time.Duration; import java.util.HashMap; import java.util.Map; import java.util.Set; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.ListTopicsResult; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.processor.TaskMetadata; @@ -29,6 +35,9 @@ import org.apache.kafka.streams.processor.ThreadMetadata; import org.springframework.boot.actuate.health.AbstractHealthIndicator; import org.springframework.boot.actuate.health.Health; import org.springframework.boot.actuate.health.Status; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.kafka.config.StreamsBuilderFactoryBean; /** @@ -39,26 +48,80 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean; */ public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator { - private final KafkaStreamsRegistry kafkaStreamsRegistry; + private final Log logger = LogFactory.getLog(getClass()); - KafkaStreamsBinderHealthIndicator(KafkaStreamsRegistry kafkaStreamsRegistry) { + private final KafkaStreamsRegistry kafkaStreamsRegistry; + private final KafkaStreamsBinderConfigurationProperties configurationProperties; + + private final Map adminClientProperties; + + private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; + + private static final ThreadLocal healthStatusThreadLocal = new ThreadLocal<>(); + + KafkaStreamsBinderHealthIndicator(KafkaStreamsRegistry kafkaStreamsRegistry, + KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties, + KafkaProperties kafkaProperties, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) { super("Kafka-streams health check failed"); + kafkaProperties.buildAdminProperties(); + this.configurationProperties = kafkaStreamsBinderConfigurationProperties; + this.adminClientProperties = kafkaProperties.buildAdminProperties(); + KafkaTopicProvisioner.normalalizeBootPropsWithBinder(this.adminClientProperties, kafkaProperties, + kafkaStreamsBinderConfigurationProperties); this.kafkaStreamsRegistry = kafkaStreamsRegistry; + this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; } @Override protected void doHealthCheck(Health.Builder builder) throws Exception { - boolean up = true; - for (KafkaStreams kStream : kafkaStreamsRegistry.getKafkaStreams()) { - up &= kStream.state().isRunning(); - builder.withDetails(buildDetails(kStream)); + AdminClient adminClient = null; + + try { + adminClient = AdminClient.create(this.adminClientProperties); + final Status status = healthStatusThreadLocal.get(); + //If one of the kafka streams binders (kstream, ktable, globalktable) was down before on the same request, + //retrieve that from the theadlocal storage where it was saved before. This is done in order to avoid + //the duration of the total health check since in the case of Kafka Streams each binder tries to do + //its own health check and since we already know that this is DOWN, simply pass that information along. + if (status != null && status.equals(Status.DOWN)) { + builder.withDetail("No topic information available", "Kafka broker is not reachable"); + builder.status(Status.DOWN); + } + else { + final ListTopicsResult listTopicsResult = adminClient.listTopics(); + listTopicsResult.listings().get(this.configurationProperties.getHealthTimeout(), TimeUnit.SECONDS); + + if (this.kafkaStreamsBindingInformationCatalogue.getStreamsBuilderFactoryBeans().isEmpty()) { + builder.withDetail("No Kafka Streams bindings have been established", "Kafka Streams binder did not detect any processors"); + builder.status(Status.UNKNOWN); + } + else { + boolean up = true; + for (KafkaStreams kStream : kafkaStreamsRegistry.getKafkaStreams()) { + up &= kStream.state().isRunning(); + builder.withDetails(buildDetails(kStream)); + } + builder.status(up ? Status.UP : Status.DOWN); + } + } + } + catch (Exception e) { + builder.withDetail("No topic information available", "Kafka broker is not reachable"); + builder.status(Status.DOWN); + //Store binder down status into a thread local storage. + healthStatusThreadLocal.set(Status.DOWN); + } + finally { + // Close admin client immmediately. + adminClient.close(Duration.ofSeconds(0)); } - builder.status(up ? Status.UP : Status.DOWN); } private Map buildDetails(KafkaStreams kafkaStreams) { final Map details = new HashMap<>(); final Map perAppdIdDetails = new HashMap<>(); + if (kafkaStreams.state().isRunning()) { for (ThreadMetadata metadata : kafkaStreams.localThreadsMetadata()) { perAppdIdDetails.put("threadName", metadata.threadName()); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java index 9668698c0..d5cd0f416 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java @@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams; import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -35,8 +37,11 @@ class KafkaStreamsBinderHealthIndicatorConfiguration { @Bean @ConditionalOnBean(KafkaStreamsRegistry.class) KafkaStreamsBinderHealthIndicator kafkaStreamsBinderHealthIndicator( - KafkaStreamsRegistry kafkaStreamsRegistry) { - return new KafkaStreamsBinderHealthIndicator(kafkaStreamsRegistry); + KafkaStreamsRegistry kafkaStreamsRegistry, KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties, + KafkaProperties kafkaProperties, KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) { + + return new KafkaStreamsBinderHealthIndicator(kafkaStreamsRegistry, kafkaStreamsBinderConfigurationProperties, + kafkaProperties, kafkaStreamsBindingInformationCatalogue); } } 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 6a048fe43..d44d18dc8 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 @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.kafka; +import java.time.Duration; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -112,8 +113,15 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe final Map topicsInUse = KafkaBinderHealthIndicator.this.binder .getTopicsInUse(); if (topicsInUse.isEmpty()) { - return Health.down().withDetail("No topic information available", - "Kafka broker is not reachable").build(); + try { + this.metadataConsumer.listTopics(Duration.ofSeconds(this.timeout)); + } + catch (Exception e) { + return Health.down().withDetail("No topic information available", + "Kafka broker is not reachable").build(); + } + return Health.unknown().withDetail("No bindings found", + "Kafka binder may not be bound to destinations on the broker").build(); } else { for (String topic : topicsInUse.keySet()) { diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java index ae663ec96..3837c0b81 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java @@ -159,7 +159,7 @@ public class KafkaBinderHealthIndicatorTest { @Test public void testIfNoTopicsRegisteredByTheBinderProvidesDownStatus() { Health health = indicator.health(); - assertThat(health.getStatus()).isEqualTo(Status.DOWN); + assertThat(health.getStatus()).isEqualTo(Status.UNKNOWN); } private List partitions(Node leader) {