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 09:01:16 UTC
[flink] branch release-1.9 updated (78e2c0d -> 323c406)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 78e2c0d [FLINK-16860][orc] Distinguish empty filters from no pushed down in expainSource
new f8852e8 [FLINK-16373] Make JobManagerLeaderListener thread safe
new fa2d98f [FLINK-16836] Clear rpcConnection field in JobManagerLeaderListener when target loses leadership
new 323c406 [FLINK-14316] Properly manage rpcConnection in JobManagerLeaderListener under leader change
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../runtime/taskexecutor/JobLeaderService.java | 153 +++++++------
.../runtime/taskexecutor/JobLeaderServiceTest.java | 240 +++++++++++++++++++++
2 files changed, 324 insertions(+), 69 deletions(-)
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/JobLeaderServiceTest.java
[flink] 03/03: [FLINK-14316] Properly manage rpcConnection in
JobManagerLeaderListener under leader change
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 323c406ef5c1cfccb17fc65053d2d13dc648366d
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Wed Apr 1 16:30:42 2020 +0200
[FLINK-14316] Properly manage rpcConnection in JobManagerLeaderListener under leader change
This commit changes how the rpcConnection is managed in JobManagerLeaderListener under leader change.
This component clears now the fields rpcConnection and currentJobMasterId if the leader loses leadership.
Moreover, it only restarts a connection attempt if the leader session id is new.
This closes #11605.
---
.../runtime/taskexecutor/JobLeaderService.java | 63 ++++++------
.../runtime/taskexecutor/JobLeaderServiceTest.java | 108 ++++++++++++++++-----
2 files changed, 119 insertions(+), 52 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 376a8c3..e6455bc 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
@@ -271,9 +271,7 @@ public class JobLeaderService {
if (!stopped) {
stopped = true;
- if (rpcConnection != null) {
- rpcConnection.close();
- }
+ closeRpcConnection();
}
}
}
@@ -310,37 +308,16 @@ public class JobLeaderService {
if (leaderAddress == null || leaderAddress.isEmpty()) {
// the leader lost leadership but there is no other leader yet.
- if (rpcConnection != null) {
- rpcConnection.close();
- rpcConnection = null;
- }
-
jobManagerLostLeadership = Optional.ofNullable(currentJobMasterId);
- currentJobMasterId = jobMasterId;
+ closeRpcConnection();
} else {
- currentJobMasterId = jobMasterId;
-
- if (rpcConnection != null) {
- // check if we are already trying to connect to this leader
- if (!Objects.equals(jobMasterId, rpcConnection.getTargetLeaderId())) {
- rpcConnection.close();
-
- rpcConnection = new JobManagerRegisteredRpcConnection(
- LOG,
- leaderAddress,
- jobMasterId,
- rpcService.getExecutor());
- }
+ // check whether we are already connecting to this leader
+ if (Objects.equals(jobMasterId, currentJobMasterId)) {
+ LOG.debug("Ongoing attempt to connect to leader of job {}. Ignoring duplicate leader information.", jobId);
} else {
- rpcConnection = new JobManagerRegisteredRpcConnection(
- LOG,
- leaderAddress,
- jobMasterId,
- rpcService.getExecutor());
+ closeRpcConnection();
+ openRpcConnectionTo(leaderAddress, jobMasterId);
}
-
- LOG.info("Try to register at job manager {} with leader id {}.", leaderAddress, leaderId);
- rpcConnection.start();
}
}
}
@@ -349,6 +326,32 @@ public class JobLeaderService {
jobManagerLostLeadership.ifPresent(oldJobMasterId -> jobLeaderListener.jobManagerLostLeadership(jobId, oldJobMasterId));
}
+ @GuardedBy("lock")
+ private void openRpcConnectionTo(String leaderAddress, JobMasterId jobMasterId) {
+ Preconditions.checkState(
+ currentJobMasterId == null && rpcConnection == null,
+ "Cannot open a new rpc connection if the previous connection has not been closed.");
+
+ currentJobMasterId = jobMasterId;
+ rpcConnection = new JobManagerRegisteredRpcConnection(
+ LOG,
+ leaderAddress,
+ jobMasterId,
+ rpcService.getExecutor());
+
+ LOG.info("Try to register at job manager {} with leader id {}.", leaderAddress, jobMasterId.toUUID());
+ rpcConnection.start();
+ }
+
+ @GuardedBy("lock")
+ private void closeRpcConnection() {
+ if (rpcConnection != null) {
+ rpcConnection.close();
+ rpcConnection = null;
+ currentJobMasterId = null;
+ }
+ }
+
@Override
public void handleError(Exception exception) {
if (stopped) {
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 7e32842..fbcf5ad 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
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.JobID;
import org.apache.flink.core.testutils.CheckedThread;
+import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServicesBuilder;
@@ -41,6 +42,10 @@ import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
+import java.util.function.Consumer;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
/**
* Tests for the {@link JobLeaderService}.
@@ -106,34 +111,25 @@ public class JobLeaderServiceTest extends TestLogger {
*/
@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 TestingJobMasterGateway jobMasterGateway = registerJobMaster();
- final String jmAddress = "foobar";
- final TestingJobMasterGateway jobMasterGateway = new TestingJobMasterGatewayBuilder().build();
- rpcServiceResource.getTestingRpcService().registerGateway(jmAddress, jobMasterGateway);
+ final OneShotLatch jobManagerGainedLeadership = new OneShotLatch();
+ final TestingJobLeaderListener testingJobLeaderListener = new TestingJobLeaderListener(ignored -> jobManagerGainedLeadership.trigger());
- final TestingJobLeaderListener testingJobLeaderListener = new TestingJobLeaderListener();
- jobLeaderService.start(
- "foobar",
- rpcServiceResource.getTestingRpcService(),
- haServices,
- testingJobLeaderListener);
+ final JobLeaderService jobLeaderService = createAndStartJobLeaderService(haServices, testingJobLeaderListener);
try {
- jobLeaderService.addJob(jobId, jmAddress);
+ jobLeaderService.addJob(jobId, jobMasterGateway.getAddress());
- leaderRetrievalService.notifyListener(jmAddress, UUID.randomUUID());
+ leaderRetrievalService.notifyListener(jobMasterGateway.getAddress(), UUID.randomUUID());
- testingJobLeaderListener.waitUntilJobManagerGainedLeadership();
+ jobManagerGainedLeadership.await();
// revoke the leadership
leaderRetrievalService.notifyListener(null, null);
@@ -145,14 +141,86 @@ public class JobLeaderServiceTest extends TestLogger {
}
}
+ /**
+ * Tests that the JobLeaderService can reconnect to an old leader which seemed
+ * to have lost the leadership in between. See FLINK-14316.
+ */
+ @Test
+ public void canReconnectToOldLeaderWithSameLeaderAddress() throws Exception {
+ final JobID jobId = new JobID();
+
+ final SettableLeaderRetrievalService leaderRetrievalService = new SettableLeaderRetrievalService();
+ final TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServicesBuilder()
+ .setJobMasterLeaderRetrieverFunction(ignored -> leaderRetrievalService)
+ .build();
+
+ final TestingJobMasterGateway jobMasterGateway = registerJobMaster();
+
+ final BlockingQueue<JobID> leadershipQueue = new ArrayBlockingQueue<>(1);
+ final TestingJobLeaderListener testingJobLeaderListener = new TestingJobLeaderListener(leadershipQueue::offer);
+
+ final JobLeaderService jobLeaderService = createAndStartJobLeaderService(haServices, testingJobLeaderListener);
+
+ try {
+ jobLeaderService.addJob(jobId, jobMasterGateway.getAddress());
+
+ final UUID leaderSessionId = UUID.randomUUID();
+ leaderRetrievalService.notifyListener(jobMasterGateway.getAddress(), leaderSessionId);
+
+ // wait for the first leadership
+ assertThat(leadershipQueue.take(), is(jobId));
+
+ // revoke the leadership
+ leaderRetrievalService.notifyListener(null, null);
+
+ testingJobLeaderListener.waitUntilJobManagerLostLeadership();
+
+ leaderRetrievalService.notifyListener(jobMasterGateway.getAddress(), leaderSessionId);
+
+ // check that we obtain the leadership a second time
+ assertThat(leadershipQueue.take(), is(jobId));
+ } finally {
+ jobLeaderService.stop();
+ }
+ }
+
+ private JobLeaderService createAndStartJobLeaderService(HighAvailabilityServices haServices, JobLeaderListener testingJobLeaderListener) {
+ final JobLeaderService jobLeaderService = new JobLeaderService(
+ new LocalTaskManagerLocation(),
+ RetryingRegistrationConfiguration.defaultConfiguration());
+
+ jobLeaderService.start(
+ "foobar",
+ rpcServiceResource.getTestingRpcService(),
+ haServices,
+ testingJobLeaderListener);
+ return jobLeaderService;
+ }
+
+ private TestingJobMasterGateway registerJobMaster() {
+ final TestingJobMasterGateway jobMasterGateway = new TestingJobMasterGatewayBuilder().build();
+ rpcServiceResource.getTestingRpcService().registerGateway(jobMasterGateway.getAddress(), jobMasterGateway);
+
+ return jobMasterGateway;
+ }
+
private static final class TestingJobLeaderListener implements JobLeaderListener {
- private final CountDownLatch jobManagerGainedLeadership = new CountDownLatch(1);
private final CountDownLatch jobManagerLostLeadership = new CountDownLatch(1);
+ private final Consumer<JobID> jobManagerGainedLeadership;
+
+ private TestingJobLeaderListener() {
+ this(ignored -> {});
+ }
+
+ private TestingJobLeaderListener(Consumer<JobID> jobManagerGainedLeadership) {
+ this.jobManagerGainedLeadership = jobManagerGainedLeadership;
+ }
+
@Override
public void jobManagerGainedLeadership(JobID jobId, JobMasterGateway jobManagerGateway, JMTMRegistrationSuccess registrationMessage) {
- jobManagerGainedLeadership.countDown();
+ jobManagerGainedLeadership.accept(jobId);
}
@Override
@@ -165,10 +233,6 @@ public class JobLeaderServiceTest extends TestLogger {
// ignored
}
- private void waitUntilJobManagerGainedLeadership() throws InterruptedException {
- jobManagerGainedLeadership.await();
- }
-
private void waitUntilJobManagerLostLeadership() throws InterruptedException {
jobManagerLostLeadership.await();
}
[flink] 02/03: [FLINK-16836] Clear rpcConnection field in
JobManagerLeaderListener when target loses leadership
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fa2d98f2d7a8d31d26a82e0017204001bef56d61
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();
+ }
}
}
[flink] 01/03: [FLINK-16373] Make JobManagerLeaderListener thread
safe
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f8852e8f764ee91b0b713bb9c90f2f170096e553
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Wed Mar 4 18:01:18 2020 +0100
[FLINK-16373] Make JobManagerLeaderListener thread safe
The JobManagerLeaderListener used by the JobLeaderService was not thread safe. Stopping the listener
while notifying a new leader could lead to an IllegalStateException where the rpcConnection which
was supposed to be started was concurrently closed by the stop call.
This closes #11313.
---
.../runtime/taskexecutor/JobLeaderService.java | 130 +++++++++++----------
.../runtime/taskexecutor/JobLeaderServiceTest.java | 116 ++++++++++++++++++
2 files changed, 186 insertions(+), 60 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 eed6f11..f89585e 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
@@ -40,9 +40,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+import javax.annotation.concurrent.ThreadSafe;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -228,20 +231,26 @@ public class JobLeaderService {
/**
* Leader listener which tries to establish a connection to a newly detected job leader.
*/
+ @ThreadSafe
private final class JobManagerLeaderListener implements LeaderRetrievalListener {
+ private final Object lock = new Object();
+
/** Job id identifying the job to look for a leader. */
private final JobID jobId;
/** Rpc connection to the job leader. */
- private volatile RegisteredRpcConnection<JobMasterId, JobMasterGateway, JMTMRegistrationSuccess> rpcConnection;
+ @GuardedBy("lock")
+ private RegisteredRpcConnection<JobMasterId, JobMasterGateway, JMTMRegistrationSuccess> rpcConnection;
+
+ /** Leader id of the current job leader. */
+ @GuardedBy("lock")
+ @Nullable
+ private JobMasterId currentJobMasterId;
/** State of the listener. */
private volatile boolean stopped;
- /** Leader id of the current job leader. */
- private volatile JobMasterId currentJobMasterId;
-
private JobManagerLeaderListener(JobID jobId) {
this.jobId = Preconditions.checkNotNull(jobId);
@@ -250,91 +259,92 @@ public class JobLeaderService {
currentJobMasterId = null;
}
+ private JobMasterId getCurrentJobMasterId() {
+ synchronized (lock) {
+ return currentJobMasterId;
+ }
+ }
+
public void stop() {
- stopped = true;
+ synchronized (lock) {
+ if (!stopped) {
+ stopped = true;
- if (rpcConnection != null) {
- rpcConnection.close();
+ if (rpcConnection != null) {
+ rpcConnection.close();
+ }
+ }
}
}
public void reconnect() {
- if (stopped) {
- LOG.debug("Cannot reconnect because the JobManagerLeaderListener has already been stopped.");
- } else {
- final RegisteredRpcConnection<JobMasterId, JobMasterGateway, JMTMRegistrationSuccess> currentRpcConnection = rpcConnection;
-
- if (currentRpcConnection != null) {
- if (currentRpcConnection.isConnected()) {
-
- if (currentRpcConnection.tryReconnect()) {
- // double check for concurrent stop operation
- if (stopped) {
- currentRpcConnection.close();
- }
- } else {
- LOG.debug("Could not reconnect to the JobMaster {}.", currentRpcConnection.getTargetAddress());
- }
+ synchronized (lock) {
+ if (stopped) {
+ LOG.debug("Cannot reconnect because the JobManagerLeaderListener has already been stopped.");
+ } else {
+ if (rpcConnection != null) {
+ Preconditions.checkState(
+ rpcConnection.tryReconnect(),
+ "Illegal concurrent modification of the JobManagerLeaderListener rpc connection.");
} else {
- LOG.debug("Ongoing registration to JobMaster {}.", currentRpcConnection.getTargetAddress());
+ LOG.debug("Cannot reconnect to an unknown JobMaster.");
}
- } else {
- LOG.debug("Cannot reconnect to an unknown JobMaster.");
}
}
}
@Override
public void notifyLeaderAddress(final @Nullable String leaderAddress, final @Nullable UUID leaderId) {
- if (stopped) {
- LOG.debug("{}'s leader retrieval listener reported a new leader for job {}. " +
- "However, the service is no longer running.", JobLeaderService.class.getSimpleName(), jobId);
- } else {
- final JobMasterId jobMasterId = JobMasterId.fromUuidOrNull(leaderId);
-
- LOG.debug("New leader information for job {}. Address: {}, leader id: {}.",
- jobId, leaderAddress, jobMasterId);
-
- if (leaderAddress == null || leaderAddress.isEmpty()) {
- // the leader lost leadership but there is no other leader yet.
- if (rpcConnection != null) {
- rpcConnection.close();
- }
+ Optional<JobMasterId> jobManagerLostLeadership = Optional.empty();
- jobLeaderListener.jobManagerLostLeadership(jobId, currentJobMasterId);
-
- currentJobMasterId = jobMasterId;
+ synchronized (lock) {
+ if (stopped) {
+ LOG.debug("{}'s leader retrieval listener reported a new leader for job {}. " +
+ "However, the service is no longer running.", JobLeaderService.class.getSimpleName(), jobId);
} else {
- currentJobMasterId = jobMasterId;
+ final JobMasterId jobMasterId = JobMasterId.fromUuidOrNull(leaderId);
- if (rpcConnection != null) {
- // check if we are already trying to connect to this leader
- if (!Objects.equals(jobMasterId, rpcConnection.getTargetLeaderId())) {
+ LOG.debug("New leader information for job {}. Address: {}, leader id: {}.",
+ jobId, leaderAddress, jobMasterId);
+
+ if (leaderAddress == null || leaderAddress.isEmpty()) {
+ // the leader lost leadership but there is no other leader yet.
+ if (rpcConnection != null) {
rpcConnection.close();
+ }
+ jobManagerLostLeadership = Optional.ofNullable(currentJobMasterId);
+ currentJobMasterId = jobMasterId;
+ } else {
+ currentJobMasterId = jobMasterId;
+
+ if (rpcConnection != null) {
+ // check if we are already trying to connect to this leader
+ if (!Objects.equals(jobMasterId, rpcConnection.getTargetLeaderId())) {
+ rpcConnection.close();
+
+ rpcConnection = new JobManagerRegisteredRpcConnection(
+ LOG,
+ leaderAddress,
+ jobMasterId,
+ rpcService.getExecutor());
+ }
+ } else {
rpcConnection = new JobManagerRegisteredRpcConnection(
LOG,
leaderAddress,
jobMasterId,
rpcService.getExecutor());
}
- } else {
- rpcConnection = new JobManagerRegisteredRpcConnection(
- LOG,
- leaderAddress,
- jobMasterId,
- rpcService.getExecutor());
- }
- // double check for a concurrent stop operation
- if (stopped) {
- rpcConnection.close();
- } else {
LOG.info("Try to register at job manager {} with leader id {}.", leaderAddress, leaderId);
rpcConnection.start();
}
}
}
+
+ // send callbacks outside of the lock scope
+ jobManagerLostLeadership.ifPresent(oldJobMasterId -> jobLeaderListener.jobManagerLostLeadership(jobId, oldJobMasterId));
}
@Override
@@ -378,7 +388,7 @@ public class JobLeaderService {
@Override
protected void onRegistrationSuccess(JMTMRegistrationSuccess success) {
// filter out old registration attempts
- if (Objects.equals(getTargetLeaderId(), currentJobMasterId)) {
+ if (Objects.equals(getTargetLeaderId(), getCurrentJobMasterId())) {
log.info("Successful registration at job manager {} for job {}.", getTargetAddress(), jobId);
jobLeaderListener.jobManagerGainedLeadership(jobId, getTargetGateway(), success);
@@ -390,7 +400,7 @@ public class JobLeaderService {
@Override
protected void onRegistrationFailure(Throwable failure) {
// filter out old registration attempts
- if (Objects.equals(getTargetLeaderId(), currentJobMasterId)) {
+ if (Objects.equals(getTargetLeaderId(), getCurrentJobMasterId())) {
log.info("Failed to register at job manager {} for job {}.", getTargetAddress(), jobId);
jobLeaderListener.handleError(failure);
} else {
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
new file mode 100644
index 0000000..51595e8
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/JobLeaderServiceTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.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.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.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.Test;
+
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Tests for the {@link JobLeaderService}.
+ */
+public class JobLeaderServiceTest extends TestLogger {
+
+ @ClassRule
+ public static final TestingRpcServiceResource RPC_SERVICE_RESOURCE = new TestingRpcServiceResource();
+
+ /**
+ * Tests that we can concurrently modify the JobLeaderService and complete the leader retrieval operation.
+ * See FLINK-16373.
+ */
+ @Test
+ public void handlesConcurrentJobAdditionsAndLeaderChanges() throws Exception {
+ final JobLeaderService jobLeaderService = new JobLeaderService(
+ new LocalTaskManagerLocation(),
+ RetryingRegistrationConfiguration.defaultConfiguration());
+
+ final TestingJobLeaderListener jobLeaderListener = new TestingJobLeaderListener();
+ final int numberOperations = 20;
+ final BlockingQueue<SettableLeaderRetrievalService> instantiatedLeaderRetrievalServices = new ArrayBlockingQueue<>(numberOperations);
+
+ final HighAvailabilityServices haServices = new TestingHighAvailabilityServicesBuilder()
+ .setJobMasterLeaderRetrieverFunction(
+ leaderForJobId -> {
+ final SettableLeaderRetrievalService leaderRetrievalService = new SettableLeaderRetrievalService();
+ instantiatedLeaderRetrievalServices.offer(leaderRetrievalService);
+ return leaderRetrievalService;
+ })
+ .build();
+
+ jobLeaderService.start(
+ "foobar",
+ RPC_SERVICE_RESOURCE.getTestingRpcService(),
+ haServices,
+ jobLeaderListener);
+
+ final CheckedThread addJobAction = new CheckedThread() {
+ @Override
+ public void go() throws Exception {
+ for (int i = 0; i < numberOperations; i++) {
+ final JobID jobId = JobID.generate();
+ jobLeaderService.addJob(jobId, "foobar");
+ Thread.yield();
+ jobLeaderService.removeJob(jobId);
+ }
+ }
+ };
+ addJobAction.start();
+
+ for (int i = 0; i < numberOperations; i++) {
+ final SettableLeaderRetrievalService leaderRetrievalService = instantiatedLeaderRetrievalServices.take();
+ leaderRetrievalService.notifyListener("foobar", UUID.randomUUID());
+ }
+
+ addJobAction.sync();
+ }
+
+ private static final class TestingJobLeaderListener implements JobLeaderListener {
+
+ @Override
+ public void jobManagerGainedLeadership(JobID jobId, JobMasterGateway jobManagerGateway, JMTMRegistrationSuccess registrationMessage) {
+ // ignored
+ }
+
+ @Override
+ public void jobManagerLostLeadership(JobID jobId, JobMasterId jobMasterId) {
+ // ignored
+ }
+
+ @Override
+ public void handleError(Throwable throwable) {
+ // ignored
+ }
+ }
+}