GGraziadei commented on code in PR #9153: URL: https://github.com/apache/storm/pull/9153#discussion_r4367162794
########## storm-client/src/jvm/org/apache/storm/utils/ConnectionAwareRetryPolicy.java: ########## @@ -0,0 +1,146 @@ +/* + * 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.utils; + +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting; +import org.apache.storm.shade.org.apache.curator.RetryPolicy; +import org.apache.storm.shade.org.apache.curator.RetrySleeper; +import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework; +import org.apache.storm.shade.org.apache.curator.framework.state.ConnectionState; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link RetryPolicy} wrapper that makes the Curator retry loop aware of the + * ZooKeeper connection state. + * + * <p>When the connection is {@link ConnectionState#SUSPENDED} or + * {@link ConnectionState#LOST}, instead of blindly sleeping and retrying (which + * races against the ZK client's {@code SendThread} reconnection), this policy + * calls {@link CuratorFramework#blockUntilConnected} to yield to the + * {@code SendThread} and wait for it to failover to another ensemble member. + * + * <p>Once the connection is re-established, the retry loop immediately retries + * the operation on the new connection. If the connection cannot be + * re-established within the session timeout, the retry is abandoned. + * + * <p>For all other connection states, this policy delegates to the wrapped + * delegate policy (typically a {@link StormBoundedExponentialBackoffRetry}). + */ +public class ConnectionAwareRetryPolicy implements RetryPolicy { + + private static final Logger LOG = LoggerFactory.getLogger(ConnectionAwareRetryPolicy.class); + + /** + * Maximum jitter (in ms) applied after a successful reconnection to avoid + * thundering-herd when many suspended clients retry in lockstep. + */ + private static final long RECONNECT_JITTER_MS = 100; + + private final RetryPolicy delegate; + private final Supplier<CuratorFramework> zkSupplier; + private final int sessionTimeoutMs; + private final AtomicReference<ConnectionState> connectionState = + new AtomicReference<>(ConnectionState.CONNECTED); + + /** + * @param delegate the underlying retry policy to delegate to for normal (connected) retries + * @param zkSupplier supplier for the {@link CuratorFramework}, used for late binding since + * the framework does not exist until {@code builder.build()} returns + * @param sessionTimeoutMs upper bound (in ms) for waiting on reconnection; typically + * {@code storm.zookeeper.session.timeout} + */ + public ConnectionAwareRetryPolicy(RetryPolicy delegate, + Supplier<CuratorFramework> zkSupplier, + int sessionTimeoutMs) { + this.delegate = delegate; + this.zkSupplier = zkSupplier; + this.sessionTimeoutMs = sessionTimeoutMs; + } + + /** + * Register a {@link org.apache.storm.shade.org.apache.curator.framework.state.ConnectionStateListener} + * on the given framework to track the current connection state. + * Must be called after the {@link CuratorFramework} has been built. + * + * @param zk the built CuratorFramework + */ + public void bind(CuratorFramework zk) { + zk.getConnectionStateListenable().addListener((client, newState) -> { Review Comment: The state is updated asynchronously on the `ConnectionStateManager` thread, so a retry right after the disconnect can still see `CONNECTED` here (in my test I had to wait for this field to flip to `SUSPENDED`). `zk.getZookeeperClient().isConnected()` is updated synchronously and would also remove the need for the late-bound `AtomicReference` and `bind()` in `CuratorUtils`. ########## storm-client/src/jvm/org/apache/storm/utils/CuratorUtils.java: ########## @@ -61,6 +62,19 @@ public static CuratorFramework newCurator(Map<String, Object> conf, List<String> CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder(); setupBuilder(builder, zkStr, conf, auth); + + // Wrap with a connection-aware policy that yields to SendThread on SUSPENDED/LOST. Review Comment: This applies to every client built by `newCurator`, not only the supervisor: Nimbus, Trident `TransactionalState` (which runs on executor threads) and `BlobStoreUtils`. With this policy a ZooKeeper call on any of those threads can block up to a session timeout per retry instead of failing. If the intent is to protect the supervisor, it would be worth either scoping it or calling the broader impact out in the description. Nit: `setupBuilder` still builds a `StormBoundedExponentialBackoffRetry` that is immediately replaced here, and its new comment (line 101) sits above `connectString` rather than the `retryPolicy` call. ########## storm-server/src/main/java/org/apache/storm/daemon/supervisor/timer/SupervisorHeartbeat.java: ########## @@ -162,7 +162,13 @@ 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) { + // Liveness is tracked via ephemeral ZK nodes, not this write. Safe to skip. Review Comment: This comment is not accurate: `supervisorHeartbeat` is the ephemeral node write (`set_ephemeral_node` on `/supervisors/<id>`), and after a `LOST` it is the only thing that recreates the node. Skipping one cycle is fine, but the catch has no bound: a persistent, non-transient failure (e.g. `NoAuth`) would leave the supervisor invisible to Nimbus indefinitely while its workers keep running, where today the process halts and gets restarted. Suggest catching only ZooKeeper failures, e.g. `Utils.exceptionCauseIsInstanceOf(KeeperException.class, e)`, and rethrowing anything else (or after N consecutive failures). ########## storm-client/test/jvm/org/apache/storm/utils/ConnectionAwareRetryPolicyTest.java: ########## @@ -0,0 +1,174 @@ +/* + * 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.utils; + +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import org.apache.storm.shade.org.apache.curator.RetryPolicy; +import org.apache.storm.shade.org.apache.curator.RetrySleeper; +import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework; +import org.apache.storm.shade.org.apache.curator.framework.listen.Listenable; +import org.apache.storm.shade.org.apache.curator.framework.state.ConnectionState; +import org.apache.storm.shade.org.apache.curator.framework.state.ConnectionStateListener; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Unit tests for {@link ConnectionAwareRetryPolicy}. The ZooKeeper connection state is driven by + * capturing the {@link ConnectionStateListener} that {@link ConnectionAwareRetryPolicy#bind} registers + * and firing state transitions at it, so no real ZooKeeper is required. + */ +public class ConnectionAwareRetryPolicyTest { Review Comment: These tests mock `CuratorFramework`, so they cannot catch thread interactions with a real client. Could you add a test against a `TestingServer` that is stopped and restarted, covering a foreground operation that should succeed after the restart, and a call from the EventThread (e.g. from a `CuratorListener` on the `Disconnected` event) that must not cause `LOST`? ########## storm-client/src/jvm/org/apache/storm/utils/ConnectionAwareRetryPolicy.java: ########## @@ -0,0 +1,146 @@ +/* + * 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.utils; + +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting; +import org.apache.storm.shade.org.apache.curator.RetryPolicy; +import org.apache.storm.shade.org.apache.curator.RetrySleeper; +import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework; +import org.apache.storm.shade.org.apache.curator.framework.state.ConnectionState; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link RetryPolicy} wrapper that makes the Curator retry loop aware of the + * ZooKeeper connection state. + * + * <p>When the connection is {@link ConnectionState#SUSPENDED} or + * {@link ConnectionState#LOST}, instead of blindly sleeping and retrying (which + * races against the ZK client's {@code SendThread} reconnection), this policy + * calls {@link CuratorFramework#blockUntilConnected} to yield to the + * {@code SendThread} and wait for it to failover to another ensemble member. + * + * <p>Once the connection is re-established, the retry loop immediately retries + * the operation on the new connection. If the connection cannot be + * re-established within the session timeout, the retry is abandoned. + * + * <p>For all other connection states, this policy delegates to the wrapped + * delegate policy (typically a {@link StormBoundedExponentialBackoffRetry}). + */ +public class ConnectionAwareRetryPolicy implements RetryPolicy { + + private static final Logger LOG = LoggerFactory.getLogger(ConnectionAwareRetryPolicy.class); + + /** + * Maximum jitter (in ms) applied after a successful reconnection to avoid + * thundering-herd when many suspended clients retry in lockstep. + */ + private static final long RECONNECT_JITTER_MS = 100; + + private final RetryPolicy delegate; + private final Supplier<CuratorFramework> zkSupplier; + private final int sessionTimeoutMs; + private final AtomicReference<ConnectionState> connectionState = + new AtomicReference<>(ConnectionState.CONNECTED); + + /** + * @param delegate the underlying retry policy to delegate to for normal (connected) retries + * @param zkSupplier supplier for the {@link CuratorFramework}, used for late binding since + * the framework does not exist until {@code builder.build()} returns + * @param sessionTimeoutMs upper bound (in ms) for waiting on reconnection; typically + * {@code storm.zookeeper.session.timeout} + */ + public ConnectionAwareRetryPolicy(RetryPolicy delegate, + Supplier<CuratorFramework> zkSupplier, + int sessionTimeoutMs) { + this.delegate = delegate; + this.zkSupplier = zkSupplier; + this.sessionTimeoutMs = sessionTimeoutMs; + } + + /** + * Register a {@link org.apache.storm.shade.org.apache.curator.framework.state.ConnectionStateListener} + * on the given framework to track the current connection state. + * Must be called after the {@link CuratorFramework} has been built. + * + * @param zk the built CuratorFramework + */ + public void bind(CuratorFramework zk) { + zk.getConnectionStateListenable().addListener((client, newState) -> { + ConnectionState prev = connectionState.getAndSet(newState); + if (prev != newState) { + LOG.debug("ZK connection state changed: {} -> {}", prev, newState); + } + }); + } + + @VisibleForTesting + RetryPolicy getDelegate() { + return delegate; + } + + @Override + public boolean allowRetry(int retryCount, long elapsedTimeMs, RetrySleeper sleepSleeper) { + ConnectionState state = connectionState.get(); + + if (state == ConnectionState.SUSPENDED || state == ConnectionState.LOST) { + // Honour the configured retry budget even while suspended. + if (!delegate.allowRetry(retryCount, elapsedTimeMs, sleepSleeper)) { + return false; + } + + CuratorFramework zk = zkSupplier.get(); + if (zk == null) { + // Framework not yet available; delegate already approved the retry + return true; + } + + LOG.info("ZK connection is {} on retry {}, waiting for reconnection (timeout {}ms)", + state, retryCount, sessionTimeoutMs); + try { + boolean reconnected = zk.blockUntilConnected(sessionTimeoutMs, TimeUnit.MILLISECONDS); Review Comment: This is the blocking point from the summary: when `allowRetry` is called from the ZooKeeper EventThread (background operation retries via `checkBackgroundRetry`), this wait prevents the `SyncConnected` event from being processed, Curator stays `SUSPENDED`, and after the session timeout it injects a session expiration. In my test the server came back after 0.6 s, but the call blocked for the full 6 s session timeout, returned `false`, and the client went to `LOST`. -- 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]
