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/02 08:59:06 UTC
[flink] 02/03: [FLINK-16836] Clear rpcConnection field in
JobManagerLeaderListener when target loses leadership
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit dede2ae753806214aac0c3552a7f2909ea881a51
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Fri Mar 27 18:08:48 2020 +0100
[FLINK-16836] Clear rpcConnection field in JobManagerLeaderListener when target loses leadership
Clearing the rpcConnection field in the JobManagerLeaderListener when target loses leadership
prevents that we try to reconnect to the target in case JobLeaderService.reconnect(JobID) is
called.
This closes #11552.
---
.../runtime/taskexecutor/JobLeaderService.java | 2 +
.../runtime/taskexecutor/JobLeaderServiceTest.java | 72 ++++++++++++++++++++--
2 files changed, 68 insertions(+), 6 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
index f89585e..376a8c3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobLeaderService.java
@@ -241,6 +241,7 @@ public class JobLeaderService {
/** Rpc connection to the job leader. */
@GuardedBy("lock")
+ @Nullable
private RegisteredRpcConnection<JobMasterId, JobMasterGateway, JMTMRegistrationSuccess> rpcConnection;
/** Leader id of the current job leader. */
@@ -311,6 +312,7 @@ public class JobLeaderService {
// the leader lost leadership but there is no other leader yet.
if (rpcConnection != null) {
rpcConnection.close();
+ rpcConnection = null;
}
jobManagerLostLeadership = Optional.ofNullable(currentJobMasterId);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/JobLeaderServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/JobLeaderServiceTest.java
index 51595e8..7e32842 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/JobLeaderServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/JobLeaderServiceTest.java
@@ -21,30 +21,34 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.JobID;
import org.apache.flink.core.testutils.CheckedThread;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
+import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServicesBuilder;
import org.apache.flink.runtime.jobmaster.JMTMRegistrationSuccess;
import org.apache.flink.runtime.jobmaster.JobMasterGateway;
import org.apache.flink.runtime.jobmaster.JobMasterId;
+import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGateway;
+import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGatewayBuilder;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
import org.apache.flink.runtime.registration.RetryingRegistrationConfiguration;
import org.apache.flink.runtime.rpc.TestingRpcServiceResource;
import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.util.TestLogger;
-import org.junit.ClassRule;
+import org.junit.Rule;
import org.junit.Test;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
/**
* Tests for the {@link JobLeaderService}.
*/
public class JobLeaderServiceTest extends TestLogger {
- @ClassRule
- public static final TestingRpcServiceResource RPC_SERVICE_RESOURCE = new TestingRpcServiceResource();
+ @Rule
+ public final TestingRpcServiceResource rpcServiceResource = new TestingRpcServiceResource();
/**
* Tests that we can concurrently modify the JobLeaderService and complete the leader retrieval operation.
@@ -71,7 +75,7 @@ public class JobLeaderServiceTest extends TestLogger {
jobLeaderService.start(
"foobar",
- RPC_SERVICE_RESOURCE.getTestingRpcService(),
+ rpcServiceResource.getTestingRpcService(),
haServices,
jobLeaderListener);
@@ -96,21 +100,77 @@ public class JobLeaderServiceTest extends TestLogger {
addJobAction.sync();
}
+ /**
+ * Tests that the JobLeaderService won't try to reconnect to JobMaster after it
+ * has lost the leadership. See FLINK-16836.
+ */
+ @Test
+ public void doesNotReconnectAfterTargetLostLeadership() throws Exception {
+ final JobLeaderService jobLeaderService = new JobLeaderService(
+ new LocalTaskManagerLocation(),
+ RetryingRegistrationConfiguration.defaultConfiguration());
+
+ final JobID jobId = new JobID();
+
+ final SettableLeaderRetrievalService leaderRetrievalService = new SettableLeaderRetrievalService();
+ final TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServicesBuilder()
+ .setJobMasterLeaderRetrieverFunction(ignored -> leaderRetrievalService)
+ .build();
+
+ final String jmAddress = "foobar";
+ final TestingJobMasterGateway jobMasterGateway = new TestingJobMasterGatewayBuilder().build();
+ rpcServiceResource.getTestingRpcService().registerGateway(jmAddress, jobMasterGateway);
+
+ final TestingJobLeaderListener testingJobLeaderListener = new TestingJobLeaderListener();
+ jobLeaderService.start(
+ "foobar",
+ rpcServiceResource.getTestingRpcService(),
+ haServices,
+ testingJobLeaderListener);
+
+ try {
+ jobLeaderService.addJob(jobId, jmAddress);
+
+ leaderRetrievalService.notifyListener(jmAddress, UUID.randomUUID());
+
+ testingJobLeaderListener.waitUntilJobManagerGainedLeadership();
+
+ // revoke the leadership
+ leaderRetrievalService.notifyListener(null, null);
+ testingJobLeaderListener.waitUntilJobManagerLostLeadership();
+
+ jobLeaderService.reconnect(jobId);
+ } finally {
+ jobLeaderService.stop();
+ }
+ }
+
private static final class TestingJobLeaderListener implements JobLeaderListener {
+ private final CountDownLatch jobManagerGainedLeadership = new CountDownLatch(1);
+ private final CountDownLatch jobManagerLostLeadership = new CountDownLatch(1);
+
@Override
public void jobManagerGainedLeadership(JobID jobId, JobMasterGateway jobManagerGateway, JMTMRegistrationSuccess registrationMessage) {
- // ignored
+ jobManagerGainedLeadership.countDown();
}
@Override
public void jobManagerLostLeadership(JobID jobId, JobMasterId jobMasterId) {
- // ignored
+ jobManagerLostLeadership.countDown();
}
@Override
public void handleError(Throwable throwable) {
// ignored
}
+
+ private void waitUntilJobManagerGainedLeadership() throws InterruptedException {
+ jobManagerGainedLeadership.await();
+ }
+
+ private void waitUntilJobManagerLostLeadership() throws InterruptedException {
+ jobManagerLostLeadership.await();
+ }
}
}