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

Merge pull request #773 from sobychacko/gh-552

Browse Source 
Kafka/Streams binder health indicator improvements
This commit is contained in:
Oleg Zhurakousky
2019-10-23 16:53:37 +02:00
committed by GitHub
parent e4fed0a52d 05e2918bc0
commit 1264434871
5 changed files with 94 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
View File

@@ -81,9 +81,9 @@ public class KafkaTopicProvisioner implements
// @checkstyle:on
InitializingBean {
private static final int DEFAULT_OPERATION_TIMEOUT = 30;
private static final Log logger = LogFactory.getLog(KafkaTopicProvisioner.class);
private final Log logger = LogFactory.getLog(getClass());
private static final int DEFAULT_OPERATION_TIMEOUT = 30;
private final KafkaBinderConfigurationProperties configurationProperties;
@@ -242,7 +242,7 @@ public class KafkaTopicProvisioner implements
* @param bootProps the boot kafka properties.
* @param binderProps the binder kafka properties.
*/
private void normalalizeBootPropsWithBinder(Map<String, Object> adminProps,
public static void normalalizeBootPropsWithBinder(Map<String, Object> adminProps,
KafkaProperties bootProps, KafkaBinderConfigurationProperties binderProps) {
// First deal with the outlier
String kafkaConnectionString = binderProps.getKafkaConnectionString();
@@ -263,8 +263,8 @@ public class KafkaTopicProvisioner implements
}
if (adminConfigNames.contains(key)) {
Object replaced = adminProps.put(key, value);
if (replaced != null && this.logger.isDebugEnabled()) {
this.logger.debug("Overrode boot property: [" + key + "], from: ["
if (replaced != null && KafkaTopicProvisioner.logger.isDebugEnabled()) {
KafkaTopicProvisioner.logger.debug("Overrode boot property: [" + key + "], from: ["
+ replaced + "] to: [" + value + "]");
}
}

78
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java
View File

@@ -16,11 +16,17 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.TaskMetadata;
@@ -29,6 +35,9 @@ import org.apache.kafka.streams.processor.ThreadMetadata;
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.autoconfigure.kafka.KafkaProperties;
import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
/**
@@ -39,26 +48,81 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean;
*/
public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator {
private final KafkaStreamsRegistry kafkaStreamsRegistry;
private final Log logger = LogFactory.getLog(getClass());
KafkaStreamsBinderHealthIndicator(KafkaStreamsRegistry kafkaStreamsRegistry) {
private final KafkaStreamsRegistry kafkaStreamsRegistry;
private final KafkaStreamsBinderConfigurationProperties configurationProperties;
private final Map<String, Object> adminClientProperties;
private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue;
private static final ThreadLocal<Status> healthStatusThreadLocal = new ThreadLocal<>();
KafkaStreamsBinderHealthIndicator(KafkaStreamsRegistry kafkaStreamsRegistry,
KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties,
KafkaProperties kafkaProperties,
KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) {
super("Kafka-streams health check failed");
kafkaProperties.buildAdminProperties();
this.configurationProperties = kafkaStreamsBinderConfigurationProperties;
this.adminClientProperties = kafkaProperties.buildAdminProperties();
KafkaTopicProvisioner.normalalizeBootPropsWithBinder(this.adminClientProperties, kafkaProperties,
kafkaStreamsBinderConfigurationProperties);
this.kafkaStreamsRegistry = kafkaStreamsRegistry;
this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue;
}
@Override
protected void doHealthCheck(Health.Builder builder) throws Exception {
boolean up = true;
for (KafkaStreams kStream : kafkaStreamsRegistry.getKafkaStreams()) {
up &= kStream.state().isRunning();
builder.withDetails(buildDetails(kStream));
AdminClient adminClient = null;
try {
final Status status = healthStatusThreadLocal.get();
//If one of the kafka streams binders (kstream, ktable, globalktable) was down before on the same request,
//retrieve that from the theadlocal storage where it was saved before. This is done in order to avoid
//the duration of the total health check since in the case of Kafka Streams each binder tries to do
//its own health check and since we already know that this is DOWN, simply pass that information along.
if (status == Status.DOWN) {
builder.withDetail("No topic information available", "Kafka broker is not reachable");
builder.status(Status.DOWN);
}
else {
adminClient = AdminClient.create(this.adminClientProperties);
final ListTopicsResult listTopicsResult = adminClient.listTopics();
listTopicsResult.listings().get(this.configurationProperties.getHealthTimeout(), TimeUnit.SECONDS);
if (this.kafkaStreamsBindingInformationCatalogue.getStreamsBuilderFactoryBeans().isEmpty()) {
builder.withDetail("No Kafka Streams bindings have been established", "Kafka Streams binder did not detect any processors");
builder.status(Status.UNKNOWN);
}
else {
boolean up = true;
for (KafkaStreams kStream : kafkaStreamsRegistry.getKafkaStreams()) {
up &= kStream.state().isRunning();
builder.withDetails(buildDetails(kStream));
}
builder.status(up ? Status.UP : Status.DOWN);
}
}
}
catch (Exception e) {
builder.withDetail("No topic information available", "Kafka broker is not reachable");
builder.status(Status.DOWN);
builder.withException(e);
//Store binder down status into a thread local storage.
healthStatusThreadLocal.set(Status.DOWN);
}
finally {
// Close admin client immediately.
adminClient.close(Duration.ofSeconds(0));
}
builder.status(up ? Status.UP : Status.DOWN);
}
private Map<String, Object> buildDetails(KafkaStreams kafkaStreams) {
final Map<String, Object> details = new HashMap<>();
final Map<String, Object> perAppdIdDetails = new HashMap<>();
if (kafkaStreams.state().isRunning()) {
for (ThreadMetadata metadata : kafkaStreams.localThreadsMetadata()) {
perAppdIdDetails.put("threadName", metadata.threadName());

9
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicatorConfiguration.java
View File

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -35,8 +37,11 @@ class KafkaStreamsBinderHealthIndicatorConfiguration {
@Bean
@ConditionalOnBean(KafkaStreamsRegistry.class)
KafkaStreamsBinderHealthIndicator kafkaStreamsBinderHealthIndicator(
KafkaStreamsRegistry kafkaStreamsRegistry) {
return new KafkaStreamsBinderHealthIndicator(kafkaStreamsRegistry);
KafkaStreamsRegistry kafkaStreamsRegistry, KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties,
KafkaProperties kafkaProperties, KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) {
return new KafkaStreamsBinderHealthIndicator(kafkaStreamsRegistry, kafkaStreamsBinderConfigurationProperties,
kafkaProperties, kafkaStreamsBindingInformationCatalogue);
}
}

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.kafka;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -112,8 +113,15 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe
final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = KafkaBinderHealthIndicator.this.binder
.getTopicsInUse();
if (topicsInUse.isEmpty()) {
return Health.down().withDetail("No topic information available",
"Kafka broker is not reachable").build();
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()) {

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

@@ -159,7 +159,7 @@ public class KafkaBinderHealthIndicatorTest {
@Test
public void testIfNoTopicsRegisteredByTheBinderProvidesDownStatus() {
Health health = indicator.health();
assertThat(health.getStatus()).isEqualTo(Status.DOWN);
assertThat(health.getStatus()).isEqualTo(Status.UNKNOWN);
}
private List<PartitionInfo> partitions(Node leader) {