$END will not be validated as a app

- Updated to check task.name instead of task.taskName
- Updated to verify that taskNode.getName is not null before assessing if it contains $END
- resolves #4387

Author: cppwfs

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

@@ -99,15 +99,17 @@ public void saveTaskDefinition(TaskDefinition taskDefinition) {
 		if (taskNode.isComposed()) {
 			// Create the child task definitions needed for the composed task
 			taskNode.getTaskApps().forEach(task -> {
-				String labelPrefix = StringUtils.hasText(task.getLabel()) ? task.getLabel() + ":" : "";
-				// Add arguments to child task definitions
-				String generatedTaskDSL = labelPrefix + task.getName() + task.getArguments().entrySet().stream()
-						.map(argument -> String.format(" --%s=%s", argument.getKey(),
-								DefinitionUtils.autoQuotes(argument.getValue())))
-						.collect(Collectors.joining());
-				TaskDefinition composedTaskElementDefinition = new TaskDefinition(task.getExecutableDSLName(),
-						generatedTaskDSL);
-				saveStandardTaskDefinition(composedTaskElementDefinition);
+				if (StringUtils.hasText(task.getName()) && !task.getName().equals("$END")) {
+					String labelPrefix = StringUtils.hasText(task.getLabel()) ? task.getLabel() + ":" : "";
+					// Add arguments to child task definitions
+					String generatedTaskDSL = labelPrefix + task.getName() + task.getArguments().entrySet().stream()
+							.map(argument -> String.format(" --%s=%s", argument.getKey(),
+									DefinitionUtils.autoQuotes(argument.getValue())))
+							.collect(Collectors.joining());
+					TaskDefinition composedTaskElementDefinition = new TaskDefinition(task.getExecutableDSLName(),
+							generatedTaskDSL);
+					saveStandardTaskDefinition(composedTaskElementDefinition);
+				}
 			});
 			taskDefinitionRepository.save(taskDefinition);
 		}

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

@@ -1355,6 +1355,28 @@ public void executeComposedTaskwithUserCTRName() {
 			assertNull(request.getDefinition().getProperties().get("globalstreamkey"));
 		}
 
+		@Test
+		@DirtiesContext
+		public void executeComposedTaskWithEnd() {
+			String dsl = "timestamp '*'->t1: timestamp 'FOO'->$END";
+			initializeSuccessfulRegistry(appRegistry);
+
+			taskSaveService.saveTaskDefinition(new TaskDefinition("transitionTask", dsl));
+			when(taskLauncher.launch(any())).thenReturn("0");
+
+			Map<String, String> properties = new HashMap<>();
+			properties.put("app.t1.timestamp.format", "YYYY");
+			assertEquals(1L, this.taskExecutionService.executeTask("transitionTask", properties, new LinkedList<>()));
+			ArgumentCaptor<AppDeploymentRequest> argumentCaptor = ArgumentCaptor.forClass(AppDeploymentRequest.class);
+			verify(this.taskLauncher, atLeast(1)).launch(argumentCaptor.capture());
+
+			AppDeploymentRequest request = argumentCaptor.getValue();
+			assertEquals("transitionTask", request.getDefinition().getProperties().get("spring.cloud.task.name"));
+			assertEquals("YYYY",
+					request.getDefinition().getProperties().get("composed-task-app-properties.app.t1.timestamp.format"));
+
+		}
+
 		@Test
 		@DirtiesContext
 		public void executeComposedTaskWithLabels() {