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 2019/02/12 20:57:58 UTC
[flink] 04/05: [hotfix] Replace RpcEndpoint#shutDown with
RpcEndpoint#closeAsync method
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 dbb175539a959a09bc8aefa20b3f7971d0b54627
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Feb 12 15:58:36 2019 +0100
[hotfix] Replace RpcEndpoint#shutDown with RpcEndpoint#closeAsync method
---
.../flink/runtime/dispatcher/Dispatcher.java | 2 +-
...tDispatcherResourceManagerComponentFactory.java | 6 ++--
.../DispatcherResourceManagerComponent.java | 6 ++--
.../apache/flink/runtime/jobmaster/JobMaster.java | 10 +------
.../flink/runtime/minicluster/MiniCluster.java | 5 ++--
.../resourcemanager/ResourceManagerRunner.java | 12 +-------
.../org/apache/flink/runtime/rpc/RpcEndpoint.java | 10 +++----
.../org/apache/flink/runtime/rpc/RpcUtils.java | 3 +-
.../runtime/taskexecutor/TaskManagerRunner.java | 3 +-
.../dispatcher/DispatcherResourceCleanupTest.java | 5 ++--
.../flink/runtime/dispatcher/DispatcherTest.java | 3 +-
.../dispatcher/ZooKeeperHADispatcherTest.java | 4 +--
.../heartbeat/TestingHeartbeatServices.java | 5 ++++
.../jobmaster/slotpool/SlotPoolResource.java | 6 +---
.../runtime/jobmaster/slotpool/SlotPoolTest.java | 9 +++---
.../resourcemanager/ResourceManagerHATest.java | 14 ++++------
.../apache/flink/runtime/rpc/AsyncCallsTest.java | 5 ++--
.../flink/runtime/rpc/FencedRpcEndpointTest.java | 15 ++++------
.../apache/flink/runtime/rpc/RpcEndpointTest.java | 6 ++--
.../flink/runtime/rpc/akka/AkkaRpcActorTest.java | 32 ++++++++++------------
.../runtime/rpc/akka/MainThreadValidationTest.java | 4 +--
.../runtime/taskexecutor/TaskExecutorTest.java | 18 ++++--------
.../retriever/impl/RpcGatewayRetrieverTest.java | 6 ++--
.../recovery/ProcessFailureCancelingITCase.java | 4 +--
.../org/apache/flink/yarn/YarnResourceManager.java | 2 +-
.../apache/flink/yarn/YarnResourceManagerTest.java | 5 +---
26 files changed, 72 insertions(+), 128 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 39a496d..47796b0 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
@@ -542,7 +542,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
@Override
public CompletableFuture<Acknowledge> shutDownCluster() {
- shutDown();
+ closeAsync();
return CompletableFuture.completedFuture(Acknowledge.get());
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java
index b350719..81138ae 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java
@@ -235,13 +235,11 @@ public abstract class AbstractDispatcherResourceManagerComponentFactory<T extend
}
if (resourceManager != null) {
- resourceManager.shutDown();
- terminationFutures.add(resourceManager.getTerminationFuture());
+ terminationFutures.add(resourceManager.closeAsync());
}
if (dispatcher != null) {
- dispatcher.shutDown();
- terminationFutures.add(dispatcher.getTerminationFuture());
+ terminationFutures.add(dispatcher.closeAsync());
}
final FutureUtils.ConjunctFuture<Void> terminationFuture = FutureUtils.completeAll(terminationFutures);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
index e9cd189..700b3c4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
@@ -169,11 +169,9 @@ public class DispatcherResourceManagerComponent<T extends Dispatcher> implements
exception = ExceptionUtils.firstOrSuppressed(e, exception);
}
- dispatcher.shutDown();
- terminationFutures.add(dispatcher.getTerminationFuture());
+ terminationFutures.add(dispatcher.closeAsync());
- resourceManager.shutDown();
- terminationFutures.add(resourceManager.getTerminationFuture());
+ terminationFutures.add(resourceManager.closeAsync());
if (exception != null) {
terminationFutures.add(FutureUtils.completedExceptionally(exception));
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index e2030a6..6eaec3d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -366,7 +366,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
suspendExecution(new FlinkException("JobManager is shutting down."));
// shut down will internally release all registered slots
- slotPool.shutDown();
+ final CompletableFuture<Void> slotPoolTerminationFuture = slotPool.closeAsync();
final CompletableFuture<Void> disposeInternalSavepointFuture;
@@ -376,8 +376,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
disposeInternalSavepointFuture = CompletableFuture.completedFuture(null);
}
- final CompletableFuture<Void> slotPoolTerminationFuture = slotPool.getTerminationFuture();
-
return FutureUtils.completeAll(Arrays.asList(disposeInternalSavepointFuture, slotPoolTerminationFuture));
}
@@ -1511,12 +1509,6 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
return getSelfGateway(JobMasterGateway.class);
}
- @Override
- public CompletableFuture<Void> closeAsync() {
- shutDown();
- return getTerminationFuture();
- }
-
//----------------------------------------------------------------------------------------------
// Utility classes
//----------------------------------------------------------------------------------------------
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 fe20237..63fa09c 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
@@ -413,8 +413,7 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
if (taskManagers != null) {
for (TaskExecutor tm : taskManagers) {
if (tm != null) {
- tm.shutDown();
- componentTerminationFutures.add(tm.getTerminationFuture());
+ componentTerminationFutures.add(tm.closeAsync());
}
}
taskManagers = null;
@@ -931,7 +930,7 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
if (currentTaskManagers != null) {
// the shutDown is asynchronous
- currentTaskManagers[index].shutDown();
+ currentTaskManagers[index].closeAsync();
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java
index 4954102..e3e6d72 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java
@@ -43,8 +43,6 @@ public class ResourceManagerRunner implements FatalErrorHandler, AutoCloseableAs
private static final Logger LOG = LoggerFactory.getLogger(ResourceManagerRunner.class);
- private final Object lock = new Object();
-
private final ResourceManagerRuntimeServices resourceManagerRuntimeServices;
private final ResourceManager<?> resourceManager;
@@ -88,10 +86,6 @@ public class ResourceManagerRunner implements FatalErrorHandler, AutoCloseableAs
jobManagerMetricGroup);
}
- public ResourceManagerGateway getResourceManageGateway() {
- return resourceManager.getSelfGateway(ResourceManagerGateway.class);
- }
-
//-------------------------------------------------------------------------------------
// Lifecycle management
//-------------------------------------------------------------------------------------
@@ -102,11 +96,7 @@ public class ResourceManagerRunner implements FatalErrorHandler, AutoCloseableAs
@Override
public CompletableFuture<Void> closeAsync() {
- synchronized (lock) {
- resourceManager.shutDown();
-
- return resourceManager.getTerminationFuture();
- }
+ return resourceManager.closeAsync();
}
//-------------------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
index 764f969..13a4d65 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.rpc;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.runtime.concurrent.ScheduledFutureAdapter;
+import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -56,7 +57,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* <p>The RPC endpoint provides {@link #runAsync(Runnable)}, {@link #callAsync(Callable, Time)}
* and the {@link #getMainThreadExecutor()} to execute code in the RPC endpoint's main thread.
*/
-public abstract class RpcEndpoint implements RpcGateway {
+public abstract class RpcEndpoint implements RpcGateway, AutoCloseableAsync {
protected final Logger log = LoggerFactory.getLogger(getClass());
@@ -157,11 +158,8 @@ public abstract class RpcEndpoint implements RpcGateway {
* <p>In order to wait on the completion of the shut down, obtain the termination future
* via {@link #getTerminationFuture()}} and wait on its completion.
*/
- public final void shutDown() {
- rpcService.stopServer(rpcServer);
- }
-
- public final CompletableFuture<Void> terminate() {
+ @Override
+ public final CompletableFuture<Void> closeAsync() {
rpcService.stopServer(rpcServer);
return getTerminationFuture();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcUtils.java
index 2f656d0..d79a170 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcUtils.java
@@ -74,8 +74,7 @@ public class RpcUtils {
* @throws TimeoutException if a timeout occurred
*/
public static void terminateRpcEndpoint(RpcEndpoint rpcEndpoint, Time timeout) throws ExecutionException, InterruptedException, TimeoutException {
- rpcEndpoint.shutDown();
- rpcEndpoint.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ rpcEndpoint.closeAsync().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
}
/**
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index 2f5ecbf..3161e33 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -176,8 +176,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
if (!shutdown) {
shutdown = true;
- taskManager.shutDown();
- final CompletableFuture<Void> taskManagerTerminationFuture = taskManager.getTerminationFuture();
+ final CompletableFuture<Void> taskManagerTerminationFuture = taskManager.closeAsync();
final CompletableFuture<Void> serviceTerminationFuture = FutureUtils.composeAfterwards(
taskManagerTerminationFuture,
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 41b719e..3dd6428 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
@@ -208,8 +208,7 @@ public class DispatcherResourceCleanupTest extends TestLogger {
@After
public void teardown() throws Exception {
if (dispatcher != null) {
- dispatcher.shutDown();
- dispatcher.getTerminationFuture().get();
+ dispatcher.close();
}
if (fatalErrorHandler != null) {
@@ -294,7 +293,7 @@ public class DispatcherResourceCleanupTest extends TestLogger {
public void testBlobServerCleanupWhenClosingDispatcher() throws Exception {
submitJob();
- dispatcher.shutDown();
+ dispatcher.closeAsync();
terminationFuture.complete(null);
dispatcher.getTerminationFuture().get();
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 15e5fc9..2c3712f 100644
--- 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
@@ -680,8 +680,7 @@ public class DispatcherTest extends TestLogger {
submissionFuture.get();
assertThat(dispatcher.getNumberJobs(TIMEOUT).get(), Matchers.is(1));
- dispatcher.shutDown();
- dispatcher.getTerminationFuture().get();
+ dispatcher.close();
assertThat(submittedJobGraphStore.contains(jobGraph.getJobID()), Matchers.is(true));
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ZooKeeperHADispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ZooKeeperHADispatcherTest.java
index 6ca17bf..4dff454 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ZooKeeperHADispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ZooKeeperHADispatcherTest.java
@@ -302,10 +302,10 @@ public class ZooKeeperHADispatcherTest extends TestLogger {
dispatcherGateway.submitJob(nonEmptyJobGraph, TIMEOUT).get();
if (dispatcher1.getAddress().equals(leaderConnectionInfo.getAddress())) {
- dispatcher1.shutDown();
+ dispatcher1.closeAsync();
assertThat(jobGraphFuture2.get().getJobID(), is(equalTo(nonEmptyJobGraph.getJobID())));
} else {
- dispatcher2.shutDown();
+ dispatcher2.closeAsync();
assertThat(jobGraphFuture1.get().getJobID(), is(equalTo(nonEmptyJobGraph.getJobID())));
}
} finally {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/TestingHeartbeatServices.java b/flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/TestingHeartbeatServices.java
index 8bf663c..a98f45a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/TestingHeartbeatServices.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/TestingHeartbeatServices.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.heartbeat;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -37,6 +38,10 @@ public class TestingHeartbeatServices extends HeartbeatServices {
this.scheduledExecutorToUse = Preconditions.checkNotNull(scheduledExecutorToUse);
}
+ public TestingHeartbeatServices() {
+ this(1000L, 10000L, TestingUtils.defaultScheduledExecutor());
+ }
+
@Override
public <I, O> HeartbeatManager<I, O> createHeartbeatManagerSender(
ResourceID resourceId,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolResource.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolResource.java
index b58b98a..fdcade2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolResource.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolResource.java
@@ -27,8 +27,6 @@ import org.junit.rules.ExternalResource;
import javax.annotation.Nonnull;
-import java.util.concurrent.CompletableFuture;
-
/**
* {@link ExternalResource} which provides a {@link SlotPool}.
*/
@@ -102,8 +100,6 @@ public class SlotPoolResource extends ExternalResource {
}
private void terminateSlotPool() {
- slotPool.shutDown();
- CompletableFuture<Void> terminationFuture = slotPool.getTerminationFuture();
- terminationFuture.join();
+ slotPool.closeAsync().join();
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
index 3a9925c..6e90eb1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
@@ -268,7 +268,7 @@ public class SlotPoolTest extends TestLogger {
assertEquals(slot1.getTaskManagerLocation(), slot2.getTaskManagerLocation());
assertEquals(slot1.getPhysicalSlotNumber(), slot2.getPhysicalSlotNumber());
} finally {
- slotPool.shutDown();
+ RpcUtils.terminateRpcEndpoint(slotPool, timeout);
}
}
@@ -336,7 +336,7 @@ public class SlotPoolTest extends TestLogger {
assertFalse(slotPoolGateway.offerSlot(taskManagerLocation, taskManagerGateway, anotherSlotOfferWithSameAllocationId).get());
assertFalse(slotPoolGateway.offerSlot(anotherTaskManagerLocation, taskManagerGateway, slotOffer).get());
} finally {
- slotPool.shutDown();
+ RpcUtils.terminateRpcEndpoint(slotPool, timeout);
}
}
@@ -393,7 +393,7 @@ public class SlotPoolTest extends TestLogger {
Thread.sleep(10);
assertFalse(future2.isDone());
} finally {
- slotPool.shutDown();
+ RpcUtils.terminateRpcEndpoint(slotPool, timeout);
}
}
@@ -577,8 +577,7 @@ public class SlotPoolTest extends TestLogger {
assertThat(acceptedSlotOffers, Matchers.equalTo(slotOffers));
// shut down the slot pool
- slotPool.shutDown();
- slotPool.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ slotPool.closeAsync().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
// the shut down operation should have freed all registered slots
ArrayList<AllocationID> freedSlots = new ArrayList<>(numSlotOffers);
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 f2810a8..aa9571c 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
@@ -22,10 +22,10 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.clusterframework.FlinkResourceManager;
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;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
-import org.apache.flink.runtime.metrics.MetricRegistryImpl;
+import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerConfiguration;
import org.apache.flink.runtime.rpc.RpcService;
@@ -40,10 +40,8 @@ import org.junit.Test;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import static org.mockito.Mockito.mock;
-
/**
- * resourceManager HA test, including grant leadership and revoke leadership
+ * ResourceManager HA test, including grant leadership and revoke leadership.
*/
public class ResourceManagerHATest extends TestLogger {
@@ -64,7 +62,7 @@ public class ResourceManagerHATest extends TestLogger {
TestingHighAvailabilityServices highAvailabilityServices = new TestingHighAvailabilityServices();
highAvailabilityServices.setResourceManagerLeaderElectionService(leaderElectionService);
- HeartbeatServices heartbeatServices = mock(HeartbeatServices.class);
+ TestingHeartbeatServices heartbeatServices = new TestingHeartbeatServices();
ResourceManagerRuntimeServicesConfiguration resourceManagerRuntimeServicesConfiguration = new ResourceManagerRuntimeServicesConfiguration(
Time.seconds(5L),
@@ -77,8 +75,6 @@ public class ResourceManagerHATest extends TestLogger {
highAvailabilityServices,
rpcService.getScheduledExecutor());
- MetricRegistryImpl metricRegistry = mock(MetricRegistryImpl.class);
-
TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
CompletableFuture<ResourceManagerId> revokedLeaderIdFuture = new CompletableFuture<>();
@@ -91,7 +87,7 @@ public class ResourceManagerHATest extends TestLogger {
highAvailabilityServices,
heartbeatServices,
resourceManagerRuntimeServices.getSlotManager(),
- metricRegistry,
+ NoOpMetricRegistry.INSTANCE,
resourceManagerRuntimeServices.getJobLeaderIdService(),
new ClusterInformation("localhost", 1234),
testingFatalErrorHandler,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
index 3927b0c..5ebdce2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
@@ -135,7 +135,7 @@ public class AsyncCallsTest extends TestLogger {
// validate that no concurrent access happened
assertFalse("Rpc Endpoint had concurrent access", concurrentAccess.get());
} finally {
- rpcEndpoint.shutDown();
+ RpcUtils.terminateRpcEndpoint(rpcEndpoint, timeout);
}
}
@@ -320,8 +320,7 @@ public class AsyncCallsTest extends TestLogger {
return result;
} finally {
- fencedTestEndpoint.shutDown();
- fencedTestEndpoint.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(fencedTestEndpoint, timeout);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java
index 344a05a..ac350de 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java
@@ -101,8 +101,7 @@ public class FencedRpcEndpointTest extends TestLogger {
assertEquals(newFencingToken, fencedGateway.getFencingToken());
assertEquals(newFencingToken, fencedTestingEndpoint.getFencingToken());
} finally {
- fencedTestingEndpoint.shutDown();
- fencedTestingEndpoint.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(fencedTestingEndpoint, timeout);
}
}
@@ -150,8 +149,7 @@ public class FencedRpcEndpointTest extends TestLogger {
}
} finally {
- fencedTestingEndpoint.shutDown();
- fencedTestingEndpoint.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(fencedTestingEndpoint, timeout);
}
}
@@ -197,8 +195,7 @@ public class FencedRpcEndpointTest extends TestLogger {
assertTrue(ExceptionUtils.stripExecutionException(e) instanceof FencingTokenException);
}
} finally {
- fencedTestingEndpoint.shutDown();
- fencedTestingEndpoint.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(fencedTestingEndpoint, timeout);
}
}
@@ -241,8 +238,7 @@ public class FencedRpcEndpointTest extends TestLogger {
}
} finally {
- fencedTestingEndpoint.shutDown();
- fencedTestingEndpoint.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(fencedTestingEndpoint, timeout);
}
}
@@ -277,8 +273,7 @@ public class FencedRpcEndpointTest extends TestLogger {
// we should not be able to call getFencingToken on an unfenced gateway
}
} finally {
- fencedTestingEndpoint.shutDown();
- fencedTestingEndpoint.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(fencedTestingEndpoint, timeout);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java
index ba0f9ca..31b5407 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java
@@ -82,7 +82,7 @@ public class RpcEndpointTest extends TestLogger {
assertEquals(Integer.valueOf(expectedValue), foobar.get());
} finally {
- baseEndpoint.shutDown();
+ RpcUtils.terminateRpcEndpoint(baseEndpoint, TIMEOUT);
}
}
@@ -102,7 +102,7 @@ public class RpcEndpointTest extends TestLogger {
fail("Expected to fail with a RuntimeException since we requested the wrong gateway type.");
} finally {
- baseEndpoint.shutDown();
+ RpcUtils.terminateRpcEndpoint(baseEndpoint, TIMEOUT);
}
}
@@ -131,7 +131,7 @@ public class RpcEndpointTest extends TestLogger {
assertEquals(Integer.valueOf(barfoo), extendedGateway.barfoo().get());
assertEquals(foo, differentGateway.foo().get());
} finally {
- endpoint.shutDown();
+ RpcUtils.terminateRpcEndpoint(endpoint, TIMEOUT);
}
}
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 1d4a1c6..47fc02e 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
@@ -135,15 +135,17 @@ public class AkkaRpcActorTest extends TestLogger {
// start the endpoint so that it can process messages
rpcEndpoint.start();
- // send the rpc again
- result = rpcGateway.foobar();
-
- // now we should receive a result :-)
- Integer actualValue = result.get(timeout.getSize(), timeout.getUnit());
+ try {
+ // send the rpc again
+ result = rpcGateway.foobar();
- assertThat("The new foobar value should have been returned.", actualValue, Is.is(expectedValue));
+ // now we should receive a result :-)
+ Integer actualValue = result.get(timeout.getSize(), timeout.getUnit());
- rpcEndpoint.shutDown();
+ assertThat("The new foobar value should have been returned.", actualValue, Is.is(expectedValue));
+ } finally {
+ RpcUtils.terminateRpcEndpoint(rpcEndpoint, timeout);
+ }
}
/**
@@ -162,7 +164,7 @@ public class AkkaRpcActorTest extends TestLogger {
assertFalse(terminationFuture.isDone());
CompletableFuture.runAsync(
- rpcEndpoint::shutDown,
+ rpcEndpoint::closeAsync,
akkaRpcService.getExecutor());
// wait until the rpc endpoint has terminated
@@ -216,9 +218,7 @@ public class AkkaRpcActorTest extends TestLogger {
FailingOnStopEndpoint rpcEndpoint = new FailingOnStopEndpoint(akkaRpcService, "FailingOnStopEndpoint");
rpcEndpoint.start();
- rpcEndpoint.shutDown();
-
- CompletableFuture<Void> terminationFuture = rpcEndpoint.getTerminationFuture();
+ CompletableFuture<Void> terminationFuture = rpcEndpoint.closeAsync();
try {
terminationFuture.get();
@@ -235,9 +235,7 @@ public class AkkaRpcActorTest extends TestLogger {
SimpleRpcEndpoint simpleRpcEndpoint = new SimpleRpcEndpoint(akkaRpcService, "SimpleRpcEndpoint");
simpleRpcEndpoint.start();
- simpleRpcEndpoint.shutDown();
-
- CompletableFuture<Void> terminationFuture = simpleRpcEndpoint.getTerminationFuture();
+ CompletableFuture<Void> terminationFuture = simpleRpcEndpoint.closeAsync();
// check that we executed the onStop method in the main thread, otherwise an exception
// would be thrown here.
@@ -281,9 +279,7 @@ public class AkkaRpcActorTest extends TestLogger {
try {
endpoint.start();
- final CompletableFuture<Void> terminationFuture = endpoint.getTerminationFuture();
-
- endpoint.shutDown();
+ final CompletableFuture<Void> terminationFuture = endpoint.closeAsync();
assertFalse(terminationFuture.isDone());
@@ -307,7 +303,7 @@ public class AkkaRpcActorTest extends TestLogger {
try {
endpoint.start();
- CompletableFuture<Void> terminationFuture = endpoint.terminate();
+ CompletableFuture<Void> terminationFuture = endpoint.closeAsync();
terminationFuture.get();
} finally {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MainThreadValidationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MainThreadValidationTest.java
index 1c9f5ca..570c181 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MainThreadValidationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MainThreadValidationTest.java
@@ -26,8 +26,6 @@ import org.apache.flink.util.TestLogger;
import org.junit.Test;
-import java.util.concurrent.CompletableFuture;
-
import static org.junit.Assert.assertTrue;
/**
@@ -68,7 +66,7 @@ public class MainThreadValidationTest extends TestLogger {
}
assertTrue("should fail with an assertion error", exceptionThrown);
- testEndpoint.shutDown();
+ testEndpoint.closeAsync();
}
finally {
akkaRpcService.stopService().get();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
index 8ddc9be..1dca276 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
@@ -645,8 +645,7 @@ public class TaskExecutorTest extends TestLogger {
eq(taskManagerAddress), eq(taskManagerLocation.getResourceID()), anyInt(), any(HardwareDescription.class), any(Time.class));
}
finally {
- taskManager.shutDown();
- taskManager.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(taskManager, timeout);
}
}
@@ -726,8 +725,7 @@ public class TaskExecutorTest extends TestLogger {
assertNotNull(taskManager.getResourceManagerConnection());
}
finally {
- taskManager.shutDown();
- taskManager.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(taskManager, timeout);
}
}
@@ -835,8 +833,7 @@ public class TaskExecutorTest extends TestLogger {
completionFuture.get();
} finally {
- taskManager.shutDown();
- taskManager.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(taskManager, timeout);
}
}
@@ -1061,8 +1058,7 @@ public class TaskExecutorTest extends TestLogger {
assertFalse(taskSlotTable.tryMarkSlotActive(jobId, allocationId2));
assertTrue(taskSlotTable.isSlotFree(1));
} finally {
- taskManager.shutDown();
- taskManager.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(taskManager, timeout);
}
}
@@ -1198,8 +1194,7 @@ public class TaskExecutorTest extends TestLogger {
// wait for the task completion
taskInTerminalState.await();
} finally {
- taskManager.shutDown();
- taskManager.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(taskManager, timeout);
}
}
@@ -1263,8 +1258,7 @@ public class TaskExecutorTest extends TestLogger {
assertEquals(jobMasterGateway, jobManagerConnection.getJobManagerGateway());
} finally {
- taskExecutor.shutDown();
- taskExecutor.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(taskExecutor, timeout);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/RpcGatewayRetrieverTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/RpcGatewayRetrieverTest.java
index d359cca..1ccf503 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/RpcGatewayRetrieverTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/RpcGatewayRetrieverTest.java
@@ -106,10 +106,8 @@ public class RpcGatewayRetrieverTest extends TestLogger {
assertEquals(dummyRpcEndpoint2.getAddress(), dummyGateway2.getAddress());
assertEquals(expectedValue2, dummyGateway2.foobar(TIMEOUT).get(TIMEOUT.toMilliseconds(), TimeUnit.MILLISECONDS));
} finally {
- dummyRpcEndpoint.shutDown();
- dummyRpcEndpoint2.shutDown();
- dummyRpcEndpoint.getTerminationFuture().get(TIMEOUT.toMilliseconds(), TimeUnit.MILLISECONDS);
- dummyRpcEndpoint2.getTerminationFuture().get(TIMEOUT.toMilliseconds(), TimeUnit.MILLISECONDS);
+ RpcUtils.terminateRpcEndpoint(dummyRpcEndpoint, TIMEOUT);
+ RpcUtils.terminateRpcEndpoint(dummyRpcEndpoint2, TIMEOUT);
}
}
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
index 1a59c2d..bd7b912 100644
--- 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
@@ -248,11 +248,11 @@ public class ProcessFailureCancelingITCase extends TestLogger {
dispatcherResourceManagerComponent.deregisterApplicationAndClose(ApplicationStatus.SUCCEEDED, null);
}
- haServices.closeAndCleanupAllData();
-
fatalErrorHandler.rethrowError();
RpcUtils.terminateRpcService(rpcService, Time.seconds(100L));
+
+ haServices.closeAndCleanupAllData();
}
}
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 1f5a2f9..375c196 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -449,7 +449,7 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
@Override
public void onShutdownRequest() {
- shutDown();
+ closeAsync();
}
@Override
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 368e95c..281bf5e 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
@@ -27,7 +27,6 @@ import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.clusterframework.types.SlotID;
-import org.apache.flink.runtime.concurrent.ScheduledExecutor;
import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
@@ -291,7 +290,6 @@ public class YarnResourceManagerTest extends TestLogger {
*/
class MockResourceManagerRuntimeServices {
- private final ScheduledExecutor scheduledExecutor;
private final TestingHighAvailabilityServices highAvailabilityServices;
private final HeartbeatServices heartbeatServices;
private final MetricRegistry metricRegistry;
@@ -302,11 +300,10 @@ public class YarnResourceManagerTest extends TestLogger {
private UUID rmLeaderSessionId;
MockResourceManagerRuntimeServices() throws Exception {
- scheduledExecutor = mock(ScheduledExecutor.class);
highAvailabilityServices = new TestingHighAvailabilityServices();
rmLeaderElectionService = new TestingLeaderElectionService();
highAvailabilityServices.setResourceManagerLeaderElectionService(rmLeaderElectionService);
- heartbeatServices = new TestingHeartbeatServices(5L, 5L, scheduledExecutor);
+ heartbeatServices = new TestingHeartbeatServices();
metricRegistry = NoOpMetricRegistry.INSTANCE;
slotManager = new SlotManager(
new ScheduledExecutorServiceAdapter(new DirectScheduledExecutorService()),