From 3b1b29e86d072bccddc22a37f675f83f2bb21a05 Mon Sep 17 00:00:00 2001
From: Soby Chacko <schacko@pivotal.io>
Date: Wed, 25 Jan 2023 13:23:38 -0500
Subject: [PATCH] Health indicator fixes when topic is pattern (#2635)

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
---
 .../kafka/KafkaBinderHealthIndicator.java     | 15 ++++++++++++-
 .../kafka/KafkaBinderHealthIndicatorTest.java | 22 ++++++++++++++++++-
 2 files changed, 35 insertions(+), 2 deletions(-)

diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
index e35d15290..d80e16f05 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
@@ -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()) {
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java
index 247446425..e748ce1cd 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java
@@ -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);