This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 41769a0 [FLINK-12057] Refactor MemoryLogger to accept termination future instead of ActorSystem 41769a0 is described below commit 41769a0e2d8f8e336685641ee2f3639b00777839 Author: Chesnay Schepler <ches...@apache.org> AuthorDate: Thu Mar 28 15:20:25 2019 +0100 [FLINK-12057] Refactor MemoryLogger to accept termination future instead of ActorSystem --- .../runtime/taskexecutor/TaskManagerRunner.java | 2 +- .../flink/runtime/taskmanager/MemoryLogger.java | 20 ++++++++++---------- .../flink/runtime/taskmanager/TaskManager.scala | 2 -- 3 files changed, 11 insertions(+), 13 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java index c8a8a1f..889221b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java @@ -159,7 +159,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync this.terminationFuture = new CompletableFuture<>(); this.shutdown = false; - MemoryLogger.startIfConfigured(LOG, configuration, metricQueryServiceActorSystem); + MemoryLogger.startIfConfigured(LOG, configuration, terminationFuture); } // -------------------------------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/MemoryLogger.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/MemoryLogger.java index 91849d7..41b5985 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/MemoryLogger.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/MemoryLogger.java @@ -21,7 +21,6 @@ package org.apache.flink.runtime.taskmanager; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.TaskManagerOptions; -import akka.actor.ActorSystem; import org.slf4j.Logger; import javax.management.MBeanServer; @@ -34,6 +33,7 @@ import java.lang.management.MemoryPoolMXBean; import java.lang.management.MemoryType; import java.lang.management.MemoryUsage; import java.util.List; +import java.util.concurrent.CompletableFuture; /** * A thread the periodically logs statistics about: @@ -57,14 +57,14 @@ public class MemoryLogger extends Thread { private final BufferPoolMXBean directBufferBean; - private final ActorSystem monitored; + private final CompletableFuture<Void> monitored; private volatile boolean running = true; public static void startIfConfigured( Logger logger, Configuration configuration, - ActorSystem taskManagerSystem) { + CompletableFuture<Void> taskManagerTerminationFuture) { if (!logger.isInfoEnabled() || !configuration.getBoolean(TaskManagerOptions.DEBUG_MEMORY_LOG)) { return; } @@ -73,19 +73,19 @@ public class MemoryLogger extends Thread { new MemoryLogger( logger, configuration.getLong(TaskManagerOptions.DEBUG_MEMORY_USAGE_LOG_INTERVAL_MS), - taskManagerSystem).start(); + taskManagerTerminationFuture).start(); } /** - * Creates a new memory logger that logs in the given interval and lives as long as the - * given actor system. + * Creates a new memory logger that logs in the given interval and lives until the + * given termination future completes. * * @param logger The logger to use for outputting the memory statistics. * @param interval The interval in which the thread logs. - * @param monitored The actor system to whose life the thread is bound. The thread terminates - * once the actor system terminates. + * @param monitored termination future for the system to whose life the thread is bound. The thread terminates + * once the system terminates. */ - public MemoryLogger(Logger logger, long interval, ActorSystem monitored) { + public MemoryLogger(Logger logger, long interval, CompletableFuture<Void> monitored) { super("Memory Logger"); setDaemon(true); setPriority(Thread.MIN_PRIORITY); @@ -125,7 +125,7 @@ public class MemoryLogger extends Thread { @Override public void run() { try { - while (running && (monitored == null || !monitored.whenTerminated().isCompleted())) { + while (running && (monitored == null || !monitored.isDone())) { logger.info(getMemoryUsageStatsAsString(memoryBean)); logger.info(getDirectMemoryStatsAsString(directBufferBean)); logger.info(getMemoryPoolStatsAsString(poolBeans)); diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala index 02f1753..35e6c28 100644 --- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala +++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala @@ -1885,8 +1885,6 @@ object TaskManager { ) } - MemoryLogger.startIfConfigured(LOG.logger, configuration, taskManagerSystem) - // block until everything is done Await.ready(taskManagerSystem.whenTerminated, Duration.Inf) } catch {