davsclaus commented on code in PR #27858:
URL: https://github.com/apache/camel/pull/27858#discussion_r4279068174
##########
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:
Nit (non-blocking): with concurrent senders the shared `connection` field is
written by several threads. A local variable keeps each message on the handle
it got here (the async branches further down read the field again):
```java
ZooKeeper zk = this.zkm.getConnection();
connection = zk;
ProductionContext context = new ProductionContext(zk, exchange);
```
The `asynchronously...(connection, context)` calls below could then also use
`zk`.
##########
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:
Nit (non-blocking): the producer now calls `getZooKeeper()` for every
message, so each send takes this monitor. Since `zookeeper` is volatile, a fast
path before the lock keeps the common case lock-free:
```java
ZooKeeper current = zookeeper;
if (current != null && current.getState().isAlive()) {
return current;
}
synchronized (this) {
current = zookeeper;
...
```
##########
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:
Suggestion (non-blocking): the PR description mentions that a stop between
the `isRunAllowed()` check and `getConnection()` can create a handle after
`doStop` closed the holder. That handle then keeps its session and its two
client threads (SendThread/EventThread) alive after the route stopped. A
re-check here would close it:
```suggestion
connection = zkm.getConnection();
if (!isRunAllowed()) {
zkm.shutdown();
return false;
}
initializeConsumer();
```
--
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]
