You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2021/09/03 10:22:44 UTC
[flink] branch release-1.14 updated: [FLINK-23654][runtime]
Splitting IO and future threadpools and adding advanced configurations
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new d3d9afd [FLINK-23654][runtime] Splitting IO and future threadpools and adding advanced configurations
d3d9afd is described below
commit d3d9afd0daac60e494feb31094c57c7b83c5972f
Author: Nicolas Raga <ra...@amazon.com>
AuthorDate: Fri Aug 20 11:10:03 2021 -0700
[FLINK-23654][runtime] Splitting IO and future threadpools and adding advanced configurations
[FLINK-23654] Use fixed thread pool for JobManagerSharedServices.ioExecutor
[FLINK-23654] Update keys of JobManagerOptions.JOB_MANAGER_FUTURE_POOL_SIZE and .JOB_MANAGER_IO_POOL_SIZE
The new key name for JobManagerOptions.JOB_MANAGER_FUTURE_POOL_SIZE is jobmanager.future-pool.size and
JobManagerOptions.JOB_MANAGER_IO_POOL_SIZE is jobmanager.io-pool.size.
This closes #16946.
---
docs/content.zh/docs/deployment/config.md | 4 +
docs/content/docs/deployment/config.md | 4 +
.../generated/all_jobmanager_section.html | 12 ++
.../generated/expert_jobmanager_section.html | 24 ++++
.../generated/job_manager_configuration.html | 12 ++
.../flink/annotation/docs/Documentation.java | 1 +
.../flink/configuration/JobManagerOptions.java | 26 ++++
.../flink/runtime/dispatcher/Dispatcher.java | 2 +-
.../JobMasterServiceLeadershipRunnerFactory.java | 2 +-
.../DefaultSlotPoolServiceSchedulerFactory.java | 9 +-
.../jobmaster/JobManagerSharedServices.java | 43 ++++--
.../apache/flink/runtime/jobmaster/JobMaster.java | 15 ++-
.../jobmaster/SlotPoolServiceSchedulerFactory.java | 5 +-
.../jobmaster/JobManagerSharedServicesTest.java | 145 +++++++++++++++++++++
.../TestingJobManagerSharedServicesBuilder.java | 20 ++-
15 files changed, 295 insertions(+), 29 deletions(-)
diff --git a/docs/content.zh/docs/deployment/config.md b/docs/content.zh/docs/deployment/config.md
index eb6a3b7..c161710 100644
--- a/docs/content.zh/docs/deployment/config.md
+++ b/docs/content.zh/docs/deployment/config.md
@@ -356,6 +356,10 @@ With the introduction of `state.backend.rocksdb.memory.managed` and `state.backe
{{< generated/expert_cluster_section >}}
+### Advanced JobManager Options
+
+{{< generated/expert_jobmanager_section >}}
+
### Advanced Scheduling Options
*These parameters can help with fine-tuning scheduling for specific situations.*
diff --git a/docs/content/docs/deployment/config.md b/docs/content/docs/deployment/config.md
index 9143fde..8195c58 100644
--- a/docs/content/docs/deployment/config.md
+++ b/docs/content/docs/deployment/config.md
@@ -358,6 +358,10 @@ With the introduction of `state.backend.rocksdb.memory.managed` and `state.backe
{{< generated/expert_cluster_section >}}
+### Advanced JobManager Options
+
+{{< generated/expert_jobmanager_section >}}
+
### Advanced Scheduling Options
*These parameters can help with fine-tuning scheduling for specific situations.*
diff --git a/docs/layouts/shortcodes/generated/all_jobmanager_section.html b/docs/layouts/shortcodes/generated/all_jobmanager_section.html
index 8a205ec..1e2b52c 100644
--- a/docs/layouts/shortcodes/generated/all_jobmanager_section.html
+++ b/docs/layouts/shortcodes/generated/all_jobmanager_section.html
@@ -45,6 +45,18 @@
<td>This option specifies how the job computation recovers from task failures. Accepted values are:<ul><li>'full': Restarts all tasks to recover the job.</li><li>'region': Restarts all tasks that could be affected by the task failure. More details can be found <a href="{{.Site.BaseURL}}{{.Site.LanguagePrefix}}/docs/ops/state/task_failure_recovery/#restart-pipelined-region-failover-strategy">here</a>.</li></ul></td>
</tr>
<tr>
+ <td><h5>jobmanager.future-pool.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The size of the future thread pool to execute future callbacks for all spawned JobMasters. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.io-pool.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+ </tr>
+ <tr>
<td><h5>jobmanager.retrieve-taskmanager-hostname</h5></td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
diff --git a/docs/layouts/shortcodes/generated/expert_jobmanager_section.html b/docs/layouts/shortcodes/generated/expert_jobmanager_section.html
new file mode 100644
index 0000000..c882966
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/expert_jobmanager_section.html
@@ -0,0 +1,24 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>jobmanager.future-pool.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The size of the future thread pool to execute future callbacks for all spawned JobMasters. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.io-pool.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/job_manager_configuration.html b/docs/layouts/shortcodes/generated/job_manager_configuration.html
index c8be41f..7419bec 100644
--- a/docs/layouts/shortcodes/generated/job_manager_configuration.html
+++ b/docs/layouts/shortcodes/generated/job_manager_configuration.html
@@ -51,6 +51,18 @@
<td>This option specifies how the job computation recovers from task failures. Accepted values are:<ul><li>'full': Restarts all tasks to recover the job.</li><li>'region': Restarts all tasks that could be affected by the task failure. More details can be found <a href="{{.Site.BaseURL}}{{.Site.LanguagePrefix}}/docs/ops/state/task_failure_recovery/#restart-pipelined-region-failover-strategy">here</a>.</li></ul></td>
</tr>
<tr>
+ <td><h5>jobmanager.future-pool.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The size of the future thread pool to execute future callbacks for all spawned JobMasters. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.io-pool.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
+ </tr>
+ <tr>
<td><h5>jobmanager.memory.enable-jvm-direct-memory-limit</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
diff --git a/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java b/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
index 9420f52..0a50318 100644
--- a/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
+++ b/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
@@ -91,6 +91,7 @@ public final class Documentation {
public static final String EXPERT_SECURITY_SSL = "expert_security_ssl";
public static final String EXPERT_ROCKSDB = "expert_rocksdb";
public static final String EXPERT_CLUSTER = "expert_cluster";
+ public static final String EXPERT_JOB_MANAGER = "expert_jobmanager";
public static final String ALL_JOB_MANAGER = "all_jobmanager";
public static final String ALL_TASK_MANAGER = "all_taskmanager";
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
index c108a44..0f51147 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
@@ -331,6 +331,32 @@ public class JobManagerOptions {
+ "JobManager could be faster, since no reverse DNS lookup is performed. "
+ "However, local input split assignment (such as for HDFS files) may be impacted.");
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_JOB_MANAGER,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Integer> JOB_MANAGER_FUTURE_POOL_SIZE =
+ key("jobmanager.future-pool.size")
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The size of the future thread pool to execute future callbacks for all spawned JobMasters. "
+ + "If no value is specified, then Flink defaults to the number of available CPU cores.");
+
+ @Documentation.Section({
+ Documentation.Sections.EXPERT_JOB_MANAGER,
+ Documentation.Sections.ALL_JOB_MANAGER
+ })
+ public static final ConfigOption<Integer> JOB_MANAGER_IO_POOL_SIZE =
+ key("jobmanager.io-pool.size")
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "The size of the IO thread pool to run blocking operations for all spawned JobMasters. "
+ + "This includes recovery and completion of checkpoints. "
+ + "Increase this value if you experience slow checkpoint operations when running many jobs. "
+ + "If no value is specified, then Flink defaults to the number of available CPU cores.");
+
/** The timeout in milliseconds for requesting a slot from Slot Pool. */
@Documentation.Section(Documentation.Sections.EXPERT_SCHEDULING)
public static final ConfigOption<Long> SLOT_REQUEST_TIMEOUT =
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index 710af29..e1bb919 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -520,7 +520,7 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
return Acknowledge.get();
},
- jobManagerSharedServices.getScheduledExecutorService());
+ jobManagerSharedServices.getIoExecutor());
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobMasterServiceLeadershipRunnerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobMasterServiceLeadershipRunnerFactory.java
index ea1f2c9..f025261 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobMasterServiceLeadershipRunnerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobMasterServiceLeadershipRunnerFactory.java
@@ -94,7 +94,7 @@ public enum JobMasterServiceLeadershipRunnerFactory implements JobManagerRunnerF
final DefaultJobMasterServiceFactory jobMasterServiceFactory =
new DefaultJobMasterServiceFactory(
- jobManagerServices.getScheduledExecutorService(),
+ jobManagerServices.getIoExecutor(),
rpcService,
jobMasterConfiguration,
jobGraph,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
index 9612698..7298fba 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
@@ -50,6 +50,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
+import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
/** Default {@link SlotPoolServiceSchedulerFactory} implementation. */
@@ -88,10 +89,10 @@ public final class DefaultSlotPoolServiceSchedulerFactory
public SchedulerNG createScheduler(
Logger log,
JobGraph jobGraph,
- ScheduledExecutorService scheduledExecutorService,
+ Executor ioExecutor,
Configuration configuration,
SlotPoolService slotPoolService,
- ScheduledExecutorService executorService,
+ ScheduledExecutorService futureExecutor,
ClassLoader userCodeLoader,
CheckpointRecoveryFactory checkpointRecoveryFactory,
Time rpcTimeout,
@@ -109,10 +110,10 @@ public final class DefaultSlotPoolServiceSchedulerFactory
return schedulerNGFactory.createInstance(
log,
jobGraph,
- scheduledExecutorService,
+ ioExecutor,
configuration,
slotPoolService,
- executorService,
+ futureExecutor,
userCodeLoader,
checkpointRecoveryFactory,
rpcTimeout,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java
index 543d9e1..4320ae9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServices.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.BlobWriter;
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
@@ -32,12 +33,17 @@ import org.apache.flink.runtime.shuffle.ShuffleMasterContextImpl;
import org.apache.flink.runtime.shuffle.ShuffleServiceLoader;
import org.apache.flink.runtime.util.Hardware;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import javax.annotation.Nonnull;
+import java.time.Duration;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -47,7 +53,11 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
public class JobManagerSharedServices {
- private final ScheduledExecutorService scheduledExecutorService;
+ private static final Duration SHUTDOWN_TIMEOUT = Duration.ofSeconds(10);
+
+ private final ScheduledExecutorService futureExecutor;
+
+ private final ExecutorService ioExecutor;
private final LibraryCacheManager libraryCacheManager;
@@ -56,19 +66,25 @@ public class JobManagerSharedServices {
@Nonnull private final BlobWriter blobWriter;
public JobManagerSharedServices(
- ScheduledExecutorService scheduledExecutorService,
+ ScheduledExecutorService futureExecutor,
+ ExecutorService ioExecutor,
LibraryCacheManager libraryCacheManager,
ShuffleMaster<?> shuffleMaster,
@Nonnull BlobWriter blobWriter) {
- this.scheduledExecutorService = checkNotNull(scheduledExecutorService);
+ this.futureExecutor = checkNotNull(futureExecutor);
+ this.ioExecutor = checkNotNull(ioExecutor);
this.libraryCacheManager = checkNotNull(libraryCacheManager);
this.shuffleMaster = checkNotNull(shuffleMaster);
this.blobWriter = blobWriter;
}
- public ScheduledExecutorService getScheduledExecutorService() {
- return scheduledExecutorService;
+ public ScheduledExecutorService getFutureExecutor() {
+ return futureExecutor;
+ }
+
+ public Executor getIoExecutor() {
+ return ioExecutor;
}
public LibraryCacheManager getLibraryCacheManager() {
@@ -97,7 +113,8 @@ public class JobManagerSharedServices {
Throwable exception = null;
try {
- scheduledExecutorService.shutdownNow();
+ ExecutorUtils.gracefulShutdown(
+ SHUTDOWN_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS, futureExecutor, ioExecutor);
} catch (Throwable t) {
exception = t;
}
@@ -147,10 +164,18 @@ public class JobManagerSharedServices {
failOnJvmMetaspaceOomError ? fatalErrorHandler : null,
checkClassLoaderLeak));
+ final int numberCPUCores = Hardware.getNumberCPUCores();
+ final int jobManagerFuturePoolSize =
+ config.getInteger(JobManagerOptions.JOB_MANAGER_FUTURE_POOL_SIZE, numberCPUCores);
final ScheduledExecutorService futureExecutor =
Executors.newScheduledThreadPool(
- Hardware.getNumberCPUCores(),
- new ExecutorThreadFactory("jobmanager-future"));
+ jobManagerFuturePoolSize, new ExecutorThreadFactory("jobmanager-future"));
+
+ final int jobManagerIoPoolSize =
+ config.getInteger(JobManagerOptions.JOB_MANAGER_IO_POOL_SIZE, numberCPUCores);
+ final ExecutorService ioExecutor =
+ Executors.newFixedThreadPool(
+ jobManagerIoPoolSize, new ExecutorThreadFactory("jobmanager-io"));
final ShuffleMasterContext shuffleMasterContext =
new ShuffleMasterContextImpl(config, fatalErrorHandler);
@@ -160,6 +185,6 @@ public class JobManagerSharedServices {
shuffleMaster.start();
return new JobManagerSharedServices(
- futureExecutor, libraryCacheManager, shuffleMaster, blobServer);
+ futureExecutor, ioExecutor, libraryCacheManager, shuffleMaster, blobServer);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index d21df82..8eb8c74 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -148,7 +148,9 @@ public class JobMaster extends PermanentlyFencedRpcEndpoint<JobMasterId>
private final HeartbeatServices heartbeatServices;
- private final ScheduledExecutorService scheduledExecutorService;
+ private final ScheduledExecutorService futureExecutor;
+
+ private final Executor ioExecutor;
private final OnCompletionActions jobCompletionActions;
@@ -276,7 +278,8 @@ public class JobMaster extends PermanentlyFencedRpcEndpoint<JobMasterId>
this.rpcTimeout = jobMasterConfiguration.getRpcTimeout();
this.highAvailabilityServices = checkNotNull(highAvailabilityService);
this.blobWriter = jobManagerSharedServices.getBlobWriter();
- this.scheduledExecutorService = jobManagerSharedServices.getScheduledExecutorService();
+ this.futureExecutor = jobManagerSharedServices.getFutureExecutor();
+ this.ioExecutor = jobManagerSharedServices.getIoExecutor();
this.jobCompletionActions = checkNotNull(jobCompletionActions);
this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
this.userCodeLoader = checkNotNull(userCodeLoader);
@@ -343,10 +346,10 @@ public class JobMaster extends PermanentlyFencedRpcEndpoint<JobMasterId>
slotPoolServiceSchedulerFactory.createScheduler(
log,
jobGraph,
- scheduledExecutorService,
+ ioExecutor,
jobMasterConfiguration.getConfiguration(),
slotPoolService,
- scheduledExecutorService,
+ futureExecutor,
userCodeLoader,
highAvailabilityServices.getCheckpointRecoveryFactory(),
rpcTimeout,
@@ -999,7 +1002,7 @@ public class JobMaster extends PermanentlyFencedRpcEndpoint<JobMasterId>
});
final ExecutionGraphInfo executionGraphInfo = schedulerNG.requestJob();
- scheduledExecutorService.execute(
+ futureExecutor.execute(
() -> jobCompletionActions.jobReachedGloballyTerminalState(executionGraphInfo));
}
}
@@ -1056,7 +1059,7 @@ public class JobMaster extends PermanentlyFencedRpcEndpoint<JobMasterId>
getFencingToken(),
resourceManagerAddress.getAddress(),
resourceManagerAddress.getResourceManagerId(),
- scheduledExecutorService);
+ futureExecutor);
resourceManagerConnection.start();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java
index ca029c0..b655602 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/SlotPoolServiceSchedulerFactory.java
@@ -36,6 +36,7 @@ import org.apache.flink.runtime.shuffle.ShuffleMaster;
import org.slf4j.Logger;
+import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
/** Factory for {@link SlotPoolService} and {@link SchedulerNG}. */
@@ -65,10 +66,10 @@ public interface SlotPoolServiceSchedulerFactory {
SchedulerNG createScheduler(
Logger log,
JobGraph jobGraph,
- ScheduledExecutorService scheduledExecutorService,
+ Executor ioExecutor,
Configuration configuration,
SlotPoolService slotPoolService,
- ScheduledExecutorService executorService,
+ ScheduledExecutorService futureExecutor,
ClassLoader userCodeLoader,
CheckpointRecoveryFactory checkpointRecoveryFactory,
Time rpcTimeout,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServicesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServicesTest.java
new file mode 100644
index 0000000..aa32242
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServicesTest.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.jobmaster;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.blob.VoidBlobStore;
+import org.apache.flink.runtime.util.Hardware;
+import org.apache.flink.runtime.util.TestingFatalErrorHandler;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.function.ThrowingRunnable;
+
+import org.junit.Test;
+
+import javax.annotation.Nonnull;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests auxiliary shared services created by {@link JobManagerSharedServices} and used by the
+ * {@link JobMaster}.
+ */
+public class JobManagerSharedServicesTest extends TestLogger {
+
+ private static final int CPU_CORES = Hardware.getNumberCPUCores();
+
+ @Test
+ public void testFutureExecutorNoConfiguration() throws Exception {
+ final Configuration config = new Configuration();
+
+ final JobManagerSharedServices jobManagerSharedServices =
+ buildJobManagerSharedServices(config);
+
+ try {
+ ScheduledExecutorService futureExecutor = jobManagerSharedServices.getFutureExecutor();
+
+ assertExecutorPoolSize(futureExecutor, CPU_CORES);
+ } finally {
+ jobManagerSharedServices.shutdown();
+ }
+ }
+
+ @Test
+ public void testFutureExecutorConfiguration() throws Exception {
+ final int futurePoolSize = 8;
+ final Configuration config = new Configuration();
+ config.setInteger(JobManagerOptions.JOB_MANAGER_FUTURE_POOL_SIZE, futurePoolSize);
+
+ final JobManagerSharedServices jobManagerSharedServices =
+ buildJobManagerSharedServices(config);
+
+ assertExecutorPoolSize(jobManagerSharedServices.getFutureExecutor(), futurePoolSize);
+
+ jobManagerSharedServices.shutdown();
+ }
+
+ @Test
+ public void testIoExecutorNoConfiguration() throws Exception {
+ final Configuration config = new Configuration();
+
+ final JobManagerSharedServices jobManagerSharedServices =
+ buildJobManagerSharedServices(config);
+
+ try {
+ assertExecutorPoolSize(jobManagerSharedServices.getIoExecutor(), CPU_CORES);
+ } finally {
+ jobManagerSharedServices.shutdown();
+ }
+ }
+
+ @Test
+ public void testIoExecutorConfiguration() throws Exception {
+ final int ioPoolSize = 5;
+ final Configuration config = new Configuration();
+ config.setInteger(JobManagerOptions.JOB_MANAGER_IO_POOL_SIZE, ioPoolSize);
+
+ final JobManagerSharedServices jobManagerSharedServices =
+ buildJobManagerSharedServices(config);
+
+ try {
+ assertExecutorPoolSize(jobManagerSharedServices.getIoExecutor(), ioPoolSize);
+ } finally {
+ jobManagerSharedServices.shutdown();
+ }
+ }
+
+ @Nonnull
+ private JobManagerSharedServices buildJobManagerSharedServices(Configuration configuration)
+ throws Exception {
+ return JobManagerSharedServices.fromConfiguration(
+ configuration,
+ new BlobServer(configuration, new VoidBlobStore()),
+ new TestingFatalErrorHandler());
+ }
+
+ private void assertExecutorPoolSize(Executor executor, int expectedPoolSize)
+ throws InterruptedException {
+ final CountDownLatch expectedPoolSizeLatch = new CountDownLatch(expectedPoolSize);
+ final int expectedPoolSizePlusOne = expectedPoolSize + 1;
+ final CountDownLatch expectedPoolSizePlusOneLatch =
+ new CountDownLatch(expectedPoolSizePlusOne);
+ final OneShotLatch releaseLatch = new OneShotLatch();
+
+ ThrowingRunnable<Exception> countsDown =
+ () -> {
+ expectedPoolSizePlusOneLatch.countDown();
+ expectedPoolSizeLatch.countDown();
+ // block the runnable to keep the thread occupied
+ releaseLatch.await();
+ };
+
+ for (int i = 0; i < expectedPoolSizePlusOne; i++) {
+ executor.execute(ThrowingRunnable.unchecked(countsDown));
+ }
+
+ // the expected pool size latch should complete since we expect to have enough threads
+ expectedPoolSizeLatch.await();
+ assertEquals(1, expectedPoolSizePlusOneLatch.getCount());
+
+ // unblock the runnables
+ releaseLatch.trigger();
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerSharedServicesBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerSharedServicesBuilder.java
index d623fb8..c1f50f6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerSharedServicesBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerSharedServicesBuilder.java
@@ -31,7 +31,9 @@ import java.util.concurrent.ScheduledExecutorService;
/** Builder for the {@link JobManagerSharedServices}. */
public class TestingJobManagerSharedServicesBuilder {
- private ScheduledExecutorService scheduledExecutorService;
+ private ScheduledExecutorService futureExecutor;
+
+ private ScheduledExecutorService ioExecutor;
private LibraryCacheManager libraryCacheManager;
@@ -40,15 +42,21 @@ public class TestingJobManagerSharedServicesBuilder {
private BlobWriter blobWriter;
public TestingJobManagerSharedServicesBuilder() {
- scheduledExecutorService = TestingUtils.defaultExecutor();
+ futureExecutor = TestingUtils.defaultExecutor();
+ ioExecutor = TestingUtils.defaultExecutor();
libraryCacheManager = ContextClassLoaderLibraryCacheManager.INSTANCE;
shuffleMaster = ShuffleTestUtils.DEFAULT_SHUFFLE_MASTER;
blobWriter = VoidBlobWriter.getInstance();
}
- public TestingJobManagerSharedServicesBuilder setScheduledExecutorService(
- ScheduledExecutorService scheduledExecutorService) {
- this.scheduledExecutorService = scheduledExecutorService;
+ public TestingJobManagerSharedServicesBuilder futureExecutor(
+ ScheduledExecutorService futureExecutor) {
+ this.futureExecutor = futureExecutor;
+ return this;
+ }
+
+ public TestingJobManagerSharedServicesBuilder ioExecutor(ScheduledExecutorService ioExecutor) {
+ this.ioExecutor = ioExecutor;
return this;
}
@@ -69,6 +77,6 @@ public class TestingJobManagerSharedServicesBuilder {
public JobManagerSharedServices build() {
return new JobManagerSharedServices(
- scheduledExecutorService, libraryCacheManager, shuffleMaster, blobWriter);
+ futureExecutor, ioExecutor, libraryCacheManager, shuffleMaster, blobWriter);
}
}