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);