reiabreu commented on code in PR #9153:
URL: https://github.com/apache/storm/pull/9153#discussion_r4431727923
##########
storm-server/src/main/java/org/apache/storm/daemon/supervisor/timer/SupervisorHeartbeat.java:
##########
@@ -162,7 +164,17 @@ public void run() {
Map<String, Object> validatedNumaMap =
SupervisorUtils.getNumaMap(conf);
Map<String, SupervisorInfo> supervisorInfoList =
buildSupervisorInfo(conf, supervisor, validatedNumaMap);
for (Map.Entry<String, SupervisorInfo> supervisorInfoEntry :
supervisorInfoList.entrySet()) {
-
stormClusterState.supervisorHeartbeat(supervisorInfoEntry.getKey(),
supervisorInfoEntry.getValue());
+ try {
+
stormClusterState.supervisorHeartbeat(supervisorInfoEntry.getKey(),
supervisorInfoEntry.getValue());
+ } catch (Exception e) {
+ if (!Utils.exceptionCauseIsInstanceOf(KeeperException.class,
e)) {
+ throw e;
+ }
+ // A transient ZK failure: the ephemeral node is recreated on
the next heartbeat.
+ // Local/config errors are rethrown so a persistent
misconfiguration still surfaces.
+ LOG.warn("Supervisor ZK heartbeat failed for {}, will retry
next cycle",
+ supervisorInfoEntry.getKey(), e);
+ }
}
}
}
Review Comment:
Narrows the catch to connection-level failures and adds the helper from the
comment above:
```suggestion
if (!isTransientZkFailure(e)) {
throw e;
}
// A transient ZK failure: the ephemeral node is recreated
on the next heartbeat.
// Local and persistent ZK errors (e.g. bad ACLs) are
rethrown so they surface.
LOG.warn("Supervisor ZK heartbeat failed for {}, will retry
next cycle",
supervisorInfoEntry.getKey(), e);
}
}
}
private static boolean isTransientZkFailure(Throwable t) {
return
Utils.exceptionCauseIsInstanceOf(KeeperException.ConnectionLossException.class,
t)
||
Utils.exceptionCauseIsInstanceOf(KeeperException.SessionExpiredException.class,
t)
||
Utils.exceptionCauseIsInstanceOf(KeeperException.OperationTimeoutException.class,
t);
}
}
```
##########
storm-server/src/test/java/org/apache/storm/daemon/supervisor/timer/SupervisorHeartbeatTest.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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 java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.storm.cluster.IStormClusterState;
+import org.apache.storm.daemon.supervisor.Supervisor;
+import org.apache.storm.generated.SupervisorInfo;
+import org.apache.storm.scheduler.ISupervisor;
+import org.apache.storm.shade.org.apache.zookeeper.KeeperException;
+import org.apache.storm.utils.Utils;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class SupervisorHeartbeatTest {
+
+ private static final String SUPERVISOR_ID = "supervisor-1";
+
+ private Supervisor mockSupervisor(IStormClusterState clusterState) {
+ Supervisor supervisor = mock(Supervisor.class);
+ ISupervisor iSupervisor = mock(ISupervisor.class);
+ when(iSupervisor.getMetadata()).thenReturn(Arrays.asList(6700, 6701));
+
+ when(supervisor.getStormClusterState()).thenReturn(clusterState);
+ when(supervisor.getId()).thenReturn(SUPERVISOR_ID);
+ when(supervisor.getiSupervisor()).thenReturn(iSupervisor);
+ when(supervisor.getCurrAssignment()).thenReturn(new
AtomicReference<>(new HashMap<>()));
+ when(supervisor.getHostName()).thenReturn("host-1");
+ when(supervisor.getAssignmentId()).thenReturn(SUPERVISOR_ID);
+ when(supervisor.getThriftServerPort()).thenReturn(6627);
+ when(supervisor.getUpTime()).thenReturn(Utils.makeUptimeComputer());
+ when(supervisor.getStormVersion()).thenReturn("test-version");
+ return supervisor;
+ }
+
+ private Map<String, Object> baseConf() {
+ return new HashMap<>(Utils.readStormConfig());
+ }
+
+ @Test
+ public void
transientZkHeartbeatFailureIsSwallowedInsteadOfKillingTheProcess() {
+ IStormClusterState clusterState = mock(IStormClusterState.class);
+ doThrow(new RuntimeException(
+ new KeeperException.ConnectionLossException()))
+ .when(clusterState).supervisorHeartbeat(anyString(),
any(SupervisorInfo.class));
+
+ Supervisor supervisor = mockSupervisor(clusterState);
+ SupervisorHeartbeat heartbeat = new SupervisorHeartbeat(baseConf(),
supervisor);
+
+ assertDoesNotThrow(heartbeat::run);
+ verify(clusterState, atLeastOnce()).supervisorHeartbeat(anyString(),
any(SupervisorInfo.class));
+ }
Review Comment:
Covers the other two transient failures, and checks that persistent ones
still propagate:
```suggestion
}
@Test
public void otherTransientZkFailuresAreSwallowed() {
for (KeeperException failure : Arrays.asList(new
KeeperException.SessionExpiredException(),
new KeeperException.OperationTimeoutException())) {
IStormClusterState clusterState = mock(IStormClusterState.class);
doThrow(new RuntimeException(failure))
.when(clusterState).supervisorHeartbeat(anyString(),
any(SupervisorInfo.class));
SupervisorHeartbeat heartbeat = new
SupervisorHeartbeat(baseConf(), mockSupervisor(clusterState));
assertDoesNotThrow(heartbeat::run);
}
}
@Test
public void persistentZkFailureStillPropagates() {
for (KeeperException failure : Arrays.asList(new
KeeperException.NoAuthException(),
new KeeperException.InvalidACLException())) {
IStormClusterState clusterState = mock(IStormClusterState.class);
doThrow(new RuntimeException(failure))
.when(clusterState).supervisorHeartbeat(anyString(),
any(SupervisorInfo.class));
SupervisorHeartbeat heartbeat = new
SupervisorHeartbeat(baseConf(), mockSupervisor(clusterState));
assertThrows(RuntimeException.class, heartbeat::run);
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
