Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-5663: Ensure that when sort Node Identifiers that we use both the node's API Address as well as API Port, in case 2 nodes are running on same host. Also ensure that when Local Node ID is determined that we update all Load Balancing Partitions, if necessary 

This closes #3048.

Signed-off-by: Koji Kawamura <ijokarumawak@apache.org>
This commit is contained in:
Mark Payne 2018-10-05 12:06:39 -04:00 committed by Koji Kawamura
parent 768bcfb509
commit c87d791938
1 changed files with 12 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
View File

@ -571,7 +571,7 @@ public class SocketLoadBalancedFlowFileQueue extends AbstractFlowFileQueue imple
// Re-define 'queuePartitions' array // Re-define 'queuePartitions' array
final List<NodeIdentifier> sortedNodeIdentifiers = new ArrayList<>(updatedNodeIdentifiers); final List<NodeIdentifier> sortedNodeIdentifiers = new ArrayList<>(updatedNodeIdentifiers);
sortedNodeIdentifiers.sort(Comparator.comparing(NodeIdentifier::getApiAddress)); sortedNodeIdentifiers.sort(Comparator.comparing(nodeId -> nodeId.getApiAddress() + ":" + nodeId.getApiPort()));
final QueuePartition[] updatedQueuePartitions; final QueuePartition[] updatedQueuePartitions;
if (sortedNodeIdentifiers.isEmpty()) { if (sortedNodeIdentifiers.isEmpty()) {
@ -990,6 +990,14 @@ public class SocketLoadBalancedFlowFileQueue extends AbstractFlowFileQueue imple
return; return;
} }
if (!nodeIdentifiers.contains(localNodeId)) {
final Set<NodeIdentifier> updatedNodeIds = new HashSet<>(nodeIdentifiers);
updatedNodeIds.add(localNodeId);
logger.debug("Local Node Identifier has now been determined to be {}. Adding to set of Node Identifiers for {}", localNodeId, SocketLoadBalancedFlowFileQueue.this);
setNodeIdentifiers(updatedNodeIds, false);
}
logger.debug("Local Node Identifier set to {}; current partitions = {}", localNodeId, queuePartitions); logger.debug("Local Node Identifier set to {}; current partitions = {}", localNodeId, queuePartitions);
for (final QueuePartition partition : queuePartitions) { for (final QueuePartition partition : queuePartitions) {
@ -1009,7 +1017,9 @@ public class SocketLoadBalancedFlowFileQueue extends AbstractFlowFileQueue imple
logger.debug("{} Local Node Identifier set to {} and found Queue Partition {} with that Node Identifier. Will force update of partitions", logger.debug("{} Local Node Identifier set to {} and found Queue Partition {} with that Node Identifier. Will force update of partitions",
SocketLoadBalancedFlowFileQueue.this, localNodeId, partition); SocketLoadBalancedFlowFileQueue.this, localNodeId, partition);
setNodeIdentifiers(SocketLoadBalancedFlowFileQueue.this.nodeIdentifiers, true); final Set<NodeIdentifier> updatedNodeIds = new HashSet<>(nodeIdentifiers);
updatedNodeIds.add(localNodeId);
setNodeIdentifiers(updatedNodeIds, true);
return; return;
} }
} }