This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 276c3ce1f92 NIFI-16182 Fixed revision equality check in
WaitNotifyProtocol (#11527)
276c3ce1f92 is described below
commit 276c3ce1f92aaf2d97f352fb38db61edae7834e6
Author: Peter Turcsanyi <[email protected]>
AuthorDate: Tue Aug 11 14:18:01 2026 +0200
NIFI-16182 Fixed revision equality check in WaitNotifyProtocol (#11527)
---
.../processors/standard/WaitNotifyProtocol.java | 3 +-
.../standard/TestWaitNotifyProtocol.java | 70 +++++++++++++++++++++-
2 files changed, 69 insertions(+), 4 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/WaitNotifyProtocol.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/WaitNotifyProtocol.java
index 4b5b04eeb9f..d510f8b70cb 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/WaitNotifyProtocol.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/WaitNotifyProtocol.java
@@ -34,6 +34,7 @@ import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.function.Consumer;
/**
@@ -312,7 +313,7 @@ public class WaitNotifyProtocol {
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)) {
+ if (expectedRevision != null && !Objects.deepEquals(expectedRevision,
actualRevision)) {
throw new ConcurrentModificationException(String.format(
"Failed to complete signal [%s]: signal was concurrently
modified (expected revision %s, found %s).",
signalId, expectedRevision, actualRevision));
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestWaitNotifyProtocol.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestWaitNotifyProtocol.java
index d2c2149f668..2c67194efc9 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestWaitNotifyProtocol.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestWaitNotifyProtocol.java
@@ -30,10 +30,12 @@ import java.io.ByteArrayOutputStream;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.UUID;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -424,8 +426,9 @@ public class TestWaitNotifyProtocol {
}
- @Test
- public void testCompleteRemovesSignalFromCache() throws Exception {
+ private void testCompleteRemovesSignalFromCache(final
AtomicDistributedMapCacheClient<Long> cache,
+ final Map<String, ?
extends AtomicCacheEntry<String, String, ?>> cacheEntries,
+ final Answer<?>
successfulReplace) throws Exception {
doAnswer(successfulReplace).when(cache).replace(any(), any(), any());
doAnswer(invocation -> {
cacheEntries.remove(invocation.getArguments()[0]);
@@ -446,7 +449,23 @@ public class TestWaitNotifyProtocol {
}
@Test
- public void testCompleteThrowsOnConcurrentModification() throws Exception {
+ public void testCompleteRemovesSignalFromCache() throws Exception {
+ testCompleteRemovesSignalFromCache(cache, cacheEntries,
successfulReplace);
+ }
+
+ @Test
+ public void testCompleteRemovesSignalFromCacheUsingByteArrayRevision()
throws Exception {
+ final ByteArrayRevisionCache byteArrayRevisionCache =
createByteArrayRevisionCache();
+
+ testCompleteRemovesSignalFromCache(
+ byteArrayRevisionCache.cache(),
+ byteArrayRevisionCache.cacheEntries(),
+ byteArrayRevisionCache.successfulReplace());
+ }
+
+ private void testCompleteThrowsOnConcurrentModification(final
AtomicDistributedMapCacheClient<Long> cache,
+ final Map<String,
? extends AtomicCacheEntry<String, String, ?>> cacheEntries,
+ final Answer<?>
successfulReplace) throws Exception {
doAnswer(successfulReplace).when(cache).replace(any(), any(), any());
final WaitNotifyProtocol protocol = new WaitNotifyProtocol(cache);
@@ -467,6 +486,21 @@ public class TestWaitNotifyProtocol {
assertTrue(cacheEntries.containsKey(signalId));
}
+ @Test
+ public void testCompleteThrowsOnConcurrentModification() throws Exception {
+ testCompleteThrowsOnConcurrentModification(cache, cacheEntries,
successfulReplace);
+ }
+
+ @Test
+ public void
testCompleteThrowsOnConcurrentModificationUsingByteArrayRevision() throws
Exception {
+ final ByteArrayRevisionCache byteArrayRevisionCache =
createByteArrayRevisionCache();
+
+ testCompleteThrowsOnConcurrentModification(
+ byteArrayRevisionCache.cache(),
+ byteArrayRevisionCache.cacheEntries(),
+ byteArrayRevisionCache.successfulReplace());
+ }
+
@Test
public void testCompleteThrowsWhenAlreadyRemoved() throws Exception {
doAnswer(successfulReplace).when(cache).replace(any(), any(), any());
@@ -488,4 +522,34 @@ public class TestWaitNotifyProtocol {
assertEquals(mapper.readTree(expected), mapper.readTree(value));
}
+ private ByteArrayRevisionCache createByteArrayRevisionCache() throws
Exception {
+ final Map<String, AtomicCacheEntry<String, String, byte[]>>
cacheEntries = new HashMap<>();
+
+ final Answer<?> successfulReplace = invocation -> {
+ final AtomicCacheEntry<String, String, byte[]> entry =
invocation.getArgument(0);
+ // generate a unique, random byte sequence for the revision
+ cacheEntries.put(entry.getKey(), new
AtomicCacheEntry<>(entry.getKey(), entry.getValue(),
UUID.randomUUID().toString().getBytes()));
+ return true;
+ };
+
+ final AtomicDistributedMapCacheClient<Long> cache =
mock(AtomicDistributedMapCacheClient.class);
+
+ doAnswer(invocation -> {
+ final AtomicCacheEntry<String, String, byte[]> entry =
cacheEntries.get(invocation.getArguments()[0]);
+ if (entry == null) {
+ return null;
+ } else {
+ // copy the revision to avoid simple reference-based equality
+ return new AtomicCacheEntry<>(entry.getKey(),
entry.getValue(), entry.getRevision().map(bytes -> Arrays.copyOf(bytes,
bytes.length)).orElse(null));
+ }
+ }).when(cache).fetch(any(), any(), any());
+
+ return new ByteArrayRevisionCache(cache, cacheEntries,
successfulReplace);
+ }
+
+ private record ByteArrayRevisionCache(
+ AtomicDistributedMapCacheClient<Long> cache,
+ Map<String, AtomicCacheEntry<String, String, byte[]>> cacheEntries,
+ Answer<?> successfulReplace) {
+ }
}
