Reactor Kafka Binder Health Indicator (#2755)

* Reactor Kafka Binder Health Indicator

 - Provide a new abstraction for general Kafka binder related HealthIndicators.
 - Refactor Kafka binder to use the new abstraction
 - Add HealthIndicator implementation for the ReactorKafkaBinder

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2752

* Addressing PR review

* Addressing PR review

* Addressing PR review

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/pom.xml

@@ -24,6 +24,11 @@
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-kafka</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-configuration-processor</artifactId>

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/AbstractKafkaBinderHealthIndicator.java

@@ -0,0 +1,205 @@
+/*
+ * Copyright 2023-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.common;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.PartitionInfo;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.boot.actuate.health.AbstractHealthIndicator;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.Status;
+import org.springframework.boot.actuate.health.StatusAggregator;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Base class that abstracts the common health indicator details for the various Kafka binder flavors.
+ *
+ * @author Soby Chacko
+ * @since 4.1.0
+ */
+public abstract class AbstractKafkaBinderHealthIndicator extends AbstractHealthIndicator implements DisposableBean {
+
+	private static final int DEFAULT_TIMEOUT = 60;
+
+	protected int timeout = DEFAULT_TIMEOUT;
+
+	private final ExecutorService executor;
+
+	protected Consumer<?, ?> metadataConsumer;
+
+	protected boolean considerDownWhenAnyPartitionHasNoLeader;
+
+	private final ConsumerFactory<?, ?> consumerFactory;
+
+	public AbstractKafkaBinderHealthIndicator(ConsumerFactory<?, ?> consumerFactory) {
+		this.consumerFactory = consumerFactory;
+		this.executor = createHealthBinderExecutorService();
+		Assert.notNull(this.executor, "The health indicator executor service must not be null");
+	}
+
+	protected abstract Map<String, TopicInformation> getTopicsInUse();
+
+	protected abstract Health buildBinderSpecificHealthDetails();
+
+	protected  abstract ExecutorService createHealthBinderExecutorService();
+
+	private void initMetadataConsumer() {
+		if (this.metadataConsumer == null) {
+			this.metadataConsumer = this.consumerFactory.createConsumer();
+		}
+	}
+
+	@Override
+	public void destroy() {
+		executor.shutdown();
+		if (this.metadataConsumer != null) {
+			this.metadataConsumer.close();
+		}
+	}
+
+	@Override
+	protected void doHealthCheck(Health.Builder builder) throws Exception {
+		Health topicsHealth = safelyBuildTopicsHealth();
+		Health listenerContainersHealth = buildBinderSpecificHealthDetails();
+		merge(topicsHealth, listenerContainersHealth, builder);
+	}
+
+	protected Health safelyBuildTopicsHealth() {
+		Future<Health> future = executor.submit(this::buildTopicsHealth);
+		try {
+			return future.get(this.timeout, TimeUnit.SECONDS);
+		}
+		catch (InterruptedException ex) {
+			Thread.currentThread().interrupt();
+			return Health.down()
+				.withDetail("Interrupted while waiting for partition information in",
+					this.timeout + " seconds")
+				.build();
+		}
+		catch (ExecutionException ex) {
+			return Health.down(ex).build();
+		}
+		catch (TimeoutException ex) {
+			return Health.down().withDetail("Failed to retrieve partition information in",
+				this.timeout + " seconds").build();
+		}
+	}
+
+	private Health buildTopicsHealth() {
+		try {
+			initMetadataConsumer();
+			Set<String> downMessages = new HashSet<>();
+			Set<String> checkedTopics = new HashSet<>();
+			final Map<String, TopicInformation> topicsInUse = 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();
+			}
+			else {
+				for (String topic : topicsInUse.keySet()) {
+					TopicInformation topicInformation = topicsInUse
+						.get(topic);
+					if (!topicInformation.isTopicPattern()) {
+						List<PartitionInfo> partitionInfos = this.metadataConsumer
+							.partitionsFor(topic);
+						for (PartitionInfo partitionInfo : partitionInfos) {
+							if (topicInformation.partitionInfos()
+								.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());
+							}
+						}
+						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()) {
+				return Health.up().withDetail("topicsInUse", checkedTopics).build();
+			}
+			else {
+				return Health.down()
+					.withDetail("Following partitions in use have no leaders: ",
+						downMessages.toString())
+					.build();
+			}
+		}
+		catch (Exception ex) {
+			return Health.down(ex).build();
+		}
+	}
+
+	private void merge(Health topicsHealth, Health listenerContainersHealth, Health.Builder builder) {
+		Status aggregatedStatus = StatusAggregator.getDefault()
+			.getAggregateStatus(topicsHealth.getStatus(), listenerContainersHealth.getStatus());
+		Map<String, Object> aggregatedDetails = new HashMap<>();
+		aggregatedDetails.putAll(topicsHealth.getDetails());
+		aggregatedDetails.putAll(listenerContainersHealth.getDetails());
+		builder.status(aggregatedStatus).withDetails(aggregatedDetails);
+	}
+
+	/**
+	 * Set the timeout in seconds to retrieve health information.
+	 *
+	 * @param timeout the timeout - default 60.
+	 */
+	public void setTimeout(int timeout) {
+		this.timeout = timeout;
+	}
+
+	public void setConsiderDownWhenAnyPartitionHasNoLeader(boolean considerDownWhenAnyPartitionHasNoLeader) {
+		this.considerDownWhenAnyPartitionHasNoLeader = considerDownWhenAnyPartitionHasNoLeader;
+	}
+}

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/TopicInformation.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2023-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.common;
+
+import java.util.Collection;
+
+import org.apache.kafka.common.PartitionInfo;
+
+/**
+ * Record to capture topic information for various binder related tasks.
+ *
+ * @param consumerGroup consumer group for the consumer
+ * @param partitionInfos collection of {@link PartitionInfo}
+ * @param isTopicPattern if the topic is specified as a pattern
+ *
+ * @author Soby Chacko (and previous authors before refactoring).
+ */
+public record TopicInformation(String consumerGroup, Collection<PartitionInfo> partitionInfos, boolean isTopicPattern) {
+
+	public boolean isConsumerTopic() {
+		return this.consumerGroup != null;
+	}
+}

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java

@@ -18,15 +18,21 @@ package org.springframework.cloud.stream.binder.reactorkafka;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.regex.Pattern;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.PartitionInfo;
 import reactor.core.publisher.Flux;
 import reactor.kafka.receiver.KafkaReceiver;
 import reactor.kafka.receiver.ReceiverOptions;
@@ -42,6 +48,7 @@ import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties;
@@ -60,6 +67,9 @@ import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.KafkaMessageHeaders;
 import org.springframework.kafka.support.converter.MessageConverter;
@@ -99,6 +109,10 @@ public class ReactorKafkaBinder
 
 	private SenderOptionsCustomizer<Object, Object> senderOptionsCustomizer = (name, opts) -> opts;
 
+	private final Map<String, TopicInformation> topicsInUse = new ConcurrentHashMap<>();
+
+	private final Map<String, MessageProducerSupport> messageProducers = new ConcurrentHashMap<>();
+
 	public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties,
 			KafkaTopicProvisioner provisioner) {
 
@@ -165,6 +179,20 @@ public class ReactorKafkaBinder
 			this.producerConfigCustomizer.configure(configs, producerProperties.getBindingName(),
 					destination.getName());
 		}
+		Map<String, Object> props = BindingUtils.createProducerConfigs(producerProperties,
+			this.configurationProperties);
+		DefaultKafkaProducerFactory<byte[], byte[]> producerFactory = new DefaultKafkaProducerFactory<>(props);
+		Collection<PartitionInfo> partitions = provisioningProvider.getPartitionsForTopic(
+			producerProperties.getPartitionCount(), false, () -> {
+				Producer<byte[], byte[]> producer = producerFactory.createProducer();
+				List<PartitionInfo> partitionsFor = producer
+					.partitionsFor(destination.getName());
+				producer.close();
+				return partitionsFor;
+			}, destination.getName());
+
+		this.topicsInUse.put(destination.getName(),
+			new TopicInformation(null, partitions, false));
 
 		SenderOptions<Object, Object> opts = this.senderOptionsCustomizer.apply(producerProperties.getBindingName(),
 				SenderOptions.create(configs));
@@ -179,13 +207,56 @@ public class ReactorKafkaBinder
 		return new ReactorMessageHandler(opts, converter, destination.getName(), resultChannel);
 	}
 
+	// TODO: Refactor to provide in a common area since KafkaMessageChannelBinder also provides this.
+	public void processTopic(final String group, final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
+							final ConsumerFactory<?, ?> consumerFactory, int partitionCount,
+							boolean usingPatterns, boolean groupManagement, String topic) {
+		Collection<PartitionInfo> listenedPartitions;
+		Collection<PartitionInfo> allPartitions = usingPatterns ? Collections.emptyList()
+			: getPartitionInfo(topic, extendedConsumerProperties, consumerFactory,
+			partitionCount);
+
+		if (groupManagement || extendedConsumerProperties.getInstanceCount() == 1) {
+			listenedPartitions = allPartitions;
+		}
+		else {
+			listenedPartitions = new ArrayList<>();
+			for (PartitionInfo partition : allPartitions) {
+				// divide partitions across modules
+				if ((partition.partition() % extendedConsumerProperties
+					.getInstanceCount()) == extendedConsumerProperties
+					.getInstanceIndex()) {
+					listenedPartitions.add(partition);
+				}
+			}
+		}
+		this.topicsInUse.put(topic,
+			new TopicInformation(group, listenedPartitions, usingPatterns));
+	}
+
+	private Collection<PartitionInfo> getPartitionInfo(String topic,
+													final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
+													final ConsumerFactory<?, ?> consumerFactory, int partitionCount) {
+		return provisioningProvider.getPartitionsForTopic(partitionCount,
+			extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(),
+			() -> {
+				try (Consumer<?, ?> consumer = consumerFactory.createConsumer()) {
+					return consumer.partitionsFor(topic);
+				}
+			}, topic);
+	}
+
+	Map<String, TopicInformation> getTopicsInUse() {
+		return this.topicsInUse;
+	}
+
 
 	@Override
 	protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group,
-			ExtendedConsumerProperties<KafkaConsumerProperties> properties) throws Exception {
+			ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
 
 		boolean anonymous = !StringUtils.hasText(group);
-		String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group;
+		String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID() : group;
 		Map<String, Object> configs = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties,
 				this.configurationProperties);
 
@@ -202,7 +273,7 @@ public class ReactorKafkaBinder
 		 *  it is still required by the provisioner, however.
 		 */
 		List<String> destList = Arrays.stream(StringUtils.commaDelimitedListToStringArray(destinations))
-				.map(dest -> dest.trim())
+				.map(String::trim)
 				.toList();
 		ReceiverOptions<Object, Object> opts = ReceiverOptions.create(configs)
 			.addAssignListener(parts -> logger.info("Assigned: " + parts));
@@ -215,6 +286,15 @@ public class ReactorKafkaBinder
 		opts = this.receiverOptionsCustomizer.apply(properties.getBindingName(), opts);
 		ReceiverOptions<Object, Object> finalOpts = opts;
 
+		Map<String, Object> props = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties,
+			this.configurationProperties);
+
+		DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(props);
+		int partitionCount = properties.getInstanceCount() * properties.getConcurrency();
+		boolean groupManagement = properties.getExtension().isAutoRebalanceEnabled();
+		processTopic(consumerGroup, properties, factory, partitionCount, properties.getExtension().isDestinationIsPattern(),
+			groupManagement, destination.getName());
+
 		class ReactorMessageProducer extends MessageProducerSupport {
 
 			private final List<KafkaReceiver<Object, Object>> receivers = new ArrayList<>();
@@ -286,7 +366,13 @@ public class ReactorKafkaBinder
 			}
 
 		}
-		return new ReactorMessageProducer();
+		ReactorMessageProducer reactorMessageProducer = new ReactorMessageProducer();
+		this.messageProducers.put(consumerGroup, reactorMessageProducer);
+		return reactorMessageProducer;
+	}
+
+	public Map<String, MessageProducerSupport> getMessageProducers() {
+		return this.messageProducers;
 	}
 
 	@Override
@@ -384,5 +470,4 @@ public class ReactorKafkaBinder
 		}
 
 	}
-
 }

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicator.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2023-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.reactorkafka;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.Status;
+import org.springframework.cloud.stream.binder.kafka.common.AbstractKafkaBinderHealthIndicator;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+
+/**
+ * {@link org.springframework.boot.actuate.health.HealthIndicator} for Reactor Kafka Binder.
+ *
+ * @author Soby Chacko
+ */
+public class ReactorKafkaBinderHealthIndicator extends AbstractKafkaBinderHealthIndicator {
+
+	private final ReactorKafkaBinder binder;
+
+	public ReactorKafkaBinderHealthIndicator(ReactorKafkaBinder binder, ConsumerFactory<?, ?> consumerFactory) {
+		super(consumerFactory);
+		this.binder = binder;
+	}
+
+	@Override
+	protected ExecutorService createHealthBinderExecutorService() {
+		return Executors.newSingleThreadExecutor(
+			new CustomizableThreadFactory("reactor-kafka-binder-health-"));
+	}
+
+	@Override
+	protected Map<String, TopicInformation> getTopicsInUse() {
+		return this.binder.getTopicsInUse();
+	}
+
+	@Override
+	protected Health buildBinderSpecificHealthDetails() {
+		Map<String, MessageProducerSupport> messageProducerSupportInfo = binder.getMessageProducers();
+		if (messageProducerSupportInfo.isEmpty()) {
+			return Health.unknown().build();
+		}
+
+		Status status = Status.UP;
+		List<Map<String, Object>> messageProducers = new ArrayList<>();
+
+		Map<String, Object> messageProducerDetails = new HashMap<>();
+		for (String groupId : messageProducerSupportInfo.keySet()) {
+			MessageProducerSupport messageProducerSupport = messageProducerSupportInfo.get(groupId);
+			boolean isRunning = messageProducerSupport.isRunning();
+			boolean isOk = messageProducerSupport.isActive();
+			if (!isOk) {
+				status = Status.DOWN;
+			}
+			messageProducerDetails.put("isRunning", isRunning);
+			messageProducerDetails.put("isStoppedAbnormally", !isRunning && !isOk);
+			//messageProducerDetails.put("isPaused", messageProducerSupport.isPaused());
+			messageProducerDetails.put("messageProducerId", messageProducerSupport.getApplicationContextId());
+			messageProducerDetails.put("groupId", groupId);
+		}
+		messageProducers.add(messageProducerDetails);
+		return Health.status(status)
+			.withDetail("messageProducers", messageProducers)
+			.build();
+	}
+}

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorTests.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2022-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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.reactorkafka;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.Status;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+
+import static java.util.Collections.singleton;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class ReactorKafkaBinderHealthIndicatorTests {
+
+	private static final String TEST_TOPIC = "test";
+
+	private ReactorKafkaBinderHealthIndicator indicator;
+
+	@Mock
+	private DefaultKafkaConsumerFactory<?, ?> consumerFactory;
+
+	@Mock
+	private KafkaConsumer consumer;
+
+	@Mock
+	MessageProducerSupport messageProducerSupport1;
+
+	@Mock
+	private ReactorKafkaBinder binder;
+
+	private final Map<String, TopicInformation> topicsInUse = new HashMap<>();
+
+	@BeforeEach
+	public void setup() {
+		MockitoAnnotations.openMocks(this);
+		org.mockito.BDDMockito.given(consumerFactory.createConsumer())
+			.willReturn((consumer));
+		org.mockito.BDDMockito.given(binder.getTopicsInUse()).willReturn(topicsInUse);
+		this.indicator = new ReactorKafkaBinderHealthIndicator(binder, consumerFactory);
+		this.indicator.setTimeout(10);
+	}
+
+	@Test
+	void reactorKafkaBinderIsUp() {
+		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
+			"group1-healthIndicator", partitions, false));
+		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
+			.willReturn(partitions);
+		org.mockito.BDDMockito.given(binder.getMessageProducers())
+			.willReturn(Map.of("group1-healthIndicator", messageProducerSupport1));
+		org.mockito.BDDMockito.given(messageProducerSupport1.isRunning()).willReturn(true);
+		org.mockito.BDDMockito.given(messageProducerSupport1.isActive()).willReturn(true);
+		Health health = indicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.UP);
+		assertThat(health.getDetails()).containsEntry("topicsInUse", singleton(TEST_TOPIC));
+		assertThat(health.getDetails()).hasEntrySatisfying("messageProducers", value ->
+			assertThat((ArrayList<?>) value).hasSize(1));
+	}
+
+	private List<PartitionInfo> partitions(Node leader) {
+		List<PartitionInfo> partitions = new ArrayList<>();
+		partitions.add(new PartitionInfo(TEST_TOPIC, 0, leader, null, null));
+		return partitions;
+	}
+
+}

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

@@ -16,28 +16,17 @@
 
 package org.springframework.cloud.stream.binder.kafka;
 
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.PartitionInfo;
-
-import org.springframework.beans.factory.DisposableBean;
-import org.springframework.boot.actuate.health.AbstractHealthIndicator;
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.boot.actuate.health.Status;
-import org.springframework.boot.actuate.health.StatusAggregator;
+import org.springframework.cloud.stream.binder.kafka.common.AbstractKafkaBinderHealthIndicator;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
@@ -55,154 +44,30 @@ import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
  * @author Chukwubuikem Ume-Ugwa
  * @author Taras Danylchuk
  */
-public class KafkaBinderHealthIndicator extends AbstractHealthIndicator implements DisposableBean {
-
-	private static final int DEFAULT_TIMEOUT = 60;
-
-	private final ExecutorService executor = Executors.newSingleThreadExecutor(
-		new CustomizableThreadFactory("kafka-binder-health-"));
+public class KafkaBinderHealthIndicator extends AbstractKafkaBinderHealthIndicator {
 
 	private final KafkaMessageChannelBinder binder;
 
-	private final ConsumerFactory<?, ?> consumerFactory;
-
-	private int timeout = DEFAULT_TIMEOUT;
-
-	private Consumer<?, ?> metadataConsumer;
-
-	private boolean considerDownWhenAnyPartitionHasNoLeader;
 
 	public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder,
 									ConsumerFactory<?, ?> consumerFactory) {
+		super(consumerFactory);
 		this.binder = binder;
-		this.consumerFactory = consumerFactory;
-	}
-
-	/**
-	 * Set the timeout in seconds to retrieve health information.
-	 * @param timeout the timeout - default 60.
-	 */
-	public void setTimeout(int timeout) {
-		this.timeout = timeout;
-	}
-
-	public void setConsiderDownWhenAnyPartitionHasNoLeader(boolean considerDownWhenAnyPartitionHasNoLeader) {
-		this.considerDownWhenAnyPartitionHasNoLeader = considerDownWhenAnyPartitionHasNoLeader;
 	}
 
 	@Override
-	protected void doHealthCheck(Health.Builder builder) {
-		Health topicsHealth = safelyBuildTopicsHealth();
-		Health listenerContainersHealth = buildListenerContainersHealth();
-		merge(topicsHealth, listenerContainersHealth, builder);
+	protected ExecutorService createHealthBinderExecutorService() {
+		return Executors.newSingleThreadExecutor(
+			new CustomizableThreadFactory("kafka-binder-health-"));
 	}
 
-	private void merge(Health topicsHealth, Health listenerContainersHealth, Health.Builder builder) {
-		Status aggregatedStatus = StatusAggregator.getDefault()
-						.getAggregateStatus(topicsHealth.getStatus(), listenerContainersHealth.getStatus());
-		Map<String, Object> aggregatedDetails = new HashMap<>();
-		aggregatedDetails.putAll(topicsHealth.getDetails());
-		aggregatedDetails.putAll(listenerContainersHealth.getDetails());
-		builder.status(aggregatedStatus).withDetails(aggregatedDetails);
+	@Override
+	protected Map<String, TopicInformation> getTopicsInUse() {
+		return this.binder.getTopicsInUse();
 	}
 
-	private Health safelyBuildTopicsHealth() {
-		Future<Health> future = executor.submit(this::buildTopicsHealth);
-		try {
-			return future.get(this.timeout, TimeUnit.SECONDS);
-		}
-		catch (InterruptedException ex) {
-			Thread.currentThread().interrupt();
-			return Health.down()
-					.withDetail("Interrupted while waiting for partition information in",
-							this.timeout + " seconds")
-					.build();
-		}
-		catch (ExecutionException ex) {
-			return Health.down(ex).build();
-		}
-		catch (TimeoutException ex) {
-			return Health.down().withDetail("Failed to retrieve partition information in",
-					this.timeout + " seconds").build();
-		}
-	}
-
-	private void initMetadataConsumer() {
-		if (this.metadataConsumer == null) {
-			this.metadataConsumer = this.consumerFactory.createConsumer();
-		}
-	}
-
-	private Health buildTopicsHealth() {
-		try {
-			initMetadataConsumer();
-			Set<String> downMessages = new HashSet<>();
-			Set<String> checkedTopics = 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();
-			}
-			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());
-							}
-						}
-						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()) {
-				return Health.up().withDetail("topicsInUse", checkedTopics).build();
-			}
-			else {
-				return Health.down()
-						.withDetail("Following partitions in use have no leaders: ",
-								downMessages.toString())
-						.build();
-			}
-		}
-		catch (Exception ex) {
-			return Health.down(ex).build();
-		}
-	}
-
-	private Health buildListenerContainersHealth() {
+	@Override
+	protected Health buildBinderSpecificHealthDetails() {
 		List<AbstractMessageListenerContainer<?, ?>> listenerContainers = binder.getKafkaMessageListenerContainers();
 		if (listenerContainers.isEmpty()) {
 			return Health.unknown().build();
@@ -230,10 +95,4 @@ public class KafkaBinderHealthIndicator extends AbstractHealthIndicator implemen
 				.withDetail("listenerContainers", containersDetails)
 				.build();
 	}
-
-	@Override
-	public void destroy() {
-		executor.shutdown();
-	}
-
 }

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java

@@ -46,6 +46,7 @@ import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 
 import org.springframework.cloud.stream.binder.BindingCreatedEvent;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 import org.springframework.context.ApplicationListener;
 import org.springframework.kafka.core.ConsumerFactory;
@@ -131,7 +132,7 @@ public class KafkaBinderMetrics
 
 		this.scheduler = Executors.newScheduledThreadPool(this.binder.getTopicsInUse().size());
 
-		for (Map.Entry<String, KafkaMessageChannelBinder.TopicInformation> topicInfo : this.binder
+		for (Map.Entry<String, TopicInformation> topicInfo : this.binder
 				.getTopicsInUse().entrySet()) {
 
 			if (!topicInfo.getValue().isConsumerTopic()) {
@@ -139,7 +140,7 @@ public class KafkaBinderMetrics
 			}
 
 			String topic = topicInfo.getKey();
-			String group = topicInfo.getValue().getConsumerGroup();
+			String group = topicInfo.getValue().consumerGroup();
 
 			ToDoubleFunction<KafkaBinderMetrics> offsetComputation = computeOffsetComputationFunction(topic, group);
 			final Gauge register = Gauge.builder(OFFSET_LAG_METRIC_NAME, this, offsetComputation)

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -65,6 +65,7 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.HeaderMode;
 import org.springframework.cloud.stream.binder.MessageValues;
 import org.springframework.cloud.stream.binder.PartitionHandler;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
 import org.springframework.cloud.stream.binder.kafka.config.ClientFactoryCustomizer;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
@@ -1595,42 +1596,6 @@ public class KafkaMessageChannelBinder extends
 		}
 	}
 
-	/**
-	 * Inner class to capture topic details.
-	 */
-	static class TopicInformation {
-
-		private final String consumerGroup;
-
-		private final Collection<PartitionInfo> partitionInfos;
-
-		private final boolean isTopicPattern;
-
-		TopicInformation(String consumerGroup, Collection<PartitionInfo> partitionInfos,
-				boolean isTopicPattern) {
-			this.consumerGroup = consumerGroup;
-			this.partitionInfos = partitionInfos;
-			this.isTopicPattern = isTopicPattern;
-		}
-
-		String getConsumerGroup() {
-			return this.consumerGroup;
-		}
-
-		boolean isConsumerTopic() {
-			return this.consumerGroup != null;
-		}
-
-		boolean isTopicPattern() {
-			return this.isTopicPattern;
-		}
-
-		Collection<PartitionInfo> getPartitionInfos() {
-			return this.partitionInfos;
-		}
-
-	}
-
 	/**
 	 * Helper class to send to DLQ.
 	 *

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

@@ -38,6 +38,7 @@ import org.mockito.MockitoAnnotations;
 
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.boot.actuate.health.Status;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 
@@ -77,7 +78,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Mock
 	private KafkaMessageChannelBinder binder;
 
-	private final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = new HashMap<>();
+	private final Map<String, TopicInformation> topicsInUse = new HashMap<>();
 
 	@BeforeEach
 	public void setup() {
@@ -92,7 +93,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Test
 	void kafkaBinderIsUpWithNoConsumers() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
-		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group1-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willReturn(partitions);
@@ -107,7 +108,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Test
 	void kafkaBinderIsUp() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
-		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group1-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willReturn(partitions);
@@ -126,7 +127,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Test
 	void kafkaBinderIsDownWhenOneOfConsumersIsNotRunning() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
-		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group1-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willReturn(partitions);
@@ -145,7 +146,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Test
 	void kafkaBinderIsDownWhenOneOfContainersWasStoppedAbnormally() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
-		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group1-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willReturn(partitions);
@@ -173,7 +174,7 @@ public class KafkaBinderHealthIndicatorTest {
 
 	@Test
 	void kafkaBinderIsUpWithRegexTopic() {
-		topicsInUse.put(REGEX_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(REGEX_TOPIC, new TopicInformation(
 				"regex-healthIndicator", null, true));
 		Health health = indicator.health();
 		// verify no consumer interaction for retrieving partitions
@@ -185,7 +186,7 @@ public class KafkaBinderHealthIndicatorTest {
 
 	@Test
 	void downWhenListTopicsThrowExceptionWithRegexTopic() {
-		topicsInUse.put(REGEX_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(REGEX_TOPIC, new TopicInformation(
 			"regex-healthIndicator", null, true));
 		org.mockito.BDDMockito.given(consumer.listTopics(any(Duration.class)))
 			.willThrow(new IllegalStateException());
@@ -204,7 +205,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Test
 	void kafkaBinderIsDown() {
 		final List<PartitionInfo> partitions = partitions(null);
-		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group2-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willReturn(partitions);
@@ -217,7 +218,7 @@ public class KafkaBinderHealthIndicatorTest {
 		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(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group2-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willReturn(partitions);
@@ -231,7 +232,7 @@ public class KafkaBinderHealthIndicatorTest {
 		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(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group2-healthIndicator", partitions(node), false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willReturn(partitions);
@@ -243,7 +244,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Timeout(5)
 	void kafkaBinderDoesNotAnswer() {
 		final List<PartitionInfo> partitions = partitions(new Node(-1, null, 0));
-		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group3-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willAnswer(invocation -> {
@@ -259,7 +260,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Test
 	void createsConsumerOnceWhenInvokedMultipleTimes() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
-		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"group4-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
 				.willReturn(partitions);
@@ -274,7 +275,7 @@ public class KafkaBinderHealthIndicatorTest {
 	@Test
 	void consumerCreationFailsFirstTime() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
-		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+		topicsInUse.put(TEST_TOPIC, new TopicInformation(
 				"foo-healthIndicator", partitions, false));
 
 		org.mockito.BDDMockito.given(consumerFactory.createConsumer())

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java

@@ -42,7 +42,7 @@ import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 
-import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.TopicInformation;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -94,7 +94,7 @@ import org.springframework.cloud.stream.binder.PollableSource;
 import org.springframework.cloud.stream.binder.RequeueCurrentMessageException;
 import org.springframework.cloud.stream.binder.Spy;
 import org.springframework.cloud.stream.binder.TestUtils;
-import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.TopicInformation;
+import org.springframework.cloud.stream.binder.kafka.common.TopicInformation;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
@@ -716,7 +716,7 @@ public class KafkaBinderTests extends
 		assertThat(topicsInUse.keySet()).contains("foo.bar");
 		TopicInformation topic = topicsInUse.get("foo.bar");
 		assertThat(topic.isConsumerTopic()).isTrue();
-		assertThat(topic.getConsumerGroup()).isEqualTo("testSendAndReceive");
+		assertThat(topic.consumerGroup()).isEqualTo("testSendAndReceive");
 
 		assertThat(KafkaTestUtils.getPropertyValue(consumerBinding, "lifecycle.recordListener.messageConverter"))
 				.isSameAs(mmc);
@@ -2305,7 +2305,7 @@ public class KafkaBinderTests extends
 		assertThat(topicsInUse.keySet()).contains("defaultGroup.0");
 		TopicInformation topic = topicsInUse.get("defaultGroup.0");
 		assertThat(topic.isConsumerTopic()).isTrue();
-		assertThat(topic.getConsumerGroup()).startsWith("anonymous");
+		assertThat(topic.consumerGroup()).startsWith("anonymous");
 
 		producerBinding.unbind();
 		binding1.unbind();