[
https://issues.apache.org/jira/browse/NIFI-16475?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Seow Alex updated NIFI-16475:
-----------------------------
Description:
In a NiFi cluster, {{ConnectionStatus.getMaxQueuedDuration()}} returns 0 on
every node except one, even when the connection has flowfiles queued on that
node. The one node that reports correctly is the node whose name sorts last.
Anything that reads {{maxQueuedDuration}} from connection status, such as a
reporting task or the connection status from the REST API, gets 0 on most nodes.
*Cause*
When the node is clustered, {{FlowController.createConnection}} creates a
{{SocketLoadBalancedFlowFileQueue}} for every connection, including ones set to
"Do not load balance". The queue keeps one partition per node.
{{setNodeIdentifiers}} sorts the partitions by node name and port
({{{}apiAddress + ":" + apiPort{}}}).
{{SocketLoadBalancedFlowFileQueue.getMinLastQueueDate()}} is effectively:
{code:java}
long min = 0;
for (QueuePartition partition : queuePartitions) {
min = (min == 0) ? partition.getMinLastQueueDate() : Long.min(min,
partition.getMinLastQueueDate());
}
return min;{code}
An empty partition returns 0. Once {{min}} holds a real date, any later empty
partition sets it back to {{{}Long.min(min, 0) = 0{}}}. On a connection that
isn't load-balanced, only the local partition holds flowfiles. So the result is
non-zero only if the local partition is the last one in the sorted array.
{{AbstractEventAccess}} then sets {{maxQueuedDuration = 0}} when
{{getMinLastQueueDate()}} returns 0, and {{now - minLastQueueDate}} otherwise.
{{SwappablePriorityQueue.getMinLastQueueDate()}} has a similar flaw. It
computes {{{}Long.min(activeMin, getMinLastQueueDate(swapQueue,
activeMin)){}}}. If the active queue is empty, {{activeMin}} is 0 and the swap
queue's dates are ignored. It falls back to the dates of swapped-out flowfiles
only when {{{}min == 0{}}}.
*Steps to reproduce*
1. Start a 3-node cluster with nodes named {{{}nifi-a{}}}, {{nifi-b}} and
{{{}nifi-c{}}}.
2. Create a flow with a connection whose load balance strategy is "Do not load
balance". Stop the destination processor so flowfiles stay queued.
3. Generate flowfiles on each node.
4. Read {{maxQueuedDuration}} for the connection from each node's local status.
*Expected:* every node reports the age of the oldest flowfile queued on it.
*Actual:* {{nifi-c}} (last in sort order) reports a non-zero value. {{nifi-a}}
and {{nifi-b}} report 0.
*Observed in production*
On three separate 3-node clusters running 1.28.1, and on 3-node clusters
running 2.7.2, only the last node by name ever reported a non-zero
{{maxQueuedDuration}} over several weeks of monitoring. The other nodes
reported 0 even with flowfiles queued for days.
was:
In a NiFi cluster, {{ConnectionStatus.getMaxQueuedDuration()}} returns 0 on
every node except one, even when the connection has flowfiles queued on that
node. The one node that reports correctly is the node whose name sorts last.
Anything that reads {{maxQueuedDuration}} from connection status, such as a
reporting task or the connection status from the REST API, gets 0 on most nodes.
*Cause*
When the node is clustered, {{FlowController.createConnection}} creates a
{{SocketLoadBalancedFlowFileQueue}} for every connection, including ones set to
"Do not load balance". The queue keeps one partition per node.
{{setNodeIdentifiers}} sorts the partitions by node name and port (`apiAddress
+ ":" + apiPort`).
{{SocketLoadBalancedFlowFileQueue.getMinLastQueueDate()}} is effectively:
{code:java}
long min = 0;
for (QueuePartition partition : queuePartitions) {
min = (min == 0) ? partition.getMinLastQueueDate() : Long.min(min,
partition.getMinLastQueueDate());
}
return min;{code}
An empty partition returns 0. Once {{min}} holds a real date, any later empty
partition sets it back to {{{}Long.min(min, 0) = 0{}}}. On a connection that
isn't load-balanced, only the local partition holds flowfiles. So the result is
non-zero only if the local partition is the last one in the sorted array.
{{AbstractEventAccess}} then sets {{maxQueuedDuration = 0}} when
{{getMinLastQueueDate()}} returns 0, and {{now - minLastQueueDate}} otherwise.
{{SwappablePriorityQueue.getMinLastQueueDate()}} has a similar flaw. It
computes {{{}Long.min(activeMin, getMinLastQueueDate(swapQueue,
activeMin)){}}}. If the active queue is empty, {{activeMin}} is 0 and the swap
queue's dates are ignored. It falls back to the dates of swapped-out flowfiles
only when {{{}min == 0{}}}.
*Steps to reproduce*
1. Start a 3-node cluster with nodes named {{{}nifi-a{}}}, {{nifi-b}} and
{{{}nifi-c{}}}.
2. Create a flow with a connection whose load balance strategy is "Do not load
balance". Stop the destination processor so flowfiles stay queued.
3. Generate flowfiles on each node.
4. Read {{maxQueuedDuration}} for the connection from each node's local status.
*Expected:* every node reports the age of the oldest flowfile queued on it.
*Actual:* {{nifi-c}} (last in sort order) reports a non-zero value. {{nifi-a}}
and {{nifi-b}} report 0.
*Observed in production*
On three separate 3-node clusters running 1.28.1, and on 3-node clusters
running 2.7.2, only the last node by name ever reported a non-zero
{{maxQueuedDuration}} over several weeks of monitoring. The other nodes
reported 0 even with flowfiles queued for days.
> ConnectionStatus.maxQueuedDuration is 0 on all but one node in a cluster
> ------------------------------------------------------------------------
>
> Key: NIFI-16475
> URL: https://issues.apache.org/jira/browse/NIFI-16475
> Project: Apache NiFi
> Issue Type: Bug
> Components: Core Framework
> Affects Versions: 1.28.1, 2.7.2
> Reporter: Seow Alex
> Priority: Major
>
> In a NiFi cluster, {{ConnectionStatus.getMaxQueuedDuration()}} returns 0 on
> every node except one, even when the connection has flowfiles queued on that
> node. The one node that reports correctly is the node whose name sorts last.
> Anything that reads {{maxQueuedDuration}} from connection status, such as a
> reporting task or the connection status from the REST API, gets 0 on most
> nodes.
> *Cause*
> When the node is clustered, {{FlowController.createConnection}} creates a
> {{SocketLoadBalancedFlowFileQueue}} for every connection, including ones set
> to "Do not load balance". The queue keeps one partition per node.
> {{setNodeIdentifiers}} sorts the partitions by node name and port
> ({{{}apiAddress + ":" + apiPort{}}}).
> {{SocketLoadBalancedFlowFileQueue.getMinLastQueueDate()}} is effectively:
> {code:java}
> long min = 0;
> for (QueuePartition partition : queuePartitions) {
> min = (min == 0) ? partition.getMinLastQueueDate() : Long.min(min,
> partition.getMinLastQueueDate());
> }
> return min;{code}
> An empty partition returns 0. Once {{min}} holds a real date, any later empty
> partition sets it back to {{{}Long.min(min, 0) = 0{}}}. On a connection that
> isn't load-balanced, only the local partition holds flowfiles. So the result
> is non-zero only if the local partition is the last one in the sorted array.
> {{AbstractEventAccess}} then sets {{maxQueuedDuration = 0}} when
> {{getMinLastQueueDate()}} returns 0, and {{now - minLastQueueDate}} otherwise.
> {{SwappablePriorityQueue.getMinLastQueueDate()}} has a similar flaw. It
> computes {{{}Long.min(activeMin, getMinLastQueueDate(swapQueue,
> activeMin)){}}}. If the active queue is empty, {{activeMin}} is 0 and the
> swap queue's dates are ignored. It falls back to the dates of swapped-out
> flowfiles only when {{{}min == 0{}}}.
> *Steps to reproduce*
> 1. Start a 3-node cluster with nodes named {{{}nifi-a{}}}, {{nifi-b}} and
> {{{}nifi-c{}}}.
> 2. Create a flow with a connection whose load balance strategy is "Do not
> load balance". Stop the destination processor so flowfiles stay queued.
> 3. Generate flowfiles on each node.
> 4. Read {{maxQueuedDuration}} for the connection from each node's local
> status.
> *Expected:* every node reports the age of the oldest flowfile queued on it.
> *Actual:* {{nifi-c}} (last in sort order) reports a non-zero value.
> {{nifi-a}} and {{nifi-b}} report 0.
> *Observed in production*
> On three separate 3-node clusters running 1.28.1, and on 3-node clusters
> running 2.7.2, only the last node by name ever reported a non-zero
> {{maxQueuedDuration}} over several weeks of monitoring. The other nodes
> reported 0 even with flowfiles queued for days.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
