PHOENIX-7990 Reconcile convergent CAS race in setHAGroupStatusIfNeeded to avoid RS aborts

phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java

@@ -110,6 +110,10 @@ public class HAGroupStoreClient implements Closeable {
   private static final long SYNC_JOB_MAX_JITTER_SECONDS = 10;
   // Exclusive upper bound for initial-delay jitter on the periodic reconciler (0..30s).
   private static final long LEGACY_CRR_SYNC_JOB_MAX_JITTER_SECONDS = 31;
+  // Total CAS attempts for the shared HA-status write. The first attempt uses the cache-derived
+  // version; each subsequent attempt re-reads the version fresh from ZK. Bounds the convergent-race
+  // reconcile loop so genuine contention still surfaces after a small number of tries.
+  private static final int SET_HA_GROUP_STATUS_MAX_ATTEMPTS = 3;
   private PhoenixHAAdmin phoenixHaAdmin;
   // Admin + NodeCache on /phoenix/ha; null when feature disabled.
   private volatile PhoenixHAAdmin legacyHaAdmin;
@@ -411,67 +415,98 @@ public long setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState haGroupStat
     if (!isHealthy) {
       throw new IOException("HAGroupStoreClient is not healthy");
     }
-    Pair<HAGroupStoreRecord, Stat> cacheRecord = fetchLocalRecordAndPopulateZKIfNeeded();
-    HAGroupStoreRecord currentHAGroupStoreRecord = cacheRecord.getLeft();
-    Stat currentHAGroupStoreRecordStat = cacheRecord.getRight();
-    if (currentHAGroupStoreRecord == null) {
-      throw new IOException("Current HAGroupStoreRecordStat in cache is null, "
-        + "cannot update HAGroupStoreRecord, the record should be initialized "
-        + "in System Table first" + haGroupName);
-    }
-    long stateTransitionWaitTime =
-      validateTransitionAndGetWaitTime(currentHAGroupStoreRecord.getHAGroupState(),
-        currentHAGroupStoreRecordStat.getMtime(), haGroupState);
-    if (stateTransitionWaitTime > 0) {
-      LOGGER.info("Not updating HAGroupStoreRecord for HA group {} with state {}", haGroupName,
-        haGroupState);
-      return stateTransitionWaitTime;
-    }
-    // We maintain last sync time as the last time cluster was in sync state.
-    // If state changes from ACTIVE_IN_SYNC to ACTIVE_NOT_IN_SYNC, record that time
-    // Once state changes back to ACTIVE_IN_SYNC or the role is
-    // NOT ACTIVE or ACTIVE_TO_STANDBY
-    // set the time to null to mark that we are current(or we don't have any reader).
-    long lastSyncTimeInMs = lastSyncTimeInMsNullable != null
-      ? lastSyncTimeInMsNullable
-      : currentHAGroupStoreRecord.getLastSyncStateTimeInMs();
-    ClusterRole clusterRole = haGroupState.getClusterRole();
-    if (
-      currentHAGroupStoreRecord.getHAGroupState() == HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC
-        && haGroupState == HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC
-    ) {
-      // We record the last round timestamp by subtracting the rotationTime and then
-      // taking the beginning of last round (floor) by first integer
-      // division and then multiplying again.
-      lastSyncTimeInMs =
-        ((System.currentTimeMillis() - rotationTimeMs) / rotationTimeMs) * rotationTimeMs;
-    }
-    HAGroupStoreRecord newHAGroupStoreRecord = new HAGroupStoreRecord(
-      currentHAGroupStoreRecord.getProtocolVersion(), currentHAGroupStoreRecord.getHaGroupName(),
-      haGroupState, lastSyncTimeInMs, currentHAGroupStoreRecord.getPolicy(),
-      currentHAGroupStoreRecord.getPeerZKUrl(), currentHAGroupStoreRecord.getClusterUrl(),
-      currentHAGroupStoreRecord.getPeerClusterUrl(), currentHAGroupStoreRecord.getHdfsUrl(),
-      currentHAGroupStoreRecord.getPeerHdfsUrl(), currentHAGroupStoreRecord.getAdminCRRVersion());
-    phoenixHaAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName, newHAGroupStoreRecord,
-      currentHAGroupStoreRecordStat.getVersion());
-    // If cluster role is changing, if so, we update,
-    // the system table on best effort basis.
-    // We also have a periodic job which syncs the ZK
-    // state with System Table periodically.
-    if (currentHAGroupStoreRecord.getClusterRole() != clusterRole) {
-      HAGroupStoreRecord peerZkRecord = getHAGroupStoreRecordFromPeer();
-      ClusterRoleRecord.ClusterRole peerClusterRole = peerZkRecord != null
-        ? peerZkRecord.getClusterRole()
-        : ClusterRoleRecord.ClusterRole.UNKNOWN;
-      SystemTableHAGroupRecord systemTableRecord = new SystemTableHAGroupRecord(
-        HighAvailabilityPolicy.valueOf(newHAGroupStoreRecord.getPolicy()), clusterRole,
-        peerClusterRole, newHAGroupStoreRecord.getClusterUrl(),
-        newHAGroupStoreRecord.getPeerClusterUrl(), this.zkUrl, newHAGroupStoreRecord.getPeerZKUrl(),
-        newHAGroupStoreRecord.getHdfsUrl(), newHAGroupStoreRecord.getPeerHdfsUrl(),
-        newHAGroupStoreRecord.getAdminCRRVersion());
-      updateSystemTableHAGroupRecordSilently(haGroupName, systemTableRecord);
-    }
-    return 0L;
+    // First attempt uses the cache-derived (record, version) — the fast, uncontended path. On a
+    // stale-version CAS loss (a peer RS already advanced the shared record), later attempts re-read
+    // fresh from ZK and reconcile: if the target is already met the write is a no-op success, so a
+    // convergent race no longer surfaces as a fatal StaleHAGroupStoreRecordVersionException.
+    Pair<HAGroupStoreRecord, Stat> currentRecordAndStat = fetchLocalRecordAndPopulateZKIfNeeded();
+    for (int attempt = 1;; attempt++) {
+      HAGroupStoreRecord currentHAGroupStoreRecord = currentRecordAndStat.getLeft();
+      Stat currentHAGroupStoreRecordStat = currentRecordAndStat.getRight();
+      if (currentHAGroupStoreRecord == null) {
+        throw new IOException("Current HAGroupStoreRecordStat in cache is null, "
+          + "cannot update HAGroupStoreRecord, the record should be initialized "
+          + "in System Table first" + haGroupName);
+      }
+      // Convergent race: on the retry path (attempt > 1) currentHAGroupStoreRecord is the fresh ZK
+      // re-read (not the watch-lagged cache), so finding it already at the target means a peer won
+      // the CAS -- treat as a no-op success without ever concluding success from stale data. Runs
+      // before validateTransitionAndGetWaitTime so a converged non-self-transitionable target (e.g.
+      // ACTIVE_IN_SYNC) is a no-op rather than an InvalidClusterRoleTransitionException on the X->X
+      // self-transition. Attempt 1 never short-circuits, so the periodic ACTIVE_NOT_IN_SYNC
+      // heartbeat still writes its mtime bump.
+      if (attempt > 1 && currentHAGroupStoreRecord.getHAGroupState() == haGroupState) {
+        LOGGER.info("HAGroupStoreRecord for HA group {} is already at state {} after a fresh "
+          + "re-read, treating update as a no-op success", haGroupName, haGroupState);
+        return 0L;
+      }
+      long stateTransitionWaitTime =
+        validateTransitionAndGetWaitTime(currentHAGroupStoreRecord.getHAGroupState(),
+          currentHAGroupStoreRecordStat.getMtime(), haGroupState);
+      if (stateTransitionWaitTime > 0) {
+        LOGGER.info("Not updating HAGroupStoreRecord for HA group {} with state {}", haGroupName,
+          haGroupState);
+        return stateTransitionWaitTime;
+      }
+      // We maintain last sync time as the last time cluster was in sync state.
+      // If state changes from ACTIVE_IN_SYNC to ACTIVE_NOT_IN_SYNC, record that time
+      // Once state changes back to ACTIVE_IN_SYNC or the role is
+      // NOT ACTIVE or ACTIVE_TO_STANDBY
+      // set the time to null to mark that we are current(or we don't have any reader).
+      long lastSyncTimeInMs = lastSyncTimeInMsNullable != null
+        ? lastSyncTimeInMsNullable
+        : currentHAGroupStoreRecord.getLastSyncStateTimeInMs();
+      ClusterRole clusterRole = haGroupState.getClusterRole();
+      if (
+        currentHAGroupStoreRecord.getHAGroupState()
+            == HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC
+          && haGroupState == HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC
+      ) {
+        // We record the last round timestamp by subtracting the rotationTime and then
+        // taking the beginning of last round (floor) by first integer
+        // division and then multiplying again.
+        lastSyncTimeInMs =
+          ((System.currentTimeMillis() - rotationTimeMs) / rotationTimeMs) * rotationTimeMs;
+      }
+      HAGroupStoreRecord newHAGroupStoreRecord = new HAGroupStoreRecord(
+        currentHAGroupStoreRecord.getProtocolVersion(), currentHAGroupStoreRecord.getHaGroupName(),
+        haGroupState, lastSyncTimeInMs, currentHAGroupStoreRecord.getPolicy(),
+        currentHAGroupStoreRecord.getPeerZKUrl(), currentHAGroupStoreRecord.getClusterUrl(),
+        currentHAGroupStoreRecord.getPeerClusterUrl(), currentHAGroupStoreRecord.getHdfsUrl(),
+        currentHAGroupStoreRecord.getPeerHdfsUrl(), currentHAGroupStoreRecord.getAdminCRRVersion());
+      try {
+        phoenixHaAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName, newHAGroupStoreRecord,
+          currentHAGroupStoreRecordStat.getVersion());
+      } catch (StaleHAGroupStoreRecordVersionException e) {
+        if (attempt >= SET_HA_GROUP_STATUS_MAX_ATTEMPTS) {
+          throw e;
+        }
+        // A peer RS advanced the record between our read and this CAS. Re-read fresh from ZK (not
+        // the watch-lagged cache) and retry; the loop head reconciles against the fresh state.
+        LOGGER.info("Stale-version CAS for HA group {} (attempt {}/{}); re-reading fresh from ZK "
+          + "and retrying", haGroupName, attempt, SET_HA_GROUP_STATUS_MAX_ATTEMPTS);
+        currentRecordAndStat = phoenixHaAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName);
+        continue;
+      }
+      // If cluster role is changing, if so, we update,
+      // the system table on best effort basis.
+      // We also have a periodic job which syncs the ZK
+      // state with System Table periodically.
+      if (currentHAGroupStoreRecord.getClusterRole() != clusterRole) {
+        HAGroupStoreRecord peerZkRecord = getHAGroupStoreRecordFromPeer();
+        ClusterRoleRecord.ClusterRole peerClusterRole = peerZkRecord != null
+          ? peerZkRecord.getClusterRole()
+          : ClusterRoleRecord.ClusterRole.UNKNOWN;
+        SystemTableHAGroupRecord systemTableRecord = new SystemTableHAGroupRecord(
+          HighAvailabilityPolicy.valueOf(newHAGroupStoreRecord.getPolicy()), clusterRole,
+          peerClusterRole, newHAGroupStoreRecord.getClusterUrl(),
+          newHAGroupStoreRecord.getPeerClusterUrl(), this.zkUrl,
+          newHAGroupStoreRecord.getPeerZKUrl(), newHAGroupStoreRecord.getHdfsUrl(),
+          newHAGroupStoreRecord.getPeerHdfsUrl(), newHAGroupStoreRecord.getAdminCRRVersion());
+        updateSystemTableHAGroupRecordSilently(haGroupName, systemTableRecord);
+      }
+      return 0L;
+    }
   }
 
   /**

phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java

@@ -364,7 +364,20 @@ public long setHAGroupStatusToSync(final String haGroupName) throws IOException,
         haGroupStoreRecord.getHAGroupState() == HAGroupState.ACTIVE_NOT_IN_SYNC_TO_STANDBY
           ? ACTIVE_IN_SYNC_TO_STANDBY
           : ACTIVE_IN_SYNC;
-      return haGroupStoreClient.setHAGroupStatusIfNeeded(targetHAGroupState);
+      try {
+        return haGroupStoreClient.setHAGroupStatusIfNeeded(targetHAGroupState);
+      } catch (InvalidClusterRoleTransitionException e) {
+        // Convergent race: co-active RS forwarders independently drive the group to the sync
+        // target. When a watch delivers a peer's winning write before this RS fires, the cache is
+        // already at the target and setHAGroupStatusIfNeeded rejects the X -> X self-transition
+        // (only ACTIVE_NOT_IN_SYNC self-transitions). The group-level goal is already met, so treat
+        // it as a no-op success. Guarded on current == target so a genuinely invalid transition
+        // still propagates.
+        if (isStateAlreadyUpdated(haGroupStoreClient, haGroupName, targetHAGroupState)) {
+          return 0L;
+        }
+        throw e;
+      }
     } else {
       throw new IOException("Current HAGroupStoreRecord is null for HA group: " + haGroupName);
     }

phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdmin.java

@@ -530,8 +530,10 @@ public void updateHAGroupStoreRecordInZooKeeper(String haGroupName,
       getCurator().setData().withVersion(currentStatVersion).forPath(toPath(haGroupName),
         HAGroupStoreRecord.toJson(newHAGroupStoreRecord));
     } catch (KeeperException.BadVersionException e) {
-      LOG.error("Failed to set HAGroupStoreRecord for HA group {}, stale stat version", haGroupName,
-        e);
+      // A stale-version CAS loss is a routine optimistic-lock outcome under concurrent writers
+      // (e.g. co-active RegionServers refreshing the same status). The typed exception below lets
+      // the caller decide whether it is benign, so log at DEBUG here to avoid stacktrace spam.
+      LOG.debug("Stale stat version setting HAGroupStoreRecord for HA group {}", haGroupName, e);
       throw new StaleHAGroupStoreRecordVersionException(
         "Failed to set HAGroupStoreRecord for HA group " + haGroupName
           + " with cached stat version " + currentStatVersion,

phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java

@@ -23,8 +23,8 @@
  * Factory for process-lifetime HAGroupStore metric sources.
  * <p>
  * Creates one source lazily per HA group and intentionally retains it for the JVM lifetime so
- * cumulative counters survive client replacement. HA-group names are expected to be a small,
- * stable set within a process.
+ * cumulative counters survive client replacement. HA-group names are expected to be a small, stable
+ * set within a process.
  */
 public final class HAGroupStoreMetricsSourceFactory {
 

phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java

@@ -58,10 +58,10 @@ public HAGroupStoreMetricsSourceImpl(String haGroupName) {
       metricsJmxContext + ",haGroup=" + ObjectName.quote(haGroupName));
     getMetricsRegistry().tag(Interns.info(HA_GROUP_TAG_NAME, HA_GROUP_TAG_DESC), haGroupName);
 
-    localCacheHealthStatus = getMetricsRegistry().newGauge(LOCAL_CACHE_HEALTH_STATUS,
-      LOCAL_CACHE_HEALTH_STATUS_DESC, 1L);
-    peerVisibilityStatus = getMetricsRegistry().newGauge(PEER_VISIBILITY_STATUS,
-      PEER_VISIBILITY_STATUS_DESC, 1L);
+    localCacheHealthStatus =
+      getMetricsRegistry().newGauge(LOCAL_CACHE_HEALTH_STATUS, LOCAL_CACHE_HEALTH_STATUS_DESC, 1L);
+    peerVisibilityStatus =
+      getMetricsRegistry().newGauge(PEER_VISIBILITY_STATUS, PEER_VISIBILITY_STATUS_DESC, 1L);
     degradedStandbyActive =
       getMetricsRegistry().newGauge(DEGRADED_STANDBY_ACTIVE, DEGRADED_STANDBY_ACTIVE_DESC, 0L);
     currentLocalState = getMetricsRegistry().newGauge(CURRENT_LOCAL_STATE, CURRENT_LOCAL_STATE_DESC,

phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java

@@ -202,7 +202,10 @@ protected void processNoMoreRoundsLeft() throws IOException {
           LOG.info("HAGroup {} updated HA state to SYNC", logGroup);
         }
       } catch (Exception e) {
-        LOG.info("Could not update status to sync for {}", logGroup, e);
+        // Convergent races now reconcile to a no-op success (PHOENIX-7990), so a throw here is a
+        // genuine anomaly -- e.g. retry-exhausted CAS contention or a truly invalid transition.
+        // Log at WARN so persistent failure to claim SYNC after processing every file is visible.
+        LOG.warn("Could not update status to sync for {}", logGroup, e);
       }
     }
   }