This is an automated email from the ASF dual-hosted git repository.
apupier pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new d46d51c24550 CAMEL-25666: Stabilize ConsumeChildrenIT - replace fixed
delay with Awaitility
d46d51c24550 is described below
commit d46d51c245506b866182d16184a76380b0e82df4
Author: Aurélien Pupier <[email protected]>
AuthorDate: Thu Oct 15 15:00:26 2026 +0200
CAMEL-25666: Stabilize ConsumeChildrenIT - replace fixed delay with
Awaitility
The test shouldAwaitCreationAndGetDataNotification fires 5 ZooKeeper
mutations in sequence and expects the mock endpoint to receive exactly
5 messages (1 NodeCreated + 4 NodeChildrenChanged).
The consumer uses a single-threaded OperationsExecutor: after each
ChildrenChangedOperation fires it must synchronously call getChildren()
in getResult(), deliver the exchange to the processor, then re-enqueue
a copy of itself via createCopy(), and finally call installWatch() on
the new copy to re-register the ZooKeeper watcher before the next event
can be captured.
The original test relied on a 200 ms pre-delay in TestZookeeperClient
(delay(200) fires before the ZooKeeper mutation, not after). On a slow
or loaded machine this is not long enough for the consumer to finish
processing + re-registering the watch before the next mutation fires,
causing a ZooKeeper NodeChildrenChanged event to be silently dropped.
The mock then receives 4 instead of 5 messages.
Fix: use Awaitility after each mutation to wait until the mock has
actually received the corresponding message before triggering the next
one. This guarantees the watch is re-registered before the next event
fires, making the sequencing deterministic regardless of machine speed.
Also rename the ZooKeeper node from /grimm to /grimm-ConsumeChildrenIT
to isolate this test from any other test that might share the same
ZooKeeper server, consistent with the same fix already applied to
ConsumeDataIT and the cluster integration tests.
Co-authored-by: IBM Bob 2.2.1
Signed-off-by: Aurélien Pupier <[email protected]>
---
.../zookeeper/integration/ConsumeChildrenIT.java | 37 ++++++++++++++++++----
1 file changed, 30 insertions(+), 7 deletions(-)
diff --git
a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/integration/ConsumeChildrenIT.java
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/integration/ConsumeChildrenIT.java
index 9160db40961d..46730fe2f5e8 100644
---
a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/integration/ConsumeChildrenIT.java
+++
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/integration/ConsumeChildrenIT.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.zookeeper.integration;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import org.apache.camel.CamelExchangeException;
import org.apache.camel.Exchange;
@@ -29,6 +30,7 @@ import org.apache.camel.component.zookeeper.ZooKeeperMessage;
import org.apache.camel.support.ExchangeHelper;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
+import org.awaitility.Awaitility;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -36,11 +38,13 @@ import static
org.junit.jupiter.api.Assertions.assertNotSame;
public class ConsumeChildrenIT extends ZooKeeperITSupport {
+ private static final String ZK_NODE = "/grimm-ConsumeChildrenIT";
+
@Override
protected RouteBuilder[] createRouteBuilders() {
return new RouteBuilder[] { new RouteBuilder() {
public void configure() {
-
from("zookeeper://{{zookeeper.connection.string}}/grimm?repeat=true&listChildren=true")
+ from("zookeeper://{{zookeeper.connection.string}}" + ZK_NODE +
"?repeat=true&listChildren=true")
.sort(body(), new
NaturalSortComparator(Order.Descending))
.to("mock:zookeeper-data");
}
@@ -52,19 +56,38 @@ public class ConsumeChildrenIT extends ZooKeeperITSupport {
MockEndpoint mock = getMockEndpoint("mock:zookeeper-data");
mock.expectedMessageCount(5);
- client.createPersistent("/grimm", "parent");
- client.create("/grimm/hansel", "child");
- client.create("/grimm/gretel", "child");
- client.delete("/grimm/hansel");
- client.delete("/grimm/gretel");
+ client.createPersistent(ZK_NODE, "parent");
+ awaitMockReceivedCount(mock, 1);
+ client.create(ZK_NODE + "/hansel", "child");
+ awaitMockReceivedCount(mock, 2);
+ client.create(ZK_NODE + "/gretel", "child");
+ awaitMockReceivedCount(mock, 3);
+ client.delete(ZK_NODE + "/hansel");
+ awaitMockReceivedCount(mock, 4);
+ client.delete(ZK_NODE + "/gretel");
- MockEndpoint.assertIsSatisfied(context);
+ MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
validateExchangesContainListings(mock, createChildListing(),
createChildListing("hansel"),
createChildListing("hansel", "gretel"),
createChildListing("gretel"),
createChildListing());
}
+ /**
+ * Waits until the mock has received at least {@code count} messages
before the next ZooKeeper mutation is fired.
+ * <p>
+ * This ensures the previous ChildrenChangedOperation copy has been taken
off the queue and its
+ * {@code installWatch()} called, which happens after the exchange is
delivered to the mock. The tiny remaining gap
+ * between the mock receipt and the actual watch registration is covered
by the 200 ms pre-delay inside
+ * {@link ZooKeeperITSupport.TestZookeeperClient#create} / {@code delete}.
If that delay is ever removed, a short
+ * post-delay or a deeper hook into the consumer would be needed here to
avoid reintroducing flakiness.
+ */
+ private void awaitMockReceivedCount(MockEndpoint mock, int count) {
+ Awaitility.await()
+ .atMost(10, TimeUnit.SECONDS)
+ .until(() -> mock.getReceivedCounter() >= count);
+ }
+
private void validateExchangesContainListings(MockEndpoint mock,
List<?>... expected)
throws CamelExchangeException {
int index = 0;
