flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [5/5] flink git commit: [FLINK-4351] [cluster management] JobManager handle TaskManager's registration
Date Sun, 16 Oct 2016 21:38:46 GMT
[FLINK-4351] [cluster management] JobManager handle TaskManager's registration


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a19cae3b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a19cae3b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a19cae3b

Branch: refs/heads/flip-6
Commit: a19cae3b07963776c07c0aae7bee806004f59429
Parents: e91b82d
Author: Kurt Young <ykt836@gmail.com>
Authored: Sun Oct 16 23:00:57 2016 +0800
Committer: Stephan Ewen <sewen@apache.org>
Committed: Sun Oct 16 22:14:41 2016 +0200

----------------------------------------------------------------------
 .../flink/runtime/jobmaster/JobMaster.java      | 75 +++++++++++++++++---
 .../runtime/jobmaster/JobMasterGateway.java     | 15 ++--
 .../runtime/taskexecutor/JobLeaderService.java  | 55 ++++++--------
 .../runtime/taskexecutor/TaskExecutor.java      |  2 +-
 .../taskexecutor/TaskManagerServices.java       |  2 +-
 .../runtime/taskexecutor/TaskExecutorTest.java  | 10 ++-
 6 files changed, 102 insertions(+), 57 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/a19cae3b/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 05c20d3..8cb9946 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
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.jobmaster;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.io.InputSplit;
 import org.apache.flink.core.io.InputSplitAssigner;
@@ -36,7 +37,9 @@ import org.apache.flink.runtime.client.SerializedJobExecutionResult;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.concurrent.ApplyFunction;
+import org.apache.flink.runtime.concurrent.BiFunction;
 import org.apache.flink.runtime.concurrent.Future;
+import org.apache.flink.runtime.concurrent.impl.FlinkCompletableFuture;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
 import org.apache.flink.runtime.executiongraph.Execution;
@@ -81,7 +84,9 @@ import org.apache.flink.runtime.rpc.RpcService;
 import org.apache.flink.runtime.rpc.StartStoppable;
 import org.apache.flink.runtime.state.CheckpointStateHandles;
 import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
 import org.apache.flink.runtime.taskmanager.TaskExecutionState;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
 import org.apache.flink.runtime.util.SerializedThrowable;
 import org.apache.flink.util.InstantiationUtil;
 import org.apache.flink.util.SerializedValue;
@@ -89,8 +94,10 @@ import org.slf4j.Logger;
 
 import javax.annotation.Nullable;
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -122,6 +129,8 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 	/** Configuration of the JobManager */
 	private final Configuration configuration;
 
+	private final Time rpcTimeout;
+
 	/** Service to contend for and retrieve the leadership of JM and RM */
 	private final HighAvailabilityServices highAvailabilityServices;
 
@@ -152,7 +161,7 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 
 	private volatile UUID leaderSessionID;
 
-	// --------- resource manager --------
+	// --------- ResourceManager --------
 
 	/** Leader retriever service used to locate ResourceManager's address */
 	private LeaderRetrievalService resourceManagerLeaderRetriever;
@@ -160,6 +169,9 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 	/** Connection with ResourceManager, null if not located address yet or we close it initiative
*/
 	private ResourceManagerConnection resourceManagerConnection;
 
+	// --------- TaskManagers --------
+
+	private final Map<ResourceID, Tuple2<TaskManagerLocation, TaskExecutorGateway>>
registeredTaskManagers;
 
 	// ------------------------------------------------------------------------
 
@@ -181,6 +193,7 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 
 		this.jobGraph = checkNotNull(jobGraph);
 		this.configuration = checkNotNull(configuration);
+		this.rpcTimeout = rpcAskTimeout;
 		this.highAvailabilityServices = checkNotNull(highAvailabilityService);
 		this.libraryCacheManager = checkNotNull(libraryCacheManager);
 		this.executionContext = checkNotNull(executorService);
@@ -243,6 +256,8 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 
 		this.slotPool = new SlotPool(executorService);
 		this.allocationTimeout = Time.of(5, TimeUnit.SECONDS);
+
+		this.registeredTaskManagers = new HashMap<>(4);
 	}
 
 	//----------------------------------------------------------------------------------------------
@@ -379,8 +394,10 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 		}
 		closeResourceManagerConnection();
 
-		// TODO: disconnect from all registered task managers
-
+		for (ResourceID taskManagerId : registeredTaskManagers.keySet()) {
+			slotPool.releaseResource(taskManagerId);
+		}
+		registeredTaskManagers.clear();
 	}
 
 	//----------------------------------------------------------------------------------------------
@@ -662,11 +679,53 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 	}
 
 	@RpcMethod
-	public RegistrationResponse registerTaskManager(
-		final String taskManagerAddress,
-		final ResourceID taskManagerProcessId,
-		final UUID leaderId) {
-		throw new UnsupportedOperationException("Has to be implemented.");
+	public Future<RegistrationResponse> registerTaskManager(
+			final TaskManagerLocation taskManagerLocation,
+			final UUID leaderId) throws Exception
+	{
+		if (!JobMaster.this.leaderSessionID.equals(leaderId)) {
+			log.warn("Discard registration from TaskExecutor {} at ({}) because the expected " +
+							"leader session ID {} did not equal the received leader session ID {}.",
+					taskManagerLocation.getResourceID(), taskManagerLocation.addressString(),
+					JobMaster.this.leaderSessionID, leaderId);
+			throw new Exception("Leader id not match, expected: " + JobMaster.this.leaderSessionID
+					+ ", actual: " + leaderId);
+		}
+
+		final ResourceID taskManagerId = taskManagerLocation.getResourceID();
+
+		if (registeredTaskManagers.containsKey(taskManagerId)) {
+			final RegistrationResponse response = new JMTMRegistrationSuccess(
+					taskManagerId, libraryCacheManager.getBlobServerPort());
+			return FlinkCompletableFuture.completed(response);
+		} else {
+			return getRpcService().execute(new Callable<TaskExecutorGateway>() {
+				@Override
+				public TaskExecutorGateway call() throws Exception {
+					return getRpcService().connect(taskManagerLocation.addressString(), TaskExecutorGateway.class)
+							.get(rpcTimeout.getSize(), rpcTimeout.getUnit());
+				}
+			}).handleAsync(new BiFunction<TaskExecutorGateway, Throwable, RegistrationResponse>()
{
+				@Override
+				public RegistrationResponse apply(TaskExecutorGateway taskExecutorGateway, Throwable
throwable) {
+					if (throwable != null) {
+						return new RegistrationResponse.Decline(throwable.getMessage());
+					}
+
+					if (!JobMaster.this.leaderSessionID.equals(leaderId)) {
+						log.warn("Discard registration from TaskExecutor {} at ({}) because the expected "
+
+										"leader session ID {} did not equal the received leader session ID {}.",
+								taskManagerLocation.getResourceID(), taskManagerLocation.addressString(),
+								JobMaster.this.leaderSessionID, leaderId);
+						return new RegistrationResponse.Decline("Invalid leader session id");
+					}
+
+					slotPool.registerResource(taskManagerId);
+					registeredTaskManagers.put(taskManagerId, Tuple2.of(taskManagerLocation, taskExecutorGateway));
+					return new JMTMRegistrationSuccess(taskManagerId, libraryCacheManager.getBlobServerPort());
+				}
+			}, getMainThreadExecutor());
+		}
 	}
 
 	//----------------------------------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/flink/blob/a19cae3b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
index 0f155a4..4c85839 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
@@ -41,6 +41,7 @@ import org.apache.flink.runtime.rpc.RpcTimeout;
 import org.apache.flink.runtime.state.KeyGroupRange;
 import org.apache.flink.runtime.state.KvState;
 import org.apache.flink.runtime.taskmanager.TaskExecutionState;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
 
 import java.util.UUID;
 
@@ -195,15 +196,13 @@ public interface JobMasterGateway extends CheckpointCoordinatorGateway
{
 	/**
 	 * Register the task manager at the job manager.
 	 *
-	 * @param taskManagerAddress address of the task manager
-	 * @param taskManagerProcessId identifying the task manager
-	 * @param leaderId identifying the job leader
-	 * @param timeout for the rpc call
+	 * @param taskManagerLocation location of the task manager
+	 * @param leaderId            identifying the job leader
+	 * @param timeout             for the rpc call
 	 * @return Future registration response indicating whether the registration was successful
or not
 	 */
 	Future<RegistrationResponse> registerTaskManager(
-		final String taskManagerAddress,
-		final ResourceID taskManagerProcessId,
-		final UUID leaderId,
-		@RpcTimeout final Time timeout);
+			final TaskManagerLocation taskManagerLocation,
+			final UUID leaderId,
+			@RpcTimeout final Time timeout);
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a19cae3b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
index e7f52e2..14d36ab 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
@@ -21,7 +21,6 @@ package org.apache.flink.runtime.taskexecutor;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.concurrent.Future;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.jobmaster.JMTMRegistrationSuccess;
@@ -32,6 +31,7 @@ import org.apache.flink.runtime.registration.RegisteredRpcConnection;
 import org.apache.flink.runtime.registration.RegistrationResponse;
 import org.apache.flink.runtime.registration.RetryingRegistration;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
 import org.apache.flink.util.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -53,8 +53,8 @@ public class JobLeaderService {
 
 	private static final Logger LOG = LoggerFactory.getLogger(JobLeaderService.class);
 
-	/** Process id of the owning process */
-	private final ResourceID ownerProcessId;
+	/** Self's location, used for the job manager connection */
+	private final TaskManagerLocation ownLocation;
 
 	/** The leader retrieval service and listener for each registered job */
 	private final Map<JobID, Tuple2<LeaderRetrievalService, JobLeaderService.JobManagerLeaderListener>>
jobLeaderServices;
@@ -62,9 +62,6 @@ public class JobLeaderService {
 	/** Internal state of the service */
 	private volatile JobLeaderService.State state;
 
-	/** Address of the owner of this service. This address is used for the job manager connection
*/
-	private String ownerAddress;
-
 	/** Rpc service to use for establishing connections */
 	private RpcService rpcService;
 
@@ -74,14 +71,13 @@ public class JobLeaderService {
 	/** Job leader listener listening for job leader changes */
 	private JobLeaderListener jobLeaderListener;
 
-	public JobLeaderService(ResourceID ownerProcessId) {
-		this.ownerProcessId = Preconditions.checkNotNull(ownerProcessId);
+	public JobLeaderService(TaskManagerLocation location) {
+		this.ownLocation = Preconditions.checkNotNull(location);
 
 		jobLeaderServices = new HashMap<>(4);
 
 		state = JobLeaderService.State.CREATED;
 
-		ownerAddress = null;
 		rpcService = null;
 		highAvailabilityServices = null;
 		jobLeaderListener = null;
@@ -94,13 +90,11 @@ public class JobLeaderService {
 	/**
 	 * Start the job leader service with the given services.
 	 *
-	 * @param initialOwnerAddress to be used for establishing connections (source address)
 	 * @param initialRpcService to be used to create rpc connections
 	 * @param initialHighAvailabilityServices to create leader retrieval services for the different
jobs
 	 * @param initialJobLeaderListener listening for job leader changes
 	 */
 	public void start(
-		final String initialOwnerAddress,
 		final RpcService initialRpcService,
 		final HighAvailabilityServices initialHighAvailabilityServices,
 		final JobLeaderListener initialJobLeaderListener) {
@@ -110,7 +104,6 @@ public class JobLeaderService {
 		} else {
 			LOG.info("Start job leader service.");
 
-			this.ownerAddress = Preconditions.checkNotNull(initialOwnerAddress);
 			this.rpcService = Preconditions.checkNotNull(initialRpcService);
 			this.highAvailabilityServices = Preconditions.checkNotNull(initialHighAvailabilityServices);
 			this.jobLeaderListener = Preconditions.checkNotNull(initialJobLeaderListener);
@@ -311,14 +304,13 @@ public class JobLeaderService {
 			@Override
 			protected RetryingRegistration<JobMasterGateway, JMTMRegistrationSuccess> generateRegistration()
{
 				return new JobLeaderService.JobManagerRetryingRegistration(
-					LOG,
-					rpcService,
-					"JobManager",
-					JobMasterGateway.class,
-					getTargetAddress(),
-					getTargetLeaderId(),
-					ownerAddress,
-					ownerProcessId);
+						LOG,
+						rpcService,
+						"JobManager",
+						JobMasterGateway.class,
+						getTargetAddress(),
+						getTargetLeaderId(),
+						ownLocation);
 			}
 
 			@Override
@@ -349,10 +341,11 @@ public class JobLeaderService {
 	/**
 	 * Retrying registration for the job manager <--> task manager connection.
 	 */
-	private static final class JobManagerRetryingRegistration extends RetryingRegistration<JobMasterGateway,
JMTMRegistrationSuccess> {
+	private static final class JobManagerRetryingRegistration
+			extends RetryingRegistration<JobMasterGateway, JMTMRegistrationSuccess>
+	{
 
-		private final String taskManagerAddress;
-		private final ResourceID taskManagerProcessId;
+		private final TaskManagerLocation taskManagerLocation;
 
 		JobManagerRetryingRegistration(
 			Logger log,
@@ -361,22 +354,18 @@ public class JobLeaderService {
 			Class<JobMasterGateway> targetType,
 			String targetAddress,
 			UUID leaderId,
-			String taskManagerAddress,
-			ResourceID taskManagerProcessId) {
+			TaskManagerLocation taskManagerLocation) {
 
 			super(log, rpcService, targetName, targetType, targetAddress, leaderId);
 
-			this.taskManagerAddress = Preconditions.checkNotNull(taskManagerAddress);
-			this.taskManagerProcessId = Preconditions.checkNotNull(taskManagerProcessId);
+			this.taskManagerLocation = Preconditions.checkNotNull(taskManagerLocation);
 		}
 
 		@Override
-		protected Future<RegistrationResponse> invokeRegistration(JobMasterGateway gateway,
UUID leaderId, long timeoutMillis) throws Exception {
-			return gateway.registerTaskManager(
-				taskManagerAddress,
-				taskManagerProcessId,
-				leaderId,
-				Time.milliseconds(timeoutMillis));
+		protected Future<RegistrationResponse> invokeRegistration(
+				JobMasterGateway gateway, UUID leaderId, long timeoutMillis) throws Exception
+		{
+			return gateway.registerTaskManager(taskManagerLocation, leaderId, Time.milliseconds(timeoutMillis));
 		}
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a19cae3b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index 3e3a544..1201281 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -206,7 +206,7 @@ public class TaskExecutor extends RpcEndpoint<TaskExecutorGateway>
{
 		taskSlotTable.start(new SlotActionsImpl());
 
 		// start the job leader service
-		jobLeaderService.start(getAddress(), getRpcService(), haServices, new JobLeaderListenerImpl());
+		jobLeaderService.start(getRpcService(), haServices, new JobLeaderListenerImpl());
 	}
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/a19cae3b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
index 7575ba3..e8de1b1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
@@ -207,7 +207,7 @@ public class TaskManagerServices {
 
 		final JobManagerTable jobManagerTable = new JobManagerTable();
 
-		final JobLeaderService jobLeaderService = new JobLeaderService(resourceID);
+		final JobLeaderService jobLeaderService = new JobLeaderService(taskManagerLocation);
 		
 		return new TaskManagerServices(
 			taskManagerLocation,

http://git-wip-us.apache.org/repos/asf/flink/blob/a19cae3b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
index 23c6833..2220f12 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
@@ -350,7 +350,7 @@ public class TaskExecutorTest extends TestLogger {
 		final TimerService<AllocationID> timerService = mock(TimerService.class);
 		final TaskSlotTable taskSlotTable = new TaskSlotTable(Arrays.asList(mock(ResourceProfile.class)),
timerService);
 		final JobManagerTable jobManagerTable = new JobManagerTable();
-		final JobLeaderService jobLeaderService = new JobLeaderService(resourceId);
+		final JobLeaderService jobLeaderService = new JobLeaderService(taskManagerLocation);
 		final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
 
 		final TestingLeaderRetrievalService resourceManagerLeaderRetrievalService = new TestingLeaderRetrievalService();
@@ -379,8 +379,7 @@ public class TaskExecutorTest extends TestLogger {
 		final JobMasterGateway jobMasterGateway = mock(JobMasterGateway.class);
 
 		when(jobMasterGateway.registerTaskManager(
-			any(String.class),
-			eq(resourceId),
+			eq(taskManagerLocation),
 			eq(jobManagerLeaderId),
 			any(Time.class)
 		)).thenReturn(FlinkCompletableFuture.<RegistrationResponse>completed(new JMTMRegistrationSuccess(jmResourceId,
blobPort)));
@@ -451,7 +450,7 @@ public class TaskExecutorTest extends TestLogger {
 		final TimerService<AllocationID> timerService = mock(TimerService.class);
 		final TaskSlotTable taskSlotTable = new TaskSlotTable(Arrays.asList(mock(ResourceProfile.class),
mock(ResourceProfile.class)), timerService);
 		final JobManagerTable jobManagerTable = new JobManagerTable();
-		final JobLeaderService jobLeaderService = new JobLeaderService(resourceId);
+		final JobLeaderService jobLeaderService = new JobLeaderService(taskManagerLocation);
 		final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
 
 		final String resourceManagerAddress = "rm";
@@ -484,8 +483,7 @@ public class TaskExecutorTest extends TestLogger {
 		final JobMasterGateway jobMasterGateway = mock(JobMasterGateway.class);
 
 		when(jobMasterGateway.registerTaskManager(
-			any(String.class),
-			eq(resourceId),
+			eq(taskManagerLocation),
 			eq(jobManagerLeaderId),
 			any(Time.class)
 		)).thenReturn(FlinkCompletableFuture.<RegistrationResponse>completed(new JMTMRegistrationSuccess(jmResourceId,
blobPort)));


Mime
View raw message