Skip to content

Commit c98a1ca

Browse files
committed
test: remove two sources of flakiness in the integration suite
Two failures in run 32109693272, neither related to region-durable. testWfSyncExecutionSignalAsync ended "Workflow ended in unexpected state: FAILED". It, testWfExecutionSignalAsync and testWfDurableExecutionSignalAsync all did: Thread.sleep(20); // "Wait for initial execution" taskClient.signalAsync(workflowId, COMPLETED, ...); 20ms is not a wait, it is a coin flip — signalling before the task exists. All three now poll until the workflow is RUNNING with a non-terminal task (TestUtil.waitUntilSignalable). Only one of the three lost the toss this run; fixing just that one would leave the other two armed. testSearchTasks failed with SocketTimeoutException from taskClient.search. It slept a fixed 5s then made a single call, which raced asynchronous search indexing and had no tolerance for a slow response. Now retries until the call answers (TestUtil.retryUntil). Also: waitForWorkflowStatus threw "ended in unexpected state: FAILED" and discarded the reason, which is why the cause of the above was not visible in CI at all. It now reports reasonForIncompletion and per-task status/reason, and the timeout path reports task states too. Caveat: the signal race is the likely cause of the FAILED workflow but is not proven — the reason was never logged. The added diagnostics mean a recurrence will say so outright.
1 parent ffa272f commit c98a1ca

2 files changed

Lines changed: 92 additions & 13 deletions

File tree

‎tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -484,8 +484,9 @@ void testWfExecutionSignalAsync() throws Exception {
484484
assertNotNull(workflow);
485485
String workflowId = workflow.getWorkflowId();
486486

487-
// Wait for initial execution
488-
Thread.sleep(20);
487+
// Wait until the workflow actually has a task to signal. A fixed 20ms sleep raced the
488+
// engine: signalling before the task existed left the workflow FAILED.
489+
TestUtil.waitUntilSignalable(workflowClient, workflowId, 30_000, 100);
489490

490491
// Signal with async execution
491492
taskClient.signalAsync(workflowId, Task.Status.COMPLETED, Map.of("result", "test"));
@@ -509,8 +510,9 @@ void testWfSyncExecutionSignalAsync() throws Exception {
509510
assertNotNull(workflow);
510511
String workflowId = workflow.getTargetWorkflowId();
511512

512-
// Wait for initial execution
513-
Thread.sleep(20);
513+
// Wait until the workflow actually has a task to signal. A fixed 20ms sleep raced the
514+
// engine: signalling before the task existed left the workflow FAILED.
515+
TestUtil.waitUntilSignalable(workflowClient, workflowId, 30_000, 100);
514516

515517
// Signal with async execution
516518
taskClient.signalAsync(workflowId, Task.Status.COMPLETED, Map.of("result", "test"));
@@ -534,8 +536,9 @@ void testWfDurableExecutionSignalAsync() throws Exception {
534536
assertNotNull(workflow);
535537
String workflowId = workflow.getTargetWorkflowId();
536538

537-
// Wait for initial execution
538-
Thread.sleep(20);
539+
// Wait until the workflow actually has a task to signal. A fixed 20ms sleep raced the
540+
// engine: signalling before the task existed left the workflow FAILED.
541+
TestUtil.waitUntilSignalable(workflowClient, workflowId, 30_000, 100);
539542

540543
// Signal with async execution
541544
taskClient.signalAsync(workflowId, Task.Status.COMPLETED, Map.of("result", "test"));
@@ -742,17 +745,19 @@ void testRequeuePendingTasksByTaskType() {
742745
// ==================== Search Tests ====================
743746

744747
@Test
745-
void testSearchTasks() {
748+
void testSearchTasks() throws Exception {
746749
StartWorkflowRequest request = new StartWorkflowRequest();
747750
request.setName(workflowName);
748751
request.setVersion(1);
749752
request.setInput(new HashMap<>());
750753
String workflowId = workflowClient.startWorkflow(request);
751754

752755
try {
753-
Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS);
754-
755-
var result = taskClient.search("workflowId='" + workflowId + "'");
756+
// Search is asynchronously indexed and occasionally slow to answer, so a single shot
757+
// after a fixed sleep was doubly fragile — it raced indexing and died on a transient
758+
// SocketTimeoutException. Retry until it responds.
759+
var result = TestUtil.retryUntil(
760+
() -> taskClient.search("workflowId='" + workflowId + "'"), 30_000, 2_000);
756761
assertNotNull(result);
757762
} finally {
758763
workflowClient.terminateWorkflow(workflowId, "cleanup");

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

Lines changed: 77 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,7 @@ public static Workflow waitForWorkflowStatus(OrkesWorkflowClient workflowClient,
114114
// Check if workflow failed or terminated
115115
if (workflowDetails.getStatus() == Workflow.WorkflowStatus.FAILED ||
116116
workflowDetails.getStatus() == Workflow.WorkflowStatus.TERMINATED) {
117-
throw new RuntimeException("Workflow ended in unexpected state: " + workflowDetails.getStatus());
117+
throw new RuntimeException(describeFailure(workflowDetails));
118118
}
119119

120120
Thread.sleep(pollIntervalMs);
@@ -123,7 +123,81 @@ public static Workflow waitForWorkflowStatus(OrkesWorkflowClient workflowClient,
123123
// Timeout
124124
Workflow finalState = workflowClient.getWorkflow(workflowId, true);
125125
throw new TimeoutException(
126-
String.format("Workflow %s did not reach status %s within %dms. Current status: %s",
127-
workflowId, expectedStatus, maxWaitTimeMs, finalState.getStatus()));
126+
String.format("Workflow %s did not reach status %s within %dms. Current status: %s, tasks: %s",
127+
workflowId, expectedStatus, maxWaitTimeMs, finalState.getStatus(), taskSummary(finalState)));
128+
}
129+
130+
/**
131+
* Signalling a workflow immediately after start is a race — the task being signalled may not
132+
* exist yet. Waits until the workflow is RUNNING with at least one non-terminal task.
133+
*/
134+
public static void waitUntilSignalable(OrkesWorkflowClient workflowClient,
135+
String workflowId,
136+
long maxWaitTimeMs,
137+
long pollIntervalMs) throws Exception {
138+
long endTime = System.currentTimeMillis() + maxWaitTimeMs;
139+
140+
while (System.currentTimeMillis() < endTime) {
141+
Workflow workflow = workflowClient.getWorkflow(workflowId, true);
142+
143+
if (workflow.getStatus() == Workflow.WorkflowStatus.FAILED
144+
|| workflow.getStatus() == Workflow.WorkflowStatus.TERMINATED) {
145+
throw new RuntimeException(describeFailure(workflow));
146+
}
147+
boolean hasPendingTask = workflow.getTasks() != null
148+
&& workflow.getTasks().stream().anyMatch(t -> !t.getStatus().isTerminal());
149+
if (workflow.getStatus() == Workflow.WorkflowStatus.RUNNING && hasPendingTask) {
150+
return;
151+
}
152+
Thread.sleep(pollIntervalMs);
153+
}
154+
155+
Workflow finalState = workflowClient.getWorkflow(workflowId, true);
156+
throw new TimeoutException(String.format(
157+
"Workflow %s was not signalable within %dms. Status: %s, tasks: %s",
158+
workflowId, maxWaitTimeMs, finalState.getStatus(), taskSummary(finalState)));
159+
}
160+
161+
/** Includes the reason and task states — without these a failure is undiagnosable from CI logs. */
162+
private static String describeFailure(Workflow workflow) {
163+
return String.format("Workflow %s ended in unexpected state: %s. Reason: %s. Tasks: %s",
164+
workflow.getWorkflowId(), workflow.getStatus(),
165+
workflow.getReasonForIncompletion() == null ? "(none given)" : workflow.getReasonForIncompletion(),
166+
taskSummary(workflow));
167+
}
168+
169+
private static String taskSummary(Workflow workflow) {
170+
if (workflow.getTasks() == null) {
171+
return "[]";
172+
}
173+
StringBuilder sb = new StringBuilder("[");
174+
workflow.getTasks().forEach(t -> sb.append(t.getReferenceTaskName()).append(':')
175+
.append(t.getStatus())
176+
.append(t.getReasonForIncompletion() == null ? "" : "(" + t.getReasonForIncompletion() + ")")
177+
.append(' '));
178+
return sb.append(']').toString();
179+
}
180+
181+
/**
182+
* Retries {@code call} until it returns without throwing. Intended for endpoints that are
183+
* eventually consistent or intermittently slow, where a single attempt after a fixed sleep
184+
* turns ordinary latency into a test failure.
185+
*/
186+
public static <T> T retryUntil(java.util.concurrent.Callable<T> call,
187+
long maxWaitTimeMs,
188+
long pollIntervalMs) throws Exception {
189+
long endTime = System.currentTimeMillis() + maxWaitTimeMs;
190+
Exception last = null;
191+
192+
while (System.currentTimeMillis() < endTime) {
193+
try {
194+
return call.call();
195+
} catch (Exception e) {
196+
last = e;
197+
Thread.sleep(pollIntervalMs);
198+
}
199+
}
200+
throw new TimeoutException("Call did not succeed within " + maxWaitTimeMs + "ms"
201+
+ (last == null ? "" : ", last error: " + last));
128202
}
129203
}

0 commit comments

Comments
 (0)