You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/10/15 10:15:50 UTC
[flink] branch release-1.5 updated: [FLINK-10282][rest] Separate
REST and Dispatcher RPC thread pools
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.5
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.5 by this push:
new 5d4d51d [FLINK-10282][rest] Separate REST and Dispatcher RPC thread pools
5d4d51d is described below
commit 5d4d51d952262809f670a87dae6f4816f689d9cb
Author: zentol <ch...@apache.org>
AuthorDate: Mon Oct 8 14:42:58 2018 +0200
[FLINK-10282][rest] Separate REST and Dispatcher RPC thread pools
---
docs/_includes/generated/rest_configuration.html | 5 +++++
.../apache/flink/configuration/RestOptions.java | 4 ++++
.../flink/docs/rest/RestAPIDocGenerator.java | 7 ++-----
.../runtime/dispatcher/DispatcherRestEndpoint.java | 4 ++--
.../runtime/entrypoint/ClusterEntrypoint.java | 7 +++++--
.../runtime/entrypoint/JobClusterEntrypoint.java | 4 ++--
.../entrypoint/SessionClusterEntrypoint.java | 4 ++--
.../jobmaster/MiniDispatcherRestEndpoint.java | 4 ++--
.../flink/runtime/minicluster/MiniCluster.java | 6 +++++-
.../runtime/webmonitor/WebMonitorEndpoint.java | 23 ++++++++++++++++++----
10 files changed, 48 insertions(+), 20 deletions(-)
diff --git a/docs/_includes/generated/rest_configuration.html b/docs/_includes/generated/rest_configuration.html
index 25da9cf..1aa963f 100644
--- a/docs/_includes/generated/rest_configuration.html
+++ b/docs/_includes/generated/rest_configuration.html
@@ -57,5 +57,10 @@
<td style="word-wrap: break-word;">104857600</td>
<td>The maximum content length in bytes that the server will handle.</td>
</tr>
+ <tr>
+ <td><h5>rest.server.numThreads</h5></td>
+ <td style="word-wrap: break-word;">4</td>
+ <td>The number of threads for the asynchronous processing of requests.</td>
+ </tr>
</tbody>
</table>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java
index c834483..edfd39b 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java
@@ -117,4 +117,8 @@ public class RestOptions {
.defaultValue(104_857_600)
.withDescription("The maximum content length in bytes that the client will handle.");
+ public static final ConfigOption<Integer> SERVER_NUM_THREADS =
+ key("rest.server.numThreads")
+ .defaultValue(4)
+ .withDescription("The number of threads for the asynchronous processing of requests.");
}
diff --git a/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java b/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java
index cebde56..8e59ed0 100644
--- a/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java
+++ b/flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java
@@ -22,7 +22,6 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestOptions;
-import org.apache.flink.runtime.concurrent.Executors;
import org.apache.flink.runtime.dispatcher.DispatcherGateway;
import org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint;
import org.apache.flink.runtime.leaderelection.LeaderContender;
@@ -68,7 +67,7 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import static org.apache.flink.docs.util.Utils.escapeCharacters;
@@ -311,7 +310,6 @@ public class RestAPIDocGenerator {
private static final Configuration config;
private static final RestServerEndpointConfiguration restConfig;
private static final RestHandlerConfiguration handlerConfig;
- private static final Executor executor;
private static final GatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever;
private static final GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever;
private static final MetricQueryServiceRetriever metricQueryServiceRetriever;
@@ -327,7 +325,6 @@ public class RestAPIDocGenerator {
throw new RuntimeException("Implementation error. RestServerEndpointConfiguration#fromConfiguration failed for default configuration.");
}
handlerConfig = RestHandlerConfiguration.fromConfiguration(config);
- executor = Executors.directExecutor();
dispatcherGatewayRetriever = () -> null;
resourceManagerGatewayRetriever = () -> null;
@@ -342,7 +339,7 @@ public class RestAPIDocGenerator {
handlerConfig,
resourceManagerGatewayRetriever,
NoOpTransientBlobService.INSTANCE,
- executor,
+ Executors.newFixedThreadPool(1),
metricQueryServiceRetriever,
NoOpElectionService.INSTANCE,
NoOpFatalErrorHandler.INSTANCE);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
index ba080c6..1bd6ad9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
@@ -43,7 +43,7 @@ import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
/**
* REST endpoint for the {@link Dispatcher} component.
@@ -59,7 +59,7 @@ public class DispatcherRestEndpoint extends WebMonitorEndpoint<DispatcherGateway
RestHandlerConfiguration restConfiguration,
GatewayRetriever<ResourceManagerGateway> resourceManagerRetriever,
TransientBlobService transientBlobService,
- Executor executor,
+ ExecutorService executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService,
FatalErrorHandler fatalErrorHandler) throws IOException {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index e0561a7..0a61d53 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -90,6 +90,7 @@ import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import scala.concurrent.duration.FiniteDuration;
@@ -332,7 +333,9 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
dispatcherGatewayRetriever,
resourceManagerGatewayRetriever,
transientBlobCache,
- rpcService.getExecutor(),
+ WebMonitorEndpoint.createExecutorService(
+ configuration.getInteger(RestOptions.SERVER_NUM_THREADS),
+ "DispatcherRestEndpoint"),
new AkkaQueryServiceRetriever(actorSystem, timeout),
highAvailabilityServices.getWebMonitorLeaderElectionService());
@@ -689,7 +692,7 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever,
LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
TransientBlobService transientBlobService,
- Executor executor,
+ ExecutorService executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService) throws Exception;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
index 80a9da2..07afe89 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
@@ -47,7 +47,7 @@ import org.apache.flink.util.FlinkException;
import javax.annotation.Nullable;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
/**
* Base class for per-job cluster entry points.
@@ -64,7 +64,7 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever,
LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
TransientBlobService transientBlobService,
- Executor executor,
+ ExecutorService executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService) throws Exception {
final RestHandlerConfiguration restHandlerConfiguration = RestHandlerConfiguration.fromConfiguration(configuration);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
index 40eb8b7..dbb9703 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
@@ -50,7 +50,7 @@ import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
/**
* Base class for session cluster entry points.
@@ -84,7 +84,7 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever,
LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
TransientBlobService transientBlobService,
- Executor executor,
+ ExecutorService executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService) throws Exception {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
index b31c46a..dae9d8e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
@@ -32,7 +32,7 @@ import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
import java.io.IOException;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
/**
* REST endpoint for the {@link JobClusterEntrypoint}.
@@ -46,7 +46,7 @@ public class MiniDispatcherRestEndpoint extends WebMonitorEndpoint<RestfulGatewa
RestHandlerConfiguration restConfiguration,
GatewayRetriever<ResourceManagerGateway> resourceManagerRetriever,
TransientBlobService transientBlobService,
- Executor executor,
+ ExecutorService executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService,
FatalErrorHandler fatalErrorHandler) throws IOException {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
index 13642dd..0628245 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.JobSubmissionResult;
import org.apache.flink.api.common.io.FileOutputFormat;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.WebOptions;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.akka.AkkaUtils;
@@ -70,6 +71,7 @@ import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.rpc.akka.AkkaRpcService;
import org.apache.flink.runtime.taskexecutor.TaskExecutor;
import org.apache.flink.runtime.taskexecutor.TaskManagerRunner;
+import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaQueryServiceRetriever;
import org.apache.flink.runtime.webmonitor.retriever.impl.RpcGatewayRetriever;
import org.apache.flink.util.AutoCloseableAsync;
@@ -348,7 +350,9 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
RestHandlerConfiguration.fromConfiguration(configuration),
resourceManagerGatewayRetriever,
blobServer.getTransientBlobService(),
- commonRpcService.getExecutor(),
+ WebMonitorEndpoint.createExecutorService(
+ configuration.getInteger(RestOptions.SERVER_NUM_THREADS, 1),
+ "DispatcherRestEndpoint"),
new AkkaQueryServiceRetriever(
metricQueryServiceActorSystem,
Time.milliseconds(configuration.getLong(WebOptions.TIMEOUT))),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
index f9b2fa8..02d92dc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
@@ -114,11 +114,13 @@ import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerLogFileHead
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerStdoutFileHeaders;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagersHeaders;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.util.ExecutorThreadFactory;
import org.apache.flink.runtime.webmonitor.history.ArchivedJson;
import org.apache.flink.runtime.webmonitor.history.JsonArchivist;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.Preconditions;
@@ -134,7 +136,9 @@ import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
/**
* Rest endpoint which serves the web frontend REST calls.
@@ -148,7 +152,7 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
protected final RestHandlerConfiguration restConfiguration;
private final GatewayRetriever<ResourceManagerGateway> resourceManagerRetriever;
private final TransientBlobService transientBlobService;
- protected final Executor executor;
+ protected final ExecutorService executor;
private final ExecutionGraphCache executionGraphCache;
private final CheckpointStatsCache checkpointStatsCache;
@@ -170,7 +174,7 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
RestHandlerConfiguration restConfiguration,
GatewayRetriever<ResourceManagerGateway> resourceManagerRetriever,
TransientBlobService transientBlobService,
- Executor executor,
+ ExecutorService executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
LeaderElectionService leaderElectionService,
FatalErrorHandler fatalErrorHandler) throws IOException {
@@ -715,7 +719,9 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
protected CompletableFuture<Void> shutDownInternal() {
executionGraphCache.close();
- final CompletableFuture<Void> shutdownFuture = super.shutDownInternal();
+ final CompletableFuture<Void> shutdownFuture = FutureUtils.runAfterwards(
+ super.shutDownInternal(),
+ () -> ExecutorUtils.gracefulShutdown(10, TimeUnit.SECONDS, executor));
final File webUiDir = restConfiguration.getWebUiDir();
@@ -776,4 +782,13 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
}
return archivedJson;
}
+
+ public static ExecutorService createExecutorService(int numThreads, String componentName) {
+ return Executors.newFixedThreadPool(
+ numThreads,
+ new ExecutorThreadFactory.Builder()
+ .setThreadPriority(Thread.MIN_PRIORITY)
+ .setPoolName("Flink-" + componentName)
+ .build());
+ }
}