[FLINK-13977] Let HighAvailabilityServices implementations implement …

…getClusterRestEndpointLeaderElectionService

Instead of implementing HighAvailabilityServices#getWebMonitorLeaderElectionService all
HighAvailabilityServices implementation now implement getClusterRestEndpointLeaderElectionService.

This closes apache#9631.

flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedHaServices.java

@@ -51,14 +51,14 @@ public class EmbeddedHaServices extends AbstractNonHaServices {
 
  private final HashMap<JobID, EmbeddedLeaderService> jobManagerLeaderServices;
 
- private final EmbeddedLeaderService webMonitorLeaderService;
+ private final EmbeddedLeaderService clusterRestEndpointLeaderService;
 
  public EmbeddedHaServices(Executor executor) {
  this.executor = Preconditions.checkNotNull(executor);
  this.resourceManagerLeaderService = createEmbeddedLeaderService(executor);
  this.dispatcherLeaderService = createEmbeddedLeaderService(executor);
  this.jobManagerLeaderServices = new HashMap<>();
- this.webMonitorLeaderService = createEmbeddedLeaderService(executor);
+ this.clusterRestEndpointLeaderService = createEmbeddedLeaderService(executor);
  }
 
  // ------------------------------------------------------------------------
@@ -103,7 +103,7 @@ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String d
 
  @Override
  public LeaderRetrievalService getClusterRestEndpointLeaderRetriever() {
- return webMonitorLeaderService.createLeaderRetrievalService();
+ return clusterRestEndpointLeaderService.createLeaderRetrievalService();
  }
 
  @Override
@@ -118,8 +118,8 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
  }
 
  @Override
- public LeaderElectionService getWebMonitorLeaderElectionService() {
- return webMonitorLeaderService.createLeaderElectionService();
+ public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
+ return clusterRestEndpointLeaderService.createLeaderElectionService();
  }
 
  // ------------------------------------------------------------------------
@@ -169,7 +169,7 @@ public void close() throws Exception {
 
  resourceManagerLeaderService.shutdown();
 
- webMonitorLeaderService.shutdown();
+ clusterRestEndpointLeaderService.shutdown();
  }
 
  super.close();

flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java

@@ -148,7 +148,7 @@ public LeaderRetrievalService getClusterRestEndpointLeaderRetriever() {
  }
 
  @Override
- public LeaderElectionService getWebMonitorLeaderElectionService() {
+ public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
  synchronized (lock) {
  checkNotShutdown();
 

flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/zookeeper/ZooKeeperHaServices.java

@@ -171,7 +171,7 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
  }
 
  @Override
- public LeaderElectionService getWebMonitorLeaderElectionService() {
+ public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
  return ZooKeeperUtils.createLeaderElectionService(client, configuration, REST_SERVER_LEADER_PATH);
  }
 

flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java

@@ -55,7 +55,7 @@ public class TestingHighAvailabilityServices implements HighAvailabilityServices
 
  private volatile LeaderElectionService dispatcherLeaderElectionService;
 
- private volatile LeaderElectionService webMonitorEndpointLeaderElectionService;
+ private volatile LeaderElectionService clusterRestEndpointLeaderElectionService;
 
  private volatile CheckpointRecoveryFactory checkpointRecoveryFactory;
 
@@ -95,8 +95,8 @@ public void setDispatcherLeaderElectionService(LeaderElectionService leaderElect
  this.dispatcherLeaderElectionService = leaderElectionService;
  }
 
- public void setWebMonitorEndpointLeaderElectionService(final LeaderElectionService webMonitorEndpointLeaderElectionService) {
- this.webMonitorEndpointLeaderElectionService = webMonitorEndpointLeaderElectionService;
+ public void setClusterRestEndpointLeaderElectionService(final LeaderElectionService clusterRestEndpointLeaderElectionService) {
+ this.clusterRestEndpointLeaderElectionService = clusterRestEndpointLeaderElectionService;
  }
 
  public void setCheckpointRecoveryFactory(CheckpointRecoveryFactory checkpointRecoveryFactory) {
@@ -197,8 +197,8 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
  }
 
  @Override
- public LeaderElectionService getWebMonitorLeaderElectionService() {
- return webMonitorEndpointLeaderElectionService;
+ public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
+ return clusterRestEndpointLeaderElectionService;
  }
 
  @Override

flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServicesBuilder.java

@@ -70,7 +70,7 @@ public TestingHighAvailabilityServices build() {
 
  testingHighAvailabilityServices.setResourceManagerLeaderElectionService(resourceManagerLeaderElectionService);
  testingHighAvailabilityServices.setDispatcherLeaderElectionService(dispatcherLeaderElectionService);
- testingHighAvailabilityServices.setWebMonitorEndpointLeaderElectionService(webMonitorEndpointLeaderElectionService);
+ testingHighAvailabilityServices.setClusterRestEndpointLeaderElectionService(webMonitorEndpointLeaderElectionService);
 
  testingHighAvailabilityServices.setCheckpointRecoveryFactory(checkpointRecoveryFactory);
  testingHighAvailabilityServices.setJobGraphStore(jobGraphStore);

flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingManualHighAvailabilityServices.java

@@ -47,13 +47,13 @@ public class TestingManualHighAvailabilityServices implements HighAvailabilitySe
 
  private final ManualLeaderService dispatcherLeaderService;
 
- private final ManualLeaderService webMonitorEndpointLeaderService;
+ private final ManualLeaderService clusterRestEndpointLeaderService;
 
  public TestingManualHighAvailabilityServices() {
  jobManagerLeaderServices = new HashMap<>(4);
  resourceManagerLeaderService = new ManualLeaderService();
  dispatcherLeaderService = new ManualLeaderService();
- webMonitorEndpointLeaderService = new ManualLeaderService();
+ clusterRestEndpointLeaderService = new ManualLeaderService();
  }
 
  @Override
@@ -80,7 +80,7 @@ public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String d
 
  @Override
  public LeaderRetrievalService getClusterRestEndpointLeaderRetriever() {
- return webMonitorEndpointLeaderService.createLeaderRetrievalService();
+ return clusterRestEndpointLeaderService.createLeaderRetrievalService();
  }
 
  @Override
@@ -101,8 +101,8 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
  }
 
  @Override
- public LeaderElectionService getWebMonitorLeaderElectionService() {
- return webMonitorEndpointLeaderService.createLeaderElectionService();
+ public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
+ return clusterRestEndpointLeaderService.createLeaderElectionService();
  }
 
  @Override

flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnIntraNonHaMasterServices.java

@@ -177,7 +177,7 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
  }
 
  @Override
- public LeaderElectionService getWebMonitorLeaderElectionService() {
+ public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
  enter();
  try {
  throw new UnsupportedOperationException();

flink-yarn/src/main/java/org/apache/flink/yarn/highavailability/YarnPreConfiguredMasterNonHaServices.java

@@ -199,7 +199,7 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
  }
 
  @Override
- public LeaderElectionService getWebMonitorLeaderElectionService() {
+ public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
  enter();
  try {
  throw new UnsupportedOperationException();