From 1e9aa60c4e2dde259470077852adcffa430c8050 Mon Sep 17 00:00:00 2001 From: cleverchuk Date: Fri, 21 Aug 2020 22:56:50 -0400 Subject: [PATCH] Fix Health indicator/partition leader issues * Fix health indicator to properly indicate partition failure * Add new flag to control binder health indicator behavior * Regardless of the consumer that is reading from a partition, if the binder detects that a partition for the topic is without a leader, mark the binder health as DOWN (if the flag is set to true). * Remove synchronize block since only one thread executes the block * Add Docs for the new binder flag * Fix checkstyle issues --- docs/src/main/asciidoc/overview.adoc | 5 ++ .../KafkaBinderConfigurationProperties.java | 11 +++ .../kafka/KafkaBinderHealthIndicator.java | 85 ++++++++++--------- ...fkaBinderHealthIndicatorConfiguration.java | 2 + .../kafka/KafkaBinderHealthIndicatorTest.java | 30 ++++++- 5 files changed, 94 insertions(+), 39 deletions(-) diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 9666ad9ad..bc58e7daf 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -144,6 +144,11 @@ If this custom `BinderHeaderMapper` bean is not made available to the binder usi + Default: none. +spring.cloud.stream.kafka.binder.considerDownWhenAnyPartitionHasNoLeader:: +Flag to set the binder health as `down`, when any partitions on the topic, regardless of the consumer that is receiving data from it, is found without a leader. ++ +Default: `false`. + [[kafka-consumer-properties]] ==== Kafka Consumer Properties diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index e64faa242..199a085fd 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -53,6 +53,7 @@ import org.springframework.util.StringUtils; * @author Rafal Zukowski * @author Aldo Sinanaj * @author Lukasz Kaminski + * @author Chukwubuikem Ume-Ugwa */ @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder") public class KafkaBinderConfigurationProperties { @@ -90,6 +91,8 @@ public class KafkaBinderConfigurationProperties { private boolean autoAddPartitions; + private boolean considerDownWhenAnyPartitionHasNoLeader; + private String requiredAcks = "1"; private short replicationFactor = -1; @@ -363,6 +366,14 @@ public class KafkaBinderConfigurationProperties { this.authorizationExceptionRetryInterval = authorizationExceptionRetryInterval; } + public boolean isConsiderDownWhenAnyPartitionHasNoLeader() { + return this.considerDownWhenAnyPartitionHasNoLeader; + } + + public void setConsiderDownWhenAnyPartitionHasNoLeader(boolean considerDownWhenAnyPartitionHasNoLeader) { + this.considerDownWhenAnyPartitionHasNoLeader = considerDownWhenAnyPartitionHasNoLeader; + } + /** * Domain class that models transaction capabilities in Kafka. */ 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 b967edf7a..2a9807d09 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 @@ -47,6 +47,7 @@ import org.springframework.scheduling.concurrent.CustomizableThreadFactory; * @author Laur Aliste * @author Soby Chacko * @author Vladislav Fefelov + * @author Chukwubuikem Ume-Ugwa */ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBean { @@ -63,6 +64,8 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe private Consumer metadataConsumer; + private boolean considerDownWhenAnyPartitionHasNoLeader; + public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder, ConsumerFactory consumerFactory) { this.binder = binder; @@ -77,6 +80,10 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe this.timeout = timeout; } + public void setConsiderDownWhenAnyPartitionHasNoLeader(boolean considerDownWhenAnyPartitionHasNoLeader) { + this.considerDownWhenAnyPartitionHasNoLeader = considerDownWhenAnyPartitionHasNoLeader; + } + @Override public Health health() { Future future = executor.submit(this::buildHealthStatus); @@ -99,57 +106,59 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe } } - private synchronized Consumer initMetadataConsumer() { + private void initMetadataConsumer() { if (this.metadataConsumer == null) { this.metadataConsumer = this.consumerFactory.createConsumer(); } - return this.metadataConsumer; } private Health buildHealthStatus() { try { initMetadataConsumer(); - synchronized (this.metadataConsumer) { - Set downMessages = new HashSet<>(); - final Map topicsInUse = KafkaBinderHealthIndicator.this.binder - .getTopicsInUse(); - if (topicsInUse.isEmpty()) { - 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(); + Set downMessages = new HashSet<>(); + final Map topicsInUse = KafkaBinderHealthIndicator.this.binder + .getTopicsInUse(); + if (topicsInUse.isEmpty()) { + try { + this.metadataConsumer.listTopics(Duration.ofSeconds(this.timeout)); } - 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()); - } + 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()) { + 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() == null || + (partitionInfo.leader() != null && partitionInfo.leader().id() == -1)) { + downMessages.add(partitionInfo.toString()); + } + else if (this.considerDownWhenAnyPartitionHasNoLeader && + partitionInfo.leader() == null || (partitionInfo.leader() != null && 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(); - } + } + 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) { diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java index c9d3f6337..c991c6caa 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java @@ -37,6 +37,7 @@ import org.springframework.util.ObjectUtils; * Configuration class for Kafka binder health indicator beans. * * @author Oleg Zhurakousky + * @author Chukwubuikem Ume-Ugwa */ @Configuration @@ -66,6 +67,7 @@ class KafkaBinderHealthIndicatorConfiguration { KafkaBinderHealthIndicator indicator = new KafkaBinderHealthIndicator( kafkaMessageChannelBinder, consumerFactory); indicator.setTimeout(configurationProperties.getHealthTimeout()); + indicator.setConsiderDownWhenAnyPartitionHasNoLeader(configurationProperties.isConsiderDownWhenAnyPartitionHasNoLeader()); return indicator; } 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 3837c0b81..7ad32169d 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 @@ -42,6 +42,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Gary Russell * @author Laur Aliste * @author Soby Chacko + * @author Chukwubuikem Ume-Ugwa */ public class KafkaBinderHealthIndicatorTest { @@ -97,7 +98,7 @@ public class KafkaBinderHealthIndicatorTest { @Test public void kafkaBinderIsDown() { - final List partitions = partitions(new Node(-1, null, 0)); + final List partitions = partitions(null); topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation( "group2-healthIndicator", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)) @@ -106,6 +107,33 @@ public class KafkaBinderHealthIndicatorTest { assertThat(health.getStatus()).isEqualTo(Status.DOWN); } + @Test + public void kafkaBinderIsDownWhenConsiderDownWhenAnyPartitionHasNoLeaderIsTrue() { + final List partitions = partitions(new Node(0, null, 0)); + partitions.add(new PartitionInfo(TEST_TOPIC, 0, null, null, null)); + indicator.setConsiderDownWhenAnyPartitionHasNoLeader(true); + topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation( + "group2-healthIndicator", partitions, false)); + org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)) + .willReturn(partitions); + Health health = indicator.health(); + assertThat(health.getStatus()).isEqualTo(Status.DOWN); + } + + @Test + public void kafkaBinderIsUpWhenConsiderDownWhenAnyPartitionHasNoLeaderIsFalse() { + Node node = new Node(0, null, 0); + final List partitions = partitions(node); + partitions.add(new PartitionInfo(TEST_TOPIC, 0, null, null, null)); + indicator.setConsiderDownWhenAnyPartitionHasNoLeader(false); + topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation( + "group2-healthIndicator", partitions(node), false)); + org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)) + .willReturn(partitions); + Health health = indicator.health(); + assertThat(health.getStatus()).isEqualTo(Status.UP); + } + @Test(timeout = 5000) public void kafkaBinderDoesNotAnswer() { final List partitions = partitions(new Node(-1, null, 0));