Complete task task resource cleanup on task deletion.

If there is no task deployment when a task definition is deleted SCDF will delete all app resources for the task definition name across all platforms.
If there is a task deployment for that task definition, then it will only delete those app resources associated with that task deployment.
This is to resolve the case where a task launch failed on the platform and thus no task deployment record is created.  Hence we don't know what platform the launch attempt occured.

resolves spring-cloud#4574

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/impl/DefaultTaskDeleteService.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.dataflow.server.service.impl;
 
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
@@ -398,8 +399,28 @@ private void destroyTask(TaskDefinition taskDefinition) {
 			}
 		}
 		else {
-			logger.info("TaskLauncher.destroy not invoked for task " +
-					taskDefinition.getTaskName() + ". Did not find a previously launched task to destroy.");
+			if(!findAndDeleteTaskResourcesAcrossPlatforms(taskDefinition)) {
+				logger.info("TaskLauncher.destroy not invoked for task " +
+						taskDefinition.getTaskName() + ". Did not find a previously launched task to destroy.");
+			}
+		}
+	}
+
+	private boolean findAndDeleteTaskResourcesAcrossPlatforms(TaskDefinition taskDefinition) {
+		boolean result = false;
+		Iterable<Launcher> launchers = launcherRepository.findAll();
+		Iterator<Launcher> launcherIterator = launchers.iterator();
+		while(launcherIterator.hasNext()) {
+			Launcher launcher = launcherIterator.next();
+			try {
+				launcher.getTaskLauncher().destroy(taskDefinition.getName());
+				logger.info(String.format("Deleted task app resources for %s in platform %s", taskDefinition.getName(), launcher.getName()));
+				result = true;
+			}
+			catch (Exception ex) {
+				logger.info(String.format("Attempted delete of app resources for %s but none found on platform %s.", taskDefinition.getName(), launcher.getName()));
+			}
 		}
+		return result;
 	}
 }

spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultTaskExecutionServiceTests.java

@@ -734,6 +734,9 @@ public void createSimpleTask() {
 			initializeSuccessfulRegistry(appRegistry);
 			taskSaveService.saveTaskDefinition(new TaskDefinition("simpleTask", "AAA --foo=bar"));
 			verifyTaskExistsInRepo("simpleTask", "AAA --foo=bar", taskDefinitionRepository);
+			taskDeleteService.deleteTaskDefinition("simpleTask", true);
+			String logEntries = outputCapture.toString();
+			assertTrue(logEntries.contains("Deleted task app resources for"));
 		}
 
 		@Test
@@ -749,6 +752,10 @@ public void executeSingleTaskTest() {
 			assertEquals(TASK_NAME_ORIG, taskDeployment.getTaskDefinitionName());
 			assertEquals("default", taskDeployment.getPlatformName());
 			assertNotNull("TaskDeployment createdOn field should not be null", taskDeployment.getCreatedOn());
+			taskDeleteService.deleteTaskDefinition(TASK_NAME_ORIG, true);
+			String logEntries = outputCapture.toString();
+			assertTrue(!logEntries.contains("Deleted task app resources for"));
+			assertTrue(!logEntries.contains("Attempted delete of app resources for"));
 		}
 
 		@Test
@@ -977,6 +984,19 @@ public void executeSameTaskDefinitionOnMultiplePlatforms() {
 			}
 		}
 
+		@Test
+		@DirtiesContext
+		public void executeDeleteNoDeploymentWithMultiplePlatforms() {
+			initializeSuccessfulRegistry(appRegistry);
+			when(taskLauncher.launch(any())).thenReturn("0");
+			this.launcherRepository.save(new Launcher("anotherPlatform", "local", taskLauncher));
+			taskDeleteService.deleteTaskDefinition(TASK_NAME_ORIG, true);
+			String logEntries = outputCapture.toString();
+			assertTrue(logEntries.contains("Deleted task app resources for myTask_ORIG in platform anotherPlatform"));
+			assertTrue(logEntries.contains("Deleted task app resources for myTask_ORIG in platform default"));
+			assertTrue(logEntries.contains("Deleted task app resources for myTask_ORIG in platform MyPlatform"));
+		}
+
 		@Test
 		@DirtiesContext
 		public void executeTaskWithNullIDReturnedTest() {