fix(governance-workflows): retry the synchronous task-complete on a deadlock (#32530)

Resolving a workflow user task calls Flowable's taskService.complete() inline
on the request thread. Under concurrent DAR-workflow writes (parallel resolves
plus the async executor advancing other instances on the shared process
definition), that command can lose an InnoDB deadlock race (MySQL errno 1213)
while flushing its ACT_RU_* deletes — surfacing to the client as a spurious 409
on the transition. READ_COMMITTED already removes the gap-lock class; this is a
record/FK-lock-ordering cycle that isolation level cannot prevent.

Flowable's async executor already retries these deadlocks automatically; the
synchronous resolve path was the one caller that did not. Wrap the
taskService.complete() command in the existing DeadlockRetry — the same helper
already guarding runtimeService.startProcessInstanceById(). The retry scope is
the single self-contained Flowable command: when InnoDB rolls the transaction
back it has already released every lock, so the replay runs in a fresh
transaction with nothing held while it backs off. DeadlockRetry only retries a
genuine deadlock (errno 1213/1205, SQLState 40001/40P01); any other failure,
including a not-yet-committed task, propagates on the first attempt unchanged.

Adds a DeadlockRetry.run(Runnable) overload for void commands and a unit test
covering success, deadlock-then-success, non-deadlock propagation, attempt
bounding, and the deadlock predicate.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: yan-3005

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowHandler.java

@@ -1002,12 +1002,12 @@ private boolean resolveTaskWithFreshReads(UUID customTaskId, Map<String, Object>
                         "[WorkflowTask] Completing with variables: taskId='{}' vars={}",
                         task.getId(),
                         variablesValue);
-                    taskService.complete(task.getId(), variablesValue);
+                    DeadlockRetry.run(() -> taskService.complete(task.getId(), variablesValue));
                   },
                   () -> {
                     LOG.debug(
                         "[WorkflowTask] Completing without variables: taskId='{}'", task.getId());
-                    taskService.complete(task.getId());
+                    DeadlockRetry.run(() -> taskService.complete(task.getId()));
                   });
           LOG.debug("[WorkflowTask] SUCCESS: Task '{}' resolved", customTaskId);
         }
@@ -1247,7 +1247,7 @@ private boolean handleMultiApproval(
       Object rejectValue =
           resolveMultiApprovalResult(task.getProcessDefinitionId(), nodeName, false, Boolean.FALSE);
       variables.put(resultVariable, rejectValue);
-      taskService.complete(task.getId(), variables);
+      DeadlockRetry.run(() -> taskService.complete(task.getId(), variables));
       return true;
     }
 
@@ -1260,7 +1260,7 @@ private boolean handleMultiApproval(
       Object approveValue =
           resolveMultiApprovalResult(task.getProcessDefinitionId(), nodeName, true, Boolean.TRUE);
       variables.put(resultVariable, approveValue);
-      taskService.complete(task.getId(), variables);
+      DeadlockRetry.run(() -> taskService.complete(task.getId(), variables));
       return true;
     }
 

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DeadlockRetry.java

@@ -19,19 +19,22 @@
 import lombok.extern.slf4j.Slf4j;
 
 /**
- * Retry wrapper for JDBI {@code @Transaction}-annotated methods that can lose a deadlock race on
- * hot rows.
+ * Retry wrapper for a self-contained unit of work that can lose a deadlock race on hot rows.
  *
- * <p>The retry scope is the full transaction: when JDBI rolls the transaction back on a deadlock,
- * we re-invoke the enclosing method so the entire unit of work replays in a fresh transaction. Do
- * not push this down into {@code CollectionDAO} — retrying one DAO statement outside its original
- * transaction context would leave earlier writes in that txn lost.
+ * <p>The retry scope is the whole enclosing unit of work — a JDBI {@code @Transaction}-annotated
+ * method, or a single self-contained Flowable command (e.g. {@code taskService.complete(...)},
+ * {@code runtimeService.startProcessInstanceById(...)}). When the database rolls the transaction
+ * back on a deadlock it has already released every lock it held, so re-invoking the enclosing
+ * operation replays it atomically in a fresh transaction. Do not push this down into
+ * {@code CollectionDAO} — retrying one DAO statement outside its original transaction context
+ * would leave earlier writes in that txn lost.
  *
  * <p>Backoff: retries are synchronous when invoked via {@link Retry#executeSupplier(Supplier)} —
  * the calling thread waits between attempts according to the configured interval. This matches
  * the existing retry pattern in {@code SearchRetryUtil} so operators see consistent behaviour
  * across subsystems. Exponential base 50 ms × 2^(attempt-1) with 50% jitter — attempt 1 ≈ 25-75
- * ms, attempt 2 ≈ 50-150 ms, attempt 3 ≈ 100-300 ms.
+ * ms, attempt 2 ≈ 50-150 ms, attempt 3 ≈ 100-300 ms. The wait is bounded and happens after the
+ * transaction has been rolled back, so no database lock is held while the thread backs off.
  */
 @Slf4j
 public final class DeadlockRetry {
@@ -57,13 +60,23 @@ public final class DeadlockRetry {
 
   private DeadlockRetry() {}
 
-  /** Execute {@code operation} with deadlock retry. {@code operation} must open its own JDBI
-   * transaction (typically via {@code @Transaction} on the method it delegates to) so each retry
-   * runs in a fresh, atomic unit of work. */
+  /** Execute {@code operation} with deadlock retry. {@code operation} must open its own
+   * transaction (a JDBI {@code @Transaction} method, or a self-contained Flowable command that
+   * commits on its own) so each retry runs in a fresh, atomic unit of work. */
   public static <T> T execute(Supplier<T> operation) {
     return RETRY.executeSupplier(operation);
   }
 
+  /** Run a void {@code operation} with deadlock retry — the {@link Runnable} equivalent of
+   * {@link #execute(Supplier)} for a self-contained command with no return value. */
+  public static void run(Runnable operation) {
+    RETRY.executeSupplier(
+        () -> {
+          operation.run();
+          return null;
+        });
+  }
+
   /** {@code true} if {@code throwable} (or any cause in its chain) is a MySQL/Postgres deadlock or
    * lock-wait timeout that is safe to retry as a fresh transaction. */
   public static boolean isDeadlock(Throwable throwable) {

openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/DeadlockRetryTest.java

@@ -0,0 +1,128 @@
+/*
+ *  Copyright 2024 Collate
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ */
+
+package org.openmetadata.service.jdbi3;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.sql.SQLException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Behavioural tests for {@link DeadlockRetry}. The exception shape mirrors production: a runtime
+ * exception (Flowable / MyBatis {@code PersistenceException}) wrapping a {@link SQLException} whose
+ * cause chain carries the MySQL deadlock (errno 1213). A bare checked {@code SQLException} cannot
+ * escape a {@code Supplier}/{@code Runnable}, so it would never exercise the real path.
+ */
+class DeadlockRetryTest {
+
+  private static final int MAX_ATTEMPTS = 4;
+
+  private static RuntimeException deadlock() {
+    SQLException sql =
+        new SQLException(
+            "Deadlock found when trying to get lock; try restarting transaction", "40001", 1213);
+    return new RuntimeException("### Error updating database", sql);
+  }
+
+  @Test
+  void executeReturnsWithoutRetryOnSuccess() {
+    AtomicInteger calls = new AtomicInteger();
+    String result =
+        DeadlockRetry.execute(
+            () -> {
+              calls.incrementAndGet();
+              return "ok";
+            });
+    assertEquals("ok", result);
+    assertEquals(1, calls.get(), "no retry on the happy path");
+  }
+
+  @Test
+  void executeRetriesDeadlockThenSucceeds() {
+    AtomicInteger calls = new AtomicInteger();
+    String result =
+        DeadlockRetry.execute(
+            () -> {
+              if (calls.incrementAndGet() < 3) {
+                throw deadlock();
+              }
+              return "ok";
+            });
+    assertEquals("ok", result);
+    assertEquals(3, calls.get(), "replays until the deadlock clears");
+  }
+
+  @Test
+  void executeDoesNotRetryNonDeadlock() {
+    AtomicInteger calls = new AtomicInteger();
+    IllegalStateException boom = new IllegalStateException("not a deadlock");
+    IllegalStateException thrown =
+        assertThrows(
+            IllegalStateException.class,
+            () ->
+                DeadlockRetry.execute(
+                    () -> {
+                      calls.incrementAndGet();
+                      throw boom;
+                    }));
+    assertSame(boom, thrown, "non-deadlock errors propagate unchanged");
+    assertEquals(1, calls.get(), "no retry for a non-deadlock error");
+  }
+
+  @Test
+  void executeStopsAfterMaxAttemptsWhenDeadlockPersists() {
+    AtomicInteger calls = new AtomicInteger();
+    assertThrows(
+        RuntimeException.class,
+        () ->
+            DeadlockRetry.execute(
+                () -> {
+                  calls.incrementAndGet();
+                  throw deadlock();
+                }));
+    assertEquals(MAX_ATTEMPTS, calls.get(), "bounded at the configured max attempts");
+  }
+
+  @Test
+  void runReplaysVoidCommandOnDeadlock() {
+    AtomicInteger calls = new AtomicInteger();
+    DeadlockRetry.run(
+        () -> {
+          if (calls.incrementAndGet() < 2) {
+            throw deadlock();
+          }
+        });
+    assertEquals(2, calls.get(), "void command replays once then commits");
+  }
+
+  @Test
+  void isDeadlockRecognisesRetryableCodesAndMessage() {
+    assertTrue(DeadlockRetry.isDeadlock(deadlock()), "errno 1213 in the cause chain");
+    assertTrue(
+        DeadlockRetry.isDeadlock(new SQLException("lock wait timeout", "HY000", 1205)),
+        "MySQL lock-wait timeout");
+    assertTrue(
+        DeadlockRetry.isDeadlock(new SQLException("deadlock detected", "40P01")),
+        "Postgres deadlock SQLState");
+    assertTrue(
+        DeadlockRetry.isDeadlock(
+            new RuntimeException("Deadlock found when trying to get lock; try restarting")),
+        "message match without a SQLException");
+    assertFalse(DeadlockRetry.isDeadlock(new IllegalStateException("unrelated")), "not a deadlock");
+    assertFalse(DeadlockRetry.isDeadlock(null), "null is not a deadlock");
+  }
+}