tillrohrmann commented on a change in pull request #13004:
URL: https://github.com/apache/flink/pull/13004#discussion_r469048052
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
##########
@@ -1064,6 +1076,13 @@ public void handleError(final Exception exception) {
*/
protected abstract void initialize() throws ResourceManagerException;
+ /**
+ * Terminates the framework specific components.
+ *
+ * @throws Throwable with occurs during termination.
+ */
+ protected abstract void terminate() throws Throwable;
Review comment:
Can we weaken the signature to `throws Exception`?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/active/ActiveResourceManager.java
##########
@@ -0,0 +1,287 @@
+/*
+ * 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.resourcemanager.active;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
+import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceIDRetrievable;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+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.ResourceManagerPartitionTrackerFactory;
+import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup;
+import org.apache.flink.runtime.resourcemanager.JobLeaderIdService;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.WorkerResourceSpec;
+import
org.apache.flink.runtime.resourcemanager.exceptions.ResourceManagerException;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * An active implementation of {@link ResourceManager}.
+ *
+ * <p>This resource manager actively requests and releases resources from/to
the external resource management frameworks.
+ * With different {@link ResourceManagerDriver} provided, this resource
manager can work with various frameworks.
+ */
+public class ActiveResourceManager<WorkerType extends ResourceIDRetrievable>
+ extends ResourceManager<WorkerType> implements
ResourceEventHandler<WorkerType> {
+
+ protected final Configuration flinkConfig;
+
+ private final ResourceManagerDriver<WorkerType> resourceManagerDriver;
+
+ /** All workers maintained by {@link ActiveResourceManager}. */
+ private final Map<ResourceID, WorkerType> workerNodeMap;
+
+ /** Number of requested and not registered workers per worker resource
spec. */
+ private final PendingWorkerCounter pendingWorkerCounter;
+
+ /** Identifiers and worker resource spec of requested not registered
workers. */
+ private final Map<ResourceID, WorkerResourceSpec>
currentAttemptUnregisteredWorkers;
+
+ public ActiveResourceManager(
+ ResourceManagerDriver<WorkerType> resourceManagerDriver,
+ Configuration flinkConfig,
+ RpcService rpcService,
+ ResourceID resourceId,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ SlotManager slotManager,
+ ResourceManagerPartitionTrackerFactory
clusterPartitionTrackerFactory,
+ JobLeaderIdService jobLeaderIdService,
+ ClusterInformation clusterInformation,
+ FatalErrorHandler fatalErrorHandler,
+ ResourceManagerMetricGroup resourceManagerMetricGroup) {
+ super(
+ rpcService,
+ resourceId,
+ highAvailabilityServices,
+ heartbeatServices,
+ slotManager,
+ clusterPartitionTrackerFactory,
+ jobLeaderIdService,
+ clusterInformation,
+ fatalErrorHandler,
+ resourceManagerMetricGroup,
+
AkkaUtils.getTimeoutAsTime(Preconditions.checkNotNull(flinkConfig)));
+
+ this.flinkConfig = flinkConfig;
+ this.resourceManagerDriver = resourceManagerDriver;
+ this.workerNodeMap = new HashMap<>();
+ this.pendingWorkerCounter = new PendingWorkerCounter();
+ this.currentAttemptUnregisteredWorkers = new HashMap<>();
+ }
+
+ //
------------------------------------------------------------------------
+ // ResourceManager
+ //
------------------------------------------------------------------------
+
+ @Override
+ protected void initialize() throws ResourceManagerException {
+ try {
+ resourceManagerDriver.initialize(
+ this,
+ command ->
getMainThreadExecutor().execute(command)); // always execute on the current
main thread executor.
+ } catch (Throwable t) {
+ throw new ResourceManagerException("Cannot initialize
resource provider.", t);
+ }
+ }
+
+ @Override
+ protected void terminate() throws ResourceManagerException {
+ try {
+ resourceManagerDriver.terminate().get();
+ } catch (Throwable t) {
+ throw new ResourceManagerException("Cannot terminate
resource provider.", t);
+ }
+ }
+
+ @Override
+ protected void internalDeregisterApplication(ApplicationStatus
finalStatus, @Nullable String optionalDiagnostics)
+ throws ResourceManagerException {
+ try {
+
resourceManagerDriver.deregisterApplication(finalStatus, optionalDiagnostics);
+ } catch (Throwable t) {
Review comment:
Same here.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/active/ActiveResourceManager.java
##########
@@ -0,0 +1,287 @@
+/*
+ * 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.resourcemanager.active;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
+import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceIDRetrievable;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+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.ResourceManagerPartitionTrackerFactory;
+import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup;
+import org.apache.flink.runtime.resourcemanager.JobLeaderIdService;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.WorkerResourceSpec;
+import
org.apache.flink.runtime.resourcemanager.exceptions.ResourceManagerException;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * An active implementation of {@link ResourceManager}.
+ *
+ * <p>This resource manager actively requests and releases resources from/to
the external resource management frameworks.
+ * With different {@link ResourceManagerDriver} provided, this resource
manager can work with various frameworks.
+ */
+public class ActiveResourceManager<WorkerType extends ResourceIDRetrievable>
+ extends ResourceManager<WorkerType> implements
ResourceEventHandler<WorkerType> {
+
+ protected final Configuration flinkConfig;
+
+ private final ResourceManagerDriver<WorkerType> resourceManagerDriver;
+
+ /** All workers maintained by {@link ActiveResourceManager}. */
+ private final Map<ResourceID, WorkerType> workerNodeMap;
+
+ /** Number of requested and not registered workers per worker resource
spec. */
+ private final PendingWorkerCounter pendingWorkerCounter;
+
+ /** Identifiers and worker resource spec of requested not registered
workers. */
+ private final Map<ResourceID, WorkerResourceSpec>
currentAttemptUnregisteredWorkers;
+
+ public ActiveResourceManager(
+ ResourceManagerDriver<WorkerType> resourceManagerDriver,
+ Configuration flinkConfig,
+ RpcService rpcService,
+ ResourceID resourceId,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ SlotManager slotManager,
+ ResourceManagerPartitionTrackerFactory
clusterPartitionTrackerFactory,
+ JobLeaderIdService jobLeaderIdService,
+ ClusterInformation clusterInformation,
+ FatalErrorHandler fatalErrorHandler,
+ ResourceManagerMetricGroup resourceManagerMetricGroup) {
+ super(
+ rpcService,
+ resourceId,
+ highAvailabilityServices,
+ heartbeatServices,
+ slotManager,
+ clusterPartitionTrackerFactory,
+ jobLeaderIdService,
+ clusterInformation,
+ fatalErrorHandler,
+ resourceManagerMetricGroup,
+
AkkaUtils.getTimeoutAsTime(Preconditions.checkNotNull(flinkConfig)));
+
+ this.flinkConfig = flinkConfig;
+ this.resourceManagerDriver = resourceManagerDriver;
+ this.workerNodeMap = new HashMap<>();
+ this.pendingWorkerCounter = new PendingWorkerCounter();
+ this.currentAttemptUnregisteredWorkers = new HashMap<>();
+ }
+
+ //
------------------------------------------------------------------------
+ // ResourceManager
+ //
------------------------------------------------------------------------
+
+ @Override
+ protected void initialize() throws ResourceManagerException {
+ try {
+ resourceManagerDriver.initialize(
+ this,
+ command ->
getMainThreadExecutor().execute(command)); // always execute on the current
main thread executor.
+ } catch (Throwable t) {
+ throw new ResourceManagerException("Cannot initialize
resource provider.", t);
+ }
+ }
+
+ @Override
+ protected void terminate() throws ResourceManagerException {
+ try {
+ resourceManagerDriver.terminate().get();
+ } catch (Throwable t) {
Review comment:
Same here with `Throwable`.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/active/ActiveResourceManager.java
##########
@@ -0,0 +1,287 @@
+/*
+ * 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.resourcemanager.active;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
+import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceIDRetrievable;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+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.ResourceManagerPartitionTrackerFactory;
+import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup;
+import org.apache.flink.runtime.resourcemanager.JobLeaderIdService;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.WorkerResourceSpec;
+import
org.apache.flink.runtime.resourcemanager.exceptions.ResourceManagerException;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * An active implementation of {@link ResourceManager}.
+ *
+ * <p>This resource manager actively requests and releases resources from/to
the external resource management frameworks.
+ * With different {@link ResourceManagerDriver} provided, this resource
manager can work with various frameworks.
+ */
+public class ActiveResourceManager<WorkerType extends ResourceIDRetrievable>
+ extends ResourceManager<WorkerType> implements
ResourceEventHandler<WorkerType> {
+
+ protected final Configuration flinkConfig;
+
+ private final ResourceManagerDriver<WorkerType> resourceManagerDriver;
+
+ /** All workers maintained by {@link ActiveResourceManager}. */
+ private final Map<ResourceID, WorkerType> workerNodeMap;
+
+ /** Number of requested and not registered workers per worker resource
spec. */
+ private final PendingWorkerCounter pendingWorkerCounter;
+
+ /** Identifiers and worker resource spec of requested not registered
workers. */
+ private final Map<ResourceID, WorkerResourceSpec>
currentAttemptUnregisteredWorkers;
+
+ public ActiveResourceManager(
+ ResourceManagerDriver<WorkerType> resourceManagerDriver,
+ Configuration flinkConfig,
+ RpcService rpcService,
+ ResourceID resourceId,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ SlotManager slotManager,
+ ResourceManagerPartitionTrackerFactory
clusterPartitionTrackerFactory,
+ JobLeaderIdService jobLeaderIdService,
+ ClusterInformation clusterInformation,
+ FatalErrorHandler fatalErrorHandler,
+ ResourceManagerMetricGroup resourceManagerMetricGroup) {
+ super(
+ rpcService,
+ resourceId,
+ highAvailabilityServices,
+ heartbeatServices,
+ slotManager,
+ clusterPartitionTrackerFactory,
+ jobLeaderIdService,
+ clusterInformation,
+ fatalErrorHandler,
+ resourceManagerMetricGroup,
+
AkkaUtils.getTimeoutAsTime(Preconditions.checkNotNull(flinkConfig)));
+
+ this.flinkConfig = flinkConfig;
+ this.resourceManagerDriver = resourceManagerDriver;
+ this.workerNodeMap = new HashMap<>();
+ this.pendingWorkerCounter = new PendingWorkerCounter();
+ this.currentAttemptUnregisteredWorkers = new HashMap<>();
+ }
+
+ //
------------------------------------------------------------------------
+ // ResourceManager
+ //
------------------------------------------------------------------------
+
+ @Override
+ protected void initialize() throws ResourceManagerException {
+ try {
+ resourceManagerDriver.initialize(
+ this,
+ command ->
getMainThreadExecutor().execute(command)); // always execute on the current
main thread executor.
Review comment:
Just as a note: If the `ResourceManagerDriver` tries to run something on
this executor while the `ResourceManager` has no leadership, then it will
simply be ignored and not executed. Is this ok with the current behaviour of
the `ResourceManagerDriver`?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/active/ActiveResourceManager.java
##########
@@ -0,0 +1,287 @@
+/*
+ * 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.resourcemanager.active;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
+import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceIDRetrievable;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+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.ResourceManagerPartitionTrackerFactory;
+import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup;
+import org.apache.flink.runtime.resourcemanager.JobLeaderIdService;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.WorkerResourceSpec;
+import
org.apache.flink.runtime.resourcemanager.exceptions.ResourceManagerException;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * An active implementation of {@link ResourceManager}.
+ *
+ * <p>This resource manager actively requests and releases resources from/to
the external resource management frameworks.
+ * With different {@link ResourceManagerDriver} provided, this resource
manager can work with various frameworks.
+ */
+public class ActiveResourceManager<WorkerType extends ResourceIDRetrievable>
+ extends ResourceManager<WorkerType> implements
ResourceEventHandler<WorkerType> {
+
+ protected final Configuration flinkConfig;
+
+ private final ResourceManagerDriver<WorkerType> resourceManagerDriver;
+
+ /** All workers maintained by {@link ActiveResourceManager}. */
+ private final Map<ResourceID, WorkerType> workerNodeMap;
+
+ /** Number of requested and not registered workers per worker resource
spec. */
+ private final PendingWorkerCounter pendingWorkerCounter;
+
+ /** Identifiers and worker resource spec of requested not registered
workers. */
+ private final Map<ResourceID, WorkerResourceSpec>
currentAttemptUnregisteredWorkers;
+
+ public ActiveResourceManager(
+ ResourceManagerDriver<WorkerType> resourceManagerDriver,
+ Configuration flinkConfig,
+ RpcService rpcService,
+ ResourceID resourceId,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ SlotManager slotManager,
+ ResourceManagerPartitionTrackerFactory
clusterPartitionTrackerFactory,
+ JobLeaderIdService jobLeaderIdService,
+ ClusterInformation clusterInformation,
+ FatalErrorHandler fatalErrorHandler,
+ ResourceManagerMetricGroup resourceManagerMetricGroup) {
+ super(
+ rpcService,
+ resourceId,
+ highAvailabilityServices,
+ heartbeatServices,
+ slotManager,
+ clusterPartitionTrackerFactory,
+ jobLeaderIdService,
+ clusterInformation,
+ fatalErrorHandler,
+ resourceManagerMetricGroup,
+
AkkaUtils.getTimeoutAsTime(Preconditions.checkNotNull(flinkConfig)));
+
+ this.flinkConfig = flinkConfig;
+ this.resourceManagerDriver = resourceManagerDriver;
+ this.workerNodeMap = new HashMap<>();
+ this.pendingWorkerCounter = new PendingWorkerCounter();
+ this.currentAttemptUnregisteredWorkers = new HashMap<>();
+ }
+
+ //
------------------------------------------------------------------------
+ // ResourceManager
+ //
------------------------------------------------------------------------
+
+ @Override
+ protected void initialize() throws ResourceManagerException {
+ try {
+ resourceManagerDriver.initialize(
+ this,
+ command ->
getMainThreadExecutor().execute(command)); // always execute on the current
main thread executor.
+ } catch (Throwable t) {
Review comment:
I'm wondering whether we should really catch all `Throwables` here. I
think that JVM `Errors` should not be caught at this point.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
