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 2017/09/04 06:36:57 UTC
flink git commit: [FLINK-7500] Set JobMaster leader session id in
main thread
Repository: flink
Updated Branches:
refs/heads/master a3df5a2ca -> a1578710a
[FLINK-7500] Set JobMaster leader session id in main thread
This commit changes the JobMaster such that it's leader session id is only set
in the main thread. This is done by passing the leader session id to the
startJobExecution method. Moreover, this commit returns a future from the
start and suspend methods which can be used to wait on the completion of these
operations.
This closes #4579.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a1578710
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a1578710
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a1578710
Branch: refs/heads/master
Commit: a1578710a8aa4c2cc862ae9ed3cf807133273d67
Parents: a3df5a2
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Aug 24 13:24:15 2017 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Mon Sep 4 08:34:27 2017 +0200
----------------------------------------------------------------------
.../runtime/jobmaster/JobManagerRunner.java | 27 +++++-
.../flink/runtime/jobmaster/JobMaster.java | 88 ++++++++++----------
.../jobmaster/JobManagerRunnerMockTest.java | 17 ++--
.../flink/runtime/jobmaster/JobMasterTest.java | 15 +++-
4 files changed, 91 insertions(+), 56 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a1578710/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 8f1be4c..8766fab 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
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.client.JobExecutionException;
@@ -33,6 +34,7 @@ import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.OnCompletionActions;
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.MetricRegistry;
import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
@@ -44,6 +46,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -82,6 +85,8 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
private final JobManagerMetricGroup jobManagerMetricGroup;
+ private final Time timeout;
+
/** flag marking the runner as shut down */
private volatile boolean shutdown;
@@ -207,6 +212,8 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
this,
this,
userCodeLoader);
+
+ this.timeout = jobManagerServices.rpcAskTimeout;
}
catch (Throwable t) {
// clean up everything
@@ -415,7 +422,15 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
runningJobsRegistry.setJobRunning(jobGraph.getJobID());
}
- jobManager.start(leaderSessionID);
+ CompletableFuture<Acknowledge> startingFuture = jobManager.start(leaderSessionID, timeout);
+
+ startingFuture.whenCompleteAsync(
+ (Acknowledge ack, Throwable throwable) -> {
+ if (throwable != null) {
+ onFatalError(new Exception("Could not start the job manager.", throwable));
+ }
+ },
+ jobManagerServices.executorService);
} catch (Exception e) {
onFatalError(new Exception("Could not start the job manager.", e));
}
@@ -434,7 +449,15 @@ public class JobManagerRunner implements LeaderContender, OnCompletionActions, F
log.info("JobManager for job {} ({}) was revoked leadership at {}.",
jobGraph.getName(), jobGraph.getJobID(), getAddress());
- jobManager.suspend(new Exception("JobManager is no longer the leader."));
+ CompletableFuture<Acknowledge> suspendFuture = jobManager.suspend(new Exception("JobManager is no longer the leader."), timeout);
+
+ suspendFuture.whenCompleteAsync(
+ (Acknowledge ack, Throwable throwable) -> {
+ if (throwable != null) {
+ onFatalError(new Exception("Could not start the job manager.", throwable));
+ }
+ },
+ jobManagerServices.executorService);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a1578710/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 3e66a34..55bc8f8 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
@@ -90,6 +90,7 @@ import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
import org.apache.flink.runtime.taskexecutor.slot.SlotOffer;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedThrowable;
@@ -109,7 +110,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -130,9 +130,6 @@ public class JobMaster extends RpcEndpoint implements JobMasterGateway {
public static final String JOB_MANAGER_NAME = "jobmanager";
public static final String ARCHIVE_NAME = "archive";
- private static final AtomicReferenceFieldUpdater<JobMaster, UUID> LEADER_ID_UPDATER =
- AtomicReferenceFieldUpdater.newUpdater(JobMaster.class, UUID.class, "leaderSessionID");
-
// ------------------------------------------------------------------------
private final JobMasterGateway selfGateway;
@@ -311,18 +308,14 @@ public class JobMaster extends RpcEndpoint implements JobMasterGateway {
* Start the rpc service and begin to run the job.
*
* @param leaderSessionID The necessary leader id for running the job.
+ * @param timeout for the operation
+ * @return Future acknowledge if the job could be started. Otherwise the future contains an exception
*/
- public void start(final UUID leaderSessionID) throws Exception {
- if (LEADER_ID_UPDATER.compareAndSet(this, null, leaderSessionID)) {
- // make sure we receive RPC and async calls
- super.start();
+ public CompletableFuture<Acknowledge> start(final UUID leaderSessionID, final Time timeout) throws Exception {
+ // make sure we receive RPC and async calls
+ super.start();
- log.info("JobManager started as leader {} for job {}", leaderSessionID, jobGraph.getJobID());
- runAsync(this::startJobExecution);
- }
- else {
- log.warn("Job already started with leader ID {}, ignoring this start request.", leaderSessionID);
- }
+ return callAsync(() -> startJobExecution(leaderSessionID), timeout);
}
/**
@@ -330,14 +323,20 @@ public class JobMaster extends RpcEndpoint implements JobMasterGateway {
* will be disposed.
*
* <p>Mostly job is suspended because of the leadership has been revoked, one can be restart this job by
- * calling the {@link #start(UUID)} method once we take the leadership back again.
+ * calling the {@link #start(UUID, Time)} method once we take the leadership back again.
*
* <p>This method is executed asynchronously
*
* @param cause The reason of why this job been suspended.
+ * @param timeout for this operation
+ * @return Future acknowledge indicating that the job has been suspended. Otherwise the future contains an exception
*/
- public void suspend(final Throwable cause) {
- runAsync(() -> suspendExecution(cause));
+ public CompletableFuture<Acknowledge> suspend(final Throwable cause, final Time timeout) {
+ CompletableFuture<Acknowledge> suspendFuture = callAsync(() -> suspendExecution(cause), timeout);
+
+ stop();
+
+ return suspendFuture;
}
/**
@@ -788,11 +787,26 @@ public class JobMaster extends RpcEndpoint implements JobMasterGateway {
//-- job starting and stopping -----------------------------------------------------------------
- private void startJobExecution() {
- // double check that the leader status did not change
+ private Acknowledge startJobExecution(UUID newLeaderSessionId) throws Exception {
+ Preconditions.checkNotNull(newLeaderSessionId, "The new leader session id must not be null.");
+
if (leaderSessionID == null) {
- log.info("Aborting job startup - JobManager lost leader status");
- return;
+ log.info("Start job execution with leader id {}.", newLeaderSessionId);
+
+ leaderSessionID = newLeaderSessionId;
+ } else if (Objects.equals(leaderSessionID, newLeaderSessionId)) {
+ log.info("Already started the job execution with leader id {}.", leaderSessionID);
+
+ return Acknowledge.get();
+ } else {
+ log.info("Restarting old job with leader id {}. The new leader id is {}.", leaderSessionID, newLeaderSessionId);
+
+ // first we have to suspend the current execution
+ suspendExecution(new FlinkException("Old job with leader id " + leaderSessionID +
+ " is restarted with a new leader id " + newLeaderSessionId + '.'));
+
+ // set new leader id
+ leaderSessionID = newLeaderSessionId;
}
log.info("Starting execution of job {} ({})", jobGraph.getName(), jobGraph.getJobID());
@@ -801,13 +815,7 @@ public class JobMaster extends RpcEndpoint implements JobMasterGateway {
// start the slot pool make sure the slot pool now accepts messages for this leader
log.debug("Staring SlotPool component");
slotPool.start(leaderSessionID, getAddress());
- } catch (Exception e) {
- log.error("Faild to start job {} ({})", jobGraph.getName(), jobGraph.getJobID(), e);
-
- handleFatalError(new Exception("Could not start job execution: Failed to start the slot pool.", e));
- }
- try {
// 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
@@ -817,24 +825,21 @@ public class JobMaster extends RpcEndpoint implements JobMasterGateway {
catch (Throwable t) {
log.error("Failed to start job {} ({})", jobGraph.getName(), jobGraph.getJobID(), t);
- handleFatalError(new Exception(
- "Could not start job execution: Failed to start leader service for Resource Manager", t));
-
- return;
+ throw new Exception("Could not start job execution: Failed to start JobMaster services.", t);
}
// start scheduling job in another thread
- executor.execute(new Runnable() {
- @Override
- public void run() {
+ executor.execute(
+ () -> {
try {
executionGraph.scheduleForExecution();
}
catch (Throwable t) {
executionGraph.failGlobal(t);
}
- }
- });
+ });
+
+ return Acknowledge.get();
}
/**
@@ -842,14 +847,14 @@ public class JobMaster extends RpcEndpoint implements JobMasterGateway {
* will be disposed.
*
* <p>Mostly job is suspended because of the leadership has been revoked, one can be restart this job by
- * calling the {@link #start(UUID)} method once we take the leadership back again.
+ * calling the {@link #start(UUID, Time)} method once we take the leadership back again.
*
* @param cause The reason of why this job been suspended.
*/
- private void suspendExecution(final Throwable cause) {
+ private Acknowledge suspendExecution(final Throwable cause) {
if (leaderSessionID == null) {
log.debug("Job has already been suspended or shutdown.");
- return;
+ return Acknowledge.get();
}
// not leader any more - should not accept any leader messages any more
@@ -864,14 +869,13 @@ public class JobMaster extends RpcEndpoint implements JobMasterGateway {
// tell the execution graph (JobManager is still processing messages here)
executionGraph.suspend(cause);
- // receive no more messages until started again, should be called before we clear self leader id
- stop();
-
// the slot pool stops receiving messages and clears its pooled slots
slotPoolGateway.suspend();
// disconnect from resource manager:
closeResourceManagerConnection(new Exception("Execution was suspended.", cause));
+
+ return Acknowledge.get();
}
//----------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a1578710/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java
index 998e803..0f38db2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.BlobStore;
@@ -132,7 +133,7 @@ public class JobManagerRunnerMockTest extends TestLogger {
assertTrue(!jobCompletion.isJobFinished());
assertTrue(!jobCompletion.isJobFailed());
- verify(jobManager).start(any(UUID.class));
+ verify(jobManager).start(any(UUID.class), any(Time.class));
runner.shutdown();
verify(leaderElectionService).stop();
@@ -166,7 +167,7 @@ public class JobManagerRunnerMockTest extends TestLogger {
UUID leaderSessionID = UUID.randomUUID();
runner.grantLeadership(leaderSessionID);
- verify(jobManager).start(leaderSessionID);
+ verify(jobManager).start(eq(leaderSessionID), any(Time.class));
assertTrue(!jobCompletion.isJobFinished());
// runner been told by JobManager that job is finished
@@ -186,7 +187,7 @@ public class JobManagerRunnerMockTest extends TestLogger {
UUID leaderSessionID = UUID.randomUUID();
runner.grantLeadership(leaderSessionID);
- verify(jobManager).start(leaderSessionID);
+ verify(jobManager).start(eq(leaderSessionID), any(Time.class));
assertTrue(!jobCompletion.isJobFinished());
// runner been told by JobManager that job is failed
@@ -205,11 +206,11 @@ public class JobManagerRunnerMockTest extends TestLogger {
UUID leaderSessionID = UUID.randomUUID();
runner.grantLeadership(leaderSessionID);
- verify(jobManager).start(leaderSessionID);
+ verify(jobManager).start(eq(leaderSessionID), any(Time.class));
assertTrue(!jobCompletion.isJobFinished());
runner.revokeLeadership();
- verify(jobManager).suspend(any(Throwable.class));
+ verify(jobManager).suspend(any(Throwable.class), any(Time.class));
assertFalse(runner.isShutdown());
}
@@ -220,16 +221,16 @@ public class JobManagerRunnerMockTest extends TestLogger {
UUID leaderSessionID = UUID.randomUUID();
runner.grantLeadership(leaderSessionID);
- verify(jobManager).start(leaderSessionID);
+ verify(jobManager).start(eq(leaderSessionID), any(Time.class));
assertTrue(!jobCompletion.isJobFinished());
runner.revokeLeadership();
- verify(jobManager).suspend(any(Throwable.class));
+ verify(jobManager).suspend(any(Throwable.class), any(Time.class));
assertFalse(runner.isShutdown());
UUID leaderSessionID2 = UUID.randomUUID();
runner.grantLeadership(leaderSessionID2);
- verify(jobManager).start(leaderSessionID2);
+ verify(jobManager).start(eq(leaderSessionID2), any(Time.class));
}
private static class TestingOnCompletionActions implements OnCompletionActions, FatalErrorHandler {
http://git-wip-us.apache.org/repos/asf/flink/blob/a1578710/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 2c17b5a..76a0b93 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
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.OnCompletionActions;
import org.apache.flink.runtime.leaderelection.TestingLeaderRetrievalService;
+import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.registration.RegistrationResponse;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.rpc.TestingRpcService;
@@ -110,13 +111,16 @@ public class JobMasterTest extends TestLogger {
blobServer,
mock(BlobLibraryCacheManager.class),
mock(RestartStrategyFactory.class),
- Time.of(10, TimeUnit.SECONDS),
+ testingTimeout,
null,
mock(OnCompletionActions.class),
testingFatalErrorHandler,
new FlinkUserCodeClassLoader(new URL[0]));
- jobMaster.start(jmLeaderId);
+ CompletableFuture<Acknowledge> startFuture = jobMaster.start(jmLeaderId, testingTimeout);
+
+ // wait for the start to complete
+ startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
final JobMasterGateway jobMasterGateway = jobMaster.getSelfGateway(JobMasterGateway.class);
@@ -209,13 +213,16 @@ public class JobMasterTest extends TestLogger {
mock(BlobServer.class),
mock(BlobLibraryCacheManager.class),
mock(RestartStrategyFactory.class),
- Time.of(10, TimeUnit.SECONDS),
+ testingTimeout,
null,
mock(OnCompletionActions.class),
testingFatalErrorHandler,
new FlinkUserCodeClassLoader(new URL[0]));
- jobMaster.start(jmLeaderId);
+ CompletableFuture<Acknowledge> startFuture = jobMaster.start(jmLeaderId, testingTimeout);
+
+ // wait for the start operation to complete
+ startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
// define a leader and see that a registration happens
rmLeaderRetrievalService.notifyListener(resourceManagerAddress, rmLeaderId);