turcsanyip commented on code in PR #11275:
URL: https://github.com/apache/nifi/pull/11275#discussion_r3617322980
##########
nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/WaitNotifyProtocol.java:
##########
@@ -281,22 +284,55 @@ public Signal getSignal(final String signalId) throws
IOException, Deserializati
/**
* Finish protocol and remove the cache entry.
- * @param signalId a key in the underlying cache engine
+ *
+ * <p>This method performs a best-effort version check before removing the
entry. If the signal
+ * was concurrently modified by a Notify processor after the caller last
read it, a
+ * {@link ConcurrentModificationException} is thrown so the caller can
roll back and retry
+ * rather than silently discarding the concurrent notification.</p>
+ *
+ * <p>Note: there is a small inherent TOCTOU window between the version
re-fetch and the
+ * remove call. A {@link AtomicDistributedMapCacheClient} API extension
for atomic
+ * compare-and-delete would eliminate this entirely, but this approach
covers the common case.</p>
+ *
+ * @param signal the Signal obtained from the most recent {@link
#getSignal(String)} call;
+ * its cached revision is used to detect concurrent
modifications
* @throws IOException thrown when it failed interacting with the cache
engine
+ * @throws ConcurrentModificationException thrown if the signal was
concurrently modified
+ * or removed since the caller last read it
*/
- public void complete(final String signalId) throws IOException {
+ public void complete(final Signal signal) throws IOException,
ConcurrentModificationException {
+ final String signalId = signal.identifier;
+
+ // Re-fetch to detect concurrent updates since the signal was last
read.
+ final Signal current = getSignal(signalId);
+ if (current == null) {
+ throw new ConcurrentModificationException(String.format(
+ "Failed to complete signal [%s]: signal was concurrently
removed.", signalId));
+ }
+
+ final Object expectedRevision = signal.cachedEntry != null ?
signal.cachedEntry.getRevision().orElse(null) : null;
+ final Object actualRevision =
current.cachedEntry.getRevision().orElse(null);
+ if (expectedRevision != null &&
!expectedRevision.equals(actualRevision)) {
+ throw new ConcurrentModificationException(String.format(
+ "Failed to complete signal [%s]: signal was concurrently
modified (expected revision %s, found %s).",
+ signalId, expectedRevision, actualRevision));
+ }
+
cache.remove(signalId, stringSerializer);
Review Comment:
As you mentioned, the ideal solution would be the atomic compare-and-delete
supported by `AtomicDistributedMapCacheClient`. However, the TOCTOU problem can
be minimized further if we check the return value of `remove()`.
```suggestion
if (!cache.remove(signalId, stringSerializer)) {
throw new ConcurrentModificationException(String.format(
"Failed to complete signal [%s]: signal was concurrently
removed.", signalId));
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
