Skip to content

Commit 5394310

Browse files
sojanmathewsojanmathew
authored
Set default workflow task failure cause to unhandled failure (#2706)
Set default workflow task failure cause to unhandled failure Unhandled exceptions previously lacked specific failure causes, making error categorization difficult in observability tools. Add WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE as default cause for unhandled exceptions, while preserving existing NonDeterministicException behavior. Improves error categorization.
1 parent be353bc commit 5394310

File tree

3 files changed

+138
-1
lines changed

3 files changed

+138
-1
lines changed

‎temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowTaskHandler.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -310,6 +310,10 @@ private Result failureToWFTResult(
310310
if (e instanceof NonDeterministicException) {
311311
failedRequest.setCause(
312312
WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_NON_DETERMINISTIC_ERROR);
313+
} else {
314+
// Default task failure cause to "workflow worker unhandled failure"
315+
failedRequest.setCause(
316+
WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE);
313317
}
314318
return new WorkflowTaskHandler.Result(
315319
workflowType, null, failedRequest.build(), null, null, false, null);

‎temporal-sdk/src/test/java/io/temporal/internal/replay/ReplayWorkflowTaskHandlerFailureCauseTest.java‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
package io.temporal.internal.replay;
2+
3+
import static org.junit.Assert.assertEquals;
4+
5+
import com.google.protobuf.ByteString;
6+
import io.temporal.api.enums.v1.WorkflowTaskFailedCause;
7+
import io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponse;
8+
import io.temporal.api.workflowservice.v1.RespondWorkflowTaskFailedRequest;
9+
import io.temporal.worker.NonDeterministicException;
10+
import java.lang.reflect.Method;
11+
import org.junit.Test;
12+
13+
/**
14+
* Unit tests for ReplayWorkflowTaskHandler failure cause assignment. Tests the failureToWFTResult
15+
* method to ensure correct WorkflowTaskFailedCause is set.
16+
*/
17+
public class ReplayWorkflowTaskHandlerFailureCauseTest {
18+
19+
@Test
20+
public void testUnhandledExceptionSetsCorrectFailureCause() throws Exception {
21+
// Create a simple workflow task with minimal required fields
22+
PollWorkflowTaskQueueResponse workflowTask =
23+
PollWorkflowTaskQueueResponse.newBuilder()
24+
.setTaskToken(ByteString.copyFromUtf8("test-token"))
25+
.build();
26+
27+
// Test with RuntimeException (unhandled failure)
28+
RuntimeException unhandledException = new RuntimeException("Simulated unhandled failure");
29+
RespondWorkflowTaskFailedRequest result =
30+
testFailureToWFTResult(workflowTask, unhandledException);
31+
32+
assertEquals(
33+
"Should set WORKFLOW_WORKER_UNHANDLED_FAILURE cause",
34+
WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE,
35+
result.getCause());
36+
}
37+
38+
@Test
39+
public void testNonDeterministicExceptionSetsCorrectFailureCause() throws Exception {
40+
// Create a simple workflow task with minimal required fields
41+
PollWorkflowTaskQueueResponse workflowTask =
42+
PollWorkflowTaskQueueResponse.newBuilder()
43+
.setTaskToken(ByteString.copyFromUtf8("test-token"))
44+
.build();
45+
46+
// Test with NonDeterministicException
47+
NonDeterministicException nonDetException =
48+
new NonDeterministicException("Simulated non-deterministic failure");
49+
RespondWorkflowTaskFailedRequest result = testFailureToWFTResult(workflowTask, nonDetException);
50+
51+
assertEquals(
52+
"Should set NON_DETERMINISTIC_ERROR cause",
53+
WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_NON_DETERMINISTIC_ERROR,
54+
result.getCause());
55+
}
56+
57+
@Test
58+
public void testErrorExceptionSetsCorrectFailureCause() throws Exception {
59+
// Create a simple workflow task with minimal required fields
60+
PollWorkflowTaskQueueResponse workflowTask =
61+
PollWorkflowTaskQueueResponse.newBuilder()
62+
.setTaskToken(ByteString.copyFromUtf8("test-token"))
63+
.build();
64+
65+
// Test with Error (unhandled failure)
66+
Error error = new Error("Simulated error");
67+
RespondWorkflowTaskFailedRequest result = testFailureToWFTResult(workflowTask, error);
68+
69+
assertEquals(
70+
"Should set WORKFLOW_WORKER_UNHANDLED_FAILURE cause for Error",
71+
WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE,
72+
result.getCause());
73+
}
74+
75+
/** Helper method to test the private failureToWFTResult method using reflection. */
76+
private RespondWorkflowTaskFailedRequest testFailureToWFTResult(
77+
PollWorkflowTaskQueueResponse workflowTask, Throwable failure) throws Exception {
78+
79+
// Create a minimal handler instance just for testing the failureToWFTResult method
80+
// We need to provide non-null values for required parameters
81+
com.uber.m3.tally.NoopScope metricsScope = new com.uber.m3.tally.NoopScope();
82+
io.temporal.internal.worker.SingleWorkerOptions options =
83+
io.temporal.internal.worker.SingleWorkerOptions.newBuilder()
84+
.setDataConverter(
85+
io.temporal.common.converter.DefaultDataConverter.newDefaultInstance())
86+
.setIdentity("test-worker")
87+
.setMetricsScope(metricsScope)
88+
.build();
89+
90+
io.temporal.internal.worker.WorkflowExecutorCache cache =
91+
new io.temporal.internal.worker.WorkflowExecutorCache(
92+
10, new io.temporal.internal.worker.WorkflowRunLockManager(), metricsScope);
93+
94+
// Mock the required service parameter
95+
io.temporal.serviceclient.WorkflowServiceStubs mockService =
96+
org.mockito.Mockito.mock(io.temporal.serviceclient.WorkflowServiceStubs.class);
97+
98+
ReplayWorkflowTaskHandler handler =
99+
new ReplayWorkflowTaskHandler(
100+
"test-namespace",
101+
null, // workflowFactory - can be null for this test
102+
cache,
103+
options,
104+
null, // stickyTaskQueue - can be null
105+
null, // stickyTaskQueueScheduleToStartTimeout - can be null
106+
mockService, // service - must be non-null
107+
null // localActivityDispatcher - can be null
108+
);
109+
110+
// Use reflection to access the private failureToWFTResult method
111+
Method method =
112+
ReplayWorkflowTaskHandler.class.getDeclaredMethod(
113+
"failureToWFTResult",
114+
io.temporal.api.workflowservice.v1.PollWorkflowTaskQueueResponseOrBuilder.class,
115+
Throwable.class,
116+
io.temporal.common.converter.DataConverter.class);
117+
method.setAccessible(true);
118+
io.temporal.internal.worker.WorkflowTaskHandler.Result result =
119+
(io.temporal.internal.worker.WorkflowTaskHandler.Result)
120+
method.invoke(
121+
handler,
122+
workflowTask,
123+
failure,
124+
io.temporal.common.converter.DefaultDataConverter.newDefaultInstance());
125+
126+
// Extract the RespondWorkflowTaskFailedRequest from the Result
127+
return result.getTaskFailed();
128+
}
129+
}

‎temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,11 @@ private Channel applyTailStandardInterceptors(Channel channel) {
145145
private Channel applyHeadStandardInterceptors(Channel channel) {
146146
Metadata headers = new Metadata();
147147
headers.merge(options.getHeaders());
148-
headers.put(LIBRARY_VERSION_HEADER_KEY, Version.LIBRARY_VERSION);
148+
// Don't set the client header if it wasn't parsed properly when building. The server will
149+
// fail RPCs if it's not semver.
150+
if (Version.LIBRARY_VERSION.contains(".")) {
151+
headers.put(LIBRARY_VERSION_HEADER_KEY, Version.LIBRARY_VERSION);
152+
}
149153
headers.put(SUPPORTED_SERVER_VERSIONS_HEADER_KEY, Version.SUPPORTED_SERVER_VERSIONS);
150154
headers.put(CLIENT_NAME_HEADER_KEY, Version.SDK_NAME);
151155
if (options instanceof CloudServiceStubsOptions) {

0 commit comments

Comments
 (0)