Can't run gradle on the train so I'm just gonna send it

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

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

@@ -392,8 +392,8 @@ class AzBatchService implements Closeable {
             throw new IllegalArgumentException("Missing Azure Blob storage SAS token")
 
         final container = task.getContainer()
-        if( !container )
-            log.warn "Missing container image for process: $task.name"
+        if( !container && config.batch().requireContainer )
+            throw new IllegalArgumentException("Missing container image for process: $task.name\nYou can disable this behaviour setting `azure.batch.requireContainer=false` in the nextflow config file")
         final taskId = "nf-${task.hash.toString()}"
         // get the pool config
         final pool = getPoolSpec(poolId)

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

@@ -60,6 +60,7 @@ class AzBatchTaskHandler extends TaskHandler implements FusionAwareTask {
         this.outputFile = task.workDir.resolve(TaskRun.CMD_OUTFILE)
         this.errorFile = task.workDir.resolve(TaskRun.CMD_ERRFILE)
         this.exitFile = task.workDir.resolve(TaskRun.CMD_EXIT)
+        validateConfiguration()
     }
 
     /** only for testing purpose - DO NOT USE */
@@ -69,6 +70,12 @@ class AzBatchTaskHandler extends TaskHandler implements FusionAwareTask {
         return executor.batchService
     }
 
+    void validateConfiguration() {
+        if (!task.container && config.batch().requireContainer ) {
+            throw new ProcessUnrecoverableException("No container image specified for process $task.name -- Either specify the container to use in the process definition or with 'process.container' value in your config. You can disable this behaviour setting `azure.batch.requireContainer=false` in the nextflow config file")
+        }
+    }
+
     protected BashWrapperBuilder createBashWrapper() {
         fusionEnabled()
                 ? fusionLauncher()

plugins/nf-azure/src/main/nextflow/cloud/azure/config/AzBatchOpts.groovy

@@ -54,6 +54,7 @@ class AzBatchOpts implements CloudTransferOptions {
     Boolean deleteJobsOnCompletion
     Boolean deletePoolsOnCompletion
     Boolean deleteTasksOnCompletion
+    Boolean requireContainer
     CopyToolInstallMode copyToolInstallMode
 
     Map<String,AzPoolOpts> pools
@@ -67,6 +68,7 @@ class AzBatchOpts implements CloudTransferOptions {
         location = config.location
         autoPoolMode = config.autoPoolMode
         allowPoolCreation = config.allowPoolCreation
+        requireContainer = config.requireContainer ?: true
         terminateJobsOnCompletion = config.terminateJobsOnCompletion != Boolean.FALSE
         deleteJobsOnCompletion = config.deleteJobsOnCompletion
         deletePoolsOnCompletion = config.deletePoolsOnCompletion

plugins/nf-azure/src/test/nextflow/cloud/azure/batch/AzBatchTaskHandlerTest.groovy

@@ -18,6 +18,45 @@ import spock.lang.Specification
  */
 class AzBatchTaskHandlerTest extends Specification {
 
+    def 'should validate config' () {
+        when:
+        def task = Mock(TaskRun) { getName() >> 'foo'; }
+        and:
+        new AzBatchTaskHandler(task: task)
+                .validateConfiguration()
+        then:
+        def e = thrown(ProcessUnrecoverableException)
+        e.message.startsWith('No container image specified for process foo')
+
+
+        when:
+        task = Mock(TaskRun) { getName() >> 'foo'; getContainer() >> 'ubuntu' }
+        and:
+        new AzBatchTaskHandler(task: task)
+                .validateConfiguration()
+        then:
+        noExceptionThrown()
+    }
+
+    def 'should ignore missing container if disabled' () {
+        when:
+        def task = Mock(TaskRun) { getName() >> 'foo'; }
+        and:
+        new AzBatchTaskHandler(task: task, config: new AzConfig([batch: [requireContainer: false]]))
+                .validateConfiguration()
+        then:
+        noExceptionThrown()
+
+
+        when:
+        task = Mock(TaskRun) { getName() >> 'foo'; getContainer() >> 'ubuntu' }
+        and:
+        new AzBatchTaskHandler(task: task)
+                .validateConfiguration()
+        then:
+        noExceptionThrown()
+    }
+
     def 'should submit task' () {
         given:
         def builder = Mock(BashWrapperBuilder)