Skip stale worker heartbeats from orphaned worker directories

The supervisor builds the heartbeat batch it sends to Nimbus from every
worker directory present on disk, with no check on whether the worker is
still alive. If a worker directory is ever left behind (a worker that died
before the supervisor finished cleanup), its frozen heartbeat keeps being
reported until the next supervisor restart -- the only time orphaned
directories are reclaimed. Nimbus then repeatedly reads the (often deleted)
topology conf, flooding its log with "Exception when getting heartbeat
timeout" / NotAliveException.

Filter out heartbeats older than supervisor.worker.timeout.secs in
ReportWorkerHeartbeats: such a worker is already considered dead by the
same timeout the slot uses, so it should not be reported. A live worker
always refreshes its heartbeat well within the timeout, so this never
drops a valid heartbeat.

Author: mwkang

storm-server/src/main/java/org/apache/storm/daemon/supervisor/timer/ReportWorkerHeartbeats.java

@@ -15,6 +15,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import org.apache.storm.Config;
 import org.apache.storm.daemon.supervisor.Supervisor;
 import org.apache.storm.daemon.supervisor.SupervisorUtils;
 import org.apache.storm.generated.LSWorkerHeartbeat;
@@ -23,6 +24,8 @@
 import org.apache.storm.thrift.TException;
 import org.apache.storm.utils.ConfigUtils;
 import org.apache.storm.utils.NimbusClient;
+import org.apache.storm.utils.ObjectReader;
+import org.apache.storm.utils.Time;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,10 +39,12 @@ public class ReportWorkerHeartbeats implements Runnable {
 
     private Supervisor supervisor;
     private Map<String, Object> conf;
+    private final int workerTimeoutSecs;
 
     public ReportWorkerHeartbeats(Map<String, Object> conf, Supervisor supervisor) {
         this.conf = conf;
         this.supervisor = supervisor;
+        this.workerTimeoutSecs = ObjectReader.getInt(conf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS));
     }
 
     @Override
@@ -59,7 +64,7 @@ private SupervisorWorkerHeartbeats getAndResetWorkerHeartbeats() {
         }
     }
 
-    private SupervisorWorkerHeartbeats getSupervisorWorkerHeartbeatsFromLocal(Map<String, LSWorkerHeartbeat> localHeartbeats) {
+    SupervisorWorkerHeartbeats getSupervisorWorkerHeartbeatsFromLocal(Map<String, LSWorkerHeartbeat> localHeartbeats) {
         SupervisorWorkerHeartbeats supervisorWorkerHeartbeats = new SupervisorWorkerHeartbeats();
 
         List<SupervisorWorkerHeartbeat> heartbeatList = new ArrayList<>();
@@ -70,6 +75,18 @@ private SupervisorWorkerHeartbeats getSupervisorWorkerHeartbeatsFromLocal(Map<St
                 continue;
             }
 
+            // Skip stale heartbeats left by worker directories that were never cleaned up
+            // (e.g. a worker that died before the supervisor finished cleanup). Such a worker
+            // has not heartbeat within the timeout, so it is already considered dead; forwarding
+            // it would make Nimbus repeatedly read the (often deleted) topology conf and log noise.
+            // A live worker always refreshes its heartbeat well within the timeout.
+            int hbAgeSecs = Time.currentTimeSecs() - lsWorkerHeartbeat.get_time_secs();
+            if (hbAgeSecs > workerTimeoutSecs) {
+                LOG.debug("Skipping stale heartbeat for topology {}: age {}s > worker timeout {}s",
+                    lsWorkerHeartbeat.get_topology_id(), hbAgeSecs, workerTimeoutSecs);
+                continue;
+            }
+
             SupervisorWorkerHeartbeat supervisorWorkerHeartbeat = new SupervisorWorkerHeartbeat();
             supervisorWorkerHeartbeat.set_storm_id(lsWorkerHeartbeat.get_topology_id());
             supervisorWorkerHeartbeat.set_executors(lsWorkerHeartbeat.get_executors());

storm-server/src/test/java/org/apache/storm/daemon/supervisor/timer/ReportWorkerHeartbeatsTest.java

@@ -0,0 +1,96 @@
+/*
+ * 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.storm.daemon.supervisor.timer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.storm.Config;
+import org.apache.storm.daemon.supervisor.Supervisor;
+import org.apache.storm.generated.LSWorkerHeartbeat;
+import org.apache.storm.generated.SupervisorWorkerHeartbeat;
+import org.apache.storm.generated.SupervisorWorkerHeartbeats;
+import org.apache.storm.utils.Time;
+import org.junit.jupiter.api.Test;
+
+public class ReportWorkerHeartbeatsTest {
+
+    private static final int WORKER_TIMEOUT_SECS = 30;
+    private static final String SUPERVISOR_ID = "test-supervisor";
+
+    private static LSWorkerHeartbeat mkHeartbeat(String topologyId, int timeSecs) {
+        LSWorkerHeartbeat hb = new LSWorkerHeartbeat();
+        hb.set_topology_id(topologyId);
+        hb.set_time_secs(timeSecs);
+        return hb;
+    }
+
+    private ReportWorkerHeartbeats mkReporter() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put(Config.SUPERVISOR_WORKER_TIMEOUT_SECS, WORKER_TIMEOUT_SECS);
+        Supervisor supervisor = mock(Supervisor.class);
+        when(supervisor.getId()).thenReturn(SUPERVISOR_ID);
+        return new ReportWorkerHeartbeats(conf, supervisor);
+    }
+
+    private static Set<String> reportedTopologies(SupervisorWorkerHeartbeats heartbeats) {
+        return heartbeats.get_worker_heartbeats().stream()
+            .map(SupervisorWorkerHeartbeat::get_storm_id)
+            .collect(Collectors.toSet());
+    }
+
+    @Test
+    public void freshHeartbeatsAreReportedAndStaleOnesAreFilteredOut() {
+        try (Time.SimulatedTime ignored = new Time.SimulatedTime()) {
+            Time.advanceTimeSecs(100_000);
+            int now = Time.currentTimeSecs();
+
+            Map<String, LSWorkerHeartbeat> local = new LinkedHashMap<>();
+            // live worker: heartbeat just refreshed
+            local.put("w-fresh", mkHeartbeat("topo-fresh", now));
+            // exactly at the timeout boundary is still considered alive (age == timeout, not > timeout)
+            local.put("w-boundary", mkHeartbeat("topo-boundary", now - WORKER_TIMEOUT_SECS));
+            // just past the timeout: the worker is already considered dead
+            local.put("w-stale", mkHeartbeat("topo-stale", now - WORKER_TIMEOUT_SECS - 1));
+            // orphaned worker directory left behind long ago for a topology that no longer exists
+            local.put("w-orphan", mkHeartbeat("topo-orphan", now - 86_400));
+
+            SupervisorWorkerHeartbeats result = mkReporter().getSupervisorWorkerHeartbeatsFromLocal(local);
+
+            assertEquals(SUPERVISOR_ID, result.get_supervisor_id());
+            assertEquals(Set.of("topo-fresh", "topo-boundary"), reportedTopologies(result));
+        }
+    }
+
+    @Test
+    public void nullLocalHeartbeatsAreSkipped() {
+        try (Time.SimulatedTime ignored = new Time.SimulatedTime()) {
+            Time.advanceTimeSecs(100_000);
+            int now = Time.currentTimeSecs();
+
+            Map<String, LSWorkerHeartbeat> local = new LinkedHashMap<>();
+            local.put("w-null", null);
+            local.put("w-fresh", mkHeartbeat("topo-fresh", now));
+
+            SupervisorWorkerHeartbeats result = mkReporter().getSupervisorWorkerHeartbeatsFromLocal(local);
+
+            assertEquals(Set.of("topo-fresh"), reportedTopologies(result));
+        }
+    }
+}