Skip to content

Commit 2ec5b62

Browse files
longquanzhenglongquanzheng
authored
Add and improve handling workflow uncompleted status return (#132)
1 parent ae0386b commit 2ec5b62

14 files changed

Lines changed: 452 additions & 20 deletions

‎README.md‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,15 +9,15 @@ See [samples](https://github.com/indeedeng/iwf-java-samples) for how to use this
99
## Gradle
1010
```gradle
1111
// https://mvnrepository.com/artifact/io.iworkflow/iwf-java-sdk
12-
implementation 'io.iworkflow:iwf-java-sdk:1.2.+'
12+
implementation 'io.iworkflow:iwf-java-sdk:1.3.+'
1313
```
1414
## Maven
1515
```
1616
<!-- https://mvnrepository.com/artifact/io.iworkflow/iwf-java-sdk -->
1717
<dependency>
1818
<groupId>io.iworkflow</groupId>
1919
<artifactId>iwf-java-sdk</artifactId>
20-
<version>1.2.+</version>
20+
<version>1.3.+</version>
2121
<type>pom</type>
2222
</dependency>
2323
@@ -88,6 +88,11 @@ Run the command `git submodule update --remote --merge` to update IDL to the lat
8888
- [x] Skip timer API for testing/operation
8989
- [x] Decider trigger type: any command combination
9090

91+
## 1.3
92+
93+
- [x] Support failing workflow with results
94+
- [x] Improve workflow uncompleted error return(canceled, failed, timeout, terminated)
95+
9196
## Future
9297

9398
- [ ] WaitForMoreResults in StateDecision

‎build.gradle‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,7 @@ signing {
150150
}
151151

152152
group = "io.iworkflow"
153-
version = "1.2.7"
153+
version = "1.3.0"
154154

155155
nexusPublishing {
156156
repositories {

‎iwf-idl‎

‎src/main/java/io/iworkflow/core/StateDecision.java‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -13,10 +13,6 @@ public abstract class StateDecision {
1313

1414
public static final StateDecision DEAD_END = ImmutableStateDecision.builder().build();
1515

16-
public static final StateDecision FORCE_FAILING_WORKFLOW = ImmutableStateDecision.builder()
17-
.nextStates(Arrays.asList(StateMovement.FORCE_FAILING_WORKFLOW_MOVEMENT))
18-
.build();
19-
2016
public static ImmutableStateDecision.Builder builder() {
2117
return ImmutableStateDecision.builder();
2218
}
@@ -27,15 +23,15 @@ public static StateDecision gracefulCompleteWorkflow(final Object output) {
2723
)).build();
2824
}
2925

30-
public static StateDecision forceCompleteWorkflow(final Object output) {
26+
public static StateDecision gracefulCompleteWorkflow() {
3127
return ImmutableStateDecision.builder().nextStates(Arrays.asList(
32-
StateMovement.forceCompleteWorkflow(output)
28+
StateMovement.gracefulCompleteWorkflow()
3329
)).build();
3430
}
3531

36-
public static StateDecision gracefulCompleteWorkflow() {
32+
public static StateDecision forceCompleteWorkflow(final Object output) {
3733
return ImmutableStateDecision.builder().nextStates(Arrays.asList(
38-
StateMovement.gracefulCompleteWorkflow()
34+
StateMovement.forceCompleteWorkflow(output)
3935
)).build();
4036
}
4137

@@ -45,6 +41,20 @@ public static StateDecision forceCompleteWorkflow() {
4541
)).build();
4642
}
4743

44+
public static StateDecision forceFailWorkflow(final Object output) {
45+
return ImmutableStateDecision.builder().nextStates(Arrays.asList(
46+
StateMovement.forceFailWorkflow(output)
47+
)).build();
48+
}
49+
50+
public static StateDecision forceFailWorkflow() {
51+
return FORCE_FAILING_WORKFLOW;
52+
}
53+
54+
public static final StateDecision FORCE_FAILING_WORKFLOW = ImmutableStateDecision.builder()
55+
.nextStates(Arrays.asList(StateMovement.FORCE_FAILING_WORKFLOW_MOVEMENT))
56+
.build();
57+
4858
public static StateDecision singleNextState(final Class<? extends WorkflowState> stateClass) {
4959
return singleNextState(stateClass.getSimpleName());
5060
}

‎src/main/java/io/iworkflow/core/StateMovement.java‎

Lines changed: 14 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -15,27 +15,34 @@ public abstract class StateMovement {
1515
private final static String GRACEFUL_COMPLETING_WORKFLOW_STATE_ID = "_SYS_GRACEFUL_COMPLETING_WORKFLOW";
1616
private final static String FORCE_COMPLETING_WORKFLOW_STATE_ID = "_SYS_FORCE_COMPLETING_WORKFLOW";
1717
private final static String FORCE_FAILING_WORKFLOW_STATE_ID = "_SYS_FORCE_FAILING_WORKFLOW";
18-
public static final StateMovement FORCE_FAILING_WORKFLOW_MOVEMENT = ImmutableStateMovement.builder().stateId(FORCE_FAILING_WORKFLOW_STATE_ID).build();
18+
19+
public static StateMovement gracefulCompleteWorkflow() {
20+
return ImmutableStateMovement.builder().stateId(GRACEFUL_COMPLETING_WORKFLOW_STATE_ID)
21+
.build();
22+
}
1923

2024
public static StateMovement gracefulCompleteWorkflow(final Object output) {
2125
return ImmutableStateMovement.builder().stateId(GRACEFUL_COMPLETING_WORKFLOW_STATE_ID)
2226
.stateInput(output)
2327
.build();
2428
}
2529

26-
public static StateMovement forceCompleteWorkflow(final Object output) {
30+
public static StateMovement forceCompleteWorkflow() {
2731
return ImmutableStateMovement.builder().stateId(FORCE_COMPLETING_WORKFLOW_STATE_ID)
28-
.stateInput(output)
2932
.build();
3033
}
3134

32-
public static StateMovement gracefulCompleteWorkflow() {
33-
return ImmutableStateMovement.builder().stateId(GRACEFUL_COMPLETING_WORKFLOW_STATE_ID)
35+
public static StateMovement forceCompleteWorkflow(final Object output) {
36+
return ImmutableStateMovement.builder().stateId(FORCE_COMPLETING_WORKFLOW_STATE_ID)
37+
.stateInput(output)
3438
.build();
3539
}
3640

37-
public static StateMovement forceCompleteWorkflow() {
38-
return ImmutableStateMovement.builder().stateId(FORCE_COMPLETING_WORKFLOW_STATE_ID)
41+
public static final StateMovement FORCE_FAILING_WORKFLOW_MOVEMENT = ImmutableStateMovement.builder().stateId(FORCE_FAILING_WORKFLOW_STATE_ID).build();
42+
43+
public static StateMovement forceFailWorkflow(final Object output) {
44+
return ImmutableStateMovement.builder().stateId(FORCE_FAILING_WORKFLOW_STATE_ID)
45+
.stateInput(output)
3946
.build();
4047
}
4148

‎src/main/java/io/iworkflow/core/UnregisteredClient.java‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import io.iworkflow.gen.models.WorkflowStartOptions;
2323
import io.iworkflow.gen.models.WorkflowStartRequest;
2424
import io.iworkflow.gen.models.WorkflowStartResponse;
25+
import io.iworkflow.gen.models.WorkflowStatus;
2526
import io.iworkflow.gen.models.WorkflowStopRequest;
2627

2728
import java.util.List;
@@ -162,19 +163,32 @@ public <T> T getSimpleWorkflowResultWithWait(
162163
throw IwfHttpException.fromFeignException(clientOptions.getObjectEncoder(), exp);
163164
}
164165

166+
if (workflowGetResponse.getWorkflowStatus() != WorkflowStatus.COMPLETED) {
167+
throwUncompletedException(workflowGetResponse);
168+
}
169+
165170
if (workflowGetResponse.getResults() == null || workflowGetResponse.getResults().size() == 0) {
166171
return null;
167172
}
168173

169174
String checkErrorMessage = "this workflow should have one or zero state output for using this API";
170175
Preconditions.checkNotNull(workflowGetResponse.getResults(), checkErrorMessage);
171176
Preconditions.checkArgument(workflowGetResponse.getResults().size() == 1, checkErrorMessage);
172-
//Preconditions.checkNotNull(workflowGetResponse.getResults().get(0).getCompletedStateOutput(), checkErrorMessage);
173177

174178
final StateCompletionOutput output = workflowGetResponse.getResults().get(0);
175179
return clientOptions.getObjectEncoder().decode(output.getCompletedStateOutput(), valueClass);
176180
}
177181

182+
private void throwUncompletedException(final WorkflowGetResponse workflowGetResponse) {
183+
throw new WorkflowUncompletedException(
184+
workflowGetResponse.getWorkflowRunId(),
185+
workflowGetResponse.getWorkflowStatus(),
186+
workflowGetResponse.getErrorType(),
187+
workflowGetResponse.getErrorMessage(),
188+
workflowGetResponse.getResults(),
189+
this.clientOptions.getObjectEncoder());
190+
}
191+
178192
public <T> T getSimpleWorkflowResultWithWait(
179193
Class<T> valueClass,
180194
final String workflowId) {
@@ -199,6 +213,10 @@ public List<StateCompletionOutput> getComplexWorkflowResultWithWait(
199213
.workflowRunId(workflowRunId)
200214
);
201215

216+
if (workflowGetResponse.getWorkflowStatus() != WorkflowStatus.COMPLETED) {
217+
throwUncompletedException(workflowGetResponse);
218+
}
219+
202220
return workflowGetResponse.getResults();
203221
} catch (FeignException.FeignClientException exp) {
204222
throw IwfHttpException.fromFeignException(clientOptions.getObjectEncoder(), exp);

‎src/main/java/io/iworkflow/core/WorkflowUncompletedException.java‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
package io.iworkflow.core;
2+
3+
import io.iworkflow.gen.models.StateCompletionOutput;
4+
import io.iworkflow.gen.models.WorkflowErrorType;
5+
import io.iworkflow.gen.models.WorkflowStatus;
6+
7+
import java.util.List;
8+
9+
public class WorkflowUncompletedException extends RuntimeException {
10+
private final String runId;
11+
private final WorkflowStatus closedStatus;
12+
private final WorkflowErrorType errorType;
13+
private final String errorMessage;
14+
private final List<StateCompletionOutput> stateResults;
15+
private final ObjectEncoder encoder;
16+
17+
public WorkflowUncompletedException(
18+
final String runId, final WorkflowStatus closedStatus, final WorkflowErrorType errorType, final String errorMessage,
19+
final List<StateCompletionOutput> stateResults, final ObjectEncoder encoder) {
20+
this.runId = runId;
21+
this.closedStatus = closedStatus;
22+
this.errorType = errorType;
23+
this.errorMessage = errorMessage;
24+
this.stateResults = stateResults;
25+
this.encoder = encoder;
26+
}
27+
28+
public String getRunId() {
29+
return runId;
30+
}
31+
32+
public WorkflowStatus getClosedStatus() {
33+
return closedStatus;
34+
}
35+
36+
// Today, this only applies to FAILED as closedStatus to differentiate different failed types
37+
public WorkflowErrorType getErrorSubType() {
38+
return errorType;
39+
}
40+
41+
public String getErrorMessage() {
42+
return errorMessage;
43+
}
44+
45+
public int getStateResultsSize() {
46+
if (stateResults == null) {
47+
return 0;
48+
}
49+
return stateResults.size();
50+
}
51+
52+
public <T> T getStateResult(final int index, Class<T> type) {
53+
final StateCompletionOutput output = stateResults.get(index);
54+
return encoder.decode(output.getCompletedStateOutput(), type);
55+
}
56+
57+
}

‎src/test/java/io/iworkflow/integ/WorkflowUncompletedTest.java‎

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
package io.iworkflow.integ;
2+
3+
import io.iworkflow.core.Client;
4+
import io.iworkflow.core.ClientOptions;
5+
import io.iworkflow.core.WorkflowUncompletedException;
6+
import io.iworkflow.gen.models.WorkflowErrorType;
7+
import io.iworkflow.gen.models.WorkflowStatus;
8+
import io.iworkflow.integ.forcefail.ForceFailWorkflow;
9+
import io.iworkflow.integ.signal.BasicSignalWorkflow;
10+
import io.iworkflow.integ.stateapifail.StateApiFailWorkflow;
11+
import io.iworkflow.integ.stateapitimeout.StateApiTimeoutFailWorkflow;
12+
import io.iworkflow.spring.TestSingletonWorkerService;
13+
import io.iworkflow.spring.controller.WorkflowRegistry;
14+
import org.junit.jupiter.api.Assertions;
15+
import org.junit.jupiter.api.BeforeEach;
16+
import org.junit.jupiter.api.Test;
17+
18+
import java.util.concurrent.ExecutionException;
19+
20+
public class WorkflowUncompletedTest {
21+
22+
@BeforeEach
23+
public void setup() throws ExecutionException, InterruptedException {
24+
TestSingletonWorkerService.startWorkerIfNotUp();
25+
}
26+
27+
@Test
28+
public void testWorkflowTimeout() throws InterruptedException {
29+
final Client client = new Client(WorkflowRegistry.registry, ClientOptions.localDefault);
30+
final String wfId = "testWorkflowTimeout" + System.currentTimeMillis() / 1000;
31+
final Integer input = 1;
32+
final String runId = client.startWorkflow(
33+
BasicSignalWorkflow.class, wfId, 1, input);
34+
35+
try {
36+
client.getSimpleWorkflowResultWithWait(Integer.class, wfId);
37+
} catch (WorkflowUncompletedException e) {
38+
Assertions.assertEquals(runId, e.getRunId());
39+
Assertions.assertEquals(WorkflowStatus.TIMEOUT, e.getClosedStatus());
40+
Assertions.assertNull(e.getErrorSubType());
41+
Assertions.assertNull(e.getErrorMessage());
42+
Assertions.assertEquals(0, e.getStateResultsSize());
43+
return;
44+
}
45+
Assertions.fail("no exception caught");
46+
}
47+
48+
@Test
49+
public void testWorkflowCanceled() throws InterruptedException {
50+
final Client client = new Client(WorkflowRegistry.registry, ClientOptions.localDefault);
51+
final String wfId = "testWorkflowTimeout" + System.currentTimeMillis() / 1000;
52+
final Integer input = 1;
53+
final String runId = client.startWorkflow(
54+
BasicSignalWorkflow.class, wfId, 10, input);
55+
56+
client.stopWorkflow(wfId, "");
57+
58+
try {
59+
client.getSimpleWorkflowResultWithWait(Integer.class, wfId);
60+
} catch (WorkflowUncompletedException e) {
61+
Assertions.assertEquals(runId, e.getRunId());
62+
Assertions.assertEquals(WorkflowStatus.CANCELED, e.getClosedStatus());
63+
Assertions.assertNull(e.getErrorSubType());
64+
Assertions.assertNull(e.getErrorMessage());
65+
Assertions.assertEquals(0, e.getStateResultsSize());
66+
return;
67+
}
68+
Assertions.fail("no exception caught");
69+
}
70+
71+
@Test
72+
public void testForceFailWorkflow() throws InterruptedException {
73+
final Client client = new Client(WorkflowRegistry.registry, ClientOptions.localDefault);
74+
final long startTs = System.currentTimeMillis();
75+
final String wfId = "testForceFailWorkflow" + startTs / 1000;
76+
final Integer input = 5;
77+
78+
final String runId = client.startWorkflow(
79+
ForceFailWorkflow.class, wfId, 10, input);
80+
81+
try {
82+
client.getSimpleWorkflowResultWithWait(Integer.class, wfId);
83+
} catch (WorkflowUncompletedException e) {
84+
Assertions.assertEquals(runId, e.getRunId());
85+
Assertions.assertEquals(WorkflowStatus.FAILED, e.getClosedStatus());
86+
Assertions.assertEquals(WorkflowErrorType.STATE_DECISION_FAILING_WORKFLOW_ERROR_TYPE, e.getErrorSubType());
87+
Assertions.assertNull(e.getErrorMessage());
88+
Assertions.assertEquals(1, e.getStateResultsSize());
89+
String out = e.getStateResult(0, String.class);
90+
Assertions.assertEquals("a failing message", out);
91+
return;
92+
}
93+
Assertions.fail("no exception caught");
94+
}
95+
96+
@Test
97+
public void testStateApiFailWorkflow() throws InterruptedException {
98+
final Client client = new Client(WorkflowRegistry.registry, ClientOptions.localDefault);
99+
final long startTs = System.currentTimeMillis();
100+
final String wfId = "testStateApiFailWorkflow" + startTs / 1000;
101+
final Integer input = 5;
102+
103+
final String runId = client.startWorkflow(
104+
StateApiFailWorkflow.class, wfId, 10, input);
105+
106+
try {
107+
client.getSimpleWorkflowResultWithWait(Integer.class, wfId);
108+
} catch (WorkflowUncompletedException e) {
109+
Assertions.assertEquals(runId, e.getRunId());
110+
Assertions.assertEquals(WorkflowStatus.FAILED, e.getClosedStatus());
111+
Assertions.assertEquals(WorkflowErrorType.STATE_API_FAIL_MAX_OUT_RETRY_ERROR_TYPE, e.getErrorSubType());
112+
Assertions.assertTrue(e.getErrorMessage().contains("/api/v1/workflowState/decide"));
113+
Assertions.assertEquals(0, e.getStateResultsSize());
114+
return;
115+
}
116+
Assertions.fail("no exception caught");
117+
}
118+
119+
@Test
120+
public void testStateApiTimeoutWorkflow() throws InterruptedException {
121+
final Client client = new Client(WorkflowRegistry.registry, ClientOptions.localDefault);
122+
final long startTs = System.currentTimeMillis();
123+
final String wfId = "testStateApiTimeoutWorkflow" + startTs / 1000;
124+
final Integer input = 5;
125+
126+
final String runId = client.startWorkflow(
127+
StateApiTimeoutFailWorkflow.class, wfId, 10, input);
128+
129+
try {
130+
client.getSimpleWorkflowResultWithWait(Integer.class, wfId);
131+
} catch (WorkflowUncompletedException e) {
132+
Assertions.assertEquals(runId, e.getRunId());
133+
Assertions.assertEquals(WorkflowStatus.FAILED, e.getClosedStatus());
134+
Assertions.assertEquals(WorkflowErrorType.STATE_API_FAIL_MAX_OUT_RETRY_ERROR_TYPE, e.getErrorSubType());
135+
Assertions.assertTrue(
136+
e.getErrorMessage().contains("activity StartToClose timeout"),
137+
e.getErrorMessage()
138+
);
139+
Assertions.assertEquals(0, e.getStateResultsSize());
140+
return;
141+
}
142+
Assertions.fail("no exception caught");
143+
}
144+
}

0 commit comments

Comments
 (0)