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 2020/04/23 16:28:41 UTC

[flink] branch master updated (f469d9f -> c7ecfbe)

This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.


    from f469d9f  [hotfix] fix broken links in documentation
     new b118186  [hotfix] Add TimeUtils.toDuration
     new ab7bdf0  [hotfix] Simplify AkkaRpcService.connectInternal by relying on Akka APIs
     new 9c86077  [hotfix] Remove unused jobManagerAddress from StandaloneHAServices
     new e3b2504  [hotfix] Remove unused RESOURCE_MANAGER_RPC_ENDPOINT_NAME from StandaloneHaServices
     new 82f20cf  [hotfix] Fix checkstyle violations in StandaloneHaServices
     new 543cd85  [FLINK-17072] Add test case for random actor resolution
     new 04d6bc6  [FLINK-17072] Let ResourceManager use random endpoint id
     new c7ecfbe  [FLINK-17072] Let Dispatcher use random endpoint id

The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.


Summary of changes:
 .../main/java/org/apache/flink/util/TimeUtils.java | 26 +++++++
 .../java/org/apache/flink/util/TimeUtilsTest.java  | 14 ++++
 .../kubernetes/KubernetesResourceManager.java      |  2 -
 .../KubernetesResourceManagerFactory.java          |  1 -
 .../kubernetes/KubernetesResourceManagerTest.java  |  3 -
 .../clusterframework/MesosResourceManager.java     |  2 -
 .../MesosResourceManagerFactory.java               |  1 -
 .../clusterframework/MesosResourceManagerTest.java |  1 -
 .../flink/runtime/dispatcher/Dispatcher.java       |  4 +-
 .../runtime/dispatcher/DispatcherFactory.java      | 10 ---
 .../runtime/dispatcher/JobDispatcherFactory.java   |  1 -
 .../flink/runtime/dispatcher/MiniDispatcher.java   |  2 -
 .../dispatcher/SessionDispatcherFactory.java       |  1 -
 .../runtime/dispatcher/StandaloneDispatcher.java   |  2 -
 .../HighAvailabilityServicesUtils.java             | 12 +---
 .../nonha/standalone/StandaloneHaServices.java     | 16 ++---
 .../flink/runtime/minicluster/MiniCluster.java     |  2 +-
 .../resourcemanager/ActiveResourceManager.java     |  2 -
 .../runtime/resourcemanager/ResourceManager.java   |  4 +-
 .../resourcemanager/ResourceManagerFactory.java    | 10 ---
 .../resourcemanager/StandaloneResourceManager.java |  2 -
 .../StandaloneResourceManagerFactory.java          |  1 -
 .../flink/runtime/rpc/akka/AkkaRpcService.java     | 31 ++++----
 .../runtime/rpc/akka/AkkaRpcServiceUtils.java      | 53 +++++++++++++-
 .../dispatcher/DispatcherResourceCleanupTest.java  |  2 -
 .../flink/runtime/dispatcher/DispatcherTest.java   |  1 -
 .../runtime/dispatcher/MiniDispatcherTest.java     |  2 -
 .../runtime/dispatcher/TestingDispatcher.java      |  2 -
 .../runner/DefaultDispatcherRunnerITCase.java      |  5 +-
 ...tDispatcherResourceManagerComponentFactory.java | 38 ----------
 .../nonha/standalone/StandaloneHaServicesTest.java |  7 +-
 .../SessionDispatcherWithUUIDFactory.java          | 51 -------------
 .../StandaloneResourceManagerWithUUIDFactory.java  | 84 ----------------------
 .../runtime/minicluster/TestingMiniCluster.java    | 10 +--
 .../resourcemanager/ResourceManagerHATest.java     |  1 -
 .../ResourceManagerJobMasterTest.java              |  1 -
 .../ResourceManagerPartitionLifecycleTest.java     |  1 -
 .../ResourceManagerTaskExecutorTest.java           |  1 -
 .../resourcemanager/ResourceManagerTest.java       |  1 -
 .../StandaloneResourceManagerTest.java             |  4 --
 .../resourcemanager/TestingResourceManager.java    |  2 -
 .../flink/runtime/rpc/akka/AkkaRpcActorTest.java   | 21 ++++++
 .../org/apache/flink/yarn/YarnResourceManager.java |  2 -
 .../entrypoint/YarnResourceManagerFactory.java     |  1 -
 .../apache/flink/yarn/YarnResourceManagerTest.java |  4 --
 45 files changed, 141 insertions(+), 303 deletions(-)
 delete mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/TestingDefaultDispatcherResourceManagerComponentFactory.java
 delete mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/SessionDispatcherWithUUIDFactory.java
 delete mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/StandaloneResourceManagerWithUUIDFactory.java

[flink] 05/08: [hotfix] Fix checkstyle violations in StandaloneHaServices

Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 82f20cfe4d6d83f9fa355b47d6564f5cf8582490
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Apr 9 15:28:39 2020 +0200

    [hotfix] Fix checkstyle violations in StandaloneHaServices
---
 .../highavailability/nonha/standalone/StandaloneHaServices.java     | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

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 cbe80733..7731476 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
@@ -32,17 +32,17 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
  * An implementation of the {@link HighAvailabilityServices} for the non-high-availability case.
  * This implementation can be used for testing, and for cluster setups that do not
  * tolerate failures of the master processes (JobManager, ResourceManager).
- * 
+ *
  * <p>This implementation has no dependencies on any external services. It returns a fix
  * pre-configured ResourceManager and JobManager, and stores checkpoints and metadata simply on the
  * heap or on a local file system and therefore in a storage without guarantees.
  */
 public class StandaloneHaServices extends AbstractNonHaServices {
 
-	/** The fix address of the ResourceManager */
+	/** The fix address of the ResourceManager. */
 	private final String resourceManagerAddress;
 
-	/** The fix address of the Dispatcher */
+	/** The fix address of the Dispatcher. */
 	private final String dispatcherAddress;
 
 	private final String clusterRestEndpointAddress;

[flink] 01/08: [hotfix] Add TimeUtils.toDuration

Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit b118186f272eb95455e75e62e883e20414a727ff
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Apr 9 10:57:39 2020 +0200

    [hotfix] Add TimeUtils.toDuration
---
 .../main/java/org/apache/flink/util/TimeUtils.java | 26 ++++++++++++++++++++++
 .../java/org/apache/flink/util/TimeUtilsTest.java  | 14 ++++++++++++
 2 files changed, 40 insertions(+)

diff --git a/flink-core/src/main/java/org/apache/flink/util/TimeUtils.java b/flink-core/src/main/java/org/apache/flink/util/TimeUtils.java
index aafdc87..e2e6b00 100644
--- a/flink-core/src/main/java/org/apache/flink/util/TimeUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/util/TimeUtils.java
@@ -18,6 +18,8 @@
 
 package org.apache.flink.util;
 
+import org.apache.flink.api.common.time.Time;
+
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
@@ -26,6 +28,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -222,4 +225,27 @@ public class TimeUtils {
 			return timeUnit.name() + ": (" + String.join(" | ", timeUnit.getLabels()) + ")";
 		}
 	}
+
+	/**
+	 * Translates {@link Time} to {@link Duration}.
+	 *
+	 * @param time time to transform into duration
+	 * @return duration equal to the given time
+	 */
+	public static Duration toDuration(Time time) {
+		return Duration.of(time.getSize(), toChronoUnit(time.getUnit()));
+	}
+
+	private static ChronoUnit toChronoUnit(java.util.concurrent.TimeUnit timeUnit) {
+		switch(timeUnit) {
+			case NANOSECONDS: return ChronoUnit.NANOS;
+			case MICROSECONDS: return ChronoUnit.MICROS;
+			case MILLISECONDS: return ChronoUnit.MILLIS;
+			case SECONDS: return ChronoUnit.SECONDS;
+			case MINUTES: return ChronoUnit.MINUTES;
+			case HOURS: return ChronoUnit.HOURS;
+			case DAYS: return ChronoUnit.DAYS;
+			default: throw new IllegalArgumentException(String.format("Unsupported time unit %s.", timeUnit));
+		}
+	}
 }
diff --git a/flink-core/src/test/java/org/apache/flink/util/TimeUtilsTest.java b/flink-core/src/test/java/org/apache/flink/util/TimeUtilsTest.java
index 6953c18..ac3960c 100644
--- a/flink-core/src/test/java/org/apache/flink/util/TimeUtilsTest.java
+++ b/flink-core/src/test/java/org/apache/flink/util/TimeUtilsTest.java
@@ -18,12 +18,18 @@
 
 package org.apache.flink.util;
 
+import org.apache.flink.api.common.time.Time;
+
 import org.junit.Test;
 
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
+import java.util.concurrent.TimeUnit;
 
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
 
 /**
@@ -176,4 +182,12 @@ public class TimeUtilsTest {
 		assertEquals("4567000ms", TimeUtils.getStringInMillis(Duration.ofSeconds(4567L)));
 		assertEquals("4ms", TimeUtils.getStringInMillis(Duration.of(4567L, ChronoUnit.MICROS)));
 	}
+
+	@Test
+	public void testToDuration() {
+		final Time time = Time.of(1337, TimeUnit.MICROSECONDS);
+		final Duration duration = TimeUtils.toDuration(time);
+
+		assertThat(duration.toNanos(), is(equalTo(time.getUnit().toNanos(time.getSize()))));
+	}
 }

[flink] 02/08: [hotfix] Simplify AkkaRpcService.connectInternal by relying on Akka APIs

Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit ab7bdf0227a7062329d753cfb4b9cbdfa5130450
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Apr 9 11:04:29 2020 +0200

    [hotfix] Simplify AkkaRpcService.connectInternal by relying on Akka APIs
---
 .../flink/runtime/rpc/akka/AkkaRpcService.java     | 31 +++++++++-------------
 1 file changed, 13 insertions(+), 18 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
index 7862b35..ae4dfcc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
@@ -39,14 +39,13 @@ import org.apache.flink.runtime.rpc.messages.RemoteHandshakeMessage;
 import org.apache.flink.runtime.util.ExecutorThreadFactory;
 import org.apache.flink.util.AutoCloseableAsync;
 import org.apache.flink.util.ExecutorUtils;
+import org.apache.flink.util.TimeUtils;
 
 import akka.actor.AbstractActor;
-import akka.actor.ActorIdentity;
 import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 import akka.actor.ActorSystem;
 import akka.actor.Address;
-import akka.actor.Identify;
 import akka.actor.Props;
 import akka.dispatch.Futures;
 import akka.pattern.Patterns;
@@ -482,22 +481,7 @@ public class AkkaRpcService implements RpcService {
 		LOG.debug("Try to connect to remote RPC endpoint with address {}. Returning a {} gateway.",
 			address, clazz.getName());
 
-		final ActorSelection actorSel = actorSystem.actorSelection(address);
-
-		final Future<ActorIdentity> identify = Patterns
-			.ask(actorSel, new Identify(42), configuration.getTimeout().toMilliseconds())
-			.<ActorIdentity>mapTo(ClassTag$.MODULE$.<ActorIdentity>apply(ActorIdentity.class));
-
-		final CompletableFuture<ActorIdentity> identifyFuture = FutureUtils.toJava(identify);
-
-		final CompletableFuture<ActorRef> actorRefFuture = identifyFuture.thenApply(
-			(ActorIdentity actorIdentity) -> {
-				if (actorIdentity.getRef() == null) {
-					throw new CompletionException(new RpcConnectionException("Could not connect to rpc endpoint under address " + address + '.'));
-				} else {
-					return actorIdentity.getRef();
-				}
-			});
+		final CompletableFuture<ActorRef> actorRefFuture = resolveActorAddress(address);
 
 		final CompletableFuture<HandshakeSuccessMessage> handshakeFuture = actorRefFuture.thenCompose(
 			(ActorRef actorRef) -> FutureUtils.toJava(
@@ -526,6 +510,17 @@ public class AkkaRpcService implements RpcService {
 			actorSystem.dispatcher());
 	}
 
+	private CompletableFuture<ActorRef> resolveActorAddress(String address) {
+		final ActorSelection actorSel = actorSystem.actorSelection(address);
+
+		return actorSel.resolveOne(TimeUtils.toDuration(configuration.getTimeout()))
+			.toCompletableFuture()
+			.exceptionally(error -> {
+				throw new CompletionException(
+					new RpcConnectionException(String.format("Could not connect to rpc endpoint under address %s.", address), error));
+			});
+	}
+
 	// ---------------------------------------------------------------------------------------
 	// Private inner classes
 	// ---------------------------------------------------------------------------------------

[flink] 08/08: [FLINK-17072] Let Dispatcher use random endpoint id

Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit c7ecfbe8bd47292702e4e2b94d6d54f3989f4998
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Apr 9 15:54:34 2020 +0200

    [FLINK-17072] Let Dispatcher use random endpoint id
    
    This commit ensures that every Dispatcher, running in the same JVM,
    is started with a different endpoint id. This ensures that no new
    Dispatcher instance will receive messages sent to an old instance.
    
    This closes #11693.
---
 .../flink/runtime/dispatcher/Dispatcher.java       |  4 +-
 .../runtime/dispatcher/DispatcherFactory.java      | 10 -----
 .../runtime/dispatcher/JobDispatcherFactory.java   |  1 -
 .../flink/runtime/dispatcher/MiniDispatcher.java   |  2 -
 .../dispatcher/SessionDispatcherFactory.java       |  1 -
 .../runtime/dispatcher/StandaloneDispatcher.java   |  2 -
 .../HighAvailabilityServicesUtils.java             |  2 +-
 .../flink/runtime/minicluster/MiniCluster.java     |  2 +-
 .../dispatcher/DispatcherResourceCleanupTest.java  |  2 -
 .../flink/runtime/dispatcher/DispatcherTest.java   |  1 -
 .../runtime/dispatcher/MiniDispatcherTest.java     |  2 -
 .../runtime/dispatcher/TestingDispatcher.java      |  2 -
 .../runner/DefaultDispatcherRunnerITCase.java      |  5 +--
 ...tDispatcherResourceManagerComponentFactory.java | 38 ----------------
 .../SessionDispatcherWithUUIDFactory.java          | 51 ----------------------
 .../runtime/minicluster/TestingMiniCluster.java    | 11 +----
 16 files changed, 7 insertions(+), 129 deletions(-)

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 726e5e8..d487122 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
@@ -61,6 +61,7 @@ import org.apache.flink.runtime.rest.handler.legacy.backpressure.OperatorBackPre
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.PermanentlyFencedRpcEndpoint;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils;
 import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.FlinkException;
@@ -134,11 +135,10 @@ public abstract class Dispatcher extends PermanentlyFencedRpcEndpoint<Dispatcher
 
 	public Dispatcher(
 			RpcService rpcService,
-			String endpointId,
 			DispatcherId fencingToken,
 			DispatcherBootstrap dispatcherBootstrap,
 			DispatcherServices dispatcherServices) throws Exception {
-		super(rpcService, endpointId, fencingToken);
+		super(rpcService, AkkaRpcServiceUtils.createRandomName(DISPATCHER_NAME), fencingToken);
 		checkNotNull(dispatcherServices);
 
 		this.configuration = dispatcherServices.getConfiguration();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherFactory.java
index ba259fc..71376c2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherFactory.java
@@ -20,8 +20,6 @@ package org.apache.flink.runtime.dispatcher;
 
 import org.apache.flink.runtime.rpc.RpcService;
 
-import java.util.UUID;
-
 /**
  * {@link Dispatcher} factory interface.
  */
@@ -35,12 +33,4 @@ public interface DispatcherFactory {
 		DispatcherId fencingToken,
 		DispatcherBootstrap dispatcherBootstrap,
 		PartialDispatcherServicesWithJobGraphStore partialDispatcherServicesWithJobGraphStore) throws Exception;
-
-	default String generateEndpointIdWithUUID() {
-		return getEndpointId() + UUID.randomUUID();
-	}
-
-	default String getEndpointId() {
-		return Dispatcher.DISPATCHER_NAME;
-	}
 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobDispatcherFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobDispatcherFactory.java
index 3dc65a2..cc17177 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobDispatcherFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobDispatcherFactory.java
@@ -42,7 +42,6 @@ public enum JobDispatcherFactory implements DispatcherFactory {
 
 		return new MiniDispatcher(
 			rpcService,
-			getEndpointId(),
 			fencingToken,
 			DispatcherServices.from(partialDispatcherServicesWithJobGraphStore, DefaultJobManagerRunnerFactory.INSTANCE),
 			dispatcherBootstrap,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
index 5a3b05c..53ec47d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
@@ -52,14 +52,12 @@ public class MiniDispatcher extends Dispatcher {
 
 	public MiniDispatcher(
 			RpcService rpcService,
-			String endpointId,
 			DispatcherId fencingToken,
 			DispatcherServices dispatcherServices,
 			DispatcherBootstrap dispatcherBootstrap,
 			JobClusterEntrypoint.ExecutionMode executionMode) throws Exception {
 		super(
 			rpcService,
-			endpointId,
 			fencingToken,
 			dispatcherBootstrap,
 			dispatcherServices);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/SessionDispatcherFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/SessionDispatcherFactory.java
index 361bac9..cc867ca 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/SessionDispatcherFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/SessionDispatcherFactory.java
@@ -35,7 +35,6 @@ public enum SessionDispatcherFactory implements DispatcherFactory {
 		// create the default dispatcher
 		return new StandaloneDispatcher(
 			rpcService,
-			getEndpointId(),
 			fencingToken,
 			dispatcherBootstrap,
 			DispatcherServices.from(partialDispatcherServicesWithJobGraphStore, DefaultJobManagerRunnerFactory.INSTANCE));
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
index 0ade8b1..4ba50c5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
@@ -30,13 +30,11 @@ import org.apache.flink.runtime.rpc.RpcService;
 public class StandaloneDispatcher extends Dispatcher {
 	public StandaloneDispatcher(
 			RpcService rpcService,
-			String endpointId,
 			DispatcherId fencingToken,
 			DispatcherBootstrap dispatcherBootstrap,
 			DispatcherServices dispatcherServices) throws Exception {
 		super(
 			rpcService,
-			endpointId,
 			fencingToken,
 			dispatcherBootstrap,
 			dispatcherServices);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
index 3be16d8..c9d82d2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
@@ -102,7 +102,7 @@ public class HighAvailabilityServicesUtils {
 				final String dispatcherRpcUrl = AkkaRpcServiceUtils.getRpcUrl(
 					hostnamePort.f0,
 					hostnamePort.f1,
-					Dispatcher.DISPATCHER_NAME,
+					AkkaRpcServiceUtils.createWildcardName(Dispatcher.DISPATCHER_NAME),
 					addressResolution,
 					configuration);
 				final String webMonitorAddress = getWebMonitorAddress(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
index 832e7d7..b4e1e6a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
@@ -416,7 +416,7 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
 	}
 
 	@Nonnull
-	private DispatcherResourceManagerComponentFactory createDispatcherResourceManagerComponentFactory() {
+	DispatcherResourceManagerComponentFactory createDispatcherResourceManagerComponentFactory() {
 		return DefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory(StandaloneResourceManagerFactory.INSTANCE);
 	}
 
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
index 91b8ce6..9cb3888 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
@@ -75,7 +75,6 @@ import javax.annotation.Nonnull;
 import java.io.File;
 import java.io.IOException;
 import java.util.Collections;
-import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -194,7 +193,6 @@ public class DispatcherResourceCleanupTest extends TestLogger {
 		final MemoryArchivedExecutionGraphStore archivedExecutionGraphStore = new MemoryArchivedExecutionGraphStore();
 		dispatcher = new TestingDispatcher(
 			rpcService,
-			Dispatcher.DISPATCHER_NAME + UUID.randomUUID(),
 			DispatcherId.generate(),
 			new DefaultDispatcherBootstrap(Collections.emptyList()),
 			new DispatcherServices(
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 b6ff8e7..e56461d 100755
--- 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
@@ -244,7 +244,6 @@ public class DispatcherTest extends TestLogger {
 
 			return new TestingDispatcher(
 				rpcService,
-				Dispatcher.DISPATCHER_NAME + '_' + name.getMethodName(),
 				DispatcherId.generate(),
 				dispatcherBootstrap,
 				new DispatcherServices(
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 d189c5c..7733d53 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
@@ -52,7 +52,6 @@ import javax.annotation.Nonnull;
 
 import java.io.IOException;
 import java.util.Collections;
-import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
@@ -215,7 +214,6 @@ public class MiniDispatcherTest extends TestLogger {
 	private MiniDispatcher createMiniDispatcher(ClusterEntrypoint.ExecutionMode executionMode) throws Exception {
 		return new MiniDispatcher(
 			rpcService,
-			UUID.randomUUID().toString(),
 			DispatcherId.generate(),
 			new DispatcherServices(
 				configuration,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingDispatcher.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingDispatcher.java
index e72e275..22e5d74 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingDispatcher.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingDispatcher.java
@@ -37,13 +37,11 @@ class TestingDispatcher extends Dispatcher {
 
 	TestingDispatcher(
 			RpcService rpcService,
-			String endpointId,
 			DispatcherId fencingToken,
 			DispatcherBootstrap dispatcherBootstrap,
 			DispatcherServices dispatcherServices) throws Exception {
 		super(
 			rpcService,
-			endpointId,
 			fencingToken,
 			dispatcherBootstrap,
 			dispatcherServices);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerITCase.java
index fca4303..bcd1eb5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerITCase.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.dispatcher.JobManagerRunnerFactory;
 import org.apache.flink.runtime.dispatcher.MemoryArchivedExecutionGraphStore;
 import org.apache.flink.runtime.dispatcher.PartialDispatcherServices;
 import org.apache.flink.runtime.dispatcher.PartialDispatcherServicesWithJobGraphStore;
+import org.apache.flink.runtime.dispatcher.SessionDispatcherFactory;
 import org.apache.flink.runtime.dispatcher.SingleJobJobGraphStore;
 import org.apache.flink.runtime.dispatcher.StandaloneDispatcher;
 import org.apache.flink.runtime.dispatcher.TestingJobManagerRunnerFactory;
@@ -43,7 +44,6 @@ import org.apache.flink.runtime.jobmanager.JobGraphStore;
 import org.apache.flink.runtime.jobmaster.TestingJobManagerRunner;
 import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
 import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
-import org.apache.flink.runtime.minicluster.SessionDispatcherWithUUIDFactory;
 import org.apache.flink.runtime.rpc.RpcService;
 import org.apache.flink.runtime.rpc.TestingRpcServiceResource;
 import org.apache.flink.runtime.testingUtils.TestingUtils;
@@ -102,7 +102,7 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
 
 	@Before
 	public void setup() {
-		dispatcherRunnerFactory = DefaultDispatcherRunnerFactory.createSessionRunner(SessionDispatcherWithUUIDFactory.INSTANCE);
+		dispatcherRunnerFactory = DefaultDispatcherRunnerFactory.createSessionRunner(SessionDispatcherFactory.INSTANCE);
 		jobGraph = createJobGraph();
 		dispatcherLeaderElectionService = new TestingLeaderElectionService();
 		fatalErrorHandler = new TestingFatalErrorHandler();
@@ -210,7 +210,6 @@ public class DefaultDispatcherRunnerITCase extends TestLogger {
 			PartialDispatcherServicesWithJobGraphStore partialDispatcherServicesWithJobGraphStore) throws Exception {
 			return new StandaloneDispatcher(
 				rpcService,
-				generateEndpointIdWithUUID(),
 				fencingToken,
 				dispatcherBootstrap,
 				DispatcherServices.from(partialDispatcherServicesWithJobGraphStore, jobManagerRunnerFactory));
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/TestingDefaultDispatcherResourceManagerComponentFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/TestingDefaultDispatcherResourceManagerComponentFactory.java
deleted file mode 100644
index 03fe15f..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/TestingDefaultDispatcherResourceManagerComponentFactory.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.entrypoint.component;
-
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.dispatcher.runner.DispatcherRunnerFactory;
-import org.apache.flink.runtime.resourcemanager.ResourceManagerFactory;
-import org.apache.flink.runtime.rest.SessionRestEndpointFactory;
-
-/**
- * Testing extension of the {@link DefaultDispatcherResourceManagerComponentFactory}.
- */
-public class TestingDefaultDispatcherResourceManagerComponentFactory {
-	public static DispatcherResourceManagerComponentFactory createSessionComponentFactory(
-			DispatcherRunnerFactory dispatcherRunnerFactory,
-			ResourceManagerFactory<ResourceID> resourceManagerFactory) {
-		return new DefaultDispatcherResourceManagerComponentFactory(
-			dispatcherRunnerFactory,
-			resourceManagerFactory,
-			SessionRestEndpointFactory.INSTANCE);
-	}
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/SessionDispatcherWithUUIDFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/SessionDispatcherWithUUIDFactory.java
deleted file mode 100644
index 908a59f..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/SessionDispatcherWithUUIDFactory.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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.minicluster;
-
-import org.apache.flink.runtime.dispatcher.DefaultJobManagerRunnerFactory;
-import org.apache.flink.runtime.dispatcher.DispatcherBootstrap;
-import org.apache.flink.runtime.dispatcher.DispatcherFactory;
-import org.apache.flink.runtime.dispatcher.DispatcherId;
-import org.apache.flink.runtime.dispatcher.DispatcherServices;
-import org.apache.flink.runtime.dispatcher.PartialDispatcherServicesWithJobGraphStore;
-import org.apache.flink.runtime.dispatcher.StandaloneDispatcher;
-import org.apache.flink.runtime.rpc.RpcService;
-
-/**
- * {@link DispatcherFactory} which creates a {@link StandaloneDispatcher} which has an
- * endpoint id with a random UUID suffix.
- */
-public enum SessionDispatcherWithUUIDFactory implements DispatcherFactory {
-	INSTANCE;
-
-	@Override
-	public StandaloneDispatcher createDispatcher(
-			RpcService rpcService,
-			DispatcherId fencingToken,
-			DispatcherBootstrap dispatcherBootstrap,
-			PartialDispatcherServicesWithJobGraphStore partialDispatcherServicesWithJobGraphStore) throws Exception {
-		// create the default dispatcher
-		return new StandaloneDispatcher(
-			rpcService,
-			generateEndpointIdWithUUID(),
-			fencingToken,
-			dispatcherBootstrap,
-			DispatcherServices.from(partialDispatcherServicesWithJobGraphStore, DefaultJobManagerRunnerFactory.INSTANCE));
-	}
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java
index 7c1de5e..affd334 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java
@@ -22,14 +22,11 @@ import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.blob.BlobServer;
 import org.apache.flink.runtime.dispatcher.DispatcherGateway;
 import org.apache.flink.runtime.dispatcher.MemoryArchivedExecutionGraphStore;
-import org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory;
 import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent;
 import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory;
-import org.apache.flink.runtime.entrypoint.component.TestingDefaultDispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.heartbeat.HeartbeatServices;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.metrics.MetricRegistry;
-import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
 
@@ -103,7 +100,7 @@ public class TestingMiniCluster extends MiniCluster {
 			MetricRegistry metricRegistry,
 			MetricQueryServiceRetriever metricQueryServiceRetriever,
 			FatalErrorHandler fatalErrorHandler) throws Exception {
-		DispatcherResourceManagerComponentFactory dispatcherResourceManagerComponentFactory = createTestingDispatcherResourceManagerComponentFactory();
+		DispatcherResourceManagerComponentFactory dispatcherResourceManagerComponentFactory = createDispatcherResourceManagerComponentFactory();
 
 		final List<DispatcherResourceManagerComponent> result = new ArrayList<>(numberDispatcherResourceManagerComponents);
 
@@ -129,10 +126,4 @@ public class TestingMiniCluster extends MiniCluster {
 	public CompletableFuture<DispatcherGateway> getDispatcherGatewayFuture() {
 		return super.getDispatcherGatewayFuture();
 	}
-
-	private DispatcherResourceManagerComponentFactory createTestingDispatcherResourceManagerComponentFactory() {
-		return TestingDefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory(
-			DefaultDispatcherRunnerFactory.createSessionRunner(SessionDispatcherWithUUIDFactory.INSTANCE),
-			StandaloneResourceManagerFactory.INSTANCE);
-	}
 }

[flink] 03/08: [hotfix] Remove unused jobManagerAddress from StandaloneHAServices

Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 9c86077cb7251f855312bd678c07241283aa159a
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Apr 9 15:27:30 2020 +0200

    [hotfix] Remove unused jobManagerAddress from StandaloneHAServices
---
 .../runtime/highavailability/HighAvailabilityServicesUtils.java   | 8 --------
 .../highavailability/nonha/standalone/StandaloneHaServices.java   | 7 +------
 .../nonha/standalone/StandaloneHaServicesTest.java                | 7 +++----
 3 files changed, 4 insertions(+), 18 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
index 24219f4..4a915f0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
@@ -34,7 +34,6 @@ import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneHaSe
 import org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperClientHAServices;
 import org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServices;
 import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
-import org.apache.flink.runtime.jobmaster.JobMaster;
 import org.apache.flink.runtime.net.SSLUtils;
 import org.apache.flink.runtime.resourcemanager.ResourceManager;
 import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils;
@@ -94,12 +93,6 @@ public class HighAvailabilityServicesUtils {
 			case NONE:
 				final Tuple2<String, Integer> hostnamePort = getJobManagerAddress(configuration);
 
-				final String jobManagerRpcUrl = AkkaRpcServiceUtils.getRpcUrl(
-					hostnamePort.f0,
-					hostnamePort.f1,
-					JobMaster.JOB_MANAGER_NAME,
-					addressResolution,
-					configuration);
 				final String resourceManagerRpcUrl = AkkaRpcServiceUtils.getRpcUrl(
 					hostnamePort.f0,
 					hostnamePort.f1,
@@ -119,7 +112,6 @@ public class HighAvailabilityServicesUtils {
 				return new StandaloneHaServices(
 					resourceManagerRpcUrl,
 					dispatcherRpcUrl,
-					jobManagerRpcUrl,
 					webMonitorAddress);
 			case ZOOKEEPER:
 				BlobStoreService blobStoreService = BlobUtils.createBlobStoreFromConfig(configuration);
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 854a04a..c8e648b 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
@@ -48,9 +48,6 @@ public class StandaloneHaServices extends AbstractNonHaServices {
 	/** The fix address of the Dispatcher */
 	private final String dispatcherAddress;
 
-	/** The fix address of the JobManager */
-	private final String jobManagerAddress;
-
 	private final String clusterRestEndpointAddress;
 
 	/**
@@ -62,11 +59,9 @@ public class StandaloneHaServices extends AbstractNonHaServices {
 	public StandaloneHaServices(
 			String resourceManagerAddress,
 			String dispatcherAddress,
-			String jobManagerAddress,
 			String clusterRestEndpointAddress) {
 		this.resourceManagerAddress = checkNotNull(resourceManagerAddress, "resourceManagerAddress");
 		this.dispatcherAddress = checkNotNull(dispatcherAddress, "dispatcherAddress");
-		this.jobManagerAddress = checkNotNull(jobManagerAddress, "jobManagerAddress");
 		this.clusterRestEndpointAddress = checkNotNull(clusterRestEndpointAddress, clusterRestEndpointAddress);
 	}
 
@@ -116,7 +111,7 @@ public class StandaloneHaServices extends AbstractNonHaServices {
 		synchronized (lock) {
 			checkNotShutdown();
 
-			return new StandaloneLeaderRetrievalService(jobManagerAddress, DEFAULT_LEADER_ID);
+			return new StandaloneLeaderRetrievalService("UNKNOWN", DEFAULT_LEADER_ID);
 		}
 	}
 
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 737b5ee..9bea103 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
@@ -25,6 +25,7 @@ import org.apache.flink.runtime.leaderelection.LeaderElectionService;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 import org.apache.flink.util.TestLogger;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -38,7 +39,6 @@ import static org.mockito.Mockito.verify;
  */
 public class StandaloneHaServicesTest extends TestLogger {
 
-	private final String jobManagerAddress = "jobManager";
 	private final String dispatcherAddress = "dispatcher";
 	private final String resourceManagerAddress = "resourceManager";
 	private final String webMonitorAddress = "webMonitor";
@@ -51,7 +51,6 @@ public class StandaloneHaServicesTest extends TestLogger {
 		standaloneHaServices = new StandaloneHaServices(
 			resourceManagerAddress,
 			dispatcherAddress,
-			jobManagerAddress,
 			webMonitorAddress);
 	}
 
@@ -103,8 +102,8 @@ public class StandaloneHaServicesTest extends TestLogger {
 		jmLeaderRetrievalService2.start(jmListener2);
 		rmLeaderRetrievalService.start(rmListener);
 
-		verify(jmListener1).notifyLeaderAddress(eq(jobManagerAddress), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
-		verify(jmListener2).notifyLeaderAddress(eq(jobManagerAddress), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
+		verify(jmListener1).notifyLeaderAddress(eq("UNKNOWN"), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
+		verify(jmListener2).notifyLeaderAddress(eq("UNKNOWN"), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
 		verify(rmListener).notifyLeaderAddress(eq(resourceManagerAddress), eq(HighAvailabilityServices.DEFAULT_LEADER_ID));
 	}

[flink] 04/08: [hotfix] Remove unused RESOURCE_MANAGER_RPC_ENDPOINT_NAME from StandaloneHaServices

Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit e3b2504333ab05a3f187648999fa71e5205fb381
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Apr 9 15:28:02 2020 +0200

    [hotfix] Remove unused RESOURCE_MANAGER_RPC_ENDPOINT_NAME from StandaloneHaServices
---
 .../highavailability/nonha/standalone/StandaloneHaServices.java        | 3 ---
 1 file changed, 3 deletions(-)

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 c8e648b..cbe80733 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
@@ -39,9 +39,6 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
  */
 public class StandaloneHaServices extends AbstractNonHaServices {
 
-	/** The constant name of the ResourceManager RPC endpoint */
-	private static final String RESOURCE_MANAGER_RPC_ENDPOINT_NAME = "resource_manager";
-
 	/** The fix address of the ResourceManager */
 	private final String resourceManagerAddress;

[flink] 06/08: [FLINK-17072] Add test case for random actor resolution

Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 543cd85dec948a677f66e8baaf97a6884525aa26
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Apr 9 15:18:24 2020 +0200

    [FLINK-17072] Add test case for random actor resolution
---
 .../runtime/rpc/akka/AkkaRpcServiceUtils.java      | 53 ++++++++++++++++++++--
 .../flink/runtime/rpc/akka/AkkaRpcActorTest.java   | 21 +++++++++
 2 files changed, 71 insertions(+), 3 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceUtils.java
index 5b464f4..bf003e6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceUtils.java
@@ -158,8 +158,6 @@ public class AkkaRpcServiceUtils {
 		checkNotNull(endpointName, "endpointName is null");
 		checkArgument(isValidClientPort(port), "port must be in [1, 65535]");
 
-		final String protocolPrefix = akkaProtocol == AkkaProtocol.SSL_TCP ? AKKA_SSL_TCP : AKKA_TCP;
-
 		if (addressResolution == AddressResolution.TRY_ADDRESS_RESOLUTION) {
 			// Fail fast if the hostname cannot be resolved
 			//noinspection ResultOfMethodCallIgnored
@@ -168,7 +166,46 @@ public class AkkaRpcServiceUtils {
 
 		final String hostPort = NetUtils.unresolvedHostAndPortToNormalizedString(hostname, port);
 
-		return String.format("%s://flink@%s/user/%s/%s", protocolPrefix, hostPort, SUPERVISOR_NAME, endpointName);
+		return internalRpcUrl(endpointName, Optional.of(new RemoteAddressInformation(hostPort, akkaProtocol)));
+	}
+
+	public static String getLocalRpcUrl(String endpointName) {
+		return internalRpcUrl(endpointName, Optional.empty());
+	}
+
+	private static final class RemoteAddressInformation {
+		private final String hostnameAndPort;
+		private final AkkaProtocol akkaProtocol;
+
+		private RemoteAddressInformation(String hostnameAndPort, AkkaProtocol akkaProtocol) {
+			this.hostnameAndPort = hostnameAndPort;
+			this.akkaProtocol = akkaProtocol;
+		}
+
+		private String getHostnameAndPort() {
+			return hostnameAndPort;
+		}
+
+		private AkkaProtocol getAkkaProtocol() {
+			return akkaProtocol;
+		}
+	}
+
+	private static String internalRpcUrl(String endpointName, Optional<RemoteAddressInformation> remoteAddressInformation) {
+		final String protocolPrefix = remoteAddressInformation.map(rai -> akkaProtocolToString(rai.getAkkaProtocol())).orElse("akka");
+		final Optional<String> optionalHostnameAndPort = remoteAddressInformation.map(RemoteAddressInformation::getHostnameAndPort);
+
+		final StringBuilder url = new StringBuilder(String.format("%s://flink", protocolPrefix));
+		optionalHostnameAndPort.ifPresent(hostPort -> url.append("@").append(hostPort));
+
+		url.append("/user/").append(SUPERVISOR_NAME).append("/").append(endpointName);
+
+		// protocolPrefix://flink[@hostname:port]/user/rpc/endpointName
+		return url.toString();
+	}
+
+	private static String akkaProtocolToString(AkkaProtocol akkaProtocol) {
+		return akkaProtocol == AkkaProtocol.SSL_TCP ? AKKA_SSL_TCP : AKKA_TCP;
 	}
 
 	/**
@@ -198,6 +235,16 @@ public class AkkaRpcServiceUtils {
 		return prefix + '_' + nameOffset;
 	}
 
+	/**
+	 * Creates a wildcard name symmetric to {@link #createRandomName(String)}.
+	 *
+	 * @param prefix prefix of the wildcard name
+	 * @return wildcard name starting with the prefix
+	 */
+	public static String createWildcardName(String prefix) {
+		return prefix + "_*";
+	}
+
 	// ------------------------------------------------------------------------
 	//  RPC service configuration
 	// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
index 5c104f4..eaf1d7e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
@@ -466,6 +466,27 @@ public class AkkaRpcActorTest extends TestLogger {
 		}
 	}
 
+	@Test
+	public void resolvesRunningAkkaRpcActor() throws Exception {
+		final String endpointName = "foobar";
+
+		try (RpcEndpoint simpleRpcEndpoint1 = createRpcEndpointWithRandomNameSuffix(endpointName);
+			RpcEndpoint simpleRpcEndpoint2 = createRpcEndpointWithRandomNameSuffix(endpointName)) {
+
+			simpleRpcEndpoint1.closeAsync().join();
+
+			final String wildcardName = AkkaRpcServiceUtils.createWildcardName(endpointName);
+			final String wildcardAddress = AkkaRpcServiceUtils.getLocalRpcUrl(wildcardName);
+			final RpcGateway rpcGateway = akkaRpcService.connect(wildcardAddress, RpcGateway.class).join();
+
+			assertThat(rpcGateway.getAddress(), is(equalTo(simpleRpcEndpoint2.getAddress())));
+		}
+	}
+
+	private RpcEndpoint createRpcEndpointWithRandomNameSuffix(String prefix) {
+		return new SimpleRpcEndpoint(akkaRpcService, AkkaRpcServiceUtils.createRandomName(prefix));
+	}
+
 	// ------------------------------------------------------------------------
 	//  Test Actors and Interfaces
 	// ------------------------------------------------------------------------

[flink] 07/08: [FLINK-17072] Let ResourceManager use random endpoint id

Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 04d6bc6fe54297bcff93cc130e60ef9074f3ad54
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Apr 9 15:45:50 2020 +0200

    [FLINK-17072] Let ResourceManager use random endpoint id
    
    This commit ensures that every ResourceManager started by the same JVM process
    will have a different endpoint id. This ensures that no new RM will receive
    messages sent to an old RM.
---
 .../kubernetes/KubernetesResourceManager.java      |  2 -
 .../KubernetesResourceManagerFactory.java          |  1 -
 .../kubernetes/KubernetesResourceManagerTest.java  |  3 -
 .../clusterframework/MesosResourceManager.java     |  2 -
 .../MesosResourceManagerFactory.java               |  1 -
 .../clusterframework/MesosResourceManagerTest.java |  1 -
 .../HighAvailabilityServicesUtils.java             |  2 +-
 .../resourcemanager/ActiveResourceManager.java     |  2 -
 .../runtime/resourcemanager/ResourceManager.java   |  4 +-
 .../resourcemanager/ResourceManagerFactory.java    | 10 ---
 .../resourcemanager/StandaloneResourceManager.java |  2 -
 .../StandaloneResourceManagerFactory.java          |  1 -
 .../StandaloneResourceManagerWithUUIDFactory.java  | 84 ----------------------
 .../runtime/minicluster/TestingMiniCluster.java    |  3 +-
 .../resourcemanager/ResourceManagerHATest.java     |  1 -
 .../ResourceManagerJobMasterTest.java              |  1 -
 .../ResourceManagerPartitionLifecycleTest.java     |  1 -
 .../ResourceManagerTaskExecutorTest.java           |  1 -
 .../resourcemanager/ResourceManagerTest.java       |  1 -
 .../StandaloneResourceManagerTest.java             |  4 --
 .../resourcemanager/TestingResourceManager.java    |  2 -
 .../org/apache/flink/yarn/YarnResourceManager.java |  2 -
 .../entrypoint/YarnResourceManagerFactory.java     |  1 -
 .../apache/flink/yarn/YarnResourceManagerTest.java |  4 --
 24 files changed, 5 insertions(+), 131 deletions(-)

diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManager.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManager.java
index 02975bc..29a487e 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManager.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManager.java
@@ -93,7 +93,6 @@ public class KubernetesResourceManager extends ActiveResourceManager<KubernetesW
 
 	public KubernetesResourceManager(
 			RpcService rpcService,
-			String resourceManagerEndpointId,
 			ResourceID resourceId,
 			Configuration flinkConfig,
 			HighAvailabilityServices highAvailabilityServices,
@@ -110,7 +109,6 @@ public class KubernetesResourceManager extends ActiveResourceManager<KubernetesW
 			flinkConfig,
 			System.getenv(),
 			rpcService,
-			resourceManagerEndpointId,
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesResourceManagerFactory.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesResourceManagerFactory.java
index a10dd6f..74edb72 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesResourceManagerFactory.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesResourceManagerFactory.java
@@ -81,7 +81,6 @@ public class KubernetesResourceManagerFactory extends ActiveResourceManagerFacto
 
 		return new KubernetesResourceManager(
 			rpcService,
-			getEndpointId(),
 			resourceId,
 			configuration,
 			highAvailabilityServices,
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesResourceManagerTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesResourceManagerTest.java
index 31ac358..a983af8 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesResourceManagerTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesResourceManagerTest.java
@@ -145,7 +145,6 @@ public class KubernetesResourceManagerTest extends KubernetesTestBase {
 
 		TestingKubernetesResourceManager(
 				RpcService rpcService,
-				String resourceManagerEndpointId,
 				ResourceID resourceId,
 				Configuration flinkConfig,
 				HighAvailabilityServices highAvailabilityServices,
@@ -159,7 +158,6 @@ public class KubernetesResourceManagerTest extends KubernetesTestBase {
 				KubernetesResourceManagerConfiguration configuration) {
 			super(
 				rpcService,
-				resourceManagerEndpointId,
 				resourceId,
 				flinkConfig,
 				highAvailabilityServices,
@@ -359,7 +357,6 @@ public class KubernetesResourceManagerTest extends KubernetesTestBase {
 
 		final TestingKubernetesResourceManager kubernetesResourceManager = new TestingKubernetesResourceManager(
 			rpcService,
-			"kubernetesResourceManager",
 			ResourceID.generate(),
 			configuration,
 			rmServices.highAvailabilityServices,
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
index 42515cc..02625b3 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
@@ -149,7 +149,6 @@ public class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerN
 	public MesosResourceManager(
 			// base class
 			RpcService rpcService,
-			String resourceManagerEndpointId,
 			ResourceID resourceId,
 			HighAvailabilityServices highAvailabilityServices,
 			HeartbeatServices heartbeatServices,
@@ -168,7 +167,6 @@ public class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerN
 			ResourceManagerMetricGroup resourceManagerMetricGroup) {
 		super(
 			rpcService,
-			resourceManagerEndpointId,
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerFactory.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerFactory.java
index 471dc1e..9aab235 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerFactory.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerFactory.java
@@ -84,7 +84,6 @@ public class MesosResourceManagerFactory extends ActiveResourceManagerFactory<Re
 
 		return new MesosResourceManager(
 			rpcService,
-			getEndpointId(),
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java
index 9590819..b9c6ac3 100644
--- a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java
+++ b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java
@@ -179,7 +179,6 @@ public class MesosResourceManagerTest extends TestLogger {
 			ResourceManagerMetricGroup resourceManagerMetricGroup) {
 			super(
 				rpcService,
-				resourceManagerEndpointId,
 				resourceId,
 				highAvailabilityServices,
 				heartbeatServices,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
index 4a915f0..3be16d8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java
@@ -96,7 +96,7 @@ public class HighAvailabilityServicesUtils {
 				final String resourceManagerRpcUrl = AkkaRpcServiceUtils.getRpcUrl(
 					hostnamePort.f0,
 					hostnamePort.f1,
-					ResourceManager.RESOURCE_MANAGER_NAME,
+					AkkaRpcServiceUtils.createWildcardName(ResourceManager.RESOURCE_MANAGER_NAME),
 					addressResolution,
 					configuration);
 				final String dispatcherRpcUrl = AkkaRpcServiceUtils.getRpcUrl(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ActiveResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ActiveResourceManager.java
index 65df9df..1ce3be6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ActiveResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ActiveResourceManager.java
@@ -67,7 +67,6 @@ public abstract class ActiveResourceManager <WorkerType extends ResourceIDRetrie
 			Configuration flinkConfig,
 			Map<String, String> env,
 			RpcService rpcService,
-			String resourceManagerEndpointId,
 			ResourceID resourceId,
 			HighAvailabilityServices highAvailabilityServices,
 			HeartbeatServices heartbeatServices,
@@ -79,7 +78,6 @@ public abstract class ActiveResourceManager <WorkerType extends ResourceIDRetrie
 			ResourceManagerMetricGroup resourceManagerMetricGroup) {
 		super(
 			rpcService,
-			resourceManagerEndpointId,
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index 537204b..2759a38 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -62,6 +62,7 @@ import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.FencedRpcEndpoint;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils;
 import org.apache.flink.runtime.taskexecutor.FileType;
 import org.apache.flink.runtime.taskexecutor.SlotReport;
 import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
@@ -156,7 +157,6 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
 
 	public ResourceManager(
 			RpcService rpcService,
-			String resourceManagerEndpointId,
 			ResourceID resourceId,
 			HighAvailabilityServices highAvailabilityServices,
 			HeartbeatServices heartbeatServices,
@@ -168,7 +168,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
 			ResourceManagerMetricGroup resourceManagerMetricGroup,
 			Time rpcTimeout) {
 
-		super(rpcService, resourceManagerEndpointId, null);
+		super(rpcService, AkkaRpcServiceUtils.createRandomName(RESOURCE_MANAGER_NAME), null);
 
 		this.resourceId = checkNotNull(resourceId);
 		this.highAvailabilityServices = checkNotNull(highAvailabilityServices);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java
index 113ca66..f7d5400 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java
@@ -30,8 +30,6 @@ import org.apache.flink.runtime.rpc.RpcService;
 
 import javax.annotation.Nullable;
 
-import java.util.UUID;
-
 /**
  * {@link ResourceManager} factory.
  *
@@ -49,12 +47,4 @@ public interface ResourceManagerFactory<T extends ResourceIDRetrievable> {
 		ClusterInformation clusterInformation,
 		@Nullable String webInterfaceUrl,
 		ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception;
-
-	default String generateEndpointIdWithUUID() {
-		return getEndpointId() + UUID.randomUUID();
-	}
-
-	default String getEndpointId() {
-		return ResourceManager.RESOURCE_MANAGER_NAME;
-	}
 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
index cffbffa..f188442 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
@@ -49,7 +49,6 @@ public class StandaloneResourceManager extends ResourceManager<ResourceID> {
 
 	public StandaloneResourceManager(
 			RpcService rpcService,
-			String resourceManagerEndpointId,
 			ResourceID resourceId,
 			HighAvailabilityServices highAvailabilityServices,
 			HeartbeatServices heartbeatServices,
@@ -63,7 +62,6 @@ public class StandaloneResourceManager extends ResourceManager<ResourceID> {
 			Time rpcTimeout) {
 		super(
 			rpcService,
-			resourceManagerEndpointId,
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactory.java
index 3401693..5fcd3f3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactory.java
@@ -61,7 +61,6 @@ public enum StandaloneResourceManagerFactory implements ResourceManagerFactory<R
 
 		return new StandaloneResourceManager(
 			rpcService,
-			getEndpointId(),
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/StandaloneResourceManagerWithUUIDFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/StandaloneResourceManagerWithUUIDFactory.java
deleted file mode 100644
index ae86165..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/StandaloneResourceManagerWithUUIDFactory.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.minicluster;
-
-import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ConfigurationUtils;
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.entrypoint.ClusterInformation;
-import org.apache.flink.runtime.heartbeat.HeartbeatServices;
-import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
-import org.apache.flink.runtime.io.network.partition.NoOpResourceManagerPartitionTracker;
-import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup;
-import org.apache.flink.runtime.resourcemanager.ArbitraryWorkerResourceSpecFactory;
-import org.apache.flink.runtime.resourcemanager.ResourceManager;
-import org.apache.flink.runtime.resourcemanager.ResourceManagerFactory;
-import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServices;
-import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServicesConfiguration;
-import org.apache.flink.runtime.resourcemanager.StandaloneResourceManager;
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.rpc.RpcUtils;
-
-import javax.annotation.Nullable;
-
-/**
- * {@link ResourceManagerFactory} which starts the {@link ResourceManager} with an endpoint id which
- * has a random UUID suffix.
- */
-public enum StandaloneResourceManagerWithUUIDFactory implements ResourceManagerFactory<ResourceID> {
-	INSTANCE;
-
-	@Override
-	public ResourceManager<ResourceID> createResourceManager(
-		Configuration configuration,
-		ResourceID resourceId,
-		RpcService rpcService,
-		HighAvailabilityServices highAvailabilityServices,
-		HeartbeatServices heartbeatServices,
-		FatalErrorHandler fatalErrorHandler,
-		ClusterInformation clusterInformation,
-		@Nullable String webInterfaceUrl,
-		ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception {
-		final ResourceManagerRuntimeServicesConfiguration resourceManagerRuntimeServicesConfiguration =
-			ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration, ArbitraryWorkerResourceSpecFactory.INSTANCE);
-		final ResourceManagerRuntimeServices resourceManagerRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(
-			resourceManagerRuntimeServicesConfiguration,
-			highAvailabilityServices,
-			rpcService.getScheduledExecutor());
-
-		final Time standaloneClusterStartupPeriodTime = ConfigurationUtils.getStandaloneClusterStartupPeriodTime(configuration);
-
-		return new StandaloneResourceManager(
-			rpcService,
-			generateEndpointIdWithUUID(),
-			resourceId,
-			highAvailabilityServices,
-			heartbeatServices,
-			resourceManagerRuntimeServices.getSlotManager(),
-			NoOpResourceManagerPartitionTracker::get,
-			resourceManagerRuntimeServices.getJobLeaderIdService(),
-			clusterInformation,
-			fatalErrorHandler,
-			resourceManagerMetricGroup,
-			standaloneClusterStartupPeriodTime,
-			RpcUtils.INF_TIMEOUT);
-	}
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java
index 4f2bd2f..7c1de5e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java
@@ -29,6 +29,7 @@ import org.apache.flink.runtime.entrypoint.component.TestingDefaultDispatcherRes
 import org.apache.flink.runtime.heartbeat.HeartbeatServices;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
 
@@ -132,6 +133,6 @@ public class TestingMiniCluster extends MiniCluster {
 	private DispatcherResourceManagerComponentFactory createTestingDispatcherResourceManagerComponentFactory() {
 		return TestingDefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory(
 			DefaultDispatcherRunnerFactory.createSessionRunner(SessionDispatcherWithUUIDFactory.INSTANCE),
-			StandaloneResourceManagerWithUUIDFactory.INSTANCE);
+			StandaloneResourceManagerFactory.INSTANCE);
 	}
 }
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java
index aeb0f8c..ff274b2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java
@@ -87,7 +87,6 @@ public class ResourceManagerHATest extends TestLogger {
 		final ResourceManager resourceManager =
 			new StandaloneResourceManager(
 				rpcService,
-				ResourceManager.RESOURCE_MANAGER_NAME,
 				rmResourceId,
 				highAvailabilityServices,
 				heartbeatServices,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerJobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerJobMasterTest.java
index 09ce92d..b3814fa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerJobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerJobMasterTest.java
@@ -143,7 +143,6 @@ public class ResourceManagerJobMasterTest extends TestLogger {
 
 		ResourceManager<?> resourceManager = new StandaloneResourceManager(
 			rpcService,
-			ResourceManager.RESOURCE_MANAGER_NAME,
 			rmResourceId,
 			haServices,
 			heartbeatServices,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerPartitionLifecycleTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerPartitionLifecycleTest.java
index aa20ff4..0aecdbc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerPartitionLifecycleTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerPartitionLifecycleTest.java
@@ -207,7 +207,6 @@ public class ResourceManagerPartitionLifecycleTest extends TestLogger {
 
 		final TestingResourceManager resourceManager = new TestingResourceManager(
 			rpcService,
-			ResourceManager.RESOURCE_MANAGER_NAME + UUID.randomUUID(),
 			ResourceID.generate(),
 			highAvailabilityServices,
 			new HeartbeatServices(100000L, 1000000L),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTaskExecutorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTaskExecutorTest.java
index b80f8c6..838235c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTaskExecutorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTaskExecutorTest.java
@@ -154,7 +154,6 @@ public class ResourceManagerTaskExecutorTest extends TestLogger {
 		StandaloneResourceManager resourceManager =
 			new StandaloneResourceManager(
 				rpcService,
-				ResourceManager.RESOURCE_MANAGER_NAME + UUID.randomUUID(),
 				resourceManagerResourceID,
 				highAvailabilityServices,
 				heartbeatServices,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
index 29c22b5..5571cca 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
@@ -257,7 +257,6 @@ public class ResourceManagerTest extends TestLogger {
 
 		final TestingResourceManager resourceManager = new TestingResourceManager(
 			rpcService,
-			ResourceManager.RESOURCE_MANAGER_NAME + UUID.randomUUID(),
 			resourceManagerResourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerTest.java
index 1974ede..1583e47 100755
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerTest.java
@@ -39,7 +39,6 @@ import org.apache.flink.util.TestLogger;
 import org.junit.ClassRule;
 import org.junit.Test;
 
-import java.util.UUID;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
@@ -114,7 +113,6 @@ public class StandaloneResourceManagerTest extends TestLogger {
 
 		final TestingStandaloneResourceManager rm = new TestingStandaloneResourceManager(
 			rmServices.rpcService,
-			UUID.randomUUID().toString(),
 			ResourceID.generate(),
 			rmServices.highAvailabilityServices,
 			rmServices.heartbeatServices,
@@ -137,7 +135,6 @@ public class StandaloneResourceManagerTest extends TestLogger {
 
 		private TestingStandaloneResourceManager(
 				RpcService rpcService,
-				String resourceManagerEndpointId,
 				ResourceID resourceId,
 				HighAvailabilityServices highAvailabilityServices,
 				HeartbeatServices heartbeatServices,
@@ -150,7 +147,6 @@ public class StandaloneResourceManagerTest extends TestLogger {
 				MockResourceManagerRuntimeServices rmServices) {
 			super(
 				rpcService,
-				resourceManagerEndpointId,
 				resourceId,
 				highAvailabilityServices,
 				heartbeatServices,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManager.java
index 0272daa..19e810b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManager.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManager.java
@@ -40,7 +40,6 @@ public class TestingResourceManager extends ResourceManager<ResourceID> {
 
 	public TestingResourceManager(
 			RpcService rpcService,
-			String resourceManagerEndpointId,
 			ResourceID resourceId,
 			HighAvailabilityServices highAvailabilityServices,
 			HeartbeatServices heartbeatServices,
@@ -51,7 +50,6 @@ public class TestingResourceManager extends ResourceManager<ResourceID> {
 			ResourceManagerMetricGroup resourceManagerMetricGroup) {
 		super(
 			rpcService,
-			resourceManagerEndpointId,
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index d032bb6..dc09349 100755
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -123,7 +123,6 @@ public class YarnResourceManager extends ActiveResourceManager<YarnWorkerNode>
 
 	public YarnResourceManager(
 			RpcService rpcService,
-			String resourceManagerEndpointId,
 			ResourceID resourceId,
 			Configuration flinkConfig,
 			Map<String, String> env,
@@ -140,7 +139,6 @@ public class YarnResourceManager extends ActiveResourceManager<YarnWorkerNode>
 			flinkConfig,
 			env,
 			rpcService,
-			resourceManagerEndpointId,
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnResourceManagerFactory.java b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnResourceManagerFactory.java
index 9da72c4..d5e3f25 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnResourceManagerFactory.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnResourceManagerFactory.java
@@ -70,7 +70,6 @@ public class YarnResourceManagerFactory extends ActiveResourceManagerFactory<Yar
 
 		return new YarnResourceManager(
 			rpcService,
-			getEndpointId(),
 			resourceId,
 			configuration,
 			System.getenv(),
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
index 122029c..08fa7ab 100755
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
@@ -185,7 +185,6 @@ public class YarnResourceManagerTest extends TestLogger {
 
 		TestingYarnResourceManager(
 				RpcService rpcService,
-				String resourceManagerEndpointId,
 				ResourceID resourceId,
 				Configuration flinkConfig,
 				Map<String, String> env,
@@ -201,7 +200,6 @@ public class YarnResourceManagerTest extends TestLogger {
 				ResourceManagerMetricGroup resourceManagerMetricGroup) {
 			super(
 				rpcService,
-				resourceManagerEndpointId,
 				resourceId,
 				flinkConfig,
 				env,
@@ -248,7 +246,6 @@ public class YarnResourceManagerTest extends TestLogger {
 
 		// RM
 		final ResourceID rmResourceID;
-		static final String RM_ADDRESS = "resourceManager";
 		final TestingYarnResourceManager resourceManager;
 
 		final int dataPort = 1234;
@@ -283,7 +280,6 @@ public class YarnResourceManagerTest extends TestLogger {
 			resourceManager =
 					new TestingYarnResourceManager(
 							rpcService,
-							RM_ADDRESS,
 							rmResourceID,
 							configuration,
 							env,