Fix polling (ECS) command executors fail to run tasks on retry

[builtinnya]

This PR fixes a bug that ECS command executor fails to submit ECS tasks on retry.

Reproducible steps

digdag version: v0.10.1 (and some earlier versions probably)

Assuming that ECS command executor is set up correctly:

1. Add and run the following workflow

   +ecs-retry-test:
     sh>: |
       echo "Task has been executed!"
       exit 1
     _retry: 1
   

2. See task logs to confirm that the command executor actually prints Task has been executed! once.

Expected behavior

In the above steps, the command executor should print Task has been executed! twice.

Cause

Each operator runs a command (= submits an ECS task) only when "commandStatus" doesn't exist on state params.
However, ECS command executor's polling mechanism persists "commandStatus" even on retry, which tries to poll an ECS task that has already exited.

Approach in this PR

This PR takes minimal change approach and simply removes "commandStatus" from state params on retry.

Considerations

• Maybe we should also use TaskExecutionException for commands' failure to propagate state params.
• Currently, each operator tries to remove "commandStatus" from state params on failure (e.g. sh> op) but it has no effect on the state params on retry because it just throws RuntimeException right after that. Should we remove these confusing lines?

[szyn]

Thank you for creating this PR!
First of all, we haven't been able to reproduce this issue in our environment yet. It seems that task retry works appropriately. Thus let me confirm, is this issue happening for another executor type such as Simple/DockerCommandExecutor as well?
I understand this approach is quite simple, but I'm worried that this change for BaseOperator affects all command executors. If this problem is happening only with EcsCommandExecutor, we would better take a different approach to fix it.

[builtinnya]

@szyn
This only applies for EcsCommandExecutor. Retry works correctly for other command executors because it doesn't rely on "commandStatus" and can't be affected by the previous execution anyway.

I'm worried that this change for BaseOperator affects all command executors. If this problem is happening only with EcsCommandExecutor, we would better take a different approach to fix it.

I understand your concern but we probably need to change codes other than EcsCommandExecutor and could affect all other command executors anyway because EcsCommandExecutor's polling behavior itself is achieved by coordination of BaseOperator, ShOperatorFactory, and so on.

At least the current executors can't be affected by this change as it can't rely on "commandStatus" presence.

digdag/digdag-standards/src/main/java/io/digdag/standards/operator/ShOperatorFactory.java

Lines 105 to 113 in 0c6e58a

	if (!state.has("commandStatus")) {
	// Run the code since command state doesn't exist
	status = runCommand(params, commandContext);
	}
	else {
	// Check the status of the running command
	final ObjectNode previousStatusJson = state.get("commandStatus", ObjectNode.class);
	status = exec.poll(commandContext, previousStatusJson);
	}

Operators run commands only if "commandStatus" doesn't exist in state params and the else branch is never executed on other executors (poll() is currently implemented only on EcsCommandExecutor.)

To summarize my points:

• The current EcsCommandExecutor's polling behavior is already relying on the implementation of BaseOperator, ShOperatorFactory, PyOperatorFactory, and so on. So changing some of these codes would be inevitable.
• Command executors other than EcsCommandExecutor aren't affected by this PR's changes.

Please let me know if some of my changes aren't clear to you and any suggestion for better changes.

[myui]

I could reproduce it with ECS command executor.
Since commandStatus is not used by other command executors, no drawbacks in this PR. LGTM.

While sh and py operator tried to remove commandStatus for non-zero exits, I guess it's different objects and no effect for retrying.
https://github.com/treasure-data/digdag/blob/master/digdag-standards/src/main/java/io/digdag/standards/operator/ShOperatorFactory.java#L120

[myui]

Sho agreed to merge this PR. Thank you for contributing!

[szyn]

Yes, finally, I could reproduce this issue... and this approach looks good to me 👍 Thank you for contributing!