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

Health indicator fixes when topic is pattern (#2635)

Browse Source 
When destination is pattern, even when the cluster is down,
health indicator is erroneously reporting the status as UP.
Addressing this issue by calling a simple listTopics on the
consumer when the destination is pattern, and if it throws an
exception, report the status as DOWN. If the call succeeds, we
assume that the cluster is accessible.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2628
This commit is contained in:
Soby Chacko
2023-01-25 13:23:38 -05:00
committed by GitHub
parent 7bff24af68
commit 3b1b29e86d
2 changed files with 35 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -171,6 +171,19 @@ public class KafkaBinderHealthIndicator implements KafkaBinderHealth, Disposable
}
checkedTopics.add(topic);
}
else {
try {
// Since destination is a pattern, all we are doing is just to make sure that
// we can connect to the cluster and query the topics.
this.metadataConsumer.listTopics(Duration.ofSeconds(this.timeout));
}
catch (Exception ex) {
return Health.down()
.withDetail("Cluster not connected",
"Destination provided is a pattern, but cannot connect to the cluster for any verification")
.build();
}
}
}
}
if (downMessages.isEmpty()) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2022 the original author or authors.
* Copyright 2017-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.kafka;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -42,6 +43,7 @@ import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import static java.util.Collections.singleton;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
/**
* @author Barry Commins
@@ -181,6 +183,24 @@ public class KafkaBinderHealthIndicatorTest {
assertThat(health.getStatus()).isEqualTo(Status.UP);
}
@Test
void downWhenListTopicsThrowExceptionWithRegexTopic() {
topicsInUse.put(REGEX_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
"regex-healthIndicator", null, true));
org.mockito.BDDMockito.given(consumer.listTopics(any(Duration.class)))
.willThrow(new IllegalStateException());
Health health = indicator.health();
// Ensuring the normal health check returns with status "up"
assertThat(health.getStatus()).isEqualTo(Status.DOWN);
Map<String, Object> details = health.getDetails();
assertThat(details.containsKey("Cluster not connected")).isTrue();
assertThat(details
.containsValue("Destination provided is a pattern, but cannot connect to the cluster for any verification")).isTrue();
}
@Test
void kafkaBinderIsDown() {
final List<PartitionInfo> partitions = partitions(null);