Fixed: Show new error message when output schema is not maching input schema

Author: vjosa@cirus.com

pom.xml

@@ -29,7 +29,7 @@
     <data.pipeline.parent>system:cdap-data-pipeline[6.0.0-SNAPSHOT,7.0.0-SNAPSHOT)</data.pipeline.parent>
     <data.stream.parent>system:cdap-data-streams[6.0.0-SNAPSHOT,7.0.0-SNAPSHOT)</data.stream.parent>
 
-    <cdap.version>6.2.0-SNAPSHOT</cdap.version>
+    <cdap.version>6.2.0</cdap.version>
     <spark.version>2.1.3</spark.version>
     <logback.version>1.0.9</logback.version>
 

src/main/java/io/cdap/plugin/spark/dynamic/ScalaSparkCompute.java

@@ -24,6 +24,7 @@
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.api.plugin.PluginConfig;
 import io.cdap.cdap.api.spark.sql.DataFrames;
+import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.StageConfigurer;
 import io.cdap.cdap.etl.api.batch.SparkCompute;
@@ -110,6 +111,13 @@ public JavaRDD<StructuredRecord> transform(SparkExecutionPluginContext context,
       // If there is no output schema configured, derive it from the DataFrame
       // Otherwise, assume the DataFrame has the correct schema already
       outputSchema = DataFrames.toSchema((DataType) invokeDataFrameMethod(result, "schema"));
+    } else {
+      Schema dataSchema = DataFrames.toSchema((DataType) invokeDataFrameMethod(result, "schema"));
+      if (!dataSchema.isCompatible(outputSchema)) {
+        FailureCollector collector = context.getFailureCollector();
+        collector.addFailure("Schema mismatch.", "Output schema is not matching input schema.");
+        collector.getOrThrowException();
+      }
     }
     //noinspection unchecked
     return ((JavaRDD<Row>) invokeDataFrameMethod(result, "toJavaRDD")).map(new RowToRecord(outputSchema));