refactor(nf-azure): remove AzBatchProcessObserver; reuse terminateJobsOnCompletion; refactor task submission; reduce logging

- Remove Azure Batch process observer:
  - Delete AzBatchProcessObserver, its factory, and tests
  - Remove registration from META-INF/extensions.idx
- Reuse `azure.batch.terminateJobsOnCompletion` to set Azure Batch job `OnAllTasksComplete` (eager auto-termination)
- Refactor task submission flow in AzBatchService:
  - Extract helpers: `submitTaskToJob`, `recreateJobForTask`, `setAutoTerminateIfEnabled`
  - Handle 409 conflicts by recreating the job and retrying submission

Signed-off-by: adamrtalbot <[email protected]>

Author: adamrtalbot

docs/reference/config.md

@@ -381,7 +381,7 @@ The following settings are available:
 `azure.batch.poolIdentityClientId`
 : :::{versionadded} 25.05.0-edge
   :::
-: The client ID for an Azure [managed identity](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/overview) that is available on all Azure Batch node pools. This identity is used by Fusion to authenticate to Azure storage. If set to `'auto'`, Fusion will use the first available managed identity.
+: When the workflow completes, set all jobs to terminate on task completion. (default: `true`).
 
 `azure.batch.pools.<name>.autoScale`
 : Enable autoscaling feature for the pool identified with `<name>`.
@@ -450,7 +450,7 @@ The following settings are available:
 `azure.batch.terminateJobsOnCompletion`
 : :::{versionadded} 23.05.0-edge
   :::
-: When the workflow completes, set all jobs to terminate on task completion (default: `true`).
+: Set all jobs to terminate on task completion (default: `true`).
 
 `azure.managedIdentity.clientId`
 : The client ID for an Azure [managed identity](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/overview). Defaults to environment variable `AZURE_MANAGED_IDENTITY_USER`.

plugins/nf-azure/src/main/nextflow/cloud/azure/batch/AzBatchProcessObserver.groovy

@@ -336,7 +336,7 @@ class AzBatchService implements Closeable {
     }
 
     protected AzureNamedKeyCredential createBatchCredentialsWithKey() {
-        log.debug "[AZURE BATCH] Creating Azure Batch client using shared key creddentials"
+        log.debug "[AZURE BATCH] Creating Azure Batch client using shared key credentials"
 
         if( !config.batch().endpoint )
             throw new IllegalArgumentException("Missing Azure Batch endpoint -- Specify it in the nextflow.config file using the setting 'azure.batch.endpoint'")
@@ -568,9 +568,57 @@ class AzBatchService implements Closeable {
     }
 
     AzTaskKey runTask(String poolId, String jobId, TaskRun task) {
-        final taskToAdd = createTask(poolId, jobId, task)
-        apply(() -> client.createTask(jobId, taskToAdd))
-        return new AzTaskKey(jobId, taskToAdd.getId())
+        final BatchTaskCreateContent taskToAdd = createTask(poolId, jobId, task)
+        final AzTaskKey submitted = submitTaskToJob(poolId, jobId, task, taskToAdd)
+        setAutoTerminateIfEnabled(submitted.jobId, taskToAdd.getId())
+        return submitted
+    }
+
+    /**
+     * Submit a task and transparently recover from a 409 job state conflict by
+     * creating a fresh job and retrying the submission.
+     */
+    protected AzTaskKey submitTaskToJob(String poolId, String jobId, TaskRun task, BatchTaskCreateContent taskToAdd) {
+        try {
+            apply(() -> client.createTask(jobId, taskToAdd))
+            return new AzTaskKey(jobId, taskToAdd.getId())
+        }
+        catch( HttpResponseException e ) {
+            if( e.response.statusCode == 409 && config.batch().terminateJobsOnCompletion ) {
+                final String newJobId = recreateJobForTask(poolId, task, jobId, taskToAdd.getId())
+                apply(() -> client.createTask(newJobId, taskToAdd))
+                return new AzTaskKey(newJobId, taskToAdd.getId())
+            }
+            throw e
+        }
+    }
+
+    /**
+     * Create a new Azure Batch job for the given process and update internal bookkeeping
+     * so subsequent tasks for the same (process,pool) use the fresh job id.
+     */
+    protected String recreateJobForTask(String poolId, TaskRun task, String oldJobId, String taskId) {
+        log.debug "Job ${oldJobId} is in completed/terminating state, creating a new job for task ${taskId}"
+        allJobIds.values().removeAll { it == oldJobId }
+        final String newJobId = createJob0(poolId, task)
+        final AzJobKey mapKey = new AzJobKey(task.processor, poolId)
+        allJobIds[mapKey] = newJobId
+        return newJobId
+    }
+
+    /**
+     * If configured, set the job to auto-terminate when all tasks complete.
+     */
+    protected void setAutoTerminateIfEnabled(String jobId, String taskId) {
+        if( !config.batch().terminateJobsOnCompletion )
+            return
+        log.trace "Eagerly setting job ${jobId} to auto-terminate after submitting task ${taskId}"
+        try {
+            setJobAutoTermination(jobId)
+        }
+        catch( Exception e ) {
+            log.trace "Failed to eagerly set auto-termination for job ${jobId} after task submission - ${e.message ?: e}"
+        }
     }
 
     protected List<String> getShareVolumeMounts(AzVmPoolSpec spec) {
@@ -984,6 +1032,7 @@ class AzBatchService implements Closeable {
      * @param jobId The Azure Batch job ID to set for auto-termination
      */
     void setJobAutoTermination(String jobId) {
+        log.trace "Eagerly setting Azure Batch job ${jobId} to auto-terminate when all tasks complete"
         setJobTermination(jobId)
     }
 
@@ -994,22 +1043,20 @@ class AzBatchService implements Closeable {
      */
     protected void setJobTermination(String jobId) {
         try {
-            log.trace "Setting Azure job ${jobId} to terminate on completion"
-
             final job = apply(() -> client.getJob(jobId))
+            
             final poolInfo = job.poolInfo
 
             final jobParameter = new BatchJobUpdateContent()
                     .setOnAllTasksComplete(OnAllBatchTasksComplete.TERMINATE_JOB)
                     .setPoolInfo(poolInfo)
-
             apply(() -> client.updateJob(jobId, jobParameter))
         }
         catch (HttpResponseException e) {
             if (e.response.statusCode == 409) {
-                log.debug "Azure Batch job ${jobId} already terminated, skipping auto-termination setup"
+                log.debug "Azure Batch job ${jobId} already terminated or in terminal state, skipping auto-termination setup"
             } else {
-                log.warn "Unable to set auto-termination for Azure Batch job ${jobId} - Status: ${e.response.statusCode}, Reason: ${e.message ?: e}"
+                log.warn "Unable to set auto-termination for Azure Batch job ${jobId} - HTTP Status: ${e.response.statusCode}, Reason: ${e.message ?: e}"
             }
         }
         catch (Exception e) {

plugins/nf-azure/src/main/nextflow/cloud/azure/batch/AzBatchProcessObserverFactory.groovy

@@ -119,7 +119,7 @@ class AzureConfigTest extends Specification {
                                              location: LOCATION,
                                              autoPoolMode: true,
                                              allowPoolCreation: true,
-                                             terminateJobsOnCompletion: false,
+                                              terminateJobsOnCompletion: false,
                                              deleteJobsOnCompletion: true,
                                              deletePoolsOnCompletion: true,
                                              deleteTasksOnCompletion: false,