amaechler commented on code in PR #20476:
URL: https://github.com/apache/druid/pull/20476#discussion_r4174752803
##########
server/src/main/java/org/apache/druid/curator/announcement/NodeAnnouncer.java:
##########
@@ -94,6 +96,17 @@ 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 -> {
Review Comment:
Fixed in 90ba8da901: `NodeAnnouncer` now registers a connection-state
listener that re-announces every path on `RECONNECTED`, so the create hits
`NODEEXISTS` again and the watch is set again from the new session. Added
`testReinstatesWhenOwnSessionExpiresWhileWatchingStaleNode`, which fails
without the listener. In practice the cache usually recovers here because it
still holds the stale node after reconnecting; this only matters when the cache
missed it too. `PathChildrenAnnouncer` already re-creates its paths after a
session loss.
##########
server/src/main/java/org/apache/druid/curator/announcement/NodeAnnouncer.java:
##########
@@ -327,7 +336,47 @@ public void update(final String path, final byte[] bytes)
private void createAnnouncement(final String path, byte[] value) throws
Exception
{
-
curator.create().compressed().withMode(CreateMode.EPHEMERAL).inBackground().forPath(path,
value);
+ 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);
+ }
+
+ /**
+ * 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) throws Exception
+ {
+ final byte[] value = announcedPaths.get(path);
Review Comment:
Fixed in 90ba8da901: the create is now queued inside `computeIfPresent` on
the announcement's entry, so a concurrent `unannounce` either removes the entry
first (no create) or waits until the create is queued, and its delete follows
the create (one session's requests are handled in order while connected). This
race already existed in the cache's removal path; I reproduced it with a forced
interleaving before the fix and confirmed it no longer happens after.
##########
server/src/main/java/org/apache/druid/curator/announcement/PathChildrenAnnouncer.java:
##########
@@ -388,7 +392,49 @@ public void update(final String path, final byte[] bytes)
private void createAnnouncement(final String path, byte[] value) throws
Exception
{
-
curator.create().compressed().withMode(CreateMode.EPHEMERAL).inBackground().forPath(path,
value);
+ 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);
+ }
+
+ /**
+ * 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) throws Exception
+ {
+ final ZKPaths.PathAndNode pathAndNode = ZKPaths.getPathAndNode(path);
+ final ConcurrentMap<String, byte[]> subPaths =
announcements.get(pathAndNode.getPath());
Review Comment:
Same fix as in `NodeAnnouncer` (90ba8da901): the create is queued inside
`computeIfPresent` on the entry, so a concurrent `unannounce` either prevents
it or deletes the node afterwards.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
