Add IT/AT for updating multi-instance apps streams

  Extend the lifecycle stream test to allow testing with multiple instances.

  Resolves #4504

Author: tzolov

spring-cloud-dataflow-rest-client/src/main/java/org/springframework/cloud/dataflow/rest/client/dsl/Stream.java

@@ -217,6 +217,10 @@ public String logs(StreamApplication app) {
         return this.client.streamOperations().streamExecutionLog(this.name, appDeploymentId);
     }
 
+	/**
+	 * @return Returns a map of the stream applications, associating every application with its applications instances
+	 * and their current runtime states: (App -> (AppInstanceId -> AppInstanceState)).
+	 */
 	public Map<StreamApplication, Map<String, String>> runtimeApps() {
 
 		StreamStatusResource streamStatus = client.runtimeOperations()

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

@@ -464,18 +464,36 @@ public void streamAppCrossVersion() {
 
 	@Test
 	public void streamLifecycle() {
+		streamLifecycleHelper(1, s -> { });
+	}
+
+	@Test
+	public void streamLifecycleWithTwoInstance() {
+		final int numberOfInstancePerApp = 2;
+		streamLifecycleHelper(numberOfInstancePerApp, stream -> {
+			Map<StreamApplication, Map<String, String>> streamApps = stream.runtimeApps();
+			assertThat(streamApps.size()).isEqualTo(2);
+			for (Map<String, String> instanceMap : streamApps.values()) {
+				assertThat(instanceMap.size()).isEqualTo(numberOfInstancePerApp); //every apps should have 2 instances.
+			}
+		});
+	}
+
+	private void streamLifecycleHelper(int appInstanceCount, Consumer<Stream> streamAssertions) {
 		logger.info("stream-lifecycle-test: DEPLOY");
 		try (Stream stream = Stream.builder(dataFlowOperations)
 				.name("lifecycle-test" + randomSuffix())
 				.definition("time | log --log.name='TEST' --log.expression='TICKTOCK - TIMESTAMP: '.concat(payload)")
 				.create()
-				.deploy(testDeploymentProperties())) {
-
-			assertThat(stream.getStatus()).is(
-					condition(status -> status.equals(DEPLOYING) || status.equals(PARTIAL)));
+				.deploy(new DeploymentPropertiesBuilder()
+						.putAll(testDeploymentProperties())
+						.put("deployer.*.count", "" + appInstanceCount)
+						.build())) {
 
 			Awaitility.await().until(() -> stream.getStatus().equals(DEPLOYED));
 
+			streamAssertions.accept(stream);
+
 			Awaitility.await().until(
 					() -> stream.logs(app("log")).contains("TICKTOCK - TIMESTAMP:"));
 
@@ -489,12 +507,13 @@ public void streamLifecycle() {
 			logger.info("stream-lifecycle-test: UPDATE");
 			stream.update(new DeploymentPropertiesBuilder()
 					.put("app.log.log.expression", "'Updated TICKTOCK - TIMESTAMP: '.concat(payload)")
-					// TODO investigate why on update the app-starters-core overrides the original web.exposure.include!!!
 					.put("app.*.management.endpoints.web.exposure.include", "*")
 					.build());
 
 			Awaitility.await().until(() -> stream.getStatus().equals(DEPLOYED));
 
+			streamAssertions.accept(stream);
+
 			Awaitility.await().until(
 					() -> stream.logs(app("log")).contains("Updated TICKTOCK - TIMESTAMP:"));
 
@@ -507,7 +526,8 @@ public void streamLifecycle() {
 			stream.rollback(0);
 
 			Awaitility.await().until(() -> stream.getStatus().equals(DEPLOYED));
-			assertThat(stream.getStatus()).isEqualTo(DEPLOYED);
+
+			streamAssertions.accept(stream);
 
 			Awaitility.await().until(
 					() -> stream.logs(app("log")).contains("TICKTOCK - TIMESTAMP:"));
@@ -522,7 +542,6 @@ public void streamLifecycle() {
 			stream.undeploy();
 
 			Awaitility.await().until(() -> stream.getStatus().equals(UNDEPLOYED));
-			assertThat(stream.getStatus()).isEqualTo(UNDEPLOYED);
 
 			assertThat(stream.history().size()).isEqualTo(3);
 			Awaitility.await().until(() -> stream.history().get(1).equals(DELETED));