Re: [PR] fix: Restarted the watch after stream reset #19055 (druid)

["FrankChen021 (via GitHub)" <gi...@apache.org>]

FrankChen021 commented on code in PR #19056:
URL: https://github.com/apache/druid/pull/19056#discussion_r3934176578


##########
extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java:
##########
@@ -254,50 +254,48 @@ private void watch()
     private void keepWatching(String labelSelector, String resourceVersion)
     {
       String nextResourceVersion = resourceVersion;
-      while (lifecycleLock.isStarted()) {
-        try {
-          WatchResult iter =
-              k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, nextResourceVersion, nodeRole);
 
+      while (lifecycleLock.awaitStarted(1, TimeUnit.MILLISECONDS)) {
+
+        try (WatchResult iter = k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, nextResourceVersion, nodeRole)) {
           if (iter == null) {
             // history not available, we need to start from scratch
             return;
           }
 
-          try {
-            while (iter.hasNext()) {
-              Watch.Response<DiscoveryDruidNodeAndResourceVersion> item = iter.next();
-              if (item != null && item.type != null && item.object != null) {
-                switch (item.type) {
-                  case WatchResult.ADDED:
-                    baseNodeRoleWatcher.childAdded(item.object.getNode());
-                    break;
-                  case WatchResult.DELETED:
-                  case WatchResult.NOT_READY:
-                    // 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:
-                }
-
-                // This should be updated after the action has been dealt with successfully
-                nextResourceVersion = item.object.getResourceVersion();
-
-              } else {
-                // Try again by starting the watch from the beginning. This can happen if the
-                // watch goes bad.
-                LOGGER.debug("Received NULL item while watching role[%s]. Restarting watch.", this.nodeRole);
-                return;
-              }
+          while (iter.hasNext()) {
+            Watch.Response<DiscoveryDruidNodeAndResourceVersion> item = iter.next();
+
+            if (item == null || item.type == null || item.object == null) {
+              LOGGER.debug("Received NULL item while watching role[%s]. Restarting watch.", this.nodeRole);
+              return;
             }
-          }
-          finally {
-            iter.close();
+
+            switch (item.type) {
+              case WatchResult.ADDED:
+                baseNodeRoleWatcher.childAdded(item.object.getNode());
+                break;
+              case WatchResult.DELETED:
+              case WatchResult.NOT_READY:
+                // 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:
+            }
+
+            // This should be updated after the action has been dealt with successfully
+            nextResourceVersion = item.object.getResourceVersion();
           }
 
+          LOGGER.trace("Watch closed normally for role[%s]", this.nodeRole);
+          return;
+        }
+        catch (ChannelResetException ex) {
+          LOGGER.debug("Watch stream terminated normally for role[%s], restarting", this.nodeRole);
+          return;

Review Comment:
   [P2] Persistent resets can cause a tight relist loop
   
   Returning from this handler immediately re-enters the outer `watch()` loop, which performs `listPods()` and opens another watch without using `watcherErrorRetryWaitMS`. If the API server or an intermediary persistently resets every watch stream, this becomes an unbounded list/watch loop that can hammer the Kubernetes API and consume the watcher thread. Apply the configured retry backoff (or otherwise rate-limit full resyncs) before retrying this failure path.



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org