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 0e118f6a105..9771c0b777b 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 long setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState haGroupStat if (!isHealthy) { throw new IOException("HAGroupStoreClient is not healthy"); } - Pair 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 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; + } } /** 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 e579d109c1d..2e33c5b0c8e 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 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); } 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 0eaa35e163c..5c9396e0fd9 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 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, 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 2d61a42fd3f..ad905046807 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 @@ * Factory for process-lifetime HAGroupStore metric sources. *

* 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 f7d731fe763..1626fc279a8 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 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, 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 6c7c965b0f3..097760bb8f3 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 @@ 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); } } } 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 4ea96ed0ba6..c530a4eda54 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,234 @@ public void testSetHAGroupStatusIfNeededMultipleTransitions() throws Exception { 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 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 (docs/HA_Status_CAS_Stale_Cache_BadVersion.md): 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 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 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. + */ + @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); + + HAGroupStoreClient haGroupStoreClient = HAGroupStoreClient + .getInstanceForZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration(), haGroupName, zkUrl); + Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS); + assertEquals(HAGroupStoreRecord.HAGroupState.STANDBY, + haGroupStoreClient.getHAGroupStoreRecord().getHAGroupState()); + + // A peer RS wins the CAS: advance the shared znode to STANDBY_TO_ACTIVE out from under this + // client's cache. This is the LAST ZK op before the client call (no read-back) so the client's + // watch has not fired and its cache is still stale at STANDBY -- the stale-CAS path. The + // winner's + // setData bumps the znode version by exactly one, so winnerVersion == versionBefore + 1. + Pair current = haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName); + int versionBefore = current.getRight().getVersion(); + haAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName, + current.getLeft().withHAGroupState(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE), + versionBefore); + + // Must not throw InvalidClusterRoleTransitionException on the converged X -> X target; the + // stale-CAS loser reconciles to a no-op success. + assertEquals(0L, haGroupStoreClient + .setHAGroupStatusIfNeeded(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE)); + + Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS); + Pair 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()); + } + + /** + * 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 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 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 912ac32ce49..af70a3b37ef 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 void testSetHAGroupStatusToSync() throws Exception { 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 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);