[jira] [Created] (NIFI-6781) ThreadPoolRequestReplicator

["Marc Parisi (Jira)" <ji...@apache.org>]

Marc Parisi created NIFI-6781:
---------------------------------

             Summary: ThreadPoolRequestReplicator
                 Key: NIFI-6781
                 URL: https://issues.apache.org/jira/browse/NIFI-6781
             Project: Apache NiFi
          Issue Type: Improvement
          Components: Core Framework
            Reporter: Marc Parisi
            Assignee: Marc Parisi


I've noticed that replication is attempted locally. I tested a simple change to eliminate the local node; however, I suspect this is not a big deal or I've missed something or the cluster states does not include the local identifier. All tests allow for local instances with different ports, implying that pruning is potentially unnecessary or incorrect logic. Therefore I've created this as a an "Improvement" as I dive further into the code to validate my change. If anyone has the immediate answer regarding this code I'm happy to close this as OBE.

 
{code:java}
 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
@@ -217,11 +217,17 @@ public class ThreadPoolRequestReplicator implements RequestReplicator {
             }
         }
 
-        final List<NodeIdentifier> nodeIds = stateMap.get(NodeConnectionState.CONNECTED);
+        // get nodes that do not match this node.
+        final List<NodeIdentifier> nodeIds = stateMap.get(NodeConnectionState.CONNECTED).stream().filter(x -> {
+            return clusterCoordinator.getLocalNodeIdentifier() == null || x != clusterCoordinator.getLocalNodeIdentifier();
+        }).collect(Collectors.toList());
+
         if (nodeIds == null || nodeIds.isEmpty()) {
             throw new NoConnectedNodesException();
         }
 
+        logger.debug("Attempting to replicate to {} nodes", nodeIds.size());
{code}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)