This is an automated email from the ASF dual-hosted git repository.
amaechler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new c55da2ae85f fix: re-announce a restarted service once old ZooKeeper
entry expires (#20476)
c55da2ae85f is described below
commit c55da2ae85ff46dff067808aa3bda9cfa82f0337
Author: Andreas Maechler <[email protected]>
AuthorDate: Tue Oct 20 08:59:39 2026 -0600
fix: re-announce a restarted service once old ZooKeeper entry expires
(#20476)
* Re-create announcements blocked by a stale ephemeral node
When a process restarts within the ZooKeeper session timeout, the ephemeral
node of its previous session still exists at the announcement path, so the
announcer's create fails with NODEEXISTS. Both announcers relied on their
cache to report the removal of that node, but PathChildrenCache and
CuratorCache only report removals of nodes they have already loaded. If
ZooKeeper deletes the stale node before the cache's initial load sees it,
the announcement is never re-created and the process stays unregistered
until it restarts.
Handle NODEEXISTS in a create callback: set an exists watch on the path and
re-create the announcement once the node is gone, as Curator's
PersistentNode does. All re-create paths, including the existing cache
listeners and the reconnect handler, now go through reinstateAnnouncement,
which uses the current payload and skips paths that were unannounced in the
meantime. Creates and exists checks that fail for any other reason are
logged as warnings instead of being dropped.
The existing tests waited for CuratorListener CREATE events, which Curator
does not deliver once a create has a callback, so they now poll until the
node exists and belongs to the announcer's session, using helpers in
CuratorTestBase. New tests reproduce the race deterministically, both when
the stale node is removed before the exists watch is set and after it.
* Re-arm stale-node watch after reconnect, guard reinstate vs unannounce
NodeAnnouncer had no reconnect handling, so if its own session expired
while it waited for a stale node to go away, the exists watch was lost
and nothing re-created the announcement unless the cache happened to
hold the stale node. Re-announce every path on RECONNECTED: the create
hits NODEEXISTS again and watches the node from the new session.
Both announcers read the payload and then queued the create, so an
unannounce running in between could leave a node behind. Queue the
create inside computeIfPresent on the announcement's entry: a concurrent
unannounce now either removes the entry first or waits, and its delete
follows the create.
createAnnouncement now wraps Curator's checked exception itself, which
removes two try/catch blocks.
* Drop unused throws clause from PathChildrenAnnouncer listener
---
.../druid/curator/announcement/NodeAnnouncer.java | 94 ++++++++++---
.../announcement/PathChildrenAnnouncer.java | 86 +++++++++---
.../org/apache/druid/curator/CuratorTestBase.java | 74 +++++++++-
.../curator/announcement/NodeAnnouncerTest.java | 152 ++++++++++++++++-----
.../announcement/PathChildrenAnnouncerTest.java | 122 +++++++++++------
5 files changed, 418 insertions(+), 110 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/curator/announcement/NodeAnnouncer.java
b/server/src/main/java/org/apache/druid/curator/announcement/NodeAnnouncer.java
index d38bf79fa38..a5270d5b89e 100644
---
a/server/src/main/java/org/apache/druid/curator/announcement/NodeAnnouncer.java
+++
b/server/src/main/java/org/apache/druid/curator/announcement/NodeAnnouncer.java
@@ -22,10 +22,13 @@ package org.apache.druid.curator.announcement;
import com.google.common.annotations.VisibleForTesting;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.api.transaction.CuratorMultiTransaction;
import org.apache.curator.framework.api.transaction.CuratorOp;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.utils.ZKPaths;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.ISE;
@@ -35,6 +38,7 @@ import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
import java.io.IOException;
import java.util.ArrayList;
@@ -94,6 +98,26 @@ public class NodeAnnouncer implements ServiceAnnouncer
@GuardedBy("toAnnounce")
private final List<String> parentsIBuilt = new CopyOnWriteArrayList<>();
+ /**
+ * Watches a node that already existed when we tried to create our
announcement, see {@link #watchExistingNode}.
+ */
+ private final CuratorWatcher existingNodeWatcher = event -> {
+ if (event.getType() == Watcher.Event.EventType.NodeDeleted) {
+ reinstateAnnouncement(event.getPath());
+ } else if (event.getType() == Watcher.Event.EventType.NodeDataChanged) {
+ watchExistingNode(event.getPath());
+ }
+ };
+
+ /**
+ * Re-announces every path after a reconnect, since watches set on an
expired session are gone.
+ */
+ private final ConnectionStateListener reconnectListener = (client, newState)
-> {
+ if (newState == ConnectionState.RECONNECTED) {
+ announcedPaths.keySet().forEach(this::reinstateAnnouncement);
+ }
+ };
+
public NodeAnnouncer(CuratorFramework curator, ExecutorService exec)
{
this.curator = curator;
@@ -118,6 +142,7 @@ public class NodeAnnouncer implements ServiceAnnouncer
}
started = true;
+ curator.getConnectionStateListenable().addListener(reconnectListener);
for (Announceable announceable : toAnnounce) {
announce(announceable.path, announceable.bytes,
announceable.removeParentsIfCreated);
@@ -150,6 +175,7 @@ public class NodeAnnouncer implements ServiceAnnouncer
@GuardedBy("toAnnounce")
private void closeResources()
{
+ curator.getConnectionStateListenable().removeListener(reconnectListener);
try {
// Close all caches...
CloseableUtils.closeAll(listeners.values());
@@ -263,12 +289,7 @@ public class NodeAnnouncer implements ServiceAnnouncer
}
if (created) {
- try {
- createAnnouncement(path, bytes);
- }
- catch (Exception e) {
- throw new RuntimeException(e);
- }
+ createAnnouncement(path, bytes);
}
}
@@ -280,16 +301,7 @@ public class NodeAnnouncer implements ServiceAnnouncer
cache.listenable().addListener(
(type, oldData, data) -> {
if (type == CuratorCacheListener.Type.NODE_DELETED) {
- final byte[] previouslyAnnouncedData = announcedPaths.get(path);
- if (previouslyAnnouncedData != null) {
- try {
- log.info("ZooKeeper Node[%s] dropped, reinstating...", path);
- createAnnouncement(path, previouslyAnnouncedData);
- }
- catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
+ reinstateAnnouncement(path);
}
}, nodeCacheExecutor
);
@@ -325,9 +337,55 @@ public class NodeAnnouncer implements ServiceAnnouncer
}
}
- private void createAnnouncement(final String path, byte[] value) throws
Exception
+ private void createAnnouncement(final String path, byte[] value)
+ {
+ try {
+
curator.create().compressed().withMode(CreateMode.EPHEMERAL).inBackground(
+ (client, event) -> {
+ if (event.getResultCode() ==
KeeperException.Code.NODEEXISTS.intValue()) {
+ watchExistingNode(path);
+ } else if (event.getResultCode() !=
KeeperException.Code.OK.intValue()) {
+ log.warn("Failed to announce node[%s]: %s", path,
KeeperException.Code.get(event.getResultCode()));
+ }
+ }
+ ).forPath(path, value);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Watches a node that was already present when we tried to create our
announcement at its path, for example one
+ * left behind by a previous session of this process. The announcement is
re-created once that node is removed,
+ * even if the {@link CuratorCache} never saw it.
+ */
+ private void watchExistingNode(final String path) throws Exception
+ {
+ curator.checkExists().usingWatcher(existingNodeWatcher).inBackground(
+ (client, event) -> {
+ if (event.getResultCode() == KeeperException.Code.NONODE.intValue())
{
+ reinstateAnnouncement(path);
+ } else if (event.getResultCode() ==
KeeperException.Code.OK.intValue()) {
+ final long owner = event.getStat().getEphemeralOwner();
+ if (owner !=
curator.getZookeeperClient().getZooKeeper().getSessionId()) {
+ log.info("Node[%s] already exists with session[0x%x], will
reinstate once it is removed.", path, owner);
+ }
+ } else {
+ log.warn("Failed to watch existing node[%s]: %s", path,
KeeperException.Code.get(event.getResultCode()));
+ }
+ }
+ ).forPath(path);
+ }
+
+ private void reinstateAnnouncement(final String path)
{
-
curator.create().compressed().withMode(CreateMode.EPHEMERAL).inBackground().forPath(path,
value);
+ // Create while holding the entry, so a concurrent unannounce either
prevents it or deletes the node afterwards.
+ announcedPaths.computeIfPresent(path, (p, value) -> {
+ log.info("ZooKeeper Node[%s] dropped, reinstating...", path);
+ createAnnouncement(path, value);
+ return value;
+ });
}
private void updateAnnouncement(final String path, final byte[] value)
throws Exception
diff --git
a/server/src/main/java/org/apache/druid/curator/announcement/PathChildrenAnnouncer.java
b/server/src/main/java/org/apache/druid/curator/announcement/PathChildrenAnnouncer.java
index ec754d96be2..ad9f1073c80 100644
---
a/server/src/main/java/org/apache/druid/curator/announcement/PathChildrenAnnouncer.java
+++
b/server/src/main/java/org/apache/druid/curator/announcement/PathChildrenAnnouncer.java
@@ -22,9 +22,9 @@ package org.apache.druid.curator.announcement;
import com.google.common.annotations.VisibleForTesting;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.api.transaction.CuratorMultiTransaction;
import org.apache.curator.framework.api.transaction.CuratorOp;
-import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
@@ -38,6 +38,7 @@ import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
import java.io.IOException;
import java.util.ArrayList;
@@ -84,6 +85,17 @@ public class PathChildrenAnnouncer implements
ServiceAnnouncer
private boolean started = false;
+ /**
+ * Watches a node that already existed when we tried to create our
announcement, see {@link #watchExistingNode}.
+ */
+ private final CuratorWatcher existingNodeWatcher = event -> {
+ if (event.getType() == Watcher.Event.EventType.NodeDeleted) {
+ reinstateAnnouncement(event.getPath());
+ } else if (event.getType() == Watcher.Event.EventType.NodeDataChanged) {
+ watchExistingNode(event.getPath());
+ }
+ };
+
public PathChildrenAnnouncer(
CuratorFramework curator,
ExecutorService exec
@@ -252,20 +264,14 @@ public class PathChildrenAnnouncer implements
ServiceAnnouncer
private final AtomicReference<Set<String>> pathsLost = new
AtomicReference<>(null);
@Override
- public void childEvent(CuratorFramework client,
PathChildrenCacheEvent event) throws Exception
+ public void childEvent(CuratorFramework client,
PathChildrenCacheEvent event)
{
// NOTE: ZooKeeper does not guarantee that we will get every
event, and thus PathChildrenCache doesn't
// as well. If one of the below events are missed, Announcer
might not work properly.
log.debug("Path[%s] got event[%s]", parentPath, event);
switch (event.getType()) {
case CHILD_REMOVED:
- final ChildData child = event.getData();
- final ZKPaths.PathAndNode childPath =
ZKPaths.getPathAndNode(child.getPath());
- final byte[] value =
finalSubPaths.get(childPath.getNode());
- if (value != null) {
- log.info("Node[%s] dropped, reinstating.",
child.getPath());
- createAnnouncement(child.getPath(), value);
- }
+ reinstateAnnouncement(event.getData().getPath());
break;
case CONNECTION_LOST:
// Lost connection, which means session is broken, take
inventory of what has been seen.
@@ -292,9 +298,7 @@ public class PathChildrenAnnouncer implements
ServiceAnnouncer
if (thePathsLost != null) {
for (String path : thePathsLost) {
- log.info("Reinstating [%s]", path);
- final ZKPaths.PathAndNode split =
ZKPaths.getPathAndNode(path);
- createAnnouncement(path,
announcements.get(split.getPath()).get(split.getNode()));
+ reinstateAnnouncement(path);
}
}
break;
@@ -341,12 +345,7 @@ public class PathChildrenAnnouncer implements
ServiceAnnouncer
}
if (created) {
- try {
- createAnnouncement(path, bytes);
- }
- catch (Exception e) {
- throw new RuntimeException(e);
- }
+ createAnnouncement(path, bytes);
}
}
@@ -386,9 +385,56 @@ public class PathChildrenAnnouncer implements
ServiceAnnouncer
}
}
- private void createAnnouncement(final String path, byte[] value) throws
Exception
+ private void createAnnouncement(final String path, byte[] value)
{
-
curator.create().compressed().withMode(CreateMode.EPHEMERAL).inBackground().forPath(path,
value);
+ try {
+
curator.create().compressed().withMode(CreateMode.EPHEMERAL).inBackground(
+ (client, event) -> {
+ if (event.getResultCode() ==
KeeperException.Code.NODEEXISTS.intValue()) {
+ watchExistingNode(path);
+ } else if (event.getResultCode() !=
KeeperException.Code.OK.intValue()) {
+ log.warn("Failed to announce node[%s]: %s", path,
KeeperException.Code.get(event.getResultCode()));
+ }
+ }
+ ).forPath(path, value);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Watches a node that was already present when we tried to create our
announcement at its path, for example one
+ * left behind by a previous session of this process. The announcement is
re-created once that node is removed,
+ * even if the {@link PathChildrenCache} never saw it.
+ */
+ private void watchExistingNode(final String path) throws Exception
+ {
+ curator.checkExists().usingWatcher(existingNodeWatcher).inBackground(
+ (client, event) -> {
+ if (event.getResultCode() == KeeperException.Code.NONODE.intValue())
{
+ reinstateAnnouncement(path);
+ } else if (event.getResultCode() ==
KeeperException.Code.OK.intValue()) {
+ final long owner = event.getStat().getEphemeralOwner();
+ if (owner !=
curator.getZookeeperClient().getZooKeeper().getSessionId()) {
+ log.info("Node[%s] already exists with session[0x%x], will
reinstate once it is removed.", path, owner);
+ }
+ } else {
+ log.warn("Failed to watch existing node[%s]: %s", path,
KeeperException.Code.get(event.getResultCode()));
+ }
+ }
+ ).forPath(path);
+ }
+
+ private void reinstateAnnouncement(final String path)
+ {
+ final ZKPaths.PathAndNode pathAndNode = ZKPaths.getPathAndNode(path);
+ // Create while holding the entry, so a concurrent unannounce either
prevents it or deletes the node afterwards.
+
announcements.get(pathAndNode.getPath()).computeIfPresent(pathAndNode.getNode(),
(node, value) -> {
+ log.info("Node[%s] dropped, reinstating.", path);
+ createAnnouncement(path, value);
+ return value;
+ });
}
private void updateAnnouncement(final String path, final byte[] value)
throws Exception
diff --git a/server/src/test/java/org/apache/druid/curator/CuratorTestBase.java
b/server/src/test/java/org/apache/druid/curator/CuratorTestBase.java
index c58999565d7..a481fa3e059 100644
--- a/server/src/test/java/org/apache/druid/curator/CuratorTestBase.java
+++ b/server/src/test/java/org/apache/druid/curator/CuratorTestBase.java
@@ -24,8 +24,15 @@ import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.test.TestingServer;
import org.apache.curator.test.Timing;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.Stat;
+import org.junit.jupiter.api.Assertions;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
/**
*/
@@ -39,7 +46,12 @@ public class CuratorTestBase
{
server = new TestingServer();
timing = new Timing();
- curator = CuratorFrameworkFactory
+ curator = createCurator();
+ }
+
+ private CuratorFramework createCurator()
+ {
+ return CuratorFrameworkFactory
.builder()
.connectString(server.getConnectString())
.sessionTimeoutMs(timing.session())
@@ -64,4 +76,64 @@ public class CuratorTestBase
{
return server.getConnectString();
}
+
+ /**
+ * Starts a new client, with its own session, that owns an ephemeral node at
the given path.
+ */
+ public CuratorFramework createEphemeralNodeInNewSession(final String path,
final byte[] bytes) throws Exception
+ {
+ final CuratorFramework client = createCurator();
+ try {
+ client.start();
+ client.blockUntilConnected();
+
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path,
bytes);
+ return client;
+ }
+ catch (Exception e) {
+ client.close();
+ throw e;
+ }
+ }
+
+ /**
+ * Waits until the node at the given path holds the given bytes and belongs
to the session of {@link #curator}.
+ */
+ public void awaitAnnounced(final String path, final byte[] bytes) throws
Exception
+ {
+ final long deadline = System.currentTimeMillis() +
timing.forWaiting().milliseconds();
+ while (!isAnnounced(path, bytes)) {
+ Assertions.assertTrue(System.currentTimeMillis() < deadline, "Timed out
waiting for " + path);
+ Thread.sleep(100);
+ }
+ }
+
+ private boolean isAnnounced(final String path, final byte[] bytes) throws
Exception
+ {
+ final Stat stat = new Stat();
+ try {
+ final byte[] data =
curator.getData().decompressed().storingStatIn(stat).forPath(path);
+ return Arrays.equals(bytes, data)
+ && stat.getEphemeralOwner() ==
curator.getZookeeperClient().getZooKeeper().getSessionId();
+ }
+ catch (KeeperException e) {
+ // Missing node, or the client is reconnecting.
+ return false;
+ }
+ }
+
+ /**
+ * Blocks the ZooKeeper event thread of {@link #curator}, which delays all
of its background callbacks and watch
+ * notifications, until the returned latch is counted down.
+ */
+ public CountDownLatch blockEventThread() throws Exception
+ {
+ final CountDownLatch blocked = new CountDownLatch(1);
+ final CountDownLatch release = new CountDownLatch(1);
+ curator.checkExists().inBackground((client, event) -> {
+ blocked.countDown();
+ release.await(timing.forWaiting().milliseconds(), TimeUnit.MILLISECONDS);
+ }).forPath("/");
+ Assertions.assertTrue(timing.awaitLatch(blocked));
+ return release;
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/curator/announcement/NodeAnnouncerTest.java
b/server/src/test/java/org/apache/druid/curator/announcement/NodeAnnouncerTest.java
index 6841f87b314..578536a311e 100644
---
a/server/src/test/java/org/apache/druid/curator/announcement/NodeAnnouncerTest.java
+++
b/server/src/test/java/org/apache/druid/curator/announcement/NodeAnnouncerTest.java
@@ -19,7 +19,7 @@
package org.apache.druid.curator.announcement;
-import org.apache.curator.framework.api.CuratorEventType;
+import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.transaction.CuratorOp;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.test.KillSession;
@@ -29,6 +29,7 @@ import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.testing.junit.LoggerCaptureExtension;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;
import org.junit.jupiter.api.AfterEach;
@@ -36,14 +37,18 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
-import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
public class NodeAnnouncerTest extends CuratorTestBase
{
+ @RegisterExtension
+ public final LoggerCaptureExtension announcerLogs = new
LoggerCaptureExtension(NodeAnnouncer.class);
+
private ExecutorService exec;
@BeforeEach
@@ -194,19 +199,13 @@ public class NodeAnnouncerTest extends CuratorTestBase
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath1), "/test1 still has data");
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath2), "/somewhere/test2 has
data");
- final CountDownLatch latch = new CountDownLatch(1);
- curator.getCuratorListenable().addListener((client, event) -> {
- if (event.getType() == CuratorEventType.CREATE &&
event.getPath().equals(testPath1)) {
- latch.countDown();
- }
- });
final CuratorOp deleteOp =
curator.transactionOp().delete().forPath(testPath1);
final Collection<CuratorTransactionResult> results =
curator.transaction().forOperations(deleteOp);
Assertions.assertEquals(1, results.size(), "Expected one result from the
delete op");
final CuratorTransactionResult result = results.iterator().next();
Assertions.assertEquals(Code.OK.intValue(), result.getError(), "Expected
OK code on delete");
- Assertions.assertTrue(timing.forWaiting().awaitLatch(latch), "Wait for
/test1 to be recreated");
+ awaitAnnounced(testPath1, billy);
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath1), "Expected /test1 data to
be restored");
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath2), "Expected /somewhere/test2
data to remain");
@@ -236,17 +235,16 @@ public class NodeAnnouncerTest extends CuratorTestBase
final byte[] billy = StringUtils.toUtf8("billy");
final String testPath1 = "/test1";
final String testPath2 = "/somewhere/test2";
- final String[] paths = new String[]{testPath1, testPath2};
announcer.announce(testPath1, billy);
announcer.announce(testPath2, billy);
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath1));
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath2));
- final CountDownLatch latch = createCountdownLatchForPaths(paths);
KillSession.kill(curator.getZookeeperClient().getZooKeeper(),
server.getConnectString());
- Assertions.assertTrue(timing.forWaiting().awaitLatch(latch), "Await
latch after killing session");
+ awaitAnnounced(testPath1, billy);
+ awaitAnnounced(testPath2, billy);
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath1));
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath2));
@@ -279,7 +277,9 @@ public class NodeAnnouncerTest extends CuratorTestBase
try {
Assertions.assertNull(curator.checkExists().forPath(parent));
- awaitAnnounce(announcer, testPath, billy, true);
+ announcer.announce(testPath, billy, true);
+
+ awaitAnnounced(testPath, billy);
Assertions.assertNotNull(curator.checkExists().forPath(parent));
}
@@ -306,7 +306,9 @@ public class NodeAnnouncerTest extends CuratorTestBase
try {
Assertions.assertEquals(initialStat.getMzxid(),
curator.checkExists().forPath(parent).getMzxid());
- awaitAnnounce(announcer, testPath, billy, true);
+ announcer.announce(testPath, billy, true);
+
+ awaitAnnounced(testPath, billy);
Assertions.assertEquals(initialStat.getMzxid(),
curator.checkExists().forPath(parent).getMzxid());
}
@@ -330,7 +332,9 @@ public class NodeAnnouncerTest extends CuratorTestBase
try {
Assertions.assertNull(curator.checkExists().forPath(parent));
- awaitAnnounce(announcer, testPath, billy, false);
+ announcer.announce(testPath, billy, false);
+
+ awaitAnnounced(testPath, billy);
Assertions.assertNotNull(curator.checkExists().forPath(parent));
}
@@ -341,27 +345,111 @@ public class NodeAnnouncerTest extends CuratorTestBase
Assertions.assertNotNull(curator.checkExists().forPath(parent));
}
- private void awaitAnnounce(
- final NodeAnnouncer announcer,
- final String path,
- final byte[] bytes,
- boolean removeParentsIfCreated
- ) throws InterruptedException
+ /**
+ * A restarted process announces a path whose node still belongs to its
previous session, and that session expires
+ * before the announcer's cache has seen the node.
+ */
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
+ public void testReinstatesWhenStaleNodeIsRemovedBeforeCacheSeesIt() throws
Exception
{
- final CountDownLatch latch = createCountdownLatchForPaths(path);
- announcer.announce(path, bytes, removeParentsIfCreated);
- latch.await();
+ final byte[] billy = StringUtils.toUtf8("billy");
+ final String testPath = "/somewhere/test";
+ final NodeAnnouncer announcer = new NodeAnnouncer(curator, exec);
+
+ try (CuratorFramework staleCurator =
createEphemeralNodeInNewSession(testPath, StringUtils.toUtf8("stale"))) {
+ final long staleSessionId =
staleCurator.getZookeeperClient().getZooKeeper().getSessionId();
+
+ final CountDownLatch release = blockEventThread();
+ try {
+ announcer.start();
+ announcer.announce(testPath, billy);
+
+ // Requests of a session are processed in order, so the announcer's
create has already failed with NODEEXISTS.
+ Assertions.assertEquals(staleSessionId,
curator.checkExists().forPath(testPath).getEphemeralOwner());
+
+ staleCurator.close();
+ Assertions.assertNull(curator.checkExists().forPath(testPath));
+ }
+ finally {
+ release.countDown();
+ }
+
+ awaitAnnounced(testPath, billy);
+ }
+ finally {
+ announcer.stop();
+ }
}
- private CountDownLatch createCountdownLatchForPaths(String... paths)
+ /**
+ * The stale node is removed only after the announcer started watching it,
while the announcer's cache cannot react
+ * because its executor is busy.
+ */
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
+ public void testReinstatesWhenWatchedStaleNodeIsRemoved() throws Exception
{
- final CountDownLatch latch = new CountDownLatch(paths.length);
- curator.getCuratorListenable().addListener((client, event) -> {
- if (event.getType() == CuratorEventType.CREATE &&
Arrays.asList(paths).contains(event.getPath())) {
- latch.countDown();
- }
- });
+ final byte[] billy = StringUtils.toUtf8("billy");
+ final String testPath = "/somewhere/test";
+ final NodeAnnouncer announcer = new NodeAnnouncer(curator, exec);
+ final CountDownLatch releaseExec = new CountDownLatch(1);
+
+ try (CuratorFramework staleCurator =
createEphemeralNodeInNewSession(testPath, StringUtils.toUtf8("stale"))) {
+ exec.submit(() -> releaseExec.await(1, TimeUnit.MINUTES));
+ announcer.start();
+ announcer.announce(testPath, billy);
- return latch;
+ // Logged once the exists watch on the stale node is set.
+ awaitLogEvents("already exists", 1);
+
+ staleCurator.close();
+ awaitAnnounced(testPath, billy);
+ }
+ finally {
+ releaseExec.countDown();
+ announcer.stop();
+ }
+ }
+
+ /**
+ * The announcer's own session expires while it waits for the stale node to
go away, which drops its watch.
+ */
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
+ public void testReinstatesWhenOwnSessionExpiresWhileWatchingStaleNode()
throws Exception
+ {
+ final byte[] billy = StringUtils.toUtf8("billy");
+ final String testPath = "/somewhere/test";
+ final NodeAnnouncer announcer = new NodeAnnouncer(curator, exec);
+ final CountDownLatch releaseExec = new CountDownLatch(1);
+
+ try (CuratorFramework staleCurator =
createEphemeralNodeInNewSession(testPath, StringUtils.toUtf8("stale"))) {
+ exec.submit(() -> releaseExec.await(1, TimeUnit.MINUTES));
+ announcer.start();
+ announcer.announce(testPath, billy);
+ awaitLogEvents("already exists", 1);
+
+ KillSession.kill(curator.getZookeeperClient().getZooKeeper());
+ // After reconnecting, the announcer finds the stale node again and
watches it from the new session.
+ awaitLogEvents("already exists", 2);
+
+ staleCurator.close();
+ awaitAnnounced(testPath, billy);
+ }
+ finally {
+ releaseExec.countDown();
+ announcer.stop();
+ }
+ }
+
+ private void awaitLogEvents(final String text, final long count) throws
InterruptedException
+ {
+ while (announcerLogs.getLogEvents()
+ .stream()
+ .filter(event ->
event.getMessage().getFormattedMessage().contains(text))
+ .count() < count) {
+ Thread.sleep(10);
+ }
}
}
diff --git
a/server/src/test/java/org/apache/druid/curator/announcement/PathChildrenAnnouncerTest.java
b/server/src/test/java/org/apache/druid/curator/announcement/PathChildrenAnnouncerTest.java
index 54d955b4ce4..408913b0e41 100644
---
a/server/src/test/java/org/apache/druid/curator/announcement/PathChildrenAnnouncerTest.java
+++
b/server/src/test/java/org/apache/druid/curator/announcement/PathChildrenAnnouncerTest.java
@@ -20,9 +20,6 @@
package org.apache.druid.curator.announcement;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.api.CuratorEvent;
-import org.apache.curator.framework.api.CuratorEventType;
-import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.transaction.CuratorOp;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.test.KillSession;
@@ -31,6 +28,7 @@ import org.apache.druid.curator.CuratorTestBase;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.testing.junit.LoggerCaptureExtension;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;
import org.junit.jupiter.api.AfterEach;
@@ -38,8 +36,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
-import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -51,6 +49,10 @@ import java.util.concurrent.TimeUnit;
public class PathChildrenAnnouncerTest extends CuratorTestBase
{
private static final Logger log = new
Logger(PathChildrenAnnouncerTest.class);
+
+ @RegisterExtension
+ public final LoggerCaptureExtension announcerLogs = new
LoggerCaptureExtension(PathChildrenAnnouncer.class);
+
private ExecutorService exec;
@BeforeEach
@@ -101,15 +103,13 @@ public class PathChildrenAnnouncerTest extends
CuratorTestBase
"/somewhere/test2 has data"
);
- final CountDownLatch latch = createCountdownLatchForPaths(testPath1);
-
final CuratorOp deleteOp =
curator.transactionOp().delete().forPath(testPath1);
final Collection<CuratorTransactionResult> results =
curator.transaction().forOperations(deleteOp);
Assertions.assertEquals(1, results.size());
final CuratorTransactionResult result = results.iterator().next();
Assertions.assertEquals(Code.OK.intValue(), result.getError()); //
assert delete
- Assertions.assertTrue(timing.forWaiting().awaitLatch(latch), "Wait for
/test1 to be created");
+ awaitAnnounced(testPath1, billy);
Assertions.assertArrayEquals(
billy,
@@ -151,18 +151,16 @@ public class PathChildrenAnnouncerTest extends
CuratorTestBase
final byte[] billy = StringUtils.toUtf8("billy");
final String testPath1 = "/test1";
final String testPath2 = "/somewhere/test2";
- final String[] paths = new String[]{testPath1, testPath2};
announcer.announce(testPath1, billy);
announcer.announce(testPath2, billy);
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath1));
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath2));
- final CountDownLatch latch = createCountdownLatchForPaths(paths);
-
KillSession.kill(curator.getZookeeperClient().getZooKeeper(),
server.getConnectString());
- Assertions.assertTrue(timing.forWaiting().awaitLatch(latch));
+ awaitAnnounced(testPath1, billy);
+ awaitAnnounced(testPath2, billy);
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath1));
Assertions.assertArrayEquals(billy,
curator.getData().decompressed().forPath(testPath2));
@@ -194,7 +192,9 @@ public class PathChildrenAnnouncerTest extends
CuratorTestBase
try {
Assertions.assertNull(curator.checkExists().forPath(parent));
- awaitAnnounce(announcer, testPath, billy, true);
+ announcer.announce(testPath, billy, true);
+
+ awaitAnnounced(testPath, billy);
Assertions.assertNotNull(curator.checkExists().forPath(parent));
}
@@ -221,7 +221,9 @@ public class PathChildrenAnnouncerTest extends
CuratorTestBase
try {
Assertions.assertEquals(initialStat.getMzxid(),
curator.checkExists().forPath(parent).getMzxid());
- awaitAnnounce(announcer, testPath, billy, true);
+ announcer.announce(testPath, billy, true);
+
+ awaitAnnounced(testPath, billy);
Assertions.assertEquals(initialStat.getMzxid(),
curator.checkExists().forPath(parent).getMzxid());
}
@@ -245,7 +247,9 @@ public class PathChildrenAnnouncerTest extends
CuratorTestBase
try {
Assertions.assertNull(curator.checkExists().forPath(parent));
- awaitAnnounce(announcer, testPath, billy, false);
+ announcer.announce(testPath, billy, false);
+
+ awaitAnnounced(testPath, billy);
Assertions.assertNotNull(curator.checkExists().forPath(parent));
}
@@ -256,34 +260,74 @@ public class PathChildrenAnnouncerTest extends
CuratorTestBase
Assertions.assertNotNull(curator.checkExists().forPath(parent));
}
- private void awaitAnnounce(
- final PathChildrenAnnouncer announcer,
- final String path,
- final byte[] bytes,
- boolean removeParentsIfCreated
- ) throws InterruptedException
+ /**
+ * A restarted process announces a path whose node still belongs to its
previous session, and that session expires
+ * before the announcer's cache has seen the node.
+ */
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
+ public void testReinstatesWhenStaleNodeIsRemovedBeforeCacheSeesIt() throws
Exception
{
- CountDownLatch latch = createCountdownLatchForPaths(path);
- announcer.announce(path, bytes, removeParentsIfCreated);
- latch.await();
+ final byte[] billy = StringUtils.toUtf8("billy");
+ final String testPath = "/somewhere/test";
+ final PathChildrenAnnouncer announcer = new PathChildrenAnnouncer(curator,
exec);
+
+ try (CuratorFramework staleCurator =
createEphemeralNodeInNewSession(testPath, StringUtils.toUtf8("stale"))) {
+ final long staleSessionId =
staleCurator.getZookeeperClient().getZooKeeper().getSessionId();
+
+ final CountDownLatch release = blockEventThread();
+ try {
+ announcer.start();
+ announcer.announce(testPath, billy);
+
+ // Requests of a session are processed in order, so the announcer's
create has already failed with NODEEXISTS.
+ Assertions.assertEquals(staleSessionId,
curator.checkExists().forPath(testPath).getEphemeralOwner());
+
+ staleCurator.close();
+ Assertions.assertNull(curator.checkExists().forPath(testPath));
+ }
+ finally {
+ release.countDown();
+ }
+
+ awaitAnnounced(testPath, billy);
+ }
+ finally {
+ announcer.stop();
+ }
}
- private CountDownLatch createCountdownLatchForPaths(String... path)
+ /**
+ * The stale node is removed only after the announcer started watching it,
while the announcer's cache cannot react
+ * because its executor is busy.
+ */
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
+ public void testReinstatesWhenWatchedStaleNodeIsRemoved() throws Exception
{
- final CountDownLatch latch = new CountDownLatch(path.length);
- curator.getCuratorListenable().addListener(
- new CuratorListener()
- {
- @Override
- public void eventReceived(CuratorFramework client, CuratorEvent
event)
- {
- if (event.getType() == CuratorEventType.CREATE &&
Arrays.asList(path).contains(event.getPath())) {
- latch.countDown();
- }
- }
- }
- );
-
- return latch;
+ final byte[] billy = StringUtils.toUtf8("billy");
+ final String testPath = "/somewhere/test";
+ final PathChildrenAnnouncer announcer = new PathChildrenAnnouncer(curator,
exec);
+ final CountDownLatch releaseExec = new CountDownLatch(1);
+
+ try (CuratorFramework staleCurator =
createEphemeralNodeInNewSession(testPath, StringUtils.toUtf8("stale"))) {
+ exec.submit(() -> releaseExec.await(1, TimeUnit.MINUTES));
+ announcer.start();
+ announcer.announce(testPath, billy);
+
+ // Logged once the exists watch on the stale node is set.
+ while (announcerLogs.getLogEvents()
+ .stream()
+ .noneMatch(event ->
event.getMessage().getFormattedMessage().contains("already exists"))) {
+ Thread.sleep(10);
+ }
+
+ staleCurator.close();
+ awaitAnnounced(testPath, billy);
+ }
+ finally {
+ releaseExec.countDown();
+ announcer.stop();
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
