Add dataflowTaskLauncherSink IT test (#4652)

* Add dataflowTaskLauncherSink IT test

* Adjustments

 - Add explicit --spring.cloud.dataflow.client.serverUri=http://dataflow-server:9393 to the definition.
 - Add RuntimeApplicationHelper#httpPost with http headers support.
 - Add CT: application/json to the test's http post.

Co-authored-by: Christian Tzolov <[email protected]>

Author: dturanski

spring-cloud-dataflow-server/src/test/java/org/springframework/cloud/dataflow/integration/test/DataFlowIT.java

@@ -30,6 +30,7 @@
 import java.util.Optional;
 import java.util.Random;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -83,7 +84,9 @@
 import org.springframework.context.annotation.Import;
 import org.springframework.core.io.DefaultResourceLoader;
 import org.springframework.hateoas.PagedModel;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
+import org.springframework.http.MediaType;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import org.springframework.util.StreamUtils;
 import org.springframework.web.util.UriComponentsBuilder;
@@ -753,6 +756,50 @@ public void namedChannelDirectedGraph() {
 		}
 	}
 
+	@Test
+	public void dataflowTaskLauncherSink() {
+		logger.info("dataflow-tasklauncher-sink-test");
+		String uri = String.format("docker:springcloud/spring-cloud-dataflow-tasklauncher-sink-kafka:%s",
+				testProperties.getDatabase().getDataflowVersion());
+		dataFlowOperations.appRegistryOperations()
+				.register("dataflowTaskLauncher", ApplicationType.sink, uri, null, true);
+
+
+		String taskName = randomTaskName();
+		try (Task task = Task.builder(dataFlowOperations)
+				.name(taskName)
+				.definition("timestamp")
+				.description("Test timestamp task")
+				.build()) {
+			try (Stream stream = Stream.builder(dataFlowOperations).name("tasklauncher-test")
+					.definition("http | dataflowTaskLauncher --trigger.initialDelay=100 --trigger.maxPeriod=1000 " +
+							"--spring.cloud.dataflow.client.serverUri=http://dataflow-server:9393")
+					.create()
+					.deploy(testDeploymentProperties())) {
+
+				Awaitility.await().until(() -> stream.getStatus().equals(DEPLOYED));
+
+				HttpHeaders headers = new HttpHeaders();
+				headers.setContentType(MediaType.APPLICATION_JSON);
+				headers.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
+
+				runtimeApps.httpPost(stream.getName(), "http", "{\"name\" : \"" + taskName + "\"}", headers);
+
+				AtomicLong launchId = new AtomicLong();
+				Awaitility.await().until(() -> task.executions().stream().filter(t ->
+						t.getTaskName().equals(taskName) && t.getTaskExecutionStatus() == TaskExecutionStatus.COMPLETE)
+						.findFirst()
+						.map(t -> launchId.getAndSet(t.getExecutionId())).isPresent()
+				);
+				long id = launchId.get();
+				assertThat(task.executions().size()).isEqualTo(1);
+				assertThat(task.execution(id).isPresent()).isTrue();
+				assertThat(task.execution(id).get().getExitCode()).isEqualTo(EXIT_CODE_SUCCESS);
+			}
+		}
+	}
+
+
 	// -----------------------------------------------------------------------
 	//                       STREAM  METRICS TESTS
 	// -----------------------------------------------------------------------

spring-cloud-dataflow-server/src/test/java/org/springframework/cloud/dataflow/integration/test/util/RuntimeApplicationHelper.java

@@ -32,6 +32,8 @@
 import org.springframework.cloud.dataflow.rest.resource.AppStatusResource;
 import org.springframework.cloud.dataflow.rest.resource.DetailedAppRegistrationResource;
 import org.springframework.cloud.skipper.domain.Deployer;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -259,6 +261,12 @@ public void httpPost(String url, String message) {
 		dataFlowTemplate.getRestTemplate().postForObject(url, message, String.class);
 	}
 
+	public void httpPost(String streamName, String appName, String message, HttpHeaders headers) {
+		String url = this.getApplicationInstanceUrl(streamName, appName);
+		HttpEntity<String> entity = new HttpEntity<>(message, headers);
+		dataFlowTemplate.getRestTemplate().postForEntity(url, entity, String.class);
+	}
+
 	public String httpGet(String url) {
 		return dataFlowTemplate.getRestTemplate().getForObject(url, String.class);
 	}