Make field Macro

Author: ardian4

integration-test-remote/src/test/java/io/cdap/cdap/app/etl/gcp/GCSTest.java

@@ -79,6 +79,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
+import javax.annotation.Nullable;
 
 /**
  * Tests reading from GCS (Google Cloud Storage) and writing to GCS from within a Dataproc cluster.
@@ -811,9 +812,7 @@ public void testMultiSinkContentType() throws Exception {
     Schema outputSchema = Schema.recordOf("output.schema",
                                           Schema.Field.of("id", Schema.of(Schema.Type.INT)),
                                           Schema.Field.of("name", Schema.nullableOf(Schema.of(Schema.Type.STRING))),
-                                          Schema.Field.of("email", Schema.nullableOf(Schema.of(Schema.Type.STRING)))
-    );
-
+                                          Schema.Field.of("email", Schema.nullableOf(Schema.of(Schema.Type.STRING))));
 
     String line1 = "1,Marilyn Hawkins,[email protected],DepartmentA";
     String line2 = "2,Terry Perez,[email protected],DepartmentB";
@@ -822,7 +821,6 @@ public void testMultiSinkContentType() throws Exception {
 
     bucket.create(inputPath, String.join("\n", Arrays.asList(line1, line2, line3)).getBytes(StandardCharsets.UTF_8));
 
-
     Map<String, String> inputSourceConfig = new HashMap<>();
     inputSourceConfig.put("schema", schema.toString());
     inputSourceConfig.put("format", "${sourceFormat}");
@@ -838,7 +836,7 @@ public void testMultiSinkContentType() throws Exception {
     ETLBatchConfig.Builder pipelineConfig = ETLBatchConfig.builder().addStage(source);
 
     String path = createPath(bucket, OUTPUT_BLOB_NAME);
-    ETLStage sink = new ETLStage("multsink", createMultiSinkPlugin(path, schema, "departament"));
+    ETLStage sink = new ETLStage("multsink", createMultiSinkPlugin("csv"));
     pipelineConfig.addStage(sink).addConnection(source.getName(), sink.getName());
 
     AppRequest<ETLBatchConfig> appRequest = getBatchAppRequestV2(pipelineConfig.build());
@@ -851,7 +849,10 @@ public void testMultiSinkContentType() throws Exception {
     args.put(multisink1, outputSchema.toString());
     args.put(multisink2, outputSchema.toString());
     args.put("sourceFormat", "csv");
-    args.put("sinkFormat", "csv");
+    args.put("multiSinkPath", path);
+    args.put("multiSinkProjectId", getProjectId());
+    args.put("multiSinkSchema", schema.toString());
+    args.put("multiSinkSplitField", "departament");
     args.put("contentType", CSV_CONTENT_TYPE);
     startWorkFlow(appManager, ProgramRunStatus.COMPLETED, args);
 
@@ -879,7 +880,7 @@ private ETLPlugin createSinkPlugin(String format, String path, Schema schema) {
     return createSinkPlugin(format, path, schema, null);
   }
 
-  private ETLPlugin createSinkPlugin(String format, String path, Schema schema, String contentType) {
+  private ETLPlugin createSinkPlugin(String format, String path, Schema schema,@Nullable String contentType) {
     ImmutableMap.Builder<String, String> propertyBuilder = new ImmutableMap.Builder<String, String>()
       .put("path", path)
       .put("format", format)
@@ -892,14 +893,16 @@ private ETLPlugin createSinkPlugin(String format, String path, Schema schema, St
     return new ETLPlugin(SINK_PLUGIN_NAME, BatchSink.PLUGIN_TYPE, propertyBuilder.build(), GOOGLE_CLOUD_ARTIFACT);
   }
 
-  private ETLPlugin createMultiSinkPlugin(String path, Schema schema, String splitField) {
+  private ETLPlugin createMultiSinkPlugin(String sinkFormat) {
     Map<String, String> map = new HashMap<>();
-    map.put("path", path);
-    map.put("format", "${sinkFormat}");
-    map.put("project", getProjectId());
-    map.put("schema", schema.toString());
+    map.put("path", "${multiSinkPath}");
+    map.put("format", sinkFormat);
+    //todo make macro when https://cdap.atlassian.net/browse/PLUGIN-553 is fixed
+    //map.put("format", "${sinkFormat}");
+    map.put("project", "${multiSinkProjectId}");
+    map.put("schema", "${multiSinkSchema}");
     map.put("referenceName", "gcs-multi-input");
-    map.put("splitField", splitField);
+    map.put("splitField", "${multiSinkSplitField}");
     map.put("contentType", "${contentType}");
     return new ETLPlugin(MULTI_SINK_PLUGIN_NAME, BatchSink.PLUGIN_TYPE, map, GOOGLE_CLOUD_ARTIFACT);
   }