Fix Google Batch task array causes process to fail (#5780) [ci fast]

Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>
Signed-off-by: jorgee <jorge.ejarque@seqera.io>
Co-authored-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>

Author: jorgee

modules/nextflow/src/main/groovy/nextflow/processor/TaskHandler.groovy

@@ -259,7 +259,7 @@ abstract class TaskHandler {
 
     /**
      * Determine if a task is ready for execution or it depends on resources
-     * e.g. container that needs to be provisionied
+     * e.g. container that needs to be provisioned
      *
      * @return {@code true} when the task is ready for execution, {@code false} otherwise
      */

plugins/nf-google/src/main/nextflow/cloud/google/batch/GoogleBatchScriptLauncher.groovy

@@ -49,6 +49,7 @@ class GoogleBatchScriptLauncher extends BashWrapperBuilder implements GoogleBatc
     private Path remoteBinDir
     private Set<String> buckets = new HashSet<>()
     private PathTrie pathTrie = new PathTrie()
+    private boolean isArray
 
     /* ONLY FOR TESTING - DO NOT USE */
     protected GoogleBatchScriptLauncher() {}
@@ -125,7 +126,9 @@ class GoogleBatchScriptLauncher extends BashWrapperBuilder implements GoogleBatc
 
     @Override
     String runCommand() {
-        launchCommand(workDirMount)
+        return isArray
+            ? launchArrayCommand(workDirMount)
+            : launchCommand(workDirMount)
     }
 
     @Override
@@ -183,6 +186,20 @@ class GoogleBatchScriptLauncher extends BashWrapperBuilder implements GoogleBatc
         return this
     }
 
+    GoogleBatchScriptLauncher withIsArray(boolean value) {
+        this.isArray = value
+        return this
+    }
+
+    static String launchArrayCommand(String workDir ) {
+        // when executing a job array run directly the command script
+        // to prevent that all child jobs write on the same .command.*
+        // control files, causing an issue with the gcsfuse mount
+        // For the same reason the .command.log file is not uploaded
+        // See https://github.com/nextflow-io/nextflow/issues/5777
+        "/bin/bash ${workDir}/${TaskRun.CMD_SCRIPT}"
+    }
+
     static String launchCommand( String workDir ) {
         "trap \"{ cp ${TaskRun.CMD_LOG} ${workDir}/${TaskRun.CMD_LOG}; }\" ERR; /bin/bash ${workDir}/${TaskRun.CMD_RUN} 2>&1 | tee ${TaskRun.CMD_LOG}"
     }

plugins/nf-google/src/main/nextflow/cloud/google/batch/GoogleBatchTaskHandler.groovy

@@ -145,6 +145,7 @@ class GoogleBatchTaskHandler extends TaskHandler implements FusionAwareTask {
             final taskBean = task.toTaskBean()
             return new GoogleBatchScriptLauncher(taskBean, executor.remoteBinDir)
                 .withConfig(executor.config)
+                .withIsArray(task.isArray())
         }
     }