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/07/06 22:59:50 UTC
flink git commit: [FLINK-7117] [ha] Add
HighAvailabilityServices.getJobManagerLeaderRetriever(jobId,
defaultJobManagerAddress)
Repository: flink
Updated Branches:
refs/heads/master 12efc553c -> d1a0935e2
[FLINK-7117] [ha] Add HighAvailabilityServices.getJobManagerLeaderRetriever(jobId, defaultJobManagerAddress)
The getJobManagerLeaderRetriever method with the defaultJobManagerAddress parameter is necessary for
the standalone HA services case. The TaskExecutor don't know at creation time of the HA services,
where the JobManager will be executed. Therefore, we need this way to pass in the JobManager
address at runtime.
This closes #4276.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d1a0935e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d1a0935e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d1a0935e
Branch: refs/heads/master
Commit: d1a0935e22303c22b5be0e8346b9aac6d69ca43f
Parents: 12efc55
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Jul 7 00:32:23 2017 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Fri Jul 7 00:51:42 2017 +0200
----------------------------------------------------------------------
.../HighAvailabilityServices.java | 12 ++++++++++
.../nonha/embedded/EmbeddedHaServices.java | 5 +++++
.../nonha/standalone/StandaloneHaServices.java | 9 ++++++++
.../zookeeper/ZooKeeperHaServices.java | 5 +++++
.../runtime/taskexecutor/JobLeaderService.java | 3 ++-
.../TestingHighAvailabilityServices.java | 5 +++++
.../TestingManualHighAvailabilityServices.java | 5 +++++
.../standalone/StandaloneHaServicesTest.java | 23 ++++++++++++++++++++
.../YarnIntraNonHaMasterServices.java | 11 ++++++++++
.../YarnPreConfiguredMasterNonHaServices.java | 10 +++++++++
10 files changed, 87 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
index 69a65dd..b44905e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
@@ -77,10 +77,22 @@ public interface HighAvailabilityServices extends AutoCloseable {
*
* @param jobID The identifier of the job.
* @return Leader retrieval service to retrieve the job manager for the given job
+ * @deprecated This method should only be used by non Flip-6 code where the JobManager acts as the master.
*/
+ @Deprecated
LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID);
/**
+ * Gets the leader retriever for the job JobMaster which is responsible for the given job
+ *
+ * @param jobID The identifier of the job.
+ * @param defaultJobManagerAddress JobManager address which will be returned by
+ * a static leader retrieval service.
+ * @return Leader retrieval service to retrieve the job manager for the given job
+ */
+ LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress);
+
+ /**
* Gets the leader election service for the cluster's resource manager.
*
* @return Leader election service for the resource manager leader election
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedHaServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedHaServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedHaServices.java
index a338edc..76eb681 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedHaServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedHaServices.java
@@ -79,6 +79,11 @@ public class EmbeddedHaServices extends AbstractNonHaServices {
}
@Override
+ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+ return getJobManagerLeaderRetriever(jobID);
+ }
+
+ @Override
public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
checkNotNull(jobID);
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java
index c7b54c3..b3c6ee5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java
@@ -91,6 +91,15 @@ public class StandaloneHaServices extends AbstractNonHaServices {
}
@Override
+ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+ synchronized (lock) {
+ checkNotShutdown();
+
+ return new StandaloneLeaderRetrievalService(defaultJobManagerAddress, DEFAULT_LEADER_ID);
+ }
+ }
+
+ @Override
public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
synchronized (lock) {
checkNotShutdown();
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java
index d4748cd..9dabfa2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java
@@ -130,6 +130,11 @@ public class ZooKeeperHaServices implements HighAvailabilityServices {
}
@Override
+ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+ return getJobManagerLeaderRetriever(jobID);
+ }
+
+ @Override
public LeaderElectionService getResourceManagerLeaderElectionService() {
return ZooKeeperUtils.createLeaderElectionService(client, configuration, RESOURCE_MANAGER_LEADER_PATH);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/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 93c7bb7..2dd7964 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
@@ -191,7 +191,8 @@ public class JobLeaderService {
LOG.info("Add job {} for job leader monitoring.", jobId);
final LeaderRetrievalService leaderRetrievalService = highAvailabilityServices.getJobManagerLeaderRetriever(
- jobId);
+ jobId,
+ defaultTargetAddress);
JobLeaderService.JobManagerLeaderListener jobManagerLeaderListener = new JobManagerLeaderListener(jobId);
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java
index 830dbf9..0a7e9c8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java
@@ -101,6 +101,11 @@ public class TestingHighAvailabilityServices implements HighAvailabilityServices
}
@Override
+ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+ return getJobManagerLeaderRetriever(jobID);
+ }
+
+ @Override
public LeaderElectionService getResourceManagerLeaderElectionService() {
LeaderElectionService service = resourceManagerLeaderElectionService;
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingManualHighAvailabilityServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingManualHighAvailabilityServices.java b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingManualHighAvailabilityServices.java
index 676b0d8..0735d17 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingManualHighAvailabilityServices.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingManualHighAvailabilityServices.java
@@ -63,6 +63,11 @@ public class TestingManualHighAvailabilityServices implements HighAvailabilitySe
}
@Override
+ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+ return getJobManagerLeaderRetriever(jobID);
+ }
+
+ @Override
public LeaderElectionService getResourceManagerLeaderElectionService() {
return resourceManagerLeaderService.createLeaderElectionService();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServicesTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServicesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServicesTest.java
index 537ed20..2d51360 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServicesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServicesTest.java
@@ -101,4 +101,27 @@ public class StandaloneHaServicesTest extends TestLogger {
verify(jmListener2).notifyLeaderAddress(eq(jobManagerAddress), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
verify(rmListener).notifyLeaderAddress(eq(resourceManagerAddress), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
}
+
+ /**
+ * Tests that the standalone leader retrieval services return the given address and the
+ * fixed leader session id.
+ */
+ @Test
+ public void testJobManagerLeaderRetrievalFlip6() throws Exception {
+ JobID jobId1 = new JobID();
+ JobID jobId2 = new JobID();
+ final String jobManagerAddress1 = "foobar";
+ final String jobManagerAddress2 = "barfoo";
+ LeaderRetrievalListener jmListener1 = mock(LeaderRetrievalListener.class);
+ LeaderRetrievalListener jmListener2 = mock(LeaderRetrievalListener.class);
+
+ LeaderRetrievalService jmLeaderRetrievalService1 = standaloneHaServices.getJobManagerLeaderRetriever(jobId1, jobManagerAddress1);
+ LeaderRetrievalService jmLeaderRetrievalService2 = standaloneHaServices.getJobManagerLeaderRetriever(jobId2, jobManagerAddress2);
+
+ jmLeaderRetrievalService1.start(jmListener1);
+ jmLeaderRetrievalService2.start(jmListener2);
+
+ verify(jmListener1).notifyLeaderAddress(eq(jobManagerAddress1), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
+ verify(jmListener2).notifyLeaderAddress(eq(jobManagerAddress2), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnIntraNonHaMasterServices.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnIntraNonHaMasterServices.java b/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnIntraNonHaMasterServices.java
index accf8d5..75f8c0a 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnIntraNonHaMasterServices.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnIntraNonHaMasterServices.java
@@ -162,6 +162,17 @@ public class YarnIntraNonHaMasterServices extends AbstractYarnNonHaServices {
}
}
+ @Override
+ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+ enter();
+ try {
+ throw new UnsupportedOperationException("needs refactoring to accept default address");
+ }
+ finally {
+ exit();
+ }
+ }
+
// ------------------------------------------------------------------------
// shutdown
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d1a0935e/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnPreConfiguredMasterNonHaServices.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnPreConfiguredMasterNonHaServices.java b/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnPreConfiguredMasterNonHaServices.java
index ae8f05b..6686a52 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnPreConfiguredMasterNonHaServices.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnPreConfiguredMasterNonHaServices.java
@@ -176,4 +176,14 @@ public class YarnPreConfiguredMasterNonHaServices extends AbstractYarnNonHaServi
exit();
}
}
+
+ @Override
+ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+ enter();
+ try {
+ return new StandaloneLeaderRetrievalService(defaultJobManagerAddress, DEFAULT_LEADER_ID);
+ } finally {
+ exit();
+ }
+ }
}