[FLINK-25402] Make JobManager process resource id configurable via jo…

…bmanager.resource-id
silenceland · Jan 19, 2022 · f107710 · f107710
1 parent 6bbf57f
commit f107710
Show file tree

Hide file tree

Showing 15 changed files with 70 additions and 5 deletions.
diff --git a/docs/layouts/shortcodes/generated/all_jobmanager_section.html b/docs/layouts/shortcodes/generated/all_jobmanager_section.html
@@ -56,6 +56,12 @@
             <td>Integer</td>
             <td>The size of the IO thread pool to run blocking operations for all spawned JobMasters. This includes recovery and completion of checkpoints. Increase this value if you experience slow checkpoint operations when running many jobs. If no value is specified, then Flink defaults to the number of available CPU cores.</td>
         </tr>
+        <tr>
+            <td><h5>jobmanager.resource-id</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The JobManager's ResourceID. If not configured, the ResourceID will be generated randomly.</td>
+        </tr>
         <tr>
             <td><h5>jobmanager.retrieve-taskmanager-hostname</h5></td>
             <td style="word-wrap: break-word;">true</td>

diff --git a/docs/layouts/shortcodes/generated/job_manager_configuration.html b/docs/layouts/shortcodes/generated/job_manager_configuration.html
@@ -116,6 +116,12 @@
             <td>MemorySize</td>
             <td>Total Process Memory size for the JobManager. This includes all the memory that a JobManager JVM process consumes, consisting of Total Flink Memory, JVM Metaspace, and JVM Overhead. In containerized setups, this should be set to the container memory. See also 'jobmanager.memory.flink.size' for Total Flink Memory size configuration.</td>
         </tr>
+        <tr>
+            <td><h5>jobmanager.resource-id</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The JobManager's ResourceID. If not configured, the ResourceID will be generated randomly.</td>
+        </tr>
         <tr>
             <td><h5>jobmanager.retrieve-taskmanager-hostname</h5></td>
             <td style="word-wrap: break-word;">true</td>

diff --git a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
@@ -547,6 +547,17 @@ public enum SchedulerType {
                                             code(SchedulerType.AdaptiveBatch.name()))
                                     .build());
 
+    /**
+     * The JobManager's ResourceID. If not configured, the ResourceID will be generated randomly.
+     */
+    @Documentation.Section(Documentation.Sections.ALL_JOB_MANAGER)
+    public static final ConfigOption<String> JOB_MANAGER_RESOURCE_ID =
+            key("jobmanager.resource-id")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The JobManager's ResourceID. If not configured, the ResourceID will be generated randomly.");
+
     // ---------------------------------------------------------------------------------------------
 
     private JobManagerOptions() {

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -39,6 +39,7 @@
 import org.apache.flink.management.jmx.JMXService;
 import org.apache.flink.runtime.blob.BlobServer;
 import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.dispatcher.ExecutionGraphInfoStore;
 import org.apache.flink.runtime.dispatcher.MiniDispatcher;
 import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent;
@@ -123,6 +124,9 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro
 
     private final AtomicBoolean isShutDown = new AtomicBoolean(false);
 
+    @GuardedBy("lock")
+    private ResourceID resourceId;
+
     @GuardedBy("lock")
     private DispatcherResourceManagerComponent clusterComponent;
 
@@ -253,6 +257,7 @@ private void runCluster(Configuration configuration, PluginManager pluginManager
             clusterComponent =
                     dispatcherResourceManagerComponentFactory.create(
                             configuration,
+                            resourceId,
                             ioExecutor,
                             commonRpcService,
                             haServices,
@@ -338,6 +343,17 @@ protected void initializeServices(Configuration configuration, PluginManager plu
             executionGraphInfoStore =
                     createSerializableExecutionGraphStore(
                             configuration, commonRpcService.getScheduledExecutor());
+
+            resourceId =
+                    configuration
+                            .getOptional(JobManagerOptions.JOB_MANAGER_RESOURCE_ID)
+                            .map(ResourceID::new)
+                            .orElseGet(ResourceID::generate);
+
+            LOG.debug(
+                    "Initialize cluster entrypoint {} with resource id {}.",
+                    getClass().getSimpleName(),
+                    resourceId);
         }
     }
 

diff --git a/.../flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java b/.../flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java
@@ -22,6 +22,7 @@
 import org.apache.flink.configuration.MetricOptions;
 import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.dispatcher.DispatcherGateway;
 import org.apache.flink.runtime.dispatcher.DispatcherId;
 import org.apache.flink.runtime.dispatcher.DispatcherOperationCaches;
@@ -101,6 +102,7 @@ public DefaultDispatcherResourceManagerComponentFactory(
     @Override
     public DispatcherResourceManagerComponent create(
             Configuration configuration,
+            ResourceID resourceId,
             Executor ioExecutor,
             RpcService rpcService,
             HighAvailabilityServices highAvailabilityServices,
@@ -178,6 +180,7 @@ public DispatcherResourceManagerComponent create(
                     ResourceManagerServiceImpl.create(
                             resourceManagerFactory,
                             configuration,
+                            resourceId,
                             rpcService,
                             highAvailabilityServices,
                             heartbeatServices,

diff --git a/.../apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponentFactory.java b/.../apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponentFactory.java
@@ -20,6 +20,7 @@
 
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.dispatcher.ExecutionGraphInfoStore;
 import org.apache.flink.runtime.heartbeat.HeartbeatServices;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -35,6 +36,7 @@ public interface DispatcherResourceManagerComponentFactory {
 
     DispatcherResourceManagerComponent create(
             Configuration configuration,
+            ResourceID resourceId,
             Executor ioExecutor,
             RpcService rpcService,
             HighAvailabilityServices highAvailabilityServices,

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
@@ -482,6 +482,7 @@ private void setupDispatcherResourceManagerComponents(
         return Collections.singleton(
                 dispatcherResourceManagerComponentFactory.create(
                         configuration,
+                        ResourceID.generate(),
                         ioExecutor,
                         rpcServiceFactory.createRpcService(),
                         haServices,

diff --git a/...untime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java b/...untime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java
@@ -50,6 +50,7 @@ public abstract class ResourceManagerFactory<T extends ResourceIDRetrievable> {
 
     public ResourceManagerProcessContext createResourceManagerProcessContext(
             Configuration configuration,
+            ResourceID resourceId,
             RpcService rpcService,
             HighAvailabilityServices highAvailabilityServices,
             HeartbeatServices heartbeatServices,
@@ -77,6 +78,7 @@ public ResourceManagerProcessContext createResourceManagerProcessContext(
 
         return new ResourceManagerProcessContext(
                 rmConfig,
+                resourceId,
                 runtimeServiceConfig,
                 rpcService,
                 highAvailabilityServices,
@@ -90,8 +92,7 @@ public ResourceManagerProcessContext createResourceManagerProcessContext(
     }
 
     public ResourceManager<T> createResourceManager(
-            ResourceManagerProcessContext context, UUID leaderSessionId, ResourceID resourceId)
-            throws Exception {
+            ResourceManagerProcessContext context, UUID leaderSessionId) throws Exception {
 
         final ResourceManagerRuntimeServices resourceManagerRuntimeServices =
                 createResourceManagerRuntimeServices(
@@ -102,7 +103,7 @@ public ResourceManager<T> createResourceManager(
 
         return createResourceManager(
                 context.getRmConfig(),
-                resourceId,
+                context.getResourceId(),
                 context.getRpcService(),
                 leaderSessionId,
                 context.getHeartbeatServices(),

diff --git a/...src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerProcessContext.java b/...src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerProcessContext.java
@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.resourcemanager;
 
 import org.apache.flink.configuration.Configuration;
+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;
@@ -41,6 +42,7 @@
  */
 public class ResourceManagerProcessContext {
     private final Configuration rmConfig;
+    private final ResourceID resourceId;
     private final ResourceManagerRuntimeServicesConfiguration rmRuntimeServicesConfig;
     private final RpcService rpcService;
     private final HighAvailabilityServices highAvailabilityServices;
@@ -54,6 +56,7 @@ public class ResourceManagerProcessContext {
 
     public ResourceManagerProcessContext(
             Configuration rmConfig,
+            ResourceID resourceId,
             ResourceManagerRuntimeServicesConfiguration rmRuntimeServicesConfig,
             RpcService rpcService,
             HighAvailabilityServices highAvailabilityServices,
@@ -65,6 +68,7 @@ public ResourceManagerProcessContext(
             SlotManagerMetricGroup slotManagerMetricGroup,
             Executor ioExecutor) {
         this.rmConfig = checkNotNull(rmConfig);
+        this.resourceId = checkNotNull(resourceId);
         this.rmRuntimeServicesConfig = checkNotNull(rmRuntimeServicesConfig);
         this.rpcService = checkNotNull(rpcService);
         this.highAvailabilityServices = checkNotNull(highAvailabilityServices);
@@ -82,6 +86,10 @@ public Configuration getRmConfig() {
         return rmConfig;
     }
 
+    public ResourceID getResourceId() {
+        return resourceId;
+    }
+
     public ResourceManagerRuntimeServicesConfiguration getRmRuntimeServicesConfig() {
         return rmRuntimeServicesConfig;
     }

diff --git a/...me/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImpl.java b/...me/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImpl.java
@@ -242,8 +242,7 @@ private void startNewLeaderResourceManager(UUID newLeaderSessionID) throws Excep
 
         this.leaderSessionID = newLeaderSessionID;
         this.leaderResourceManager =
-                resourceManagerFactory.createResourceManager(
-                        rmProcessContext, newLeaderSessionID, ResourceID.generate());
+                resourceManagerFactory.createResourceManager(rmProcessContext, newLeaderSessionID);
 
         final ResourceManager<?> newLeaderResourceManager = this.leaderResourceManager;
 
@@ -334,6 +333,7 @@ public ResourceManager<?> getLeaderResourceManager() {
     public static ResourceManagerServiceImpl create(
             ResourceManagerFactory<?> resourceManagerFactory,
             Configuration configuration,
+            ResourceID resourceId,
             RpcService rpcService,
             HighAvailabilityServices highAvailabilityServices,
             HeartbeatServices heartbeatServices,
@@ -349,6 +349,7 @@ public static ResourceManagerServiceImpl create(
                 resourceManagerFactory,
                 resourceManagerFactory.createResourceManagerProcessContext(
                         configuration,
+                        resourceId,
                         rpcService,
                         highAvailabilityServices,
                         heartbeatServices,

diff --git a/...me/src/test/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStoreTest.java b/...me/src/test/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStoreTest.java
@@ -24,6 +24,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.testutils.OneShotLatch;
 import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent;
 import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory;
@@ -409,6 +410,7 @@ private class PersistingMiniCluster extends MiniCluster {
             return Collections.singleton(
                     dispatcherResourceManagerComponentFactory.create(
                             configuration,
+                            ResourceID.generate(),
                             getIOExecutor(),
                             rpcServiceFactory.createRpcService(),
                             haServices,

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
@@ -20,6 +20,7 @@
 
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.dispatcher.DispatcherGateway;
 import org.apache.flink.runtime.dispatcher.MemoryExecutionGraphInfoStore;
 import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent;
@@ -156,6 +157,7 @@ protected HighAvailabilityServices createHighAvailabilityServices(
             result.add(
                     dispatcherResourceManagerComponentFactory.create(
                             configuration,
+                            ResourceID.generate(),
                             getIOExecutor(),
                             rpcServiceFactory.createRpcService(),
                             haServices,

diff --git a/...rc/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImplTest.java b/...rc/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImplTest.java
@@ -20,6 +20,7 @@
 
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.Configuration;
+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.heartbeat.TestingHeartbeatServices;
@@ -126,6 +127,7 @@ private void createResourceManager() throws Exception {
                 ResourceManagerServiceImpl.create(
                         rmFactory,
                         new Configuration(),
+                        ResourceID.generate(),
                         rpcService,
                         haService,
                         heartbeatServices,

diff --git a/...src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManagerService.java b/...src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManagerService.java
@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.entrypoint.ClusterInformation;
 import org.apache.flink.runtime.heartbeat.TestingHeartbeatServices;
 import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
@@ -184,6 +185,7 @@ public TestingResourceManagerService build() throws Exception {
                     ResourceManagerServiceImpl.create(
                             StandaloneResourceManagerFactory.getInstance(),
                             new Configuration(),
+                            ResourceID.generate(),
                             rpcService,
                             haServices,
                             new TestingHeartbeatServices(),

diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java
@@ -33,6 +33,7 @@
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.core.testutils.CommonTestUtils;
 import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.dispatcher.MemoryExecutionGraphInfoStore;
 import org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent;
@@ -140,6 +141,7 @@ public void testCancelingOnProcessFailure() throws Throwable {
             dispatcherResourceManagerComponent =
                     resourceManagerComponentFactory.create(
                             config,
+                            ResourceID.generate(),
                             ioExecutor,
                             rpcService,
                             haServices,