Throw an error depending on the expiration of TemporalConfigStorage

[TrsNium]

objective

KubernetesCommandExecutor now throws an exception when the execution time of a pod exceeds the defaultPodTTL.
But we also need to consider that the pod execution time may exceed expiration of TemporalConfigStorage.
This is because pod error detection can be done early.

problem

if we continue processing when the pod's execution time exceed the expiration of TemporalConfigStorage, the curl command will raise an error when trying to upload the archive to the expired TemporaryConfigStorage at the end of the pod process.
It is inefficient to not notice error until the end of the pod process.

Fail if inTemporalConfigStorage has expired.

digdag/digdag-standards/src/main/java/io/digdag/standards/command/KubernetesCommandExecutor.java

Line 183 in ec9ba0e

bashArguments.add(s("curl -s \"%s\" --output %s", url, relativeArchivePath.toString()));

Fail if outTemporalConfigStorage has expired.

digdag/digdag-standards/src/main/java/io/digdag/standards/command/KubernetesCommandExecutor.java

Line 213 in ec9ba0e

bashArguments.add(s("curl -s -X PUT -T %s -L \"%s\"", outputArchivePathName, url));

changes

Make throwing an error depending on the expiration of TemporalConfigStorage.

reopened #1359

[yoyama]

@szyn Could you review this PR?
LGTM for me.

[szyn]

I guess you are using a non-blocking operator such as py>. In this case, IIRC, (defaultPodTTL.isPresent() && isRunningLongerThanTTL(previousStatusJson)) will never become true. In addtion, this block will not execute when user/WorkflowExecutionTimeoutEnforcer cancel-requested to the attempt. Is it OK/acceptable for you?
We also had the same kind of issue. Thus we introduce the cleanup function to the Operator and CommnadExecutor interfaces and implements them in PyOperatorFactory and EcsCommandExecutor. Please see #1480, and this function helps to perform the cleanup process if the attempt received cancel-requested. I feel it is a good idea to handle it by using the same way to avoid such an issue.
How about adding it to KubernetesCommandExecutor as well? How do you think about that?