SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix Health indicator/partition leader issues

Browse Source 
* 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
This commit is contained in:
cleverchuk
2020-08-21 22:56:50 -04:00
committed by Soby Chacko
parent 4161f875ed
commit 1e9aa60c4e
5 changed files with 94 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
docs/src/main/asciidoc/overview.adoc
View File

@@ -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

11
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java
View File

@@ -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.
*/

85
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
View File

@@ -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<Health> 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<String> downMessages = new HashSet<>();
final Map<String, KafkaMessageChannelBinder.TopicInformation> 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<String> downMessages = new HashSet<>();
final Map<String, KafkaMessageChannelBinder.TopicInformation> 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<PartitionInfo> 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<PartitionInfo> 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) {

2
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java
View File

@@ -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;
}

30
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java
View File

@@ -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<PartitionInfo> partitions = partitions(new Node(-1, null, 0));
final List<PartitionInfo> 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<PartitionInfo> 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<PartitionInfo> 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<PartitionInfo> partitions = partitions(new Node(-1, null, 0));