From a29b7b3bf0d564a4f4ae622c7fdc37b80fed6e96 Mon Sep 17 00:00:00 2001
From: Mark Payne <markap14@hotmail.com>
Date: Tue, 24 Nov 2015 09:38:24 -0500
Subject: [PATCH] NIFI-1203: Do not count looping connections when determining
 validity based on incoming connections

Signed-off-by: joewitt <joewitt@apache.org>
---
 .../controller/StandardProcessorNode.java     | 20 +++++++++++++++----
 1 file changed, 16 insertions(+), 4 deletions(-)

diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
index 2b0d413579..567c3c782c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
@@ -964,6 +964,18 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
         }
     }
 
+    List<Connection> getIncomingNonLoopConnections() {
+        final List<Connection> connections = getIncomingConnections();
+        final List<Connection> nonLoopConnections = new ArrayList<>(connections.size());
+        for (final Connection connection : connections) {
+            if (!connection.getSource().equals(this)) {
+                nonLoopConnections.add(connection);
+            }
+        }
+
+        return nonLoopConnections;
+    }
+
     @Override
     public boolean isValid() {
         readLock.lock();
@@ -991,13 +1003,13 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
                 case INPUT_ALLOWED:
                     break;
                 case INPUT_FORBIDDEN: {
-                    if (!getIncomingConnections().isEmpty()) {
+                    if (!getIncomingNonLoopConnections().isEmpty()) {
                         return false;
                     }
                     break;
                 }
                 case INPUT_REQUIRED: {
-                    if (getIncomingConnections().isEmpty()) {
+                    if (getIncomingNonLoopConnections().isEmpty()) {
                         return false;
                     }
                     break;
@@ -1045,7 +1057,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
                 case INPUT_ALLOWED:
                     break;
                 case INPUT_FORBIDDEN: {
-                    final int incomingConnCount = getIncomingConnections().size();
+                    final int incomingConnCount = getIncomingNonLoopConnections().size();
                     if (incomingConnCount != 0) {
                         results.add(new ValidationResult.Builder()
                             .explanation("Processor does not allow upstream connections but currently has " + incomingConnCount)
@@ -1056,7 +1068,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
                     break;
                 }
                 case INPUT_REQUIRED: {
-                    if (getIncomingConnections().isEmpty()) {
+                    if (getIncomingNonLoopConnections().isEmpty()) {
                         results.add(new ValidationResult.Builder()
                             .explanation("Processor requires an upstream connection but currently has none")
                             .subject("Upstream Connections")