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/03/28 15:24:30 UTC
[1/8] flink git commit: [hotfix] Create ExecutionGraph when JobMaster
is started
Repository: flink
Updated Branches:
refs/heads/master 7a21efbd4 -> af53f28e5
[hotfix] Create ExecutionGraph when JobMaster is started
The ExecutionGraph is not a final resource in the JobMaster. For example, it is necessary
to create a new ExecutionGraph when rescaling the job or when the JobMaster loses and
regains its leadership.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8d1545fe
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8d1545fe
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8d1545fe
Branch: refs/heads/master
Commit: 8d1545febfb7e1c02a9f5a948efc8b8eb4247742
Parents: 9a31dea
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Mar 23 15:43:20 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 28 17:01:11 2018 +0200
----------------------------------------------------------------------
.../flink/runtime/jobmaster/JobMaster.java | 244 ++++++++++++-------
.../DefaultJobManagerJobMetricGroupFactory.java | 7 +-
.../flink/runtime/jobmaster/JobMasterTest.java | 2 -
3 files changed, 156 insertions(+), 97 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8d1545fe/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 a9d8f29..9c36ddb 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
@@ -77,6 +77,7 @@ import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.query.KvStateLocation;
import org.apache.flink.runtime.query.KvStateLocationRegistry;
import org.apache.flink.runtime.query.UnknownKvStateLocation;
@@ -186,6 +187,8 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private final SlotPoolGateway slotPoolGateway;
+ private final RestartStrategy restartStrategy;
+
private final CompletableFuture<String> restAddressFuture;
private final String metricQueryServicePath;
@@ -212,6 +215,12 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private ExecutionGraph executionGraph;
@Nullable
+ private JobManagerJobStatusListener jobStatusListener;
+
+ @Nullable
+ private JobManagerJobMetricGroup jobManagerJobMetricGroup;
+
+ @Nullable
private String lastInternalSavepoint;
// ------------------------------------------------------------------------
@@ -270,7 +279,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
.deserializeValue(userCodeLoader)
.getRestartStrategy();
- final RestartStrategy restartStrategy = (restartStrategyConfiguration != null) ?
+ this.restartStrategy = (restartStrategyConfiguration != null) ?
RestartStrategyFactory.createRestartStrategy(restartStrategyConfiguration) :
jobManagerSharedServices.getRestartStrategyFactory().createRestartStrategy();
@@ -287,40 +296,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
this.slotPoolGateway = slotPool.getSelfGateway(SlotPoolGateway.class);
- this.executionGraph = ExecutionGraphBuilder.buildGraph(
- null,
- jobGraph,
- jobMasterConfiguration.getConfiguration(),
- scheduledExecutorService,
- scheduledExecutorService,
- slotPool.getSlotProvider(),
- userCodeLoader,
- highAvailabilityServices.getCheckpointRecoveryFactory(),
- rpcTimeout,
- restartStrategy,
- jobMetricGroupFactory.create(jobGraph),
- -1,
- blobServer,
- jobMasterConfiguration.getSlotRequestTimeout(),
- log);
-
- final CheckpointCoordinator checkpointCoordinator = executionGraph.getCheckpointCoordinator();
-
- if (checkpointCoordinator != null) {
- // check whether we find a valid checkpoint
- if (!checkpointCoordinator.restoreLatestCheckpointedState(
- executionGraph.getAllVertices(),
- false,
- false)) {
-
- // check whether we can restore from a savepoint
- tryRestoreExecutionGraphFromSavepoint(executionGraph, jobGraph.getSavepointRestoreSettings());
- }
- }
-
- // register self as job status change listener
- executionGraph.registerJobStatusListener(new JobManagerJobStatusListener());
-
this.registeredTaskManagers = new HashMap<>(4);
this.restAddressFuture = Optional.ofNullable(restAddress)
@@ -330,6 +305,10 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
this.metricQueryServicePath = metricQueryServicePath;
this.backPressureStatsTracker = checkNotNull(jobManagerSharedServices.getBackPressureStatsTracker());
this.lastInternalSavepoint = null;
+
+ this.jobManagerJobMetricGroup = jobMetricGroupFactory.create(jobGraph);
+ this.executionGraph = createAndRestoreExecutionGraph(jobManagerJobMetricGroup);
+ this.jobStatusListener = null;
}
//----------------------------------------------------------------------------------------------
@@ -474,32 +453,18 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
final ExecutionGraph currentExecutionGraph = executionGraph;
+ final JobManagerJobMetricGroup newJobManagerJobMetricGroup = jobMetricGroupFactory.create(jobGraph);
final ExecutionGraph newExecutionGraph;
try {
- newExecutionGraph = ExecutionGraphBuilder.buildGraph(
- null,
- jobGraph,
- jobMasterConfiguration.getConfiguration(),
- scheduledExecutorService,
- scheduledExecutorService,
- slotPool.getSlotProvider(),
- userCodeLoader,
- highAvailabilityServices.getCheckpointRecoveryFactory(),
- rpcTimeout,
- currentExecutionGraph.getRestartStrategy(),
- jobMetricGroupFactory.create(jobGraph),
- 1,
- blobServer,
- jobMasterConfiguration.getSlotRequestTimeout(),
- log);
+ newExecutionGraph = createExecutionGraph(newJobManagerJobMetricGroup);
} catch (JobExecutionException | JobException e) {
return FutureUtils.completedExceptionally(
new JobModificationException("Could not create rescaled ExecutionGraph.", e));
}
// 3. disable checkpoint coordinator to suppress subsequent checkpoints
- final CheckpointCoordinator checkpointCoordinator = executionGraph.getCheckpointCoordinator();
+ final CheckpointCoordinator checkpointCoordinator = currentExecutionGraph.getCheckpointCoordinator();
checkpointCoordinator.stopCheckpointScheduler();
// 4. take a savepoint
@@ -527,7 +492,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
// 5. suspend the current job
final CompletableFuture<JobStatus> terminationFuture = executionGraphFuture.thenComposeAsync(
(ExecutionGraph ignored) -> {
- currentExecutionGraph.suspend(new FlinkException("Job is being rescaled."));
+ suspendExecutionGraph(new FlinkException("Job is being rescaled."));
return currentExecutionGraph.getTerminationFuture();
},
getMainThreadExecutor());
@@ -546,13 +511,9 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
executionGraphFuture,
(Void ignored, ExecutionGraph restoredExecutionGraph) -> {
// check if the ExecutionGraph is still the same
- //noinspection ObjectEquality
if (executionGraph == currentExecutionGraph) {
- executionGraph = restoredExecutionGraph;
-
- // register self as job status change listener
- executionGraph.registerJobStatusListener(new JobManagerJobStatusListener());
-
+ clearExecutionGraphFields();
+ assignExecutionGraph(restoredExecutionGraph, newJobManagerJobMetricGroup);
scheduleExecutionGraph();
return Acknowledge.get();
@@ -1040,6 +1001,28 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
return Acknowledge.get();
}
+ setNewFencingToken(newJobMasterId);
+
+ log.info("Starting execution of job {} ({})", jobGraph.getName(), jobGraph.getJobID());
+
+ startJobMasterServices();
+ resetAndScheduleExecutionGraph();
+
+ return Acknowledge.get();
+ }
+
+ private void startJobMasterServices() throws Exception {
+ // start the slot pool make sure the slot pool now accepts messages for this leader
+ slotPool.start(getFencingToken(), getAddress());
+
+ // job is ready to go, try to establish connection with resource manager
+ // - activate leader retrieval for the resource manager
+ // - on notification of the leader, the connection will be established and
+ // the slot pool will start requesting slots
+ resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());
+ }
+
+ private void setNewFencingToken(JobMasterId newJobMasterId) {
if (getFencingToken() != null) {
log.info("Restarting old job with JobMasterId {}. The new JobMasterId is {}.", getFencingToken(), newJobMasterId);
@@ -1050,30 +1033,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
// set new leader id
setFencingToken(newJobMasterId);
-
- log.info("Starting execution of job {} ({})", jobGraph.getName(), jobGraph.getJobID());
-
- try {
- // start the slot pool make sure the slot pool now accepts messages for this leader
- log.debug("Staring SlotPool component");
- slotPool.start(getFencingToken(), getAddress());
-
- // job is ready to go, try to establish connection with resource manager
- // - activate leader retrieval for the resource manager
- // - on notification of the leader, the connection will be established and
- // the slot pool will start requesting slots
- resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());
- }
- catch (Throwable t) {
- log.error("Failed to start job {} ({})", jobGraph.getName(), jobGraph.getJobID(), t);
-
- throw new Exception("Could not start job execution: Failed to start JobMaster services.", t);
- }
-
- // start scheduling job in another thread
- scheduledExecutorService.execute(this::scheduleExecutionGraph);
-
- return Acknowledge.get();
}
/**
@@ -1102,8 +1061,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
log.warn("Failed to stop resource manager leader retriever when suspending.", t);
}
- // tell the execution graph (JobManager is still processing messages here)
- executionGraph.suspend(cause);
+ suspendAndClearExecutionGraphFields(cause);
// the slot pool stops receiving messages and clears its pooled slots
slotPoolGateway.suspend();
@@ -1114,10 +1072,46 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
return Acknowledge.get();
}
- /**
- * Schedules the execution of the current {@link ExecutionGraph}.
- */
+ private void assignExecutionGraph(
+ ExecutionGraph newExecutionGraph,
+ JobManagerJobMetricGroup newJobManagerJobMetricGroup) {
+ validateRunsInMainThread();
+ Preconditions.checkState(executionGraph.getState().isTerminalState());
+ Preconditions.checkState(jobManagerJobMetricGroup == null);
+
+ executionGraph = newExecutionGraph;
+ jobManagerJobMetricGroup = newJobManagerJobMetricGroup;
+ }
+
+ private void resetAndScheduleExecutionGraph() throws Exception {
+ validateRunsInMainThread();
+
+ final CompletableFuture<Void> executionGraphAssignedFuture;
+
+ if (executionGraph.getState() == JobStatus.CREATED) {
+ executionGraphAssignedFuture = CompletableFuture.completedFuture(null);
+ } else {
+ suspendAndClearExecutionGraphFields(new FlinkException("ExecutionGraph is being reset in order to be rescheduled."));
+ final JobManagerJobMetricGroup newJobManagerJobMetricGroup = jobMetricGroupFactory.create(jobGraph);
+ final ExecutionGraph newExecutionGraph = createAndRestoreExecutionGraph(newJobManagerJobMetricGroup);
+
+ executionGraphAssignedFuture = executionGraph.getTerminationFuture().handleAsync(
+ (JobStatus ignored, Throwable throwable) -> {
+ assignExecutionGraph(newExecutionGraph, newJobManagerJobMetricGroup);
+ return null;
+ },
+ getMainThreadExecutor());
+ }
+
+ executionGraphAssignedFuture.thenRun(this::scheduleExecutionGraph);
+ }
+
private void scheduleExecutionGraph() {
+ Preconditions.checkState(jobStatusListener == null);
+ // register self as job status change listener
+ jobStatusListener = new JobManagerJobStatusListener();
+ executionGraph.registerJobStatusListener(jobStatusListener);
+
try {
executionGraph.scheduleForExecution();
}
@@ -1126,6 +1120,68 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
}
}
+ private ExecutionGraph createAndRestoreExecutionGraph(JobManagerJobMetricGroup currentJobManagerJobMetricGroup) throws Exception {
+
+ ExecutionGraph newExecutionGraph = createExecutionGraph(currentJobManagerJobMetricGroup);
+
+ final CheckpointCoordinator checkpointCoordinator = newExecutionGraph.getCheckpointCoordinator();
+
+ if (checkpointCoordinator != null) {
+ // check whether we find a valid checkpoint
+ if (!checkpointCoordinator.restoreLatestCheckpointedState(
+ newExecutionGraph.getAllVertices(),
+ false,
+ false)) {
+
+ // check whether we can restore from a savepoint
+ tryRestoreExecutionGraphFromSavepoint(newExecutionGraph, jobGraph.getSavepointRestoreSettings());
+ }
+ }
+
+ return newExecutionGraph;
+ }
+
+ private ExecutionGraph createExecutionGraph(JobManagerJobMetricGroup currentJobManagerJobMetricGroup) throws JobExecutionException, JobException {
+ return ExecutionGraphBuilder.buildGraph(
+ null,
+ jobGraph,
+ jobMasterConfiguration.getConfiguration(),
+ scheduledExecutorService,
+ scheduledExecutorService,
+ slotPool.getSlotProvider(),
+ userCodeLoader,
+ highAvailabilityServices.getCheckpointRecoveryFactory(),
+ rpcTimeout,
+ restartStrategy,
+ currentJobManagerJobMetricGroup,
+ -1,
+ blobServer,
+ jobMasterConfiguration.getSlotRequestTimeout(),
+ log);
+ }
+
+ private void suspendAndClearExecutionGraphFields(Exception cause) {
+ suspendExecutionGraph(cause);
+ clearExecutionGraphFields();
+ }
+
+ private void suspendExecutionGraph(Exception cause) {
+ executionGraph.suspend(cause);
+
+ if (jobManagerJobMetricGroup != null) {
+ jobManagerJobMetricGroup.close();
+ }
+
+ if (jobStatusListener != null) {
+ jobStatusListener.stop();
+ }
+ }
+
+ private void clearExecutionGraphFields() {
+ jobManagerJobMetricGroup = null;
+ jobStatusListener = null;
+ }
+
/**
* Dispose the savepoint stored under the given path.
*
@@ -1496,6 +1552,8 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private class JobManagerJobStatusListener implements JobStatusListener {
+ private volatile boolean running = true;
+
@Override
public void jobStatusChanges(
final JobID jobId,
@@ -1503,8 +1561,14 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
final long timestamp,
final Throwable error) {
- // run in rpc thread to avoid concurrency
- runAsync(() -> jobStatusChanged(newJobStatus, timestamp, error));
+ if (running) {
+ // run in rpc thread to avoid concurrency
+ runAsync(() -> jobStatusChanged(newJobStatus, timestamp, error));
+ }
+ }
+
+ private void stop() {
+ running = false;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8d1545fe/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java
index 23cb8dc..cd94405 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.jobmaster.factories;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
@@ -26,8 +25,8 @@ import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
import javax.annotation.Nonnull;
/**
- * Implementation of {@link JobManagerJobMetricGroupFactory} which makes sure that there is always
- * at most one {@link JobManagerJobMetricGroup} for a given {@link JobID} registered.
+ * Default implementation of {@link JobManagerJobMetricGroupFactory} which creates for a given
+ * {@link JobGraph} a {@link JobManagerJobMetricGroup}.
*/
public class DefaultJobManagerJobMetricGroupFactory implements JobManagerJobMetricGroupFactory {
@@ -39,8 +38,6 @@ public class DefaultJobManagerJobMetricGroupFactory implements JobManagerJobMetr
@Override
public JobManagerJobMetricGroup create(@Nonnull JobGraph jobGraph) {
- jobManagerMetricGroup.removeJob(jobGraph.getJobID());
-
return jobManagerMetricGroup.addJob(jobGraph);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8d1545fe/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 fb2b96b..ef822a6 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
@@ -299,8 +299,6 @@ public class JobMasterTest extends TestLogger {
haServices,
new TestingJobManagerSharedServicesBuilder().build());
- jobMaster.start(JobMasterId.generate(), testingTimeout).get();
-
try {
// starting the JobMaster should have read the savepoint
final CompletedCheckpoint savepointCheckpoint = completedCheckpointStore.getLatestCheckpoint();
[4/8] flink git commit: [hotfix] Remove unused fields in JobMaster
Posted by tr...@apache.org.
[hotfix] Remove unused fields in JobMaster
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4de80cf8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4de80cf8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4de80cf8
Branch: refs/heads/master
Commit: 4de80cf8c04ff1fa7d5f4165d4c31d4f97317300
Parents: 8d1545f
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Mar 23 15:46:29 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 28 17:01:12 2018 +0200
----------------------------------------------------------------------
.../apache/flink/runtime/dispatcher/Dispatcher.java | 16 ++++------------
.../flink/runtime/jobmaster/JobManagerRunner.java | 10 ++--------
.../apache/flink/runtime/jobmaster/JobMaster.java | 15 ++-------------
.../flink/runtime/dispatcher/DispatcherTest.java | 8 ++------
.../runtime/dispatcher/MiniDispatcherTest.java | 5 +----
.../runtime/jobmaster/JobManagerRunnerTest.java | 4 +---
.../flink/runtime/jobmaster/JobMasterTest.java | 4 +---
7 files changed, 13 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4de80cf8/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 8ba84fb..2b353c0 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
@@ -263,9 +263,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
heartbeatServices,
blobServer,
jobManagerSharedServices,
- new DefaultJobManagerJobMetricGroupFactory(jobManagerMetricGroup),
- metricQueryServicePath,
- restAddress);
+ new DefaultJobManagerJobMetricGroupFactory(jobManagerMetricGroup));
jobManagerRunner.getResultFuture().whenCompleteAsync(
(ArchivedExecutionGraph archivedExecutionGraph, Throwable throwable) -> {
@@ -757,9 +755,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
HeartbeatServices heartbeatServices,
BlobServer blobServer,
JobManagerSharedServices jobManagerServices,
- JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
- @Nullable String metricQueryServicePath,
- @Nullable String restAddress) throws Exception;
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory) throws Exception;
}
/**
@@ -778,9 +774,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
HeartbeatServices heartbeatServices,
BlobServer blobServer,
JobManagerSharedServices jobManagerServices,
- JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
- @Nullable String metricQueryServicePath,
- @Nullable String restAddress) throws Exception {
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory) throws Exception {
return new JobManagerRunner(
resourceId,
jobGraph,
@@ -790,9 +784,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
heartbeatServices,
blobServer,
jobManagerServices,
- jobManagerJobMetricGroupFactory,
- metricQueryServicePath,
- restAddress);
+ jobManagerJobMetricGroupFactory);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4de80cf8/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 81f3a72..3b44ed7 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
@@ -45,8 +45,6 @@ import org.apache.flink.util.FlinkException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -107,9 +105,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
final HeartbeatServices heartbeatServices,
final BlobServer blobServer,
final JobManagerSharedServices jobManagerSharedServices,
- final JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
- @Nullable final String metricQueryServicePath,
- @Nullable final String restAddress) throws Exception {
+ final JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory) throws Exception {
this.resultFuture = new CompletableFuture<>();
this.terminationFuture = new CompletableFuture<>();
@@ -156,9 +152,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
jobManagerJobMetricGroupFactory,
this,
this,
- userCodeLoader,
- restAddress,
- metricQueryServicePath);
+ userCodeLoader);
}
catch (Throwable t) {
terminationFuture.completeExceptionally(t);
http://git-wip-us.apache.org/repos/asf/flink/blob/4de80cf8/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 9c36ddb..d116f03 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
@@ -189,10 +189,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private final RestartStrategy restartStrategy;
- private final CompletableFuture<String> restAddressFuture;
-
- private final String metricQueryServicePath;
-
// --------- BackPressure --------
private final BackPressureStatsTracker backPressureStatsTracker;
@@ -237,9 +233,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
JobManagerJobMetricGroupFactory jobMetricGroupFactory,
OnCompletionActions jobCompletionActions,
FatalErrorHandler errorHandler,
- ClassLoader userCodeLoader,
- @Nullable String restAddress,
- @Nullable String metricQueryServicePath) throws Exception {
+ ClassLoader userCodeLoader) throws Exception {
super(rpcService, AkkaRpcServiceUtils.createRandomName(JOB_MANAGER_NAME));
@@ -298,11 +292,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
this.registeredTaskManagers = new HashMap<>(4);
- this.restAddressFuture = Optional.ofNullable(restAddress)
- .map(CompletableFuture::completedFuture)
- .orElse(FutureUtils.completedExceptionally(new JobMasterException("The JobMaster has not been started with a REST endpoint.")));
-
- this.metricQueryServicePath = metricQueryServicePath;
this.backPressureStatsTracker = checkNotNull(jobManagerSharedServices.getBackPressureStatsTracker());
this.lastInternalSavepoint = null;
@@ -993,7 +982,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private Acknowledge startJobExecution(JobMasterId newJobMasterId) throws Exception {
validateRunsInMainThread();
- Preconditions.checkNotNull(newJobMasterId, "The new JobMasterId must not be null.");
+ checkNotNull(newJobMasterId, "The new JobMasterId must not be null.");
if (Objects.equals(getFencingToken(), newJobMasterId)) {
log.info("Already started the job execution with JobMasterId {}.", newJobMasterId);
http://git-wip-us.apache.org/repos/asf/flink/blob/4de80cf8/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 3d448ee..fb1ec63 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
@@ -480,9 +480,7 @@ public class DispatcherTest extends TestLogger {
HeartbeatServices heartbeatServices,
BlobServer blobServer,
JobManagerSharedServices jobManagerSharedServices,
- JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
- @Nullable String metricQueryServicePath,
- @Nullable String restAddress) throws Exception {
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory) throws Exception {
assertEquals(expectedJobId, jobGraph.getJobID());
return Dispatcher.DefaultJobManagerRunnerFactory.INSTANCE.createJobManagerRunner(
@@ -494,9 +492,7 @@ public class DispatcherTest extends TestLogger {
heartbeatServices,
blobServer,
jobManagerSharedServices,
- jobManagerJobMetricGroupFactory,
- metricQueryServicePath,
- restAddress);
+ jobManagerJobMetricGroupFactory);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4de80cf8/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
index 6fa6390..d58160e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
@@ -56,7 +56,6 @@ import org.junit.experimental.categories.Category;
import org.junit.rules.TemporaryFolder;
import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
import java.io.IOException;
import java.util.UUID;
@@ -284,9 +283,7 @@ public class MiniDispatcherTest extends TestLogger {
HeartbeatServices heartbeatServices,
BlobServer blobServer,
JobManagerSharedServices jobManagerSharedServices,
- JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
- @Nullable String metricQueryServicePath,
- @Nullable String restAddress) throws Exception {
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory) throws Exception {
jobGraphFuture.complete(jobGraph);
final JobManagerRunner mock = mock(JobManagerRunner.class);
http://git-wip-us.apache.org/repos/asf/flink/blob/4de80cf8/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
index d39bb9d..aa3f153 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
@@ -210,8 +210,6 @@ public class JobManagerRunnerTest extends TestLogger {
heartbeatServices,
blobServer,
jobManagerSharedServices,
- UnregisteredJobManagerJobMetricGroupFactory.INSTANCE,
- null,
- null);
+ UnregisteredJobManagerJobMetricGroupFactory.INSTANCE);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4de80cf8/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 ef822a6..7bf831c 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
@@ -414,9 +414,7 @@ public class JobMasterTest extends TestLogger {
UnregisteredJobManagerJobMetricGroupFactory.INSTANCE,
new NoOpOnCompletionActions(),
testingFatalErrorHandler,
- JobMasterTest.class.getClassLoader(),
- null,
- null);
+ JobMasterTest.class.getClassLoader());
}
/**
[2/8] flink git commit: [FLINK-8740] [metrics] Create new
JobManagerJobMetricGroup when creating a new ExecutionGraph
Posted by tr...@apache.org.
[FLINK-8740] [metrics] Create new JobManagerJobMetricGroup when creating a new ExecutionGraph
This closes #5755.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9a31deae
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9a31deae
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9a31deae
Branch: refs/heads/master
Commit: 9a31deae7ebdbda5a78f19ae8967e6b9d35834a4
Parents: 7a21efb
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Mar 23 14:08:49 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 28 17:01:11 2018 +0200
----------------------------------------------------------------------
.../flink/runtime/dispatcher/Dispatcher.java | 11 ++---
.../runtime/jobmaster/JobManagerRunner.java | 6 +--
.../flink/runtime/jobmaster/JobMaster.java | 12 ++---
.../DefaultJobManagerJobMetricGroupFactory.java | 46 ++++++++++++++++++++
.../JobManagerJobMetricGroupFactory.java | 38 ++++++++++++++++
...gisteredJobManagerJobMetricGroupFactory.java | 37 ++++++++++++++++
.../flink/runtime/jobmanager/JobManager.scala | 1 +
.../runtime/dispatcher/DispatcherTest.java | 6 +--
.../runtime/dispatcher/MiniDispatcherTest.java | 4 +-
.../runtime/jobmaster/JobManagerRunnerTest.java | 4 +-
.../flink/runtime/jobmaster/JobMasterTest.java | 6 ++-
11 files changed, 148 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/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 008d4dc..8ba84fb 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
@@ -42,6 +42,8 @@ import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
import org.apache.flink.runtime.jobmaster.JobNotFinishedException;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.jobmaster.RescalingBehaviour;
+import org.apache.flink.runtime.jobmaster.factories.DefaultJobManagerJobMetricGroupFactory;
+import org.apache.flink.runtime.jobmaster.factories.JobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.leaderelection.LeaderContender;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
import org.apache.flink.runtime.messages.Acknowledge;
@@ -50,7 +52,6 @@ import org.apache.flink.runtime.messages.webmonitor.ClusterOverview;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.messages.webmonitor.JobsOverview;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
-import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceOverview;
@@ -262,7 +263,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
heartbeatServices,
blobServer,
jobManagerSharedServices,
- jobManagerMetricGroup.addJob(jobGraph),
+ new DefaultJobManagerJobMetricGroupFactory(jobManagerMetricGroup),
metricQueryServicePath,
restAddress);
@@ -756,7 +757,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
HeartbeatServices heartbeatServices,
BlobServer blobServer,
JobManagerSharedServices jobManagerServices,
- JobManagerJobMetricGroup jobManagerJobMetricGroup,
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
@Nullable String metricQueryServicePath,
@Nullable String restAddress) throws Exception;
}
@@ -777,7 +778,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
HeartbeatServices heartbeatServices,
BlobServer blobServer,
JobManagerSharedServices jobManagerServices,
- JobManagerJobMetricGroup jobManagerJobMetricGroup,
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
@Nullable String metricQueryServicePath,
@Nullable String restAddress) throws Exception {
return new JobManagerRunner(
@@ -789,7 +790,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
heartbeatServices,
blobServer,
jobManagerServices,
- jobManagerJobMetricGroup,
+ jobManagerJobMetricGroupFactory,
metricQueryServicePath,
restAddress);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/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 cd2852b..81f3a72 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
@@ -32,10 +32,10 @@ import org.apache.flink.runtime.highavailability.RunningJobsRegistry;
import org.apache.flink.runtime.highavailability.RunningJobsRegistry.JobSchedulingStatus;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.OnCompletionActions;
+import org.apache.flink.runtime.jobmaster.factories.JobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.leaderelection.LeaderContender;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.util.AutoCloseableAsync;
@@ -107,7 +107,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
final HeartbeatServices heartbeatServices,
final BlobServer blobServer,
final JobManagerSharedServices jobManagerSharedServices,
- final JobManagerJobMetricGroup jobManagerJobMetricGroup,
+ final JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
@Nullable final String metricQueryServicePath,
@Nullable final String restAddress) throws Exception {
@@ -153,7 +153,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
jobManagerSharedServices,
heartbeatServices,
blobServer,
- jobManagerJobMetricGroup,
+ jobManagerJobMetricGroupFactory,
this,
this,
userCodeLoader,
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/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 bc18fab..a9d8f29 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
@@ -66,6 +66,7 @@ import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.jobmanager.OnCompletionActions;
import org.apache.flink.runtime.jobmanager.PartitionProducerDisposedException;
import org.apache.flink.runtime.jobmaster.exceptions.JobModificationException;
+import org.apache.flink.runtime.jobmaster.factories.JobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.jobmaster.message.ClassloadingProps;
import org.apache.flink.runtime.jobmaster.slotpool.SlotPool;
import org.apache.flink.runtime.jobmaster.slotpool.SlotPoolGateway;
@@ -76,7 +77,6 @@ import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
-import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.query.KvStateLocation;
import org.apache.flink.runtime.query.KvStateLocationRegistry;
import org.apache.flink.runtime.query.UnknownKvStateLocation;
@@ -165,7 +165,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private final BlobServer blobServer;
/** The metrics for the job. */
- private final JobManagerJobMetricGroup jobMetricGroup;
+ private final JobManagerJobMetricGroupFactory jobMetricGroupFactory;
/** The heartbeat manager with task managers. */
private final HeartbeatManager<AccumulatorReport, Void> taskManagerHeartbeatManager;
@@ -225,7 +225,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
JobManagerSharedServices jobManagerSharedServices,
HeartbeatServices heartbeatServices,
BlobServer blobServer,
- JobManagerJobMetricGroup jobMetricGroup,
+ JobManagerJobMetricGroupFactory jobMetricGroupFactory,
OnCompletionActions jobCompletionActions,
FatalErrorHandler errorHandler,
ClassLoader userCodeLoader,
@@ -246,7 +246,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
this.jobCompletionActions = checkNotNull(jobCompletionActions);
this.errorHandler = checkNotNull(errorHandler);
this.userCodeLoader = checkNotNull(userCodeLoader);
- this.jobMetricGroup = checkNotNull(jobMetricGroup);
+ this.jobMetricGroupFactory = checkNotNull(jobMetricGroupFactory);
this.taskManagerHeartbeatManager = heartbeatServices.createHeartbeatManagerSender(
resourceId,
@@ -298,7 +298,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
highAvailabilityServices.getCheckpointRecoveryFactory(),
rpcTimeout,
restartStrategy,
- jobMetricGroup,
+ jobMetricGroupFactory.create(jobGraph),
-1,
blobServer,
jobMasterConfiguration.getSlotRequestTimeout(),
@@ -488,7 +488,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
highAvailabilityServices.getCheckpointRecoveryFactory(),
rpcTimeout,
currentExecutionGraph.getRestartStrategy(),
- jobMetricGroup,
+ jobMetricGroupFactory.create(jobGraph),
1,
blobServer,
jobMasterConfiguration.getSlotRequestTimeout(),
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java
new file mode 100644
index 0000000..23cb8dc
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobManagerJobMetricGroupFactory.java
@@ -0,0 +1,46 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
+
+import javax.annotation.Nonnull;
+
+/**
+ * Implementation of {@link JobManagerJobMetricGroupFactory} which makes sure that there is always
+ * at most one {@link JobManagerJobMetricGroup} for a given {@link JobID} registered.
+ */
+public class DefaultJobManagerJobMetricGroupFactory implements JobManagerJobMetricGroupFactory {
+
+ private final JobManagerMetricGroup jobManagerMetricGroup;
+
+ public DefaultJobManagerJobMetricGroupFactory(@Nonnull JobManagerMetricGroup jobManagerMetricGroup) {
+ this.jobManagerMetricGroup = jobManagerMetricGroup;
+ }
+
+ @Override
+ public JobManagerJobMetricGroup create(@Nonnull JobGraph jobGraph) {
+ jobManagerMetricGroup.removeJob(jobGraph.getJobID());
+
+ return jobManagerMetricGroup.addJob(jobGraph);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/JobManagerJobMetricGroupFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/JobManagerJobMetricGroupFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/JobManagerJobMetricGroupFactory.java
new file mode 100644
index 0000000..a693bf2
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/JobManagerJobMetricGroupFactory.java
@@ -0,0 +1,38 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
+
+import javax.annotation.Nonnull;
+
+/**
+ * Factory interface for {@link JobManagerJobMetricGroup}.
+ */
+public interface JobManagerJobMetricGroupFactory {
+
+ /**
+ * Create a new {@link JobManagerJobMetricGroup}.
+ *
+ * @param jobGraph for which to create a new {@link JobManagerJobMetricGroup}.
+ * @return {@link JobManagerJobMetricGroup}
+ */
+ JobManagerJobMetricGroup create(@Nonnull JobGraph jobGraph);
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/UnregisteredJobManagerJobMetricGroupFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/UnregisteredJobManagerJobMetricGroupFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/UnregisteredJobManagerJobMetricGroupFactory.java
new file mode 100644
index 0000000..20c7e88
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/UnregisteredJobManagerJobMetricGroupFactory.java
@@ -0,0 +1,37 @@
+/*
+ * 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.factories;
+
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+
+import javax.annotation.Nonnull;
+
+/**
+ * {@link JobManagerJobMetricGroupFactory} which returns an unregistered {@link JobManagerJobMetricGroup}.
+ */
+public enum UnregisteredJobManagerJobMetricGroupFactory implements JobManagerJobMetricGroupFactory {
+ INSTANCE {
+ @Override
+ public JobManagerJobMetricGroup create(@Nonnull JobGraph jobGraph) {
+ return UnregisteredMetricGroups.createUnregisteredJobManagerJobMetricGroup();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index b9529de..2674332 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -1769,6 +1769,7 @@ class JobManager(
val futures = for ((jobID, (eg, jobInfo)) <- currentJobs) yield {
future {
eg.suspend(cause)
+ jobManagerMetricGroup.removeJob(eg.getJobID)
jobInfo.notifyNonDetachedClients(
decorateMessage(
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/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 8ea686b..3d448ee 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
@@ -43,11 +43,11 @@ import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
import org.apache.flink.runtime.jobmaster.JobManagerRunner;
import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.jobmaster.factories.JobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
-import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
@@ -480,7 +480,7 @@ public class DispatcherTest extends TestLogger {
HeartbeatServices heartbeatServices,
BlobServer blobServer,
JobManagerSharedServices jobManagerSharedServices,
- JobManagerJobMetricGroup jobManagerJobMetricGroup,
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
@Nullable String metricQueryServicePath,
@Nullable String restAddress) throws Exception {
assertEquals(expectedJobId, jobGraph.getJobID());
@@ -494,7 +494,7 @@ public class DispatcherTest extends TestLogger {
heartbeatServices,
blobServer,
jobManagerSharedServices,
- jobManagerJobMetricGroup,
+ jobManagerJobMetricGroupFactory,
metricQueryServicePath,
restAddress);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
index 651200f..6fa6390 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java
@@ -34,8 +34,8 @@ import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobmaster.JobManagerRunner;
import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.jobmaster.factories.JobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
-import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
@@ -284,7 +284,7 @@ public class MiniDispatcherTest extends TestLogger {
HeartbeatServices heartbeatServices,
BlobServer blobServer,
JobManagerSharedServices jobManagerSharedServices,
- JobManagerJobMetricGroup jobManagerJobMetricGroup,
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
@Nullable String metricQueryServicePath,
@Nullable String restAddress) throws Exception {
jobGraphFuture.complete(jobGraph);
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
index 1d7f090..d39bb9d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerTest.java
@@ -30,9 +30,9 @@ import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobmaster.factories.UnregisteredJobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
-import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
import org.apache.flink.runtime.rpc.TestingRpcService;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
@@ -210,7 +210,7 @@ public class JobManagerRunnerTest extends TestLogger {
heartbeatServices,
blobServer,
jobManagerSharedServices,
- UnregisteredMetricGroups.createUnregisteredJobManagerJobMetricGroup(),
+ UnregisteredJobManagerJobMetricGroupFactory.INSTANCE,
null,
null);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9a31deae/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 ed5a894..fb2b96b 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
@@ -45,9 +45,9 @@ import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
import org.apache.flink.runtime.jobmanager.OnCompletionActions;
+import org.apache.flink.runtime.jobmaster.factories.UnregisteredJobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.registration.RegistrationResponse;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
@@ -299,6 +299,8 @@ public class JobMasterTest extends TestLogger {
haServices,
new TestingJobManagerSharedServicesBuilder().build());
+ jobMaster.start(JobMasterId.generate(), testingTimeout).get();
+
try {
// starting the JobMaster should have read the savepoint
final CompletedCheckpoint savepointCheckpoint = completedCheckpointStore.getLatestCheckpoint();
@@ -411,7 +413,7 @@ public class JobMasterTest extends TestLogger {
jobManagerSharedServices,
fastHeartbeatServices,
blobServer,
- UnregisteredMetricGroups.createUnregisteredJobManagerJobMetricGroup(),
+ UnregisteredJobManagerJobMetricGroupFactory.INSTANCE,
new NoOpOnCompletionActions(),
testingFatalErrorHandler,
JobMasterTest.class.getClassLoader(),
[7/8] flink git commit: [FLINK-8887] Wait for JobMaster leader
election in Dispatcher
Posted by tr...@apache.org.
[FLINK-8887] Wait for JobMaster leader election in Dispatcher
Before sending requests from the Dispatcher to the JobMasters, the Dispatcher must
wait until the respective JobMaster has gained leadership. Otherwise we might risk
that the messages are ignored because no fencing token was set.
This is solved by letting the JobManagerRunner expose a CompletableFuture<JobMasterGateway>
which is only completed after the JobMaster has gained leadership. The future is cleared
once the leadership is revoked.
This closes #5767.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c0dddc1a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c0dddc1a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c0dddc1a
Branch: refs/heads/master
Commit: c0dddc1a6e3eef8ded1963e61ee4a8f8ecf66475
Parents: e6b1d5c
Author: Till Rohrmann <tr...@apache.org>
Authored: Mon Mar 26 17:55:10 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 28 17:01:13 2018 +0200
----------------------------------------------------------------------
.../flink/client/program/MiniClusterClient.java | 51 ++----
.../flink/runtime/dispatcher/Dispatcher.java | 175 +++++++++++--------
.../nonha/embedded/EmbeddedLeaderService.java | 20 ++-
.../runtime/jobmaster/JobManagerRunner.java | 121 +++++++------
.../runtime/dispatcher/DispatcherTest.java | 38 +++-
5 files changed, 233 insertions(+), 172 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c0dddc1a/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
index 802622e..a135359 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
@@ -20,27 +20,18 @@ package org.apache.flink.client.program;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobSubmissionResult;
-import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.clusterframework.messages.GetClusterStatusResponse;
-import org.apache.flink.runtime.concurrent.FutureUtils;
-import org.apache.flink.runtime.concurrent.ScheduledExecutor;
-import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter;
import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.minicluster.MiniCluster;
-import org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcException;
-import org.apache.flink.runtime.rpc.exceptions.FencingTokenException;
-import org.apache.flink.runtime.util.ExecutorThreadFactory;
import org.apache.flink.runtime.util.LeaderConnectionInfo;
import org.apache.flink.runtime.util.LeaderRetrievalUtils;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.FlinkException;
import org.apache.flink.util.OptionalFailure;
import org.apache.flink.util.SerializedValue;
@@ -54,9 +45,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.function.Supplier;
/**
* Client to interact with a {@link MiniCluster}.
@@ -64,10 +52,8 @@ import java.util.function.Supplier;
public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClusterId> {
private final MiniCluster miniCluster;
- private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(4, new ExecutorThreadFactory("Flink-MiniClusterClient"));
- private final ScheduledExecutor scheduledExecutor = new ScheduledExecutorServiceAdapter(scheduledExecutorService);
- public MiniClusterClient(@Nonnull Configuration configuration, @Nonnull MiniCluster miniCluster) throws Exception {
+ public MiniClusterClient(@Nonnull Configuration configuration, @Nonnull MiniCluster miniCluster) {
super(configuration, miniCluster.getHighAvailabilityServices(), true);
this.miniCluster = miniCluster;
@@ -76,7 +62,6 @@ public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClust
@Override
public void shutdown() throws Exception {
super.shutdown();
- scheduledExecutorService.shutdown();
}
@Override
@@ -104,32 +89,32 @@ public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClust
@Override
public void cancel(JobID jobId) throws Exception {
- guardWithSingleRetry(() -> miniCluster.cancelJob(jobId), scheduledExecutor);
+ miniCluster.cancelJob(jobId).get();
}
@Override
public String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory) throws Exception {
- return guardWithSingleRetry(() -> miniCluster.triggerSavepoint(jobId, savepointDirectory, true), scheduledExecutor).get();
+ return miniCluster.triggerSavepoint(jobId, savepointDirectory, true).get();
}
@Override
public void stop(JobID jobId) throws Exception {
- guardWithSingleRetry(() -> miniCluster.stopJob(jobId), scheduledExecutor).get();
+ miniCluster.stopJob(jobId).get();
}
@Override
- public CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory) throws FlinkException {
- return guardWithSingleRetry(() -> miniCluster.triggerSavepoint(jobId, savepointDirectory, false), scheduledExecutor);
+ public CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory) {
+ return miniCluster.triggerSavepoint(jobId, savepointDirectory, false);
}
@Override
- public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath) throws FlinkException {
- return guardWithSingleRetry(() -> miniCluster.disposeSavepoint(savepointPath), scheduledExecutor);
+ public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath) {
+ return miniCluster.disposeSavepoint(savepointPath);
}
@Override
- public CompletableFuture<Collection<JobStatusMessage>> listJobs() throws Exception {
- return guardWithSingleRetry(miniCluster::listJobs, scheduledExecutor);
+ public CompletableFuture<Collection<JobStatusMessage>> listJobs() {
+ return miniCluster.listJobs();
}
@Override
@@ -139,7 +124,7 @@ public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClust
@Override
public Map<String, OptionalFailure<Object>> getAccumulators(JobID jobID, ClassLoader loader) throws Exception {
- AccessExecutionGraph executionGraph = guardWithSingleRetry(() -> miniCluster.getExecutionGraph(jobID), scheduledExecutor).get();
+ AccessExecutionGraph executionGraph = miniCluster.getExecutionGraph(jobID).get();
Map<String, SerializedValue<OptionalFailure<Object>>> accumulatorsSerialized = executionGraph.getAccumulatorsSerialized();
Map<String, OptionalFailure<Object>> result = new HashMap<>(accumulatorsSerialized.size());
for (Map.Entry<String, SerializedValue<OptionalFailure<Object>>> acc : accumulatorsSerialized.entrySet()) {
@@ -150,7 +135,7 @@ public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClust
@Override
public CompletableFuture<JobStatus> getJobStatus(JobID jobId) {
- return guardWithSingleRetry(() -> miniCluster.getJobStatus(jobId), scheduledExecutor);
+ return miniCluster.getJobStatus(jobId);
}
@Override
@@ -202,16 +187,4 @@ public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClust
enum MiniClusterId {
INSTANCE
}
-
- private static <X> CompletableFuture<X> guardWithSingleRetry(Supplier<CompletableFuture<X>> operation, ScheduledExecutor executor) {
- return FutureUtils.retryWithDelay(
- operation,
- 1,
- Time.milliseconds(500),
- throwable -> {
- Throwable actualException = ExceptionUtils.stripCompletionException(throwable);
- return actualException instanceof FencingTokenException || actualException instanceof AkkaRpcException;
- },
- executor);
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0dddc1a/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 2b353c0..2b96a09 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
@@ -39,6 +39,7 @@ import org.apache.flink.runtime.jobmanager.SubmittedJobGraph;
import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
import org.apache.flink.runtime.jobmaster.JobManagerRunner;
import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
import org.apache.flink.runtime.jobmaster.JobNotFinishedException;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.jobmaster.RescalingBehaviour;
@@ -64,6 +65,7 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
+import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.IOException;
@@ -75,9 +77,11 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.function.Function;
import java.util.stream.Collectors;
/**
@@ -341,35 +345,25 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
@Override
public CompletableFuture<Acknowledge> cancelJob(JobID jobId, Time timeout) {
- JobManagerRunner jobManagerRunner = jobManagerRunners.get(jobId);
+ final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getJobMasterGatewayFuture(jobId);
- if (jobManagerRunner == null) {
- return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
- } else {
- return jobManagerRunner.getJobManagerGateway().cancel(timeout);
- }
+ return jobMasterGatewayFuture.thenCompose((JobMasterGateway jobMasterGateway) -> jobMasterGateway.cancel(timeout));
}
@Override
public CompletableFuture<Acknowledge> stopJob(JobID jobId, Time timeout) {
- JobManagerRunner jobManagerRunner = jobManagerRunners.get(jobId);
+ final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getJobMasterGatewayFuture(jobId);
- if (jobManagerRunner == null) {
- return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
- } else {
- return jobManagerRunner.getJobManagerGateway().stop(timeout);
- }
+ return jobMasterGatewayFuture.thenCompose((JobMasterGateway jobMasterGateway) -> jobMasterGateway.stop(timeout));
}
@Override
public CompletableFuture<Acknowledge> rescaleJob(JobID jobId, int newParallelism, RescalingBehaviour rescalingBehaviour, Time timeout) {
- JobManagerRunner jobManagerRunner = jobManagerRunners.get(jobId);
+ final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getJobMasterGatewayFuture(jobId);
- if (jobManagerRunner == null) {
- return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
- } else {
- return jobManagerRunner.getJobManagerGateway().rescaleJob(newParallelism, rescalingBehaviour, timeout);
- }
+ return jobMasterGatewayFuture.thenCompose(
+ (JobMasterGateway jobMasterGateway) ->
+ jobMasterGateway.rescaleJob(newParallelism, rescalingBehaviour, timeout));
}
@Override
@@ -385,15 +379,12 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
public CompletableFuture<ClusterOverview> requestClusterOverview(Time timeout) {
CompletableFuture<ResourceOverview> taskManagerOverviewFuture = resourceManagerGateway.requestResourceOverview(timeout);
- ArrayList<CompletableFuture<JobStatus>> jobStatus = new ArrayList<>(jobManagerRunners.size());
+ final List<CompletableFuture<Optional<JobStatus>>> optionalJobInformation = queryJobMastersForInformation(
+ (JobMasterGateway jobMasterGateway) -> jobMasterGateway.requestJobStatus(timeout));
- for (Map.Entry<JobID, JobManagerRunner> jobManagerRunnerEntry : jobManagerRunners.entrySet()) {
- CompletableFuture<JobStatus> jobStatusFuture = jobManagerRunnerEntry.getValue().getJobManagerGateway().requestJobStatus(timeout);
-
- jobStatus.add(jobStatusFuture);
- }
+ CompletableFuture<Collection<Optional<JobStatus>>> allOptionalJobsFuture = FutureUtils.combineAll(optionalJobInformation);
- CompletableFuture<Collection<JobStatus>> allJobsFuture = FutureUtils.combineAll(jobStatus);
+ CompletableFuture<Collection<JobStatus>> allJobsFuture = allOptionalJobsFuture.thenApply(this::flattenOptionalCollection);
final JobsOverview completedJobsOverview = archivedExecutionGraphStore.getStoredJobsOverview();
@@ -403,76 +394,79 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
final JobsOverview allJobsOverview = JobsOverview.create(runningJobsStatus).combine(completedJobsOverview);
return new ClusterOverview(resourceOverview, allJobsOverview);
});
-
}
@Override
public CompletableFuture<MultipleJobsDetails> requestMultipleJobDetails(Time timeout) {
- final int numberJobsRunning = jobManagerRunners.size();
+ List<CompletableFuture<Optional<JobDetails>>> individualOptionalJobDetails = queryJobMastersForInformation(
+ (JobMasterGateway jobMasterGateway) -> jobMasterGateway.requestJobDetails(timeout));
- ArrayList<CompletableFuture<JobDetails>> individualJobDetails = new ArrayList<>(numberJobsRunning);
-
- for (JobManagerRunner jobManagerRunner : jobManagerRunners.values()) {
- individualJobDetails.add(jobManagerRunner.getJobManagerGateway().requestJobDetails(timeout));
- }
+ CompletableFuture<Collection<Optional<JobDetails>>> optionalCombinedJobDetails = FutureUtils.combineAll(
+ individualOptionalJobDetails);
- CompletableFuture<Collection<JobDetails>> combinedJobDetails = FutureUtils.combineAll(individualJobDetails);
+ CompletableFuture<Collection<JobDetails>> combinedJobDetails = optionalCombinedJobDetails.thenApply(this::flattenOptionalCollection);
final Collection<JobDetails> completedJobDetails = archivedExecutionGraphStore.getAvailableJobDetails();
return combinedJobDetails.thenApply(
(Collection<JobDetails> runningJobDetails) -> {
final Collection<JobDetails> allJobDetails = new ArrayList<>(completedJobDetails.size() + runningJobDetails.size());
+
allJobDetails.addAll(runningJobDetails);
allJobDetails.addAll(completedJobDetails);
+
return new MultipleJobsDetails(allJobDetails);
});
}
@Override
public CompletableFuture<JobStatus> requestJobStatus(JobID jobId, Time timeout) {
- final JobManagerRunner jobManagerRunner = jobManagerRunners.get(jobId);
- if (jobManagerRunner != null) {
- return jobManagerRunner.getJobManagerGateway().requestJobStatus(timeout);
- } else {
- final JobDetails jobDetails = archivedExecutionGraphStore.getAvailableJobDetails(jobId);
+ final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getJobMasterGatewayFuture(jobId);
- if (jobDetails != null) {
- return CompletableFuture.completedFuture(jobDetails.getStatus());
- } else {
- return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
- }
- }
+ final CompletableFuture<JobStatus> jobStatusFuture = jobMasterGatewayFuture.thenCompose(
+ (JobMasterGateway jobMasterGateway) -> jobMasterGateway.requestJobStatus(timeout));
+
+ return jobStatusFuture.exceptionally(
+ (Throwable throwable) -> {
+ final JobDetails jobDetails = archivedExecutionGraphStore.getAvailableJobDetails(jobId);
+
+ // check whether it is a completed job
+ if (jobDetails == null) {
+ throw new CompletionException(ExceptionUtils.stripCompletionException(throwable));
+ } else {
+ return jobDetails.getStatus();
+ }
+ });
}
@Override
public CompletableFuture<OperatorBackPressureStatsResponse> requestOperatorBackPressureStats(
- final JobID jobId, final JobVertexID jobVertexId) {
- final JobManagerRunner jobManagerRunner = jobManagerRunners.get(jobId);
- if (jobManagerRunner == null) {
- return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
- } else {
- return jobManagerRunner.getJobManagerGateway().requestOperatorBackPressureStats(jobVertexId);
- }
+ final JobID jobId,
+ final JobVertexID jobVertexId) {
+ final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getJobMasterGatewayFuture(jobId);
+
+ return jobMasterGatewayFuture.thenCompose((JobMasterGateway jobMasterGateway) -> jobMasterGateway.requestOperatorBackPressureStats(jobVertexId));
}
@Override
public CompletableFuture<ArchivedExecutionGraph> requestJob(JobID jobId, Time timeout) {
- final JobManagerRunner jobManagerRunner = jobManagerRunners.get(jobId);
+ final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getJobMasterGatewayFuture(jobId);
- if (jobManagerRunner == null) {
- final ArchivedExecutionGraph serializableExecutionGraph = archivedExecutionGraphStore.get(jobId);
+ final CompletableFuture<ArchivedExecutionGraph> archivedExecutionGraphFuture = jobMasterGatewayFuture.thenCompose(
+ (JobMasterGateway jobMasterGateway) -> jobMasterGateway.requestJob(timeout));
- // check whether it is a completed job
- if (serializableExecutionGraph == null) {
- return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
- } else {
- return CompletableFuture.completedFuture(serializableExecutionGraph);
- }
- } else {
- return jobManagerRunner.getJobManagerGateway().requestJob(timeout);
- }
+ return archivedExecutionGraphFuture.exceptionally(
+ (Throwable throwable) -> {
+ final ArchivedExecutionGraph serializableExecutionGraph = archivedExecutionGraphStore.get(jobId);
+
+ // check whether it is a completed job
+ if (serializableExecutionGraph == null) {
+ throw new CompletionException(ExceptionUtils.stripCompletionException(throwable));
+ } else {
+ return serializableExecutionGraph;
+ }
+ });
}
@Override
@@ -517,13 +511,11 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
final String targetDirectory,
final boolean cancelJob,
final Time timeout) {
- if (jobManagerRunners.containsKey(jobId)) {
- return jobManagerRunners.get(jobId)
- .getJobManagerGateway()
- .triggerSavepoint(targetDirectory, cancelJob, timeout);
- } else {
- return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
- }
+ final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getJobMasterGatewayFuture(jobId);
+
+ return jobMasterGatewayFuture.thenCompose(
+ (JobMasterGateway jobMasterGateway) ->
+ jobMasterGateway.triggerSavepoint(targetDirectory, cancelJob, timeout));
}
/**
@@ -644,6 +636,49 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
jobManagerRunnerTerminationFuture));
}
+ private CompletableFuture<JobMasterGateway> getJobMasterGatewayFuture(JobID jobId) {
+ final JobManagerRunner jobManagerRunner = jobManagerRunners.get(jobId);
+
+ if (jobManagerRunner == null) {
+ return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));
+ } else {
+ final CompletableFuture<JobMasterGateway> leaderGatewayFuture = jobManagerRunner.getLeaderGatewayFuture();
+ return leaderGatewayFuture.thenApplyAsync(
+ (JobMasterGateway jobMasterGateway) -> {
+ // check whether the retrieved JobMasterGateway belongs still to a running JobMaster
+ if (jobManagerRunners.containsKey(jobId)) {
+ return jobMasterGateway;
+ } else {
+ throw new CompletionException(new FlinkJobNotFoundException(jobId));
+ }
+ },
+ getMainThreadExecutor());
+ }
+ }
+
+ private <T> List<T> flattenOptionalCollection(Collection<Optional<T>> optionalCollection) {
+ return optionalCollection.stream().filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());
+ }
+
+ @Nonnull
+ private <T> List<CompletableFuture<Optional<T>>> queryJobMastersForInformation(Function<JobMasterGateway, CompletableFuture<T>> queryFunction) {
+ final int numberJobsRunning = jobManagerRunners.size();
+
+ ArrayList<CompletableFuture<Optional<T>>> optionalJobInformation = new ArrayList<>(
+ numberJobsRunning);
+
+ for (JobID jobId : jobManagerRunners.keySet()) {
+ final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getJobMasterGatewayFuture(jobId);
+
+ final CompletableFuture<Optional<T>> optionalRequest = jobMasterGatewayFuture
+ .thenCompose(queryFunction::apply)
+ .handle((T value, Throwable throwable) -> Optional.ofNullable(value));
+
+ optionalJobInformation.add(optionalRequest);
+ }
+ return optionalJobInformation;
+ }
+
//------------------------------------------------------
// Leader contender
//------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c0dddc1a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java
index ce964a3..f89cd2c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedLeaderService.java
@@ -212,7 +212,7 @@ public class EmbeddedLeaderService {
}
/**
- * Callback from leader contenders when they confirm a leader grant
+ * Callback from leader contenders when they confirm a leader grant.
*/
void confirmLeader(final EmbeddedLeaderElectionService service, final UUID leaderSessionId) {
synchronized (lock) {
@@ -222,7 +222,7 @@ public class EmbeddedLeaderService {
}
try {
- // check if the confirmation is for the same grant, or whether it is a stale grant
+ // check if the confirmation is for the same grant, or whether it is a stale grant
if (service == currentLeaderProposed && currentLeaderSessionId.equals(leaderSessionId)) {
final String address = service.contender.getAddress();
LOG.info("Received confirmation of leadership for leader {} , session={}", address, leaderSessionId);
@@ -231,7 +231,6 @@ public class EmbeddedLeaderService {
currentLeaderConfirmed = service;
currentLeaderAddress = address;
currentLeaderProposed = null;
- service.isLeader = true;
// notify all listeners
for (EmbeddedLeaderRetrievalService listener : listeners) {
@@ -275,7 +274,7 @@ public class EmbeddedLeaderService {
leaderService.contender, leaderService.contender.getAddress());
notificationExecutor.execute(
- new GrantLeadershipCall(leaderService.contender, leaderSessionId, LOG));
+ new GrantLeadershipCall(leaderService, leaderSessionId, LOG));
}
}
}
@@ -328,7 +327,7 @@ public class EmbeddedLeaderService {
}
// ------------------------------------------------------------------------
- // election and retrieval service implementations
+ // election and retrieval service implementations
// ------------------------------------------------------------------------
private class EmbeddedLeaderElectionService implements LeaderElectionService {
@@ -440,28 +439,33 @@ public class EmbeddedLeaderService {
private static class GrantLeadershipCall implements Runnable {
- private final LeaderContender contender;
+ private final EmbeddedLeaderElectionService leaderElectionService;
private final UUID leaderSessionId;
private final Logger logger;
GrantLeadershipCall(
- LeaderContender contender,
+ EmbeddedLeaderElectionService leaderElectionService,
UUID leaderSessionId,
Logger logger) {
- this.contender = checkNotNull(contender);
+ this.leaderElectionService = checkNotNull(leaderElectionService);
this.leaderSessionId = checkNotNull(leaderSessionId);
this.logger = checkNotNull(logger);
}
@Override
public void run() {
+ leaderElectionService.isLeader = true;
+
+ final LeaderContender contender = leaderElectionService.contender;
+
try {
contender.grantLeadership(leaderSessionId);
}
catch (Throwable t) {
logger.warn("Error granting leadership to contender", t);
contender.handleError(t instanceof Exception ? (Exception) t : new Exception(t));
+ leaderElectionService.isLeader = false;
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0dddc1a/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 3b44ed7..034232e 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
@@ -76,7 +76,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
private final JobManagerSharedServices jobManagerSharedServices;
- private final JobMaster jobManager;
+ private final JobMaster jobMaster;
private final Time rpcTimeout;
@@ -87,6 +87,8 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
/** flag marking the runner as shut down. */
private volatile boolean shutdown;
+ private volatile CompletableFuture<JobMasterGateway> leaderGatewayFuture;
+
// ------------------------------------------------------------------------
/**
@@ -139,8 +141,10 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
this.rpcTimeout = jobMasterConfiguration.getRpcTimeout();
+ this.leaderGatewayFuture = new CompletableFuture<>();
+
// now start the JobManager
- this.jobManager = new JobMaster(
+ this.jobMaster = new JobMaster(
rpcService,
jobMasterConfiguration,
resourceId,
@@ -166,8 +170,8 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
// Getter
//----------------------------------------------------------------------------------------------
- public JobMasterGateway getJobManagerGateway() {
- return jobManager.getSelfGateway(JobMasterGateway.class);
+ public CompletableFuture<JobMasterGateway> getLeaderGatewayFuture() {
+ return leaderGatewayFuture;
}
public JobGraph getJobGraph() {
@@ -185,8 +189,7 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
public void start() throws Exception {
try {
leaderElectionService.start(this);
- }
- catch (Exception e) {
+ } catch (Exception e) {
log.error("Could not start the JobManager because the leader election service did not start.", e);
throw new Exception("Could not start the leader election service.", e);
}
@@ -198,9 +201,11 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
if (!shutdown) {
shutdown = true;
- jobManager.shutDown();
+ setNewLeaderGatewayFuture();
+ leaderGatewayFuture.completeExceptionally(new FlinkException("JobMaster has been shut down."));
- final CompletableFuture<Void> jobManagerTerminationFuture = jobManager.getTerminationFuture();
+ jobMaster.shutDown();
+ final CompletableFuture<Void> jobManagerTerminationFuture = jobMaster.getTerminationFuture();
jobManagerTerminationFuture.whenComplete(
(Void ignored, Throwable throwable) -> {
@@ -294,54 +299,47 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
return;
}
- log.info("JobManager runner for job {} ({}) was granted leadership with session id {} at {}.",
- jobGraph.getName(), jobGraph.getJobID(), leaderSessionID, getAddress());
-
- // The operation may be blocking, but since this runner is idle before it been granted the leadership,
- // it's okay that job manager wait for the operation complete
- leaderElectionService.confirmLeaderSessionID(leaderSessionID);
-
- final JobSchedulingStatus schedulingStatus;
try {
- schedulingStatus = runningJobsRegistry.getJobSchedulingStatus(jobGraph.getJobID());
- }
- catch (Throwable t) {
- log.error("Could not access status (running/finished) of job {}. ", jobGraph.getJobID(), t);
- onFatalError(t);
- return;
+ verifyJobSchedulingStatusAndStartJobManager(leaderSessionID);
+ } catch (Exception e) {
+ onFatalError(new FlinkException("Could not start the JobMaster.", e));
}
+ }
+ }
- if (schedulingStatus == JobSchedulingStatus.DONE) {
- log.info("Granted leader ship but job {} has been finished. ", jobGraph.getJobID());
- jobFinishedByOther();
- return;
- }
+ private void verifyJobSchedulingStatusAndStartJobManager(UUID leaderSessionId) throws Exception {
+ final JobSchedulingStatus jobSchedulingStatus = runningJobsRegistry.getJobSchedulingStatus(jobGraph.getJobID());
+
+ if (jobSchedulingStatus == JobSchedulingStatus.DONE) {
+ log.info("Granted leader ship but job {} has been finished. ", jobGraph.getJobID());
+ jobFinishedByOther();
+ } else {
+ log.info("JobManager runner for job {} ({}) was granted leadership with session id {} at {}.",
+ jobGraph.getName(), jobGraph.getJobID(), leaderSessionId, getAddress());
+
+ runningJobsRegistry.setJobRunning(jobGraph.getJobID());
+
+ final CompletableFuture<Acknowledge> startFuture = jobMaster.start(new JobMasterId(leaderSessionId), rpcTimeout);
+ final CompletableFuture<JobMasterGateway> currentLeaderGatewayFuture = leaderGatewayFuture;
- // Double check the leadership after we confirm that, there is a small chance that multiple
- // job managers schedule the same job after if they try to recover at the same time.
- // This will eventually be noticed, but can not be ruled out from the beginning.
- if (leaderElectionService.hasLeadership()) {
- try {
- // Now set the running status is after getting leader ship and
- // set finished status after job in terminated status.
- // So if finding the job is running, it means someone has already run the job, need recover.
- if (schedulingStatus == JobSchedulingStatus.PENDING) {
- runningJobsRegistry.setJobRunning(jobGraph.getJobID());
+ startFuture.whenCompleteAsync(
+ (Acknowledge ack, Throwable throwable) -> {
+ if (throwable != null) {
+ onFatalError(new FlinkException("Could not start the job manager.", throwable));
+ } else {
+ confirmLeaderSessionIdIfStillLeader(leaderSessionId, currentLeaderGatewayFuture);
}
+ },
+ jobManagerSharedServices.getScheduledExecutorService());
+ }
+ }
- CompletableFuture<Acknowledge> startingFuture = jobManager.start(new JobMasterId(leaderSessionID), rpcTimeout);
-
- startingFuture.whenCompleteAsync(
- (Acknowledge ack, Throwable throwable) -> {
- if (throwable != null) {
- onFatalError(new Exception("Could not start the job manager.", throwable));
- }
- },
- jobManagerSharedServices.getScheduledExecutorService());
- } catch (Exception e) {
- onFatalError(new Exception("Could not start the job manager.", e));
- }
- }
+ private void confirmLeaderSessionIdIfStillLeader(UUID leaderSessionId, CompletableFuture<JobMasterGateway> currentLeaderGatewayFuture) {
+ if (leaderElectionService.hasLeadership()) {
+ currentLeaderGatewayFuture.complete(jobMaster.getSelfGateway(JobMasterGateway.class));
+ leaderElectionService.confirmLeaderSessionID(leaderSessionId);
+ } else {
+ log.debug("Ignoring confirmation of leader session id because {} is no longer the leader.", getAddress());
}
}
@@ -356,7 +354,9 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
log.info("JobManager for job {} ({}) was revoked leadership at {}.",
jobGraph.getName(), jobGraph.getJobID(), getAddress());
- CompletableFuture<Acknowledge> suspendFuture = jobManager.suspend(new FlinkException("JobManager is no longer the leader."), rpcTimeout);
+ setNewLeaderGatewayFuture();
+
+ CompletableFuture<Acknowledge> suspendFuture = jobMaster.suspend(new FlinkException("JobManager is no longer the leader."), rpcTimeout);
suspendFuture.whenCompleteAsync(
(Acknowledge ack, Throwable throwable) -> {
@@ -368,9 +368,26 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
}
}
+ private void setNewLeaderGatewayFuture() {
+ final CompletableFuture<JobMasterGateway> oldLeaderGatewayFuture = leaderGatewayFuture;
+
+ leaderGatewayFuture = new CompletableFuture<>();
+
+ if (!oldLeaderGatewayFuture.isDone()) {
+ leaderGatewayFuture.whenComplete(
+ (JobMasterGateway jobMasterGateway, Throwable throwable) -> {
+ if (throwable != null) {
+ oldLeaderGatewayFuture.completeExceptionally(throwable);
+ } else {
+ oldLeaderGatewayFuture.complete(jobMasterGateway);
+ }
+ });
+ }
+ }
+
@Override
public String getAddress() {
- return jobManager.getAddress();
+ return jobMaster.getAddress();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/c0dddc1a/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 fb1ec63..cafdf54 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
@@ -96,6 +96,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.hamcrest.Matchers.contains;
@@ -103,11 +104,13 @@ import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
@@ -368,13 +371,13 @@ public class DispatcherTest extends TestLogger {
*/
@Test
public void testSavepointDisposal() throws Exception {
+ final URI externalPointer = createTestingSavepoint();
+ final Path savepointPath = Paths.get(externalPointer);
+
final DispatcherGateway dispatcherGateway = dispatcher.getSelfGateway(DispatcherGateway.class);
dispatcherLeaderElectionService.isLeader(UUID.randomUUID()).get();
- final URI externalPointer = createTestingSavepoint();
- final Path savepointPath = Paths.get(externalPointer);
-
assertThat(Files.exists(savepointPath), is(true));
dispatcherGateway.disposeSavepoint(externalPointer.toString(), TIMEOUT).get();
@@ -397,6 +400,35 @@ public class DispatcherTest extends TestLogger {
final CompletedCheckpointStorageLocation completedCheckpointStorageLocation = metadataOutputStream.closeAndFinalizeCheckpoint();
return new URI(completedCheckpointStorageLocation.getExternalPointer());
+
+ }
+
+ /**
+ * Tests that we wait until the JobMaster has gained leader ship before sending requests
+ * to it. See FLINK-8887.
+ */
+ @Test
+ public void testWaitingForJobMasterLeadership() throws ExecutionException, InterruptedException {
+ final DispatcherGateway dispatcherGateway = dispatcher.getSelfGateway(DispatcherGateway.class);
+
+ dispatcherLeaderElectionService.isLeader(UUID.randomUUID()).get();
+
+ dispatcherGateway.submitJob(jobGraph, TIMEOUT).get();
+
+ final CompletableFuture<JobStatus> jobStatusFuture = dispatcherGateway.requestJobStatus(jobGraph.getJobID(), TIMEOUT);
+
+ assertThat(jobStatusFuture.isDone(), is(false));
+
+ try {
+ jobStatusFuture.get(10, TimeUnit.MILLISECONDS);
+ fail("Should not complete.");
+ } catch (TimeoutException ignored) {
+ // ignored
+ }
+
+ jobMasterLeaderElectionService.isLeader(UUID.randomUUID()).get();
+
+ assertThat(jobStatusFuture.get(), notNullValue());
}
private static class TestingDispatcher extends Dispatcher {
[8/8] flink git commit: [FLINK-9063] Fix SpillableSubpartitionTest
testSpillFinishedBufferConsumers
Posted by tr...@apache.org.
[FLINK-9063] Fix SpillableSubpartitionTest testSpillFinishedBufferConsumers
testSpillFinishedBufferConsumers was incorrectly manually closing the BufferConsumer
after passing it's ownership to the ResultSubpartition. This was leading to a race
conditions with AsynchronousBufferFileWriter.
This closes #5765.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/af53f28e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/af53f28e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/af53f28e
Branch: refs/heads/master
Commit: af53f28e568353998dc370a780de4e09483429ea
Parents: c0dddc1
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Mon Mar 26 14:25:58 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 28 17:01:14 2018 +0200
----------------------------------------------------------------------
.../network/partition/SpillableSubpartitionTest.java | 14 ++++++--------
1 file changed, 6 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/af53f28e/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
index 15acd7c..09e0291 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
@@ -675,14 +675,12 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
SpillableSubpartition partition = createSubpartition();
BufferBuilder bufferBuilder = createBufferBuilder(BUFFER_DATA_SIZE);
- try (BufferConsumer buffer = bufferBuilder.createBufferConsumer()) {
- partition.add(buffer);
- assertEquals(0, partition.releaseMemory());
- // finally fill the buffer with some bytes
- bufferBuilder.appendAndCommit(ByteBuffer.allocate(BUFFER_DATA_SIZE));
- bufferBuilder.finish(); // so that this buffer can be removed from the queue
- assertEquals(BUFFER_DATA_SIZE, partition.spillFinishedBufferConsumers());
- }
+ partition.add(bufferBuilder.createBufferConsumer());
+ assertEquals(0, partition.releaseMemory());
+ // finally fill the buffer with some bytes
+ bufferBuilder.appendAndCommit(ByteBuffer.allocate(BUFFER_DATA_SIZE));
+ bufferBuilder.finish(); // so that this buffer can be removed from the queue
+ assertEquals(BUFFER_DATA_SIZE, partition.spillFinishedBufferConsumers());
}
/**
[5/8] flink git commit: [FLINK-9099] Assign Execution to LogicalSlot
when slot is assigned to Execution
Posted by tr...@apache.org.
[FLINK-9099] Assign Execution to LogicalSlot when slot is assigned to Execution
In order to fail fast if an allocated slot is released by the SlotPool, we assign the
Execution as payload to a LogicalSlot when the slot is assigned to the Execution.
This closes #5775.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e6b1d5ce
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e6b1d5ce
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e6b1d5ce
Branch: refs/heads/master
Commit: e6b1d5cee423831dd0f7b3b57c31c9cd899ac8f0
Parents: 5c67d36
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Mar 27 14:23:18 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 28 17:01:13 2018 +0200
----------------------------------------------------------------------
.../flink/runtime/executiongraph/Execution.java | 11 ++--
.../runtime/jobmaster/slotpool/SlotPool.java | 3 +-
.../ExecutionGraphSchedulingTest.java | 67 +++++++++++++++++---
3 files changed, 67 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b1d5ce/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 1af4fc9..64e602f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -280,7 +280,7 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
// only allow to set the assigned resource in state SCHEDULED or CREATED
// note: we also accept resource assignment when being in state CREATED for testing purposes
if (state == SCHEDULED || state == CREATED) {
- if (ASSIGNED_SLOT_UPDATER.compareAndSet(this, null, logicalSlot)) {
+ if (ASSIGNED_SLOT_UPDATER.compareAndSet(this, null, logicalSlot) && logicalSlot.tryAssignPayload(this)) {
// check for concurrent modification (e.g. cancelling call)
if (state == SCHEDULED || state == CREATED) {
checkState(!taskManagerLocationFuture.isDone(), "The TaskManagerLocationFuture should not be set if we haven't assigned a resource yet.");
@@ -559,11 +559,12 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
throw new IllegalStateException("The vertex must be in CREATED or SCHEDULED state to be deployed. Found state " + previous);
}
+ if (this != slot.getPayload()) {
+ throw new IllegalStateException(
+ String.format("The execution %s has not been assigned to the assigned slot.", this));
+ }
+
try {
- // good, we are allowed to deploy
- if (!slot.tryAssignPayload(this)) {
- throw new JobException("Could not assign the ExecutionVertex to the slot " + slot);
- }
// race double check, did we fail/cancel and do we need to release the slot?
if (this.state != DEPLOYING) {
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b1d5ce/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
index 662e71a..a4db46c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
@@ -69,6 +69,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -659,7 +660,7 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
.orTimeout(pendingRequest.getAllocatedSlotFuture(), allocationTimeout.toMilliseconds(), TimeUnit.MILLISECONDS)
.whenCompleteAsync(
(AllocatedSlot ignored, Throwable throwable) -> {
- if (throwable != null) {
+ if (throwable instanceof TimeoutException) {
timeoutPendingSlotRequest(slotRequestId);
}
},
http://git-wip-us.apache.org/repos/asf/flink/blob/e6b1d5ce/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
index 88ba446..e43137b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
+import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
import org.apache.flink.runtime.instance.SimpleSlot;
@@ -38,6 +39,7 @@ import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobmanager.slots.DummySlotOwner;
import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
import org.apache.flink.runtime.jobmanager.slots.TestingSlotOwner;
import org.apache.flink.runtime.jobmaster.LogicalSlot;
@@ -49,12 +51,16 @@ import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
import org.junit.After;
import org.junit.Test;
import org.mockito.verification.Timeout;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
import java.net.InetAddress;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
@@ -440,14 +446,7 @@ public class ExecutionGraphSchedulingTest extends TestLogger {
final CompletableFuture<?> releaseFuture = new CompletableFuture<>();
- final TestingLogicalSlot slot = new TestingLogicalSlot(
- new LocalTaskManagerLocation(),
- new SimpleAckingTaskManagerGateway(),
- 0,
- new AllocationID(),
- new SlotRequestId(),
- new SlotSharingGroupId(),
- releaseFuture);
+ final TestingLogicalSlot slot = createTestingSlot(releaseFuture);
slotFuture1.complete(slot);
// cancel should change the state of all executions to CANCELLED
@@ -465,6 +464,46 @@ public class ExecutionGraphSchedulingTest extends TestLogger {
assertThat(executionGraph.getTerminationFuture().get(), is(JobStatus.CANCELED));
}
+ /**
+ * Tests that a partially completed eager scheduling operation fails if a
+ * completed slot is released. See FLINK-9099.
+ */
+ @Test
+ public void testSlotReleasingFailsSchedulingOperation() throws Exception {
+ final int parallelism = 2;
+
+ final JobVertex jobVertex = new JobVertex("Testing job vertex");
+ jobVertex.setInvokableClass(NoOpInvokable.class);
+ jobVertex.setParallelism(parallelism);
+ final JobGraph jobGraph = new JobGraph(jobVertex);
+ jobGraph.setAllowQueuedScheduling(true);
+ jobGraph.setScheduleMode(ScheduleMode.EAGER);
+
+ final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism);
+
+ final SimpleSlot slot = createSlot(new SimpleAckingTaskManagerGateway(), jobGraph.getJobID(), new DummySlotOwner());
+ slotProvider.addSlot(jobVertex.getID(), 0, CompletableFuture.completedFuture(slot));
+
+ final CompletableFuture<LogicalSlot> slotFuture = new CompletableFuture<>();
+ slotProvider.addSlot(jobVertex.getID(), 1, slotFuture);
+
+ final ExecutionGraph executionGraph = createExecutionGraph(jobGraph, slotProvider);
+
+ executionGraph.scheduleForExecution();
+
+ assertThat(executionGraph.getState(), is(JobStatus.RUNNING));
+
+ final ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertex.getID());
+ final ExecutionVertex[] taskVertices = executionJobVertex.getTaskVertices();
+ assertThat(taskVertices[0].getExecutionState(), is(ExecutionState.SCHEDULED));
+ assertThat(taskVertices[1].getExecutionState(), is(ExecutionState.SCHEDULED));
+
+ // fail the single allocated slot --> this should fail the scheduling operation
+ slot.releaseSlot(new FlinkException("Test failure"));
+
+ assertThat(executionGraph.getTerminationFuture().get(), is(JobStatus.FAILED));
+ }
+
// ------------------------------------------------------------------------
// Utilities
// ------------------------------------------------------------------------
@@ -530,4 +569,16 @@ public class ExecutionGraphSchedulingTest extends TestLogger {
private static class TestRuntimeException extends RuntimeException {
private static final long serialVersionUID = 1L;
}
+
+ @Nonnull
+ private static TestingLogicalSlot createTestingSlot(@Nullable CompletableFuture<?> releaseFuture) {
+ return new TestingLogicalSlot(
+ new LocalTaskManagerLocation(),
+ new SimpleAckingTaskManagerGateway(),
+ 0,
+ new AllocationID(),
+ new SlotRequestId(),
+ new SlotSharingGroupId(),
+ releaseFuture);
+ }
}
[3/8] flink git commit: [hotfix] Fix race condition in
JobMaster#requestJobDetails
Posted by tr...@apache.org.
[hotfix] Fix race condition in JobMaster#requestJobDetails
Do not directly access executionGraph in another thread because it might be
altered.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fa96c21a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fa96c21a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fa96c21a
Branch: refs/heads/master
Commit: fa96c21a67ed648837ff96b7c5d4d52979322df2
Parents: 4de80cf
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Mar 27 08:00:48 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 28 17:01:12 2018 +0200
----------------------------------------------------------------------
.../main/java/org/apache/flink/runtime/jobmaster/JobMaster.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fa96c21a/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 d116f03..4313187 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
@@ -903,7 +903,8 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
@Override
public CompletableFuture<JobDetails> requestJobDetails(@RpcTimeout Time timeout) {
- return CompletableFuture.supplyAsync(() -> WebMonitorUtils.createDetailsForJob(executionGraph), scheduledExecutorService);
+ final ExecutionGraph currentExecutionGraph = executionGraph;
+ return CompletableFuture.supplyAsync(() -> WebMonitorUtils.createDetailsForJob(currentExecutionGraph), scheduledExecutorService);
}
@Override
[6/8] flink git commit: [hotfix] Remove cluttering comments from
JobMaster
Posted by tr...@apache.org.
[hotfix] Remove cluttering comments from JobMaster
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5c67d36a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5c67d36a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5c67d36a
Branch: refs/heads/master
Commit: 5c67d36a3a1a2520536737bae1a4a2ff87b0eef2
Parents: fa96c21
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Mar 28 11:53:56 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 28 17:01:13 2018 +0200
----------------------------------------------------------------------
.../org/apache/flink/runtime/jobmaster/JobMaster.java | 11 +----------
1 file changed, 1 insertion(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5c67d36a/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 4313187..d17c97d 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
@@ -154,27 +154,20 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private final ResourceID resourceId;
- /** Logical representation of the job. */
private final JobGraph jobGraph;
private final Time rpcTimeout;
- /** Service to contend for and retrieve the leadership of JM and RM. */
private final HighAvailabilityServices highAvailabilityServices;
- /** Blob server used across jobs. */
private final BlobServer blobServer;
- /** The metrics for the job. */
private final JobManagerJobMetricGroupFactory jobMetricGroupFactory;
- /** The heartbeat manager with task managers. */
private final HeartbeatManager<AccumulatorReport, Void> taskManagerHeartbeatManager;
- /** The heartbeat manager with resource manager. */
private final HeartbeatManager<Void, Void> resourceManagerHeartbeatManager;
- /** The execution context which is used to execute futures. */
private final ScheduledExecutorService scheduledExecutorService;
private final OnCompletionActions jobCompletionActions;
@@ -195,10 +188,9 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
// --------- ResourceManager --------
- /** Leader retriever service used to locate ResourceManager's address. */
private LeaderRetrievalService resourceManagerLeaderRetriever;
- /** Connection with ResourceManager, null if not located address yet or we close it initiative. */
+ @Nullable
private ResourceManagerConnection resourceManagerConnection;
// --------- TaskManagers --------
@@ -207,7 +199,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
// -------- Mutable fields ---------
- /** The execution graph of this job. */
private ExecutionGraph executionGraph;
@Nullable