address feedback from self review

Author: chrishagglund-ship-it

.github/workflows/ci.yml

@@ -149,13 +149,7 @@ jobs:
         uses: mikepenz/action-junit-report@v6
         with:
           report_paths: '**/build/test-results/test/TEST-*.xml'
-          # Distinct per suite: the action defaults to 'JUnit Test Report' for every caller, so
-          # this report and the ones published by integration-tests.yml all landed on a single
-          # check run and overwrote each other.
-          check_name: Unit Test Report
-
+          
       - name: Check Tests Status
         if: steps.tests.outcome == 'failure'
-        run: |
-          echo "::error::Tests failed. See the 'Run Tests' step above for the Gradle/test output, and the 'Publish Test Report' step's JUnit summary for which test(s) failed."
-          exit 1
+        run: exit 1

conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java

@@ -77,17 +77,9 @@ public class WorkflowExecutor {
     private final TypeReference<List<TaskDef>> listOfTaskDefs = new TypeReference<>() {
     };
 
-    /** Zero, i.e. never give up. See {@link #setMonitorFailureGiveUpMillis(long)}. */
-    private static final long DEFAULT_MONITOR_FAILURE_GIVE_UP_MILLIS = 0;
-
     private final Map<String, CompletableFuture<Workflow>> runningWorkflowFutures =
             new ConcurrentHashMap<>();
 
-    /** When the current run of consecutive polling failures started, per workflow id. */
-    private final Map<String, Long> monitorFailingSince = new ConcurrentHashMap<>();
-
-    private volatile long monitorFailureGiveUpMillis = DEFAULT_MONITOR_FAILURE_GIVE_UP_MILLIS;
-
     private final ObjectMapper objectMapper = new ObjectMapperProvider().getObjectMapper();
 
     private final TaskClient taskClient;
@@ -184,18 +176,19 @@ private void initMonitor() {
                         CompletableFuture<Workflow> future = entry.getValue();
                         try {
                             Workflow workflow = workflowClient.getWorkflow(workflowId, true);
-                            monitorFailingSince.remove(workflowId);
                             if (workflow.getStatus().isTerminal()) {
                                 future.complete(workflow);
                                 runningWorkflowFutures.remove(workflowId);
                             }
                         } catch (Exception e) {
                             // scheduleAtFixedRate silently kills all future ticks on any uncaught
-                            // exception, so one transient error here would otherwise stop completion
-                            // tracking for every workflow, forever. Catch, but do not retry forever:
-                            // a workflow id that never becomes resolvable would spin at the polling
-                            // interval indefinitely while its caller blocks with no signal.
-                            handleMonitorFailure(workflowId, future, e);
+                            // exception, so this must not escape: one failed poll would otherwise
+                            // stop completion tracking for every workflow, forever. Stop tracking
+                            // this one and let its caller see the failure.
+                            LOGGER.error("Error polling workflow {} for completion; completing its "
+                                    + "future exceptionally", workflowId, e);
+                            runningWorkflowFutures.remove(workflowId);
+                            future.completeExceptionally(e);
                         }
                     }
                 },
@@ -204,55 +197,6 @@ private void initMonitor() {
                 TimeUnit.MILLISECONDS);
     }
 
-    private void handleMonitorFailure(String workflowId, CompletableFuture<Workflow> future, Exception e) {
-        long now = System.currentTimeMillis();
-        Long failingSince = monitorFailingSince.putIfAbsent(workflowId, now);
-
-        if (failingSince == null) {
-            LOGGER.warn("Error polling workflow {} for completion; will retry on the next tick",
-                    workflowId, e);
-            return;
-        }
-
-        long giveUpMillis = monitorFailureGiveUpMillis;
-        long failingForMillis = now - failingSince;
-        if (giveUpMillis <= 0 || failingForMillis < giveUpMillis) {
-            // Already warned once for this run of failures. Staying at DEBUG keeps a persistently
-            // unresolvable workflow from emitting a stack trace on every tick.
-            LOGGER.debug("Still failing to poll workflow {} for completion ({} ms so far)",
-                    workflowId, failingForMillis, e);
-            return;
-        }
-
-        LOGGER.error("Giving up polling workflow {} for completion after {} ms of consecutive "
-                + "failures; completing its future exceptionally", workflowId, failingForMillis, e);
-        monitorFailingSince.remove(workflowId);
-        runningWorkflowFutures.remove(workflowId);
-        future.completeExceptionally(e);
-    }
-
-    /**
-     * How long the completion monitor keeps retrying a workflow whose status cannot be fetched
-     * before giving up on it.
-     *
-     * @param monitorFailureGiveUpMillis zero or negative (the default) to never give up: the
-     *     monitor retries such a workflow for as long as this executor lives, so a server outage
-     *     longer than any fixed budget — a rolling restart, a failover — does not strand futures
-     *     that would otherwise have completed once the server came back. A positive value bounds
-     *     that: once a workflow has failed to poll continuously for this long, the monitor stops
-     *     tracking it and completes its future exceptionally, so a caller blocked in
-     *     {@code executeWorkflow(...).get()} sees an {@link java.util.concurrent.ExecutionException}
-     *     rather than blocking indefinitely on a workflow id that will never resolve.
-     */
-    public void setMonitorFailureGiveUpMillis(long monitorFailureGiveUpMillis) {
-        this.monitorFailureGiveUpMillis = monitorFailureGiveUpMillis;
-    }
-
-    /** @see #setMonitorFailureGiveUpMillis(long) */
-    public long getMonitorFailureGiveUpMillis() {
-        return monitorFailureGiveUpMillis;
-    }
-
     public void initWorkers(String... packagesToScan) {
         annotatedWorkerExecutor.initWorkers(packagesToScan);
     }

conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java

@@ -16,7 +16,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
@@ -58,59 +57,30 @@ private WorkflowExecutor executorFor(WorkflowClient workflowClient) {
     }
 
     @Test
-    @DisplayName("the monitor should keep polling after a transient getWorkflow failure")
-    void monitorSurvivesTransientPollFailure() throws Exception {
-        Workflow completed = new Workflow();
-        completed.setStatus(Workflow.WorkflowStatus.COMPLETED);
-
-        WorkflowClient workflowClient = mock(WorkflowClient.class);
-        when(workflowClient.startWorkflow(any(StartWorkflowRequest.class))).thenReturn(WORKFLOW_ID);
-        when(workflowClient.getWorkflow(anyString(), anyBoolean()))
-                .thenThrow(new RuntimeException("transient failure"))
-                .thenReturn(completed);
-
-        WorkflowExecutor executor = executorFor(workflowClient);
-        try {
-            CompletableFuture<Workflow> future = executor.executeWorkflow("wf", 1, Map.of());
-
-            Workflow result = future.get(5, TimeUnit.SECONDS);
-
-            assertEquals(Workflow.WorkflowStatus.COMPLETED, result.getStatus());
-        } finally {
-            executor.shutdown();
-        }
-    }
-
-    @Test
-    @DisplayName("the monitor should give up and fail the future once the failure budget is spent")
-    void monitorGivesUpOnPersistentPollFailure() {
+    @DisplayName("a failed poll should complete that workflow's future exceptionally")
+    void monitorFailsTheFutureOnPollFailure() {
         WorkflowClient workflowClient = mock(WorkflowClient.class);
         when(workflowClient.startWorkflow(any(StartWorkflowRequest.class))).thenReturn(WORKFLOW_ID);
         when(workflowClient.getWorkflow(anyString(), anyBoolean()))
-                .thenThrow(new RuntimeException("permanent failure"));
+                .thenThrow(new RuntimeException("poll failure"));
 
         WorkflowExecutor executor = executorFor(workflowClient);
         try {
-            // 1ms budget: the monitor ticks every 100ms, so the second consecutive failure is
-            // already past it. Keeps the test off the wall clock while still exercising a real
-            // (positive, opt-in) budget rather than the never-give-up default.
-            executor.setMonitorFailureGiveUpMillis(1);
-
             CompletableFuture<Workflow> future = executor.executeWorkflow("wf", 1, Map.of());
 
             ExecutionException thrown = assertThrows(
                     ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS));
 
             assertInstanceOf(RuntimeException.class, thrown.getCause());
-            assertEquals("permanent failure", thrown.getCause().getMessage());
+            assertEquals("poll failure", thrown.getCause().getMessage());
         } finally {
             executor.shutdown();
         }
     }
 
     @Test
-    @DisplayName("by default the monitor should never give up, and one bad workflow should not stall the rest")
-    void monitorDoesNotGiveUpByDefault() throws Exception {
+    @DisplayName("one workflow's poll failure should not stop the monitor tracking the rest")
+    void monitorKeepsTrackingOtherWorkflowsAfterAFailure() throws Exception {
         Workflow completed = new Workflow();
         completed.setStatus(Workflow.WorkflowStatus.COMPLETED);
 
@@ -119,24 +89,20 @@ void monitorDoesNotGiveUpByDefault() throws Exception {
                 .thenReturn(WORKFLOW_ID)
                 .thenReturn(OTHER_WORKFLOW_ID);
         when(workflowClient.getWorkflow(eq(WORKFLOW_ID), anyBoolean()))
-                .thenThrow(new RuntimeException("permanent failure"));
+                .thenThrow(new RuntimeException("poll failure"));
         when(workflowClient.getWorkflow(eq(OTHER_WORKFLOW_ID), anyBoolean()))
                 .thenReturn(completed);
 
         WorkflowExecutor executor = executorFor(workflowClient);
         try {
-            assertEquals(0, executor.getMonitorFailureGiveUpMillis(), "give-up should be off by default");
-
             CompletableFuture<Workflow> failing = executor.executeWorkflow("wf", 1, Map.of());
-            CompletableFuture<Workflow> healthy = executor.executeWorkflow("wf", 1, Map.of());
+            assertThrows(ExecutionException.class, () -> failing.get(5, TimeUnit.SECONDS));
 
-            // The unresolvable workflow must not take the monitor down with it: a sibling
-            // registered on the same tick loop still completes.
+            // Registered only after the failure has already happened, so it can complete at all
+            // only if the tick loop survived it -- scheduleAtFixedRate would have cancelled every
+            // future tick had the exception been allowed to escape.
+            CompletableFuture<Workflow> healthy = executor.executeWorkflow("wf", 1, Map.of());
             assertEquals(Workflow.WorkflowStatus.COMPLETED, healthy.get(5, TimeUnit.SECONDS).getStatus());
-
-            // ...and the unresolvable one stays pending rather than being failed, which is the
-            // pre-bounded-retry behavior callers depend on across a server restart.
-            assertThrows(TimeoutException.class, () -> failing.get(1, TimeUnit.SECONDS));
         } finally {
             executor.shutdown();
         }

scripts/docker-compose-oss.yaml

@@ -2,8 +2,10 @@
 # Shared by scripts/run-integration-oss.sh and the integration-tests-oss job in
 # .github/workflows/integration-tests.yml.
 #
-# OSS_CONDUCTOR_VERSION defaults to `latest` for local runs; CI pins it via the
-# E2E_TEST_OSS_CONDUCTOR_VERSION org variable (or a workflow_dispatch input).
+# OSS_CONDUCTOR_VERSION defaults to `latest` for local runs. CI resolves it from the
+# E2E_TEST_OSS_CONDUCTOR_VERSION org variable (or a workflow_dispatch input); that variable is
+# currently set to `latest` too, so CI tracks whatever `latest` resolves to at run time rather
+# than a fixed version. Set the org variable to a real tag if the job needs to be deterministic.
 
 services:
   conductor-server:

tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java

@@ -38,7 +38,7 @@ void testEventHandler() {
         } catch (ConductorClientException e) {
             // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server
             // we're running against actually reports it.
-            TestUtil.assertNotFoundOrRethrow(e, "not found");
+            TestUtil.tolerateNotFound(e, "EventHandler with name");
         }
         EventHandler eventHandler = getEventHandler();
         eventClient.registerEventHandler(eventHandler);

tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java

@@ -41,7 +41,7 @@ void taskDefinition() {
         } catch (ConductorClientException e) {
             // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server
             // we're running against actually reports it.
-            TestUtil.assertNotFoundOrRethrow(e, "No such task definition");
+            TestUtil.tolerateNotFound(e, "No such task definition");
         }
         TaskDef taskDef = Commons.getTaskDef();
         metadataClient.registerTaskDefs(List.of(taskDef));
@@ -57,7 +57,7 @@ void workflow() {
         } catch (ConductorClientException e) {
             // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server
             // we're running against actually reports it.
-            TestUtil.assertNotFoundOrRethrow(e, "No such workflow definition");
+            TestUtil.tolerateNotFound(e, "No such workflow definition");
         }
         metadataClient.registerTaskDefs(List.of(Commons.getTaskDef()));
         WorkflowDef workflowDef = WorkflowUtil.getWorkflowDef();

tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java

@@ -174,7 +174,7 @@ private static boolean isTerminalFailure(Workflow workflow) {
      * OSS with it deliberately unset, and OSS returning a correct 404 for one of these endpoints
      * should not start failing the suite.
      */
-    public static void assertNotFoundOrRethrow(ConductorClientException e, String ossMessageSubstring) {
+    public static void tolerateNotFound(ConductorClientException e, String ossMessageSubstring) {
         if (e.getStatus() == 404) {
             return;
         }