[
https://issues.apache.org/jira/browse/NIFI-15883?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18098130#comment-18098130
]
ASF subversion and git services commented on NIFI-15883:
--------------------------------------------------------
Commit c7725c43340292e2231b4024b2456aa0ac284c44 in nifi's branch
refs/heads/main from Rakesh Kumar Singh
[ https://gitbox.apache.org/repos/asf?p=nifi.git;h=c7725c43340 ]
NIFI-15883: Fix Wait/Notify race condition with CAS-aware replace and complete
(#11275)
- WaitNotifyProtocol.replace(Signal): Changed from returning boolean to void.
Throws ConcurrentModificationException when the underlying cache.replace()
returns false (CAS failure due to concurrent Notify), making the API
consistent with complete(Signal) which also throws on concurrent modification.
- WaitNotifyProtocol.notify(): Updated retry loop to catch
ConcurrentModificationException from replace() instead of checking
the boolean return value.
- WaitNotifyProtocol.complete(Signal): Replaces raw cache.remove() with a
version-aware approach that re-fetches the cache entry and compares
revisions before removing. Throws ConcurrentModificationException if
a concurrent Notify updated the signal between Wait's initial read
and the remove call, preventing silent data loss.
- Wait.java: Simplified call sites - protocol.replace(signal) and
protocol.complete(signal) both now throw ConcurrentModificationException
on concurrent modification. The catch block handles both IOException and
ConcurrentModificationException with session rollback and ProcessException.
- TestWaitNotifyProtocol: Add three tests for complete(Signal):
happy-path removal, concurrent-modification detection, and no-op
when the signal is already absent.
- TestWait: Add two integration tests that verify session rollback and
ProcessException wrapping when replace() throws on CAS failure
(waitProgressed path) and when complete() detects a stale revision
(waitCompleted path).
> Wait/Notify Processing have a race condition while updating the Cache
> ---------------------------------------------------------------------
>
> Key: NIFI-15883
> URL: https://issues.apache.org/jira/browse/NIFI-15883
> Project: Apache NiFi
> Issue Type: Bug
> Components: Extensions
> Affects Versions: 1.15.3, 2.1.0, 2.9.0
> Reporter: Sheel Chand
> Assignee: Rakesh Kumar Singh
> Priority: Critical
> Time Spent: 1h
> Remaining Estimate: 0h
>
> Nifi Processor Wait calls the protocol.replace function without checking the
> return value - true or false. The replace() method may fail if a Notify
> processor has updated the counters since the last fetch() by the Wait
> Processor. As a result some updates are lost specially in a high concurrency
> case. If you have a counter to wait for a number of ACKS before releasing the
> files, the counter keeps increasing because the decrements by the Wait
> Processor get lost.
> The same issue with the entry removal. The protocol.complete() method
> removes the entry without going through the Cache Protocol Version 2. So, it
> may remove an entry even if a Notify processor has updated counters since the
> last fetch().
> The notify function in WaitNotifyPortocol.java does the right thing by
> retrying multiple times if the replace() method returns false.
> Wait Processor needs to be fixed to implement Cache Protocol Version 2 for
> replace and complete() methods.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
