From 8be8b8e81f59a75bfe1f3acf7ddeb8ffd4c2e3e3 Mon Sep 17 00:00:00 2001 From: Lokesh Khurana Date: Mon, 27 Jul 2026 15:49:39 -0700 Subject: [PATCH 1/5] PHOENIX-7973 HA client can adopt a stale (lower-version) ClusterRoleRecord when one endpoint lags the peer getClusterRoleRecordFromEndpoint() queried cluster 1 first and returned it immediately whenever it had no UNKNOWN role, without consulting cluster 2. CRR version propagation across RegionServers is not synchronized, so at startup or during an in-flight admin/failover transition one endpoint can momentarily serve a lower admin version (or an UNKNOWN role) than its peer. In that window the client adopted the staler, lower-version record and silently reverted to an older cluster-role view. The refresh path guards only with ClusterRoleRecord.equals() (which ignores version) and never called the existing isNewerThan() helper, so nothing detected the downgrade. Fix: always fetch the CRR from both cluster endpoints and reconcile via a new package-private static reconcileClusterRoleRecords(): prefer a record without an UNKNOWN role (a known-role record is usable for routing; an UNKNOWN one is not), and within the same category prefer the higher admin version. This is a strict superset of the previous UNKNOWN-only handling and guarantees the client never adopts a CRR older than one a peer already advertises. If the peer endpoint is unreachable, cluster 1's record is used as-is. Adds one endpoint RPC to the CRR refresh path only; CRR is fetched on connect/refresh (cached), not per query, so no meaningful perf impact. Client-side only; no API or wire-format change. Unit-tested via HighAvailabilityGroupTest#testReconcileClusterRoleRecords (higher-version-wins regression guard, order-independence, non-UNKNOWN beats UNKNOWN in both orders, UNKNOWN-vs-UNKNOWN higher-version). Co-Authored-By: Claude Opus 4.8 --- .../phoenix/jdbc/HighAvailabilityGroup.java | 87 +++++++++++++------ .../jdbc/HighAvailabilityGroupTest.java | 49 +++++++++++ 2 files changed, 111 insertions(+), 25 deletions(-) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java index 017f996a319..1b74dc2cfaa 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java @@ -979,8 +979,17 @@ private static void throwMalFormedConnectionUrlException(String message) throws } /** - * Method to get ClusterRoleRecord from RegionServer Endpoints from either of the clusters. - * @return ClusterRoleRecord from the first available cluster + * Method to get ClusterRoleRecord from RegionServer Endpoints of both clusters. + *

+ * CRR version propagation across the RegionServers of the two clusters is not synchronized, so at + * startup or during an in-flight admin/failover transition one endpoint can momentarily serve a + * lower admin version (or an UNKNOWN role) than its peer. To avoid the client adopting a staler + * or less usable record than one a peer already advertises, this always fetches the CRR from + * both endpoints and reconciles them via {@link #reconcileClusterRoleRecords}. If the + * peer (cluster 2) endpoint is unreachable, cluster 1's record is used as-is. The reconciliation + * subsumes the previous UNKNOWN-only handling. CRR is fetched only on connect/refresh (and cached + * in {@link #roleRecord}), not per query, so the extra endpoint RPC is not on a hot path. + * @return the reconciled ClusterRoleRecord * @throws SQLException if there is an error getting the ClusterRoleRecord */ private ClusterRoleRecord getClusterRoleRecordFromEndpoint() throws SQLException { @@ -992,31 +1001,22 @@ private ClusterRoleRecord getClusterRoleRecordFromEndpoint() throws SQLException // Get the CRR via RSEndpoint for cluster 1 ClusterRoleRecord roleRecord = GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), info.getUrl2(), info.getUrl1(), info.getName(), this, pollerInterval, properties); - // If we have unknown role for any cluster then try getting CRR from cluster 2 endpoint and if - // we get unknown role from there as well then CRR with higher adminVersion wins. - if (roleRecord.hasUnknownRole()) { - ClusterRoleRecord roleRecordFromPR; - try { - roleRecordFromPR = GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), - info.getUrl2(), info.getUrl2(), info.getName(), this, pollerInterval, properties); - } catch (Exception e) { - // As we were able to get CRR from cluster 1 but cluster 2 threw exception then just - // return - // CRR from cluster 1 and consume this exception - LOG.warn("Role Record from cluster {} has Unknown Role but cluster {} threw exception, " - + "returning {} as CRR", info.getUrl1(), info.getUrl2(), roleRecord.toPrettyString()); - return roleRecord; - } - if (roleRecordFromPR.hasUnknownRole()) { - return roleRecord.getVersion() > roleRecordFromPR.getVersion() - ? roleRecord - : roleRecordFromPR; - } else { - return roleRecordFromPR; - } - } else { + // Always consult cluster 2 as well and keep the more authoritative record (see method + // javadoc): a non-UNKNOWN record is preferred over an UNKNOWN one, and among records in the + // same category the higher admin version wins. If cluster 2 is unreachable, fall back to the + // cluster 1 record we already have and consume the exception. + ClusterRoleRecord roleRecordFromPR; + try { + roleRecordFromPR = GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), + info.getUrl2(), info.getUrl2(), info.getName(), this, pollerInterval, properties); + } catch (Exception e) { + LOG.warn( + "Fetched CRR {} from cluster {} but cluster {} endpoint threw an exception; " + + "returning cluster {} CRR without peer reconciliation", + roleRecord.toPrettyString(), info.getUrl1(), info.getUrl2(), info.getUrl1(), e); return roleRecord; } + return reconcileClusterRoleRecords(roleRecord, roleRecordFromPR); } catch (Exception e) { // If we get CRR Not Found on cluster 1, we should still try cluster 2, maybe // haGroupStoreClient @@ -1240,4 +1240,41 @@ static boolean shouldCountFailover(boolean transitionSucceeded, ClusterRoleRecor return transitionSucceeded && !oldRecord.getActiveUrl().equals(newRecord.getActiveUrl()) && newRecord.getActiveUrl().isPresent(); } + + /** + * Reconcile the two ClusterRoleRecords fetched from the cluster 1 and cluster 2 RegionServer + * endpoints and return the more authoritative one. CRR version propagation across RegionServers + * is not synchronized, so the two endpoints may disagree during startup or an in-flight + * transition; picking the more authoritative record prevents the client from adopting a staler or + * less usable view than one a peer already advertises. The order of preference is: + *

    + *
  1. A record without an UNKNOWN role beats a record with an UNKNOWN role. An UNKNOWN role means + * that endpoint could not resolve the cluster roles (e.g. a ZooKeeper problem) and is not usable + * for routing, so it must never win merely on version.
    2. + *
  2. Among records in the same UNKNOWN category, the higher admin {@code version} wins (the + * admin version only advances on an operator-driven CRR change, so higher is strictly + * fresher).
    3. + *
  3. On a tie (same category and version) the peer ({@code recordFromCluster2}) record is + * returned; both are equivalent for routing so the choice is arbitrary.
    4. + *
+ * This subsumes the previous UNKNOWN-only handling and is a strict superset of it: when cluster 1 + * is non-UNKNOWN and equal-or-newer it is still returned, and the UNKNOWN/UNKNOWN case still + * resolves to the higher version. Pure function of its inputs (no global state, no clock) so it + * is straightforward to unit-test. Package-private rather than private so + * {@code HighAvailabilityGroupTest} can call it directly. Neither argument may be {@code null}. + * @param recordFromCluster1 CRR fetched from the cluster 1 endpoint ({@code info.getUrl1()}) + * @param recordFromCluster2 CRR fetched from the cluster 2 endpoint ({@code info.getUrl2()}) + * @return the more authoritative of the two records + */ + static ClusterRoleRecord reconcileClusterRoleRecords(ClusterRoleRecord recordFromCluster1, + ClusterRoleRecord recordFromCluster2) { + // Prefer a usable (non-UNKNOWN) record over an UNKNOWN one regardless of version. + if (recordFromCluster1.hasUnknownRole() != recordFromCluster2.hasUnknownRole()) { + return recordFromCluster1.hasUnknownRole() ? recordFromCluster2 : recordFromCluster1; + } + // Same category: keep the higher admin version. Ties fall through to the peer record. + return recordFromCluster1.getVersion() > recordFromCluster2.getVersion() + ? recordFromCluster1 + : recordFromCluster2; + } } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java index cc58727838d..7a74ea39b72 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java @@ -185,4 +185,53 @@ public void testShouldCountFailoverGate() { + "should count as a failover", HighAvailabilityGroup.shouldCountFailover(true, bothStandby, aStandbyBActive)); } + + /** + * Verifies the two-endpoint reconciliation performed by + * {@link HighAvailabilityGroup#reconcileClusterRoleRecords} — exercised directly via the + * package-private helper rather than by driving two mini-cluster endpoints. The client fetches + * the CRR from both cluster endpoints and must keep the more authoritative record so that a + * momentarily stale endpoint (lower admin version) or an endpoint that cannot resolve roles + * (UNKNOWN) never causes the client to adopt a staler / less usable view than its peer. This pins + * down: (a) higher version wins when both are usable — the regression guard for the stale-revert + * bug; (b) it is order-independent; (c) a non-UNKNOWN record beats an UNKNOWN one regardless of + * version, in both argument orders; and (d) UNKNOWN vs UNKNOWN still resolves to the higher + * version (preserving the previous behavior). + */ + @Test + public void testReconcileClusterRoleRecords() { + String haGroupName = "testReconcileClusterRoleRecords"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + + ClusterRoleRecord v9 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, url1, + ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 9L); + ClusterRoleRecord v10 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 10L); + ClusterRoleRecord unknownV11 = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.UNKNOWN, url2, ClusterRole.STANDBY, 11L); + ClusterRoleRecord unknownV12 = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.UNKNOWN, url2, ClusterRole.UNKNOWN, 12L); + + // (a) Both usable (non-UNKNOWN): the higher admin version wins. This is the core fix — a + // stale endpoint reporting v9 must not override the peer's v10. Regression guard for the + // silent stale-revert. + assertTrue("Higher version must win when cluster 1 lags the peer", + v10 == HighAvailabilityGroup.reconcileClusterRoleRecords(v9, v10)); + + // (b) Order-independent: same result regardless of which endpoint is passed first. + assertTrue("Higher version must win regardless of argument order", + v10 == HighAvailabilityGroup.reconcileClusterRoleRecords(v10, v9)); + + // (c) A non-UNKNOWN record beats an UNKNOWN one even when the UNKNOWN record has a higher + // version — an UNKNOWN role is not usable for routing. Assert both argument orders. + assertTrue("Non-UNKNOWN must beat UNKNOWN even with a lower version (cluster 1 usable)", + v9 == HighAvailabilityGroup.reconcileClusterRoleRecords(v9, unknownV11)); + assertTrue("Non-UNKNOWN must beat UNKNOWN even with a lower version (cluster 2 usable)", + v9 == HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, v9)); + + // (d) UNKNOWN vs UNKNOWN → higher version wins (preserves the prior behavior). + assertTrue("Among two UNKNOWN records the higher version wins", + unknownV12 == HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, unknownV12)); + } } From fa2b265597f9f48bb3eb3ca6491950b1f330abfe Mon Sep 17 00:00:00 2001 From: Lokesh Khurana Date: Wed, 29 Jul 2026 09:44:39 -0700 Subject: [PATCH 2/5] PHOENIX-7973 Guard the CRR refresh path against rolling back to a lower version The refresh path applied any non-equals() ClusterRoleRecord fetched from the endpoints, with no version comparison. Because CRR propagation across a cluster's RegionServers is eventually consistent and the client picks an endpoint at (effectively) random per fetch, a lagging endpoint can momentarily serve an older admin version than the client has already applied, silently reverting the client to a stale cluster-role view. Add a shouldApplyRefreshedRecord(current, fetched) guard that keeps the current record when it is strictly newer than the fetched one (equivalently, !current.isNewerThan(fetched)). An equal admin version is intentionally still applied: the admin version only advances on an operator-driven change, so an autonomous state-machine transition changes the cluster roles while keeping the same version, and that legitimate same-version role change must still take effect. Only a strictly lower version is rejected, so a strict '>' guard is deliberately avoided. The decision is factored into a package-private static helper (mirroring shouldCountFailover / reconcileClusterRoleRecords) and unit-tested in HighAvailabilityGroupTest#testShouldApplyRefreshedRecord: (a) reject a strictly lower version, (b) apply a strictly higher version, (c) apply a same-version record with changed roles. Co-Authored-By: Claude Opus 4.8 --- .../phoenix/jdbc/HighAvailabilityGroup.java | 44 +++++++++++++++++++ .../jdbc/HighAvailabilityGroupTest.java | 43 ++++++++++++++++++ 2 files changed, 87 insertions(+) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java index 1b74dc2cfaa..b294f14fa12 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java @@ -1106,6 +1106,22 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio return true; } + // Do not roll back to a lower-version record. CRR propagation across a cluster's + // RegionServers is eventually consistent, so a lagging endpoint can momentarily serve an + // older admin version than the one already applied on this client. Since the endpoint is + // picked at (effectively) random on each fetch, adopting such a record would revert the + // client to a stale cluster-role view. The apply-or-reject decision is factored into the + // package-private static {@link #shouldApplyRefreshedRecord} so it can be unit-tested + // directly without driving a full mini-cluster refresh. + if (!shouldApplyRefreshedRecord(roleRecord, newRoleRecord)) { + LOG.warn( + "Fetched role record {} is older (V{}) than the current record (V{}) for HA group {};" + + " keeping the current record and not rolling back", + newRoleRecord, newRoleRecord.getVersion(), roleRecord.getVersion(), info); + lastClusterRoleRecordRefreshTime = System.currentTimeMillis(); + return true; + } + final ClusterRoleRecord oldRecord = roleRecord; state = State.IN_TRANSITION; LOG.info("HA group {} is in {} to set V{} record", info, state, newRoleRecord.getVersion()); @@ -1277,4 +1293,32 @@ static ClusterRoleRecord reconcileClusterRoleRecords(ClusterRoleRecord recordFro ? recordFromCluster1 : recordFromCluster2; } + + /** + * Decides whether a freshly fetched {@link ClusterRoleRecord} should replace the currently + * applied one on the refresh path. Returns {@code false} only when the current record is strictly + * newer (higher admin {@code version}) than the fetched one, i.e. the fetch would roll the client + * back to a stale view. This guards against eventual-consistency lag: CRR propagation across a + * cluster's RegionServers is not synchronized, so a lagging endpoint (the client picks one at + * effectively random per fetch) can momentarily serve an older admin version than the client has + * already applied. + *

+ * An equal version returns {@code true} (apply) by design. The admin {@code version} + * only advances on an operator-driven CRR change; an autonomous state-machine transition changes + * the cluster roles while keeping the same admin version, so a same-version record with different + * roles is a legitimate update that must still take effect. Only a strictly lower version is + * rejected — equivalently, this returns {@code !current.isNewerThan(fetched)}. + *

+ * Callers guarantee {@code current} and {@code fetched} are non-null and share the same HA group + * info (checked upstream), and that they are not {@code equals()} (the no-op case is handled + * before this). Pure function of its inputs (no global state, no clock) so the guard is + * unit-testable without driving a full mini-cluster refresh. Package-private rather than private + * so {@code HighAvailabilityGroupTest} can call it directly. + * @param current the currently applied {@link ClusterRoleRecord} (must be non-null) + * @param fetched the candidate record freshly fetched from the endpoints + * @return {@code true} to apply {@code fetched}; {@code false} to keep {@code current} + */ + static boolean shouldApplyRefreshedRecord(ClusterRoleRecord current, ClusterRoleRecord fetched) { + return !current.isNewerThan(fetched); + } } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java index 7a74ea39b72..99d9533e648 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java @@ -234,4 +234,47 @@ public void testReconcileClusterRoleRecords() { assertTrue("Among two UNKNOWN records the higher version wins", unknownV12 == HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, unknownV12)); } + + /** + * Verifies the refresh-path apply/reject decision performed by + * {@link HighAvailabilityGroup#shouldApplyRefreshedRecord} — exercised directly via the + * package-private helper rather than by driving a full mini-cluster refresh. CRR propagation + * across a cluster's RegionServers is eventually consistent and the client picks an endpoint at + * effectively random per fetch, so a lagging endpoint can momentarily return a lower admin + * version than the client has already applied; the client must not roll back to it. This pins + * down: (a) a strictly lower-version record is rejected (the regression guard for the + * stale-revert bug); (b) a strictly higher-version record is applied; and — the subtle case — (c) + * a SAME-version record with different roles is applied, because an autonomous state-machine + * transition changes roles while keeping the same admin version (only an operator/admin action + * bumps the version), so a strict {@code >} guard here would wrongly strand failover. + */ + @Test + public void testShouldApplyRefreshedRecord() { + String haGroupName = "testShouldApplyRefreshedRecord"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + + ClusterRoleRecord v9 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, url1, + ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 9L); + ClusterRoleRecord v10 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 10L); + // Same version as v10 but different roles — models an autonomous state-machine transition, + // which changes roles while keeping the admin version unchanged. + ClusterRoleRecord v10RolesChanged = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2, ClusterRole.STANDBY, 10L); + + // (a) Current is newer (v10) than the fetched record (v9): reject, do not roll back. This is + // the core fix — a lagging endpoint serving v9 must not override the applied v10. + assertFalse("Must not roll back from the applied v10 to a stale fetched v9", + HighAvailabilityGroup.shouldApplyRefreshedRecord(v10, v9)); + + // (b) Fetched record is newer (v10) than the current (v9): apply it. + assertTrue("Must apply a strictly newer fetched record", + HighAvailabilityGroup.shouldApplyRefreshedRecord(v9, v10)); + + // (c) SAME version, different roles (autonomous transition): must still apply. A strict '>' + // guard would reject this and strand the client on a stale role view — regression guard. + assertTrue("Must apply a same-version record with changed roles (autonomous transition)", + HighAvailabilityGroup.shouldApplyRefreshedRecord(v10, v10RolesChanged)); + } } From 45eceddd1093347cf8f94e103f48be8eb0cfa7a0 Mon Sep 17 00:00:00 2001 From: lokiore Date: Tue, 4 Aug 2026 11:45:07 -0700 Subject: [PATCH 3/5] PHOENIX-7973 :- Trim verbose Javadoc and comments on CRR reconcile helpers Condense the method Javadocs, inline comments, and test Javadocs added for the two-endpoint reconciliation and refresh guard down to the non-obvious contract (UNKNOWN-not-usable-for-routing, higher-version-wins, same-version still applied for autonomous transitions, package-private-for-test). No behavior change; comments only. Generated-by: Claude Code (Opus 4.8) Co-authored-by: Claude Opus 4.8 (1M context) --- .../phoenix/jdbc/HighAvailabilityGroup.java | 77 ++++--------------- .../jdbc/HighAvailabilityGroupTest.java | 50 ++---------- 2 files changed, 25 insertions(+), 102 deletions(-) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java index b294f14fa12..9ad1a47d821 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java @@ -979,16 +979,10 @@ private static void throwMalFormedConnectionUrlException(String message) throws } /** - * Method to get ClusterRoleRecord from RegionServer Endpoints of both clusters. - *

- * CRR version propagation across the RegionServers of the two clusters is not synchronized, so at - * startup or during an in-flight admin/failover transition one endpoint can momentarily serve a - * lower admin version (or an UNKNOWN role) than its peer. To avoid the client adopting a staler - * or less usable record than one a peer already advertises, this always fetches the CRR from - * both endpoints and reconciles them via {@link #reconcileClusterRoleRecords}. If the - * peer (cluster 2) endpoint is unreachable, cluster 1's record is used as-is. The reconciliation - * subsumes the previous UNKNOWN-only handling. CRR is fetched only on connect/refresh (and cached - * in {@link #roleRecord}), not per query, so the extra endpoint RPC is not on a hot path. + * Fetches the CRR from both cluster endpoints and returns the more authoritative one (see + * {@link #reconcileClusterRoleRecords}). Consulting both avoids adopting a staler view when one + * endpoint momentarily lags its peer. If the peer (cluster 2) is unreachable, cluster 1's record + * is used as-is. * @return the reconciled ClusterRoleRecord * @throws SQLException if there is an error getting the ClusterRoleRecord */ @@ -1001,10 +995,7 @@ private ClusterRoleRecord getClusterRoleRecordFromEndpoint() throws SQLException // Get the CRR via RSEndpoint for cluster 1 ClusterRoleRecord roleRecord = GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), info.getUrl2(), info.getUrl1(), info.getName(), this, pollerInterval, properties); - // Always consult cluster 2 as well and keep the more authoritative record (see method - // javadoc): a non-UNKNOWN record is preferred over an UNKNOWN one, and among records in the - // same category the higher admin version wins. If cluster 2 is unreachable, fall back to the - // cluster 1 record we already have and consume the exception. + // Reconcile with cluster 2; if it is unreachable, keep cluster 1's record. ClusterRoleRecord roleRecordFromPR; try { roleRecordFromPR = GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), @@ -1106,13 +1097,7 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio return true; } - // Do not roll back to a lower-version record. CRR propagation across a cluster's - // RegionServers is eventually consistent, so a lagging endpoint can momentarily serve an - // older admin version than the one already applied on this client. Since the endpoint is - // picked at (effectively) random on each fetch, adopting such a record would revert the - // client to a stale cluster-role view. The apply-or-reject decision is factored into the - // package-private static {@link #shouldApplyRefreshedRecord} so it can be unit-tested - // directly without driving a full mini-cluster refresh. + // A lagging endpoint can serve an older admin version; do not roll back to it. if (!shouldApplyRefreshedRecord(roleRecord, newRoleRecord)) { LOG.warn( "Fetched role record {} is older (V{}) than the current record (V{}) for HA group {};" @@ -1258,26 +1243,11 @@ static boolean shouldCountFailover(boolean transitionSucceeded, ClusterRoleRecor } /** - * Reconcile the two ClusterRoleRecords fetched from the cluster 1 and cluster 2 RegionServer - * endpoints and return the more authoritative one. CRR version propagation across RegionServers - * is not synchronized, so the two endpoints may disagree during startup or an in-flight - * transition; picking the more authoritative record prevents the client from adopting a staler or - * less usable view than one a peer already advertises. The order of preference is: - *

    - *
  1. A record without an UNKNOWN role beats a record with an UNKNOWN role. An UNKNOWN role means - * that endpoint could not resolve the cluster roles (e.g. a ZooKeeper problem) and is not usable - * for routing, so it must never win merely on version.
    2. - *
  2. Among records in the same UNKNOWN category, the higher admin {@code version} wins (the - * admin version only advances on an operator-driven CRR change, so higher is strictly - * fresher).
    3. - *
  3. On a tie (same category and version) the peer ({@code recordFromCluster2}) record is - * returned; both are equivalent for routing so the choice is arbitrary.
    4. - *
- * This subsumes the previous UNKNOWN-only handling and is a strict superset of it: when cluster 1 - * is non-UNKNOWN and equal-or-newer it is still returned, and the UNKNOWN/UNKNOWN case still - * resolves to the higher version. Pure function of its inputs (no global state, no clock) so it - * is straightforward to unit-test. Package-private rather than private so - * {@code HighAvailabilityGroupTest} can call it directly. Neither argument may be {@code null}. + * Returns the more authoritative of the two records fetched from the cluster 1 and cluster 2 + * endpoints. Preference: a non-UNKNOWN record beats an UNKNOWN one (an UNKNOWN role can't resolve + * cluster roles and isn't usable for routing, so it never wins on version alone); otherwise the + * higher admin {@code version} wins; a tie returns the peer ({@code recordFromCluster2}). + * Package-private for direct unit testing. Neither argument may be {@code null}. * @param recordFromCluster1 CRR fetched from the cluster 1 endpoint ({@code info.getUrl1()}) * @param recordFromCluster2 CRR fetched from the cluster 2 endpoint ({@code info.getUrl2()}) * @return the more authoritative of the two records @@ -1295,25 +1265,12 @@ static ClusterRoleRecord reconcileClusterRoleRecords(ClusterRoleRecord recordFro } /** - * Decides whether a freshly fetched {@link ClusterRoleRecord} should replace the currently - * applied one on the refresh path. Returns {@code false} only when the current record is strictly - * newer (higher admin {@code version}) than the fetched one, i.e. the fetch would roll the client - * back to a stale view. This guards against eventual-consistency lag: CRR propagation across a - * cluster's RegionServers is not synchronized, so a lagging endpoint (the client picks one at - * effectively random per fetch) can momentarily serve an older admin version than the client has - * already applied. - *

- * An equal version returns {@code true} (apply) by design. The admin {@code version} - * only advances on an operator-driven CRR change; an autonomous state-machine transition changes - * the cluster roles while keeping the same admin version, so a same-version record with different - * roles is a legitimate update that must still take effect. Only a strictly lower version is - * rejected — equivalently, this returns {@code !current.isNewerThan(fetched)}. - *

- * Callers guarantee {@code current} and {@code fetched} are non-null and share the same HA group - * info (checked upstream), and that they are not {@code equals()} (the no-op case is handled - * before this). Pure function of its inputs (no global state, no clock) so the guard is - * unit-testable without driving a full mini-cluster refresh. Package-private rather than private - * so {@code HighAvailabilityGroupTest} can call it directly. + * Whether a freshly fetched record should replace the applied one on the refresh path. Rejects + * only a strictly lower version (a rollback to a stale view from a lagging endpoint). An + * equal version is still applied by design: an autonomous transition changes roles while + * keeping the same admin version, so a same-version record with changed roles is a legitimate + * update. Equivalent to {@code !current.isNewerThan(fetched)}. Package-private for direct unit + * testing. * @param current the currently applied {@link ClusterRoleRecord} (must be non-null) * @param fetched the candidate record freshly fetched from the endpoints * @return {@code true} to apply {@code fetched}; {@code false} to keep {@code current} diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java index 99d9533e648..148c3489e1f 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java @@ -186,18 +186,7 @@ public void testShouldCountFailoverGate() { HighAvailabilityGroup.shouldCountFailover(true, bothStandby, aStandbyBActive)); } - /** - * Verifies the two-endpoint reconciliation performed by - * {@link HighAvailabilityGroup#reconcileClusterRoleRecords} — exercised directly via the - * package-private helper rather than by driving two mini-cluster endpoints. The client fetches - * the CRR from both cluster endpoints and must keep the more authoritative record so that a - * momentarily stale endpoint (lower admin version) or an endpoint that cannot resolve roles - * (UNKNOWN) never causes the client to adopt a staler / less usable view than its peer. This pins - * down: (a) higher version wins when both are usable — the regression guard for the stale-revert - * bug; (b) it is order-independent; (c) a non-UNKNOWN record beats an UNKNOWN one regardless of - * version, in both argument orders; and (d) UNKNOWN vs UNKNOWN still resolves to the higher - * version (preserving the previous behavior). - */ + /** Reconciliation prefers non-UNKNOWN over UNKNOWN, then higher version; order-independent. */ @Test public void testReconcileClusterRoleRecords() { String haGroupName = "testReconcileClusterRoleRecords"; @@ -213,41 +202,24 @@ public void testReconcileClusterRoleRecords() { ClusterRoleRecord unknownV12 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.UNKNOWN, url2, ClusterRole.UNKNOWN, 12L); - // (a) Both usable (non-UNKNOWN): the higher admin version wins. This is the core fix — a - // stale endpoint reporting v9 must not override the peer's v10. Regression guard for the - // silent stale-revert. + // Both usable: higher version wins (the stale-revert regression guard), either order. assertTrue("Higher version must win when cluster 1 lags the peer", v10 == HighAvailabilityGroup.reconcileClusterRoleRecords(v9, v10)); - - // (b) Order-independent: same result regardless of which endpoint is passed first. assertTrue("Higher version must win regardless of argument order", v10 == HighAvailabilityGroup.reconcileClusterRoleRecords(v10, v9)); - // (c) A non-UNKNOWN record beats an UNKNOWN one even when the UNKNOWN record has a higher - // version — an UNKNOWN role is not usable for routing. Assert both argument orders. + // Non-UNKNOWN beats UNKNOWN even at a lower version, either order. assertTrue("Non-UNKNOWN must beat UNKNOWN even with a lower version (cluster 1 usable)", v9 == HighAvailabilityGroup.reconcileClusterRoleRecords(v9, unknownV11)); assertTrue("Non-UNKNOWN must beat UNKNOWN even with a lower version (cluster 2 usable)", v9 == HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, v9)); - // (d) UNKNOWN vs UNKNOWN → higher version wins (preserves the prior behavior). + // UNKNOWN vs UNKNOWN: higher version wins. assertTrue("Among two UNKNOWN records the higher version wins", unknownV12 == HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, unknownV12)); } - /** - * Verifies the refresh-path apply/reject decision performed by - * {@link HighAvailabilityGroup#shouldApplyRefreshedRecord} — exercised directly via the - * package-private helper rather than by driving a full mini-cluster refresh. CRR propagation - * across a cluster's RegionServers is eventually consistent and the client picks an endpoint at - * effectively random per fetch, so a lagging endpoint can momentarily return a lower admin - * version than the client has already applied; the client must not roll back to it. This pins - * down: (a) a strictly lower-version record is rejected (the regression guard for the - * stale-revert bug); (b) a strictly higher-version record is applied; and — the subtle case — (c) - * a SAME-version record with different roles is applied, because an autonomous state-machine - * transition changes roles while keeping the same admin version (only an operator/admin action - * bumps the version), so a strict {@code >} guard here would wrongly strand failover. - */ + /** Refresh guard rejects a lower version but applies a same-version role change. */ @Test public void testShouldApplyRefreshedRecord() { String haGroupName = "testShouldApplyRefreshedRecord"; @@ -258,22 +230,16 @@ public void testShouldApplyRefreshedRecord() { ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 9L); ClusterRoleRecord v10 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 10L); - // Same version as v10 but different roles — models an autonomous state-machine transition, - // which changes roles while keeping the admin version unchanged. + // Same version as v10 but different roles — models an autonomous transition (roles change, + // admin version does not). ClusterRoleRecord v10RolesChanged = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2, ClusterRole.STANDBY, 10L); - // (a) Current is newer (v10) than the fetched record (v9): reject, do not roll back. This is - // the core fix — a lagging endpoint serving v9 must not override the applied v10. assertFalse("Must not roll back from the applied v10 to a stale fetched v9", HighAvailabilityGroup.shouldApplyRefreshedRecord(v10, v9)); - - // (b) Fetched record is newer (v10) than the current (v9): apply it. assertTrue("Must apply a strictly newer fetched record", HighAvailabilityGroup.shouldApplyRefreshedRecord(v9, v10)); - - // (c) SAME version, different roles (autonomous transition): must still apply. A strict '>' - // guard would reject this and strand the client on a stale role view — regression guard. + // A strict '>' guard would wrongly reject this same-version role change. assertTrue("Must apply a same-version record with changed roles (autonomous transition)", HighAvailabilityGroup.shouldApplyRefreshedRecord(v10, v10RolesChanged)); } From 9fb7870cfa9b3915117bee64e89adb8453a7c126 Mon Sep 17 00:00:00 2001 From: lokiore Date: Mon, 17 Aug 2026 16:50:51 -0700 Subject: [PATCH 4/5] PHOENIX-7973 :- Reconcile CRR against currently-applied record and hoist poller scheduling out of the raw fetch Two related hardening changes to the two-endpoint CRR reconciliation path: - Reconcile now takes the currently-applied record so an equal-version divergence (both endpoints at the same admin version but with different roles, one endpoint lagging) defers to the applied record instead of arbitrarily adopting the peer and flapping. The defer is scoped to current's version so a genuine version advance is never suppressed. A strictly-newer UNKNOWN-tagged record that still names an active cluster now wins over a stale fully-known record (mid-transition state advance); a newer UNKNOWN record with no active role stays masked and the non-active poller resolves it. - Poller scheduling is hoisted out of the raw endpoint fetch. getClusterRoleRecord is now a pure read; the non-active poller is scheduled at most once, after reconciliation, via maybeSchedulePoller on the resolved record. Scheduling off a single raw fetch churned a poller for an ACTIVE-plus-stale-peer state and widened a pollerLock/write-lock AB-BA inversion into a reachable deadlock; the winner tick now refreshes OUTSIDE pollerLock. - The cluster-2 fetch failure path catches broadly (an unchecked exception from the pre-RPC connect path is still a reachability failure) and restores the thread interrupt status when the failure wrapped an interruption. Unit-tested in HighAvailabilityGroupTest: reconcile UNKNOWN/version precedence incl. the strict newer-active-unknown boundary, and equal-version divergence deferral/convergence. Generated-by: Claude Code (Opus 4.8) Co-authored-by: Claude Opus 4.8 (1M context) --- .../phoenix/jdbc/HighAvailabilityGroup.java | 150 ++++++++++++++---- .../util/GetClusterRoleRecordUtil.java | 72 +++++---- .../apache/phoenix/jdbc/HAGroupMetricsIT.java | 5 +- .../jdbc/HighAvailabilityGroupTest.java | 103 ++++++++++-- 4 files changed, 252 insertions(+), 78 deletions(-) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java index 9ad1a47d821..2a52c92237d 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java @@ -24,6 +24,7 @@ import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR; import java.io.IOException; +import java.io.InterruptedIOException; import java.sql.Connection; import java.sql.Driver; import java.sql.DriverManager; @@ -979,11 +980,17 @@ private static void throwMalFormedConnectionUrlException(String message) throws } /** - * Fetches the CRR from both cluster endpoints and returns the more authoritative one (see + * Reads the CRR from both cluster endpoints and returns the more authoritative one (see * {@link #reconcileClusterRoleRecords}). Consulting both avoids adopting a staler view when one * endpoint momentarily lags its peer. If the peer (cluster 2) is unreachable, cluster 1's record - * is used as-is. - * @return the reconciled ClusterRoleRecord + * is used as-is; if cluster 1 fails, cluster 2's record is used as-is. + *

+ * Endpoints are read via {@link GetClusterRoleRecordUtil#getClusterRoleRecord} (a pure read); the + * non-active poller is scheduled at most once here, after reconciliation, via + * {@link GetClusterRoleRecordUtil#maybeSchedulePoller} on the resolved record — never off a + * single raw fetch (which churned a poller and widened a deadlock; see that method). + * @return the reconciled ClusterRoleRecord, or — on a fallback path where only one endpoint was + * reachable — that single endpoint's un-reconciled record * @throws SQLException if there is an error getting the ClusterRoleRecord */ private ClusterRoleRecord getClusterRoleRecordFromEndpoint() throws SQLException { @@ -991,44 +998,78 @@ private ClusterRoleRecord getClusterRoleRecordFromEndpoint() throws SQLException Long.parseLong(properties.getProperty(PHOENIX_HA_CRR_POLLER_INTERVAL_MS_KEY, config .get(PHOENIX_HA_CRR_POLLER_INTERVAL_MS_KEY, PHOENIX_HA_CRR_POLLER_INTERVAL_MS_DEFAULT))); + ClusterRoleRecord resolvedRecord; try { - // Get the CRR via RSEndpoint for cluster 1 - ClusterRoleRecord roleRecord = GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), - info.getUrl2(), info.getUrl1(), info.getName(), this, pollerInterval, properties); - // Reconcile with cluster 2; if it is unreachable, keep cluster 1's record. - ClusterRoleRecord roleRecordFromPR; + // Read cluster 1's CRR (read-only; no poller side effect). + ClusterRoleRecord roleRecord = GetClusterRoleRecordUtil.getClusterRoleRecord(info.getUrl1(), + info.getName(), true, properties); + // Read cluster 2's CRR and reconcile; if cluster 2 is unreachable, keep cluster 1's record. try { - roleRecordFromPR = GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), - info.getUrl2(), info.getUrl2(), info.getName(), this, pollerInterval, properties); + ClusterRoleRecord roleRecordFromPR = GetClusterRoleRecordUtil + .getClusterRoleRecord(info.getUrl2(), info.getName(), true, properties); + // Pass the currently applied record (this.roleRecord; null on first load) so an + // equal-version divergence between the endpoints defers to it rather than flapping. + resolvedRecord = reconcileClusterRoleRecords(roleRecord, roleRecordFromPR, this.roleRecord); + if (!roleRecord.equals(roleRecordFromPR)) { + LOG.info( + "Reconciled divergent CRRs for HA group {}: cluster1={} (V{}), cluster2={} (V{}); " + + "chose {} (V{})", + info.getName(), roleRecord, roleRecord.getVersion(), roleRecordFromPR, + roleRecordFromPR.getVersion(), resolvedRecord, resolvedRecord.getVersion()); + } } catch (Exception e) { + // Any cluster 2 fetch failure degrades to cluster 1's record. Catch broadly: an unchecked + // exception from the pre-RPC connect path is still a reachability failure, and letting it + // reach the outer catch would discard the cluster 1 record we already hold. Restore the + // interrupt status if the failure wrapped one, so callers can observe cancellation. + if (isCausedByInterrupt(e)) { + Thread.currentThread().interrupt(); + } LOG.warn( "Fetched CRR {} from cluster {} but cluster {} endpoint threw an exception; " + "returning cluster {} CRR without peer reconciliation", roleRecord.toPrettyString(), info.getUrl1(), info.getUrl2(), info.getUrl1(), e); - return roleRecord; + resolvedRecord = roleRecord; } - return reconcileClusterRoleRecords(roleRecord, roleRecordFromPR); } catch (Exception e) { - // If we get CRR Not Found on cluster 1, we should still try cluster 2, maybe - // haGroupStoreClient - // was not initialized somehow, but if we get any exception from cluster 2 too then we should - // throw CRR not found so that fallback can happen. + // Cluster 1 failed: fall back to cluster 2. On CRR-Not-Found, if cluster 2 also fails + // rethrow the original Not-Found so downstream fallback can trigger. if ( e instanceof SQLException && ((SQLException) e).getErrorCode() == SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getErrorCode() ) { try { - return GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), info.getUrl2(), - info.getUrl2(), info.getName(), this, pollerInterval, properties); + resolvedRecord = GetClusterRoleRecordUtil.getClusterRoleRecord(info.getUrl2(), + info.getName(), true, properties); } catch (Exception ignoredEx) { throw (SQLException) e; } + } else { + // If caught exception is not CRR not found, then just try cluster 2 endpoint. + resolvedRecord = GetClusterRoleRecordUtil.getClusterRoleRecord(info.getUrl2(), + info.getName(), true, properties); } + } - // If caught exception is not CRR not found, then just try cluster 2 endpoint. - return GetClusterRoleRecordUtil.fetchClusterRoleRecord(info.getUrl1(), info.getUrl2(), - info.getUrl2(), info.getName(), this, pollerInterval, properties); + // Schedule the non-active CRR poller at most once, gated on the resolved record. maybeSchedule + // is a no-op when the record has an active role, so this is safe to call unconditionally. + GetClusterRoleRecordUtil.maybeSchedulePoller(info.getUrl1(), info.getUrl2(), info.getName(), + this, resolvedRecord, pollerInterval, properties); + return resolvedRecord; + } + + /** + * True if {@code t}'s cause chain (bounded against cyclic causes) carries an interruption marker. + * A blocking endpoint RPC surfaces the interruption wrapped inside the thrown exception, and the + * JVM has already cleared the thread's interrupt flag, so the caller must restore it explicitly. + */ + private static boolean isCausedByInterrupt(Throwable t) { + for (int depth = 0; t != null && depth < 16; t = t.getCause(), depth++) { + if (t instanceof InterruptedException || t instanceof InterruptedIOException) { + return true; + } } + return false; } /** @@ -1245,23 +1286,70 @@ static boolean shouldCountFailover(boolean transitionSucceeded, ClusterRoleRecor /** * Returns the more authoritative of the two records fetched from the cluster 1 and cluster 2 * endpoints. Preference: a non-UNKNOWN record beats an UNKNOWN one (an UNKNOWN role can't resolve - * cluster roles and isn't usable for routing, so it never wins on version alone); otherwise the - * higher admin {@code version} wins; a tie returns the peer ({@code recordFromCluster2}). - * Package-private for direct unit testing. Neither argument may be {@code null}. + * cluster roles and isn't usable for routing, so it does not win on version alone) UNLESS the + * UNKNOWN-tagged record is strictly newer and still names an active cluster (a real state advance + * where the peer role is momentarily UNKNOWN mid-transition) — masking that behind a stale + * fully-known record would keep routing to a since-demoted cluster. Otherwise the higher admin + * {@code version} wins. On an equal-version divergence (both records at the same version but with + * different roles, one endpoint lagging its peer) there is no freshness signal to order them, so + * if {@code current} is already applied at that same version the currently applied record is + * kept, deferring any transition until the endpoints converge; this avoids flapping to a stale + * peer view and back. Once both endpoints agree on the new roles the same-version record is + * returned and the autonomous transition applies as normal. A genuine version advance is never + * dropped: the defer only fires when {@code current} sits at the endpoints' version. + * Package-private for direct unit testing. The two fetched records may not be {@code null}; + * {@code current} is {@code null} on first load. * @param recordFromCluster1 CRR fetched from the cluster 1 endpoint ({@code info.getUrl1()}) * @param recordFromCluster2 CRR fetched from the cluster 2 endpoint ({@code info.getUrl2()}) + * @param current the currently applied CRR, or {@code null} on first load * @return the more authoritative of the two records */ static ClusterRoleRecord reconcileClusterRoleRecords(ClusterRoleRecord recordFromCluster1, - ClusterRoleRecord recordFromCluster2) { - // Prefer a usable (non-UNKNOWN) record over an UNKNOWN one regardless of version. + ClusterRoleRecord recordFromCluster2, ClusterRoleRecord current) { + // Exactly one record carries an UNKNOWN role. Prefer the usable (non-UNKNOWN) record so the + // client keeps a routable view -- EXCEPT when the UNKNOWN-tagged record is strictly newer AND + // still names an active cluster (one role resolved ACTIVE while its peer is mid-transition and + // momentarily UNKNOWN). There the newer record reflects a real state advance, and masking it + // behind a stale fully-known record would keep routing to a since-demoted cluster, so the newer + // record wins. A newer UNKNOWN record with NO active role stays masked: it cannot route a + // connection, and the non-active poller picks up the true state on its next tick. if (recordFromCluster1.hasUnknownRole() != recordFromCluster2.hasUnknownRole()) { - return recordFromCluster1.hasUnknownRole() ? recordFromCluster2 : recordFromCluster1; + ClusterRoleRecord unknownRecord = + recordFromCluster1.hasUnknownRole() ? recordFromCluster1 : recordFromCluster2; + ClusterRoleRecord usableRecord = + recordFromCluster1.hasUnknownRole() ? recordFromCluster2 : recordFromCluster1; + if ( + unknownRecord.getVersion() > usableRecord.getVersion() + && unknownRecord.getActiveUrl().isPresent() + ) { + return unknownRecord; + } + return usableRecord; + } + // Different versions: keep the higher admin version. + if (recordFromCluster1.getVersion() != recordFromCluster2.getVersion()) { + return recordFromCluster1.getVersion() > recordFromCluster2.getVersion() + ? recordFromCluster1 + : recordFromCluster2; + } + // Equal version: if both endpoints agree, return it (a legitimate same-version autonomous + // transition applies once both endpoints reflect the new roles). + if (recordFromCluster1.equals(recordFromCluster2)) { + return recordFromCluster2; + } + // Equal-version divergence with no freshness signal. If the currently applied record sits at + // this same version, keep it and defer the transition until the endpoints converge, rather + // than arbitrarily adopting the peer and flapping. Scoping the defer to current's version + // ensures a genuine version advance is never suppressed. + if ( + current != null && current.hasSameInfo(recordFromCluster1) + && current.getVersion() == recordFromCluster1.getVersion() + ) { + return current; } - // Same category: keep the higher admin version. Ties fall through to the peer record. - return recordFromCluster1.getVersion() > recordFromCluster2.getVersion() - ? recordFromCluster1 - : recordFromCluster2; + // First load, or current at a lower version than the (equal) endpoint version: fall back to + // the deterministic peer record so a real advance is not dropped. + return recordFromCluster2; } /** diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java index 16c7715c17b..b25d5f46f21 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java @@ -75,16 +75,19 @@ private GetClusterRoleRecordUtil() { } /** - * Method to get ClusterRoleRecord from RegionServer Endpoints. it picks a random region server - * and gets the CRR from it. + * Reads the ClusterRoleRecord from a random live region server at {@code url}. Pure read, no side + * effects: callers needing the non-active poller must reconcile both endpoints first and then + * call {@link #maybeSchedulePoller} on the result. Scheduling off a single raw fetch churned a + * poller for an ACTIVE-plus-stale-peer state and widened a {@code pollerLock}/write-lock AB-BA + * inversion into a reachable deadlock. * @param url URL to create Connection to be used to get RegionServer Endpoint Service * @param haGroupName Name of the HA group * @param doRetry Whether to retry if the operation fails * @param properties Connection properties - * @return ClusterRoleRecord from the first available cluster + * @return ClusterRoleRecord read from the endpoint at {@code url} * @throws SQLException if there is an error getting the ClusterRoleRecord */ - private static ClusterRoleRecord getClusterRoleRecord(String url, String haGroupName, + public static ClusterRoleRecord getClusterRoleRecord(String url, String haGroupName, boolean doRetry, Properties properties) throws SQLException { Connection conn = getConnection(url, properties); PhoenixConnection connection = conn.unwrap(PhoenixConnection.class); @@ -149,44 +152,35 @@ private static ClusterRoleRecord getClusterRoleRecord(String url, String haGroup } /** - * Method to schedule a poller to fetch ClusterRoleRecord every 5 seconds (or configured value) - * until we get an Active ClusterRoleRecord (one role should be Active) if we receive an Active - * roleRecord then client this method will return the roleRecord to be consumed and used, if not - * then it will start a poller and return non-active roleRecord. - *

- * The poller alternates between {@code url1} and {@code url2} on successive ticks so a transient - * outage on one cluster does not stall progress; both URLs are passed in even though the initial - * fetch only targets one of them (selected by the caller via the {@code primaryUrl} hint). - * @param url1 URL of the RegionServer Endpoint Service for cluster 1 - * @param url2 URL of the RegionServer Endpoint Service for cluster 2 - * @param primaryUrl URL to use for the initial (non-poller) fetch; must be either url1 or - * url2 - * @param haGroupName Name of the HA group - * @param haGroup HighAvailabilityGroup object to refresh the ClusterRoleRecord when an - * Active CRR is found - * @param pollerInterval Interval in milliseconds to poll for ClusterRoleRecord - * @param properties Connection properties - * @throws SQLException if there is an error getting the ClusterRoleRecord + * Starts the non-active CRR poller for {@code haGroupName} only when {@code reconciledRecord} is + * a FAILOVER record with no active role (neither cluster can take a connection); otherwise a + * no-op, so callers may invoke it unconditionally on the reconciled record. + * {@code reconciledRecord} must be the reconciled view of both endpoints — see + * {@link #getClusterRoleRecord} for why scheduling off a single raw fetch is unsafe. The poller + * alternates {@code url1}/{@code url2} each tick (so a transient outage on one cluster does not + * stall progress) until an ACTIVE CRR appears. + * @param url1 URL of the RegionServer Endpoint Service for cluster 1 + * @param url2 URL of the RegionServer Endpoint Service for cluster 2 + * @param haGroupName Name of the HA group + * @param haGroup HighAvailabilityGroup object to refresh the ClusterRoleRecord when an + * Active CRR is found + * @param reconciledRecord the reconciled CRR (across both endpoints) to gate scheduling on + * @param pollerInterval Interval in milliseconds to poll for ClusterRoleRecord + * @param properties Connection properties */ - public static ClusterRoleRecord fetchClusterRoleRecord(String url1, String url2, - String primaryUrl, String haGroupName, HighAvailabilityGroup haGroup, long pollerInterval, - Properties properties) throws SQLException { - ClusterRoleRecord clusterRoleRecord = - getClusterRoleRecord(primaryUrl, haGroupName, true, properties); + public static void maybeSchedulePoller(String url1, String url2, String haGroupName, + HighAvailabilityGroup haGroup, ClusterRoleRecord reconciledRecord, long pollerInterval, + Properties properties) { if ( - clusterRoleRecord.getPolicy() == HighAvailabilityPolicy.FAILOVER - && !clusterRoleRecord.getRole1().isActive() && !clusterRoleRecord.getRole2().isActive() + reconciledRecord.getPolicy() == HighAvailabilityPolicy.FAILOVER + && !reconciledRecord.getRole1().isActive() && !reconciledRecord.getRole2().isActive() ) { LOGGER.info( "Non-active ClusterRoleRecord found for HA group {}. Scheduling poller to check every {} ms," + " alternating between url1 and url2 until we find an ACTIVE CRR", haGroupName, pollerInterval); - // Schedule a poller to fetch ClusterRoleRecord every pollerInterval milliseconds - // until we get an Active ClusterRoleRecord and return the Non-Active CRR schedulePoller(url1, url2, haGroupName, haGroup, pollerInterval, properties); } - - return clusterRoleRecord; } /** @@ -251,16 +245,24 @@ private static void schedulePoller(String url1, String url2, String haGroupName, LOGGER.info("Active ClusterRoleRecord found for HA group {}. Cancelling poller.", haGroupName); + // Elect a single winner tick under pollerLock, but refresh OUTSIDE it: refresh grabs + // the HA-group write lock, which the connect path holds before reaching pollerLock, so + // refreshing under pollerLock would close an AB-BA deadlock. Only the tick that removed + // the future refreshes and shuts the scheduler down, so each happens exactly once. + boolean winner; + ScheduledExecutorService scheduler; synchronized (pollerLock) { ScheduledFuture future = futureMap.remove(haGroupName); if (future != null) { future.cancel(false); } + winner = future != null; + scheduler = schedulerMap.remove(haGroupName); + } + if (winner) { try { - // Refresh ClusterRoleRecord for the HAGroup with appropriate transition haGroup.refreshClusterRoleRecord(true); } finally { - ScheduledExecutorService scheduler = schedulerMap.remove(haGroupName); if (scheduler != null) { scheduler.shutdown(); } diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupMetricsIT.java index 088ccff183a..7719b8733aa 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupMetricsIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupMetricsIT.java @@ -245,8 +245,9 @@ public void testCrrCacheAgeMs() throws Exception { @Test(timeout = 300000) public void testPollerTickCount() throws Exception { - // The poller starts only when fetchClusterRoleRecord observes both roles non-active under - // FAILOVER policy. Drive that state, await a couple of ticks, and verify the counter moved. + // The poller starts only when the reconciled CRR has both roles non-active under FAILOVER + // policy (maybeSchedulePoller). Drive that state, await a couple of ticks, and verify the + // counter moved. long beforeTicks = GLOBAL_HA_POLLER_TICK_COUNT.getMetric().getValue(); long beforeFailures = GLOBAL_HA_POLLER_TICK_FAILURES.getMetric().getValue(); diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java index 148c3489e1f..b527eb3a351 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import java.sql.SQLException; @@ -203,20 +204,102 @@ public void testReconcileClusterRoleRecords() { HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.UNKNOWN, url2, ClusterRole.UNKNOWN, 12L); // Both usable: higher version wins (the stale-revert regression guard), either order. - assertTrue("Higher version must win when cluster 1 lags the peer", - v10 == HighAvailabilityGroup.reconcileClusterRoleRecords(v9, v10)); - assertTrue("Higher version must win regardless of argument order", - v10 == HighAvailabilityGroup.reconcileClusterRoleRecords(v10, v9)); + assertSame("Higher version must win when cluster 1 lags the peer", v10, + HighAvailabilityGroup.reconcileClusterRoleRecords(v9, v10, null)); + assertSame("Higher version must win regardless of argument order", v10, + HighAvailabilityGroup.reconcileClusterRoleRecords(v10, v9, null)); // Non-UNKNOWN beats UNKNOWN even at a lower version, either order. - assertTrue("Non-UNKNOWN must beat UNKNOWN even with a lower version (cluster 1 usable)", - v9 == HighAvailabilityGroup.reconcileClusterRoleRecords(v9, unknownV11)); - assertTrue("Non-UNKNOWN must beat UNKNOWN even with a lower version (cluster 2 usable)", - v9 == HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, v9)); + assertSame("Non-UNKNOWN must beat UNKNOWN even with a lower version (cluster 1 usable)", v9, + HighAvailabilityGroup.reconcileClusterRoleRecords(v9, unknownV11, null)); + assertSame("Non-UNKNOWN must beat UNKNOWN even with a lower version (cluster 2 usable)", v9, + HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, v9, null)); // UNKNOWN vs UNKNOWN: higher version wins. - assertTrue("Among two UNKNOWN records the higher version wins", - unknownV12 == HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, unknownV12)); + assertSame("Among two UNKNOWN records the higher version wins", unknownV12, + HighAvailabilityGroup.reconcileClusterRoleRecords(unknownV11, unknownV12, null)); + + // A strictly newer UNKNOWN-tagged record that STILL names an active cluster (one role ACTIVE, + // peer momentarily UNKNOWN mid-transition) must beat a stale fully-known record: masking it + // would keep routing to the since-demoted cluster. Either order. + ClusterRoleRecord activeUnknownV13 = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2, ClusterRole.UNKNOWN, 13L); + assertSame( + "Newer UNKNOWN-tagged record that still names an ACTIVE cluster must win (cluster 2)", + activeUnknownV13, + HighAvailabilityGroup.reconcileClusterRoleRecords(v9, activeUnknownV13, null)); + assertSame( + "Newer UNKNOWN-tagged record that still names an ACTIVE cluster must win (cluster 1)", + activeUnknownV13, + HighAvailabilityGroup.reconcileClusterRoleRecords(activeUnknownV13, v9, null)); + + // But a newer UNKNOWN record with NO active role stays masked behind the usable record: it + // cannot route a connection, and the non-active poller resolves the true state on its next + // tick. unknownV11 (url1 UNKNOWN, url2 STANDBY, no active) is newer than v9 but not routable. + assertSame("Newer UNKNOWN record with no active role stays masked behind the usable record", v9, + HighAvailabilityGroup.reconcileClusterRoleRecords(v9, unknownV11, null)); + + // Equal-version boundary: an active-tagged UNKNOWN record at the SAME version as a fully-known + // usable record must NOT win. The newer-UNKNOWN carve-out is strict '>', so only a strictly + // newer active-unknown record displaces the usable one; a same-version peer cannot. This pins + // the strict boundary — a '>=' mutant would wrongly route to the active-unknown record. Either + // order. + ClusterRoleRecord activeUnknownV9 = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2, ClusterRole.UNKNOWN, 9L); + assertSame("Equal-version active-unknown must NOT displace the fully-known usable record", v9, + HighAvailabilityGroup.reconcileClusterRoleRecords(v9, activeUnknownV9, null)); + assertSame("Equal-version active-unknown must NOT displace the usable record (order swapped)", + v9, HighAvailabilityGroup.reconcileClusterRoleRecords(activeUnknownV9, v9, null)); + } + + /** + * On an equal-version divergence (endpoints at the same version but different roles, one lagging) + * reconcile keeps the currently applied record when it sits at that version, deferring the + * transition until the endpoints converge; once both endpoints agree the same-version record is + * returned. A genuine version advance is never dropped. + */ + @Test + public void testReconcileEqualVersionDivergence() { + String haGroupName = "testReconcileEqualVersionDivergence"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + + // Applied record: url1 ACTIVE, url2 STANDBY at v10. + ClusterRoleRecord current = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 10L); + // Same version, diverging roles — a lagging/leading peer mid-propagation. + ClusterRoleRecord divergentV10 = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 10L); + // Both endpoints agreeing on the new same-version roles (propagation complete). + ClusterRoleRecord agreedV10a = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2, ClusterRole.STANDBY, 10L); + ClusterRoleRecord agreedV10b = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2, ClusterRole.STANDBY, 10L); + // A genuine version advance both endpoints see. + ClusterRoleRecord v11 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 11L); + + // Equal-version divergence with current at that version → keep current (no flap), either order. + assertSame("Equal-version divergence must keep the applied record (peer diverges)", current, + HighAvailabilityGroup.reconcileClusterRoleRecords(current, divergentV10, current)); + assertSame("Equal-version divergence defers regardless of which endpoint diverges", current, + HighAvailabilityGroup.reconcileClusterRoleRecords(divergentV10, current, current)); + + // Both endpoints agree on the new same-version roles → apply it (autonomous transition). The + // equal-values branch returns recordFromCluster2, and the returned record differs from current + // so the refresh path will apply it. + assertSame("Both endpoints agreeing on a same-version role change must be applied", agreedV10b, + HighAvailabilityGroup.reconcileClusterRoleRecords(agreedV10a, agreedV10b, current)); + assertFalse("Agreed same-version record must differ from current so refresh applies it", + current.equals(agreedV10b)); + + // First load (current == null) with an equal-version divergence → deterministic peer record. + assertSame("First-load equal-version divergence falls back to the peer record", divergentV10, + HighAvailabilityGroup.reconcileClusterRoleRecords(current, divergentV10, null)); + + // A genuine version advance is applied even though current is at the older version. + assertSame("A strictly newer version must still win over the applied record", v11, + HighAvailabilityGroup.reconcileClusterRoleRecords(current, v11, current)); } /** Refresh guard rejects a lower version but applies a same-version role change. */ From 6fe0dab020e551a38d4a162ef7c0227c95d4ba8d Mon Sep 17 00:00:00 2001 From: lokiore Date: Tue, 18 Aug 2026 12:52:39 -0700 Subject: [PATCH 5/5] PHOENIX-7973 :- Address review: reconcile diagnostics, interrupt handling, and endpoint/refresh test wiring Follow-up to the reviewer's minor items on the two-endpoint CRR path (no correctness/concurrency changes to the reconcile decision tree itself): - getClusterRoleRecordFromEndpoint restructured so reconciliation runs OUTSIDE both per-endpoint fetch try/catch blocks. A bug in reconcile now surfaces as itself rather than being caught by a fetch catch, misattributed as an endpoint failure, and silently degraded to a single-endpoint record. - The cluster-1 failure path now logs the cluster-1 exception (WARN) before falling back to cluster 2, so a cluster-1 failure (including an unchecked bug such as an NPE) no longer vanishes when cluster 2 succeeds. The interrupt status the fetch cleared is restored in a finally after the fallback fetch, so a stale flag cannot pre-empt the blocking work while callers still observe cancellation. - On both cluster-1 fallback paths, if cluster 2 also fails the cluster-1 failure is retained via addSuppressed instead of being dropped: the CRR-Not- Found path rethrows the original Not-Found (cluster-2 suppressed) so downstream fallback still triggers, and the non-Not-Found path rethrows cluster 2's exception with cluster 1's attached as suppressed, so the single propagating exception carries both root causes. - Corrected the reconcile carve-out comment (and its mirror in the test): a newer-UNKNOWN-with-no-active record stays masked behind the usable record and recovery comes from the next scheduled refresh (the poller runs only if the usable record is itself non-active). Tests added in HighAvailabilityGroupTest: - testIsCausedByInterrupt: both interrupt marker types, wrapped, non-interrupt, null, the depth-16 bound (within and beyond), and a cyclic cause chain. - testRefreshDoesNotRollBackToOlderRecord: wires the refresh no-rollback branch end to end (keeps the applied record, stays READY, no failover count, returns true). - testGetClusterRoleRecordFromEndpointWiring: pins url1->cluster1 / url2->cluster2 and that the applied record is threaded as current on the equal-version defer. - Equal-version divergence with the applied record at a lower version than the endpoints now covered (the non-first-load fall-through). - Endpoint fallback/rethrow branches: NOT_FOUND-on-cluster-1 falls back to a cluster-2 record; both-fail on the NOT_FOUND path rethrows the original Not-Found with the cluster-2 failure suppressed (error code preserved); both-fail on the non-Not-Found path rethrows cluster 2's exception with cluster 1's suppressed; a cluster-2 failure with cluster 1 reachable degrades to the cluster-1 record; and an interrupt-wrapped cluster-1 failure restores the thread interrupt status on the fallback path. Small testability seams added: package-private @VisibleForTesting on getClusterRoleRecordFromEndpoint, isCausedByInterrupt, a fetchClusterRoleRecord per-endpoint seam, and getStateForTesting. Generated-by: Claude Code (Opus 4.8) Co-authored-by: Claude Opus 4.8 (1M context) --- .../phoenix/jdbc/HighAvailabilityGroup.java | 134 +++++--- .../jdbc/HighAvailabilityGroupTest.java | 296 +++++++++++++++++- 2 files changed, 386 insertions(+), 44 deletions(-) diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java index 2a52c92237d..05ce2787644 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java @@ -773,6 +773,11 @@ public ClusterRoleRecord getRoleRecord() { return roleRecord; } + @VisibleForTesting + State getStateForTesting() { + return state; + } + /** * Package private close method. *

@@ -993,77 +998,121 @@ private static void throwMalFormedConnectionUrlException(String message) throws * reachable — that single endpoint's un-reconciled record * @throws SQLException if there is an error getting the ClusterRoleRecord */ - private ClusterRoleRecord getClusterRoleRecordFromEndpoint() throws SQLException { + @VisibleForTesting + ClusterRoleRecord getClusterRoleRecordFromEndpoint() throws SQLException { long pollerInterval = Long.parseLong(properties.getProperty(PHOENIX_HA_CRR_POLLER_INTERVAL_MS_KEY, config .get(PHOENIX_HA_CRR_POLLER_INTERVAL_MS_KEY, PHOENIX_HA_CRR_POLLER_INTERVAL_MS_DEFAULT))); - ClusterRoleRecord resolvedRecord; + ClusterRoleRecord resolvedRecord = null; + ClusterRoleRecord roleRecord1 = null; try { // Read cluster 1's CRR (read-only; no poller side effect). - ClusterRoleRecord roleRecord = GetClusterRoleRecordUtil.getClusterRoleRecord(info.getUrl1(), - info.getName(), true, properties); - // Read cluster 2's CRR and reconcile; if cluster 2 is unreachable, keep cluster 1's record. + roleRecord1 = fetchClusterRoleRecord(info.getUrl1()); + } catch (Exception e) { + // Cluster 1 failed: fall back to cluster 2 (single endpoint, no reconciliation). Log first, + // otherwise a cluster-1 failure (including an unchecked bug such as an NPE) vanishes whenever + // cluster 2 succeeds. + LOG.warn("Cluster 1 endpoint {} for HA group {} threw an exception; attempting cluster 2 " + + "endpoint {}", info.getUrl1(), info.getName(), info.getUrl2(), e); + // Restore the interrupt status the cluster-1 fetch cleared in a finally, AFTER the fallback + // fetch on every exit path (normal or thrown): a stale flag set before the fetch would + // pre-empt the blocking work we depend on, but callers must still observe cancellation. + boolean cluster1Interrupted = isCausedByInterrupt(e); try { - ClusterRoleRecord roleRecordFromPR = GetClusterRoleRecordUtil - .getClusterRoleRecord(info.getUrl2(), info.getName(), true, properties); - // Pass the currently applied record (this.roleRecord; null on first load) so an - // equal-version divergence between the endpoints defers to it rather than flapping. - resolvedRecord = reconcileClusterRoleRecords(roleRecord, roleRecordFromPR, this.roleRecord); - if (!roleRecord.equals(roleRecordFromPR)) { - LOG.info( - "Reconciled divergent CRRs for HA group {}: cluster1={} (V{}), cluster2={} (V{}); " - + "chose {} (V{})", - info.getName(), roleRecord, roleRecord.getVersion(), roleRecordFromPR, - roleRecordFromPR.getVersion(), resolvedRecord, resolvedRecord.getVersion()); + // On CRR-Not-Found, if cluster 2 also fails rethrow the original Not-Found (with the + // cluster-2 failure suppressed) so downstream fallback can trigger. + if ( + e instanceof SQLException && ((SQLException) e).getErrorCode() + == SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getErrorCode() + ) { + try { + resolvedRecord = fetchClusterRoleRecord(info.getUrl2()); + } catch (Exception ignoredEx) { + ((SQLException) e).addSuppressed(ignoredEx); + throw (SQLException) e; + } + } else { + // If caught exception is not CRR-Not-Found, try the cluster 2 endpoint. If cluster 2 also + // fails, attach cluster 1's failure as suppressed so the single propagating exception + // carries both root causes (parity with the Not-Found path above). + try { + resolvedRecord = fetchClusterRoleRecord(info.getUrl2()); + } catch (Exception cluster2Ex) { + cluster2Ex.addSuppressed(e); + throw cluster2Ex; + } + } + } finally { + if (cluster1Interrupted) { + Thread.currentThread().interrupt(); } + } + } + + if (roleRecord1 != null) { + // Cluster 1 reachable. Read cluster 2's CRR; if cluster 2 is unreachable, degrade to + // cluster 1's record. + ClusterRoleRecord roleRecord2 = null; + try { + roleRecord2 = fetchClusterRoleRecord(info.getUrl2()); } catch (Exception e) { // Any cluster 2 fetch failure degrades to cluster 1's record. Catch broadly: an unchecked - // exception from the pre-RPC connect path is still a reachability failure, and letting it - // reach the outer catch would discard the cluster 1 record we already hold. Restore the - // interrupt status if the failure wrapped one, so callers can observe cancellation. + // exception from the pre-RPC connect path is still a reachability failure. No further + // blocking work follows here, so restore the interrupt status immediately if the failure + // wrapped one, so callers can observe cancellation. if (isCausedByInterrupt(e)) { Thread.currentThread().interrupt(); } LOG.warn( "Fetched CRR {} from cluster {} but cluster {} endpoint threw an exception; " + "returning cluster {} CRR without peer reconciliation", - roleRecord.toPrettyString(), info.getUrl1(), info.getUrl2(), info.getUrl1(), e); - resolvedRecord = roleRecord; + roleRecord1.toPrettyString(), info.getUrl1(), info.getUrl2(), info.getUrl1(), e); } - } catch (Exception e) { - // Cluster 1 failed: fall back to cluster 2. On CRR-Not-Found, if cluster 2 also fails - // rethrow the original Not-Found so downstream fallback can trigger. - if ( - e instanceof SQLException && ((SQLException) e).getErrorCode() - == SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getErrorCode() - ) { - try { - resolvedRecord = GetClusterRoleRecordUtil.getClusterRoleRecord(info.getUrl2(), - info.getName(), true, properties); - } catch (Exception ignoredEx) { - throw (SQLException) e; - } + if (roleRecord2 == null) { + resolvedRecord = roleRecord1; } else { - // If caught exception is not CRR not found, then just try cluster 2 endpoint. - resolvedRecord = GetClusterRoleRecordUtil.getClusterRoleRecord(info.getUrl2(), - info.getName(), true, properties); + // Both endpoints reachable. Reconcile OUTSIDE both fetch try/catch blocks: it is pure + // computation, so a bug here surfaces as itself rather than being caught by a fetch catch, + // misreported as an endpoint failure, and silently degraded to a single-endpoint record. + // Pass the currently applied record (this.roleRecord; null on first load) so an + // equal-version divergence defers to it rather than flapping. + resolvedRecord = reconcileClusterRoleRecords(roleRecord1, roleRecord2, this.roleRecord); + if (!roleRecord1.equals(roleRecord2)) { + LOG.info( + "Reconciled divergent CRRs for HA group {}: cluster1={} (V{}), cluster2={} (V{}); " + + "chose {} (V{})", + info.getName(), roleRecord1, roleRecord1.getVersion(), roleRecord2, + roleRecord2.getVersion(), resolvedRecord, resolvedRecord.getVersion()); + } } } - // Schedule the non-active CRR poller at most once, gated on the resolved record. maybeSchedule - // is a no-op when the record has an active role, so this is safe to call unconditionally. + // resolvedRecord is non-null here: either the cluster-1 fallback set it, or the cluster-1 + // reachable branch did. Schedule the non-active CRR poller at most once, gated on the resolved + // record. maybeSchedulePoller is a no-op when the record has an active role, so this is safe to + // call unconditionally. GetClusterRoleRecordUtil.maybeSchedulePoller(info.getUrl1(), info.getUrl2(), info.getName(), this, resolvedRecord, pollerInterval, properties); return resolvedRecord; } + /** + * Reads a single endpoint's CRR (a pure read; no poller side effect). Extracted as a seam so unit + * tests can stub per-endpoint fetches without a mini-cluster. + */ + @VisibleForTesting + ClusterRoleRecord fetchClusterRoleRecord(String url) throws SQLException { + return GetClusterRoleRecordUtil.getClusterRoleRecord(url, info.getName(), true, properties); + } + /** * True if {@code t}'s cause chain (bounded against cyclic causes) carries an interruption marker. * A blocking endpoint RPC surfaces the interruption wrapped inside the thrown exception, and the * JVM has already cleared the thread's interrupt flag, so the caller must restore it explicitly. */ - private static boolean isCausedByInterrupt(Throwable t) { + @VisibleForTesting + static boolean isCausedByInterrupt(Throwable t) { for (int depth = 0; t != null && depth < 16; t = t.getCause(), depth++) { if (t instanceof InterruptedException || t instanceof InterruptedIOException) { return true; @@ -1311,8 +1360,9 @@ static ClusterRoleRecord reconcileClusterRoleRecords(ClusterRoleRecord recordFro // still names an active cluster (one role resolved ACTIVE while its peer is mid-transition and // momentarily UNKNOWN). There the newer record reflects a real state advance, and masking it // behind a stale fully-known record would keep routing to a since-demoted cluster, so the newer - // record wins. A newer UNKNOWN record with NO active role stays masked: it cannot route a - // connection, and the non-active poller picks up the true state on its next tick. + // record wins. A newer UNKNOWN record with NO active role stays masked behind the usable + // record: it cannot route a connection, so recovery comes from the next scheduled refresh (and + // the poller only if the usable record is itself non-active, which schedules it after return). if (recordFromCluster1.hasUnknownRole() != recordFromCluster2.hasUnknownRole()) { ClusterRoleRecord unknownRecord = recordFromCluster1.hasUnknownRole() ? recordFromCluster1 : recordFromCluster2; diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java index b527eb3a351..5bf956fa88e 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java @@ -22,12 +22,19 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import java.io.IOException; +import java.io.InterruptedIOException; import java.sql.SQLException; import java.util.Properties; import org.apache.phoenix.exception.SQLExceptionCode; import org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole; +import org.apache.phoenix.jdbc.HighAvailabilityGroup.HAGroupInfo; +import org.apache.phoenix.jdbc.HighAvailabilityGroup.State; +import org.apache.phoenix.monitoring.GlobalClientMetrics; import org.junit.Test; +import org.mockito.Mockito; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -234,8 +241,9 @@ public void testReconcileClusterRoleRecords() { HighAvailabilityGroup.reconcileClusterRoleRecords(activeUnknownV13, v9, null)); // But a newer UNKNOWN record with NO active role stays masked behind the usable record: it - // cannot route a connection, and the non-active poller resolves the true state on its next - // tick. unknownV11 (url1 UNKNOWN, url2 STANDBY, no active) is newer than v9 but not routable. + // cannot route a connection, so recovery comes from the next scheduled refresh (the poller runs + // only if the usable record is itself non-active). unknownV11 (url1 UNKNOWN, url2 STANDBY, no + // active) is newer than v9 but not routable. assertSame("Newer UNKNOWN record with no active role stays masked behind the usable record", v9, HighAvailabilityGroup.reconcileClusterRoleRecords(v9, unknownV11, null)); @@ -297,6 +305,18 @@ public void testReconcileEqualVersionDivergence() { assertSame("First-load equal-version divergence falls back to the peer record", divergentV10, HighAvailabilityGroup.reconcileClusterRoleRecords(current, divergentV10, null)); + // Equal-version divergence where the applied record is at a LOWER version than the endpoints: + // the defer guard (current.version == endpoints' version) does not fire, so this falls through + // to the deterministic peer record rather than keeping the stale applied record. This is the + // other branch of the fall-through the first-load case cannot reach (current != null). + ClusterRoleRecord olderCurrentV9 = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 9L); + assertSame( + "Equal-version divergence with the applied record at a lower version falls through to the " + + "peer record (the defer only fires when current sits at the endpoints' version)", + divergentV10, + HighAvailabilityGroup.reconcileClusterRoleRecords(current, divergentV10, olderCurrentV9)); + // A genuine version advance is applied even though current is at the older version. assertSame("A strictly newer version must still win over the applied record", v11, HighAvailabilityGroup.reconcileClusterRoleRecords(current, v11, current)); @@ -326,4 +346,276 @@ public void testShouldApplyRefreshedRecord() { assertTrue("Must apply a same-version record with changed roles (autonomous transition)", HighAvailabilityGroup.shouldApplyRefreshedRecord(v10, v10RolesChanged)); } + + /** + * Pins {@link HighAvailabilityGroup#isCausedByInterrupt}: both interrupt marker types (direct and + * wrapped) are detected, non-interrupt chains are not, and the depth-16 bound stops the walk so a + * self-referential (cyclic) cause chain terminates rather than spinning forever. + */ + @Test + public void testIsCausedByInterrupt() { + assertTrue("Direct InterruptedException must be detected", + HighAvailabilityGroup.isCausedByInterrupt(new InterruptedException())); + assertTrue("Direct InterruptedIOException must be detected", + HighAvailabilityGroup.isCausedByInterrupt(new InterruptedIOException())); + assertTrue("A wrapped interrupt marker must be detected", HighAvailabilityGroup + .isCausedByInterrupt(new SQLException("rpc", new InterruptedIOException()))); + assertFalse("A non-interrupt cause chain must not be detected", + HighAvailabilityGroup.isCausedByInterrupt(new SQLException("rpc", new IOException("io")))); + assertFalse("null must not be detected", HighAvailabilityGroup.isCausedByInterrupt(null)); + + // An interrupt marker at chain index 15 is within the depth-16 bound → detected. + Throwable withinBound = new InterruptedException("deep"); + for (int i = 0; i < 15; i++) { + withinBound = new RuntimeException("wrap" + i, withinBound); + } + assertTrue("An interrupt marker at depth 15 (within the bound) must be detected", + HighAvailabilityGroup.isCausedByInterrupt(withinBound)); + + // An interrupt marker at chain index 16 is beyond the bound → not detected. + Throwable beyondBound = new InterruptedException("deeper"); + for (int i = 0; i < 16; i++) { + beyondBound = new RuntimeException("wrap" + i, beyondBound); + } + assertFalse("An interrupt marker at depth 16 (beyond the bound) must not be detected", + HighAvailabilityGroup.isCausedByInterrupt(beyondBound)); + + // A self-referential cause chain must terminate (the depth bound is the cycle guard). + Throwable selfCycle = new RuntimeException() { + @Override + public synchronized Throwable getCause() { + return this; + } + }; + assertFalse("A cyclic cause chain must terminate and not be detected", + HighAvailabilityGroup.isCausedByInterrupt(selfCycle)); + } + + /** + * Wires the refresh no-rollback branch end to end: when the endpoint serves a strictly older + * record than the applied one, {@code refreshClusterRoleRecord} keeps the applied record, stays + * {@code READY}, does not count a failover, and returns {@code true}. A passing + * {@code shouldApplyRefreshedRecord} unit test alone does not prove this branch is wired — an + * inverted guard would silently reintroduce the rollback with the helper test still green. + */ + @Test + public void testRefreshDoesNotRollBackToOlderRecord() throws Exception { + String haGroupName = "testRefreshDoesNotRollBackToOlderRecord"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + ClusterRoleRecord appliedV10 = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 10L); + ClusterRoleRecord staleV9 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 9L); + + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + HighAvailabilityGroup group = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), appliedV10, State.READY)); + Mockito.doReturn(staleV9).when(group).getClusterRoleRecordFromEndpoint(); + + long failoverBefore = GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue(); + assertTrue("Refresh that would roll back must be a no-op returning true", + group.refreshClusterRoleRecord(true)); + assertSame("The applied record must be kept, not rolled back to the older fetched record", + appliedV10, group.getRoleRecord()); + assertSame("HA group must stay READY after a rejected rollback", State.READY, + group.getStateForTesting()); + assertEquals("A rejected rollback must not count as a failover", failoverBefore, + GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue()); + } + + /** + * Wires {@link HighAvailabilityGroup#getClusterRoleRecordFromEndpoint}: because reconcile is + * order-independent for most cases, a url1/url2 fetch swap or a wrong {@code current} argument + * would pass every reconcile helper test yet change first-load behavior. On a first-load + * equal-version divergence reconcile returns the cluster-2 record, so the resolved record must be + * the url2 fetch (a swap would surface the url1 fetch); once a same-version record is applied the + * defer must return that applied record (proving {@code this.roleRecord} is threaded as + * {@code current}). + */ + @Test + public void testGetClusterRoleRecordFromEndpointWiring() throws Exception { + String haGroupName = "testGetClusterRoleRecordFromEndpointWiring"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + ClusterRoleRecord fromUrl1 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 10L); + ClusterRoleRecord fromUrl2 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 10L); + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + + // First load (current == null): equal-version divergence resolves to the cluster-2 (url2) + // fetch. + HighAvailabilityGroup firstLoad = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), null, State.UNINITIALIZED)); + Mockito.doReturn(fromUrl1).when(firstLoad).fetchClusterRoleRecord(url1); + Mockito.doReturn(fromUrl2).when(firstLoad).fetchClusterRoleRecord(url2); + assertSame("url1 must be fetched as cluster 1 and url2 as cluster 2 (a swap would return the " + + "url1 record)", fromUrl2, firstLoad.getClusterRoleRecordFromEndpoint()); + + // Applied record at the endpoints' version: the equal-version defer must return + // this.roleRecord, + // proving current is threaded (otherwise the fall-through would return the url2 record). + ClusterRoleRecord appliedV10 = new ClusterRoleRecord(haGroupName, + HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 10L); + HighAvailabilityGroup applied = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), appliedV10, State.READY)); + Mockito.doReturn(fromUrl1).when(applied).fetchClusterRoleRecord(url1); + Mockito.doReturn(fromUrl2).when(applied).fetchClusterRoleRecord(url2); + assertSame("Equal-version divergence must defer to the applied record (current threaded)", + appliedV10, applied.getClusterRoleRecordFromEndpoint()); + } + + /** + * Cluster-1 fails with CRR-Not-Found but cluster 2 serves a record: the method falls back to the + * cluster-2 record and returns it (no reconciliation on a single reachable endpoint). + */ + @Test + public void testEndpointNotFoundOnCluster1FallsBackToCluster2() throws Exception { + String haGroupName = "testEndpointNotFoundOnCluster1FallsBackToCluster2"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + ClusterRoleRecord fromUrl2 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 10L); + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + + HighAvailabilityGroup group = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), null, State.UNINITIALIZED)); + Mockito + .doThrow( + new SQLException("not found", SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getSQLState(), + SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getErrorCode())) + .when(group).fetchClusterRoleRecord(url1); + Mockito.doReturn(fromUrl2).when(group).fetchClusterRoleRecord(url2); + + assertSame("A cluster-1 Not-Found must fall back to the cluster-2 record", fromUrl2, + group.getClusterRoleRecordFromEndpoint()); + } + + /** + * Cluster-1 fails with CRR-Not-Found and cluster 2 also fails: the original Not-Found propagates + * (so downstream single-cluster fallback can trigger) with the cluster-2 failure attached as a + * suppressed exception, so the single thrown exception carries both root causes. + */ + @Test + public void testEndpointBothFailNotFoundRethrowsNotFoundWithSuppressed() throws Exception { + String haGroupName = "testEndpointBothFailNotFoundRethrowsNotFoundWithSuppressed"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + + HighAvailabilityGroup group = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), null, State.UNINITIALIZED)); + SQLException notFound = + new SQLException("not found", SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getSQLState(), + SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getErrorCode()); + SQLException cluster2Failure = new SQLException("cluster 2 unreachable"); + Mockito.doThrow(notFound).when(group).fetchClusterRoleRecord(url1); + Mockito.doThrow(cluster2Failure).when(group).fetchClusterRoleRecord(url2); + + try { + group.getClusterRoleRecordFromEndpoint(); + fail("Expected the cluster-1 Not-Found to propagate when both endpoints fail"); + } catch (SQLException e) { + assertSame("The original Not-Found must propagate so downstream fallback can trigger", + notFound, e); + assertEquals("Propagated Not-Found must keep its error code", + SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getErrorCode(), e.getErrorCode()); + assertEquals("The cluster-2 failure must be attached as suppressed", 1, + e.getSuppressed().length); + assertSame(cluster2Failure, e.getSuppressed()[0]); + } + } + + /** + * Cluster-1 fails with a non-Not-Found exception and cluster 2 also fails: the cluster-2 + * exception propagates (its distinct error, not the cluster-1 transport failure) with the + * cluster-1 failure attached as suppressed, so the single thrown exception carries both root + * causes. This is the else-branch parity with the Not-Found path — the scenario where cluster 1 + * is merely unreachable and cluster 2 then reports Not-Found. + */ + @Test + public void testEndpointBothFailNonNotFoundRethrowsCluster2WithSuppressed() throws Exception { + String haGroupName = "testEndpointBothFailNonNotFoundRethrowsCluster2WithSuppressed"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + + HighAvailabilityGroup group = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), null, State.UNINITIALIZED)); + SQLException cluster1Failure = new SQLException("cluster 1 unreachable"); + SQLException cluster2NotFound = + new SQLException("not found", SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getSQLState(), + SQLExceptionCode.CLUSTER_ROLE_RECORD_NOT_FOUND.getErrorCode()); + Mockito.doThrow(cluster1Failure).when(group).fetchClusterRoleRecord(url1); + Mockito.doThrow(cluster2NotFound).when(group).fetchClusterRoleRecord(url2); + + try { + group.getClusterRoleRecordFromEndpoint(); + fail("Expected the cluster-2 exception to propagate when both endpoints fail"); + } catch (SQLException e) { + assertSame("The cluster-2 exception must propagate on the non-Not-Found fallback path", + cluster2NotFound, e); + assertEquals("The cluster-1 failure must be attached as suppressed", 1, + e.getSuppressed().length); + assertSame(cluster1Failure, e.getSuppressed()[0]); + } + } + + /** + * Cluster-1 is reachable but the cluster-2 fetch fails: the method degrades to the cluster-1 + * record (no reconciliation), rather than propagating the cluster-2 failure. + */ + @Test + public void testEndpointCluster2FailureDegradesToCluster1() throws Exception { + String haGroupName = "testEndpointCluster2FailureDegradesToCluster1"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + ClusterRoleRecord fromUrl1 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 10L); + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + + HighAvailabilityGroup group = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), null, State.UNINITIALIZED)); + Mockito.doReturn(fromUrl1).when(group).fetchClusterRoleRecord(url1); + Mockito.doThrow(new SQLException("cluster 2 unreachable")).when(group) + .fetchClusterRoleRecord(url2); + + assertSame("A cluster-2 failure must degrade to the cluster-1 record", fromUrl1, + group.getClusterRoleRecordFromEndpoint()); + } + + /** + * A cluster-1 fetch failure wrapping an interruption restores the thread's interrupt status on + * the fallback path (the flag is cleared by the blocking fetch and must be re-raised so callers + * still observe cancellation). + */ + @Test + public void testEndpointRestoresInterruptStatusOnFallback() throws Exception { + String haGroupName = "testEndpointRestoresInterruptStatusOnFallback"; + String url1 = "host1\\:60010"; + String url2 = "host2\\:60010"; + ClusterRoleRecord fromUrl2 = new ClusterRoleRecord(haGroupName, HighAvailabilityPolicy.FAILOVER, + url1, ClusterRole.STANDBY, url2, ClusterRole.ACTIVE, 10L); + HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2); + + HighAvailabilityGroup group = + Mockito.spy(new HighAvailabilityGroup(info, new Properties(), null, State.UNINITIALIZED)); + // Non-Not-Found cluster-1 failure wrapping an interruption; cluster 2 then succeeds. + Mockito.doThrow(new SQLException("interrupted rpc", new InterruptedIOException())).when(group) + .fetchClusterRoleRecord(url1); + Mockito.doReturn(fromUrl2).when(group).fetchClusterRoleRecord(url2); + + // Clear any pre-existing interrupt status so the assertion is meaningful. + Thread.interrupted(); + try { + assertSame("Fallback to cluster 2 must still return its record", fromUrl2, + group.getClusterRoleRecordFromEndpoint()); + assertTrue("The interrupt status must be restored after the fallback fetch", + Thread.currentThread().isInterrupted()); + } finally { + // Consume the interrupt flag so it does not leak into a reused fork. + Thread.interrupted(); + } + } }