Perform some renamings

Author: dardanxh

src/main/java/io/cdap/plugin/jms/common/SchemaValidationUtils.java

@@ -208,7 +208,7 @@ public static void validateObjectMessageSchema(Schema schema, FailureCollector c
    * @param schema the user defined schema
    * @param collector the failure collector
    */
-  public static void validateBytesMessageInputSchema(Schema schema, FailureCollector collector) {
+  public static void validateBytesMessageSchema(Schema schema, FailureCollector collector) {
     Schema.Type type = schema.getField(JMSMessageParts.BODY).getSchema().getType();
     boolean isTypeString = type.equals(Schema.Type.STRING);
     boolean isTypeRecord = type.equals(Schema.Type.RECORD);

src/main/java/io/cdap/plugin/jms/sink/JMSBatchSink.java

@@ -39,7 +39,7 @@
  */
 @Plugin(type = BatchSink.PLUGIN_TYPE)
 @Name("JMS")
-@Description("JMS sink to write events to JMS")
+@Description("JMSSink")
 public class JMSBatchSink extends ReferenceBatchSink<StructuredRecord, NullWritable, StructuredRecord> {
 
   private final JMSBatchSinkConfig config;

src/main/java/io/cdap/plugin/jms/sink/JMSBatchSinkConfig.java

@@ -24,15 +24,13 @@
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.plugin.jms.common.JMSConfig;
 import io.cdap.plugin.jms.common.JMSMessageType;
-import io.cdap.plugin.jms.source.JMSSourceUtils;
 
 import java.io.IOException;
 import java.io.Serializable;
 import javax.annotation.Nullable;
 
 /**
- * Holds configuration required for configuring {@link JMSSourceUtils ;} and
- * {@link JMSBatchSink}.
+ * Holds the necessary configurations for the JMS source plugin
  */
 public class JMSBatchSinkConfig extends JMSConfig implements Serializable {
 

src/main/java/io/cdap/plugin/jms/source/JMSSourceUtils.java

@@ -27,17 +27,19 @@
 import io.cdap.cdap.etl.api.streaming.StreamingSource;
 import io.cdap.cdap.etl.api.streaming.StreamingSourceContext;
 import io.cdap.plugin.common.LineageRecorder;
+import org.apache.spark.storage.StorageLevel;
 import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.receiver.Receiver;
 
 import java.util.stream.Collectors;
 
 /**
- * This class <code>JMSStreamingSource</code> is a plugin that allows consuming messages from a specified JMS
- * Queue/Topic and generate StructuredRecords out of them.
+ * This class is a plugin that allows consuming messages from a specified JMS Queue/Topic and generate
+ * StructuredRecords out of them.
  */
 @Plugin(type = StreamingSource.PLUGIN_TYPE)
 @Name("JMS")
-@Description("JMS (Java Messaging Service) Source")
+@Description("JMSSource")
 public class JMSStreamingSource extends ReferenceStreamingSource<StructuredRecord> {
 
   private JMSStreamingSourceConfig config;
@@ -57,7 +59,6 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
   @Override
   public void prepareRun(StreamingSourceContext context) throws Exception {
     Schema schema = config.getSchema();
-    // record dataset lineage
     context.registerLineage(config.referenceName, schema);
 
     if (schema.getFields() != null) {
@@ -72,6 +73,12 @@ public JavaDStream<StructuredRecord> getStream(StreamingContext context) throws
     FailureCollector collector = context.getFailureCollector();
     config.validate(collector);
     collector.getOrThrowException();
-    return JMSSourceUtils.getJavaDStream(context, config);
+    return getJavaDStream(context, config);
+  }
+
+  private static JavaDStream<StructuredRecord> getJavaDStream(StreamingContext context,
+                                                              JMSStreamingSourceConfig config) {
+    Receiver<StructuredRecord> jmsReceiver = new JMSReceiver(StorageLevel.MEMORY_AND_DISK_SER_2(), config);
+    return context.getSparkStreamingContext().receiverStream(jmsReceiver);
   }
 }

src/main/java/io/cdap/plugin/jms/source/JMSStreamingSource.java

@@ -27,11 +27,11 @@
 import io.cdap.plugin.jms.common.JMSMessageHeader;
 import io.cdap.plugin.jms.common.JMSMessageParts;
 import io.cdap.plugin.jms.common.JMSMessageType;
+import io.cdap.plugin.jms.common.SchemaValidationUtils;
 
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
@@ -113,52 +113,46 @@ public void validate(FailureCollector failureCollector) {
     }
 
     if (!containsMacro(NAME_SCHEMA)) {
-
       Schema schema = getSchema();
 
-      boolean otherFieldsExist = schema
-        .getFields()
-        .stream()
-        .anyMatch(field -> !JMSMessageParts.getJMSMessageParts().contains(field.getName()));
-
-      if (otherFieldsExist) {
-        failureCollector
-          .addFailure("New fields detected in the output schema!",
-                      String.format("Only \"%s\", \"%s\" and \"%s\" fields are allowed.",
-                                    JMSMessageParts.HEADER, JMSMessageParts.BODY, JMSMessageParts.PROPERTIES))
-          .withConfigProperty(NAME_SCHEMA);
+      SchemaValidationUtils.validateIfAnyNotSupportedRootFieldExists(schema, failureCollector);
+
+      if (getMessageHeader()) {
+        SchemaValidationUtils.validateHeaderSchema(schema, failureCollector);
+      }
+
+      if (getMessageProperties()) {
+        SchemaValidationUtils.validatePropertiesSchema(schema, failureCollector);
       }
 
       switch (messageType) {
         case JMSMessageType.TEXT:
-          if (!schema.getField(JMSMessageParts.BODY).getSchema().getType().equals(Schema.Type.STRING)) {
-            failureCollector
-              .addFailure(String.format("Wrong data type for field \"%s\".", JMSMessageParts.BODY),
-                          String.format("For JMS %s type of message, \"%s\" must be of String data type.",
-                                        messageType, JMSMessageParts.BODY)).withConfigProperty(NAME_SCHEMA);
-          }
+          SchemaValidationUtils.validateIfBodyNotInSchema(schema, failureCollector);
+          SchemaValidationUtils.validateTextMessageSchema(schema, failureCollector);
+
           break;
 
         case JMSMessageType.OBJECT:
-          if (!schema.getField(JMSMessageParts.BODY).getSchema().getType().equals(Schema.Type.ARRAY)) {
-            failureCollector
-              .addFailure(String.format("Wrong data type for field \"%s\".", JMSMessageParts.BODY),
-                          String.format("For JMS %s type of message, \"%s\" must be of Array[Bytes] data type.",
-                                        messageType, JMSMessageParts.BODY)).withConfigProperty(NAME_SCHEMA);
-          }
+          SchemaValidationUtils.validateIfBodyNotInSchema(schema, failureCollector);
+          SchemaValidationUtils.validateObjectMessageSchema(schema, failureCollector);
           break;
 
         case JMSMessageType.BYTES:
+          SchemaValidationUtils.validateIfBodyNotInSchema(schema, failureCollector);
+          SchemaValidationUtils.validateBytesMessageSchema(schema, failureCollector);
+
+          break;
+
         case JMSMessageType.MAP:
-        case JMSMessageType.MESSAGE:
-          if (!Arrays.asList(Schema.Type.STRING, Schema.Type.RECORD)
-            .contains(schema.getField(JMSMessageParts.BODY).getSchema().getType())) {
-            failureCollector
-              .addFailure(String.format("Wrong data type for field \"%s\"", JMSMessageParts.BODY),
-                          String.format("For JMS %s type of message, \"%s\" must be of String or Record data type.",
-                                        messageType, JMSMessageParts.BODY)).withConfigProperty(NAME_SCHEMA);
-          }
+          SchemaValidationUtils.validateIfBodyNotInSchema(schema, failureCollector);
+          SchemaValidationUtils.validateMapMessageSchema(schema, failureCollector);
+
           break;
+
+        case JMSMessageType.MESSAGE:
+          SchemaValidationUtils.validateMessageSchema(schema, failureCollector);
+
+
       }
     }
   }

src/main/java/io/cdap/plugin/jms/source/JMSStreamingSourceConfig.java

@@ -172,7 +172,7 @@ public void validateByteMessageSchema_WithNonStringOrRecordBodyRootField_ShouldT
     Schema schema = Schema
       .recordOf("record", Schema.Field.of(JMSMessageParts.BODY, Schema.of(Schema.Type.INT)));
 
-    SchemaValidationUtils.validateBytesMessageInputSchema(schema, null);
+    SchemaValidationUtils.validateBytesMessageSchema(schema, null);
   }
 
   @Test