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

Kafka Streams binder health indicator improvements

Browse Source 
Caching AdminClient in Kafk Streams binder health indicator.

Resolves #791
This commit is contained in:
Soby Chacko
2019-10-31 16:11:59 -04:00
parent 06e5739fbd
commit e2f1092173
1 changed files with 33 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -59,6 +59,8 @@ public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator {
private static final ThreadLocal<Status> healthStatusThreadLocal = new ThreadLocal<>();
private AdminClient adminClient;
KafkaStreamsBinderHealthIndicator(KafkaStreamsRegistry kafkaStreamsRegistry,
KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties,
KafkaProperties kafkaProperties,
@@ -75,34 +77,34 @@ public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator {
@Override
protected void doHealthCheck(Health.Builder builder) throws Exception {
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);
initAdminClient();
synchronized (this.adminClient) {
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 thead local 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 {
boolean up = true;
for (KafkaStreams kStream : kafkaStreamsRegistry.getKafkaStreams()) {
up &= kStream.state().isRunning();
builder.withDetails(buildDetails(kStream));
final ListTopicsResult listTopicsResult = this.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);
}
builder.status(up ? Status.UP : Status.DOWN);
}
}
}
@@ -121,6 +123,13 @@ public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator {
}
}
private synchronized AdminClient initAdminClient() {
if (this.adminClient == null) {
this.adminClient = AdminClient.create(this.adminClientProperties);
}
return this.adminClient;
}
private Map<String, Object> buildDetails(KafkaStreams kafkaStreams) {
final Map<String, Object> details = new HashMap<>();
final Map<String, Object> perAppdIdDetails = new HashMap<>();