diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/pom.xml b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/pom.xml index bb0b3a552..6091fcbd4 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/pom.xml +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/pom.xml @@ -24,6 +24,11 @@ org.springframework.integration spring-integration-kafka + + org.springframework.boot + spring-boot-starter-actuator + true + org.springframework.boot spring-boot-configuration-processor diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/AbstractKafkaBinderHealthIndicator.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/AbstractKafkaBinderHealthIndicator.java new file mode 100644 index 000000000..e4cf87d8b --- /dev/null +++ b/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 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 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 downMessages = new HashSet<>(); + Set checkedTopics = new HashSet<>(); + final Map 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 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 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; + } +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/TopicInformation.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/TopicInformation.java new file mode 100644 index 000000000..bac2b6901 --- /dev/null +++ b/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 partitionInfos, boolean isTopicPattern) { + + public boolean isConsumerTopic() { + return this.consumerGroup != null; + } +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java index 371ac967f..91ec7fc41 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ b/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 senderOptionsCustomizer = (name, opts) -> opts; + private final Map topicsInUse = new ConcurrentHashMap<>(); + + private final Map 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 props = BindingUtils.createProducerConfigs(producerProperties, + this.configurationProperties); + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(props); + Collection partitions = provisioningProvider.getPartitionsForTopic( + producerProperties.getPartitionCount(), false, () -> { + Producer producer = producerFactory.createProducer(); + List partitionsFor = producer + .partitionsFor(destination.getName()); + producer.close(); + return partitionsFor; + }, destination.getName()); + + this.topicsInUse.put(destination.getName(), + new TopicInformation(null, partitions, false)); SenderOptions 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 extendedConsumerProperties, + final ConsumerFactory consumerFactory, int partitionCount, + boolean usingPatterns, boolean groupManagement, String topic) { + Collection listenedPartitions; + Collection 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 getPartitionInfo(String topic, + final ExtendedConsumerProperties extendedConsumerProperties, + final ConsumerFactory consumerFactory, int partitionCount) { + return provisioningProvider.getPartitionsForTopic(partitionCount, + extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(), + () -> { + try (Consumer consumer = consumerFactory.createConsumer()) { + return consumer.partitionsFor(topic); + } + }, topic); + } + + Map getTopicsInUse() { + return this.topicsInUse; + } + @Override protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, - ExtendedConsumerProperties properties) throws Exception { + ExtendedConsumerProperties properties) { boolean anonymous = !StringUtils.hasText(group); - String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; + String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID() : group; Map configs = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties, this.configurationProperties); @@ -202,7 +273,7 @@ public class ReactorKafkaBinder * it is still required by the provisioner, however. */ List destList = Arrays.stream(StringUtils.commaDelimitedListToStringArray(destinations)) - .map(dest -> dest.trim()) + .map(String::trim) .toList(); ReceiverOptions 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 finalOpts = opts; + Map props = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties, + this.configurationProperties); + + DefaultKafkaConsumerFactory 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> 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 getMessageProducers() { + return this.messageProducers; } @Override @@ -384,5 +470,4 @@ public class ReactorKafkaBinder } } - } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicator.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicator.java new file mode 100644 index 000000000..3f4887df8 --- /dev/null +++ b/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 getTopicsInUse() { + return this.binder.getTopicsInUse(); + } + + @Override + protected Health buildBinderSpecificHealthDetails() { + Map messageProducerSupportInfo = binder.getMessageProducers(); + if (messageProducerSupportInfo.isEmpty()) { + return Health.unknown().build(); + } + + Status status = Status.UP; + List> messageProducers = new ArrayList<>(); + + Map 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(); + } +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderHealthIndicatorTests.java new file mode 100644 index 000000000..c8b39414c --- /dev/null +++ b/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 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 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 partitions(Node leader) { + List partitions = new ArrayList<>(); + partitions.add(new PartitionInfo(TEST_TOPIC, 0, leader, null, null)); + return partitions; + } + +} 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 a82f573f0..dc412dd2a 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 @@ -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 aggregatedDetails = new HashMap<>(); - aggregatedDetails.putAll(topicsHealth.getDetails()); - aggregatedDetails.putAll(listenerContainersHealth.getDetails()); - builder.status(aggregatedStatus).withDetails(aggregatedDetails); + @Override + protected Map getTopicsInUse() { + return this.binder.getTopicsInUse(); } - private Health safelyBuildTopicsHealth() { - Future 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 downMessages = new HashSet<>(); - Set checkedTopics = new HashSet<>(); - final Map 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 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> 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(); - } - } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java index 94d456e20..c5bba0eb5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java +++ b/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 topicInfo : this.binder + for (Map.Entry 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 offsetComputation = computeOffsetComputationFunction(topic, group); final Gauge register = Gauge.builder(OFFSET_LAG_METRIC_NAME, this, offsetComputation) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 6629039ce..77d12d37f 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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 partitionInfos; - - private final boolean isTopicPattern; - - TopicInformation(String consumerGroup, Collection 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 getPartitionInfos() { - return this.partitionInfos; - } - - } - /** * Helper class to send to DLQ. * 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 e748ce1cd..ba22ace6a 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 @@ -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 topicsInUse = new HashMap<>(); + private final Map topicsInUse = new HashMap<>(); @BeforeEach public void setup() { @@ -92,7 +93,7 @@ public class KafkaBinderHealthIndicatorTest { @Test void kafkaBinderIsUpWithNoConsumers() { final List 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 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 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 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 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 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 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 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 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 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()) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java index 54f912114..5bcaf02f1 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java +++ b/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; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 766278e5b..bb44daa9f 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/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();