You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/10/24 07:40:48 UTC
[GitHub] [flink] reswqa opened a new pull request, #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
reswqa opened a new pull request, #21137:
URL: https://github.com/apache/flink/pull/21137
## What is the purpose of the change
*Currently, in JobMasterServiceLeadershipRunner, Leader event callback such as grantLeadership and revokeLeadership will be invoked with nested lock(e.g. curator's event lock and DefaultLeaderElectionService's lock) and try to acquire JobMasterServiceLeadershipRunner's own lock. This will lead to many deadlock risks, such as the simultaneous close and grantLeadership. To avoid this, we should execute this call back in a separate thread to get rid of the nested structure of locks.*
## Brief change log
- *JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock*
## Verifying this change
This change is covered partially by existing tests and added a new test*.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
xintongsong commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1320061643
@reswqa, as previously discussed, please remember to open a ticket for fixing the leader election service side and all other contenders.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1007555607
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -683,21 +665,66 @@ public void testJobAlreadyDone() throws Exception {
jobManagerRunner.getResultFuture();
JobManagerRunnerResult result = resultFuture.get();
- assertEquals(
- JobStatus.FAILED,
- result.getExecutionGraphInfo().getArchivedExecutionGraph().getState());
+ assertThat(result.getExecutionGraphInfo().getArchivedExecutionGraph().getState())
+ .isEqualTo(JobStatus.FAILED);
}
}
+ @Test
+ void testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip()
+ throws Exception {
+ final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
+ testingLeaderElectionDriverFactory =
+ new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
+ final LeaderElectionService defaultLeaderElectionService =
+ new DefaultLeaderElectionService(testingLeaderElectionDriverFactory);
+
+ final JobMasterServiceLeadershipRunner jobManagerRunner =
+ newJobMasterServiceLeadershipRunnerBuilder()
+ .setJobMasterServiceProcessFactory(
+ TestingJobMasterServiceProcessFactory.newBuilder().build())
+ .setLeaderElectionService(defaultLeaderElectionService)
+ .build();
+
+ jobManagerRunner.start();
+ final TestingLeaderElectionDriver currentLeaderDriver =
+ Preconditions.checkNotNull(
+ testingLeaderElectionDriverFactory.getCurrentLeaderDriver());
+
+ final CheckedThread contenderCloseThread =
+ new CheckedThread() {
+ @Override
+ public void go() {
+ try {
+ jobManagerRunner.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ contenderCloseThread.start();
+
+ // grant leadership.
+ currentLeaderDriver.isLeader();
Review Comment:
@XComp Thank you for your advice.
At the beginning, I didn't think of a good way to control the completion of completableFuture, and the probability of deadlock reproduce in my local environment is very high. But your suggestion did find a good control injection point, so i decided to test it in this way.
But if we want to reproduce the problem 100%, it seems that we need another control point. Maybe TestingLeaderElectionDriver#isLeader is a suitable choice. What's your opinion?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -683,21 +665,66 @@ public void testJobAlreadyDone() throws Exception {
jobManagerRunner.getResultFuture();
JobManagerRunnerResult result = resultFuture.get();
- assertEquals(
- JobStatus.FAILED,
- result.getExecutionGraphInfo().getArchivedExecutionGraph().getState());
+ assertThat(result.getExecutionGraphInfo().getArchivedExecutionGraph().getState())
+ .isEqualTo(JobStatus.FAILED);
}
}
+ @Test
+ void testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip()
+ throws Exception {
+ final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
+ testingLeaderElectionDriverFactory =
+ new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
+ final LeaderElectionService defaultLeaderElectionService =
+ new DefaultLeaderElectionService(testingLeaderElectionDriverFactory);
+
+ final JobMasterServiceLeadershipRunner jobManagerRunner =
+ newJobMasterServiceLeadershipRunnerBuilder()
+ .setJobMasterServiceProcessFactory(
+ TestingJobMasterServiceProcessFactory.newBuilder().build())
+ .setLeaderElectionService(defaultLeaderElectionService)
+ .build();
+
+ jobManagerRunner.start();
+ final TestingLeaderElectionDriver currentLeaderDriver =
+ Preconditions.checkNotNull(
+ testingLeaderElectionDriverFactory.getCurrentLeaderDriver());
+
+ final CheckedThread contenderCloseThread =
+ new CheckedThread() {
+ @Override
+ public void go() {
+ try {
+ jobManagerRunner.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ contenderCloseThread.start();
+
+ // grant leadership.
+ currentLeaderDriver.isLeader();
Review Comment:
@XComp Thank you for your advice.
At the beginning, I didn't think of a good way to control the completion of completableFuture, and the probability of deadlock reproduce in my local environment is very high. But your suggestion did find a good control injection point, so i decided to test it in this way.
But if we want to reproduce the problem 100%, it seems that we need another control point. Maybe `TestingLeaderElectionDriver#isLeader` is a suitable choice. What's your opinion?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
XComp commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1300806942
> To me the issue stems more from both the runner and election service calling into each other under locks (== fundamental issue that should never happen), and locks maybe being way too broad.
For example, why is Runner#closeAsync doing the entire shutdown under the lock? Modifying the state should suffice, because all other operations are checking that it's running.
As far as we concluded, the problem appears if the `CompletableFuture` that's returned by `JobMasterServiceProcess#closeAsync` (see [JobMasterServiceLeadershipRunner.java:145](https://github.com/apache/flink/blob/113299701cc0c41bf7fc4bbe86cebd3beea8dbe3/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java#L145)) completes before triggering the callback that follows (that releases the `ClassLoaderLease` and stops the `DefaultLeaderElectionService`). In that case, the entire callback will be executed in the synchronized block rightaway, which is not what we want. We could work around that issue by calling `runAfterwardsAsync`, instead, which would make sure that the followup calls are not executed within the synchronized block. It feels like a similar pattern (i.e. introducing an async execution in a separate thread) to what we have in the PR right now (just in a different location). Does that sound more reasonable?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
xintongsong commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1301630684
Thanks @reswqa for the fix, and thanks all for the discussion.
For this specific case, I think it indeed can be fixed either way, introducing a thread or making sure `leaderElectionService.stop()` is called outside the lock scope. However, from the perspective of maintenance, I think having a dedicated `handleLeaderEventExecutor` is easier to understand and maintain compared to carefully decide which lines of code needs to be executed under lock and which needn't. Admittedly, it comes with a price of an extra thread, which is probably fine in this one-per-contender case.
So +1 for the current approach. The PR LGTM.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1013588761
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -46,45 +49,36 @@
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
+import org.apache.flink.util.function.ThrowingRunnable;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import javax.annotation.Nonnull;
-import java.time.Duration;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsMessage;
-import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link JobMasterServiceLeadershipRunner}. */
-public class JobMasterServiceLeadershipRunnerTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
Review Comment:
We already have `src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension` in flink-runtime module.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1007555607
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -683,21 +665,66 @@ public void testJobAlreadyDone() throws Exception {
jobManagerRunner.getResultFuture();
JobManagerRunnerResult result = resultFuture.get();
- assertEquals(
- JobStatus.FAILED,
- result.getExecutionGraphInfo().getArchivedExecutionGraph().getState());
+ assertThat(result.getExecutionGraphInfo().getArchivedExecutionGraph().getState())
+ .isEqualTo(JobStatus.FAILED);
}
}
+ @Test
+ void testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip()
+ throws Exception {
+ final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
+ testingLeaderElectionDriverFactory =
+ new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
+ final LeaderElectionService defaultLeaderElectionService =
+ new DefaultLeaderElectionService(testingLeaderElectionDriverFactory);
+
+ final JobMasterServiceLeadershipRunner jobManagerRunner =
+ newJobMasterServiceLeadershipRunnerBuilder()
+ .setJobMasterServiceProcessFactory(
+ TestingJobMasterServiceProcessFactory.newBuilder().build())
+ .setLeaderElectionService(defaultLeaderElectionService)
+ .build();
+
+ jobManagerRunner.start();
+ final TestingLeaderElectionDriver currentLeaderDriver =
+ Preconditions.checkNotNull(
+ testingLeaderElectionDriverFactory.getCurrentLeaderDriver());
+
+ final CheckedThread contenderCloseThread =
+ new CheckedThread() {
+ @Override
+ public void go() {
+ try {
+ jobManagerRunner.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ contenderCloseThread.start();
+
+ // grant leadership.
+ currentLeaderDriver.isLeader();
Review Comment:
@XComp Thank you for your advice.
At the beginning, I didn't think of a good way to control the completion of completableFuture, and the probability of deadlock reproduce in my local environment is very high. But your suggestion did find a good control injection point, so i decided to test it in this way.
But if we want to reproduce the problem 100%, it seems that we need another control point. Maybe `TestingLeaderElectionDriver#isLeader` is a suitable choice. What's your opinion?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1301765413
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong closed pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
xintongsong closed pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
URL: https://github.com/apache/flink/pull/21137
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1292379575
> Thanks @reswqa for this PR. I'm wondering how executing the leadership granting/revocation call from within another thread would help fixing the issue. The locks might be still acquired concurrently in opposite orders leading to the deadlock situation.
>
> The usecase that was described in [FLINK-29234](https://issues.apache.org/jira/browse/FLINK-29234) essentially happens because the Dispatcher is stopped (which, as a consequence, would stop `JobMasterServiceLeadershipRunner`) while the `JobMasterServiceLeadershipRunner` is granted leadership causing the locks to be acquired in the opposite order.
>
> I think the problem is that we're still trying to acquire the lock in [JobMasterServiceLeadershipRunner#runIfStateRunning:453](https://github.com/apache/flink/blob/bfe4f9cc3d67d37a2258ab4226d70b6a7d24f22c/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java#L453) even though the `JobMasterServiceLeadershipRunner` is already switched to `STOPPED` state. I'm wondering whether we could make `JobMasterServiceLeadershipRunner#state` volatile and check the instance being in `RUNNING` state outside of the lock. But this wouldn't solve the issue entirely because there's still a slight chance that the state changes after the state check is processed but before entering the lock... Essentially, we would need another lock guarding the state field. 🤔
Thanks @XComp for your feedback. I agree that the reason for deadlock is that the `JobMasterServiceLeadershipRunner` was granted leadership when it was being closed.
At the beginning, I tried to make the `leaderContender.grantLeadership (issuedLeaderSessionID)` in `onGrantLeadership ` method moved outside the lock of `DefaultLeaderElectionService` to avoid nesting locks. But later, I found that there is another layer of lock nesting relationship. When executing `isLeader` and other methods, the curator framework will add a lock itself and can still reproduce the deadlock phenomenon.
So I began to consider whether there was a way to completely avoid the nesting of locks when calling the `grantLeadership` of the `JobMasterServiceLeadershipRunner`. Inspired by `ResourceManagerServiceImpl` (which executes the leader event in a separate single thread executor), I thought that the same way should also be used to avoid the nesting of lock structures. When we let a piece of code to another thread for execution, the locks placed on the current thread will not be moved to another thread, which will enable the logic of `granteLeadership` in `JobMasterServiceLeadershipRunner` has chance to get rid of the locks that may be imposed by the caller (whether from the leaderSelectionService or from the Curator).
We record the lock in `DefaultLeaderSelectionService` as A and the lock in `JobMasterServiceLeadershipRunner` as B.
FLINK-29234's case is as following:
Thread t1 invoke `JobMasterServiceLeadershipRunner#closeAsync`, holds lock B, and then attempts to call `DefaultLeaderElectionService#stop` to obtain lock A.
Thread t2 invoke `DefaultLeaderElectionService#onGrantLeadership`, hold lock A, and then try to call `JobMasterServiceLeadershipRunner#grantLeadership` to obtain lock B.
After introducing an executor to process `grantLeadership`, t2 calling `onGrantLeadership` will hold lock A, move the code logic into the executor, and then release lock A without attempting to acquire lock B. It is the thread in executor's responsibility to obtain B and do the real work.
If I miss something, you can tell me. I'm willing to listen to your opinion, because I'm not particularly familiar with this part code.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1301815374
I don't see how this is supposed to be good from a maintenance perspective.
If you look at the code without context you now have 2 arbitrary calls go through an executor and you make similar decisions as with locking ("what operations should run in the executor").
The javadocs states that all operations are serialized, yet we need an executor for certain operations to make things work. That's just an accident waiting to happen.
Additionally the proposal changes the behavior slightly, since calling grant/revokeLeadership no longer has an immediate effect (as it now effectively runs in the background). If this is ok (not sure if it is), then I'd rather see us pushing more code into the `sequentialOperation` and always running that in the executor. Then you at least have a consistent behavior.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1303034139
Thanks everyone for your comments and discussion, which is very meaningful to me. Based on the above discussion, I think @xintongsong's suggestion looks good to me, +1 for this plan.
We should first deal with the problem of too broad lock(the real reason of this ticket) at `JobMasterServiceLeadershipRunner#closeAsync`, which should be consistent with @zentol 's opinion. At the same time, I still think we should be committed to preventing the same thing from happening, so I think the second proposal is also very make sense. WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
XComp commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1006684103
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -683,21 +665,66 @@ public void testJobAlreadyDone() throws Exception {
jobManagerRunner.getResultFuture();
JobManagerRunnerResult result = resultFuture.get();
- assertEquals(
- JobStatus.FAILED,
- result.getExecutionGraphInfo().getArchivedExecutionGraph().getState());
+ assertThat(result.getExecutionGraphInfo().getArchivedExecutionGraph().getState())
+ .isEqualTo(JobStatus.FAILED);
}
}
+ @Test
+ void testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip()
+ throws Exception {
+ final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
+ testingLeaderElectionDriverFactory =
+ new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
+ final LeaderElectionService defaultLeaderElectionService =
+ new DefaultLeaderElectionService(testingLeaderElectionDriverFactory);
+
+ final JobMasterServiceLeadershipRunner jobManagerRunner =
+ newJobMasterServiceLeadershipRunnerBuilder()
+ .setJobMasterServiceProcessFactory(
+ TestingJobMasterServiceProcessFactory.newBuilder().build())
+ .setLeaderElectionService(defaultLeaderElectionService)
+ .build();
+
+ jobManagerRunner.start();
+ final TestingLeaderElectionDriver currentLeaderDriver =
+ Preconditions.checkNotNull(
+ testingLeaderElectionDriverFactory.getCurrentLeaderDriver());
+
+ final CheckedThread contenderCloseThread =
+ new CheckedThread() {
+ @Override
+ public void go() {
+ try {
+ jobManagerRunner.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ contenderCloseThread.start();
+
+ // grant leadership.
+ currentLeaderDriver.isLeader();
Review Comment:
This test is not necessarily failing when running it without the fix you're proposing in this PR. The `close()` call of `jobManagerRunner` needs to be halted to influence the concurrent processing of the `currentLeaderDriver.isLeader()` and the `jobManagerRunner.close()` method. Ideally, this method should fail/block consistently without the fix but should succeed with the fix.
I guess, you could achieve that by inducing your own `TestingJobMasterServiceProcess` through the `TestingJobMasterServiceProcessFactory` in [line 685](https://github.com/apache/flink/pull/21137/files#diff-c010bcd788571261ea6c43a17ad7aaea719a64a7b8a080d76939e7dbbc5a4b32R685). This `TestingJobMasterServiceProcess` gives you control over the `CompletableFuture` that's returned by [JobMasterServiceProcess.closeAsync() in JobMasterServiceLeadershipRunner.closeAsync()](https://github.com/apache/flink/blob/bfe4f9cc3d67d37a2258ab4226d70b6a7d24f22c/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java#L145). The `JobMasterServiceLeadershipRunner` will only proceeed with stopping the `DefaultLeaderElectionService` if this `CompletableFuture` completes. WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1296507911
> We might also want to check the other `LeaderContender` implementations (as you already mentioned, `ResourceManagerImpl` does not suffer from this bug but the others might because they don't use this `handleLeaderEventHandler` pattern.
@XComp I totally agree that we should also check the other `LeaderContender` to ensure they are not suffer from this. However, from my point of view, in consideration of limiting the scope of this pull request, I suggest that only the `JobMasterServiceLeadershipRunner` be tested in this PR. For other `LeaderContender`, I will open a separate ticket for tracking. WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1013672492
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -93,48 +87,47 @@ public class JobMasterServiceLeadershipRunnerTest extends TestLogger {
private JobResultStore jobResultStore;
- @BeforeClass
- public static void setupClass() {
+ @BeforeAll
+ static void setupClass() {
final JobVertex jobVertex = new JobVertex("Test vertex");
jobVertex.setInvokableClass(NoOpInvokable.class);
jobGraph = JobGraphTestUtils.streamingJobGraph(jobVertex);
}
- @Before
- public void setup() {
+ @BeforeEach
+ void setup() {
leaderElectionService = new TestingLeaderElectionService();
jobResultStore = new EmbeddedJobResultStore();
fatalErrorHandler = new TestingFatalErrorHandler();
}
- @After
- public void tearDown() throws Exception {
+ @AfterEach
+ void tearDown() throws Exception {
fatalErrorHandler.rethrowError();
}
@Test
- public void testShutDownSignalsJobAsNotFinished() throws Exception {
+ void testShutDownSignalsJobAsNotFinished() throws Exception {
try (JobManagerRunner jobManagerRunner =
newJobMasterServiceLeadershipRunnerBuilder().build()) {
jobManagerRunner.start();
final CompletableFuture<JobManagerRunnerResult> resultFuture =
jobManagerRunner.getResultFuture();
- assertThat(resultFuture.isDone(), is(false));
+ assertThat(resultFuture).isNotDone();
jobManagerRunner.closeAsync();
assertJobNotFinished(resultFuture);
- assertThat(
- jobManagerRunner.getJobMasterGateway(),
- FlinkMatchers.futureWillCompleteExceptionally(Duration.ofMillis(5L)));
+ assertThat(jobManagerRunner.getJobMasterGateway())
+ .failsWithin(5L, TimeUnit.MILLISECONDS);
Review Comment:
BTW, I have not removed the timeout in the test at present, because according to our discussion, the production code maybe should change, and I will modify it together after we reach an agreement on the implementation.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1296821182
@XComp About your [proposal](https://github.com/reswqa/flink/pull/1) , I add a very short sleep after `triggerClassLoaderLeaseRelease.await();`, the deadlock problem can be 100% reproduced on my own machine. Although theoretically there is still the possibility of non recurrence, but I am sure that this short sleep can almost absolutely trigger `grantLeadership` before `running` state becomes false.
Personally, I think it is acceptable to introduce a very short sleep (I used 5ms) to fight for the deadlock problem to recur almost every time, and it will not significantly increase the running time of CI. Of course, if we can think of a better way to solve this problem, I will be more willing to accept it.
In addition, I found that if sleep is not introduced, run the whole test class on my and my colleagues' machines, the deadlock problem can hardly be reproduced. However, if run `testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip` separately, the probability of deadlock will be greatly increased. I don't have a deep understanding of Junit's testing framework, so I don't know why there is such a difference. You can try to run the entire test class directly to help you reproduce the problem.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1302209878
> Well, that should also work. I don't see much differences compared to the current approach, except for more code changes. But I'm not opposed to it.
One advantage of my proposal is that you can't run into TOCTOU races. If the state check runs as part of the sequential operation we are sure that it only runs if we're actually still running by the time the executor processes is.
As is, you check that its currently running, schedule an operation, but by the time the operation runs we may have already shut down.
The second main defect is that some calls prepared in closeAsync may _or may not_ run under the lock. That's something we must fix in any case (and is the actual bug fix!); the introduction of the executor could be a means to that end, or just be a refactoring.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
xintongsong commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1303003420
This is indeed a fascinating discussion.
I see the point that there are some calls in `cloaseAsync` that may or may not run under the lock. I also agree that it would be nice the lock only covers the calls that need to be guarded and leave the unnecessary calls outside.
I think my focus is more about preventing (or at least reducing the chances of) similar problems. In that sense, it's probably better that the runner does not need to (or as least as possible) worry about another lock from the leader election service. To that end, I think @XComp's proposal makes more sense than mine. Ideally, leader election service should not call `contender.grant/revokeLeadership()` under a lock while the same lock can be accessed by the contender via `stop()`.
To clarify, fixing the problem on the service side does not justify that calls in `cloaseAsync` may or may not run under the lock. Ideally, `A` should not call `B.b()` under a lock while another method `A.a()` called by `B` can access the same lock. That applies to both the service and the runner, and maybe we should fix that for both.
All in all, I'd suggest:
- Fix the runner side issue in `closeAsync` in this PR. For this purpose solely, we probably don't need a dedicated executor.
- Fix the leader election service side issue with a dedicated executor, in a follow up ticket. That would affect all contenders and we need to carefully check that no existing contenders are relying on the current behavior that `grant/revokeLeadership` are called under lock. We should also clean up things like `ResourceManagerServiceImpl.handleLeaderEventExecutor`.
WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1013644830
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java:
##########
@@ -151,10 +155,12 @@ public CompletableFuture<Void> closeAsync() {
FutureUtils.forward(serviceTerminationFuture, terminationFuture);
terminationFuture.whenComplete(
- (unused, throwable) ->
- LOG.debug(
- "Leadership runner for job {} has been terminated.",
- getJobID()));
+ (unused, throwable) -> {
+ LOG.debug(
+ "Leadership runner for job {} has been terminated.",
+ getJobID());
+ handleLeaderEventExecutor.shutdown();
+ });
Review Comment:
good idea.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1007555509
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -683,21 +665,66 @@ public void testJobAlreadyDone() throws Exception {
jobManagerRunner.getResultFuture();
JobManagerRunnerResult result = resultFuture.get();
- assertEquals(
- JobStatus.FAILED,
- result.getExecutionGraphInfo().getArchivedExecutionGraph().getState());
+ assertThat(result.getExecutionGraphInfo().getArchivedExecutionGraph().getState())
+ .isEqualTo(JobStatus.FAILED);
}
}
+ @Test
+ void testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip()
+ throws Exception {
+ final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
+ testingLeaderElectionDriverFactory =
+ new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
+ final LeaderElectionService defaultLeaderElectionService =
+ new DefaultLeaderElectionService(testingLeaderElectionDriverFactory);
+
+ final JobMasterServiceLeadershipRunner jobManagerRunner =
+ newJobMasterServiceLeadershipRunnerBuilder()
+ .setJobMasterServiceProcessFactory(
+ TestingJobMasterServiceProcessFactory.newBuilder().build())
+ .setLeaderElectionService(defaultLeaderElectionService)
+ .build();
+
+ jobManagerRunner.start();
+ final TestingLeaderElectionDriver currentLeaderDriver =
+ Preconditions.checkNotNull(
+ testingLeaderElectionDriverFactory.getCurrentLeaderDriver());
+
+ final CheckedThread contenderCloseThread =
+ new CheckedThread() {
+ @Override
+ public void go() {
+ try {
+ jobManagerRunner.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ contenderCloseThread.start();
+
+ // grant leadership.
+ currentLeaderDriver.isLeader();
Review Comment:
@XComp Thank you for your advice.
At the beginning, I didn't think of a good way to control the completion of completableFuture, and the probability of deadlock reproduce in my local environment is very high. But your suggestion did find a good control injection point, so i decided to test it in this way.
But if we want to reproduce the problem 100%, it seems that we need another control point. Maybe `TestingLeaderElectionDriver#isLeader` is a suitable choice. What's your opinion?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1291488778
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
XComp commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1295315230
We might also want to check the other `LeaderContender` implementations (as you already mentioned, `ResourceManagerImpl` does not suffer from this bug but the others might because they don't use this `handleLeaderEventHandler` pattern.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
XComp commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1296841251
> @XComp I totally agree that we should also check the other LeaderContender to ensure they are not suffer from this. However, from my point of view, in consideration of limiting the scope of this pull request, I suggest that only the JobMasterServiceLeadershipRunner be tested in this PR. For other LeaderContender, I will open a separate ticket for tracking. WDYT?
Yup, I agree. Creating follow-up Jira issues makes sense.
> testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip separately, the probability of deadlock will be greatly increased. I don't have a deep understanding of Junit's testing framework, so I don't know why there is such a difference. Maybe you can try to run the entire test class directly to help you reproduce the problem.
Thanks for the hint. :+1:
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1303507652
Sounds good to me.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1026158401
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -46,45 +49,36 @@
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
+import org.apache.flink.util.function.ThrowingRunnable;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import javax.annotation.Nonnull;
-import java.time.Duration;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsMessage;
-import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link JobMasterServiceLeadershipRunner}. */
-public class JobMasterServiceLeadershipRunnerTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
Review Comment:
> We already have src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension in flink-runtime module.
Which means you don't need to add `@ExtendWith` :)
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
XComp commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1008330056
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -683,21 +665,66 @@ public void testJobAlreadyDone() throws Exception {
jobManagerRunner.getResultFuture();
JobManagerRunnerResult result = resultFuture.get();
- assertEquals(
- JobStatus.FAILED,
- result.getExecutionGraphInfo().getArchivedExecutionGraph().getState());
+ assertThat(result.getExecutionGraphInfo().getArchivedExecutionGraph().getState())
+ .isEqualTo(JobStatus.FAILED);
}
}
+ @Test
+ void testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip()
+ throws Exception {
+ final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
+ testingLeaderElectionDriverFactory =
+ new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
+ final LeaderElectionService defaultLeaderElectionService =
+ new DefaultLeaderElectionService(testingLeaderElectionDriverFactory);
+
+ final JobMasterServiceLeadershipRunner jobManagerRunner =
+ newJobMasterServiceLeadershipRunnerBuilder()
+ .setJobMasterServiceProcessFactory(
+ TestingJobMasterServiceProcessFactory.newBuilder().build())
+ .setLeaderElectionService(defaultLeaderElectionService)
+ .build();
+
+ jobManagerRunner.start();
+ final TestingLeaderElectionDriver currentLeaderDriver =
+ Preconditions.checkNotNull(
+ testingLeaderElectionDriverFactory.getCurrentLeaderDriver());
+
+ final CheckedThread contenderCloseThread =
+ new CheckedThread() {
+ @Override
+ public void go() {
+ try {
+ jobManagerRunner.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ contenderCloseThread.start();
+
+ // grant leadership.
+ currentLeaderDriver.isLeader();
Review Comment:
I went ahead and created a PR with a running example. See reswqa/flink#1 and feel free to edit it. Please see the PRs description. It's not meant to be merged as is but rather an example on how you would be able to test it. I'm happy to discuss any questions you have. But I hope the inline comments are helpful already.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1013656229
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -93,48 +87,47 @@ public class JobMasterServiceLeadershipRunnerTest extends TestLogger {
private JobResultStore jobResultStore;
- @BeforeClass
- public static void setupClass() {
+ @BeforeAll
+ static void setupClass() {
final JobVertex jobVertex = new JobVertex("Test vertex");
jobVertex.setInvokableClass(NoOpInvokable.class);
jobGraph = JobGraphTestUtils.streamingJobGraph(jobVertex);
}
- @Before
- public void setup() {
+ @BeforeEach
+ void setup() {
leaderElectionService = new TestingLeaderElectionService();
jobResultStore = new EmbeddedJobResultStore();
fatalErrorHandler = new TestingFatalErrorHandler();
}
- @After
- public void tearDown() throws Exception {
+ @AfterEach
+ void tearDown() throws Exception {
fatalErrorHandler.rethrowError();
}
@Test
- public void testShutDownSignalsJobAsNotFinished() throws Exception {
+ void testShutDownSignalsJobAsNotFinished() throws Exception {
try (JobManagerRunner jobManagerRunner =
newJobMasterServiceLeadershipRunnerBuilder().build()) {
jobManagerRunner.start();
final CompletableFuture<JobManagerRunnerResult> resultFuture =
jobManagerRunner.getResultFuture();
- assertThat(resultFuture.isDone(), is(false));
+ assertThat(resultFuture).isNotDone();
jobManagerRunner.closeAsync();
assertJobNotFinished(resultFuture);
- assertThat(
- jobManagerRunner.getJobMasterGateway(),
- FlinkMatchers.futureWillCompleteExceptionally(Duration.ofMillis(5L)));
+ assertThat(jobManagerRunner.getJobMasterGateway())
+ .failsWithin(5L, TimeUnit.MILLISECONDS);
Review Comment:
Thank you for reminding me about this. If I understand correctly, can I change this timeout assertion to the approach like `assertThatThrownBy(() -> jobManagerRunner.getJobMasterGateway().get())` to rely on Azure's global timeout in this way?
As for the migration of JUnit5, I saw this prompt: "Unless there is a specific reason, make sure you use JUnit 5 and AssertJ when contributing to Flink with new tests and even when modifying existing tests." in the `Apache Flink Code Style and Quality Guide`. From my perspective, it seems a bit bad to have JUnt4 and JUnit5 dependencies in the test class at the same time, so I think this may be necessary.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1301817458
> In that case, the entire callback will be executed in the synchronized block rightaway, which is not what we want.
Which implies that the future calls should never run under the lock in the first place. So let's not put them into the synchronized block.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
xintongsong commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1301899690
> If you look at the code without context you now have 2 arbitrary calls go through an executor and you make similar decisions as with locking ("what operations should run in the executor").
It would be ideal if `JobMasterServiceLeadershipRunner` need not be aware of the lock inside the leader election service. Unfortunately, I don't see a good way to do so.
In the current approach, the name of the executor clearly states that it is supposed to be used for handling the leadership related events, which are granting and revoking. You can safely access the runner lock in any other methods of the runner, without worrying about causing deadlocks with another lock from another class. Otherwise, imagine you called a method `a()` in `grantLeadership()`, and `b()` in `a()`, and `c()` in `b()`, etc. One can easily get confused whether `c()` is called under the leader election service lock and whether it is safe to access the runner lock. That is to explain why I think the current approach is good for the maintenance.
>The javadocs states that all operations are serialized, yet we need an executor for certain operations to make things work. That's just an accident waiting to happen.
I don't get what do you mean by an accident waiting to happen. The JavaDoc says the granting and revoking of leadership are serialized, not the other public methods of the class.
> calling grant/revokeLeadership no longer has an immediate effect (as it now effectively runs in the background)
Exactly. The change is based on the fact that the runner neither need to do the works synchronously in grant/revokeLeadership, nor need/want to be aware of the lock that the leader election service put upon the two methods.
> I'd rather see us pushing more code into the sequentialOperation and always running that in the executor
Well, that should also work. I don't see much differences compare the current approach, except for more code changes. But I'm not opposed to it.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1026170047
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -46,45 +49,36 @@
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
+import org.apache.flink.util.function.ThrowingRunnable;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import javax.annotation.Nonnull;
-import java.time.Duration;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsMessage;
-import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link JobMasterServiceLeadershipRunner}. */
-public class JobMasterServiceLeadershipRunnerTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
Review Comment:
Ahh, Thank you for your clarification, added this annotation accidentally~
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1288564912
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "30b8aee141ae28b687fa2d1eaf47521074686f66",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "30b8aee141ae28b687fa2d1eaf47521074686f66",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 30b8aee141ae28b687fa2d1eaf47521074686f66 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1289898634
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
XComp commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1292206053
Thanks @reswqa for this PR. I'm wondering how executing the leadership granting/revocation being called from within another thread would help fixing the issue. The locks might be still acquired concurrently in opposite orders leading to the deadlock situation.
The usecase that was described in FLINK-29234 essentially happens because the Dispatcher is stopped (which, as a consequence, would stop `JobMasterServiceLeadershipRunner`) while the `JobMasterServiceLeadershipRunner` is granted leadership causing the locks to be acquired in the opposite order.
I think the problem is that we're still trying to acquire the lock in [JobMasterServiceLeadershipRunner#runIfStateRunning:453](https://github.com/apache/flink/blob/bfe4f9cc3d67d37a2258ab4226d70b6a7d24f22c/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java#L453) even though the `JobMasterServiceLeadershipRunner` is already switched to `STOPPED` state. I'm wondering whether we could make `JobMasterServiceLeadershipRunner#state` volatile and check the instance being in `RUNNING` state outside of the lock. But this wouldn't solve the issue entirely because there's still a slight chance that the state changes after the state check is processed but before entering the lock... :thinking:
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
XComp commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1011489421
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -46,45 +49,36 @@
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
+import org.apache.flink.util.function.ThrowingRunnable;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import javax.annotation.Nonnull;
-import java.time.Duration;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsMessage;
-import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link JobMasterServiceLeadershipRunner}. */
-public class JobMasterServiceLeadershipRunnerTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
Review Comment:
Usually, we would create a file `org.junit.jupiter.api.extension.Extension` in `src/test/resources/META-INF/services` containing `org.apache.flink.util.TestLoggerExtension`. This way, the extension applies to all JUnit5 tests in the module (e.g. see `flink-clients` module).
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobMasterServiceProcess.java:
##########
@@ -20,124 +20,114 @@
import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGatewayBuilder;
-import javax.annotation.Nullable;
-
import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
/** Testing implementation of {@link JobMasterServiceProcess}. */
public class TestingJobMasterServiceProcess implements JobMasterServiceProcess {
- private final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture;
-
- private final CompletableFuture<JobManagerRunnerResult> jobManagerRunnerResultFuture;
-
- private final CompletableFuture<String> leaderAddressFuture;
-
- private final boolean isInitialized;
-
- private final CompletableFuture<Void> terminationFuture;
-
- private final boolean manualTerminationFutureCompletion;
-
- private TestingJobMasterServiceProcess(
- CompletableFuture<JobMasterGateway> jobMasterGatewayFuture,
- CompletableFuture<JobManagerRunnerResult> jobManagerRunnerResultFuture,
- CompletableFuture<String> leaderAddressFuture,
- boolean isInitialized,
- CompletableFuture<Void> terminationFuture,
- boolean manualTerminationFutureCompletion) {
- this.jobMasterGatewayFuture = jobMasterGatewayFuture;
- this.jobManagerRunnerResultFuture = jobManagerRunnerResultFuture;
- this.leaderAddressFuture = leaderAddressFuture;
- this.isInitialized = isInitialized;
- this.terminationFuture = terminationFuture;
- this.manualTerminationFutureCompletion = manualTerminationFutureCompletion;
+ private final Supplier<CompletableFuture<Void>> closeAsyncSupplier;
+ private final Supplier<Boolean> isInitializedAndRunningSupplier;
+ private final Supplier<CompletableFuture<JobMasterGateway>> getJobMasterGatewayFutureSupplier;
+ private final Supplier<CompletableFuture<JobManagerRunnerResult>> getResultFutureSupplier;
+ private final Supplier<CompletableFuture<String>> getLeaderAddressFutureSupplier;
+
+ public TestingJobMasterServiceProcess(
+ Supplier<CompletableFuture<Void>> closeAsyncSupplier,
+ Supplier<Boolean> isInitializedAndRunningSupplier,
+ Supplier<CompletableFuture<JobMasterGateway>> getJobMasterGatewayFutureSupplier,
+ Supplier<CompletableFuture<JobManagerRunnerResult>> getResultFutureSupplier,
+ Supplier<CompletableFuture<String>> getLeaderAddressFutureSupplier) {
+ this.closeAsyncSupplier = closeAsyncSupplier;
+ this.isInitializedAndRunningSupplier = isInitializedAndRunningSupplier;
+ this.getJobMasterGatewayFutureSupplier = getJobMasterGatewayFutureSupplier;
+ this.getResultFutureSupplier = getResultFutureSupplier;
+ this.getLeaderAddressFutureSupplier = getLeaderAddressFutureSupplier;
}
@Override
public CompletableFuture<Void> closeAsync() {
- if (!manualTerminationFutureCompletion) {
- terminationFuture.complete(null);
- }
-
- return terminationFuture;
+ return closeAsyncSupplier.get();
}
@Override
public boolean isInitializedAndRunning() {
- return isInitialized && !terminationFuture.isDone();
+ return isInitializedAndRunningSupplier.get();
}
@Override
public CompletableFuture<JobMasterGateway> getJobMasterGatewayFuture() {
- return jobMasterGatewayFuture;
+ return getJobMasterGatewayFutureSupplier.get();
}
@Override
public CompletableFuture<JobManagerRunnerResult> getResultFuture() {
- return jobManagerRunnerResultFuture;
+ return getResultFutureSupplier.get();
}
@Override
public CompletableFuture<String> getLeaderAddressFuture() {
- return leaderAddressFuture;
+ return getLeaderAddressFutureSupplier.get();
}
public static Builder newBuilder() {
return new Builder();
}
public static final class Builder {
- private CompletableFuture<JobMasterGateway> jobMasterGatewayFuture =
- CompletableFuture.completedFuture(new TestingJobMasterGatewayBuilder().build());
- private CompletableFuture<JobManagerRunnerResult> jobManagerRunnerResultFuture =
- new CompletableFuture<>();
- private CompletableFuture<String> leaderAddressFuture =
- CompletableFuture.completedFuture("foobar");
- private boolean isInitialized = true;
- @Nullable private CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
- private boolean manualTerminationFutureCompletion = false;
-
- public Builder setJobMasterGatewayFuture(
- CompletableFuture<JobMasterGateway> jobMasterGatewayFuture) {
- this.jobMasterGatewayFuture = jobMasterGatewayFuture;
- return this;
+ private Supplier<CompletableFuture<Void>> closeAsyncSupplier = unsupportedOperation();
+ private Supplier<Boolean> isInitializedAndRunningSupplier = unsupportedOperation();
+ private Supplier<CompletableFuture<JobMasterGateway>> getJobMasterGatewayFutureSupplier =
+ () ->
+ CompletableFuture.completedFuture(
+ new TestingJobMasterGatewayBuilder().build());;
Review Comment:
```suggestion
new TestingJobMasterGatewayBuilder().build());
```
Checkstyle complaining here again
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -683,21 +686,148 @@ public void testJobAlreadyDone() throws Exception {
jobManagerRunner.getResultFuture();
JobManagerRunnerResult result = resultFuture.get();
- assertEquals(
- JobStatus.FAILED,
- result.getExecutionGraphInfo().getArchivedExecutionGraph().getState());
+ assertThat(result.getExecutionGraphInfo().getArchivedExecutionGraph().getState())
+ .isEqualTo(JobStatus.FAILED);
}
}
+ @Test
+ void testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip()
+ throws Exception {
+ final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
+ testingLeaderElectionDriverFactory =
+ new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
+
+ // we need to use DefaultLeaderElectionService here because JobMasterServiceLeadershipRunner
+ // in connection with the DefaultLeaderElectionService generates the nested locking
+ final LeaderElectionService defaultLeaderElectionService =
+ new DefaultLeaderElectionService(testingLeaderElectionDriverFactory);
+
+ // latch to detect when we reached the first synchronized section having a lock on the
+ // JobMasterServiceProcess#stop side
+ final OneShotLatch closeAsyncCalledTrigger = new OneShotLatch();
+ // latch to halt the JobMasterServiceProcess#stop before calling stop on the
+ // DefaultLeaderElectionService instance (and entering the LeaderElectionService's
+ // synchronized block)
+ final OneShotLatch triggerClassLoaderLeaseRelease = new OneShotLatch();
+
+ final JobMasterServiceProcess jobMasterServiceProcess =
+ TestingJobMasterServiceProcess.newBuilder()
+ .setGetJobMasterGatewayFutureSupplier(CompletableFuture::new)
+ .setGetResultFutureSupplier(CompletableFuture::new)
+ .setGetLeaderAddressFutureSupplier(
+ () -> CompletableFuture.completedFuture("unused address"))
+ .setCloseAsyncSupplier(
+ () -> {
+ closeAsyncCalledTrigger.trigger();
+ // we have to return a completed future because we need the
+ // follow-up task to run in the calling thread to make the
+ // follow-up logic block be executed in the synchronized block
Review Comment:
```suggestion
// follow-up code block being executed in the synchronized block
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java:
##########
@@ -151,10 +155,12 @@ public CompletableFuture<Void> closeAsync() {
FutureUtils.forward(serviceTerminationFuture, terminationFuture);
terminationFuture.whenComplete(
- (unused, throwable) ->
- LOG.debug(
- "Leadership runner for job {} has been terminated.",
- getJobID()));
+ (unused, throwable) -> {
+ LOG.debug(
+ "Leadership runner for job {} has been terminated.",
+ getJobID());
+ handleLeaderEventExecutor.shutdown();
+ });
Review Comment:
Could we make the resulting `CompletableFuture` of this call being returned by `closeAsync`. Shutting down the executor is part of the closing procedure and, therefore, should be reflected in the method's result.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobMasterServiceProcess.java:
##########
@@ -20,124 +20,114 @@
import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGatewayBuilder;
-import javax.annotation.Nullable;
-
import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
/** Testing implementation of {@link JobMasterServiceProcess}. */
public class TestingJobMasterServiceProcess implements JobMasterServiceProcess {
- private final CompletableFuture<JobMasterGateway> jobMasterGatewayFuture;
-
- private final CompletableFuture<JobManagerRunnerResult> jobManagerRunnerResultFuture;
-
- private final CompletableFuture<String> leaderAddressFuture;
-
- private final boolean isInitialized;
-
- private final CompletableFuture<Void> terminationFuture;
-
- private final boolean manualTerminationFutureCompletion;
-
- private TestingJobMasterServiceProcess(
- CompletableFuture<JobMasterGateway> jobMasterGatewayFuture,
- CompletableFuture<JobManagerRunnerResult> jobManagerRunnerResultFuture,
- CompletableFuture<String> leaderAddressFuture,
- boolean isInitialized,
- CompletableFuture<Void> terminationFuture,
- boolean manualTerminationFutureCompletion) {
- this.jobMasterGatewayFuture = jobMasterGatewayFuture;
- this.jobManagerRunnerResultFuture = jobManagerRunnerResultFuture;
- this.leaderAddressFuture = leaderAddressFuture;
- this.isInitialized = isInitialized;
- this.terminationFuture = terminationFuture;
- this.manualTerminationFutureCompletion = manualTerminationFutureCompletion;
+ private final Supplier<CompletableFuture<Void>> closeAsyncSupplier;
+ private final Supplier<Boolean> isInitializedAndRunningSupplier;
+ private final Supplier<CompletableFuture<JobMasterGateway>> getJobMasterGatewayFutureSupplier;
+ private final Supplier<CompletableFuture<JobManagerRunnerResult>> getResultFutureSupplier;
+ private final Supplier<CompletableFuture<String>> getLeaderAddressFutureSupplier;
+
+ public TestingJobMasterServiceProcess(
Review Comment:
```suggestion
private TestingJobMasterServiceProcess(
```
Intellij rightfully claimed that `private` is good enough here.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -93,48 +87,47 @@ public class JobMasterServiceLeadershipRunnerTest extends TestLogger {
private JobResultStore jobResultStore;
- @BeforeClass
- public static void setupClass() {
+ @BeforeAll
+ static void setupClass() {
final JobVertex jobVertex = new JobVertex("Test vertex");
jobVertex.setInvokableClass(NoOpInvokable.class);
jobGraph = JobGraphTestUtils.streamingJobGraph(jobVertex);
}
- @Before
- public void setup() {
+ @BeforeEach
+ void setup() {
leaderElectionService = new TestingLeaderElectionService();
jobResultStore = new EmbeddedJobResultStore();
fatalErrorHandler = new TestingFatalErrorHandler();
}
- @After
- public void tearDown() throws Exception {
+ @AfterEach
+ void tearDown() throws Exception {
fatalErrorHandler.rethrowError();
}
@Test
- public void testShutDownSignalsJobAsNotFinished() throws Exception {
+ void testShutDownSignalsJobAsNotFinished() throws Exception {
try (JobManagerRunner jobManagerRunner =
newJobMasterServiceLeadershipRunnerBuilder().build()) {
jobManagerRunner.start();
final CompletableFuture<JobManagerRunnerResult> resultFuture =
jobManagerRunner.getResultFuture();
- assertThat(resultFuture.isDone(), is(false));
+ assertThat(resultFuture).isNotDone();
jobManagerRunner.closeAsync();
assertJobNotFinished(resultFuture);
- assertThat(
- jobManagerRunner.getJobMasterGateway(),
- FlinkMatchers.futureWillCompleteExceptionally(Duration.ofMillis(5L)));
+ assertThat(jobManagerRunner.getJobMasterGateway())
+ .failsWithin(5L, TimeUnit.MILLISECONDS);
Review Comment:
We agreed on not relying on timeouts in JUnit tests anymore to enable the thread dump and prevent test instabilities (see [Flink's Common Code Guidelines](https://flink.apache.org/contributing/code-style-and-quality-common.html#avoid-timeouts-in-junit-tests)). This also applies to the other test implementations if you want to migrate this test class as part of this PR.
It would be also reasonable to not touch the JUnit5 migration here. I didn't check whether you actually did it to use some JUnit5 feature in that test. That might help focusing on the actual topic of the PR. ...considering that it's not that straight-forward.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on code in PR #21137:
URL: https://github.com/apache/flink/pull/21137#discussion_r1013588761
##########
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java:
##########
@@ -46,45 +49,36 @@
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
+import org.apache.flink.util.function.ThrowingRunnable;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
import javax.annotation.Nonnull;
-import java.time.Duration;
import java.util.ArrayDeque;
import java.util.Arrays;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsMessage;
-import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link JobMasterServiceLeadershipRunner}. */
-public class JobMasterServiceLeadershipRunnerTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
Review Comment:
We already have src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension in flink-runtime module.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1292957867
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1293090172
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
XComp commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1299313832
Thanks for your patience on that PR. I gave it another thought: You're right about the issue with the 5ms delay being necessary to enter the synchronized block of the `grantLeadership` call hierarchy. The issue is that we don't have any means of verify that we're in the synchronized block of `DefaultLeaderElectionService#lock` in `DefaultLeaderElectionService#onGrantLeadership` but we trigger this implicly through `TestingLeaderElectionDriver#isLeader()`. This applies to your and my proposal and, therefore, the delay (i.e. the 5ms) are actually necessary. I'm not a big fan of these 5ms because they tend to make these tests flaky. We should be really clear in an inline comment on why this is necessary.
That said, I tend to prefer my proposal, because it keeps the test-specific logic in the test method. Your proposal adds a test-specific callback to `TestingLeaderElectionDriver` which is some kind of generic test utility class. Mixing test utility classes with actual test implementations messes things up in the long run usually. That's the reason we started to come up with `Testing*` implementations like I proposed with `TestingJobMasterServiceProcess2` in my PR. WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1299649118
@XComp Thank you very much for your suggestions and review. I totally agree with you that introduction of such callbacks in the `TestingLeaderSelectionDriver` will make things a bit troublesome and strange. So I decided to take your proposal. At the same time, I refactor the 'TestingJobMasterServiceProcess', and introduced `Thread.sleep (5ms)` to your proposal and added the corresponding explanation. By the way, I added you as a co-author to thank you for your proposal.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1300443324
I'm not really sold on the idea of solving a deadlock by introducing yet another thread into the mix.
To me the issue stems more from both the runner and election service calling into each other under locks, and locks maybe being way too broad.
For example, why is `Runner#closeAsync` doing the entire shutdown under the lock? Modifying the state _should_ suffice, because all other operations are checking that it's running.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on pull request #21137: [FLINK-29234][runtime] JobMasterServiceLeadershipRunner handle leader event in a separate executor to avoid dead lock
Posted by GitBox <gi...@apache.org>.
reswqa commented on PR #21137:
URL: https://github.com/apache/flink/pull/21137#issuecomment-1301664533
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org