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

NIFI-5919: Addressed a race condition that can exist if adding FlowFiles to a FlowFileQueue before adjusting the size of the queue to account for the FlowFiles 

Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com>

This closes #3238.
This commit is contained in:
Mark Payne 2018-12-28 10:35:48 -05:00 committed by Pierre Villard
parent 595a2decc6
commit fea17d0ca8
No known key found for this signature in database
GPG Key ID: BEE1599F0726E9CD
1 changed files with 19 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -769,8 +769,26 @@ public class SocketLoadBalancedFlowFileQueue extends AbstractFlowFileQueue imple
putAll(flowFiles);
} else {
logger.debug("Received the following FlowFiles from Peer: {}. Will accept FlowFiles to the local partition", flowFiles);
localPartition.putAll(flowFiles);
// As explained in the putAllAndGetPartitions() method, we must ensure that we call adjustSize() before we
// put the FlowFiles on the queue. Otherwise, we will encounter a race condition. Specifically, that race condition
// can play out like so:
//
// Thread 1: Call localPartition.putAll() when the queue is empty (has a queue size of 0) but has not yet adjusted the size.
// Thread 2: Call poll() to obtain the FlowFile just received.
// Thread 2: Transfer the FlowFile to some Relationship
// Thread 2: Commit the session, which will call acknowledge on this queue.
// Thread 2: The acknowledge() method attempts to decrement the size of the queue to -1.
// This causes an Exception to be thrown and the queue size to remain at 0.
// However, the FlowFile has already been successfully transferred to the next Queue.
// Thread 1: Call adjustSize() to increment the size of the queue to 1 FlowFile.
//
// In this scenario, we now have no FlowFiles in the queue. However, the queue size is set to 1.
// We can avoid this race condition by simply ensuring that we call adjustSize() before making the FlowFiles
// available on the queue. This way, we cannot possibly obtain the FlowFiles and process/acknowledge them before the queue
// size has been updated to account for them and therefore we will not attempt to assign a negative queue size.
adjustSize(flowFiles.size(), flowFiles.stream().mapToLong(FlowFileRecord::getSize).sum());
localPartition.putAll(flowFiles);
}
} finally {
partitionReadLock.unlock();