fix(core): don't retry a task if the execution is killed (#2057)

Note that this will make the task not killing but ends with the "previous" state (failed), the execution will then be marked as failed and not killed at the moment, but if #1747 is merged the execution will correctly be marked as KILLED.
Anyway, currently, without this , the exuction is also marked as FAILED but the attempts continue so this PR improve the current process by shortcirciting the attempt for killed execution.

core/src/main/java/io/kestra/core/runners/Worker.java

@@ -308,7 +308,8 @@ private WorkerTaskResult run(WorkerTask workerTask, Boolean cleanUp) throws Queu
         WorkerTask finalWorkerTask = Failsafe
             .with(AbstractRetry.<WorkerTask>retryPolicy(workerTask.getTask().getRetry())
                 .handleResultIf(result -> result.getTaskRun().lastAttempt() != null &&
-                    Objects.requireNonNull(result.getTaskRun().lastAttempt()).getState().getCurrent() == State.Type.FAILED
+                        result.getTaskRun().lastAttempt().getState().getCurrent() == State.Type.FAILED &&
+                        !killedExecution.contains(result.getTaskRun().getExecutionId())
                 )
                 .onRetry(e -> {
                     WorkerTask lastResult = e.getLastResult();
@@ -443,7 +444,7 @@ private WorkerTask runAttempt(WorkerTask workerTask) {
 
         Logger logger = runContext.logger();
 
-        if (!(workerTask.getTask() instanceof RunnableTask)) {
+        if (!(workerTask.getTask() instanceof RunnableTask<?> task)) {
             // This should never happen but better to deal with it than crashing the Worker
             TaskRunAttempt attempt = TaskRunAttempt.builder().state(new State().withState(State.Type.FAILED)).build();
             List<TaskRunAttempt> attempts = this.addAttempt(workerTask, attempt);
@@ -453,8 +454,6 @@ private WorkerTask runAttempt(WorkerTask workerTask) {
             return workerTask.withTaskRun(taskRun);
         }
 
-        RunnableTask<?> task = (RunnableTask<?>) workerTask.getTask();
-
         TaskRunAttempt.TaskRunAttemptBuilder builder = TaskRunAttempt.builder()
             .state(new State().withState(State.Type.RUNNING));
 
@@ -502,7 +501,7 @@ private WorkerTask runAttempt(WorkerTask workerTask) {
             log.debug("Outputs\n{}", JacksonMapper.log(workerThread.getTaskOutput()));
         }
 
-        if (runContext.metrics().size() > 0 && log.isTraceEnabled()) {
+        if (!runContext.metrics().isEmpty() && log.isTraceEnabled()) {
             log.trace("Metrics\n{}", JacksonMapper.log(runContext.metrics()));
         }