[flink] branch release-1.14 updated: [FLINK-15550][runtime] Debug logging for TaskTest

[dw...@apache.org]

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

dwysakowicz pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.14 by this push:
     new 706d1a5  [FLINK-15550][runtime] Debug logging for TaskTest
706d1a5 is described below

commit 706d1a548962c3f4d6eda5118f439a0f3cb84d22
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Fri Mar 4 17:04:57 2022 +0100

    [FLINK-15550][runtime] Debug logging for TaskTest
    
    (cherry picked from commit 16e9b7325881b45741f3591f73a7b7bcfd39322b)
---
 .../apache/flink/runtime/taskmanager/TaskTest.java   | 20 +++++++++++++++++++-
 1 file changed, 19 insertions(+), 1 deletion(-)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
index 53b1e57..a193f4e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
@@ -59,6 +59,8 @@ import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
 
@@ -112,6 +114,11 @@ public class TaskTest extends TestLogger {
     public void setup() {
         awaitLatch = new OneShotLatch();
         triggerLatch = new OneShotLatch();
+        // Logging is only for debugging FLINK-15550.
+        log.info(
+                "Preparing trigger latch, {}, isTriggered = {}",
+                triggerLatch,
+                triggerLatch.isTriggered());
 
         shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build();
         wasCleanedUp = false;
@@ -633,6 +640,7 @@ public class TaskTest extends TestLogger {
         awaitLatch.await();
 
         task.failExternally(new Exception("external"));
+        assertFalse(triggerLatch.isTriggered());
         assertEquals(ExecutionState.FAILED, task.getExecutionState());
 
         // Either we cause the CancelTaskException or the TaskCanceler
@@ -1437,13 +1445,23 @@ public class TaskTest extends TestLogger {
 
     /** {@link AbstractInvokable} which throws {@link CancelTaskException} on invoke. */
     public static final class InvokableWithCancelTaskExceptionInInvoke extends AbstractInvokable {
+        static final Logger LOG = LoggerFactory.getLogger(InvokableWithExceptionOnTrigger.class);
+
         public InvokableWithCancelTaskExceptionInInvoke(Environment environment) {
             super(environment);
         }
 
         @Override
         public void invoke() {
-            awaitTriggerLatch();
+            // Logging and try-catch block are only for debugging FLINK-15550.
+            LOG.info("Await for {}, isTriggered = {}", triggerLatch, triggerLatch.isTriggered());
+            try {
+                awaitTriggerLatch();
+            } catch (Throwable ex) {
+                LOG.error("Fail on awaiting trigger latch", ex);
+
+                throw ex;
+            }
 
             throw new CancelTaskException();
         }