You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2022/04/14 07:03:10 UTC
[flink] branch master updated: [FLINK-27050][runtime] Removes default RpcSystem instance
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ccf4d24ee71 [FLINK-27050][runtime] Removes default RpcSystem instance
ccf4d24ee71 is described below
commit ccf4d24ee71264eb32d1bf2fcc61c2a8aef7b617
Author: Matthias Pohl <ma...@ververica.com>
AuthorDate: Mon Apr 4 14:59:16 2022 +0200
[FLINK-27050][runtime] Removes default RpcSystem instance
Having a default RpcSystem in TestingDispatcher.Builder
caused threads being spawned without cleanup. Instead, we
should rely on the RpcSystem instance provided by the test.
---
.../runtime/dispatcher/AbstractDispatcherTest.java | 1 -
.../runtime/dispatcher/DispatcherCleanupITCase.java | 8 ++++----
.../dispatcher/DispatcherResourceCleanupTest.java | 6 ++++--
.../flink/runtime/dispatcher/DispatcherTest.java | 20 +++++++++++---------
.../flink/runtime/dispatcher/TestingDispatcher.java | 9 +--------
5 files changed, 20 insertions(+), 24 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/AbstractDispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/AbstractDispatcherTest.java
index 31f034e94ed..77cb28f169f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/AbstractDispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/AbstractDispatcherTest.java
@@ -103,7 +103,6 @@ public class AbstractDispatcherTest extends TestLogger {
protected TestingDispatcher.Builder createTestingDispatcherBuilder() {
return TestingDispatcher.builder()
- .setRpcService(rpcService)
.setConfiguration(configuration)
.setHeartbeatServices(heartbeatServices)
.setHighAvailabilityServices(haServices)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherCleanupITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherCleanupITCase.java
index d305011a7e9..2383eae943c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherCleanupITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherCleanupITCase.java
@@ -177,7 +177,7 @@ public class DispatcherCleanupITCase extends AbstractDispatcherTest {
haServices,
UnregisteredMetricGroups
.createUnregisteredJobManagerMetricGroup()))
- .build();
+ .build(rpcService);
dispatcher.start();
toTerminate.add(dispatcher);
@@ -223,7 +223,7 @@ public class DispatcherCleanupITCase extends AbstractDispatcherTest {
final Dispatcher dispatcher =
createTestingDispatcherBuilder()
.setJobManagerRunnerRegistry(jobManagerRunnerRegistry)
- .build();
+ .build(rpcService);
dispatcher.start();
toTerminate.add(dispatcher);
@@ -290,7 +290,7 @@ public class DispatcherCleanupITCase extends AbstractDispatcherTest {
configuration.set(
CleanupOptions.CLEANUP_STRATEGY,
CleanupOptions.NONE_PARAM_VALUES.iterator().next());
- final Dispatcher dispatcher = createTestingDispatcherBuilder().build();
+ final Dispatcher dispatcher = createTestingDispatcherBuilder().build(rpcService);
dispatcher.start();
toTerminate.add(dispatcher);
@@ -332,7 +332,7 @@ public class DispatcherCleanupITCase extends AbstractDispatcherTest {
final Dispatcher secondDispatcher =
createTestingDispatcherBuilder()
.setRecoveredDirtyJobs(haServices.getJobResultStore().getDirtyResults())
- .build();
+ .build(rpcService);
secondDispatcher.start();
toTerminate.add(secondDispatcher);
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 f007a82b5e1..f8611d394da 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
@@ -165,7 +165,10 @@ public class DispatcherResourceCleanupTest extends TestLogger {
TestingDispatcher.Builder dispatcherBuilder,
JobManagerRunnerFactory jobManagerRunnerFactory)
throws Exception {
- dispatcher = dispatcherBuilder.setJobManagerRunnerFactory(jobManagerRunnerFactory).build();
+ dispatcher =
+ dispatcherBuilder
+ .setJobManagerRunnerFactory(jobManagerRunnerFactory)
+ .build(rpcService);
dispatcher.start();
@@ -176,7 +179,6 @@ public class DispatcherResourceCleanupTest extends TestLogger {
final JobManagerRunnerRegistry jobManagerRunnerRegistry =
new DefaultJobManagerRunnerRegistry(2);
return TestingDispatcher.builder()
- .setRpcService(rpcService)
.setBlobServer(blobServer)
.setJobManagerRunnerRegistry(jobManagerRunnerRegistry)
.setFatalErrorHandler(testingFatalErrorHandlerResource.getFatalErrorHandler())
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 e68c52750de..e806583c8cc 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
@@ -169,7 +169,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
.setJobManagerRunnerFactory(jobManagerRunnerFactory)
.setJobGraphWriter(haServices.getJobGraphStore())
.setJobResultStore(haServices.getJobResultStore())
- .build();
+ .build(rpcService);
dispatcher.start();
return dispatcher;
}
@@ -251,7 +251,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
new ExpectedJobIdJobManagerRunnerFactory(
jobId, createdJobManagerRunnerLatch))
.setRecoveredJobs(Collections.singleton(jobGraph))
- .build();
+ .build(rpcService);
dispatcher.start();
final DispatcherGateway dispatcherGateway =
dispatcher.getSelfGateway(DispatcherGateway.class);
@@ -483,7 +483,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
archiveAttemptFuture.complete(null);
return CompletableFuture.completedFuture(null);
})
- .build();
+ .build(rpcService);
dispatcher.start();
jobMasterLeaderElectionService.isLeader(UUID.randomUUID());
DispatcherGateway dispatcherGateway = dispatcher.getSelfGateway(DispatcherGateway.class);
@@ -678,7 +678,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
createTestingDispatcherBuilder()
.setJobManagerRunnerFactory(jobManagerRunnerFactory)
.setRecoveredJobs(Collections.singleton(JobGraphTestUtils.emptyJobGraph()))
- .build();
+ .build(rpcService);
dispatcher.start();
@@ -723,7 +723,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
createTestingDispatcherBuilder()
.setRecoveredJobs(Collections.singleton(jobGraph))
.setRecoveredDirtyJobs(Collections.singleton(jobResult))
- .build();
+ .build(rpcService);
}
@Test
@@ -752,7 +752,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
dispatcherBootstrapLatch.trigger();
return new NoOpDispatcherBootstrap();
})
- .build();
+ .build(rpcService);
dispatcher.start();
@@ -779,7 +779,9 @@ public class DispatcherTest extends AbstractDispatcherTest {
haServices.setJobGraphStore(submittedJobGraphStore);
dispatcher =
- createTestingDispatcherBuilder().setJobGraphWriter(submittedJobGraphStore).build();
+ createTestingDispatcherBuilder()
+ .setJobGraphWriter(submittedJobGraphStore)
+ .build(rpcService);
dispatcher.start();
@@ -899,7 +901,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
createTestingDispatcherBuilder()
.setRecoveredJobs(Collections.singleton(jobGraph))
.setJobGraphWriter(testingJobGraphStore)
- .build();
+ .build(rpcService);
dispatcher.start();
final CompletableFuture<Void> processFuture =
@@ -1068,7 +1070,7 @@ public class DispatcherTest extends AbstractDispatcherTest {
dispatcher =
createTestingDispatcherBuilder()
.setRecoveredJobs(Collections.singleton(new JobGraph(jobId1, "foobar")))
- .build();
+ .build(rpcService);
Assertions.assertThat(blobServer.getFile(jobId1, blobKey1)).hasBinaryContent(fileContent);
Assertions.assertThatThrownBy(() -> blobServer.getFile(jobId2, blobKey2))
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 444c77e8a55..9c18622d136 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
@@ -41,7 +41,6 @@ import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.rpc.TestingRpcService;
import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
@@ -177,7 +176,6 @@ class TestingDispatcher extends Dispatcher {
}
public static class Builder {
- private RpcService rpcService = new TestingRpcService();
private DispatcherId fencingToken = DispatcherId.generate();
private Collection<JobGraph> recoveredJobs = Collections.emptyList();
@Nullable private Collection<JobResult> recoveredDirtyJobs = null;
@@ -217,11 +215,6 @@ class TestingDispatcher extends Dispatcher {
new DefaultJobManagerRunnerRegistry(1);
@Nullable private ResourceCleanerFactory resourceCleanerFactory;
- public Builder setRpcService(RpcService rpcService) {
- this.rpcService = rpcService;
- return this;
- }
-
public Builder setFencingToken(DispatcherId fencingToken) {
this.fencingToken = fencingToken;
return this;
@@ -354,7 +347,7 @@ class TestingDispatcher extends Dispatcher {
jobManagerMetricGroup);
}
- public TestingDispatcher build() throws Exception {
+ public TestingDispatcher build(RpcService rpcService) throws Exception {
return new TestingDispatcher(
rpcService,
fencingToken,