[flink] 10/10: [hotfix][tests] Simplify and harden CoordinatorEventsExactlyOnceITCase

[se...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit e3d2e1c49d6bb89d11e9dfc9c171f9d00aff2215
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Fri Apr 16 16:48:17 2021 +0200

    [hotfix][tests] Simplify and harden CoordinatorEventsExactlyOnceITCase
---
 .../CoordinatorEventsExactlyOnceITCase.java        | 42 +++++++++++-----------
 1 file changed, 21 insertions(+), 21 deletions(-)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/CoordinatorEventsExactlyOnceITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/CoordinatorEventsExactlyOnceITCase.java
index 33222c3..edc9ede 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/CoordinatorEventsExactlyOnceITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/CoordinatorEventsExactlyOnceITCase.java
@@ -71,6 +71,7 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -303,7 +304,7 @@ public class CoordinatorEventsExactlyOnceITCase extends TestLogger {
         private final int failAtMessage;
         private boolean failedBefore;
 
-        private final ArrayDeque<CompletableFuture<?>> recoveredTaskRunning = new ArrayDeque<>();
+        private final ArrayDeque<CountDownLatch> recoveredTaskRunning = new ArrayDeque<>();
 
         private SubtaskGateway subtaskGateway;
         private boolean workLoopRunning;
@@ -348,14 +349,16 @@ public class CoordinatorEventsExactlyOnceITCase extends TestLogger {
                         String.format("Don't recognize event '%s' from task %d.", event, subtask));
             }
 
+            // We complete all events that were enqueued. We may need to complete
+            // multiple ones here, because it can happen that after a failure no real recovery
+            // happens that results in an event being sent (and this method being called), but that
+            // immediately another failure comes, triggered by the other operator coordinator (or
+            // its task).
             synchronized (recoveredTaskRunning) {
-                // signal the previous task that its recovered task is now running
-                final CompletableFuture<?> prevTaskFuture = recoveredTaskRunning.peekLast();
-                if (prevTaskFuture != null) {
-                    prevTaskFuture.complete(null);
+                for (CountDownLatch latch : recoveredTaskRunning) {
+                    latch.countDown();
                 }
-                // add a future for this task
-                recoveredTaskRunning.addLast(new CompletableFuture<>());
+                recoveredTaskRunning.clear();
             }
 
             // first, we hand this over to the mailbox thread, so we preserve order on operations,
@@ -372,6 +375,14 @@ public class CoordinatorEventsExactlyOnceITCase extends TestLogger {
 
         @Override
         public void subtaskFailed(int subtask, @Nullable Throwable reason) {
+            // we need to create and enqueue this outside the mailbox, so that the
+            // enqueuing is strictly ordered with the completion (which also happens outside
+            // the mail box executor).
+            final CountDownLatch successorIsRunning = new CountDownLatch(1);
+            synchronized (recoveredTaskRunning) {
+                recoveredTaskRunning.addLast(successorIsRunning);
+            }
+
             // simulate a heavy thread race here: the mailbox has a last enqueued action before the
             // cancellation is processed. But through a race, the mailbox freezes for a while and in
             // that time, the task already went through a recovery cycle. By the time the mailbox
@@ -380,20 +391,9 @@ public class CoordinatorEventsExactlyOnceITCase extends TestLogger {
             // coordinator before unfreezing the mailbox
             runInMailbox(
                     () -> {
-                        final CompletableFuture<?> thisTasksFuture;
-                        synchronized (recoveredTaskRunning) {
-                            thisTasksFuture = recoveredTaskRunning.peekFirst();
-                        }
-
-                        if (thisTasksFuture != null) {
-                            try {
-                                thisTasksFuture.get();
-                            } catch (Exception ignored) {
-                            }
-
-                            synchronized (recoveredTaskRunning) {
-                                recoveredTaskRunning.removeFirst();
-                            }
+                        try {
+                            successorIsRunning.await();
+                        } catch (Exception ignored) {
                         }
 
                         executeSingleAction();