This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 2a30744c443a CAMEL-25655: camel-zookeeper - consumer waits for ever
and producer fails every message after the ZooKeeper session expired (#27858)
2a30744c443a is described below
commit 2a30744c443a6d0cd871283993db46fd862de4f8
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 15 17:31:20 2026 +0530
CAMEL-25655: camel-zookeeper - consumer waits for ever and producer fails
every message after the ZooKeeper session expired (#27858)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
components/camel-zookeeper/pom.xml | 14 ++
.../component/zookeeper/ConnectionHolder.java | 81 +++++--
.../component/zookeeper/ZooKeeperConsumer.java | 32 ++-
.../component/zookeeper/ZooKeeperProducer.java | 13 +-
.../operations/ChildrenChangedOperation.java | 4 +-
.../zookeeper/operations/DataChangedOperation.java | 4 +-
.../operations/ExistenceChangedOperation.java | 5 +
.../operations/FutureEventDrivenOperation.java | 43 +++-
.../ZooKeeperConsumerSessionExpiryTest.java | 253 +++++++++++++++++++++
parent/pom.xml | 1 +
10 files changed, 415 insertions(+), 35 deletions(-)
diff --git a/components/camel-zookeeper/pom.xml
b/components/camel-zookeeper/pom.xml
index e7bff0878ebb..05c2cf15e453 100644
--- a/components/camel-zookeeper/pom.xml
+++ b/components/camel-zookeeper/pom.xml
@@ -84,6 +84,20 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <!-- the ZooKeeper server that ZooKeeperConsumerSessionExpiryTest runs
in the test JVM needs these two
+ (both are "provided" dependencies of zookeeper) -->
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ <version>${metrics-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
+ <version>${snappy-java-version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-test-spring-junit6</artifactId>
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java
index f52a9419e37d..4037550c69e8 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java
@@ -33,31 +33,60 @@ import org.slf4j.LoggerFactory;
public class ConnectionHolder implements Watcher {
private static final Logger LOG =
LoggerFactory.getLogger(ConnectionHolder.class);
- private CountDownLatch connectionLatch = new CountDownLatch(1);
+ private volatile CountDownLatch connectionLatch = new CountDownLatch(1);
private ZooKeeperConfiguration configuration;
- private ZooKeeper zookeeper;
+ private volatile ZooKeeper zookeeper;
public ConnectionHolder(ZooKeeperConfiguration configuration) {
this.configuration = configuration;
}
+ /**
+ * Returns the ZooKeeper handle, and creates one when there is none or
when the current one can no longer be used: a
+ * handle whose session expired (or that was closed) stays closed for
good, so a new handle with a new session is
+ * needed. Waits until the new handle is connected.
+ */
public ZooKeeper getZooKeeper() {
- if (zookeeper != null) {
- return zookeeper;
+ // fast path without the lock: the producer calls this for every
message
+ ZooKeeper current = zookeeper;
+ if (current != null && current.getState().isAlive()) {
+ return current;
}
- if (configuration.getConnectString() == null) {
- throw new RuntimeCamelException(
- "Cannot create ZooKeeper connection as connection string
is null. Have servers been configured?");
- }
- try {
- zookeeper = new ZooKeeper(configuration.getConnectString(),
configuration.getTimeout(), this);
- } catch (Exception e) {
- throw RuntimeCamelException.wrapRuntimeCamelException(e);
+ ZooKeeper created;
+ CountDownLatch latch;
+ synchronized (this) {
+ current = zookeeper;
+ if (current != null) {
+ if (current.getState().isAlive()) {
+ return current;
+ }
+ LOG.info("The ZooKeeper session 0x{} to {} can no longer be
used ({}). Creating a new session.",
+ Long.toHexString(current.getSessionId()),
configuration.getConnectString(), current.getState());
+ closeQuietly(current);
+ zookeeper = null;
+ }
+ if (configuration.getConnectString() == null) {
+ throw new RuntimeCamelException(
+ "Cannot create ZooKeeper connection as connection
string is null. Have servers been configured?");
+ }
+ latch = new CountDownLatch(1);
+ connectionLatch = latch;
+ try {
+ // each handle counts down its own latch, so that a late event
of a previous handle cannot release
+ // the wait for a new one
+ created = new ZooKeeper(configuration.getConnectString(),
configuration.getTimeout(), event -> {
+ process(event, latch);
+ });
+ zookeeper = created;
+ } catch (Exception e) {
+ throw RuntimeCamelException.wrapRuntimeCamelException(e);
+ }
}
- awaitConnection();
- return zookeeper;
+ // wait on the latch of this handle: another thread may already have
replaced the field
+ awaitConnection(latch);
+ return created;
}
public boolean isConnected() {
@@ -65,11 +94,15 @@ public class ConnectionHolder implements Watcher {
}
public void awaitConnection() {
+ awaitConnection(connectionLatch);
+ }
+
+ private void awaitConnection(CountDownLatch latch) {
if (LOG.isDebugEnabled()) {
LOG.debug("Awaiting Connection event from Zookeeper cluster {}",
configuration.getConnectString());
}
try {
- connectionLatch.await();
+ latch.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw RuntimeCamelException.wrapRuntimeCamelException(e);
@@ -78,13 +111,25 @@ public class ConnectionHolder implements Watcher {
@Override
public void process(WatchedEvent event) {
+ process(event, connectionLatch);
+ }
+
+ private static void process(WatchedEvent event, CountDownLatch latch) {
if (event.getState() == KeeperState.SyncConnected) {
- connectionLatch.countDown();
+ latch.countDown();
+ }
+ latch.countDown();
+ }
+
+ private void closeQuietly(ZooKeeper handle) {
+ try {
+ handle.close();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
}
- connectionLatch.countDown();
}
- public void closeConnection() {
+ public synchronized void closeConnection() {
try {
if (zookeeper != null) {
zookeeper.close();
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java
index 9e73fc6cffe5..275c661ecf2e 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java
@@ -50,7 +50,7 @@ public class ZooKeeperConsumer extends DefaultConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(ZooKeeperConsumer.class);
private final ZooKeeperConnectionManager zkm;
- private ZooKeeper connection;
+ private volatile ZooKeeper connection;
private ZooKeeperConfiguration configuration;
private LinkedBlockingQueue<ZooKeeperOperation> operations = new
LinkedBlockingQueue<>();
private ExecutorService executor;
@@ -154,6 +154,7 @@ public class ZooKeeperConsumer extends DefaultConsumer {
continue;
}
String node = current.getNode();
+ boolean restarted = false;
try {
OperationResult result = current.get();
if (ZooKeeperUtils.hasWatchedEvent(current)) {
@@ -166,12 +167,14 @@ public class ZooKeeperConsumer extends DefaultConsumer {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
handleException(e);
- backoffAndThenRestart();
+ restarted = backoffAndThenRestart();
} catch (Exception e) {
handleException(e);
- backoffAndThenRestart();
+ restarted = backoffAndThenRestart();
} finally {
- if (configuration.isRepeat()) {
+ // after a restart the queue holds the operations of the
new sequence; a copy of the failed one
+ // (which may use a handle whose session expired) must not
be added to them
+ if (configuration.isRepeat() && !restarted) {
try {
operations.offer(current.createCopy());
} catch (Exception e) {
@@ -182,7 +185,13 @@ public class ZooKeeperConsumer extends DefaultConsumer {
}
}
- private void backoffAndThenRestart() {
+ /**
+ * Waits for the backoff, then starts the sequence of operations again
on the current connection, which is a new
+ * one when the session of the previous one expired.
+ *
+ * @return true when the sequence was started again
+ */
+ private boolean backoffAndThenRestart() {
try {
if (isRunAllowed()) {
// Use Camel's task API for reconnection backoff delay
instead of Thread.sleep()
@@ -199,11 +208,22 @@ public class ZooKeeperConsumer extends DefaultConsumer {
.build()
.run(getEndpoint().getCamelContext(), () -> true);
- initializeConsumer();
+ if (isRunAllowed()) {
+ connection = zkm.getConnection();
+ if (!isRunAllowed()) {
+ // stopped while the connection was created:
doStop may already have closed the holder,
+ // so close the new handle (and its client
threads) here
+ zkm.shutdown();
+ return false;
+ }
+ initializeConsumer();
+ return true;
+ }
}
} catch (Exception e) {
// ignore
}
+ return false;
}
}
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
index aad9c2d99ffb..e8c01cba654c 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
@@ -66,10 +66,11 @@ public class ZooKeeperProducer extends DefaultProducer {
@Override
public void process(Exchange exchange) throws Exception {
- if (connection == null) {
- connection = this.zkm.getConnection();
- }
- ProductionContext context = new ProductionContext(connection,
exchange);
+ // a new connection when the session of the previous one expired; a
local variable keeps this message on the
+ // handle it got here, as other threads may send concurrently
+ ZooKeeper zk = this.zkm.getConnection();
+ connection = zk;
+ ProductionContext context = new ProductionContext(zk, exchange);
String operation =
exchange.getIn().getHeader(ZooKeeperMessage.ZOOKEEPER_OPERATION, String.class);
boolean isDelete = ZK_OPERATION_DELETE.equals(operation);
@@ -94,9 +95,9 @@ public class ZooKeeperProducer extends DefaultProducer {
}
} else {
if (isDelete) {
- asynchronouslyDeleteNode(connection, context);
+ asynchronouslyDeleteNode(zk, context);
} else {
- asynchronouslySetDataOnNode(connection, context);
+ asynchronouslySetDataOnNode(zk, context);
}
}
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java
index 1600ab35900d..b460b8760a3f 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java
@@ -41,8 +41,8 @@ public class ChildrenChangedOperation extends
FutureEventDrivenOperation<List<St
@Override
protected void installWatch() {
- connection.getChildren(getNode(), this, (int rc, String path, Object
ctx, List<String> children, Stat stat) -> {
- }, null);
+ connection.getChildren(getNode(), this,
+ (int rc, String path, Object ctx, List<String> children, Stat
stat) -> watchRequestCompleted(rc), null);
}
@Override
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java
index ed9d5a3264ea..85fe2f325822 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java
@@ -44,8 +44,8 @@ public class DataChangedOperation extends
FutureEventDrivenOperation<byte[]> {
@Override
protected void installWatch() {
- connection.getData(getNode(), this, (int rc, String path, Object ctx,
byte[] data, Stat stat) -> {
- }, null);
+ connection.getData(getNode(), this,
+ (int rc, String path, Object ctx, byte[] data, Stat stat) ->
watchRequestCompleted(rc), null);
}
@Override
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java
index 24c4cd54cbaf..13a1f9a62369 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.zookeeper.operations;
+import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
@@ -33,6 +34,10 @@ public class ExistenceChangedOperation extends
FutureEventDrivenOperation<String
@Override
protected void installWatch() {
connection.exists(getNode(), this, (int rc, String path, Object ctx,
Stat stat) -> {
+ // a node that does not exist is what this operation waits for:
ZooKeeper registers the watch then too
+ if (rc != KeeperException.Code.NONODE.intValue()) {
+ watchRequestCompleted(rc);
+ }
}, null);
if (LOG.isDebugEnabled()) {
LOG.debug("Installed exists watch");
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 771ca5627978..86a819c6ca3e 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
@@ -22,9 +22,11 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import static java.lang.String.format;
@@ -42,6 +44,8 @@ public abstract class FutureEventDrivenOperation<ResultType>
extends ZooKeeperOp
private WatchedEvent event;
+ private volatile ExecutionException failure;
+
protected FutureEventDrivenOperation(ZooKeeper connection, String node,
EventType... awaitedTypes) {
super(connection, node);
this.awaitedTypes = awaitedTypes;
@@ -55,6 +59,16 @@ public abstract class FutureEventDrivenOperation<ResultType>
extends ZooKeeperOp
LOG.debug(format("Received event of type %s for node '%s'",
received, event.getPath()));
}
+ // When the session expires, ZooKeeper tells every watcher with an
event of type None and state Expired. The
+ // awaited event can no longer come (the handle is closed for good),
so stop waiting and fail.
+ if (received == EventType.None && event.getState() ==
KeeperState.Expired) {
+ failure = new ExecutionException(
+ "The ZooKeeper session expired while waiting for an event
on node '" + getNode() + "'",
+ new KeeperException.SessionExpiredException());
+ waitForAnyWatchedType.countDown();
+ return;
+ }
+
for (EventType watched : awaitedTypes) {
if (watched.equals(received)) {
result = getResult();
@@ -74,6 +88,7 @@ public abstract class FutureEventDrivenOperation<ResultType>
extends ZooKeeperOp
installWatch();
waitingThreads.add(Thread.currentThread());
waitForAnyWatchedType.await();
+ throwIfFailed();
return result;
}
@@ -83,13 +98,39 @@ public abstract class
FutureEventDrivenOperation<ResultType> extends ZooKeeperOp
installWatch();
waitingThreads.add(Thread.currentThread());
waitForAnyWatchedType.await(timeout, unit);
+ throwIfFailed();
return result;
}
+ private void throwIfFailed() throws ExecutionException {
+ ExecutionException e = failure;
+ if (e != null) {
+ throw e;
+ }
+ }
+
+ /**
+ * 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.
+ *
+ * @param rc the result code of the request
+ */
+ protected void watchRequestCompleted(int rc) {
+ if (rc != KeeperException.Code.OK.intValue()) {
+ failure = new ExecutionException(
+ "Could not watch node '" + getNode() + "'",
+ KeeperException.create(KeeperException.Code.get(rc),
getNode()));
+ waitForAnyWatchedType.countDown();
+ }
+ }
+
/**
* Install the watcher to receive {@link WatchedEvent}s. It should use the
appropriate asynchronous ZooKeeper call
* to do this so as not to block the route from starting. Once one of the
watched for types of event is received a
- * call is made to getResult, which can use the appropriate synchronous
call to retrieve the actual data.
+ * call is made to getResult, which can use the appropriate synchronous
call to retrieve the actual data. The
+ * callback of the asynchronous call must pass its result code to {@link
#watchRequestCompleted(int)}.
*/
protected abstract void installWatch();
diff --git
a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperConsumerSessionExpiryTest.java
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperConsumerSessionExpiryTest.java
new file mode 100644
index 000000000000..16c87f767270
--- /dev/null
+++
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperConsumerSessionExpiryTest.java
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.zookeeper;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import
org.apache.camel.component.zookeeper.operations.FutureEventDrivenOperation;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A ZooKeeper consumer must keep consuming after the server expired its
session. The server runs in the test JVM. The
+ * session is expired the usual way: a second client opens the same session
and closes it; the consumer's client then
+ * gets the "Expired" event when it reconnects, and its handle is closed for
good. Every wait is bounded (20 seconds).
+ */
+class ZooKeeperConsumerSessionExpiryTest extends CamelTestSupport {
+
+ private static final String NODE = "/camel-session-expiry";
+ private static final String BLOCKED_NODE = "/camel-session-expiry-blocked";
+ private static final String ONCE_NODE = "/camel-session-expiry-once";
+
+ @TempDir
+ Path dataDir;
+
+ private final int port = AvailablePortFinder.getNextAvailable();
+
+ private final CountDownLatch processing = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ private ZooKeeperServer server;
+ private NIOServerCnxnFactory factory;
+ private ZooKeeper client;
+
+ @Override
+ protected void doPreSetup() throws Exception {
+ File dir = dataDir.toFile();
+ server = new ZooKeeperServer(dir, dir, 2000);
+ factory = new NIOServerCnxnFactory();
+ factory.configure(new InetSocketAddress("127.0.0.1", port), 60);
+ factory.startup(server);
+ client = connect(0, null);
+ client.create(NODE, bytes("one"), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
+ client.create(BLOCKED_NODE, bytes("one"), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
+ client.create(ONCE_NODE, bytes("one"), Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
+ }
+
+ @Override
+ protected void doPostTearDown() throws Exception {
+ release.countDown();
+ if (client != null) {
+ client.close();
+ }
+ if (factory != null) {
+ factory.shutdown();
+ }
+ if (server != null) {
+ server.shutdown();
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // each test starts the route it needs, so that only one
consumer runs at a time
+ from("zookeeper://127.0.0.1:" + port + NODE +
"?repeat=true&backoff=200").routeId("watch")
+ .autoStartup(false)
+ .to("mock:data");
+
+ // the first exchange holds the operations thread of the
consumer until the test releases it
+ from("zookeeper://127.0.0.1:" + port + BLOCKED_NODE +
"?repeat=true&backoff=200").routeId("blocked")
+ .autoStartup(false)
+ .process(e -> {
+ if (processing.getCount() > 0) {
+ processing.countDown();
+ assertTrue(release.await(20,
TimeUnit.SECONDS), "the test did not release the route");
+ }
+ })
+ .to("mock:blocked");
+
+ from("zookeeper://127.0.0.1:" + port + ONCE_NODE +
"?backoff=200").routeId("once")
+ .autoStartup(false)
+ .to("mock:once");
+ }
+ };
+ }
+
+ @Test
+ void consumesAgainAfterTheSessionExpired() throws Exception {
+ context.getRouteController().startRoute("watch");
+ MockEndpoint mock = getMockEndpoint("mock:data");
+ await().atMost(20, TimeUnit.SECONDS).until(() -> received(mock,
"one"));
+
+ ZooKeeperEndpoint endpoint = (ZooKeeperEndpoint)
context.getRoute("watch").getEndpoint();
+ ZooKeeper consumerHandle =
endpoint.getConnectionManager().getConnection();
+ expire(consumerHandle);
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
!consumerHandle.getState().isAlive());
+
+ client.setData(NODE, bytes("two"), -1);
+
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() ->
assertTrue(received(mock, "two"),
+ "the consumer did not consume the update made after its
session had expired"));
+
+ // "two" may come from the read of the restarted sequence; "three" can
only come from its new watch
+ updateWhenTheConsumerWaitsForAChange(endpoint, NODE, "three");
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() ->
assertTrue(received(mock, "three"),
+ "the consumer did not watch the node again on its new
session"));
+ }
+
+ @Test
+ void consumesAgainAfterTheSessionExpiredWhileAnExchangeWasProcessed()
throws Exception {
+ context.getRouteController().startRoute("blocked");
+ assertTrue(processing.await(20, TimeUnit.SECONDS), "the route did not
receive the first data");
+
+ // the session expires while the operations thread routes the first
exchange: no watch of the consumer is
+ // registered at that time, so the consumer gets no Expired event; the
next watch request fails instead
+ ZooKeeperEndpoint endpoint = (ZooKeeperEndpoint)
context.getRoute("blocked").getEndpoint();
+ ZooKeeper consumerHandle =
endpoint.getConnectionManager().getConnection();
+ expire(consumerHandle);
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
!consumerHandle.getState().isAlive());
+ client.setData(BLOCKED_NODE, bytes("two"), -1);
+ release.countDown();
+
+ MockEndpoint mock = getMockEndpoint("mock:blocked");
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() ->
assertTrue(received(mock, "two"),
+ "the consumer did not consume the update made after its
session had expired during an exchange"));
+
+ updateWhenTheConsumerWaitsForAChange(endpoint, BLOCKED_NODE, "three");
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() ->
assertTrue(received(mock, "three"),
+ "the consumer did not watch the node again on its new
session"));
+ }
+
+ @Test
+ void emitsTheCurrentDataAgainAfterTheSessionExpiredWithoutRepeat() throws
Exception {
+ context.getRouteController().startRoute("once");
+ MockEndpoint mock = getMockEndpoint("mock:once");
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
mock.getReceivedCounter() == 1);
+
+ ZooKeeperEndpoint endpoint = (ZooKeeperEndpoint)
context.getRoute("once").getEndpoint();
+ ZooKeeper consumerHandle =
endpoint.getConnectionManager().getConnection();
+ expire(consumerHandle);
+
+ // the consumer starts its sequence again on a new session, which
reads the (unchanged) data once more
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(2, mock.getReceivedCounter(),
+ "the consumer did not read the data again after its session
had expired"));
+ assertTrue(mock.getReceivedExchanges().stream()
+ .allMatch(e -> "one".equals(new
String(e.getIn().getBody(byte[].class), StandardCharsets.UTF_8))));
+ }
+
+ @Test
+ void producerWritesAgainAfterTheSessionExpired() throws Exception {
+ template.sendBody("zookeeper://127.0.0.1:" + port + NODE, "three");
+ ZooKeeperEndpoint endpoint =
context.getEndpoint("zookeeper://127.0.0.1:" + port + NODE,
ZooKeeperEndpoint.class);
+ ZooKeeper producerHandle =
endpoint.getConnectionManager().getConnection();
+ expire(producerHandle);
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
!producerHandle.getState().isAlive());
+
+ Exchange reply = template.request("zookeeper://127.0.0.1:" + port +
NODE, e -> e.getIn().setBody("four"));
+ assertNull(reply.getException(), "the producer failed after its
session had expired");
+ assertEquals("four", new String(client.getData(NODE, false, null),
StandardCharsets.UTF_8));
+ }
+
+ /**
+ * Waits until the consumer's operations thread waits for a change of a
node (its watch request is sent), then
+ * updates the node through the consumer's own session: ZooKeeper handles
the requests of one session in order, so
+ * the update comes after the watch is registered and the change cannot be
missed.
+ */
+ private static void updateWhenTheConsumerWaitsForAChange(ZooKeeperEndpoint
endpoint, String node, String data)
+ throws Exception {
+ await().atMost(20,
TimeUnit.SECONDS).until(ZooKeeperConsumerSessionExpiryTest::operationsThreadWaitsForAChange);
+ endpoint.getConnectionManager().getConnection().setData(node,
bytes(data), -1);
+ }
+
+ private static boolean operationsThreadWaitsForAChange() {
+ return Thread.getAllStackTraces().entrySet().stream()
+ .filter(e -> e.getKey().getName().contains("Camel-Zookeeper
OperationsExecutor"))
+ .filter(e -> e.getKey().getState() == Thread.State.WAITING)
+ .anyMatch(e -> Arrays.stream(e.getValue())
+ .anyMatch(f ->
FutureEventDrivenOperation.class.getName().equals(f.getClassName())
+ && "get".equals(f.getMethodName())));
+ }
+
+ private static boolean received(MockEndpoint mock, String body) {
+ return mock.getReceivedExchanges().stream()
+ .anyMatch(e -> body.equals(new
String(e.getIn().getBody(byte[].class), StandardCharsets.UTF_8)));
+ }
+
+ private ZooKeeper connect(long sessionId, byte[] password) throws
Exception {
+ CountDownLatch connected = new CountDownLatch(1);
+ ZooKeeper zk = password == null
+ ? new ZooKeeper("127.0.0.1:" + port, 10000, e ->
countDownWhenConnected(e.getState(), connected))
+ : new ZooKeeper(
+ "127.0.0.1:" + port, 10000, e ->
countDownWhenConnected(e.getState(), connected), sessionId,
+ password);
+ assertTrue(connected.await(20, TimeUnit.SECONDS), "the test client did
not connect");
+ return zk;
+ }
+
+ private static void countDownWhenConnected(KeeperState state,
CountDownLatch connected) {
+ if (state == KeeperState.SyncConnected) {
+ connected.countDown();
+ }
+ }
+
+ /**
+ * Closes the session of the given handle from a second client, so that
the server expires it for the handle.
+ */
+ private void expire(ZooKeeper handle) throws Exception {
+ ZooKeeper other = connect(handle.getSessionId(),
handle.getSessionPasswd());
+ other.close();
+ }
+
+ private static byte[] bytes(String s) {
+ return s.getBytes(StandardCharsets.UTF_8);
+ }
+}
diff --git a/parent/pom.xml b/parent/pom.xml
index e9a4b88d7dc0..bcbddcb349ef 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -501,6 +501,7 @@
<smooks-version>2.2.1</smooks-version>
<snakeyaml-version>2.7</snakeyaml-version>
<snakeyaml-engine-version>3.2</snakeyaml-engine-version>
+ <snappy-java-version>1.1.10.8</snappy-java-version>
<snmp4j-version>3.13.1</snmp4j-version>
<solr-version>10.0.0</solr-version>
<splunk-version>1.9.5_1</splunk-version>
