Add deploy props param to "create stream" REST API (#5143)

Fixes #5043

Author: onobc

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/StreamDefinitionController.java

@@ -20,6 +20,7 @@
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -44,8 +45,10 @@
 import org.springframework.hateoas.server.ExposesResourceFor;
 import org.springframework.hateoas.server.RepresentationModelAssembler;
 import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
 import org.springframework.util.Assert;
 import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RequestParam;
@@ -129,7 +132,10 @@ public PagedModel<? extends StreamDefinitionResource> list(Pageable pageable,
 	}
 
 	/**
-	 * Create a new stream.
+	 * Create a new stream and optionally deploy it.
+	 * <p>
+	 * Differs from {@link #saveWithDeployProps} by accepting deployment properties and consuming
+	 * {@link MediaType#APPLICATION_FORM_URLENCODED} request content (required by the Dataflow Shell).
 	 *
 	 * @param name stream name
 	 * @param dsl DSL definition for stream
@@ -139,15 +145,46 @@ public PagedModel<? extends StreamDefinitionResource> list(Pageable pageable,
 	 * @return the created stream definition
 	 * @throws DuplicateStreamDefinitionException if a stream definition with the same name
 	 * already exists
-	 * @throws InvalidStreamDefinitionException if there errors in parsing the stream DSL,
+	 * @throws InvalidStreamDefinitionException if there are errors parsing the stream DSL,
 	 * resolving the name, or type of applications in the stream
 	 */
-	@RequestMapping(value = "", method = RequestMethod.POST)
+	@RequestMapping(value = "", method = RequestMethod.POST, consumes = MediaType.APPLICATION_FORM_URLENCODED_VALUE)
 	@ResponseStatus(HttpStatus.CREATED)
-	public StreamDefinitionResource save(@RequestParam("name") String name, @RequestParam("definition") String dsl,
-											@RequestParam(value = "description", defaultValue = "") String description,
-											@RequestParam(value = "deploy", defaultValue = "false") boolean deploy) {
-		StreamDefinition streamDefinition = this.streamService.createStream(name, dsl, description, deploy);
+	public StreamDefinitionResource save(@RequestParam("name") String name,
+										 @RequestParam("definition") String dsl,
+										 @RequestParam(value = "description", defaultValue = "") String description,
+										 @RequestParam(value = "deploy", defaultValue = "false") boolean deploy) {
+		StreamDefinition streamDefinition = this.streamService.createStream(name, dsl, description, deploy, null);
+		return ((RepresentationModelAssembler<StreamDefinition, ? extends StreamDefinitionResource>)
+				this.streamDefinitionAssemblerProvider.getStreamDefinitionAssembler(Collections.singletonList(streamDefinition))).toModel(streamDefinition);
+	}
+
+	/**
+	 * Create a new stream and optionally deploy it.
+	 * <p>
+	 * Differs from {@link #save} by accepting deployment properties and consuming
+	 * {@link MediaType#APPLICATION_JSON} request content.
+	 *
+	 * @param name stream name
+	 * @param dsl DSL definition for stream
+	 * @param deploy if {@code true}, the stream is deployed upon creation (default is
+	 * {@code false})
+	 * @param deploymentProperties the optional deployment properties to use when the stream is deployed upon creation
+	 * @param description description of the stream definition
+	 * @return the created stream definition
+	 * @throws DuplicateStreamDefinitionException if a stream definition with the same name
+	 * already exists
+	 * @throws InvalidStreamDefinitionException if there are errors parsing the stream DSL,
+	 * resolving the name, or type of applications in the stream
+	 */
+	@RequestMapping(value = "", method = RequestMethod.POST, consumes = MediaType.APPLICATION_JSON_VALUE)
+	@ResponseStatus(HttpStatus.CREATED)
+	public StreamDefinitionResource saveWithDeployProps(@RequestParam("name") String name,
+										 @RequestParam("definition") String dsl,
+										 @RequestParam(value = "description", defaultValue = "") String description,
+										 @RequestParam(value = "deploy", defaultValue = "false") boolean deploy,
+										 @RequestBody(required = false) Map<String, String> deploymentProperties) {
+		StreamDefinition streamDefinition = this.streamService.createStream(name, dsl, description, deploy, deploymentProperties);
 		return ((RepresentationModelAssembler<StreamDefinition, ? extends StreamDefinitionResource>)
 				this.streamDefinitionAssemblerProvider.getStreamDefinitionAssembler(Collections.singletonList(streamDefinition))).toModel(streamDefinition);
 	}

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

@@ -96,11 +96,13 @@ public interface StreamService {
 	 * @param description description of the stream definition
 	 * @param deploy if {@code true}, the stream is deployed upon creation (default is
 	 * {@code false})
+	 * @param deploymentProperties the optional deployment properties to use when the stream is deployed upon creation
 	 * @return the created stream definition already exists
 	 * @throws InvalidStreamDefinitionException if there are errors in parsing the stream DSL,
 	 * resolving the name, or type of applications in the stream
 	 */
-	StreamDefinition createStream(String streamName, String dsl, String description, boolean deploy);
+	StreamDefinition createStream(String streamName, String dsl, String description, boolean deploy,
+								  Map<String, String> deploymentProperties);
 
 	/**
 	 * Deploys the stream with the user provided deployment properties.

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

@@ -393,19 +393,9 @@ public StreamDeployment info(String streamName) {
 		return this.skipperStreamDeployer.getStreamInfo(streamName);
 	}
 
-	/**
-	 * Create a new stream.
-	 *
-	 * @param streamName stream name
-	 * @param dsl DSL definition for stream
-	 * @param description description of the stream definition
-	 * @param deploy if {@code true}, the stream is deployed upon creation (default is
-	 *     {@code false})
-	 * @return the created stream definition already exists
-	 * @throws InvalidStreamDefinitionException if there are errors in parsing the stream DSL,
-	 *     resolving the name, or type of applications in the stream
-	 */
-	public StreamDefinition createStream(String streamName, String dsl, String description, boolean deploy) {
+	@Override
+	public StreamDefinition createStream(String streamName, String dsl, String description, boolean deploy,
+										 Map<String, String> deploymentProperties) {
 		StreamDefinition streamDefinition = createStreamDefinition(streamName, dsl, description);
 		List<String> errorMessages = new ArrayList<>();
 
@@ -432,7 +422,7 @@ public StreamDefinition createStream(String streamName, String dsl, String descr
 		final StreamDefinition savedStreamDefinition = this.streamDefinitionRepository.save(streamDefinition);
 
 		if (deploy) {
-			this.deployStream(streamName, new HashMap<>());
+			this.deployStream(streamName, deploymentProperties);
 		}
 
 		auditRecordService.populateAndSaveAuditRecord(

spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/StreamControllerTests.java

@@ -24,15 +24,18 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Stream;
 
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.MappingIterator;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLMapper;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.ArgumentCaptor;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -74,13 +77,16 @@
 import org.springframework.http.MediaType;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.annotation.DirtiesContext.ClassMode;
-import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder;
 import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
 import org.springframework.web.client.RestClientException;
 import org.springframework.web.context.WebApplicationContext;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.startsWith;
@@ -110,7 +116,6 @@
  * @author Christian Tzolov
  * @author Daniel Serleg
  */
-@RunWith(SpringRunner.class)
 @SpringBootTest(classes = TestDependencies.class)
 @DirtiesContext(classMode = ClassMode.BEFORE_EACH_TEST_METHOD)
 @AutoConfigureTestDatabase(replace = Replace.ANY)
@@ -138,7 +143,7 @@ public class StreamControllerTests {
 	@Autowired
 	private StreamDefinitionService streamDefinitionService;
 
-	@Before
+	@BeforeEach
 	public void setupMocks() {
 		this.mockMvc = MockMvcBuilders.webAppContextSetup(wac)
 				.defaultRequest(get("/").accept(MediaType.APPLICATION_JSON)).build();
@@ -156,24 +161,51 @@ public void setupMocks() {
 		when(skipperClient.search(anyString(), eq(false))).thenReturn(new ArrayList<PackageMetadata>());
 	}
 
-	@After
+	@AfterEach
 	public void tearDown() {
 		repository.deleteAll();
 		auditRecordRepository.deleteAll();
 		assertThat(repository.count()).isZero();
 		assertThat(auditRecordRepository.count()).isZero();
 	}
 
-	@Test(expected = IllegalArgumentException.class)
+	@Test
 	public void testConstructorMissingStreamService() {
-		new StreamDefinitionController(null, null, null, null, null);
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> {
+					new StreamDefinitionController(null, null, null, null, null);
+				});
 	}
 
 	@Test
-	public void testSave() throws Exception {
+	public void testSaveNoDeployJsonEncoded() throws Exception {
 		assertThat(repository.count()).isZero();
-		mockMvc.perform(post("/streams/definitions/").param("name", "myStream").param("definition", "time | log")
-				.accept(MediaType.APPLICATION_JSON)).andDo(print()).andExpect(status().isCreated());
+		mockMvc.perform(post("/streams/definitions/")
+						.param("name", "myStream")
+						.param("definition", "time | log")
+						.contentType(MediaType.APPLICATION_JSON)
+						.accept(MediaType.APPLICATION_JSON))
+				.andDo(print())
+				.andExpect(status().isCreated());
+		assertThatStreamSavedWithoutDeploy();
+	}
+
+	@Test
+	public void testSaveNoDeployFormEncoded() throws Exception {
+		assertThat(repository.count()).isZero();
+		MultiValueMap<String, String> values = new LinkedMultiValueMap<>();
+		values.add("name", "myStream");
+		values.add("definition", "time | log");
+		mockMvc.perform(post("/streams/definitions/")
+						.params(values)
+						.contentType(MediaType.APPLICATION_FORM_URLENCODED)
+						.accept(MediaType.APPLICATION_JSON))
+				.andDo(print())
+				.andExpect(status().isCreated());
+		assertThatStreamSavedWithoutDeploy();
+	}
+
+	private void assertThatStreamSavedWithoutDeploy() {
 		assertThat(repository.count()).isEqualTo(1);
 		StreamDefinition myStream = repository.findById("myStream").get();
 		assertThat(myStream.getDslText()).isEqualTo("time | log");
@@ -189,6 +221,60 @@ public void testSave() throws Exception {
 		assertThat(logDefinition.getProperties().get(BindingPropertyKeys.INPUT_GROUP)).isEqualTo("myStream");
 	}
 
+	@ParameterizedTest
+	@MethodSource("testSaveAndDeployWithDeployPropsProvider")
+	public void testSaveAndDeploy(Map<String, String> deploymentProps, Map<String, String> expectedPropsOnApps) throws Exception {
+		assertThat(repository.count()).isZero();
+		String definition = "time | log";
+		String streamName = "testSaveAndDeploy-stream";
+		MockHttpServletRequestBuilder requestBuilder = post("/streams/definitions/")
+				.param("name", streamName)
+				.param("definition", definition)
+				.param("deploy", "true")
+				.accept(MediaType.APPLICATION_JSON);
+		if (deploymentProps != null) {
+			requestBuilder
+					.content(new ObjectMapper().writeValueAsBytes(deploymentProps))
+					.contentType(MediaType.APPLICATION_JSON);
+		}
+		mockMvc.perform(requestBuilder)
+				.andDo(print())
+				.andExpect(status().isCreated());
+
+		assertThat(repository.count()).isEqualTo(1);
+		assertThat(repository.findById(streamName)).isPresent()
+				.hasValueSatisfying((sd) -> {
+					assertThat(sd.getName()).isEqualTo(streamName);
+					assertThat(sd.getDslText()).isEqualTo(definition);
+				})
+				.map(this.streamDefinitionService::getAppDefinitions).get().asList()
+				.extracting("name").containsExactly("time", "log");
+
+		ArgumentCaptor<UploadRequest> uploadRequestCaptor = ArgumentCaptor.forClass(UploadRequest.class);
+		verify(skipperClient, times(1)).upload(uploadRequestCaptor.capture());
+
+		List<UploadRequest> uploadRequests = uploadRequestCaptor.getAllValues();
+		assertThat(uploadRequests).hasSize(1);
+		Package pkg = SkipperPackageUtils.loadPackageFromBytes(uploadRequestCaptor);
+
+		Package timePackage = findChildPackageByName(pkg, "time");
+		assertThat(timePackage).isNotNull();
+		SpringCloudDeployerApplicationSpec timeSpec = parseSpec(timePackage.getConfigValues().getRaw());
+		assertThat(timeSpec.getDeploymentProperties()).containsAllEntriesOf(expectedPropsOnApps);
+
+		Package logPackage = findChildPackageByName(pkg, "log");
+		assertThat(logPackage).isNotNull();
+		SpringCloudDeployerApplicationSpec logAppSpec = parseSpec(logPackage.getConfigValues().getRaw());
+		assertThat(logAppSpec.getDeploymentProperties()).containsAllEntriesOf(expectedPropsOnApps);
+	}
+
+	private static Stream<Arguments> testSaveAndDeployWithDeployPropsProvider() {
+		return Stream.of(
+				Arguments.of(Collections.singletonMap("deployer.*.count", "2"), Collections.singletonMap("spring.cloud.deployer.count", "2")),
+				Arguments.of(Collections.emptyMap(), Collections.emptyMap()),
+				Arguments.of(null, Collections.emptyMap()));
+	}
+
 	@Test
 	public void testSaveWithSensitiveProperties() throws Exception {
 		assertThat(repository.count()).isZero();

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

@@ -127,7 +127,7 @@ public void createStream() {
 		final StreamDefinition expectedStreamDefinition = new StreamDefinition("testStream", "time | log");
 		when(streamDefinitionRepository.save(expectedStreamDefinition)).thenReturn(expectedStreamDefinition);
 
-		this.defaultStreamService.createStream("testStream", "time | log", "demo stream", false);
+		this.defaultStreamService.createStream("testStream", "time | log", "demo stream", false, null);
 
 		verify(this.streamValidationService).isRegistered("time", ApplicationType.source);
 		verify(this.streamValidationService).isRegistered("log", ApplicationType.sink);
@@ -150,7 +150,7 @@ public void createStreamWithMissingApps() {
 		thrown.expectMessage("Application name 'time' with type 'source' does not exist in the app registry.\n" +
 				"Application name 'log' with type 'sink' does not exist in the app registry.");
 
-		this.defaultStreamService.createStream("testStream", "time | log", "demo stream", false);
+		this.defaultStreamService.createStream("testStream", "time | log", "demo stream", false, null);
 	}
 
 	@Test
@@ -161,7 +161,7 @@ public void createStreamInvalidDsl() {
 		thrown.expect(InvalidStreamDefinitionException.class);
 		thrown.expectMessage("Application name 'koza' with type 'app' does not exist in the app registry.");
 
-		this.defaultStreamService.createStream("testStream", "koza", "demo stream", false);
+		this.defaultStreamService.createStream("testStream", "koza", "demo stream", false, null);
 	}
 
 	@Test