zentol closed pull request #6870: [BP-1.5][FLINK-10582] Make REST executor's
thread priority configurable
URL: https://github.com/apache/flink/pull/6870
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/docs/_includes/generated/rest_configuration.html
b/docs/_includes/generated/rest_configuration.html
index 1aa963fb3e2..2c5f539a480 100644
--- a/docs/_includes/generated/rest_configuration.html
+++ b/docs/_includes/generated/rest_configuration.html
@@ -62,5 +62,10 @@
<td style="word-wrap: break-word;">4</td>
<td>The number of threads for the asynchronous processing of
requests.</td>
</tr>
+ <tr>
+ <td><h5>rest.server.thread-priority</h5></td>
+ <td style="word-wrap: break-word;">5</td>
+ <td>Thread priority of the REST server's executor for processing
asynchronous requests. Lowering the thread priority will give Flink's main
components more CPU time whereas increasing will allocate more time for the
REST server's processing.</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 edfd39be808..11c38de06c6 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
@@ -19,6 +19,7 @@
package org.apache.flink.configuration;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.description.Description;
import static org.apache.flink.configuration.ConfigOptions.key;
@@ -121,4 +122,12 @@
key("rest.server.numThreads")
.defaultValue(4)
.withDescription("The number of threads for the
asynchronous processing of requests.");
+
+ public static final ConfigOption<Integer> SERVER_THREAD_PRIORITY =
key("rest.server.thread-priority")
+ .defaultValue(Thread.NORM_PRIORITY)
+ .withDescription(Description.builder()
+ .text("Thread priority of the REST server's executor
for processing asynchronous requests. " +
+ "Lowering the thread priority will give Flink's
main components more CPU time whereas " +
+ "increasing will allocate more time for the
REST server's processing.")
+ .build());
}
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 2c34b530804..c4c583853bf 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
@@ -330,6 +330,7 @@ protected void startClusterComponents(
transientBlobCache,
WebMonitorEndpoint.createExecutorService(
configuration.getInteger(RestOptions.SERVER_NUM_THREADS),
+
configuration.getInteger(RestOptions.SERVER_THREAD_PRIORITY),
"DispatcherRestEndpoint"),
new AkkaQueryServiceRetriever(actorSystem,
timeout),
highAvailabilityServices.getWebMonitorLeaderElectionService());
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 a7840d64830..74ff770e69a 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
@@ -350,6 +350,7 @@ public void start() throws Exception {
blobServer.getTransientBlobService(),
WebMonitorEndpoint.createExecutorService(
configuration.getInteger(RestOptions.SERVER_NUM_THREADS, 1),
+
configuration.getInteger(RestOptions.SERVER_THREAD_PRIORITY),
"DispatcherRestEndpoint"),
new AkkaQueryServiceRetriever(
actorSystem,
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 02d92dc54fb..c480c33c671 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
@@ -783,11 +783,20 @@ public void handleError(final Exception exception) {
return archivedJson;
}
- public static ExecutorService createExecutorService(int numThreads,
String componentName) {
+ public static ExecutorService createExecutorService(int numThreads, int
threadPriority, String componentName) {
+ if (threadPriority < Thread.MIN_PRIORITY || threadPriority >
Thread.MAX_PRIORITY) {
+ throw new IllegalArgumentException(
+ String.format(
+ "The thread priority must be within
(%s, %s) but it was %s.",
+ Thread.MIN_PRIORITY,
+ Thread.MAX_PRIORITY,
+ threadPriority));
+ }
+
return Executors.newFixedThreadPool(
numThreads,
new ExecutorThreadFactory.Builder()
- .setThreadPriority(Thread.MIN_PRIORITY)
+ .setThreadPriority(threadPriority)
.setPoolName("Flink-" + componentName)
.build());
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
