SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Regex topics in health indicator

Browse Source 
If the destination topic is pattern based, only do the basic up/down check
in the health indicator. In this case, the health indicator does not do any
partitions queries as it does for normal topics.

Resolves #430
This commit is contained in:
Soby Chacko
2018-10-11 12:43:39 -04:00
committed by Gary Russell
parent a5cec9a25c
commit 0227c9ecc9
4 changed files with 66 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
View File

@@ -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<Health> future = exec.submit(new Callable<Health>() {
@Override
public Health call() {
try {
if (metadataConsumer == null) {
synchronized(KafkaBinderHealthIndicator.this) {
if (metadataConsumer == null) {
metadataConsumer = consumerFactory.createConsumer();
}
Future<Health> future = exec.submit(() -> {
try {
if (metadataConsumer == null) {
synchronized(KafkaBinderHealthIndicator.this) {
if (metadataConsumer == null) {
metadataConsumer = consumerFactory.createConsumer();
}
}
synchronized (metadataConsumer) {
Set<String> downMessages = new HashSet<>();
for (String topic : KafkaBinderHealthIndicator.this.binder.getTopicsInUse().keySet()) {
}
synchronized (metadataConsumer) {
Set<String> downMessages = new HashSet<>();
final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse =
KafkaBinderHealthIndicator.this.binder.getTopicsInUse();
for (String topic : topicsInUse.keySet()) {
KafkaMessageChannelBinder.TopicInformation topicInformation = topicsInUse.get(topic);
if (!topicInformation.isTopicPattern()) {
List<PartitionInfo> 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);

17
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -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<PartitionInfo> 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<PartitionInfo> 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<PartitionInfo> partitionInfos;
TopicInformation(String consumerGroup, 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() {
@@ -951,6 +954,10 @@ public class KafkaMessageChannelBinder extends
return consumerGroup != null;
}
boolean isTopicPattern() {
return isTopicPattern;
}
Collection<PartitionInfo> getPartitionInfos() {
return partitionInfos;
}

23
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java
View File

@@ -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<PartitionInfo> 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<PartitionInfo> 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<PartitionInfo> 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<Object>() {
@Override
@@ -110,7 +123,7 @@ public class KafkaBinderHealthIndicatorTest {
@Test
public void createsConsumerOnceWhenInvokedMultipleTimes() {
final List<PartitionInfo> 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();

15
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
View File

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