(impala) 02/03: IMPALA-13399: Don't take the stub lock in subscriber RPC handlers

[la...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git

commit 15cc045e2dfdcb129d4c8c7c49c4e2c460c320d0
Author: Petr Fedchenkov <gi...@gmail.com>
AuthorDate: Tue Oct 20 15:50:58 2026 +0300

    IMPALA-13399: Don't take the stub lock in subscriber RPC handlers
    
    With statestored HA, the standby statestored asks every subscriber in
    its heartbeats for the state of the connection with the active
    statestored. StatestoreSubscriber::Heartbeat() answers with
    GetStatestoreConnState() of the other StatestoreStub, which takes that
    stub's lock_. RecoveryModeChecker() holds this lock for its whole
    re-registration loop. So while the active statestored is down and the
    subscribers try to re-register with it, the heartbeats of the standby
    statestored time out, and it unregisters every subscriber, catalogds
    included. When it takes over, it designates the catalogd that
    re-registers first, which can switch the active catalogd. It also keeps
    the standby statestored from learning that the subscribers lost the
    active one, which delays the takeover; which matches the heartbeat
    timeouts the statestored HA tests saw.
    
    GetStatestoreConnState() and IncCountForUpdateStatestoredRoleRPC() no
    longer take lock_. The failure detector has its own lock, and the
    counter is atomic. The stale "Not thread safe" comments of the failure
    detectors are fixed.
    
    Testing:
     - Added test_standby_heartbeats_during_subscriber_recovery. It keeps
       the standby statestored from taking over for 90 s after the active
       one is killed. Without the fix the standby statestored unregistered
       the subscribers; with the fix it passed.
     - Ran test_statestored_ha.py and test_catalogd_ha.py with the core
       exploration strategy, and all tests passed.
    
    Change-Id: I4dfcf768b2b5ebde9e37bd0950e90de383ad8681
    Assisted-by: Claude Opus 5.5 (Claude Code)
    Reviewed-on: http://gerrit.cloudera.org:8080/25151
    Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
    Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
 be/src/statestore/failure-detector.h        |  4 +--
 be/src/statestore/statestore-subscriber.cc  |  2 --
 tests/custom_cluster/test_statestored_ha.py | 38 +++++++++++++++++++++++++++++
 3 files changed, 40 insertions(+), 4 deletions(-)

diff --git a/be/src/statestore/failure-detector.h b/be/src/statestore/failure-detector.h
index 6d2700a3e9..6955a7bee3 100644
--- a/be/src/statestore/failure-detector.h
+++ b/be/src/statestore/failure-detector.h
@@ -70,7 +70,7 @@ class FailureDetector {
 /// considered suspected.
 /// A timeout failure detector is most appropriate for a client which is
 /// receiving periodic heartbeats.
-/// Not thread safe.
+/// Thread safe.
 class TimeoutFailureDetector : public FailureDetector {
  public:
   TimeoutFailureDetector(boost::posix_time::time_duration failure_timeout,
@@ -106,7 +106,7 @@ class TimeoutFailureDetector : public FailureDetector {
 /// The MissedHeartbeatFailureDetector is most appropriate when heartbeats are
 /// being sent, not being received, because it is easy in that situation to tell
 /// when a heartbeat has not been successful.
-/// Not thread safe.
+/// Thread safe.
 class MissedHeartbeatFailureDetector : public FailureDetector {
  public:
   /// max_missed_heartbeats -> the number of heartbeats that can be missed before a
diff --git a/be/src/statestore/statestore-subscriber.cc b/be/src/statestore/statestore-subscriber.cc
index 5ce6d392b5..c339c20fea 100644
--- a/be/src/statestore/statestore-subscriber.cc
+++ b/be/src/statestore/statestore-subscriber.cc
@@ -1162,13 +1162,11 @@ void StatestoreSubscriber::StatestoreStub::GetRegistrationIdAndStatestoreId(
 }
 
 void StatestoreSubscriber::StatestoreStub::IncCountForUpdateStatestoredRoleRPC() {
-  lock_guard<shared_mutex> exclusive_lock(lock_);
   update_statestored_role_rpc_metric_->Increment(1);
 }
 
 TStatestoreConnState::type
 StatestoreSubscriber::StatestoreStub::GetStatestoreConnState() {
-  lock_guard<shared_mutex> exclusive_lock(lock_);
   FailureDetector::PeerState state = failure_detector_->GetPeerState(STATESTORE_ID);
   switch (state) {
     case FailureDetector::FAILED:
diff --git a/tests/custom_cluster/test_statestored_ha.py b/tests/custom_cluster/test_statestored_ha.py
index 1c3673cf0d..9d2465520d 100644
--- a/tests/custom_cluster/test_statestored_ha.py
+++ b/tests/custom_cluster/test_statestored_ha.py
@@ -923,6 +923,44 @@ class TestStatestoredHA(CustomClusterTestSuite):
           if "Skipped updating catalogd message" in line: skipped += 1
       LOG.info("{0} skipped {1} UpdateCatalogd RPCs".format(standby_log, skipped))
 
+  @pytest.mark.execute_serially
+  @CustomClusterTestSuite.with_args(
+    statestored_args="--use_network_address_as_statestore_priority=true "
+                     "--statestore_peer_timeout_seconds=120",
+    impalad_args="--statestore_subscriber_timeout_seconds=10",
+    catalogd_args="--statestore_subscriber_timeout_seconds=10",
+    start_args="--enable_statestored_ha --enable_catalogd_ha",
+    disable_log_buffering=True)
+  def test_standby_heartbeats_during_subscriber_recovery(self):
+    """The active statestoreds[0] dies and the standby statestoreds[1] does not take over
+    yet. The subscribers try to re-register with statestoreds[0] in recovery mode. Their
+    replies to the heartbeats of statestoreds[1] must not wait for that, or
+    statestoreds[1] unregisters all subscribers, including the catalogds."""
+    statestoreds = self.cluster.statestoreds()
+    assert len(statestoreds) == 2
+    statestore_service_1 = statestoreds[1].service
+    statestore_service_1.wait_for_live_subscribers(5, timeout=60)
+    statestoreds[0].kill()
+    # The subscribers lose statestoreds[0] and keep re-registering with it.
+    for daemon in ("impalad", "impalad_node1", "impalad_node2", "catalogd",
+                   "catalogd_node1"):
+      self.assert_log_contains(daemon, "INFO",
+          "Connection with statestore lost, entering recovery mode", expected_count=-1,
+          timeout_s=60)
+    # Meanwhile statestoreds[1] must keep heartbeating them; before the fix the heartbeats
+    # timed out. Wait for 10 rounds (statestore_max_missed_heartbeats) to 5 subscribers.
+    num_heartbeats = statestore_service_1.get_metric_value(
+        "statestore.heartbeat-durations")["count"]
+    start_time = time.time()
+    while statestore_service_1.get_metric_value(
+        "statestore.heartbeat-durations")["count"] < num_heartbeats + 50:
+      assert time.time() - start_time < 60, "Heartbeats of statestoreds[1] stalled"
+      sleep(1)
+    assert not statestore_service_1.get_metric_value("statestore.active-status")
+    self.assert_log_contains("statestored_node1", "INFO",
+        "has failed, disconnected or re-registered", expected_count=0)
+    assert statestore_service_1.get_metric_value("statestore.live-backends") == 5
+
 
 class TestStatestoredHAStartupDelay(CustomClusterTestSuite):
   """This test injects a real delay in statestored startup. The impalads and catalogd are