From b8eb41cb87733c861cabb6b2318f6d10325d4ade Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 21 Oct 2019 21:59:06 -0400 Subject: [PATCH] Kafka/Streams binder health indicator improvements When both binders are present, there were ambiguities in the way the binders were reporting health status. If one binder does not have any bindings, the total health status was reported as down. Fixing these ambiguiltes as below. If both binders have bindings present and Kafka broker is reachable, report the status as UP and the associated details. If one of the binder does not have bindings, but Kafka broker can be reached, then that particular binder's status will be marked as UNKNOWN and the overall status is reported as UP. If Kafka broker is down, then both binders are reported as DOWN and the overall status is marked as DOWN. Resolves #552 --- .../provisioning/KafkaTopicProvisioner.java | 10 +-- .../KafkaStreamsBinderHealthIndicator.java | 77 +++++++++++++++++-- ...amsBinderHealthIndicatorConfiguration.java | 9 ++- .../kafka/KafkaBinderHealthIndicator.java | 12 ++- .../kafka/KafkaBinderHealthIndicatorTest.java | 2 +- 5 files changed, 93 insertions(+), 17 deletions(-) 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) {