root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 4 Packages Projects Releases Wiki Activity

MINOR: Make PushHttpMetricsReporter use daemon threads. 

This is a safe guard against users that do not properly close() the reporter, which causes
the process to hang even if its main() method returns. This was the case with Kafka Streams
apps in some cases in Kafka < 1.1.0 (KAFKA-6383). Without this fix, this behavior can make
it difficult to use this class in some types of system tests.

Author: Ewen Cheslack-Postava <me@ewencp.org>

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Ismael Juma <ismael@juma.me.uk>

Closes #4629 from ewencp/http-metrics-dont-block-shutdown
This commit is contained in:
Ewen Cheslack-Postava 2018-03-02 13:16:21 -08:00
parent c5fbea622e
commit 73fb1bffe1
1 changed files with 13 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java
View File

@ -48,6 +48,8 @@ import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
/**
@ -96,7 +98,17 @@ public class PushHttpMetricsReporter implements MetricsReporter {
public PushHttpMetricsReporter() {
time = new SystemTime();
executor = Executors.newSingleThreadScheduledExecutor();
executor = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = Executors.defaultThreadFactory().newThread(r);
thread.setName("PushHttpMetricsReporterThread");
// Ensure these are daemon threads so they won't block shutdown if the MetricsReporter is not properly
// closed (e.g. as might happen with Kafka Streams < 1.1.0)
thread.setDaemon(true);
return thread;
}
});
}
PushHttpMetricsReporter(Time mockTime, ScheduledExecutorService mockExecutor) {