If an activity is failing, stuck the workflow and make it queriable #10121
Conversation
github-actions
bot
added
area/platform
| * activity. | ||
| */ | ||
| @SignalMethod | ||
| void retryFailActivity(); |
| @@ -77,4 +91,22 @@ | |||
| @QueryMethod | |||
| JobInformation getJobInformation(); | |||
|
|
|||
| @Data | |||
| @NoArgsConstructor | |||
There was a problem hiding this comment.
why do we need a no args constructor?
| private <INPUT> void runActivity(Consumer<INPUT> consumer, INPUT input) { | ||
| try { | ||
| consumer.accept(input); | ||
| } catch (Exception e) { | ||
| log.error("Failed to run an activity for the connection " + connectionId, e); | ||
| workflowState.setStuck(true); | ||
| workflowState.setRetryFailedActivity(false); | ||
| Workflow.await(() -> workflowState.isRetryFailedActivity()); | ||
| log.error("Retrying an activity for the connection " + connectionId, e); | ||
| workflowState.setStuck(false); | ||
| workflowState.setRetryFailedActivity(false); | ||
| runActivity(consumer, input); | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
can we dry this up. the duplicate code here is going to be easy to have go out of sync.
| private <INPUT> void runActivity(Consumer<INPUT> consumer, INPUT input) { | |
| try { | |
| consumer.accept(input); | |
| } catch (Exception e) { | |
| log.error("Failed to run an activity for the connection " + connectionId, e); | |
| workflowState.setStuck(true); | |
| workflowState.setRetryFailedActivity(false); | |
| Workflow.await(() -> workflowState.isRetryFailedActivity()); | |
| log.error("Retrying an activity for the connection " + connectionId, e); | |
| workflowState.setStuck(false); | |
| workflowState.setRetryFailedActivity(false); | |
| runActivity(consumer, input); | |
| } | |
| } | |
| private <INPUT> void runActivity2(final Consumer<INPUT> consumer, final INPUT input) { | |
| // make the consumer look like a function so that we can reuse logic in runActivityWithOutput | |
| runActivityWithOutput((inputInternal) -> { | |
| consumer.accept(inputInternal); | |
| return null; | |
| }, input); | |
| } |
| @@ -334,4 +365,34 @@ private void continueAsNew(final ConnectionUpdaterInput connectionUpdaterInput) | |||
| } | |||
| } | |||
|
|
|||
| private <INPUT, OUTPUT> OUTPUT runActivityWithOutput(Function<INPUT, OUTPUT> mapper, INPUT input) { | |||
There was a problem hiding this comment.
this is only for a subset of activities. should the method name say short activities or mandatory activities or something? also can we add a javadoc comment to this method explaining why it is necessary, please?
| @Data | ||
| @NoArgsConstructor | ||
| @AllArgsConstructor | ||
| class StuckInformation { |
There was a problem hiding this comment.
can we replace the word "stuck" with "quarantined", please? quarantined is a common state name for a workflow that needs manual intervention. within that we can add information on the reason it is stuck.
Yes, but it should be in another PR. |
Done |
It is needed for the internal temporal serialization |
Done |
What
After an activity failure, we are blocking the workflow. A new query method is available to query the workflows to get the list of workflow being stuck.
Then the activity can be retry with a signal.