diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java index 7677ff6c6..7af3af917 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java @@ -18,8 +18,8 @@ package org.springframework.cloud.stream.binder.kafka; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; -import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -42,6 +42,7 @@ import org.springframework.kafka.core.ConsumerFactory; * @author Henryk Konsek * @author Gary Russell * @author Laur Aliste + * @author Soby Chacko */ public class KafkaBinderHealthIndicator implements HealthIndicator { @@ -72,44 +73,44 @@ public class KafkaBinderHealthIndicator implements HealthIndicator { @Override public Health health() { ExecutorService exec = Executors.newSingleThreadExecutor(); - Future future = exec.submit(new Callable() { - - @Override - public Health call() { - try { - if (metadataConsumer == null) { - synchronized(KafkaBinderHealthIndicator.this) { - if (metadataConsumer == null) { - metadataConsumer = consumerFactory.createConsumer(); - } + Future future = exec.submit(() -> { + try { + if (metadataConsumer == null) { + synchronized(KafkaBinderHealthIndicator.this) { + if (metadataConsumer == null) { + metadataConsumer = consumerFactory.createConsumer(); } } - synchronized (metadataConsumer) { - Set downMessages = new HashSet<>(); - for (String topic : KafkaBinderHealthIndicator.this.binder.getTopicsInUse().keySet()) { + } + synchronized (metadataConsumer) { + Set downMessages = new HashSet<>(); + final Map topicsInUse = + KafkaBinderHealthIndicator.this.binder.getTopicsInUse(); + for (String topic : topicsInUse.keySet()) { + KafkaMessageChannelBinder.TopicInformation topicInformation = topicsInUse.get(topic); + if (!topicInformation.isTopicPattern()) { List partitionInfos = metadataConsumer.partitionsFor(topic); for (PartitionInfo partitionInfo : partitionInfos) { - if (KafkaBinderHealthIndicator.this.binder.getTopicsInUse().get(topic).getPartitionInfos() + if (topicInformation.getPartitionInfos() .contains(partitionInfo) && partitionInfo.leader().id() == -1) { downMessages.add(partitionInfo.toString()); } } } - if (downMessages.isEmpty()) { - return Health.up().build(); - } - else { - return Health.down() - .withDetail("Following partitions in use have no leaders: ", downMessages.toString()) - .build(); - } + } + if (downMessages.isEmpty()) { + return Health.up().build(); + } + else { + return Health.down() + .withDetail("Following partitions in use have no leaders: ", downMessages.toString()) + .build(); } } - catch (Exception e) { - return Health.down(e).build(); - } } - + catch (Exception e) { + return Health.down(e).build(); + } }); try { return future.get(this.timeout, TimeUnit.SECONDS); diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 3fc9fbc81..926e19475 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -255,7 +255,7 @@ public class KafkaMessageChannelBinder extends ((DisposableBean) producerFB).destroy(); return partitionsFor; }); - this.topicsInUse.put(destination.getName(), new TopicInformation(null, partitions)); + this.topicsInUse.put(destination.getName(), new TopicInformation(null, partitions, false)); if (producerProperties.isPartitioned() && producerProperties.getPartitionCount() < partitions.size()) { if (this.logger.isInfoEnabled()) { this.logger.info("The `partitionCount` of the producer for topic " + destination.getName() + " is " @@ -488,7 +488,7 @@ public class KafkaMessageChannelBinder extends } } } - this.topicsInUse.put(topic, new TopicInformation(group, listenedPartitions)); + this.topicsInUse.put(topic, new TopicInformation(group, listenedPartitions, usingPatterns)); return listenedPartitions; } @@ -567,13 +567,13 @@ public class KafkaMessageChannelBinder extends // not just the ones this binding is listening to; doesn't seem right for a health check. Collection partitionInfos = getPartitionInfo(destination.getName(), consumerProperties, consumerFactory, -1); - this.topicsInUse.put(destination.getName(), new TopicInformation(group, partitionInfos)); + this.topicsInUse.put(destination.getName(), new TopicInformation(group, partitionInfos, false)); } else { for (int i = 0; i < topics.length; i++) { Collection partitionInfos = getPartitionInfo(topics[i], consumerProperties, consumerFactory, -1); - this.topicsInUse.put(topics[i], new TopicInformation(group, partitionInfos)); + this.topicsInUse.put(topics[i], new TopicInformation(group, partitionInfos, false)); } } @@ -938,9 +938,12 @@ public class KafkaMessageChannelBinder extends private final Collection partitionInfos; - TopicInformation(String consumerGroup, Collection partitionInfos) { + private final boolean isTopicPattern; + + TopicInformation(String consumerGroup, Collection partitionInfos, boolean isTopicPattern) { this.consumerGroup = consumerGroup; this.partitionInfos = partitionInfos; + this.isTopicPattern = isTopicPattern; } String getConsumerGroup() { @@ -951,6 +954,10 @@ public class KafkaMessageChannelBinder extends return consumerGroup != null; } + boolean isTopicPattern() { + return isTopicPattern; + } + Collection getPartitionInfos() { return partitionInfos; } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java index bec846086..8b3db635d 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -43,11 +43,14 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Barry Commins * @author Gary Russell * @author Laur Aliste + * @author Soby Chacko */ public class KafkaBinderHealthIndicatorTest { private static final String TEST_TOPIC = "test"; + private static final String REGEX_TOPIC = "regex*"; + private KafkaBinderHealthIndicator indicator; @Mock @@ -73,16 +76,26 @@ public class KafkaBinderHealthIndicatorTest { @Test public void kafkaBinderIsUp() { final List partitions = partitions(new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group1-healthIndicator", partitions)); + topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group1-healthIndicator", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.UP); } + @Test + public void kafkaBinderIsUpWithRegexTopic() { + topicsInUse.put(REGEX_TOPIC, new KafkaMessageChannelBinder.TopicInformation("regex-healthIndicator", null, true)); + Health health = indicator.health(); + //verify no consumer interaction for retrieving partitions + org.mockito.BDDMockito.verify(consumer, Mockito.never()).partitionsFor(REGEX_TOPIC); + //Ensuring the normal health check returns with status "up" + assertThat(health.getStatus()).isEqualTo(Status.UP); + } + @Test public void kafkaBinderIsDown() { final List partitions = partitions(new Node(-1, null, 0)); - topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group2-healthIndicator", partitions)); + topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group2-healthIndicator", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); Health health = indicator.health(); assertThat(health.getStatus()).isEqualTo(Status.DOWN); @@ -91,7 +104,7 @@ public class KafkaBinderHealthIndicatorTest { @Test(timeout = 5000) public void kafkaBinderDoesNotAnswer() { final List partitions = partitions(new Node(-1, null, 0)); - topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group3-healthIndicator", partitions)); + topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group3-healthIndicator", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willAnswer(new Answer() { @Override @@ -110,7 +123,7 @@ public class KafkaBinderHealthIndicatorTest { @Test public void createsConsumerOnceWhenInvokedMultipleTimes() { final List partitions = partitions(new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group4-healthIndicator", partitions)); + topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation("group4-healthIndicator", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); indicator.health(); diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java index dd3dc2b90..956397967 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -46,6 +46,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Henryk Konsek * @author Thomas Cheyney + * @author Soby Chacko */ public class KafkaBinderMetricsTest { @@ -83,7 +84,7 @@ public class KafkaBinderMetricsTest { public void shouldIndicateLag() { org.mockito.BDDMockito.given(consumer.committed(ArgumentMatchers.any(TopicPartition.class))).willReturn(new OffsetAndMetadata(500)); List partitions = partitions(new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, new TopicInformation("group1-metrics", partitions)); + topicsInUse.put(TEST_TOPIC, new TopicInformation("group1-metrics", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); metrics.bindTo(meterRegistry); assertThat(meterRegistry.getMeters()).hasSize(1); @@ -99,7 +100,7 @@ public class KafkaBinderMetricsTest { org.mockito.BDDMockito.given(consumer.endOffsets(ArgumentMatchers.anyCollection())).willReturn(endOffsets); org.mockito.BDDMockito.given(consumer.committed(ArgumentMatchers.any(TopicPartition.class))).willReturn(new OffsetAndMetadata(500)); List partitions = partitions(new Node(0, null, 0), new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, new TopicInformation("group2-metrics", partitions)); + topicsInUse.put(TEST_TOPIC, new TopicInformation("group2-metrics", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); metrics.bindTo(meterRegistry); assertThat(meterRegistry.getMeters()).hasSize(1); @@ -110,7 +111,7 @@ public class KafkaBinderMetricsTest { @Test public void shouldIndicateFullLagForNotCommittedGroups() { List partitions = partitions(new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, new TopicInformation("group3-metrics", partitions)); + topicsInUse.put(TEST_TOPIC, new TopicInformation("group3-metrics", partitions, false)); org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC)).willReturn(partitions); metrics.bindTo(meterRegistry); assertThat(meterRegistry.getMeters()).hasSize(1); @@ -121,7 +122,7 @@ public class KafkaBinderMetricsTest { @Test public void shouldNotCalculateLagForProducerTopics() { List partitions = partitions(new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, new TopicInformation(null, partitions)); + topicsInUse.put(TEST_TOPIC, new TopicInformation(null, partitions, false)); metrics.bindTo(meterRegistry); assertThat(meterRegistry.getMeters()).isEmpty(); } @@ -129,7 +130,7 @@ public class KafkaBinderMetricsTest { @Test public void createsConsumerOnceWhenInvokedMultipleTimes() { final List partitions = partitions(new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, new TopicInformation("group4-metrics", partitions)); + topicsInUse.put(TEST_TOPIC, new TopicInformation("group4-metrics", partitions, false)); metrics.bindTo(meterRegistry); @@ -146,7 +147,7 @@ public class KafkaBinderMetricsTest { .willReturn(consumer); final List partitions = partitions(new Node(0, null, 0)); - topicsInUse.put(TEST_TOPIC, new TopicInformation("group5-metrics", partitions)); + topicsInUse.put(TEST_TOPIC, new TopicInformation("group5-metrics", partitions, false)); metrics.bindTo(meterRegistry);