FrankChen021 commented on code in PR #20476:
URL: https://github.com/apache/druid/pull/20476#discussion_r4173130107
##########
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:
[P1] Re-arm the stale-node watch after reconnect
**Finding:** This is a one-shot ZooKeeper watch tied to the announcer's
current session. If that session expires after the create returns NODEEXISTS
but before the old ephemeral node expires, ZooKeeper removes the watch; if the
old node is then deleted while the client is disconnected, CuratorCache can
still have no stored entry in exactly the cache-miss race this change targets,
so reconnecting produces no deletion event and the service remains
undiscoverable until another restart.
**Suggestion:** Register and remove a connection-state listener that
reissues the existence watch on every reconnect, or use a managed persistent
watcher that covers the session-expiration gap.
##########
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:
[P2] Guard reannounce against concurrent unannounce
**Finding:** The payload is read without the lifecycle lock and the create
is queued afterward. If a NodeDeleted callback is already in flight while
unannounce or stop removes this path, the old node can be deleted before this
create runs, leaving a new ephemeral node after the caller has successfully
unannounced or stopped it.
**Suggestion:** Synchronize the payload and started-state check with
unannounce and stop, then revalidate immediately before issuing the create.
##########
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:
[P2] Guard reannounce against concurrent unannounce
**Finding:** The payload is read without the lifecycle lock and the create
is queued afterward. If a NodeDeleted callback is already in flight while
unannounce or stop removes this path, the old node can be deleted before this
create runs, leaving a new ephemeral node after the caller has successfully
unannounced or stopped it; the identical race exists in PathChildrenAnnouncer.
**Suggestion:** Synchronize the payload and started-state check with
unannounce and stop, then revalidate immediately before issuing the create in
both announcers.
--
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]
