From dbe34a862da1656c717e71aeada8b8499a83699e Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 11 Mar 2026 13:59:36 -0500
Subject: [PATCH 01/13] Potential fix

---
 .../k8s/discovery/DefaultK8sApiClient.java    |  80 ++++++++++++-
 .../K8sDruidNodeDiscoveryProvider.java        |   5 +
 .../druid/k8s/discovery/WatchResult.java      |   9 ++
 .../discovery/DefaultK8sApiClientTest.java    | 107 ++++++++++++++++++
 .../druid/discovery/BaseNodeRoleWatcher.java  |  20 ++++
 5 files changed, 216 insertions(+), 5 deletions(-)
 create mode 100644 extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java

diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
index adf10a42dd49..c50e3ea61f12 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
@@ -28,6 +28,7 @@
 import io.kubernetes.client.openapi.ApiClient;
 import io.kubernetes.client.openapi.ApiException;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1ContainerStatus;
 import io.kubernetes.client.openapi.models.V1Pod;
 import io.kubernetes.client.openapi.models.V1PodList;
 import io.kubernetes.client.util.PatchUtils;
@@ -41,6 +42,7 @@
 import java.io.IOException;
 import java.net.SocketTimeoutException;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -114,6 +116,14 @@ public DiscoveryDruidNodeList listPods(
 
       Map<String, DiscoveryDruidNode> allNodes = new HashMap();
       for (V1Pod podDef : podList.getItems()) {
+        if (!isPodReady(podDef)) {
+          LOGGER.warn(
+              "Ignoring pod [%s] for role [%s] during list: poad has discovery label but is not ready.",
+              podDef.getMetadata().getName(),
+              nodeRole
+          );
+          continue;
+        }
         DiscoveryDruidNode node = getDiscoveryDruidNodeFromPodDef(nodeRole, podDef);
         allNodes.put(node.getDruidNode().getHostAndPortToUse(), node);
       }
@@ -124,6 +134,23 @@ public DiscoveryDruidNodeList listPods(
     }
   }
 
+  /**
+   * Check whether a pod's containers are all running and ready. This is used to filter out pods
+   * whose containers have been OOM-killed or are otherwise not serving traffic, even though the
+   * pod itself still exists and retains its Druid announcement labels.
+   */
+  static boolean isPodReady(V1Pod pod)
+  {
+    if (pod.getStatus() == null) {
+      return false;
+    }
+    List<V1ContainerStatus> containerStatuses = pod.getStatus().getContainerStatuses();
+    if (containerStatuses == null || containerStatuses.isEmpty()) {
+      return false;
+    }
+    return containerStatuses.stream().allMatch(cs -> Boolean.TRUE.equals(cs.getReady()));
+  }
+
   private DiscoveryDruidNode getDiscoveryDruidNodeFromPodDef(NodeRole nodeRole, V1Pod podDef)
   {
     String jsonStr = podDef.getMetadata().getAnnotations().get(K8sDruidNodeAnnouncer.getInfoAnnotation(nodeRole));
@@ -174,11 +201,54 @@ public boolean hasNext() throws SocketTimeoutException
               Watch.Response<V1Pod> item = watch.next();
               if (item != null && item.type != null && !item.type.equals(WatchResult.BOOKMARK)) {
                 DiscoveryDruidNodeAndResourceVersion result = null;
+                String effectiveType = item.type;
+
                 if (item.object != null) {
-                  result = new DiscoveryDruidNodeAndResourceVersion(
-                    item.object.getMetadata().getResourceVersion(),
-                    getDiscoveryDruidNodeFromPodDef(nodeRole, item.object)
-                  );
+                  if (!isPodReady(item.object)) {
+                    if (WatchResult.MODIFIED.equals(item.type)) {
+                      // Pod was previously ready but is now unready (e.g., OOM-killed
+                      // container). Convert to DELETED to remove from discovery cache.
+                      LOGGER.warn(
+                          "Pod [%s] for role [%s] notified that it was modified and is now showing as not ready, "
+                          + "treating as removed for discovery purposes.",
+                          item.object.getMetadata().getName(),
+                          nodeRole
+                      );
+                      effectiveType = WatchResult.NOT_READY;
+                    } else if (WatchResult.ADDED.equals(item.type)) {
+                      // Pod is not ready yet (e.g., still starting up). Skip this event
+                      // entirely — it will appear via a MODIFIED event once it becomes ready.
+                      LOGGER.debug(
+                          "Pod [%s] for role [%s] is not ready on ADDED event, skipping until it becomes ready.",
+                          item.object.getMetadata().getName(),
+                          nodeRole
+                      );
+                      continue;
+                    }
+                  } else if (WatchResult.MODIFIED.equals(item.type)) {
+                    // Pod is ready on a MODIFIED event — it may have just become ready
+                    // (e.g., container restarted after OOM). Treat as ADDED so it gets
+                    // (re-)added to the discovery cache.
+                    effectiveType = WatchResult.ADDED;
+                  }
+
+                  try {
+                    result = new DiscoveryDruidNodeAndResourceVersion(
+                        item.object.getMetadata().getResourceVersion(),
+                        getDiscoveryDruidNodeFromPodDef(nodeRole, item.object)
+                    );
+                  }
+                  catch (Exception ex) {
+                    LOGGER.warn(
+                        ex,
+                        "Failed to deserialize node info from pod [%s] for role [%s] on [%s] event. "
+                        + "Passing null to trigger watch restart and full resync.",
+                        item.object.getMetadata() != null ? item.object.getMetadata().getName() : "unknown",
+                        nodeRole,
+                        item.type
+                    );
+                    // result stays null, caller will restart the watch and do a full listPods resync
+                  }
                 } else {
                   // The item's object can be null in some cases -- likely due to a blip
                   // in the k8s watch. Handle that by passing the null upwards. The caller
@@ -187,7 +257,7 @@ public boolean hasNext() throws SocketTimeoutException
                 }
 
                 obj = new Watch.Response<>(
-                    item.type,
+                    effectiveType,
                     result
                 );
                 return true;
diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
index d2472a9fde4d..e7c44e2d609b 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
@@ -273,6 +273,11 @@ private void keepWatching(String labelSelector, String resourceVersion)
                   case WatchResult.DELETED:
                     baseNodeRoleWatcher.childRemoved(item.object.getNode());
                     break;
+                  case WatchResult.NOT_READY:
+                    // Pod container became not-ready (e.g., OOM-killed). Remove if present
+                    // in cache, but don't log errors for repeated events during CrashLoopBackOff.
+                    baseNodeRoleWatcher.childRemoved(item.object.getNode(), false);
+                    break;
                   default:
                 }
 
diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java
index f47b13757320..c5457cdde065 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java
@@ -26,9 +26,18 @@
 public interface WatchResult
 {
   String ADDED = "ADDED";
+  String MODIFIED = "MODIFIED";
   String DELETED = "DELETED";
   String BOOKMARK = "BOOKMARK";
 
+  /**
+   * Synthetic event type: pod's container became not-ready (e.g., OOM-killed) but the pod
+   * still exists. Should be treated as a removal from discovery, but without error-level
+   * logging for nodes that aren't in the cache (repeated events are expected during
+   * CrashLoopBackOff).
+   */
+  String NOT_READY = "NOT_READY";
+
   boolean hasNext() throws SocketTimeoutException;
 
   Watch.Response<DiscoveryDruidNodeAndResourceVersion> next();
diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java
new file mode 100644
index 000000000000..0b8cb77c6a37
--- /dev/null
+++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.k8s.discovery;
+
+import io.kubernetes.client.openapi.models.V1ContainerStatus;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodStatus;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+public class DefaultK8sApiClientTest
+{
+  @Test
+  public void testIsPodReady_nullStatus()
+  {
+    V1Pod pod = new V1Pod();
+    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+  }
+
+  @Test
+  public void testIsPodReady_nullContainerStatuses()
+  {
+    V1Pod pod = new V1Pod();
+    pod.setStatus(new V1PodStatus());
+    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+  }
+
+  @Test
+  public void testIsPodReady_emptyContainerStatuses()
+  {
+    V1Pod pod = new V1Pod();
+    V1PodStatus status = new V1PodStatus();
+    status.setContainerStatuses(Collections.emptyList());
+    pod.setStatus(status);
+    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+  }
+
+  @Test
+  public void testIsPodReady_allContainersReady()
+  {
+    V1Pod pod = new V1Pod();
+    V1PodStatus status = new V1PodStatus();
+    V1ContainerStatus cs = new V1ContainerStatus();
+    cs.setReady(true);
+    status.setContainerStatuses(Collections.singletonList(cs));
+    pod.setStatus(status);
+    Assert.assertTrue(DefaultK8sApiClient.isPodReady(pod));
+  }
+
+  @Test
+  public void testIsPodReady_containerNotReady()
+  {
+    V1Pod pod = new V1Pod();
+    V1PodStatus status = new V1PodStatus();
+    V1ContainerStatus cs = new V1ContainerStatus();
+    cs.setReady(false);
+    status.setContainerStatuses(Collections.singletonList(cs));
+    pod.setStatus(status);
+    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+  }
+
+  @Test
+  public void testIsPodReady_mixedContainerReadiness()
+  {
+    V1Pod pod = new V1Pod();
+    V1PodStatus status = new V1PodStatus();
+    V1ContainerStatus cs1 = new V1ContainerStatus();
+    cs1.setReady(true);
+    V1ContainerStatus cs2 = new V1ContainerStatus();
+    cs2.setReady(false);
+    status.setContainerStatuses(Arrays.asList(cs1, cs2));
+    pod.setStatus(status);
+    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+  }
+
+  @Test
+  public void testIsPodReady_containerReadyNull()
+  {
+    V1Pod pod = new V1Pod();
+    V1PodStatus status = new V1PodStatus();
+    V1ContainerStatus cs = new V1ContainerStatus();
+    // ready is null (not set)
+    status.setContainerStatuses(Collections.singletonList(cs));
+    pod.setStatus(status);
+    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+  }
+}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
index f17b43cf2c70..5c872d9cc1a3 100644
--- a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
+++ b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
@@ -174,6 +174,17 @@ private void addNode(DiscoveryDruidNode druidNode)
   }
 
   public void childRemoved(DiscoveryDruidNode druidNode)
+  {
+    childRemoved(druidNode, true);
+  }
+
+  /**
+   * Remove a node from the discovery cache. If {@code warnIfUnknown} is true, logs at ERROR
+   * level when the node was not in the cache (appropriate for real K8s DELETED events). If false,
+   * silently skips unknown nodes (appropriate for readiness-based removals where repeated
+   * NOT_READY events are expected during CrashLoopBackOff).
+   */
+  public void childRemoved(DiscoveryDruidNode druidNode, boolean warnIfUnknown)
   {
     synchronized (lock) {
       if (!nodeRole.equals(druidNode.getNodeRole())) {
@@ -186,6 +197,15 @@ public void childRemoved(DiscoveryDruidNode druidNode)
         return;
       }
 
+      if (!warnIfUnknown && !nodes.containsKey(druidNode.getDruidNode().getHostAndPortToUse())) {
+        LOGGER.debug(
+            "Ignoring removal of node [%s] of role [%s] that is not in the cache (pod may still be not ready).",
+            druidNode.getDruidNode().getUriToUse(),
+            nodeRole.getJsonName()
+        );
+        return;
+      }
+
       LOGGER.warn("Node [%s] of role [%s] went offline.", druidNode.getDruidNode().getUriToUse(), nodeRole.getJsonName());
 
       removeNode(druidNode);

From e53945205d049c0691cba66a2733099083e1ff6b Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 11 Mar 2026 15:02:20 -0500
Subject: [PATCH 02/13] self review iteration

---
 .../k8s/discovery/DefaultK8sApiClient.java    |  6 +-
 .../discovery/DefaultK8sApiClientTest.java    |  2 +-
 .../K8sDruidNodeDiscoveryProviderTest.java    | 65 ++++++++++++++++++
 .../druid/discovery/BaseNodeRoleWatcher.java  | 10 +--
 .../discovery/BaseNodeRoleWatcherTest.java    | 66 +++++++++++++++++++
 5 files changed, 140 insertions(+), 9 deletions(-)

diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
index c50e3ea61f12..a19e7ecf062e 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
@@ -117,8 +117,8 @@ public DiscoveryDruidNodeList listPods(
       Map<String, DiscoveryDruidNode> allNodes = new HashMap();
       for (V1Pod podDef : podList.getItems()) {
         if (!isPodReady(podDef)) {
-          LOGGER.warn(
-              "Ignoring pod [%s] for role [%s] during list: poad has discovery label but is not ready.",
+          LOGGER.info(
+              "Ignoring pod [%s] for role [%s] during list: pod has discovery label but is not yet reporting as ready.",
               podDef.getMetadata().getName(),
               nodeRole
           );
@@ -207,7 +207,7 @@ public boolean hasNext() throws SocketTimeoutException
                   if (!isPodReady(item.object)) {
                     if (WatchResult.MODIFIED.equals(item.type)) {
                       // Pod was previously ready but is now unready (e.g., OOM-killed
-                      // container). Convert to DELETED to remove from discovery cache.
+                      // container). Convert to NOT_READY to ensure the host is removed from discovery cache if it exists there.
                       LOGGER.warn(
                           "Pod [%s] for role [%s] notified that it was modified and is now showing as not ready, "
                           + "treating as removed for discovery purposes.",
diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java
index 0b8cb77c6a37..df98b447d1a6 100644
--- a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java
+++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java
@@ -104,4 +104,4 @@ public void testIsPodReady_containerReadyNull()
     pod.setStatus(status);
     Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
   }
-}
\ No newline at end of file
+}
diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
index 12a3a6118da8..fa734433f419 100644
--- a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
+++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
@@ -226,6 +226,71 @@ public void testNodeRoleWatcherHandlesNullFromAPIByRestarting() throws Exception
     discoveryProvider.stop();
   }
 
+  @Test(timeout = 60_000)
+  public void testNotReadyEventRemovesNodeAndReAddOnReady() throws Exception
+  {
+    String labelSelector = "druidDiscoveryAnnouncement-cluster-identifier=druid-cluster,druidDiscoveryAnnouncement-router=true";
+    K8sApiClient mockK8sApiClient = EasyMock.createMock(K8sApiClient.class);
+
+    // Initial list returns two healthy nodes
+    EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)).andReturn(
+        new DiscoveryDruidNodeList(
+            "v1",
+            ImmutableMap.of(
+                testNode1.getDruidNode().getHostAndPortToUse(), testNode1,
+                testNode2.getDruidNode().getHostAndPortToUse(), testNode2
+            )
+        )
+    );
+
+    // Watch returns: testNode1 becomes NOT_READY (OOM), then comes back as ADDED (recovered)
+    EasyMock.expect(mockK8sApiClient.watchPods(
+        podInfo.getPodNamespace(), labelSelector, "v1", NodeRole.ROUTER)).andReturn(
+        new MockWatchResult(
+            ImmutableList.of(
+                new Watch.Response<>(WatchResult.NOT_READY, new DiscoveryDruidNodeAndResourceVersion("v2", testNode1)),
+                // Repeated NOT_READY during CrashLoopBackOff — should be silently ignored
+                new Watch.Response<>(WatchResult.NOT_READY, new DiscoveryDruidNodeAndResourceVersion("v3", testNode1)),
+                // Pod recovers and becomes ready again
+                new Watch.Response<>(WatchResult.ADDED, new DiscoveryDruidNodeAndResourceVersion("v4", testNode1))
+            ),
+            false,
+            false
+        )
+    );
+    EasyMock.replay(mockK8sApiClient);
+
+    K8sDruidNodeDiscoveryProvider discoveryProvider = new K8sDruidNodeDiscoveryProvider(
+        podInfo,
+        discoveryConfig,
+        mockK8sApiClient,
+        1
+    );
+    discoveryProvider.start();
+
+    K8sDruidNodeDiscoveryProvider.NodeRoleWatcher nodeDiscovery = discoveryProvider.getForNodeRole(NodeRole.ROUTER, false);
+
+    MockListener testListener = new MockListener(
+        ImmutableList.of(
+            MockListener.Event.added(testNode1),
+            MockListener.Event.added(testNode2),
+            MockListener.Event.inited(),
+            // testNode1 goes NOT_READY — removed
+            MockListener.Event.deleted(testNode1),
+            // Second NOT_READY is silently skipped (not in cache)
+            // testNode1 recovers — re-added
+            MockListener.Event.added(testNode1)
+        )
+    );
+    nodeDiscovery.registerListener(testListener);
+
+    nodeDiscovery.start();
+
+    testListener.assertSuccess();
+
+    discoveryProvider.stop();
+  }
+
   @Test(timeout = 10_000)
   public void testNodeRoleWatcherLoopOnNullItems() throws Exception
   {
diff --git a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
index 5c872d9cc1a3..70d6d92e10d6 100644
--- a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
+++ b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
@@ -179,10 +179,10 @@ public void childRemoved(DiscoveryDruidNode druidNode)
   }
 
   /**
-   * Remove a node from the discovery cache. If {@code warnIfUnknown} is true, logs at ERROR
-   * level when the node was not in the cache (appropriate for real K8s DELETED events). If false,
-   * silently skips unknown nodes (appropriate for readiness-based removals where repeated
-   * NOT_READY events are expected during CrashLoopBackOff).
+   * Remove a node from the discovery cache.
+   * <p>
+   * If {@code warnIfUnknown} is false and the node is not known to be present in the cache,
+   * the removal will be ignored and a debug message will be logged.
    */
   public void childRemoved(DiscoveryDruidNode druidNode, boolean warnIfUnknown)
   {
@@ -199,7 +199,7 @@ public void childRemoved(DiscoveryDruidNode druidNode, boolean warnIfUnknown)
 
       if (!warnIfUnknown && !nodes.containsKey(druidNode.getDruidNode().getHostAndPortToUse())) {
         LOGGER.debug(
-            "Ignoring removal of node [%s] of role [%s] that is not in the cache (pod may still be not ready).",
+            "Ignoring removal of node [%s] of role [%s] because it is not known to be present in the cache.",
             druidNode.getDruidNode().getUriToUse(),
             nodeRole.getJsonName()
         );
diff --git a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java
index 0decfcdef386..30d2e7843c02 100644
--- a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java
+++ b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java
@@ -244,6 +244,72 @@ public void testDuplicateChildAddedAfterResetNodesDoesNotNotifyListeners()
     Assert.assertEquals(ImmutableSet.of(broker1, broker2), new HashSet<>(nodeRoleWatcher.getAllNodes()));
   }
 
+  @Test(timeout = 60_000L)
+  public void testChildRemovedIfPresentRemovesKnownNode()
+  {
+    BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER);
+
+    DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1");
+
+    nodeRoleWatcher.childAdded(broker1);
+    nodeRoleWatcher.cacheInitialized();
+
+    TestListener listener = new TestListener();
+    nodeRoleWatcher.registerListener(listener);
+
+    Assert.assertEquals(ImmutableList.of(broker1), listener.nodesAddedList);
+
+    // Remove with warnIfUnknown=false — node IS in cache, should remove and notify
+    nodeRoleWatcher.childRemoved(broker1, false);
+
+    Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList);
+    Assert.assertTrue(nodeRoleWatcher.getAllNodes().isEmpty());
+  }
+
+  @Test(timeout = 60_000L)
+  public void testChildRemovedIfPresentSkipsUnknownNode()
+  {
+    BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER);
+
+    DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1");
+    DiscoveryDruidNode broker2 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker2");
+
+    nodeRoleWatcher.childAdded(broker1);
+    nodeRoleWatcher.cacheInitialized();
+
+    TestListener listener = new TestListener();
+    nodeRoleWatcher.registerListener(listener);
+
+    // Remove broker2 with warnIfUnknown=false — node is NOT in cache, should silently skip
+    nodeRoleWatcher.childRemoved(broker2, false);
+
+    Assert.assertTrue(listener.nodesRemovedList.isEmpty());
+    Assert.assertEquals(1, nodeRoleWatcher.getAllNodes().size());
+  }
+
+  @Test(timeout = 60_000L)
+  public void testChildRemovedIfPresentRepeatedRemovalsAreIdempotent()
+  {
+    BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER);
+
+    DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1");
+
+    nodeRoleWatcher.childAdded(broker1);
+    nodeRoleWatcher.cacheInitialized();
+
+    TestListener listener = new TestListener();
+    nodeRoleWatcher.registerListener(listener);
+
+    // First removal should remove and notify
+    nodeRoleWatcher.childRemoved(broker1, false);
+    Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList);
+
+    // Second removal should silently skip (node already removed)
+    nodeRoleWatcher.childRemoved(broker1, false);
+    // Still only one removal notification
+    Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList);
+  }
+
   private DiscoveryDruidNode buildDiscoveryDruidNode(NodeRole role, String host)
   {
     return new DiscoveryDruidNode(

From cb7127938441840919a98002b21a72ec93b1bd91 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 11 Mar 2026 16:01:38 -0500
Subject: [PATCH 03/13] minor fixes

---
 .../org/apache/druid/k8s/discovery/DefaultK8sApiClient.java   | 2 +-
 .../java/org/apache/druid/discovery/BaseNodeRoleWatcher.java  | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
index a19e7ecf062e..288fd035c183 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
@@ -208,7 +208,7 @@ public boolean hasNext() throws SocketTimeoutException
                     if (WatchResult.MODIFIED.equals(item.type)) {
                       // Pod was previously ready but is now unready (e.g., OOM-killed
                       // container). Convert to NOT_READY to ensure the host is removed from discovery cache if it exists there.
-                      LOGGER.warn(
+                      LOGGER.info(
                           "Pod [%s] for role [%s] notified that it was modified and is now showing as not ready, "
                           + "treating as removed for discovery purposes.",
                           item.object.getMetadata().getName(),
diff --git a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
index 70d6d92e10d6..cbeaf4e04433 100644
--- a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
+++ b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
@@ -181,8 +181,8 @@ public void childRemoved(DiscoveryDruidNode druidNode)
   /**
    * Remove a node from the discovery cache.
    * <p>
-   * If {@code warnIfUnknown} is false and the node is not known to be present in the cache,
-   * the removal will be ignored and a debug message will be logged.
+   * If {@code warnIfUnknown} is true, the removal is attempted unconditionally. If false,
+   * the removal is skipped if the node is not already present in the cache.
    */
   public void childRemoved(DiscoveryDruidNode druidNode, boolean warnIfUnknown)
   {

From fcce511e6180853e3e59c6c2f768c0a33913090b Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 11 Mar 2026 16:14:53 -0500
Subject: [PATCH 04/13] clarify why remapping modified to added is safe

---
 .../org/apache/druid/k8s/discovery/DefaultK8sApiClient.java   | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
index 288fd035c183..50534e5b636e 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
@@ -228,7 +228,9 @@ public boolean hasNext() throws SocketTimeoutException
                   } else if (WatchResult.MODIFIED.equals(item.type)) {
                     // Pod is ready on a MODIFIED event — it may have just become ready
                     // (e.g., container restarted after OOM). Treat as ADDED so it gets
-                    // (re-)added to the discovery cache.
+                    // (re-)added to the discovery cache. This is safe even if the node
+                    // is already in the cache — BaseNodeRoleWatcher.childAdded() uses
+                    // putIfAbsent, so duplicates are silently ignored.
                     effectiveType = WatchResult.ADDED;
                   }
 

From 474c33f6aaa842759ddd61121be6fc2d2bbc76e5 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 11 Mar 2026 16:38:35 -0500
Subject: [PATCH 05/13] Add more info about how readiness probes come into play
 for k8s discovery

---
 .../development/extensions-core/kubernetes.md | 22 ++++++++++++++++++-
 1 file changed, 21 insertions(+), 1 deletion(-)

diff --git a/docs/development/extensions-core/kubernetes.md b/docs/development/extensions-core/kubernetes.md
index 25696546dfeb..d1b03af378d9 100644
--- a/docs/development/extensions-core/kubernetes.md
+++ b/docs/development/extensions-core/kubernetes.md
@@ -54,10 +54,30 @@ Additionally, this extension has following configuration.
 
 ### Gotchas
 
-- Label/Annotation path in each pod spec MUST EXIST, which is easily satisfied if there is at least one label/annotation in the pod spec already. 
+- Label/Annotation path in each pod spec MUST EXIST, which is easily satisfied if there is at least one label/annotation in the pod spec already.
 - All Druid Pods belonging to one Druid cluster must be inside same kubernetes namespace.
 - All Druid Pods need permissions to be able to add labels to self-pod, List and Watch other Pods, create and read ConfigMap for leader election. Assuming, "default" service account is used by Druid pods, you might need to add following or something similar Kubernetes Role and Role Binding.
 
+### Readiness probes and discovery
+
+This extension uses Kubernetes container readiness to decide whether a pod is available for service discovery. When a container becomes not-ready (for example, after an OOM kill or crash), the pod is automatically removed from discovery until it becomes ready again. This prevents routing requests to pods that are unable to serve them.
+
+Because of this, you should configure [readiness probes](https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/) on all Druid pods. Without a readiness probe, Kubernetes marks a container as ready the moment the process starts, which may be before the Druid service is fully initialized and able to handle requests.
+
+**Important:** Readiness probe configuration directly affects discovery behavior. If a probe is too aggressive (low timeout, low failure threshold), a pod under heavy load could temporarily fail its probe, be removed from discovery, and shift its load onto other pods — potentially causing a cascade. To avoid this, tune your probes to tolerate brief periods of high load:
+
+```yaml
+readinessProbe:
+  httpGet:
+    path: /status/health
+    port: 8088
+  periodSeconds: 10
+  failureThreshold: 3
+  timeoutSeconds: 10
+```
+
+With this configuration, a pod must fail its readiness check 3 times in a row (30 seconds) before it is removed from discovery. Adjust these values based on your workload and tolerance for routing to temporarily unhealthy pods.
+
 ```
 apiVersion: rbac.authorization.k8s.io/v1
 kind: Role

From 23af3818396abb87122dcec5abae9b055b91d3f8 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 11 Mar 2026 16:42:49 -0500
Subject: [PATCH 06/13] more comment and log cleanup

---
 .../k8s/discovery/DefaultK8sApiClient.java    | 22 +++++++++----------
 1 file changed, 10 insertions(+), 12 deletions(-)

diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
index 50534e5b636e..fe902a4f99f7 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
@@ -118,7 +118,7 @@ public DiscoveryDruidNodeList listPods(
       for (V1Pod podDef : podList.getItems()) {
         if (!isPodReady(podDef)) {
           LOGGER.info(
-              "Ignoring pod [%s] for role [%s] during list: pod has discovery label but is not yet reporting as ready.",
+              "Ignoring pod[%s] for role[%s] during list: pod has discovery label but is not yet reporting as ready.",
               podDef.getMetadata().getName(),
               nodeRole
           );
@@ -206,30 +206,28 @@ public boolean hasNext() throws SocketTimeoutException
                 if (item.object != null) {
                   if (!isPodReady(item.object)) {
                     if (WatchResult.MODIFIED.equals(item.type)) {
-                      // Pod was previously ready but is now unready (e.g., OOM-killed
-                      // container). Convert to NOT_READY to ensure the host is removed from discovery cache if it exists there.
+                      // Pod was previously ready but is now unready (e.g., OOM-killed container).
+                      // Remap to NOT_READY to ensure the host is removed from discovery cache if is cached
                       LOGGER.info(
-                          "Pod [%s] for role [%s] notified that it was modified and is now showing as not ready, "
+                          "Pod[%s] for role[%s] notified that it was modified and is now showing as not ready, "
                           + "treating as removed for discovery purposes.",
                           item.object.getMetadata().getName(),
                           nodeRole
                       );
                       effectiveType = WatchResult.NOT_READY;
                     } else if (WatchResult.ADDED.equals(item.type)) {
-                      // Pod is not ready yet (e.g., still starting up). Skip this event
-                      // entirely — it will appear via a MODIFIED event once it becomes ready.
+                      // Pod is not ready yet (e.g., still starting up). Skip this event entirely.
+                      // It will appear via a MODIFIED event that remaps to ADDED for dicovery, once it becomes ready.
                       LOGGER.debug(
-                          "Pod [%s] for role [%s] is not ready on ADDED event, skipping until it becomes ready.",
+                          "Pod[%s] for role[%s] is not ready on ADDED event, skipping until it becomes ready.",
                           item.object.getMetadata().getName(),
                           nodeRole
                       );
                       continue;
                     }
                   } else if (WatchResult.MODIFIED.equals(item.type)) {
-                    // Pod is ready on a MODIFIED event — it may have just become ready
-                    // (e.g., container restarted after OOM). Treat as ADDED so it gets
-                    // (re-)added to the discovery cache. This is safe even if the node
-                    // is already in the cache — BaseNodeRoleWatcher.childAdded() uses
+                    // Remap MODIFIED (pod ready) events to ADDED for discovery cache purposes.
+                    // This is safe even if the node is already in the cache because BaseNodeRoleWatcher.childAdded() uses
                     // putIfAbsent, so duplicates are silently ignored.
                     effectiveType = WatchResult.ADDED;
                   }
@@ -243,7 +241,7 @@ public boolean hasNext() throws SocketTimeoutException
                   catch (Exception ex) {
                     LOGGER.warn(
                         ex,
-                        "Failed to deserialize node info from pod [%s] for role [%s] on [%s] event. "
+                        "Failed to deserialize node info from pod[%s] for role[%s] on [%s] event. "
                         + "Passing null to trigger watch restart and full resync.",
                         item.object.getMetadata() != null ? item.object.getMetadata().getName() : "unknown",
                         nodeRole,

From 1340130771b11efc1f8d09fac65135a74e8f7666 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 11 Mar 2026 16:51:28 -0500
Subject: [PATCH 07/13] Fix doc spelling

---
 docs/development/extensions-core/kubernetes.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/development/extensions-core/kubernetes.md b/docs/development/extensions-core/kubernetes.md
index d1b03af378d9..48ec2823cbc2 100644
--- a/docs/development/extensions-core/kubernetes.md
+++ b/docs/development/extensions-core/kubernetes.md
@@ -60,7 +60,7 @@ Additionally, this extension has following configuration.
 
 ### Readiness probes and discovery
 
-This extension uses Kubernetes container readiness to decide whether a pod is available for service discovery. When a container becomes not-ready (for example, after an OOM kill or crash), the pod is automatically removed from discovery until it becomes ready again. This prevents routing requests to pods that are unable to serve them.
+This extension uses Kubernetes container readiness to decide whether a pod is available for service discovery. When a container becomes not-ready (for example, after an out of memory kill or crash), the pod is automatically removed from discovery until it becomes ready again. This prevents routing requests to pods that are unable to serve them.
 
 Because of this, you should configure [readiness probes](https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/) on all Druid pods. Without a readiness probe, Kubernetes marks a container as ready the moment the process starts, which may be before the Druid service is fully initialized and able to handle requests.
 

From 23f557ebd24e4202425e597fb96c6b7d50448340 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 11 Mar 2026 17:14:02 -0500
Subject: [PATCH 08/13] refactor to make reasoning about new behavior more easy

---
 .../k8s/discovery/K8sDruidNodeDiscoveryProvider.java |  2 +-
 .../apache/druid/discovery/BaseNodeRoleWatcher.java  | 10 +++++-----
 .../druid/discovery/BaseNodeRoleWatcherTest.java     | 12 ++++++------
 3 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
index e7c44e2d609b..397c816d40d6 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
@@ -276,7 +276,7 @@ private void keepWatching(String labelSelector, String resourceVersion)
                   case WatchResult.NOT_READY:
                     // Pod container became not-ready (e.g., OOM-killed). Remove if present
                     // in cache, but don't log errors for repeated events during CrashLoopBackOff.
-                    baseNodeRoleWatcher.childRemoved(item.object.getNode(), false);
+                    baseNodeRoleWatcher.childRemoved(item.object.getNode(), true);
                     break;
                   default:
                 }
diff --git a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
index cbeaf4e04433..6c38a5dc18a2 100644
--- a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
+++ b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java
@@ -175,16 +175,16 @@ private void addNode(DiscoveryDruidNode druidNode)
 
   public void childRemoved(DiscoveryDruidNode druidNode)
   {
-    childRemoved(druidNode, true);
+    childRemoved(druidNode, false);
   }
 
   /**
    * Remove a node from the discovery cache.
    * <p>
-   * If {@code warnIfUnknown} is true, the removal is attempted unconditionally. If false,
-   * the removal is skipped if the node is not already present in the cache.
+   * If {@code skipIfUnknown} is true, the removal is skipped if the node is not already
+   * present in the cache. If false, the removal is attempted unconditionally.
    */
-  public void childRemoved(DiscoveryDruidNode druidNode, boolean warnIfUnknown)
+  public void childRemoved(DiscoveryDruidNode druidNode, boolean skipIfUnknown)
   {
     synchronized (lock) {
       if (!nodeRole.equals(druidNode.getNodeRole())) {
@@ -197,7 +197,7 @@ public void childRemoved(DiscoveryDruidNode druidNode, boolean warnIfUnknown)
         return;
       }
 
-      if (!warnIfUnknown && !nodes.containsKey(druidNode.getDruidNode().getHostAndPortToUse())) {
+      if (skipIfUnknown && !nodes.containsKey(druidNode.getDruidNode().getHostAndPortToUse())) {
         LOGGER.debug(
             "Ignoring removal of node [%s] of role [%s] because it is not known to be present in the cache.",
             druidNode.getDruidNode().getUriToUse(),
diff --git a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java
index 30d2e7843c02..ec7b24cbdc0f 100644
--- a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java
+++ b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java
@@ -259,8 +259,8 @@ public void testChildRemovedIfPresentRemovesKnownNode()
 
     Assert.assertEquals(ImmutableList.of(broker1), listener.nodesAddedList);
 
-    // Remove with warnIfUnknown=false — node IS in cache, should remove and notify
-    nodeRoleWatcher.childRemoved(broker1, false);
+    // Remove with skipIfUnknown=true — node IS in cache, should remove and notify
+    nodeRoleWatcher.childRemoved(broker1, true);
 
     Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList);
     Assert.assertTrue(nodeRoleWatcher.getAllNodes().isEmpty());
@@ -280,8 +280,8 @@ public void testChildRemovedIfPresentSkipsUnknownNode()
     TestListener listener = new TestListener();
     nodeRoleWatcher.registerListener(listener);
 
-    // Remove broker2 with warnIfUnknown=false — node is NOT in cache, should silently skip
-    nodeRoleWatcher.childRemoved(broker2, false);
+    // Remove broker2 with skipIfUnknown=true — node is NOT in cache, should silently skip
+    nodeRoleWatcher.childRemoved(broker2, true);
 
     Assert.assertTrue(listener.nodesRemovedList.isEmpty());
     Assert.assertEquals(1, nodeRoleWatcher.getAllNodes().size());
@@ -301,11 +301,11 @@ public void testChildRemovedIfPresentRepeatedRemovalsAreIdempotent()
     nodeRoleWatcher.registerListener(listener);
 
     // First removal should remove and notify
-    nodeRoleWatcher.childRemoved(broker1, false);
+    nodeRoleWatcher.childRemoved(broker1, true);
     Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList);
 
     // Second removal should silently skip (node already removed)
-    nodeRoleWatcher.childRemoved(broker1, false);
+    nodeRoleWatcher.childRemoved(broker1, true);
     // Still only one removal notification
     Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList);
   }

From 106fe387f1fe3fa2c20c5ffa6eac8b0ef835a376 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Mon, 16 Mar 2026 17:03:16 -0500
Subject: [PATCH 09/13] change example in docs to new readiness probe

---
 docs/development/extensions-core/kubernetes.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/development/extensions-core/kubernetes.md b/docs/development/extensions-core/kubernetes.md
index 48ec2823cbc2..df3f744692be 100644
--- a/docs/development/extensions-core/kubernetes.md
+++ b/docs/development/extensions-core/kubernetes.md
@@ -69,7 +69,7 @@ Because of this, you should configure [readiness probes](https://kubernetes.io/d
 ```yaml
 readinessProbe:
   httpGet:
-    path: /status/health
+    path: /status/ready
     port: 8088
   periodSeconds: 10
   failureThreshold: 3

From 5ca2069ecfa6547fdae5dd6c8468a9664dd9cb44 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Mon, 16 Mar 2026 17:07:00 -0500
Subject: [PATCH 10/13] Fix tests file

---
 .../k8s/discovery/DefaultK8sApiClientTest.java | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java
index df98b447d1a6..74b95d3cfef3 100644
--- a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java
+++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientTest.java
@@ -22,8 +22,8 @@
 import io.kubernetes.client.openapi.models.V1ContainerStatus;
 import io.kubernetes.client.openapi.models.V1Pod;
 import io.kubernetes.client.openapi.models.V1PodStatus;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -34,7 +34,7 @@ public class DefaultK8sApiClientTest
   public void testIsPodReady_nullStatus()
   {
     V1Pod pod = new V1Pod();
-    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+    Assertions.assertFalse(DefaultK8sApiClient.isPodReady(pod));
   }
 
   @Test
@@ -42,7 +42,7 @@ public void testIsPodReady_nullContainerStatuses()
   {
     V1Pod pod = new V1Pod();
     pod.setStatus(new V1PodStatus());
-    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+    Assertions.assertFalse(DefaultK8sApiClient.isPodReady(pod));
   }
 
   @Test
@@ -52,7 +52,7 @@ public void testIsPodReady_emptyContainerStatuses()
     V1PodStatus status = new V1PodStatus();
     status.setContainerStatuses(Collections.emptyList());
     pod.setStatus(status);
-    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+    Assertions.assertFalse(DefaultK8sApiClient.isPodReady(pod));
   }
 
   @Test
@@ -64,7 +64,7 @@ public void testIsPodReady_allContainersReady()
     cs.setReady(true);
     status.setContainerStatuses(Collections.singletonList(cs));
     pod.setStatus(status);
-    Assert.assertTrue(DefaultK8sApiClient.isPodReady(pod));
+    Assertions.assertTrue(DefaultK8sApiClient.isPodReady(pod));
   }
 
   @Test
@@ -76,7 +76,7 @@ public void testIsPodReady_containerNotReady()
     cs.setReady(false);
     status.setContainerStatuses(Collections.singletonList(cs));
     pod.setStatus(status);
-    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+    Assertions.assertFalse(DefaultK8sApiClient.isPodReady(pod));
   }
 
   @Test
@@ -90,7 +90,7 @@ public void testIsPodReady_mixedContainerReadiness()
     cs2.setReady(false);
     status.setContainerStatuses(Arrays.asList(cs1, cs2));
     pod.setStatus(status);
-    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+    Assertions.assertFalse(DefaultK8sApiClient.isPodReady(pod));
   }
 
   @Test
@@ -102,6 +102,6 @@ public void testIsPodReady_containerReadyNull()
     // ready is null (not set)
     status.setContainerStatuses(Collections.singletonList(cs));
     pod.setStatus(status);
-    Assert.assertFalse(DefaultK8sApiClient.isPodReady(pod));
+    Assertions.assertFalse(DefaultK8sApiClient.isPodReady(pod));
   }
 }

From 0a9b72f572943bc3d45ffb19ecab2e30dd27e46d Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Tue, 17 Mar 2026 17:23:44 -0500
Subject: [PATCH 11/13] Some doc sharpening

---
 .../development/extensions-core/kubernetes.md | 42 +++++++++++++------
 1 file changed, 30 insertions(+), 12 deletions(-)

diff --git a/docs/development/extensions-core/kubernetes.md b/docs/development/extensions-core/kubernetes.md
index df3f744692be..3115ee9056b2 100644
--- a/docs/development/extensions-core/kubernetes.md
+++ b/docs/development/extensions-core/kubernetes.md
@@ -38,9 +38,11 @@ This extension works together with HTTP-based segment and task management in Dru
 `druid.indexer.runner.type=httpRemote`
 `druid.discovery.type=k8s`
 
-For Node Discovery, Each Druid process running inside a pod "announces" itself by adding few "labels" and "annotations" in the pod spec. Druid process needs to be aware of pod name and namespace which it reads from environment variables `POD_NAME` and `POD_NAMESPACE`. These variable names can be changed, see configuration below. But in the end, each pod needs to have self pod name and namespace added as environment variables.
+For node discovery, each Druid process running inside a pod "announces" itself by adding labels and annotations to the pod spec. A pod is discoverable by other Druid processes when it has the required labels and annotations and Kubernetes considers the container ready. Without a readiness probe, Kubernetes marks a container as ready the moment the process starts — see [Readiness Probes](#readiness-probes) for why you should configure one.
 
-Additionally, this extension has following configuration.
+Each Druid process needs to be aware of its own pod name and namespace, which it reads from environment variables `POD_NAME` and `POD_NAMESPACE`. These variable names can be changed (see configuration below), but every pod must have its own name and namespace available as environment variables.
+
+Additionally, this extension has the following configuration.
 
 ### Properties
 |Property|Possible Values|Description|Default|required|
@@ -52,31 +54,47 @@ Additionally, this extension has following configuration.
 |`druid.discovery.k8s.renewDeadline`|`Duration`|Lease renewal period used by Leader.|PT17S|No|
 |`druid.discovery.k8s.retryPeriod`|`Duration`|Retry wait used by Leader Election algorithm on failed operations.|PT5S|No|
 
-### Gotchas
+### Readiness Probes
+
+:::info
+Readiness probe configuration directly affects discovery behavior. If a probe is too aggressive (low timeout, low failure threshold), a pod under heavy load could temporarily fail its probe, be removed from discovery, and shift its load onto other pods — potentially causing a cascade. To avoid this, tune your probes to tolerate brief periods of high load.
+:::
+
+This extension uses Kubernetes container readiness, in addition to labels and annotations, to decide whether a pod is available for service discovery.
+
+You should configure [readiness probes](https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/) on all Druid pods. Without a readiness probe, Kubernetes marks a container as ready the moment the process starts, which may be before the Druid service is fully initialized and able to handle requests.
 
-- Label/Annotation path in each pod spec MUST EXIST, which is easily satisfied if there is at least one label/annotation in the pod spec already.
-- All Druid Pods belonging to one Druid cluster must be inside same kubernetes namespace.
-- All Druid Pods need permissions to be able to add labels to self-pod, List and Watch other Pods, create and read ConfigMap for leader election. Assuming, "default" service account is used by Druid pods, you might need to add following or something similar Kubernetes Role and Role Binding.
+A container may become unready if:
+* The process is killed or crashes. It will be immediately marked as unready without waiting for the readiness probe to cross its failure threshold.
+* The process is alive but not healthy. It will be marked as unready after it fails its readiness probe a configured number of times (`failureThreshold`).
 
-### Readiness probes and discovery
+Once marked as unready, a container must pass its readiness probe `successThreshold` times (default: 1) before it is considered ready again and re-enters discovery.
 
-This extension uses Kubernetes container readiness to decide whether a pod is available for service discovery. When a container becomes not-ready (for example, after an out of memory kill or crash), the pod is automatically removed from discovery until it becomes ready again. This prevents routing requests to pods that are unable to serve them.
+#### Recommendations
 
-Because of this, you should configure [readiness probes](https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/) on all Druid pods. Without a readiness probe, Kubernetes marks a container as ready the moment the process starts, which may be before the Druid service is fully initialized and able to handle requests.
+The `/status/ready` endpoint is a good candidate for readiness checks, as it indicates whether the Druid node is ready to serve requests. Services use this endpoint to decide if they should announce themselves, so it is a natural choice for the readiness probe. However, you can choose a different endpoint if it better suits your needs.
 
-**Important:** Readiness probe configuration directly affects discovery behavior. If a probe is too aggressive (low timeout, low failure threshold), a pod under heavy load could temporarily fail its probe, be removed from discovery, and shift its load onto other pods — potentially causing a cascade. To avoid this, tune your probes to tolerate brief periods of high load:
+For Druid processes that have long startup times, consider using a [startup probe](https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/#define-startup-probes) so that the readiness probe does not run (and fail) during initialization.
+
+Baseline readiness probe configuration for Druid might look like this. Replace the port value with the port your Druid process listens on (e.g., `8888` for the Router, `8081` for the Broker):
 
 ```yaml
 readinessProbe:
   httpGet:
     path: /status/ready
-    port: 8088
+    port: 8888
   periodSeconds: 10
   failureThreshold: 3
   timeoutSeconds: 10
 ```
 
-With this configuration, a pod must fail its readiness check 3 times in a row (30 seconds) before it is removed from discovery. Adjust these values based on your workload and tolerance for routing to temporarily unhealthy pods.
+With this configuration, a pod must fail its readiness check 3 times in a row (30 seconds) before it is marked as not ready. Adjust these values based on your workload and tolerance for routing to temporarily unhealthy pods.
+
+### Gotchas
+
+- The label/annotation path in each pod spec must exist, which is easily satisfied if there is at least one label or annotation in the pod spec already.
+- All Druid pods belonging to one Druid cluster must be inside the same Kubernetes namespace.
+- All Druid pods need permissions to add labels to their own pod, list and watch other pods, and create and read ConfigMaps for leader election. Assuming the `default` service account is used by Druid pods, you might need to add the following (or similar) Kubernetes Role and RoleBinding.
 
 ```
 apiVersion: rbac.authorization.k8s.io/v1

From 4e1db73710b19b50fa38829f5209be9528e96679 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Tue, 17 Mar 2026 17:30:47 -0500
Subject: [PATCH 12/13] add some common k8s language to spelling

---
 website/.spelling | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/website/.spelling b/website/.spelling
index 706c3845d753..f0c7e5cf6091 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -219,6 +219,7 @@ pull-deps
 RDBMS
 RDDs
 RDS
+RoleBinding
 ROUTINE_CATALOG
 ROUTINE_NAME
 ROUTINE_SCHEMA
@@ -1131,6 +1132,7 @@ Env
 POD_NAME
 POD_NAMESPACE
 ConfigMap
+ConfigMaps
 PT17S
 GCS
 gcs-connector

From 818bffcbdb6b3e5e038d1ccde9b9e24a981d0f09 Mon Sep 17 00:00:00 2001
From: Lucas Capistrant <capistrant@users.noreply.github.com>
Date: Wed, 18 Mar 2026 15:05:14 -0500
Subject: [PATCH 13/13] Address review comments on log noiseness and javadocs

---
 .../k8s/discovery/DefaultK8sApiClient.java    |  2 +-
 .../K8sDruidNodeDiscoveryProvider.java        |  8 +--
 .../druid/k8s/discovery/WatchResult.java      |  6 ++
 .../K8sDruidNodeDiscoveryProviderTest.java    | 62 +++++++++++++++++++
 4 files changed, 73 insertions(+), 5 deletions(-)

diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
index fe902a4f99f7..b7992a3fb2ad 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java
@@ -217,7 +217,7 @@ public boolean hasNext() throws SocketTimeoutException
                       effectiveType = WatchResult.NOT_READY;
                     } else if (WatchResult.ADDED.equals(item.type)) {
                       // Pod is not ready yet (e.g., still starting up). Skip this event entirely.
-                      // It will appear via a MODIFIED event that remaps to ADDED for dicovery, once it becomes ready.
+                      // It will appear via a MODIFIED event that remaps to ADDED for discovery, once it becomes ready.
                       LOGGER.debug(
                           "Pod[%s] for role[%s] is not ready on ADDED event, skipping until it becomes ready.",
                           item.object.getMetadata().getName(),
diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
index 397c816d40d6..96497d4510f6 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
@@ -271,11 +271,11 @@ private void keepWatching(String labelSelector, String resourceVersion)
                     baseNodeRoleWatcher.childAdded(item.object.getNode());
                     break;
                   case WatchResult.DELETED:
-                    baseNodeRoleWatcher.childRemoved(item.object.getNode());
-                    break;
                   case WatchResult.NOT_READY:
-                    // Pod container became not-ready (e.g., OOM-killed). Remove if present
-                    // in cache, but don't log errors for repeated events during CrashLoopBackOff.
+                    // Use skipIfUnknown=true for all k8s discovery removals.
+                    // DELETED can fire after NOT_READY (so the service is already removed), or before ADDED (pod deleted before becoming ready).
+                    // NOT_READY can repeat during CrashLoopBackOff. None of these warrant the error-level logging that
+                    // comes with trying to remove an unknown service.
                     baseNodeRoleWatcher.childRemoved(item.object.getNode(), true);
                     break;
                   default:
diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java
index c5457cdde065..2b3ef7b584a9 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java
@@ -23,6 +23,12 @@
 
 import java.net.SocketTimeoutException;
 
+/**
+ * Iterator over k8s pod watch events that is aligned with the needs of Druid service discovery rather than
+ * raw Kubernetes watch semantics. Implementations may remap or synthesize event types: for example,
+ * a Kubernetes MODIFIED event for a pod whose containers are no longer ready (according to k8s readiness state) is surfaced as the
+ * synthetic {@link #NOT_READY} type so that consumers can handle it as a removal from k8s service discovery.
+ */
 public interface WatchResult
 {
   String ADDED = "ADDED";
diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
index b31bb2df2cb6..18e030f25073 100644
--- a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
+++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
@@ -296,6 +296,68 @@ public void testNotReadyEventRemovesNodeAndReAddOnReady() throws Exception
     discoveryProvider.stop();
   }
 
+  @Test
+  @Timeout(value = 60_000, unit = TimeUnit.MILLISECONDS)
+  public void testDeletedAfterNotReadyIssilentlyIgnored() throws Exception
+  {
+    String labelSelector = "druidDiscoveryAnnouncement-cluster-identifier=druid-cluster,druidDiscoveryAnnouncement-router=true";
+    K8sApiClient mockK8sApiClient = EasyMock.createMock(K8sApiClient.class);
+
+    // Initial list returns two healthy nodes
+    EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)).andReturn(
+        new DiscoveryDruidNodeList(
+            "v1",
+            ImmutableMap.of(
+                testNode1.getDruidNode().getHostAndPortToUse(), testNode1,
+                testNode2.getDruidNode().getHostAndPortToUse(), testNode2
+            )
+        )
+    );
+
+    // Watch: testNode1 goes NOT_READY, then DELETED arrives (node already removed from cache)
+    EasyMock.expect(mockK8sApiClient.watchPods(
+        podInfo.getPodNamespace(), labelSelector, "v1", NodeRole.ROUTER)).andReturn(
+        new MockWatchResult(
+            ImmutableList.of(
+                new Watch.Response<>(WatchResult.NOT_READY, new DiscoveryDruidNodeAndResourceVersion("v2", testNode1)),
+                // DELETED after NOT_READY — node already removed, should be silently skipped
+                new Watch.Response<>(WatchResult.DELETED, new DiscoveryDruidNodeAndResourceVersion("v3", testNode1))
+            ),
+            false,
+            false
+        )
+    );
+    EasyMock.replay(mockK8sApiClient);
+
+    K8sDruidNodeDiscoveryProvider discoveryProvider = new K8sDruidNodeDiscoveryProvider(
+        podInfo,
+        discoveryConfig,
+        mockK8sApiClient,
+        1
+    );
+    discoveryProvider.start();
+
+    K8sDruidNodeDiscoveryProvider.NodeRoleWatcher nodeDiscovery = discoveryProvider.getForNodeRole(NodeRole.ROUTER, false);
+
+    MockListener testListener = new MockListener(
+        ImmutableList.of(
+            MockListener.Event.added(testNode1),
+            MockListener.Event.added(testNode2),
+            MockListener.Event.inited(),
+            // testNode1 goes NOT_READY — removed
+            MockListener.Event.deleted(testNode1)
+            // DELETED is silently skipped (not in cache) — no second deleted event
+        )
+    );
+    nodeDiscovery.registerListener(testListener);
+
+    nodeDiscovery.start();
+
+    testListener.assertSuccess();
+
+    discoveryProvider.stop();
+  }
+
   @Test
   @Timeout(value = 10_000, unit = TimeUnit.MILLISECONDS)
   public void testNodeRoleWatcherLoopOnNullItems() throws Exception