Error handling deleted the original content & added new Processor to convert protobuf to Avro

README.md

@@ -45,8 +45,9 @@ and everything should be fine ! :)
 See the installation section to learn how to integrate this processor in Apache NiFi.
 This projects add 2 different new processors in NiFi:
 
-- `ProtobufDecoder`, which **decodes** a Protobuf-encoded payload to different kind of structured formats ;
-- `ProtobufEncoder`, which **encodes** a payload in a structured format using a Protobuf schema.
+- `DecodeProtobuf`, which **decodes** a Protobuf-encoded payload to different kind of structured formats.
+- `EncodeProtobuf`, which **encodes** a payload in a structured format using a Protobuf schema.
+- `ConvertProtobufToAvro` which decodes a Protobuf-encoded payload into Avro.
 
 ### Specifying the schema file
 In both processors, you have to specify a schema file to use for data encoding/decoding. You can do so either

pom.xml

@@ -20,24 +20,24 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compiler.source>1.7</maven.compiler.source>
         <maven.compiler.target>1.7</maven.compiler.target>
-        <nifi.version>1.4.0</nifi.version>
+        <nifi.version>1.16.3</nifi.version>
     </properties>
 
     <dependencies>
         <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
-            <version>3.5.0</version>
+            <version>3.9.0</version>
         </dependency>
         <dependency>
             <groupId>com.github.os72</groupId>
             <artifactId>protobuf-dynamic</artifactId>
-            <version>0.9.3</version>
+            <version>1.0.0</version>
         </dependency>
         <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java-util</artifactId>
-            <version>3.5.0</version>
+            <version>3.9.0</version>
         </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
@@ -52,7 +52,7 @@
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-processor-utils</artifactId>
-            <version>${nifi.version}</version>
+            <version>1.15.3</version>
         </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
@@ -66,19 +66,45 @@
             <version>4.12</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-record-serialization-service-api</artifactId>
+            <version>${nifi.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-record</artifactId>
+            <version>${nifi.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.github.os72</groupId>
             <artifactId>protoc-jar</artifactId>
             <version>3.5.0</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro-protobuf</artifactId>
+            <version>1.11.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+            <version>1.11.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-standard-services-api-nar</artifactId>
+            <version>${nifi.version}</version>
+            <type>nar</type>
+        </dependency>
     </dependencies>
 
     <build>
         <plugins>
             <plugin>
                 <groupId>org.apache.nifi</groupId>
                 <artifactId>nifi-nar-maven-plugin</artifactId>
-                <version>1.2.0</version>
+                <version>1.3.1</version>
                 <extensions>true</extensions>
             </plugin>
             <plugin>

src/main/java/com/github/whiver/nifi/mapper/JSONMapper.java

@@ -38,11 +38,15 @@ public class JSONMapper {
     /**
      * Format a Protocol Buffers Message to a JSON string
      * @param data  The Message to be formatted
+     * @param preserveFieldNames whether to preserve original field names
      * @return  A JSON String representing the data
      * @throws InvalidProtocolBufferException   Thrown in case of invalid Message data
      */
-    public static String toJSON(Message data) throws InvalidProtocolBufferException {
+    public static String toJSON(Message data, boolean preserveFieldNames) throws InvalidProtocolBufferException {
         JsonFormat.Printer printer = JsonFormat.printer();
+        if (preserveFieldNames) {
+            printer = printer.preservingProtoFieldNames();
+        }
         return printer.print(data);
     }
 

src/main/java/com/github/whiver/nifi/processor/ProtobufProcessor.java → src/main/java/com/github/whiver/nifi/processor/AbstractProtobufProcessor.java

@@ -31,8 +31,16 @@
 import com.github.whiver.nifi.exception.SchemaLoadingException;
 import com.github.whiver.nifi.parser.SchemaParser;
 import com.google.protobuf.Descriptors;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
 import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.ProcessorInitializationContext;
 import org.apache.nifi.processor.Relationship;
 import org.apache.nifi.processor.util.StandardValidators;
@@ -41,7 +49,16 @@
 import java.io.IOException;
 import java.util.*;
 
-public abstract class ProtobufProcessor extends AbstractProcessor {
+import static org.apache.nifi.components.Validator.VALID;
+
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@ReadsAttributes({
+        @ReadsAttribute(attribute = "protobuf.schemaPath", description = "will use this attribute as default for schema path."),
+        @ReadsAttribute(attribute = "protobuf.messageType", description = "will use this attribute as default for message type.")
+})
+@Tags({"Protobuf", "decoder", "Google Protocol Buffer"})
+public abstract class AbstractProtobufProcessor extends AbstractProcessor {
     /**
      * NiFi properties of the processor, that can be configured using the Web UI
      */
@@ -56,26 +73,39 @@ public abstract class ProtobufProcessor extends AbstractProcessor {
      * The compiled descriptor used to parse incoming binaries in case where the schema has been specified in the
      * processor level (protobuf.schema property)
      */
-    protected DynamicSchema schema;
-
-    /**
-     * Reflects the value of the COMPILE_SCHEMA property, so that it can be used by the onPropertyModified method
-     */
-    private boolean compileSchema;
-
+    protected static DynamicSchema schema = null;
 
     /*          PROPERTIES          */
 
     static final PropertyDescriptor PROTOBUF_SCHEMA = new PropertyDescriptor.Builder()
             .name("protobuf.schemaPath")
-            .displayName("Schema path")
+            .displayName("Schema Path")
             .required(false)
             .description("Path to the Protocol Buffers schema to use to encode or decode the data. If set, this schema will " +
                     "be used when the flowfile protobuf.schemaPath is missing.")
-            .expressionLanguageSupported(false)
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
             .addValidator(StandardValidators.createURLorFileValidator())
             .build();
 
+    static final PropertyDescriptor DEMARCATOR = new PropertyDescriptor.Builder()
+            .name("demarcator")
+            .displayName("Demarcator")
+            .required(false)
+            .description("This property is used to produce/consume messages separated by a demarcator")
+            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+            .addValidator(Validator.VALID)
+            .build();
+
+    static final PropertyDescriptor PROTOBUF_MESSAGE_TYPE = new PropertyDescriptor.Builder()
+            .name("protobuf.messageType")
+            .displayName("Message Type")
+            .required(false)
+            .description("Path to the Protocol Buffers message type to use to encode or decode the data. If set, this message type will " +
+                    "be used when the flowfile protobuf.messageType is missing.")
+            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+            .addValidator(VALID)
+            .build();
+
     static final PropertyDescriptor COMPILE_SCHEMA = new PropertyDescriptor.Builder()
             .name("protobuf.compileSchema")
             .displayName("Compile schema")
@@ -85,6 +115,17 @@ public abstract class ProtobufProcessor extends AbstractProcessor {
                     "processing the data. It is useful if the given schema file is in .proto format. Try to always use " +
                     "precompiled .desc schema whenever possible, since it is more performant.")
             .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .build();
+
+    static final PropertyDescriptor MAX_MESSAGE_SIZE = new PropertyDescriptor.Builder()
+            .name("max_message_size")
+            .displayName("Max Message Size")
+            .required(true)
+            .description("Maximum size of a single message when using a demarcator.")
+            .defaultValue("1 MB")
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
             .build();
 
 
@@ -106,9 +147,10 @@ public abstract class ProtobufProcessor extends AbstractProcessor {
             .build();
 
     @Override
-    public void init(final ProcessorInitializationContext context){
+    public void init(final ProcessorInitializationContext context) {
         List<PropertyDescriptor> properties = new ArrayList<>();
         properties.add(PROTOBUF_SCHEMA);
+        properties.add(PROTOBUF_MESSAGE_TYPE);
         properties.add(COMPILE_SCHEMA);
         this.properties = Collections.unmodifiableList(properties);
 
@@ -117,33 +159,25 @@ public void init(final ProcessorInitializationContext context){
         relationships.add(INVALID_SCHEMA);
         relationships.add(ERROR);
         this.relationships = Collections.unmodifiableSet(relationships);
-
-        this.compileSchema = false;
     }
 
+
+
     /**
      * Compile the given schema file when the protobuf.schemaPath property is given
      *
-     * @see AbstractProcessor
      */
-    @Override
-    public void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue) {
-        super.onPropertyModified(descriptor, oldValue, newValue);
-
-        if (descriptor == PROTOBUF_SCHEMA) {
-
-            // If the property is unset, just delete the existing descriptor
-            if (newValue == null || newValue.isEmpty()) {
-                this.schema = null;
-                return;
-            }
-
-            if (!newValue.equals(oldValue)) {
+    @OnScheduled
+    public void setUpSchema(final ProcessContext processContext) {
+        String protoSchemaPath = processContext.getProperty(PROTOBUF_SCHEMA).evaluateAttributeExpressions().getValue();
+        // If the property is unset, just delete the existing descriptor
+        if (protoSchemaPath == null || protoSchemaPath.isEmpty()) {
                 this.schema = null;
+            } else {
                 try {
-                    this.schema = SchemaParser.parseSchema(newValue, this.compileSchema);
+                    this.schema = SchemaParser.parseSchema(protoSchemaPath, processContext.getProperty(COMPILE_SCHEMA).evaluateAttributeExpressions().asBoolean());
                 } catch (FileNotFoundException e) {
-                    getLogger().error("File " + newValue + " not found on the disk.", e);
+                    getLogger().error("File " + protoSchemaPath + " not found on the disk.", e);
                 } catch (Descriptors.DescriptorValidationException e) {
                     getLogger().error("Invalid schema file: " + e.getMessage(), e);
                 } catch (IOException e) {
@@ -154,9 +188,6 @@ public void onPropertyModified(PropertyDescriptor descriptor, String oldValue, S
                     getLogger().error("Unable to compile schema: " + e.getMessage(), e);
                 }
             }
-        } else if (descriptor == COMPILE_SCHEMA) {
-            this.compileSchema = Boolean.parseBoolean(newValue);
-        }
     }
 
     @Override