You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by tk...@apache.org on 2026/08/21 22:10:23 UTC
(phoenix) branch PHOENIX-7562-feature-new updated: PHOENIX-7990 Reconcile convergent CAS race in setHAGroupStatusIfNeeded to avoid RS aborts (#2604)
This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by this push:
new ce14f19c79 PHOENIX-7990 Reconcile convergent CAS race in setHAGroupStatusIfNeeded to avoid RS aborts (#2604)
ce14f19c79 is described below
commit ce14f19c79157936e94810404320a5f6a5527e9d
Author: tkhurana <kh...@gmail.com>
AuthorDate: Fri Aug 21 15:10:17 2026 -0700
PHOENIX-7990 Reconcile convergent CAS race in setHAGroupStatusIfNeeded to avoid RS aborts (#2604)
* PHOENIX-7990 Reconcile convergent CAS race in setHAGroupStatusIfNeeded to avoid RS aborts
When >=2 co-active RegionServers drive the shared HA-status znode to the same
target state, each does an optimistic CAS whose expected version comes from the
watch-lagged local cache. The first writer wins; the others were aborting the
RegionServer even though the shared status was already at the target. This hits
the peer-degrade path (ACTIVE_IN_SYNC -> ACTIVE_NOT_IN_SYNC) and, more commonly,
the forwarders racing ACTIVE_NOT_IN_SYNC -> ACTIVE_IN_SYNC.
The two convergent losers differ by whether the winner's write reached the loser's
cache before it fired, and each is reconciled from a source that is authoritative
for its path -- never the watch-lagged cache used as a false-success signal:
Cache-stale loser (client, HAGroupStoreClient.setHAGroupStatusIfNeeded): the CAS
is attempted and loses stale; the loop re-reads fresh from ZK and retries,
bounded by SET_HA_GROUP_STATUS_MAX_ATTEMPTS. Move the existing
'attempt > 1 && current == target' no-op ahead of validateTransitionAndGetWaitTime
so a converged non-self-transitionable target (e.g. ACTIVE_IN_SYNC) returns a
no-op success from the fresh re-read instead of throwing
InvalidClusterRoleTransitionException on the X -> X self-transition. The first
attempt never short-circuits, so the periodic STORE_AND_FORWARD heartbeat still
writes its znode mtime bump that gates SYNC promotion.
Watch-won loser (server, HAGroupStoreManager.setHAGroupStatusToSync): the winner's
write already propagated to the cache, so validate rejects the ACTIVE_IN_SYNC ->
ACTIVE_IN_SYNC self-transition before any CAS is attempted -- no stale-version
exception is ever thrown, so the client cannot catch it. Swallow the
InvalidClusterRoleTransitionException as a no-op success, guarded on
isStateAlreadyUpdated(current == target) so a genuinely invalid transition still
propagates. The ACTIVE_IN_SYNC local listener in ReplicationLogGroup already drives
the loser to the correct end state (resume rotation, SYNC_AND_FORWARD -> SYNC), so
the group-level goal is met and the loser need not retry its own write.
Add HAGroupStoreManagerIT#testSetHAGroupStatusToSyncConvergentRaceIsNoOp covering
the watch-won path: with the record already at ACTIVE_IN_SYNC, setHAGroupStatusToSync
returns 0L with no ZK write instead of throwing. The existing HAGroupStoreClientIT
convergent-race and heartbeat cases continue to cover the cache-stale path.
Follow-up (review): add HAGroupStoreClientIT#
testSetHAGroupStatusIfNeededConvergentRaceNonSelfTransitionableTarget, a subcase-A
regression that converges on a non-self-transitionable target (STANDBY ->
STANDBY_TO_ACTIVE): the existing convergent test converges on ACTIVE_NOT_IN_SYNC (the
lone self-transitionable state) and so passes even without the reorder, whereas this
one throws InvalidClusterRoleTransitionException if the no-op short-circuit is not moved
ahead of validate. Also raise the ReplicationLogDiscoveryForwarder swallowed
setHAGroupStatusToSync failure from INFO to WARN: convergent races now reconcile to a
no-op success, so a throw there is a genuine anomaly (retry-exhausted contention or a
truly invalid transition) and must be visible.
* PHOENIX-7990 Address review NITs: drop stale doc ref, de-flake convergent-race IT
Replace the dangling docs/ reference in the convergent-race regression test
javadoc with the PHOENIX-7990 JIRA reference (the doc is a local, uncommitted
analysis file).
De-flake testSetHAGroupStatusIfNeededConvergentRaceNonSelfTransitionableTarget:
the prior test relied on the client's PathChildrenCache watch not delivering the
peer's STANDBY_TO_ACTIVE write before the call, a rare CI race that would make
attempt 1 read the converged non-self-transitionable target and throw
InvalidClusterRoleTransitionException. Inject the stale attempt-1 read
deterministically by overriding fetchLocalRecordAndPopulateZKIfNeeded (widened to
package-private @VisibleForTesting) to hand back the captured pre-advance
(record, version) exactly once, so the stale-CAS reconcile path is exercised
regardless of watch timing.
---
.../apache/phoenix/jdbc/HAGroupStoreClient.java | 164 ++++++++------
.../apache/phoenix/jdbc/HAGroupStoreManager.java | 15 +-
.../org/apache/phoenix/jdbc/PhoenixHAAdmin.java | 6 +-
.../metrics/HAGroupStoreMetricsSourceFactory.java | 4 +-
.../metrics/HAGroupStoreMetricsSourceImpl.java | 8 +-
.../ReplicationLogDiscoveryForwarder.java | 5 +-
.../apache/phoenix/jdbc/HAGroupStoreClientIT.java | 242 +++++++++++++++++++++
.../apache/phoenix/jdbc/HAGroupStoreManagerIT.java | 32 +++
8 files changed, 403 insertions(+), 73 deletions(-)
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
index 0e118f6a10..f80cf1174e 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
+++ b/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 class HAGroupStoreClient implements Closeable {
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;
+ }
}
/**
@@ -983,9 +1018,12 @@ public class HAGroupStoreClient implements Closeable {
/**
* Read the local record from the cache; if absent, rebuild once from the system table (the znode
- * may have been deleted) and re-read. Returns (null, null) when still absent.
+ * may have been deleted) and re-read. Returns (null, null) when still absent. Package-private and
+ * overridable only so tests can seed attempt 1 of {@link #setHAGroupStatusIfNeeded} with a stale
+ * (record, version) and exercise the stale-CAS reconcile path deterministically.
*/
- private Pair<HAGroupStoreRecord, Stat> fetchLocalRecordAndPopulateZKIfNeeded() {
+ @VisibleForTesting
+ Pair<HAGroupStoreRecord, Stat> fetchLocalRecordAndPopulateZKIfNeeded() {
if (pathChildrenCache == null) {
LOGGER.warn("LOCAL HAGroupStoreClient cache is null for HA group {}, returning null",
haGroupName);
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java
index e579d109c1..2e33c5b0c8 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java
@@ -364,7 +364,20 @@ public class HAGroupStoreManager {
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);
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdmin.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdmin.java
index 0eaa35e163..5c9396e0fd 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdmin.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdmin.java
@@ -530,8 +530,10 @@ public class PhoenixHAAdmin implements Closeable {
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,
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java
index 2d61a42fd3..ad90504680 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceFactory.java
@@ -23,8 +23,8 @@ import java.util.concurrent.ConcurrentHashMap;
* 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 {
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java
index f7d731fe76..1626fc279a 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/metrics/HAGroupStoreMetricsSourceImpl.java
@@ -58,10 +58,10 @@ public class HAGroupStoreMetricsSourceImpl extends BaseSourceImpl
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,
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
index 6c7c965b0f..097760bb8f 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
@@ -202,7 +202,10 @@ public class ReplicationLogDiscoveryForwarder extends ReplicationLogDiscovery {
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);
}
}
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreClientIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreClientIT.java
index 4ea96ed0ba..181bfd53e8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreClientIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreClientIT.java
@@ -1537,6 +1537,248 @@ public class HAGroupStoreClientIT extends HABaseIT {
assertEquals(HAGroupStoreRecord.HAGroupState.STANDBY, afterSecond.getHAGroupState());
}
+ /**
+ * A same-state write on the first attempt is an intentional refresh, not a no-op: the periodic
+ * STORE_AND_FORWARD heartbeat re-writes ACTIVE_NOT_IN_SYNC to bump the znode mtime so the
+ * standby's staleness check stays fresh (see
+ * StoreAndForwardModeImpl#startHAGroupStoreUpdateTask). It must proceed to the CAS and bump the
+ * version. The no-op short-circuit applies only on the reconcile retry path after a stale-version
+ * CAS loss (see the convergent-race test).
+ */
+ @Test
+ public void testSetHAGroupStatusIfNeededSameStateRefreshBumpsVersion() throws Exception {
+ String haGroupName = testName.getMethodName();
+
+ HAGroupStoreRecord initialRecord = new HAGroupStoreRecord("v1.0", haGroupName,
+ HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC, 0L,
+ HighAvailabilityPolicy.FAILOVER.toString(), this.peerZKUrl, this.masterUrl,
+ this.peerMasterUrl, CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L);
+ createOrUpdateHAGroupStoreRecordOnZookeeper(haAdmin, haGroupName, initialRecord);
+
+ HAGroupStoreClient haGroupStoreClient = HAGroupStoreClient
+ .getInstanceForZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration(), haGroupName, zkUrl);
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+ int versionBefore =
+ haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName).getRight().getVersion();
+
+ // Requested state == current state on the first attempt: this is the heartbeat refresh; it must
+ // write and bump the version so the mtime advances.
+ assertEquals(0L, haGroupStoreClient
+ .setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC));
+
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+ Pair<HAGroupStoreRecord, Stat> after = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName);
+ assertEquals(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC,
+ after.getLeft().getHAGroupState());
+ assertTrue("Same-state heartbeat refresh must bump the znode version",
+ after.getRight().getVersion() > versionBefore);
+ }
+
+ /**
+ * Convergent-race regression (PHOENIX-7990): a peer RS advances the shared record to
+ * ACTIVE_NOT_IN_SYNC while this client still holds the pre-bump cached version. The client's own
+ * transition to the same target must reconcile (stale CAS -> re-read fresh -> observe target
+ * already met -> no-op success) rather than aborting with
+ * StaleHAGroupStoreRecordVersionException. The external advance happens immediately before the
+ * client call so the local cache is still at the stale AIS version and the stale-CAS path is
+ * exercised. Version is asserted >= the winner's: if the watch had already caught the cache up,
+ * attempt 1 would instead do a legitimate same-state heartbeat refresh (one extra bump) — also
+ * correct; the invariant this test pins is "no abort, converges to ACTIVE_NOT_IN_SYNC".
+ */
+ @Test
+ public void testSetHAGroupStatusIfNeededConvergentRaceReconciles() throws Exception {
+ String haGroupName = testName.getMethodName();
+
+ HAGroupStoreRecord initialRecord =
+ new HAGroupStoreRecord("v1.0", haGroupName, HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC,
+ 0L, HighAvailabilityPolicy.FAILOVER.toString(), this.peerZKUrl, this.masterUrl,
+ this.peerMasterUrl, CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L);
+ createOrUpdateHAGroupStoreRecordOnZookeeper(haAdmin, haGroupName, initialRecord);
+
+ HAGroupStoreClient haGroupStoreClient = HAGroupStoreClient
+ .getInstanceForZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration(), haGroupName, zkUrl);
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+ assertEquals(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC,
+ haGroupStoreClient.getHAGroupStoreRecord().getHAGroupState());
+
+ // A peer RS wins the CAS: advance the shared znode to ACTIVE_NOT_IN_SYNC out from under this
+ // client's cache, then immediately drive this client to the same target before its watch fires.
+ Pair<HAGroupStoreRecord, Stat> current = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName);
+ haAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName,
+ current.getLeft().withHAGroupState(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC),
+ current.getRight().getVersion());
+ int winnerVersion =
+ haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName).getRight().getVersion();
+
+ // Must not throw StaleHAGroupStoreRecordVersionException; converges as a no-op success.
+ assertEquals(0L, haGroupStoreClient
+ .setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC));
+
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+ Pair<HAGroupStoreRecord, Stat> after = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName);
+ assertEquals(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC,
+ after.getLeft().getHAGroupState());
+ assertTrue(
+ "Loser must converge without clobbering: version >= winner's (equal on the stale-CAS "
+ + "reconcile path; winner+1 if the watch caught the cache up first and attempt 1 did a "
+ + "legitimate same-state refresh)",
+ after.getRight().getVersion() >= winnerVersion);
+ }
+
+ /**
+ * Subcase-A regression for a non-self-transitionable convergence target (PHOENIX-7990). Unlike
+ * {@link #testSetHAGroupStatusIfNeededConvergentRaceReconciles}, which converges on
+ * ACTIVE_NOT_IN_SYNC (the one self-transitionable state, so it reconciles even without the fix),
+ * this drives STANDBY -> STANDBY_TO_ACTIVE: an allowed, ungated transition whose target is NOT
+ * self-transitionable. A peer wins the CAS to STANDBY_TO_ACTIVE while this client's cache is
+ * still stale at STANDBY; attempt 1 loses the stale CAS, re-reads fresh (now STANDBY_TO_ACTIVE),
+ * and the no-op short-circuit — which the fix moved ahead of validateTransitionAndGetWaitTime —
+ * returns a no-op success. This is the case the reorder guards: revert it and attempt 2's
+ * validate(STANDBY_TO_ACTIVE -> STANDBY_TO_ACTIVE) throws InvalidClusterRoleTransitionException
+ * on the X -> X self-transition.
+ * <p>
+ * The stale attempt-1 read is injected deterministically by overriding
+ * fetchLocalRecordAndPopulateZKIfNeeded to hand back the captured pre-advance (record, version)
+ * exactly once, so the test does not depend on the client's watch losing a race with the peer's
+ * write (which would otherwise flake if the watch delivered the advance before the call).
+ */
+ @Test
+ public void testSetHAGroupStatusIfNeededConvergentRaceNonSelfTransitionableTarget()
+ throws Exception {
+ String haGroupName = testName.getMethodName();
+
+ HAGroupStoreRecord initialRecord =
+ new HAGroupStoreRecord("v1.0", haGroupName, HAGroupStoreRecord.HAGroupState.STANDBY, 0L,
+ HighAvailabilityPolicy.FAILOVER.toString(), this.peerZKUrl, this.masterUrl,
+ this.peerMasterUrl, CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L);
+ createOrUpdateHAGroupStoreRecordOnZookeeper(haAdmin, haGroupName, initialRecord);
+
+ // Capture the pre-advance (record, version) -- exactly what a watch-lagged cache would hold --
+ // and seed it as attempt 1's read via the override below, so the stale-CAS path is exercised
+ // deterministically regardless of whether the client's watch fires before the call.
+ Pair<HAGroupStoreRecord, Stat> stale = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName);
+ int versionBefore = stale.getRight().getVersion();
+ AtomicReference<Pair<HAGroupStoreRecord, Stat>> attempt1Seed = new AtomicReference<>(stale);
+
+ HAGroupStoreClient haGroupStoreClient = new HAGroupStoreClient(
+ CLUSTERS.getHBaseCluster1().getConfiguration(), null, haGroupName, zkUrl) {
+ @Override
+ Pair<HAGroupStoreRecord, Stat> fetchLocalRecordAndPopulateZKIfNeeded() {
+ Pair<HAGroupStoreRecord, Stat> seed = attempt1Seed.getAndSet(null);
+ return seed != null ? seed : super.fetchLocalRecordAndPopulateZKIfNeeded();
+ }
+ };
+ try {
+ // A peer RS wins the CAS: advance the shared znode to STANDBY_TO_ACTIVE, bumping the znode
+ // version by exactly one, so winnerVersion == versionBefore + 1.
+ haAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName,
+ stale.getLeft().withHAGroupState(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE),
+ versionBefore);
+
+ // Attempt 1 reads the seeded stale STANDBY@versionBefore, loses the CAS on the stale version,
+ // re-reads fresh (now STANDBY_TO_ACTIVE), and the reordered no-op short-circuit returns a
+ // no-op success rather than throwing InvalidClusterRoleTransitionException on the X -> X
+ // target.
+ assertEquals(0L, haGroupStoreClient
+ .setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE));
+
+ Pair<HAGroupStoreRecord, Stat> after = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName);
+ assertEquals(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE,
+ after.getLeft().getHAGroupState());
+ assertEquals("Convergent-race no-op must not write to ZK: only the winner's write lands",
+ versionBefore + 1, after.getRight().getVersion());
+ } finally {
+ haGroupStoreClient.close();
+ }
+ }
+
+ /**
+ * Multi-RegionServer STORE_AND_FORWARD heartbeat under contention. N independent clients — each
+ * with its own watch-lagged cache, exactly like N co-active RegionServers — re-write
+ * ACTIVE_NOT_IN_SYNC every cycle to keep the znode mtime fresh for the standby's staleness check
+ * (see StoreAndForwardModeImpl#startHAGroupStoreUpdateTask). Verifies the convergent-race fix
+ * preserves the heartbeat guarantee: each cycle the shared znode's version and mtime advance (one
+ * RS wins the CAS and bumps; the losers hit StaleHAGroupStoreRecordVersionException, re-read
+ * fresh, and reconcile to a no-op success) and no RS ever aborts. The per-cycle propagation pause
+ * mirrors the production cadence (heartbeat interval >> watch propagation): every RS enters the
+ * cycle with a fresh-enough version for its attempt-1 same-state write to land, so exactly one
+ * write per cycle succeeds and bumps.
+ */
+ @Test
+ public void testConcurrentStoreAndForwardHeartbeatBumpsVersionAndMtimeEachCycle()
+ throws Exception {
+ String haGroupName = testName.getMethodName();
+ final int numRegionServers = 4;
+ final int numCycles = 5;
+
+ HAGroupStoreRecord initialRecord = new HAGroupStoreRecord("v1.0", haGroupName,
+ HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC, 0L,
+ HighAvailabilityPolicy.FAILOVER.toString(), this.peerZKUrl, this.masterUrl,
+ this.peerMasterUrl, CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L);
+ createOrUpdateHAGroupStoreRecordOnZookeeper(haAdmin, haGroupName, initialRecord);
+
+ // Distinct instances (not the zkUrl-keyed singleton) so each has its own lagging cache, like
+ // separate RegionServers.
+ List<HAGroupStoreClient> clients = new ArrayList<>();
+ ExecutorService pool = Executors.newFixedThreadPool(numRegionServers);
+ try {
+ for (int i = 0; i < numRegionServers; i++) {
+ clients.add(new HAGroupStoreClient(CLUSTERS.getHBaseCluster1().getConfiguration(), null,
+ haGroupName, zkUrl));
+ }
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+
+ Stat before = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName).getRight();
+ int prevVersion = before.getVersion();
+ long prevMtime = before.getMtime();
+
+ for (int cycle = 0; cycle < numCycles; cycle++) {
+ ConcurrentLinkedQueue<Throwable> failures = new ConcurrentLinkedQueue<>();
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(numRegionServers);
+ for (HAGroupStoreClient client : clients) {
+ pool.submit(() -> {
+ try {
+ start.await();
+ client.setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC);
+ } catch (Throwable t) {
+ failures.add(t);
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ start.countDown();
+ assertTrue("Cycle " + cycle + ": heartbeat writers timed out",
+ done.await(30, TimeUnit.SECONDS));
+ assertTrue(
+ "Cycle " + cycle + ": no RegionServer may abort on the heartbeat CAS; the "
+ + "convergent race must reconcile to a no-op success, but saw " + failures,
+ failures.isEmpty());
+
+ // Let the winning CAS propagate to every client's cache before the next cycle, mirroring
+ // the production heartbeat cadence (interval >> watch propagation).
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+
+ Stat after = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName).getRight();
+ assertTrue("Cycle " + cycle + ": znode version must advance (a heartbeat write landed)",
+ after.getVersion() > prevVersion);
+ assertTrue("Cycle " + cycle + ": znode mtime must advance to stay fresh for the standby",
+ after.getMtime() > prevMtime);
+ assertEquals("Cycle " + cycle + ": state must remain ACTIVE_NOT_IN_SYNC",
+ HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC,
+ haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName).getLeft().getHAGroupState());
+ prevVersion = after.getVersion();
+ prevMtime = after.getMtime();
+ }
+ } finally {
+ pool.shutdownNow();
+ for (HAGroupStoreClient client : clients) {
+ client.close();
+ }
+ }
+ }
+
/**
* Regression test for the startCache INITIALIZED-latch contract: a LOCAL cache listener that
* throws while handling the INITIALIZED event must not strand startup. startCache releases its
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreManagerIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreManagerIT.java
index 912ac32ce4..af70a3b37e 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreManagerIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreManagerIT.java
@@ -448,6 +448,38 @@ public class HAGroupStoreManagerIT extends HABaseIT {
assertEquals(anisRecord.getLastSyncStateTimeInMs(), updatedRecord.getLastSyncStateTimeInMs());
}
+ /**
+ * Convergent sync race (the watch-won case): a losing co-active RS forwarder drives
+ * ACTIVE_NOT_IN_SYNC -> ACTIVE_IN_SYNC, but the winner's write has already propagated to this
+ * RS's cache before it fires. setHAGroupStatusIfNeeded then rejects the ACTIVE_IN_SYNC ->
+ * ACTIVE_IN_SYNC self-transition with InvalidClusterRoleTransitionException (ACTIVE_IN_SYNC is
+ * not self-transitionable). Since the group is already at the sync target, setHAGroupStatusToSync
+ * must swallow that as a no-op success (return 0L, no write) rather than propagate a failure.
+ */
+ @Test
+ public void testSetHAGroupStatusToSyncConvergentRaceIsNoOp() throws Exception {
+ String haGroupName = testName.getMethodName();
+ HAGroupStoreManager haGroupStoreManager = HAGroupStoreManager.getInstance(conf1);
+
+ // @Before seeds the record at ACTIVE_IN_SYNC; let the cache catch up so the target is already
+ // met, mirroring a losing forwarder whose watch delivered the winner's write first.
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+ HAGroupStoreRecord before = haGroupStoreManager.getHAGroupStoreRecord(haGroupName).orElse(null);
+ assertNotNull(before);
+ assertEquals(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, before.getHAGroupState());
+ int versionBefore =
+ haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName).getRight().getVersion();
+
+ // Must not throw InvalidClusterRoleTransitionException; converges as a no-op success.
+ assertEquals(0L, haGroupStoreManager.setHAGroupStatusToSync(haGroupName));
+
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+ Pair<HAGroupStoreRecord, Stat> after = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName);
+ assertEquals(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, after.getLeft().getHAGroupState());
+ assertEquals("Convergent-race no-op must not write to ZK", versionBefore,
+ after.getRight().getVersion());
+ }
+
@Test
public void testGetHAGroupNamesFiltersCorrectlyByZkUrl() throws Exception {
HAGroupStoreManager haGroupStoreManager = HAGroupStoreManager.getInstance(conf1);