allthingssecurity commented on code in PR #27858:
URL: https://github.com/apache/camel/pull/27858#discussion_r4279102281
##########
components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java:
##########
@@ -33,43 +33,71 @@
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;
- }
- 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) {
Review Comment:
Done in a233c3b: `getZooKeeper()` now returns the volatile handle without
taking the monitor when it is alive, and re-reads it inside the lock otherwise.
camel-zookeeper unit suite: 10 tests, 0 failures (incl. the 4 session-expiry
tests).
_Claude Code on behalf of allthingssecurity_
##########
components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java:
##########
@@ -199,11 +208,16 @@ private void backoffAndThenRestart() {
.build()
.run(getEndpoint().getCamelContext(), () -> true);
- initializeConsumer();
+ if (isRunAllowed()) {
+ connection = zkm.getConnection();
+ initializeConsumer();
Review Comment:
Done in a233c3b: after `getConnection()` the restart re-checks
`isRunAllowed()` and calls `zkm.shutdown()` when the consumer was stopped
meanwhile, so the handle and its client threads do not outlive the route. If
the stop comes after the re-check, `doStop` closes the new handle as before. No
dedicated test for this window (it needs a stop timed between the two calls);
the module suite passes.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java:
##########
@@ -66,9 +66,8 @@ public ZooKeeperProducer(ZooKeeperEndpoint endpoint) {
@Override
public void process(Exchange exchange) throws Exception {
- if (connection == null) {
- connection = this.zkm.getConnection();
- }
+ // a new connection when the session of the previous one expired
+ connection = this.zkm.getConnection();
Review Comment:
Done in a233c3b: `process()` keeps the handle in a local `zk`, which the
context and both asynchronous calls use.
_Claude Code on behalf of allthingssecurity_
--
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]
