diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/HeartbeatCache.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/HeartbeatCache.java index bb6449c9132..c69aa2c2500 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/HeartbeatCache.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/HeartbeatCache.java @@ -27,6 +27,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; + import org.apache.storm.generated.Assignment; import org.apache.storm.generated.ExecutorInfo; import org.apache.storm.generated.SupervisorWorkerHeartbeat; @@ -72,7 +73,16 @@ public synchronized void updateTimeout(Integer timeout) { isTimedOut = Time.deltaSecs(getNimbusTimeSecs()) >= timeout; } - public synchronized void updateFromHb(Integer timeout, Map newBeat) { + // Used for RPC heartbeats: nimbusTimeSecs is refreshed on every heartbeat so that + // idle-but-alive executors (whose stats TIME_SECS may not advance) are not falsely timed out. + public synchronized void updateFromRpcHb(Integer timeout) { + nimbusTimeSecs = Time.currentTimeSecs(); + updateTimeout(timeout); + } + + // Used for ZK heartbeats: nimbusTimeSecs is only refreshed when the executor's stats + // TIME_SECS advances, preserving zombie detection for legacy topologies. + public synchronized void updateFromZkHb(Integer timeout, Map newBeat) { if (newBeat != null) { Integer newReportedTime = (Integer) newBeat.getOrDefault(ClientStatsUtil.TIME_SECS, 0); if (!newReportedTime.equals(executorReportedTimeSecs)) { @@ -96,6 +106,7 @@ public HeartbeatCache() { /** * Add an empty topology to the cache for testing purposes. + * * @param topoId the id of the topology to add. */ @VisibleForTesting @@ -105,6 +116,7 @@ public void addEmptyTopoForTests(String topoId) { /** * Get the number of topologies with cached heartbeats. + * * @return the number of topologies with cached heartbeats. */ @VisibleForTesting @@ -114,6 +126,7 @@ public int getNumToposCached() { /** * Get the topology ids with cached heartbeats. + * * @return the set of topology ids with cached heartbeats. */ @VisibleForTesting @@ -123,6 +136,7 @@ public Set getTopologyIds() { /** * Remove a specific topology from the cache. + * * @param topoId the id of the topology to remove. */ public void removeTopo(String topoId) { @@ -131,7 +145,8 @@ public void removeTopo(String topoId) { /** * Go through all executors and time them out if needed. - * @param topoId the id of the topology to look at. + * + * @param topoId the id of the topology to look at. * @param taskTimeoutSecs the timeout to know if they are too old. */ public void timeoutOldHeartbeats(String topoId, Integer taskTimeoutSecs) { @@ -143,10 +158,11 @@ public void timeoutOldHeartbeats(String topoId, Integer taskTimeoutSecs) { /** * Update the cache with heartbeats from a worker through zookeeper. - * @param topoId the id to the topology. + * + * @param topoId the id to the topology. * @param executorBeats the HB data. - * @param allExecutors the executors. - * @param timeout the timeout. + * @param allExecutors the executors. + * @param timeout the timeout. */ public void updateFromZkHeartbeat(String topoId, Map, Map> executorBeats, Set> allExecutors, Integer timeout) { @@ -158,12 +174,13 @@ public void updateFromZkHeartbeat(String topoId, Map, Map executor : allExecutors) { final Map newBeat = executorBeats.get(executor); ExecutorCache currBeat = topoCache.computeIfAbsent(executor, (k) -> new ExecutorCache(newBeat)); - currBeat.updateFromHb(timeout, newBeat); + currBeat.updateFromZkHb(timeout, newBeat); } } /** * Update the heartbeats for a given worker. + * * @param workerHeartbeat the heartbeats from the worker. * @param taskTimeoutSecs the timeout we should be looking at. */ @@ -176,22 +193,23 @@ public void updateHeartbeat(SupervisorWorkerHeartbeat workerHeartbeat, Integer t List executor = Arrays.asList(executorInfo.get_task_start(), executorInfo.get_task_end()); final Map newBeat = executorBeats.get(executor); ExecutorCache currBeat = topoCache.computeIfAbsent(executor, (k) -> new ExecutorCache(newBeat)); - currBeat.updateFromHb(taskTimeoutSecs, newBeat); + currBeat.updateFromRpcHb(taskTimeoutSecs); } } /** * Get all of the alive executors for a given topology. - * @param topoId the id of the topology we are looking for. - * @param allExecutors all of the executors for this topology. - * @param assignment the current topology assignment. + * + * @param topoId the id of the topology we are looking for. + * @param allExecutors all of the executors for this topology. + * @param assignment the current topology assignment. * @param taskLaunchSecs timeout for right after a worker is launched. * @return the set of tasks that are alive. */ public Set> getAliveExecutors(String topoId, Set> allExecutors, Assignment assignment, int taskLaunchSecs) { Map, ExecutorCache> topoCache = cache.computeIfAbsent(topoId, MAKE_MAP); LOG.debug("Computing alive executors for {}\nExecutors: {}\nAssignment: {}\nHeartbeat cache: {}", - topoId, allExecutors, assignment, topoCache); + topoId, allExecutors, assignment, topoCache); Set> ret = new HashSet<>(); Map, Long> execToStartTimes = assignment.get_executor_start_time_secs(); diff --git a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/HeartbeatCacheTest.java b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/HeartbeatCacheTest.java new file mode 100644 index 00000000000..5ae1ee31546 --- /dev/null +++ b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/HeartbeatCacheTest.java @@ -0,0 +1,195 @@ +/* + * 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.nimbus; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.storm.generated.Assignment; +import org.apache.storm.generated.ExecutorInfo; +import org.apache.storm.generated.NodeInfo; +import org.apache.storm.generated.SupervisorWorkerHeartbeat; +import org.apache.storm.stats.ClientStatsUtil; +import org.apache.storm.utils.Time; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class HeartbeatCacheTest { + private static final String TOPO_ID = "test-topology-1"; + private static final int TIMEOUT_SECS = 30; + + @Test + void testExecutorRemainsAliveWhenHeartbeatTimestampDoesNotAdvance() { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + HeartbeatCache cache = new HeartbeatCache(); + Set> allExecutors = Collections.singleton(Arrays.asList(1, 1)); + Assignment assignment = mkAssignment(Time.currentTimeSecs(), 1, 1); + + // First heartbeat at t=0 with TIME_SECS=100 + cache.updateHeartbeat(mkWorkerHeartbeat(TOPO_ID, 100, 1, 1), TIMEOUT_SECS); + + // Advance time to just before timeout + Time.advanceTimeSecs(TIMEOUT_SECS - 1); + + // Second heartbeat arrives but TIME_SECS is still 100 (same second, stats not rotated) + cache.updateHeartbeat(mkWorkerHeartbeat(TOPO_ID, 100, 1, 1), TIMEOUT_SECS); + + // Advance just 2 more seconds: now t = TIMEOUT_SECS + 1, which is past the original + // timeout window (rooted at t=0) but well within the refreshed window (rooted at t=TIMEOUT_SECS-1). + Time.advanceTimeSecs(2); + + // Simulate the scheduling cycle timeout check + cache.timeoutOldHeartbeats(TOPO_ID, TIMEOUT_SECS); + + // Executor should still be alive because a fresh heartbeat was received at t=(TIMEOUT_SECS-1) + Set> alive = cache.getAliveExecutors(TOPO_ID, allExecutors, assignment, TIMEOUT_SECS); + assertFalse(alive.isEmpty(), "Executor should be alive after receiving a recent heartbeat even if TIME_SECS did not advance"); + } + } + + @Test + void testExecutorTimesOutWhenNoHeartbeatReceived() { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + HeartbeatCache cache = new HeartbeatCache(); + Set> allExecutors = Collections.singleton(Arrays.asList(1, 1)); + Assignment assignment = mkAssignment(Time.currentTimeSecs(), 1, 1); + + // Single heartbeat at t=0 + cache.updateHeartbeat(mkWorkerHeartbeat(TOPO_ID, 100, 1, 1), TIMEOUT_SECS); + + // No more heartbeats — advance time past the timeout + Time.advanceTimeSecs(TIMEOUT_SECS + 1); + + // Simulate the scheduling cycle timeout check + cache.timeoutOldHeartbeats(TOPO_ID, TIMEOUT_SECS); + + Set> alive = cache.getAliveExecutors(TOPO_ID, allExecutors, assignment, TIMEOUT_SECS); + assertTrue(alive.isEmpty(), "Executor should be timed out after no heartbeat for longer than timeout"); + } + } + + @Test + void testExecutorAliveWithRegularHeartbeats() { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + HeartbeatCache cache = new HeartbeatCache(); + Set> allExecutors = Collections.singleton(Arrays.asList(1, 1)); + Assignment assignment = mkAssignment(Time.currentTimeSecs(), 1, 1); + + // Send heartbeats every second for 60 seconds + for (int t = 0; t < 60; t++) { + cache.updateHeartbeat(mkWorkerHeartbeat(TOPO_ID, Time.currentTimeSecs(), 1, 1), TIMEOUT_SECS); + Time.advanceTimeSecs(1); + } + + // Simulate the scheduling cycle timeout check + cache.timeoutOldHeartbeats(TOPO_ID, TIMEOUT_SECS); + + Set> alive = cache.getAliveExecutors(TOPO_ID, allExecutors, assignment, TIMEOUT_SECS); + assertFalse(alive.isEmpty(), "Executor should be alive when receiving regular heartbeats"); + } + } + + @Test + void testZkExecutorTimesOutWhenTimeSecsStopsAdvancing() { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + HeartbeatCache cache = new HeartbeatCache(); + Set> allExecutors = Collections.singleton(Arrays.asList(1, 1)); + Assignment assignment = mkAssignment(Time.currentTimeSecs(), 1, 1); + + // Heartbeats with advancing TIME_SECS — executor is healthy + for (int t = 0; t < 5; t++) { + cache.updateFromZkHeartbeat(TOPO_ID, mkZkExecutorBeats(1, 1, t * 10), allExecutors, TIMEOUT_SECS); + Time.advanceTimeSecs(1); + } + + // TIME_SECS freezes — zombie executor keeps sending heartbeats but stats are stuck + int frozenTimeSecs = 40; + for (int t = 0; t < TIMEOUT_SECS + 1; t++) { + cache.updateFromZkHeartbeat(TOPO_ID, mkZkExecutorBeats(1, 1, frozenTimeSecs), allExecutors, TIMEOUT_SECS); + Time.advanceTimeSecs(1); + } + + cache.timeoutOldHeartbeats(TOPO_ID, TIMEOUT_SECS); + + Set> alive = cache.getAliveExecutors(TOPO_ID, allExecutors, assignment, TIMEOUT_SECS); + assertTrue(alive.isEmpty(), "ZK executor should be timed out when TIME_SECS stops advancing (zombie detection)"); + } + } + + @Test + void testZkExecutorAliveWhenTimeSecsAdvances() { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + HeartbeatCache cache = new HeartbeatCache(); + Set> allExecutors = Collections.singleton(Arrays.asList(1, 1)); + Assignment assignment = mkAssignment(Time.currentTimeSecs(), 1, 1); + + // Heartbeats with advancing TIME_SECS every second + for (int t = 0; t < 60; t++) { + cache.updateFromZkHeartbeat(TOPO_ID, mkZkExecutorBeats(1, 1, t), allExecutors, TIMEOUT_SECS); + Time.advanceTimeSecs(1); + } + + cache.timeoutOldHeartbeats(TOPO_ID, TIMEOUT_SECS); + + Set> alive = cache.getAliveExecutors(TOPO_ID, allExecutors, assignment, TIMEOUT_SECS); + assertFalse(alive.isEmpty(), "ZK executor should be alive when TIME_SECS advances regularly"); + } + } + + private SupervisorWorkerHeartbeat mkWorkerHeartbeat(String topoId, int timeSecs, int... executors) { + SupervisorWorkerHeartbeat hb = new SupervisorWorkerHeartbeat(); + hb.set_storm_id(topoId); + hb.set_time_secs(timeSecs); + for (int i = 0; i < executors.length - 1; i += 2) { + ExecutorInfo info = new ExecutorInfo(); + info.set_task_start(executors[i]); + info.set_task_end(executors[i + 1]); + hb.add_to_executors(info); + } + return hb; + } + + + private Map, Map> mkZkExecutorBeats(int taskStart, int taskEnd, int timeSecs) { + Map beat = new HashMap<>(); + beat.put(ClientStatsUtil.TIME_SECS, timeSecs); + return Collections.singletonMap(Arrays.asList(taskStart, taskEnd), beat); + } + + private Assignment mkAssignment(int startTimeSecs, int... executors) { + Assignment assignment = new Assignment(); + Map, Long> execToStartTime = new HashMap<>(); + Map, NodeInfo> execToNodePort = new HashMap<>(); + NodeInfo nodeInfo = new NodeInfo("node1", Collections.singleton(6700L)); + for (int i = 0; i < executors.length - 1; i += 2) { + List exec = Arrays.asList((long) executors[i], (long) executors[i + 1]); + execToStartTime.put(exec, (long) startTimeSecs); + execToNodePort.put(exec, nodeInfo); + } + assignment.set_executor_start_time_secs(execToStartTime); + assignment.set_executor_node_port(execToNodePort); + return assignment; + } +}