NIFI-9259 Rebase and transfer original flowfiles to original relation…

…ship for every routing strategy
apache · Nov 29, 2021 · c63b7e2 · c63b7e2
1 parent 59c41df
commit c63b7e2
Show file tree

Hide file tree

Showing 6 changed files with 20 additions and 17 deletions.
diff --git a/nifi-assembly/pom.xml b/nifi-assembly/pom.xml
@@ -407,7 +407,7 @@ language governing permissions and limitations under the License. -->
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-geohash-nar</artifactId>
-            <version>1.15.0-SNAPSHOT</version>
+            <version>1.16.0-SNAPSHOT</version>
             <type>nar</type>
         </dependency>
         <dependency>

diff --git a/nifi-nar-bundles/nifi-geohash-bundle/nifi-geohash-nar/pom.xml b/nifi-nar-bundles/nifi-geohash-bundle/nifi-geohash-nar/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.nifi</groupId>
         <artifactId>nifi-geohash-bundle</artifactId>
-        <version>1.15.0-SNAPSHOT</version>
+        <version>1.16.0-SNAPSHOT</version>
     </parent>
 
     <artifactId>nifi-geohash-nar</artifactId>
@@ -39,7 +39,7 @@
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-geohash-processors</artifactId>
-            <version>1.15.0-SNAPSHOT</version>
+            <version>1.16.0-SNAPSHOT</version>
         </dependency>
     </dependencies>
 

diff --git a/nifi-nar-bundles/nifi-geohash-bundle/nifi-geohash-processors/pom.xml b/nifi-nar-bundles/nifi-geohash-bundle/nifi-geohash-processors/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.nifi</groupId>
         <artifactId>nifi-geohash-bundle</artifactId>
-        <version>1.15.0-SNAPSHOT</version>
+        <version>1.16.0-SNAPSHOT</version>
     </parent>
 
     <artifactId>nifi-geohash-processors</artifactId>
@@ -33,7 +33,7 @@
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-utils</artifactId>
-            <version>1.15.0-SNAPSHOT</version>
+            <version>1.16.0-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>ch.hsr</groupId>
@@ -48,7 +48,7 @@
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-record-path</artifactId>
-            <version>1.15.0-SNAPSHOT</version>
+            <version>1.16.0-SNAPSHOT</version>
             <scope>compile</scope>
         </dependency>
         <dependency>
@@ -60,19 +60,19 @@
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-mock</artifactId>
-            <version>1.15.0-SNAPSHOT</version>
+            <version>1.16.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-mock-record-utils</artifactId>
-            <version>1.15.0-SNAPSHOT</version>
+            <version>1.16.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-record-serialization-services</artifactId>
-            <version>1.15.0-SNAPSHOT</version>
+            <version>1.16.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
         <dependency>

diff --git a/...fi-geohash-processors/src/main/java/org/apache/nifi/processors/geohash/GeohashRecord.java b/...fi-geohash-processors/src/main/java/org/apache/nifi/processors/geohash/GeohashRecord.java
@@ -334,17 +334,20 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
             output = session.putAllAttributes(output, buildAttributes(foundCount, writer.getMimeType(), writeResult));
             if (routingStrategy != RoutingStrategy.SPLIT) {
                 session.transfer(output, targetRelationship);
-                session.remove(input);
             } else {
                 if (notFoundCount > 0) {
                     notFound = session.putAllAttributes(notFound, buildAttributes(notFoundCount, writer.getMimeType(), notFoundWriterResult));
                     session.transfer(notFound, REL_FAILURE);
                 } else {
                     session.remove(notFound);
                 }
-                session.transfer(output, REL_SUCCESS);
-                session.transfer(input, REL_ORIGINAL);
+                if (foundCount > 0) {
+                    session.transfer(output, REL_SUCCESS);
+                } else {
+                    session.remove(output);
+                }
             }
+            session.transfer(input, REL_ORIGINAL);
 
         } catch (Exception ex) {
             getLogger().error("Failed to {} due to {}", encode ? "encode" : "decode", ex.getLocalizedMessage(), ex);

diff --git a/...eohash-processors/src/test/java/org/apache/nifi/processors/geohash/GeohashRecordTest.java b/...eohash-processors/src/test/java/org/apache/nifi/processors/geohash/GeohashRecordTest.java
@@ -98,7 +98,7 @@ public void testEncodeSendToSuccessDefault() throws Exception {
         runner.setProperty(GeohashRecord.ROUTING_STRATEGY, GeohashRecord.RoutingStrategy.SKIP_UNENRICHED.toString());
         runner.assertValid();
 
-        assertTransfers("/record_encode.json", 0, 1, 0);
+        assertTransfers("/record_encode.json", 0, 1, 1);
 
         MockFlowFile ff = runner.getFlowFilesForRelationship(GeohashRecord.REL_SUCCESS).get(0);
         byte[] raw = runner.getContentAsByteArray(ff);
@@ -120,7 +120,7 @@ public void testEncodeSendToFailureDefault() {
         runner.setProperty(GeohashRecord.ROUTING_STRATEGY, GeohashRecord.RoutingStrategy.SKIP_UNENRICHED.toString());
         runner.assertValid();
 
-        assertTransfers("/record_decode.json", 1, 0, 0);
+        assertTransfers("/record_decode.json", 1, 0, 1);
     }
 
     @Test
@@ -138,7 +138,7 @@ public void testRequireAllEnrichedSendToFailure() {
         runner.setProperty(GeohashRecord.ROUTING_STRATEGY, GeohashRecord.RoutingStrategy.REQUIRE_ALL_ENRICHED.toString());
         runner.assertValid();
 
-        assertTransfers("/record_encode.json", 1, 0, 0);
+        assertTransfers("/record_encode.json", 1, 0, 1);
     }
 
     @Test
@@ -147,6 +147,6 @@ public void testRequireAllEnrichedSendToSuccess() {
         runner.setProperty(GeohashRecord.ROUTING_STRATEGY, GeohashRecord.RoutingStrategy.REQUIRE_ALL_ENRICHED.toString());
         runner.assertValid();
 
-        assertTransfers("/record_decode.json", 0, 1, 0);
+        assertTransfers("/record_decode.json", 0, 1, 1);
     }
 }
diff --git a/nifi-nar-bundles/nifi-geohash-bundle/pom.xml b/nifi-nar-bundles/nifi-geohash-bundle/pom.xml
@@ -19,7 +19,7 @@
     <parent>
         <groupId>org.apache.nifi</groupId>
         <artifactId>nifi-nar-bundles</artifactId>
-        <version>1.15.0-SNAPSHOT</version>
+        <version>1.16.0-SNAPSHOT</version>
     </parent>
 
     <artifactId>nifi-geohash-bundle</artifactId>