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

Use single Executor Service in Kafka Binder health indicator

Browse Source 
Resolves #665
This commit is contained in:
Vladislav Fefelov
2019-06-04 08:45:07 +03:00
committed by Oleg Zhurakousky
parent d213b4ff2c
commit 4d59670096
1 changed files with 12 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -30,9 +30,11 @@ 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.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
/**
* Health indicator for Kafka.
@@ -44,10 +46,13 @@ import org.springframework.kafka.core.ConsumerFactory;
* @author Laur Aliste
* @author Soby Chacko
*/
public class KafkaBinderHealthIndicator implements HealthIndicator {
public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBean {
private static final int DEFAULT_TIMEOUT = 60;
private final ExecutorService executor = Executors.newSingleThreadExecutor(
new CustomizableThreadFactory("kafka-binder-health-"));
private final KafkaMessageChannelBinder binder;
private final ConsumerFactory<?, ?> consumerFactory;
@@ -72,8 +77,7 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
@Override
public Health health() {
ExecutorService exec = Executors.newSingleThreadExecutor();
Future<Health> future = exec.submit(this::buildHealthStatus);
Future<Health> future = executor.submit(this::buildHealthStatus);
try {
return future.get(this.timeout, TimeUnit.SECONDS);
}
@@ -91,9 +95,6 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
return Health.down().withDetail("Failed to retrieve partition information in",
this.timeout + " seconds").build();
}
finally {
exec.shutdownNow();
}
}
private Health buildHealthStatus() {
@@ -146,4 +147,9 @@ public class KafkaBinderHealthIndicator implements HealthIndicator {
}
}
@Override
public void destroy() throws Exception {
executor.shutdown();
}
}