Repository: kestra-io/kestra
File: script/src/main/java/io/kestra/plugin/scripts/runner/docker/Docker.java (Java)

Code before:
                    .withLabelFilter(labels)
                    .exec();

                if (!existing.isEmpty()) {
                    containerId = existing.get(0).getId();
                    logger.debug("Resuming existing container: {}", containerId);
                }
            }

            List<Path> relativeWorkingDirectoryFilesPaths = taskCommands.relativeWorkingDirectoryFilesPaths(true);
            boolean hasFilesToUpload = !ListUtils.isEmpty(relativeWorkingDirectoryFilesPaths);
            boolean hasFilesToDownload = !ListUtils.isEmpty(filesToDownload);
            boolean outputDirectoryEnabled = taskCommands.outputDirectoryEnabled();
            boolean needVolume = hasFilesToDownload || hasFilesToUpload || outputDirectoryEnabled;
            String filesVolumeName = null;
            var strategy = runContext.render(this.fileHandlingStrategy).as(FileHandlingStrategy.class).orElse(null);

            // pull image only if we will create a new container
            if (containerId == null) {
                var renderedPolicy = runContext.render(this.getPullPolicy()).as(PullPolicy.class).orElseThrow();
                if (!PullPolicy.NEVER.equals(renderedPolicy)) {
                    pullImage(dockerClient, image, renderedPolicy, logger);
                }

                if (getIsKilled().get()) {
                    throw new IllegalStateException("Task was killed during image pull");
                }

                // create container
                CreateContainerCmd container = configure(taskCommands, dockerClient, runContext, additionalVars);
                CreateContainerResponse exec = container.exec();
                containerId = exec.getId();
                containerIdRef.set(containerId);

                if (logger.isTraceEnabled()) {
                    logger.trace("Container created: {}", containerId);
                }

                // create a volume if we need to handle files
                if (needVolume && FileHandlingStrategy.VOLUME.equals(strategy)) {
                    CreateVolumeCmd files = dockerClient.createVolumeCmd()
                        .withLabels(labels);
                    filesVolumeName = files.exec().getName();
                    if (logger.isTraceEnabled()) {
                        logger.trace("Volume created: {}", filesVolumeName);
                    }

                    String remotePath = windowsToUnixPath(taskCommands.getWorkingDirectory().toString());

                    // first, create an archive
                    Path fileArchive = runContext.workingDir().createFile("inputFiles.tar");

Review comment (refactor):
Do you really need to throw an exception here?
Returning null didn't work?

I wonder if we should not move that to the parent TaskRunner as it would be usefull for other task runners?

If so:
- create a method `checkKilled()`  in TaskRunner
- throw a dedicated Exception: `KilledException` wich extends `KestraRuntimeException`

So it would be reusable.

Code after:
                    .withShowAll(true)
                    .withLabelFilter(labels)
                    .exec();

                if (!existing.isEmpty()) {
                    containerId = existing.getFirst().getId();
                    containerIdRef.set(containerId);
                    logger.debug("Resuming existing container: {}", containerId);
                }
            }

            List<Path> relativeWorkingDirectoryFilesPaths = taskCommands.relativeWorkingDirectoryFilesPaths(true);
            boolean hasFilesToUpload = !ListUtils.isEmpty(relativeWorkingDirectoryFilesPaths);
            boolean hasFilesToDownload = !ListUtils.isEmpty(filesToDownload);
            boolean outputDirectoryEnabled = taskCommands.outputDirectoryEnabled();
            boolean needVolume = hasFilesToDownload || hasFilesToUpload || outputDirectoryEnabled;
            String filesVolumeName = null;
            var strategy = runContext.render(this.fileHandlingStrategy).as(FileHandlingStrategy.class).orElse(null);

            // pull image only if we will create a new container
            if (containerId == null) {
                var renderedPolicy = runContext.render(this.getPullPolicy()).as(PullPolicy.class).orElseThrow();
                if (!PullPolicy.NEVER.equals(renderedPolicy)) {
                    pullImage(dockerClient, image, renderedPolicy, logger);
                }

                // we check, if killed during image pull.
                checkKilled("Execution was killed during image pull.");

                // create container
                CreateContainerCmd container = configure(taskCommands, dockerClient, runContext, additionalVars);
                CreateContainerResponse exec = container.exec();
                containerId = exec.getId();
                containerIdRef.set(containerId);

                if (logger.isTraceEnabled()) {
                    logger.trace("Container created: {}", containerId);
                }

                // create a volume if we need to handle files
                if (needVolume && FileHandlingStrategy.VOLUME.equals(strategy)) {
                    CreateVolumeCmd files = dockerClient.createVolumeCmd()
                        .withLabels(labels);
                    filesVolumeName = files.exec().getName();
                    if (logger.isTraceEnabled()) {
                        logger.trace("Volume created: {}", filesVolumeName);
                    }

                    String remotePath = windowsToUnixPath(taskCommands.getWorkingDirectory().toString());

                    // first, create an archive