This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch CAMEL-25782-flakyDataChangedOperationIT in repository https://gitbox.apache.org/repos/asf/camel.git
commit 29d0764c466f904c261bee949351268378089b34 Author: Aurélien Pupier <[email protected]> AuthorDate: Mon Oct 19 13:58:10 2026 +0200 CAMEL-25782: camel-zookeeper - Do not fail a watch operation whose awaited event was already received FutureEventDrivenOperation.watchRequestCompleted() (added by CAMEL-25655) fails the operation as soon as the asynchronous request installing the watch returns a non-OK result code, because ZooKeeper does not register the watch in that case and the awaited event would never come. However the request and the watched event are independent: when the node is deleted right after the request was sent, the NodeDeleted event can be processed first and the request then completes with NONODE. The operation had already got what it was waiting for, but the late failure was still recorded and get() threw "Could not watch node", which made DataChangedOperationIT.getsNotifiedWhenNodeIsDeleted flaky. Only record the failure when the awaited event has not been received yet. A failed request with no event received (session expired, connection loss) still fails the operation as before. FutureEventDrivenOperationTest covers both cases. Co-authored-by: IBM Bob 2.2.1 Signed-off-by: Aurélien Pupier <[email protected]> --- .../operations/FutureEventDrivenOperation.java | 6 ++-- .../operations/FutureEventDrivenOperationTest.java | 37 ++++++++++++++++++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java index 86a819c6ca3e..725acd143f6a 100644 --- a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java +++ b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java @@ -113,12 +113,14 @@ public abstract class FutureEventDrivenOperation<ResultType> extends ZooKeeperOp * To be called from the asynchronous callback of the request made by {@link #installWatch()}, with its result code. * When the request failed (for example with SESSIONEXPIRED on a handle whose session expired, or CONNECTIONLOSS), * ZooKeeper did not register the watch, so the awaited event will never come: the operation stops waiting and fails - * with the {@link KeeperException} of that result code. + * with the {@link KeeperException} of that result code. If the awaited event was already received there is nothing + * left to wait for, so the failure of the request (for example NONODE because the node was deleted in the meantime) + * is ignored. * * @param rc the result code of the request */ protected void watchRequestCompleted(int rc) { - if (rc != KeeperException.Code.OK.intValue()) { + if (rc != KeeperException.Code.OK.intValue() && waitForAnyWatchedType.getCount() > 0) { failure = new ExecutionException( "Could not watch node '" + getNode() + "'", KeeperException.create(KeeperException.Code.get(rc), getNode())); diff --git a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperationTest.java b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperationTest.java index d8d135f792b8..7dc2d8a4fcec 100644 --- a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperationTest.java +++ b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperationTest.java @@ -16,12 +16,17 @@ */ package org.apache.camel.component.zookeeper.operations; +import java.util.concurrent.ExecutionException; + +import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher.Event.EventType; import org.apache.zookeeper.data.Stat; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; public class FutureEventDrivenOperationTest { private String data = "Event Received"; @@ -49,6 +54,38 @@ public class FutureEventDrivenOperationTest { assertEquals(event, future.getWatchedEvent()); } + @Test + public void shouldIgnoreFailedWatchRequestWhenAwaitedEventAlreadyReceived() throws Exception { + // the event arrives before the callback of the watch request, which fails because the node is gone + final FutureEventDrivenOperation<String> future = failingWatchOperation(); + future.process(new WatchedEvent(EventType.NodeDeleted, null, "somepath")); + + assertEquals(data, future.get().getResult()); + } + + @Test + public void shouldFailWhenWatchRequestFailedAndNoEventReceived() { + final FutureEventDrivenOperation<String> future = failingWatchOperation(); + + ExecutionException e = assertThrows(ExecutionException.class, future::get); + assertInstanceOf(KeeperException.NoNodeException.class, e.getCause()); + } + + private FutureEventDrivenOperation<String> failingWatchOperation() { + return new FutureEventDrivenOperation<String>(null, "somepath", EventType.NodeDeleted) { + + @Override + protected void installWatch() { + watchRequestCompleted(KeeperException.Code.NONODE.intValue()); + } + + @Override + public OperationResult<String> getResult() { + return new OperationResult<>(data, statistics); + } + }; + } + private void fireEventIn( final FutureEventDrivenOperation<String> future, final WatchedEvent event, final int millisecondsTillFire) { new Thread(new Runnable() {
