From 7f5c0dfb54309cd4383799139992898f4e8e76a1 Mon Sep 17 00:00:00 2001
From: Mark Payne <markap14@hotmail.com>
Date: Fri, 3 Mar 2017 13:18:41 -0500
Subject: [PATCH] NIFI-3553: If IOException is thrown when completing FTP
 transfer, do not update the FlowFile to point to the content that was
 streamed but instead leave FlowFile pointing to original version

Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com>

This closes #1562.
---
 .../apache/nifi/processors/standard/FetchFileTransfer.java  | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchFileTransfer.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchFileTransfer.java
index a7ae5ef059..6182b0aa84 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchFileTransfer.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchFileTransfer.java
@@ -239,9 +239,9 @@ public abstract class FetchFileTransfer extends AbstractProcessor {
                 @Override
                 public void process(final OutputStream out) throws IOException {
                     StreamUtils.copy(in, out);
+                    transfer.flush();
                 }
             });
-            transfer.flush();
             transferQueue.offer(new FileTransferIdleWrapper(transfer, System.nanoTime()));
         } catch (final FileNotFoundException e) {
             getLogger().error("Failed to fetch content for {} from filename {} on remote host {} because the file could not be found on the remote system; routing to {}",
@@ -255,14 +255,14 @@ public abstract class FetchFileTransfer extends AbstractProcessor {
             session.transfer(session.penalize(flowFile), REL_PERMISSION_DENIED);
             session.getProvenanceReporter().route(flowFile, REL_PERMISSION_DENIED);
             return;
-        } catch (final IOException e) {
+        } catch (final ProcessException | IOException e) {
             try {
                 transfer.close();
             } catch (final IOException e1) {
                 getLogger().warn("Failed to close connection to {}:{} due to {}", new Object[] {host, port, e.toString()}, e);
             }
 
-            getLogger().error("Failed to fetch content for {} from filename {} on remote host {}:{} due to {}; routing to failure",
+            getLogger().error("Failed to fetch content for {} from filename {} on remote host {}:{} due to {}; routing to comms.failure",
                 new Object[] {flowFile, filename, host, port, e.toString()}, e);
             session.transfer(session.penalize(flowFile), REL_COMMS_FAILURE);
             return;