This is an automated email from the ASF dual-hosted git repository.
lokiore pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new 9628412934 PHOENIX-7973 HA client can adopt a stale (lower-version)
ClusterRoleRecord when one endpoint lags the peer (#2589)
9628412934 is described below
commit 9628412934b66a25c91f70d9201b8d7db642936c
Author: Lokesh Khurana <[email protected]>
AuthorDate: Wed Aug 19 09:47:25 2026 -0700
PHOENIX-7973 HA client can adopt a stale (lower-version) ClusterRoleRecord
when one endpoint lags the peer (#2589)
* 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 <[email protected]>
---
.../apache/phoenix/jdbc/HighAvailabilityGroup.java | 266 ++++++++++---
.../phoenix/util/GetClusterRoleRecordUtil.java | 72 ++--
.../org/apache/phoenix/jdbc/HAGroupMetricsIT.java | 5 +-
.../phoenix/jdbc/HighAvailabilityGroupTest.java | 433 +++++++++++++++++++++
4 files changed, 694 insertions(+), 82 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 017f996a31..05ce278764 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.query.QueryServicesOptions.DEFAULT_CLIENT_CONNE
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;
@@ -772,6 +773,11 @@ public class HighAvailabilityGroup {
return roleRecord;
}
+ @VisibleForTesting
+ State getStateForTesting() {
+ return state;
+ }
+
/**
* Package private close method.
* <p>
@@ -979,65 +985,140 @@ public class HighAvailabilityGroup {
}
/**
- * Method to get ClusterRoleRecord from RegionServer Endpoints from either
of the clusters.
- * @return ClusterRoleRecord from the first available cluster
+ * 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; if cluster 1 fails, cluster 2's record is used as-is.
+ * <p>
+ * 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 {
+ @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 = null;
+ ClusterRoleRecord roleRecord1 = null;
try {
- // 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;
+ // Read cluster 1's CRR (read-only; no poller side effect).
+ 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 {
+ // 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 {
- return roleRecordFromPR;
+ // 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();
}
- } else {
- return roleRecord;
}
- } 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.
- 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);
- } catch (Exception ignoredEx) {
- throw (SQLException) e;
+ }
+
+ 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. 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",
+ roleRecord1.toPrettyString(), info.getUrl1(), info.getUrl2(),
info.getUrl1(), e);
+ }
+ if (roleRecord2 == null) {
+ resolvedRecord = roleRecord1;
+ } else {
+ // 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());
}
}
+ }
+
+ // 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;
+ }
- // 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);
+ /**
+ * 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.
+ */
+ @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;
+ }
}
+ return false;
}
/**
@@ -1106,6 +1187,16 @@ public class HighAvailabilityGroup {
return true;
}
+ // 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 {};"
+ + " 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());
@@ -1240,4 +1331,89 @@ public class HighAvailabilityGroup {
return transitionSucceeded &&
!oldRecord.getActiveUrl().equals(newRecord.getActiveUrl())
&& newRecord.getActiveUrl().isPresent();
}
+
+ /**
+ * 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 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, 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
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;
+ 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;
+ }
+ // 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;
+ }
+
+ /**
+ * 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
+ * <em>equal</em> 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}
+ */
+ static boolean shouldApplyRefreshedRecord(ClusterRoleRecord current,
ClusterRoleRecord fetched) {
+ return !current.isNewerThan(fetched);
+ }
}
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 16c7715c17..b25d5f46f2 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 @@ public class 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 @@ public class GetClusterRoleRecordUtil {
}
/**
- * 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.
- * <p>
- * 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 @@ public class GetClusterRoleRecordUtil {
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 9589204cb5..7f02288cc6 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
@@ -248,8 +248,9 @@ public class HAGroupMetricsIT extends HABaseIT {
@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 cc58727838..5bf956fa88 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,13 +20,21 @@ package org.apache.phoenix.jdbc;
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 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;
@@ -185,4 +193,429 @@ public class HighAvailabilityGroupTest {
+ "should count as a failover",
HighAvailabilityGroup.shouldCountFailover(true, bothStandby,
aStandbyBActive));
}
+
+ /** Reconciliation prefers non-UNKNOWN over UNKNOWN, then higher version;
order-independent. */
+ @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);
+
+ // Both usable: higher version wins (the stale-revert regression guard),
either order.
+ 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.
+ 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.
+ 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, 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));
+
+ // 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));
+
+ // 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));
+ }
+
+ /** Refresh guard rejects a lower version but applies a same-version role
change. */
+ @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
transition (roles change,
+ // admin version does not).
+ ClusterRoleRecord v10RolesChanged = new ClusterRoleRecord(haGroupName,
+ HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2,
ClusterRole.STANDBY, 10L);
+
+ assertFalse("Must not roll back from the applied v10 to a stale fetched
v9",
+ HighAvailabilityGroup.shouldApplyRefreshedRecord(v10, v9));
+ assertTrue("Must apply a strictly newer fetched record",
+ HighAvailabilityGroup.shouldApplyRefreshedRecord(v9, v10));
+ // 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));
+ }
+
+ /**
+ * 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();
+ }
+ }
}
