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 2018/02/06 16:53:18 UTC
[03/11] flink git commit: [FLINK-8502] [flip6] Remove
LibraryCacheManager from JobMaster
[FLINK-8502] [flip6] Remove LibraryCacheManager from JobMaster
This commit removes the LibraryCacheManager from the JobMaster since it is
no longer needed. The JobMaster is started with the correct user code class
loader and, thus, does not need the LibraryCacheManager.
This commit also corrects that the BlobServer is not closed by the
JobManagerServices#shutdown method.
This closes #5352.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2af2b734
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2af2b734
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2af2b734
Branch: refs/heads/master
Commit: 2af2b7347c7036bae4b1fb5f37782eccc6749911
Parents: 0f0a637
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Jan 24 13:42:41 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Tue Feb 6 12:15:31 2018 +0100
----------------------------------------------------------------------
.../flink/runtime/dispatcher/Dispatcher.java | 13 +++++---
.../dispatcher/StandaloneDispatcher.java | 2 ++
.../entrypoint/JobClusterEntrypoint.java | 3 ++
.../runtime/jobmaster/JobManagerRunner.java | 5 +--
.../runtime/jobmaster/JobManagerServices.java | 31 +++++------------
.../flink/runtime/jobmaster/JobMaster.java | 15 ---------
.../minicluster/MiniClusterJobDispatcher.java | 35 +++++++++++---------
.../runtime/dispatcher/DispatcherTest.java | 17 ++++++++--
.../flink/runtime/jobmaster/JobMasterTest.java | 9 -----
9 files changed, 59 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
----------------------------------------------------------------------
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 cb2682c..d13233a 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
@@ -89,6 +89,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
private final ResourceManagerGateway resourceManagerGateway;
private final JobManagerServices jobManagerServices;
private final HeartbeatServices heartbeatServices;
+ private final BlobServer blobServer;
private final MetricRegistry metricRegistry;
private final FatalErrorHandler fatalErrorHandler;
@@ -119,13 +120,15 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
this.configuration = Preconditions.checkNotNull(configuration);
this.highAvailabilityServices = Preconditions.checkNotNull(highAvailabilityServices);
this.resourceManagerGateway = Preconditions.checkNotNull(resourceManagerGateway);
- this.jobManagerServices = JobManagerServices.fromConfiguration(
- configuration,
- Preconditions.checkNotNull(blobServer));
this.heartbeatServices = Preconditions.checkNotNull(heartbeatServices);
+ this.blobServer = Preconditions.checkNotNull(blobServer);
this.metricRegistry = Preconditions.checkNotNull(metricRegistry);
this.fatalErrorHandler = Preconditions.checkNotNull(fatalErrorHandler);
+ this.jobManagerServices = JobManagerServices.fromConfiguration(
+ configuration,
+ this.blobServer);
+
this.submittedJobGraphStore = highAvailabilityServices.getSubmittedJobGraphStore();
this.runningJobsRegistry = highAvailabilityServices.getRunningJobsRegistry();
@@ -228,6 +231,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
getRpcService(),
highAvailabilityServices,
heartbeatServices,
+ blobServer,
jobManagerServices,
metricRegistry,
new DispatcherOnCompleteActions(jobGraph.getJobID()),
@@ -396,7 +400,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
@Override
public CompletableFuture<Integer> getBlobServerPort(Time timeout) {
- return CompletableFuture.completedFuture(jobManagerServices.blobServer.getPort());
+ return CompletableFuture.completedFuture(blobServer.getPort());
}
@Override
@@ -526,6 +530,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
RpcService rpcService,
HighAvailabilityServices highAvailabilityServices,
HeartbeatServices heartbeatServices,
+ BlobServer blobServer,
JobManagerServices jobManagerServices,
MetricRegistry metricRegistry,
OnCompletionActions onCompleteActions,
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
index b71b6bf..5533d59 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
@@ -75,6 +75,7 @@ public class StandaloneDispatcher extends Dispatcher {
RpcService rpcService,
HighAvailabilityServices highAvailabilityServices,
HeartbeatServices heartbeatServices,
+ BlobServer blobServer,
JobManagerServices jobManagerServices,
MetricRegistry metricRegistry,
OnCompletionActions onCompleteActions,
@@ -87,6 +88,7 @@ public class StandaloneDispatcher extends Dispatcher {
rpcService,
highAvailabilityServices,
heartbeatServices,
+ blobServer,
jobManagerServices,
metricRegistry,
onCompleteActions,
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
----------------------------------------------------------------------
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 1cf3de2..fe2636d 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
@@ -144,6 +144,7 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
highAvailabilityServices,
jobManagerServices,
heartbeatServices,
+ blobServer,
metricRegistry,
this,
jobMasterRestEndpoint.getRestAddress());
@@ -191,6 +192,7 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
HighAvailabilityServices highAvailabilityServices,
JobManagerServices jobManagerServices,
HeartbeatServices heartbeatServices,
+ BlobServer blobServer,
MetricRegistry metricRegistry,
FatalErrorHandler fatalErrorHandler,
@Nullable String restAddress) throws Exception {
@@ -204,6 +206,7 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
rpcService,
highAvailabilityServices,
heartbeatServices,
+ blobServer,
jobManagerServices,
metricRegistry,
new TerminatingOnCompleteActions(jobGraph.getJobID()),
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
index 5af6c67..e9d1c2d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
@@ -113,6 +114,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
final RpcService rpcService,
final HighAvailabilityServices haServices,
final HeartbeatServices heartbeatServices,
+ final BlobServer blobServer,
final JobManagerServices jobManagerServices,
final MetricRegistry metricRegistry,
final OnCompletionActions toNotifyOnComplete,
@@ -161,8 +163,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
haServices,
heartbeatServices,
jobManagerServices.executorService,
- jobManagerServices.blobServer,
- jobManagerServices.libraryCacheManager,
+ blobServer,
jobManagerServices.restartStrategyFactory,
jobManagerServices.rpcAskTimeout,
jobManagerMetrics,
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java
index 6abb1af..b46e0fa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java
@@ -30,13 +30,12 @@ import org.apache.flink.runtime.executiongraph.restart.RestartStrategyFactory;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
import org.apache.flink.runtime.util.Hardware;
import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.Preconditions;
-
-import scala.concurrent.duration.FiniteDuration;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import scala.concurrent.duration.FiniteDuration;
+
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -46,7 +45,6 @@ public class JobManagerServices {
public final ScheduledExecutorService executorService;
- public final BlobServer blobServer;
public final BlobLibraryCacheManager libraryCacheManager;
public final RestartStrategyFactory restartStrategyFactory;
@@ -55,24 +53,23 @@ public class JobManagerServices {
public JobManagerServices(
ScheduledExecutorService executorService,
- BlobServer blobServer,
BlobLibraryCacheManager libraryCacheManager,
RestartStrategyFactory restartStrategyFactory,
Time rpcAskTimeout) {
this.executorService = checkNotNull(executorService);
- this.blobServer = checkNotNull(blobServer);
this.libraryCacheManager = checkNotNull(libraryCacheManager);
this.restartStrategyFactory = checkNotNull(restartStrategyFactory);
this.rpcAskTimeout = checkNotNull(rpcAskTimeout);
}
/**
- *
+ * Shutdown the {@link JobMaster} services.
+ *
* <p>This method makes sure all services are closed or shut down, even when an exception occurred
* in the shutdown of one component. The first encountered exception is thrown, with successive
* exceptions added as suppressed exceptions.
- *
+ *
* @throws Exception The first Exception encountered during shutdown.
*/
public void shutdown() throws Exception {
@@ -85,16 +82,6 @@ public class JobManagerServices {
}
libraryCacheManager.shutdown();
- try {
- blobServer.close();
- }
- catch (Throwable t) {
- if (firstException == null) {
- firstException = t;
- } else {
- firstException.addSuppressed(t);
- }
- }
if (firstException != null) {
ExceptionUtils.rethrowException(firstException, "Error while shutting down JobManager services");
@@ -102,16 +89,15 @@ public class JobManagerServices {
}
// ------------------------------------------------------------------------
- // Creating the components from a configuration
+ // Creating the components from a configuration
// ------------------------------------------------------------------------
-
public static JobManagerServices fromConfiguration(
Configuration config,
BlobServer blobServer) throws Exception {
- Preconditions.checkNotNull(config);
- Preconditions.checkNotNull(blobServer);
+ checkNotNull(config);
+ checkNotNull(blobServer);
final String classLoaderResolveOrder =
config.getString(CoreOptions.CLASSLOADER_RESOLVE_ORDER);
@@ -139,7 +125,6 @@ public class JobManagerServices {
return new JobManagerServices(
futureExecutor,
- blobServer,
libraryCacheManager,
RestartStrategyFactory.createRestartStrategyFactory(config),
Time.of(timeout.length(), timeout.unit()));
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
----------------------------------------------------------------------
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 f137f39..7bcea19 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
@@ -38,7 +38,6 @@ import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.Execution;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -149,9 +148,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
/** Logical representation of the job. */
private final JobGraph jobGraph;
- /** Configuration of the JobManager. */
- private final Configuration configuration;
-
private final Time rpcTimeout;
/** Service to contend for and retrieve the leadership of JM and RM. */
@@ -160,12 +156,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
/** Blob server used across jobs. */
private final BlobServer blobServer;
- /** Blob library cache manager used across jobs. */
- private final BlobLibraryCacheManager libraryCacheManager;
-
- /** The metrics for the JobManager itself. */
- private final MetricGroup jobManagerMetricGroup;
-
/** The metrics for the job. */
private final MetricGroup jobMetricGroup;
@@ -218,7 +208,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
HeartbeatServices heartbeatServices,
ScheduledExecutorService executor,
BlobServer blobServer,
- BlobLibraryCacheManager libraryCacheManager,
RestartStrategyFactory restartStrategyFactory,
Time rpcAskTimeout,
@Nullable JobManagerMetricGroup jobManagerMetricGroup,
@@ -234,11 +223,9 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
this.resourceId = checkNotNull(resourceId);
this.jobGraph = checkNotNull(jobGraph);
- this.configuration = checkNotNull(configuration);
this.rpcTimeout = rpcAskTimeout;
this.highAvailabilityServices = checkNotNull(highAvailabilityService);
this.blobServer = checkNotNull(blobServer);
- this.libraryCacheManager = checkNotNull(libraryCacheManager);
this.executor = checkNotNull(executor);
this.jobCompletionActions = checkNotNull(jobCompletionActions);
this.errorHandler = checkNotNull(errorHandler);
@@ -260,10 +247,8 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
final JobID jid = jobGraph.getJobID();
if (jobManagerMetricGroup != null) {
- this.jobManagerMetricGroup = jobManagerMetricGroup;
this.jobMetricGroup = jobManagerMetricGroup.addJob(jobGraph);
} else {
- this.jobManagerMetricGroup = UnregisteredMetricGroups.createUnregisteredJobManagerMetricGroup();
this.jobMetricGroup = UnregisteredMetricGroups.createUnregisteredJobManagerJobMetricGroup();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java
index 7e49cef..e5ee9d7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobDispatcher.java
@@ -60,46 +60,49 @@ public class MiniClusterJobDispatcher {
// ------------------------------------------------------------------------
- /** lock to ensure that this dispatcher executes only one job at a time */
+ /** lock to ensure that this dispatcher executes only one job at a time. */
private final Object lock = new Object();
- /** the configuration with which the mini cluster was started */
+ /** the configuration with which the mini cluster was started. */
private final Configuration configuration;
- /** the RPC services to use by the job managers */
+ /** the RPC services to use by the job managers. */
private final RpcService[] rpcServices;
- /** services for discovery, leader election, and recovery */
+ /** services for discovery, leader election, and recovery. */
private final HighAvailabilityServices haServices;
- /** services for heartbeating */
+ /** services for heartbeating. */
private final HeartbeatServices heartbeatServices;
- /** all the services that the JobManager needs, such as BLOB service, factories, etc */
+ /** BlobServer for storing blobs. */
+ private final BlobServer blobServer;
+
+ /** all the services that the JobManager needs, such as BLOB service, factories, etc. */
private final JobManagerServices jobManagerServices;
- /** Registry for all metrics in the mini cluster */
+ /** Registry for all metrics in the mini cluster. */
private final MetricRegistry metricRegistry;
- /** The number of JobManagers to launch (more than one simulates a high-availability setup) */
+ /** The number of JobManagers to launch (more than one simulates a high-availability setup). */
private final int numJobManagers;
- /** The runner for the job and master. non-null if a job is currently running */
+ /** The runner for the job and master. non-null if a job is currently running. */
private volatile JobManagerRunner[] runners;
- /** flag marking the dispatcher as hut down */
+ /** flag marking the dispatcher as hut down. */
private volatile boolean shutdown;
/**
* Starts a mini cluster job dispatcher.
- *
+ *
* <p>The dispatcher kicks off one JobManager per job, a behavior similar to a
* non-highly-available setup.
- *
+ *
* @param config The configuration of the mini cluster
* @param haServices Access to the discovery, leader election, and recovery services
- *
+ *
* @throws Exception Thrown, if the services for the JobMaster could not be started.
*/
public MiniClusterJobDispatcher(
@@ -124,11 +127,11 @@ public class MiniClusterJobDispatcher {
*
* <p>The dispatcher may kick off more than one JobManager per job, thus simulating
* a highly-available setup.
- *
+ *
* @param config The configuration of the mini cluster
* @param haServices Access to the discovery, leader election, and recovery services
* @param numJobManagers The number of JobMasters to start for each job.
- *
+ *
* @throws Exception Thrown, if the services for the JobMaster could not be started.
*/
public MiniClusterJobDispatcher(
@@ -147,6 +150,7 @@ public class MiniClusterJobDispatcher {
this.rpcServices = rpcServices;
this.haServices = checkNotNull(haServices);
this.heartbeatServices = checkNotNull(heartbeatServices);
+ this.blobServer = checkNotNull(blobServer);
this.metricRegistry = checkNotNull(metricRegistry);
this.numJobManagers = numJobManagers;
@@ -280,6 +284,7 @@ public class MiniClusterJobDispatcher {
rpcServices[i],
haServices,
heartbeatServices,
+ blobServer,
jobManagerServices,
metricRegistry,
onCompletion,
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
index 1b399c0..0d58f62 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -391,15 +391,26 @@ public class DispatcherTest extends TestLogger {
RpcService rpcService,
HighAvailabilityServices highAvailabilityServices,
HeartbeatServices heartbeatServices,
+ BlobServer blobServer,
JobManagerServices jobManagerServices,
MetricRegistry metricRegistry,
OnCompletionActions onCompleteActions,
FatalErrorHandler fatalErrorHandler) throws Exception {
assertEquals(expectedJobId, jobGraph.getJobID());
- return new JobManagerRunner(resourceId, jobGraph, configuration, rpcService,
- highAvailabilityServices, heartbeatServices, jobManagerServices, metricRegistry,
- onCompleteActions, fatalErrorHandler, null);
+ return new JobManagerRunner(
+ resourceId,
+ jobGraph,
+ configuration,
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ blobServer,
+ jobManagerServices,
+ metricRegistry,
+ onCompleteActions,
+ fatalErrorHandler,
+ null);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/2af2b734/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index d20e1fe..386ed95 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -27,7 +27,6 @@ import org.apache.flink.runtime.blob.VoidBlobStore;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
-import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
import org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
@@ -113,10 +112,6 @@ public class JobMasterTest extends TestLogger {
heartbeatServices,
Executors.newScheduledThreadPool(1),
blobServer,
- new BlobLibraryCacheManager(
- blobServer,
- FlinkUserCodeClassLoaders.ResolveOrder.CHILD_FIRST,
- new String[0]),
new NoRestartStrategy.NoRestartStrategyFactory(),
testingTimeout,
null,
@@ -217,10 +212,6 @@ public class JobMasterTest extends TestLogger {
heartbeatServices,
Executors.newScheduledThreadPool(1),
blobServer,
- new BlobLibraryCacheManager(
- blobServer,
- FlinkUserCodeClassLoaders.ResolveOrder.CHILD_FIRST,
- new String[0]),
new NoRestartStrategy.NoRestartStrategyFactory(),
testingTimeout,
null,