Merge branch 'develop' of https://git-wip-us.apache.org/repos/asf/incubator-nifi into develop

nifi/nifi-api/src/main/java/org/apache/nifi/controller/FlowFileQueue.java

@@ -100,6 +100,15 @@ public interface FlowFileQueue {
 
     QueueSize getActiveQueueSize();
 
+    /**
+     * Returns a QueueSize that represents all FlowFiles that are 'unacknowledged'. A FlowFile
+     * is considered to be unacknowledged if it has been pulled from the queue by some component
+     * but the session that pulled the FlowFile has not yet been committed or rolled back.
+     *
+     * @return a QueueSize that represents all FlowFiles that are 'unacknowledged'.
+     */
+    QueueSize getUnacknowledgedQueueSize();
+
     void acknowledge(FlowFileRecord flowFile);
 
     void acknowledge(Collection<FlowFileRecord> flowFiles);

nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/StandardFlowFileQueue.java

@@ -961,6 +961,11 @@ public final class StandardFlowFileQueue implements FlowFileQueue {
         writeLock.unlock("external unlock");
     }
 
+    @Override
+    public QueueSize getUnacknowledgedQueueSize() {
+        return unacknowledgedSizeRef.get();
+    }
+
     private void updateUnacknowledgedSize(final int addToCount, final long addToSize) {
         boolean updated = false;
 

nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/connectable/StandardConnection.java

@@ -181,6 +181,10 @@ public final class StandardConnection implements Connection {
             throw new IllegalStateException("Cannot change destination of Connection because the current destination is running");
         }
 
+        if (getFlowFileQueue().getUnacknowledgedQueueSize().getObjectCount() > 0) {
+            throw new IllegalStateException("Cannot change destination of Connection because FlowFiles from this Connection are currently held by " + previousDestination);
+        }
+
         try {
             previousDestination.removeConnection(this);
             this.destination.set(newDestination);