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 <[email protected]>
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);
