From 26cf0b511e3b4000b3bc7272b54e7b21a38e045f Mon Sep 17 00:00:00 2001 From: yeomin4242 Date: Sun, 11 Oct 2026 23:26:45 +0900 Subject: [PATCH] Add counter for unaccessed failed promise warnings --- CHANGELOG.md | 2 + .../sync/DeterministicRunnerImpl.java | 5 +++ .../java/io/temporal/worker/MetricsType.java | 9 +++++ .../sync/DeterministicRunnerTest.java | 39 +++++++++++++++++++ 4 files changed, 55 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index af4bac6c19..aabdf88e4f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,8 @@ to docs, or any other relevant information. ## [Unreleased] ### Added +- Added `temporal_workflow_unaccessed_failed_promise`, counting failed promises reported by the + existing unaccessed-promise warning during runner cleanup. - Added experimental `ChildWorkflowOptions.Builder.setVersioningOverride` and `VersioningOverride.OneTimeVersioningOverride` for explicit pinned, auto-upgrade, and one-time child workflow routing. Invalid child overrides are reported as `InvalidVersioningOverrideFailure` diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java index 0bf086f537..e74e1d9bd7 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java @@ -8,6 +8,7 @@ import io.temporal.internal.context.ContextThreadLocal; import io.temporal.internal.worker.WorkflowExecutorCache; import io.temporal.serviceclient.CheckedExceptionWrapper; +import io.temporal.worker.MetricsType; import io.temporal.workflow.Promise; import java.util.ArrayList; import java.util.HashMap; @@ -339,6 +340,10 @@ private void close(boolean fromWorkflowThread) { f.get(); throw new Error("unreachable"); } catch (RuntimeException e) { + workflowContext + .getMetricsScope() + .counter(MetricsType.WORKFLOW_UNACCESSED_FAILED_PROMISE_COUNTER) + .inc(1); log.warn( "Promise completed with exception and was never accessed. The ignored exception:", CheckedExceptionWrapper.unwrap(e)); diff --git a/temporal-sdk/src/main/java/io/temporal/worker/MetricsType.java b/temporal-sdk/src/main/java/io/temporal/worker/MetricsType.java index c9bcc49550..d1f9e8f5e2 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/MetricsType.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/MetricsType.java @@ -17,6 +17,15 @@ private MetricsType() {} public static final String WORKFLOW_FAILED_COUNTER = TEMPORAL_METRICS_PREFIX + "workflow_failed"; public static final String WORKFLOW_CONTINUE_AS_NEW_COUNTER = TEMPORAL_METRICS_PREFIX + "workflow_continue_as_new"; + + /** + * Counts failed promises reported by the unaccessed-promise warning during runner cleanup, + * including cache eviction and Worker shutdown. Replay-time increments are suppressed; separate + * runners may report the same failure. + */ + public static final String WORKFLOW_UNACCESSED_FAILED_PROMISE_COUNTER = + TEMPORAL_METRICS_PREFIX + "workflow_unaccessed_failed_promise"; + public static final String CORRUPTED_SIGNALS_COUNTER = TEMPORAL_METRICS_PREFIX + "corrupted_signals"; diff --git a/temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java index cb29a53bb6..6aef4c813a 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java @@ -4,6 +4,7 @@ import static org.junit.Assert.assertFalse; import static org.mockito.Mockito.*; +import com.uber.m3.tally.Counter; import com.uber.m3.tally.NoopScope; import com.uber.m3.tally.RootScopeBuilder; import com.uber.m3.tally.Scope; @@ -1000,4 +1001,42 @@ public void testSupplierCalledMultipleWithoutCaching() { }); d.runUntilAllBlocked(DeterministicRunner.DEFAULT_DEADLOCK_DETECTION_TIMEOUT_MS); } + + @Test + public void testUnaccessedFailedPromiseMetric() { + Counter counter = closeRunnerWithFailedPromises(false); + verify(counter, times(2)).inc(1); + } + + @Test + public void testAccessedFailedPromisesDoNotIncrementMetric() { + Counter counter = closeRunnerWithFailedPromises(true); + verifyNoInteractions(counter); + } + + private Counter closeRunnerWithFailedPromises(boolean accessFailures) { + Counter counter = mock(Counter.class); + Scope scope = mock(Scope.class); + when(scope.counter(MetricsType.WORKFLOW_UNACCESSED_FAILED_PROMISE_COUNTER)).thenReturn(counter); + SyncWorkflowContext context = spy(DummySyncWorkflowContext.newDummySyncWorkflowContext()); + doReturn(scope).when(context).getMetricsScope(); + DeterministicRunner runner = + DeterministicRunner.newRunner( + threadPool::submit, + context, + () -> { + Promise read = Workflow.newFailedPromise(new RuntimeException("read")); + Promise handled = Workflow.newFailedPromise(new RuntimeException("handled")); + if (accessFailures) { + read.getFailure(); + handled.exceptionally(failure -> null); + } + WorkflowThread.await("blocked", () -> false); + }); + runner.runUntilAllBlocked(DeterministicRunner.DEFAULT_DEADLOCK_DETECTION_TIMEOUT_MS); + assertFalse(runner.isDone()); + runner.close(); + runner.close(); + return counter; + } }