[
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. `totalQueuedDuration` is not affected.
**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:
```java
long min = 0;
for (QueuePartition partition : queuePartitions) {
min = (min == 0) ? partition.getMinLastQueueDate()
: Long.min(min, partition.getMinLastQueueDate());
}
return min;
```
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)
