This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new c8c54d6e4a PHOENIX-7920 Add replay-side handling for direct
DEGRADED_STANDBY → STANDBY_TO_ACTIVE transition (ANISTS → AISTS) (#2577)
c8c54d6e4a is described below
commit c8c54d6e4af4c8d051588cf08b873f7dcb6df9cd
Author: Himanshu Gwalani <[email protected]>
AuthorDate: Fri Aug 7 19:42:20 2026 +0530
PHOENIX-7920 Add replay-side handling for direct DEGRADED_STANDBY →
STANDBY_TO_ACTIVE transition (ANISTS → AISTS) (#2577)
* Add replay-side handling for direct DEGRADED_STANDBY → STANDBY_TO_ACTIVE
transition (ANISTS → AISTS)
* Adding E2E test including HA Store changes
* PHOENIX-7920 Apply spotless formatting and inline findLogFiles test helper
* PHOENIX-7920 SAF failover: restart into STANDBY_TO_ACTIVE inits to SYNC
when nothing to rewind
On a restart while already STANDBY_TO_ACTIVE (e.g. an RS bounce after a
direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE), enter SYNCED_RECOVERY only when
a rewind is actually pending (lastRoundInSync behind lastRoundProcessed). When
the rounds are equal there is nothing to rewind, so initialize directly to SYNC
and let the first replay() promote immediately, instead of waiting ~one round
window for the SYNCED_RECOVERY -> SYNC CAS that only fires inside the
round-processing loop. Arm fail [...]
Add regression guard
testReplay_RestartIntoStandbyToActive_NothingToReplay_PromotesOnFirstTick
(promotes on the first replay tick) and update the existing STANDBY_TO_ACTIVE
init tests for the SYNC outcome. In StoreAndForwardFailoverIT, gate
transitionCluster2's peer-reaction settle-poll to the DEGRADED_STANDBY target
(with named constants) and note that PHOENIX_REPLICATION_REPLAY_ENABLED also
gates the compaction consistency-point guard.
Document degradedListener's authoritative (non-CAS) set and the init-time
compareAndSet(NOT_INITIALIZED, ...) yield-to-listener semantics.
* PHOENIX-7920 Add E2E IT for direct failover from ACTIVE_NOT_IN_SYNC with
automatic state transitions
* PHOENIX-7920 recoveryListener uses compareAndSet(DEGRADED,
SYNCED_RECOVERY), not unconditional set
A spurious LOCAL STANDBY event arriving while already SYNC (e.g. an
aborted-failover round trip, or a cache-reconnect redelivery) would otherwise
flip SYNC -> SYNCED_RECOVERY and needlessly re-process the frontier round. CAS
from DEGRADED only makes recovery safe by construction, symmetric with
triggerFailoverListner.
- Add guard IT: Phase 1 proves a spurious STANDBY-while-SYNC stays SYNC;
Phase 2 proves a genuine DEGRADED_STANDBY -> STANDBY still reaches
SYNCED_RECOVERY. Adds effectiveLocalStateIs helper.
- Tighten HealthyStandbyToActiveKeepsSync: use failoverPending as the
synchronization barrier and re-assert both invariants after settling.
- Document that the LOCAL HA-state listeners run on the notification
callback and must be non-blocking.
* PHOENIX-7920 Address PR review: log CAS result, add init-race regression
test, dedup findLogFiles
- ReplicationLogDiscoveryReplay: capture and log the compareAndSet result
(rewindScheduled) in recoveryListener and triggerFailoverListner so the
guarded-transition outcome is visible in logs.
- ReplicationLogDiscoveryReplayTestIT: add a deterministic init-window race
regression test (direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE reconciles)
and fix a stale ordering comment.
- Route findLogFiles through CrossClusterReplicationTestUtil and remove the
pass-through delegate; repoint ReplicationLogGroupBaseIT and
ReplicationLogGroupIT call sites to the shared util.
---------
Co-authored-by: Himanshu Gwalani
<[email protected]>
---
.../replication/ReplicationLogDiscovery.java | 27 +-
.../reader/ReplicationLogDiscoveryReplay.java | 215 +++++++--
.../apache/phoenix/jdbc/HAGroupStoreManagerIT.java | 116 +++++
.../CrossClusterReplicationTestUtil.java | 117 +++++
.../replication/ReplicationLogGroupBaseIT.java | 69 +--
.../phoenix/replication/ReplicationLogGroupIT.java | 2 +-
.../replication/ReplicationLogGroupTestAccess.java | 67 +++
.../ReplicationLogDiscoveryReplayTestIT.java | 525 ++++++++++++++++++++-
.../reader/StoreAndForwardFailoverIT.java | 395 ++++++++++++++++
9 files changed, 1417 insertions(+), 116 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java
index ccd0e82caf..638d95280f 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java
@@ -104,7 +104,7 @@ public abstract class ReplicationLogDiscovery {
protected final ReplicationLogTracker replicationLogTracker;
protected ScheduledExecutorService scheduler;
protected volatile boolean isRunning = false;
- protected ReplicationRound lastRoundProcessed;
+ protected volatile ReplicationRound lastRoundProcessed;
protected MetricsReplicationLogDiscovery metrics;
protected long roundTimeMills;
protected long bufferMillis;
@@ -382,16 +382,30 @@ public abstract class ReplicationLogDiscovery {
/** Creates a new metrics source for monitoring operations. */
protected abstract MetricsReplicationLogDiscovery createMetricsSource();
+ /**
+ * Initializes lastRoundProcessed, sampling the current time up front to use
as the no-files
+ * fallback (see {@link #initializeLastRoundProcessed(long)}). Sampling here
- before the file
+ * scans - keeps the fallback anchored to when initialization began.
+ * @throws IOException if there's an error reading file timestamps
+ */
+ protected void initializeLastRoundProcessed() throws IOException {
+ initializeLastRoundProcessed(EnvironmentEdgeManager.currentTime());
+ }
+
/**
* Initializes lastRoundProcessed based on minimum timestamp from 1.
In-progress files (highest
* priority) - indicates partially processed rounds 2. New files (medium
priority) - indicates
- * unprocessed rounds waiting to be replayed 3. Current time (fallback) -
used when no files
- * exist, starts from current time The minimum timestamp is converted to a
replication round using
+ * unprocessed rounds waiting to be replayed 3. fallbackCurrentTime - used
when no files exist,
+ * starts from the supplied time The minimum timestamp is converted to a
replication round using
* getReplicationRoundFromEndTime(), which rounds down to the nearest round
boundary to ensure we
* start from a complete round.
+ * @param fallbackCurrentTime the timestamp to start from when no files
exist; sampled by the
+ * caller at the start of initialization rather
than re-read here, so a
+ * slow init (or a state transition during init)
cannot push the
+ * starting round forward
* @throws IOException if there's an error reading file timestamps
*/
- protected void initializeLastRoundProcessed() throws IOException {
+ protected void initializeLastRoundProcessed(long fallbackCurrentTime) throws
IOException {
Optional<Long> minTimestampFromInProgressFiles =
getMinTimestampFromInProgressFiles();
if (minTimestampFromInProgressFiles.isPresent()) {
LOG.info(
@@ -412,11 +426,10 @@ public abstract class ReplicationLogDiscovery {
this.lastRoundProcessed =
replicationLogTracker.getReplicationShardDirectoryManager()
.getReplicationRoundFromEndTime(minTimestampFromNewFiles.get());
} else {
- long currentTime = EnvironmentEdgeManager.currentTime();
LOG.info("Initializing lastRoundProcessed for haGroup: {} from current
time {}",
- haGroupName, currentTime);
+ haGroupName, fallbackCurrentTime);
this.lastRoundProcessed =
replicationLogTracker.getReplicationShardDirectoryManager()
- .getReplicationRoundFromEndTime(currentTime);
+ .getReplicationRoundFromEndTime(fallbackCurrentTime);
}
}
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java
index ba36df9a0b..a3ae1ca1f7 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java
@@ -121,12 +121,23 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
LOG.info("Initializing ReplicationLogDiscoveryReplay for haGroup: {}",
haGroupName);
+ // The LOCAL HA-state listeners below run synchronously on the HA store's
state-change
+ // notification callback, so they must not block: each does only fast,
wait-free work
+ // (an atomic state transition plus logging). Blocking here would stall
delivery of
+ // subsequent HA state notifications for this group.
HAGroupStateListener degradedListener =
(groupName, fromState, toState, modifiedTime, clusterType,
lastSyncStateTimeInMs) -> {
if (
clusterType == ClusterType.LOCAL
&& HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(toState)
) {
+ // Unconditional set (not compareAndSet): degradation is an
authoritative, fail-closed
+ // signal. Whatever the current state (SYNC, SYNCED_RECOVERY, or
even NOT_INITIALIZED if a
+ // degrade lands mid-init), replay must stop advancing the sync
point. There is no single
+ // valid "from" state to CAS against, so a CAS would silently drop
legitimate degrade
+ // signals arriving from the other states. Contrast
triggerFailoverListner below, which is
+ // deliberately conditional (compareAndSet from DEGRADED only) so it
cannot clobber a
+ // healthy SYNC or an already-pending SYNCED_RECOVERY.
replicationReplayState.set(ReplicationReplayState.DEGRADED);
LOG.info("Cluster degraded detected for {}.
replicationReplayState={}", haGroupName,
ReplicationReplayState.DEGRADED);
@@ -139,9 +150,18 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
clusterType == ClusterType.LOCAL
&& HAGroupStoreRecord.HAGroupState.STANDBY.equals(toState)
) {
- replicationReplayState.set(ReplicationReplayState.SYNCED_RECOVERY);
- LOG.info("Cluster recovered detected for {}.
replicationReplayState={}", haGroupName,
- getReplicationReplayState());
+ // compareAndSet from DEGRADED only, not an unconditional set():
DEGRADED is the only
+ // state whose lastRoundInSync lags lastRoundProcessed, hence the
only state a recovery
+ // to STANDBY must rewind from. A STANDBY event that lands while
already SYNC (e.g.
+ // ABORT_TO_STANDBY -> STANDBY after a prior rewind, or a
cache-reconnect redelivery)
+ // would otherwise flip SYNC -> SYNCED_RECOVERY and needlessly
re-process the frontier
+ // round. Symmetric with triggerFailoverListner below; contrast
degradedListener above,
+ // whose unconditional fail-closed set() is intentional.
+ boolean rewindScheduled = replicationReplayState
+ .compareAndSet(ReplicationReplayState.DEGRADED,
ReplicationReplayState.SYNCED_RECOVERY);
+ LOG.info(
+ "Cluster recovered detected for {}. replicationReplayState={},
rewindScheduled={}",
+ haGroupName, getReplicationReplayState(), rewindScheduled);
}
};
@@ -151,11 +171,24 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
clusterType == ClusterType.LOCAL
&&
HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(toState)
) {
+ // Direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE skips the STANDBY
event that normally
+ // drives recovery. If we are DEGRADED, schedule the rewind so
replay() re-syncs from
+ // lastRoundInSync before shouldTriggerFailover() (which gates on
SYNC) can promote.
+ //
+ // Arm failoverPending BEFORE the state CAS. This ordering is a
contract with the
+ // init-window reconcile in initializeLastRoundProcessed(): if this
listener fires while
+ // init still holds NOT_INITIALIZED, the CAS below no-ops and only
the arm survives, so
+ // arming first guarantees the reconcile sees the arm whenever it
still sees DEGRADED.
+ //
+ // compareAndSet, not set: a listener firing while already SYNC
(healthy failover) or
+ // SYNCED_RECOVERY (rewind already pending) must not clobber a good
state.
failoverPending.set(true);
+ boolean rewindScheduled = replicationReplayState
+ .compareAndSet(ReplicationReplayState.DEGRADED,
ReplicationReplayState.SYNCED_RECOVERY);
LOG.info(
- "Failover trigger detected for {}. replicationReplayState={}. "
+ "Failover trigger detected for {}. replicationReplayState={},
rewindScheduled={}. "
+ "Setting failover pending to {}",
- haGroupName, getReplicationReplayState(), failoverPending.get());
+ haGroupName, getReplicationReplayState(), rewindScheduled,
failoverPending.get());
}
};
@@ -202,50 +235,80 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
}
/**
- * Initializes lastRoundProcessed and lastRoundInSync based on HA group
state. For DEGRADED states
- * (DEGRADED_STANDBY, DEGRADED_STANDBY_FOR_WRITER): - Sets
replicationReplayState to DEGRADED -
- * Initializes lastRoundProcessed from minimum of: in-progress files, new
files, or current time -
- * Initializes lastRoundInSync from minimum of: lastSyncStateTimeInMs (from
HA Store) or minimum
- * timestamp from IN and IN PROGRESS files - This ensures lastRoundInSync
represents the last
- * known good sync point before degradation For SYNC states (STANDBY): - Sets
- * replicationReplayState to SYNC - Calls parent's
initializeLastRoundProcessed() to initialize
- * lastRoundProcessed - Sets lastRoundInSync equal to lastRoundProcessed
(both are in sync)
+ * Initializes lastRoundProcessed and lastRoundInSync based on the persisted
HA group state.
+ * <ul>
+ * <li>DEGRADED_STANDBY: sets replicationReplayState to DEGRADED and
initializes both rounds from
+ * the last known good sync point via {@link
#initLastRoundsFromLastSyncPoint(HAGroupStoreRecord)}
+ * (lastRoundInSync from the minimum of lastSyncStateTimeInMs and the file
frontier, so it
+ * represents the last consistent point before degradation).</li>
+ * <li>STANDBY_TO_ACTIVE: a restart while already mid-failover (e.g. after a
direct
+ * DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). Initializes both
rounds from the last
+ * known good sync point, then sets replicationReplayState to
SYNCED_RECOVERY only when a rewind
+ * is actually pending (lastRoundInSync behind lastRoundProcessed) so the
first replay() rewinds
+ * before failover can promote; when the rounds are already equal it
initializes directly to SYNC
+ * so promotion is not delayed by a round window.</li>
+ * <li>Other states (e.g. STANDBY): sets replicationReplayState to SYNC,
delegates to the parent
+ * to initialize lastRoundProcessed, and sets lastRoundInSync equal to
lastRoundProcessed.</li>
+ * </ul>
+ * When the state is STANDBY_TO_ACTIVE, failoverPending is also armed.
* @throws IOException if there's an error reading HA group state or file
timestamps
*/
@Override
protected void initializeLastRoundProcessed() throws IOException {
LOG.info("Initializing last round processed for haGroup: {}", haGroupName);
+ // Sample current time BEFORE reading the HA group state, so if the group
transitions (e.g.
+ // SYNC -> DEGRADED_STANDBY) during init, the starting round still anchors
to when init began
+ // rather than to a later point after the state read and file scans.
+ long frontierStartTime = EnvironmentEdgeManager.currentTime();
HAGroupStoreRecord haGroupStoreRecord = getHAGroupRecord();
- LOG.info("Found HA Group state during initialization as {} for haGroup:
{}",
- haGroupStoreRecord.getHAGroupState(), haGroupName);
- if (
-
HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupStoreRecord.getHAGroupState())
- ) {
+ HAGroupStoreRecord.HAGroupState haGroupState =
haGroupStoreRecord.getHAGroupState();
+ LOG.info("Found HA Group state during initialization as {} for haGroup:
{}", haGroupState,
+ haGroupName);
+ // Each branch below sets the initial state with
compareAndSet(NOT_INITIALIZED, ...), not a
+ // plain set: the LOCAL state listeners are subscribed in init() before
this runs, so a
+ // concurrent LOCAL transition may have already advanced the state off
NOT_INITIALIZED. When
+ // that happens the CAS no-ops and we deliberately keep the listener's
value -- a live
+ // transition is more recent (and thus more authoritative) than the record
snapshot read above.
+ if (HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupState))
{
replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED,
ReplicationReplayState.DEGRADED);
- long minimumTimestampFromFiles = EnvironmentEdgeManager.currentTime();
- Optional<Long> minTimestampFromInProgressFiles =
getMinTimestampFromInProgressFiles();
- Optional<Long> minTimestampFromNewFiles = getMinTimestampFromNewFiles();
- if (minTimestampFromInProgressFiles.isPresent()) {
- LOG.info("Found minimum timestamp from IN PROGRESS files as {}",
- minTimestampFromInProgressFiles.get());
- minimumTimestampFromFiles =
- Math.min(minimumTimestampFromFiles,
minTimestampFromInProgressFiles.get());
+ initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime);
+ } else if
(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) {
+ // Restarted while already in STANDBY_TO_ACTIVE (e.g. an RS bounce after
a direct
+ // DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). No listener fires
on a fresh process, so
+ // initialize as if recovering: lastRoundInSync at the last good sync
point,
+ // lastRoundProcessed
+ // from the file frontier.
+ initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime);
+ // Enter SYNCED_RECOVERY only when there is real rewind work
(lastRoundInSync behind
+ // lastRoundProcessed) so the first replay() rewinds before
shouldTriggerFailover() promotes.
+ // When the rounds are equal (the common "already fully replayed, RS
bounced before
+ // setHAGroupStatusToSync() completed" case) there is nothing to rewind,
so initialize
+ // directly to SYNC. Otherwise the SYNCED_RECOVERY -> SYNC CAS (which
only fires inside the
+ // round-processing loop) would not run until a round becomes
time-eligible, delaying
+ // promotion by ~one round window.
+ ReplicationReplayState initialState =
+ lastRoundInSync.getEndTime() < lastRoundProcessed.getEndTime()
+ ? ReplicationReplayState.SYNCED_RECOVERY
+ : ReplicationReplayState.SYNC;
+ // Distinct, greppable signal that this process restarted mid-failover,
and which sub-case it
+ // hit, so a restart-driven promotion is diagnosable from logs (not
inferred from the generic
+ // init summary below).
+ if (initialState == ReplicationReplayState.SYNCED_RECOVERY) {
+ LOG.info(
+ "Restart into STANDBY_TO_ACTIVE with rewind pending for haGroup {}:
lastRoundInSync={} "
+ + "behind lastRoundProcessed={}; entering SYNCED_RECOVERY to
re-sync before promotion.",
+ haGroupName, lastRoundInSync, lastRoundProcessed);
+ } else {
+ LOG.info("Restart into STANDBY_TO_ACTIVE with nothing to rewind for
haGroup {}: "
+ + "lastRoundInSync == lastRoundProcessed ({}); entering SYNC,
promotion eligible "
+ + "immediately.", haGroupName, lastRoundProcessed);
}
- if (minTimestampFromNewFiles.isPresent()) {
- LOG.info("Found minimum timestamp from IN files as {}",
minTimestampFromNewFiles.get());
- minimumTimestampFromFiles =
- Math.min(minimumTimestampFromFiles, minTimestampFromNewFiles.get());
- }
- this.lastRoundProcessed =
replicationLogTracker.getReplicationShardDirectoryManager()
- .getReplicationRoundFromEndTime(minimumTimestampFromFiles);
- this.lastRoundInSync =
-
replicationLogTracker.getReplicationShardDirectoryManager().getReplicationRoundFromEndTime(
- Math.min(haGroupStoreRecord.getLastSyncStateTimeInMs(),
minimumTimestampFromFiles));
+
replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED,
initialState);
} else {
replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED,
ReplicationReplayState.SYNC);
- super.initializeLastRoundProcessed();
+ super.initializeLastRoundProcessed(frontierStartTime);
this.lastRoundInSync =
new ReplicationRound(lastRoundProcessed.getStartTime(),
lastRoundProcessed.getEndTime());
}
@@ -254,12 +317,80 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
+ "replication replay state as {}",
lastRoundProcessed, lastRoundInSync, replicationReplayState);
- // Update the failoverPending variable during initialization
- if (
-
HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupStoreRecord.getHAGroupState())
- ) {
- failoverPending.compareAndSet(false, true);
+ // Arm failoverPending during initialization when restarting mid-failover.
Plain set (not
+ // compareAndSet): it starts false and this is the only initializer, so
there is no prior value
+ // to guard; this also matches triggerFailoverListner, which arms the same
flag with set(true).
+ if
(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) {
+ failoverPending.set(true);
+ }
+
+ // Reconcile the init-window race between the (possibly stale) record read
above and the live
+ // LOCAL listeners subscribed in init(). A direct DEGRADED_STANDBY ->
STANDBY_TO_ACTIVE flip can
+ // fire triggerFailoverListner while init still holds NOT_INITIALIZED: its
CAS(DEGRADED,
+ // SYNCED_RECOVERY) no-ops and only failoverPending survives; then the
DEGRADED_STANDBY branch's
+ // CAS(NOT_INITIALIZED, DEGRADED) lands, leaving state DEGRADED with a
failover pending. Nothing
+ // re-fires (the record is already STANDBY_TO_ACTIVE; a direct failover
never revisits STANDBY),
+ // so shouldTriggerFailover() -- gated on SYNC -- never promotes until an
RS restart heals it.
+ // Finish the promotion the listener intended. No-op (no warn) on a normal
STANDBY_TO_ACTIVE
+ // restart (state already SYNCED_RECOVERY or SYNC) and skipped for a
legitimate DEGRADED standby
+ // (failoverPending false). triggerFailoverListner arms failoverPending
before its CAS, so a
+ // still-DEGRADED state here guarantees the arm is visible, closing the
window.
+ if (failoverPending.get()) {
+ if (
+ replicationReplayState.compareAndSet(ReplicationReplayState.DEGRADED,
+ ReplicationReplayState.SYNCED_RECOVERY)
+ ) {
+ LOG.warn(
+ "Reconciled init-window failover race for haGroup {}: state was
DEGRADED with a pending "
+ + "failover (a DEGRADED_STANDBY -> STANDBY_TO_ACTIVE flip raced
init); promoted to "
+ + "SYNCED_RECOVERY so replay can rewind from lastRoundInSync and
then promote.",
+ haGroupName);
+ }
+ }
+ }
+
+ /**
+ * Initializes {@link #lastRoundProcessed} and {@link #lastRoundInSync} from
the last known good
+ * sync point. Used by the DEGRADED_STANDBY and STANDBY_TO_ACTIVE branches,
which both need a
+ * rewind to the last consistent point. lastRoundProcessed is derived from
the minimum timestamp
+ * across IN-PROGRESS and IN files (or {@code frontierStartTime} when no
files exist);
+ * lastRoundInSync is derived from the minimum of the record's
lastSyncStateTimeInMs and that file
+ * frontier, so it never sits ahead of the last synced data. When
lastSyncStateTimeInMs is 0 (no
+ * known sync point), it falls back to the file frontier so lastRoundInSync
collapses onto
+ * lastRoundProcessed instead of rewinding to the epoch.
+ * @param haGroupStoreRecord the persisted HA group record supplying
lastSyncStateTimeInMs
+ * @param frontierStartTime current time sampled at the start of
initialization; the file
+ * frontier upper bound, and the sole basis when
no files exist
+ */
+ private void initLastRoundsFromLastSyncPoint(HAGroupStoreRecord
haGroupStoreRecord,
+ long frontierStartTime) throws IOException {
+ long minimumTimestampFromFiles = frontierStartTime;
+ Optional<Long> minTimestampFromInProgressFiles =
getMinTimestampFromInProgressFiles();
+ Optional<Long> minTimestampFromNewFiles = getMinTimestampFromNewFiles();
+ if (minTimestampFromInProgressFiles.isPresent()) {
+ LOG.info("Found minimum timestamp from IN PROGRESS files as {}",
+ minTimestampFromInProgressFiles.get());
+ minimumTimestampFromFiles =
+ Math.min(minimumTimestampFromFiles,
minTimestampFromInProgressFiles.get());
+ }
+ if (minTimestampFromNewFiles.isPresent()) {
+ LOG.info("Found minimum timestamp from IN files as {}",
minTimestampFromNewFiles.get());
+ minimumTimestampFromFiles =
+ Math.min(minimumTimestampFromFiles, minTimestampFromNewFiles.get());
}
+ this.lastRoundProcessed =
replicationLogTracker.getReplicationShardDirectoryManager()
+ .getReplicationRoundFromEndTime(minimumTimestampFromFiles);
+ // A lastSyncStateTimeInMs of 0 means "no known sync point" (never synced,
or a record that
+ // predates the field). Do NOT feed 0 into the min:
getReplicationRoundFromEndTime(0) returns
+ // ReplicationRound(0, 0), which would rewind SYNCED_RECOVERY to the epoch
and drive the
+ // consistency point to 0 (retain-everything). Fall back to the file
frontier so lastRoundInSync
+ // collapses onto lastRoundProcessed (no rewind) when there is no known
sync point.
+ long lastSyncStateTimeInMs = haGroupStoreRecord.getLastSyncStateTimeInMs();
+ long lastSyncBasis = lastSyncStateTimeInMs > 0L
+ ? Math.min(lastSyncStateTimeInMs, minimumTimestampFromFiles)
+ : minimumTimestampFromFiles;
+ this.lastRoundInSync =
replicationLogTracker.getReplicationShardDirectoryManager()
+ .getReplicationRoundFromEndTime(lastSyncBasis);
}
/**
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreManagerIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreManagerIT.java
index f2697ae719..183d538e0f 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreManagerIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreManagerIT.java
@@ -18,11 +18,15 @@
package org.apache.phoenix.jdbc;
import static org.apache.hadoop.hbase.HConstants.ZK_SESSION_TIMEOUT;
+import static org.apache.hadoop.test.GenericTestUtils.waitFor;
import static
org.apache.phoenix.jdbc.HAGroupStoreClient.ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE;
import static org.apache.phoenix.jdbc.PhoenixHAAdmin.getLocalZkUrl;
import static org.apache.phoenix.jdbc.PhoenixHAAdmin.toPath;
import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
+import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STORE_CLIENT_PREWARM_ENABLED;
import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.DEFAULT_REPLICATION_ROUND_DURATION_SECONDS;
+import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY;
+import static
org.apache.phoenix.replication.reader.ReplicationLogReplayService.PHOENIX_REPLICATION_REPLAY_ENABLED;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
@@ -30,6 +34,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import java.io.IOException;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Optional;
@@ -38,8 +43,10 @@ import java.util.stream.Collectors;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.exception.InvalidClusterRoleTransitionException;
+import org.apache.phoenix.replication.ReplicationLogGroup;
import org.apache.phoenix.replication.reader.ReplicationLogReplay;
import org.apache.phoenix.util.HAGroupStoreTestUtil;
import org.apache.zookeeper.data.Stat;
@@ -61,6 +68,9 @@ public class HAGroupStoreManagerIT extends HABaseIT {
private PhoenixHAAdmin haAdmin;
private static final Long ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS = 2000L;
+ private static final int HA_STATE_POLL_INTERVAL_MS = 100;
+ private static final int HA_STATE_TRANSITION_TIMEOUT_MS = 120000;
+ private static final int TEST_REPLICATION_ROUND_DURATION_SECONDS = 2;
private String zkUrl;
private String peerZKUrl;
@@ -70,6 +80,12 @@ public class HAGroupStoreManagerIT extends HABaseIT {
conf1.setLong(ZK_SESSION_TIMEOUT, 30 * 1000);
conf2.setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
conf2.setLong(ZK_SESSION_TIMEOUT, 30 * 1000);
+ // Both mini-clusters run in one JVM. Keep background initialization from
claiming shared
+ // replay and log-group singletons before each test binds them to the
intended cluster.
+ conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false);
+ conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false);
+ conf1.setBoolean(HA_GROUP_STORE_CLIENT_PREWARM_ENABLED, false);
+ conf2.setBoolean(HA_GROUP_STORE_CLIENT_PREWARM_ENABLED, false);
CLUSTERS.start();
}
@@ -623,6 +639,106 @@ public class HAGroupStoreManagerIT extends HABaseIT {
}
}
+ private static void waitForHAGroupState(HAGroupStoreManager manager, String
haGroupName,
+ HAGroupStoreRecord.HAGroupState expectedState) throws Exception {
+ waitFor(() -> {
+ try {
+ Optional<HAGroupStoreRecord> record =
manager.getHAGroupStoreRecord(haGroupName);
+ return record.isPresent() && record.get().getHAGroupState() ==
expectedState;
+ } catch (IOException e) {
+ return false;
+ }
+ }, HA_STATE_POLL_INTERVAL_MS, HA_STATE_TRANSITION_TIMEOUT_MS);
+ }
+
+ private static void waitForEffectiveHAGroupState(HAGroupStoreManager
manager, String haGroupName,
+ HAGroupStoreRecord.HAGroupState expectedState) throws Exception {
+ waitFor(() -> {
+ try {
+ Optional<HAGroupStoreRecord> record =
manager.getEffectiveHAGroupStoreRecord(haGroupName);
+ return record.isPresent() && record.get().getHAGroupState() ==
expectedState;
+ } catch (IOException e) {
+ return false;
+ }
+ }, HA_STATE_POLL_INTERVAL_MS, HA_STATE_TRANSITION_TIMEOUT_MS);
+ }
+
+ @Test(timeout = 240000)
+ public void
testE2EFailoverFromActiveNotInSyncWithAutomaticStateTransitions() throws
Exception {
+ String haGroupName = testName.getMethodName();
+ Configuration cluster1Conf =
CLUSTERS.getHBaseCluster1().getConfiguration();
+ Configuration cluster2Conf =
CLUSTERS.getHBaseCluster2().getConfiguration();
+ String zkUrl1 = getLocalZkUrl(cluster1Conf);
+ String zkUrl2 = getLocalZkUrl(cluster2Conf);
+
+ cluster1Conf.setLong(ZK_SESSION_TIMEOUT, 10 * 1000);
+ cluster2Conf.setLong(ZK_SESSION_TIMEOUT, 10 * 1000);
+ cluster1Conf.setInt(PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY,
+ TEST_REPLICATION_ROUND_DURATION_SECONDS);
+ cluster2Conf.setInt(PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY,
+ TEST_REPLICATION_ROUND_DURATION_SECONDS);
+
+ HAGroupStoreTestUtil.upsertHAGroupRecordInSystemTable(haGroupName, zkUrl1,
zkUrl2,
+ CLUSTERS.getMasterAddress1(), CLUSTERS.getMasterAddress2(),
+ ClusterRoleRecord.ClusterRole.ACTIVE,
ClusterRoleRecord.ClusterRole.STANDBY, null,
+ CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2());
+ HAGroupStoreTestUtil.upsertHAGroupRecordInSystemTable(haGroupName, zkUrl1,
zkUrl2,
+ CLUSTERS.getMasterAddress1(), CLUSTERS.getMasterAddress2(),
+ ClusterRoleRecord.ClusterRole.ACTIVE,
ClusterRoleRecord.ClusterRole.STANDBY, zkUrl2,
+ CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2());
+
+ HAGroupStoreManager cluster1HAManager =
HAGroupStoreManager.getInstance(cluster1Conf);
+ HAGroupStoreManager cluster2HAManager =
HAGroupStoreManager.getInstance(cluster2Conf);
+ ReplicationLogGroup replicationLogGroup = null;
+ ReplicationLogReplay replicationLogReplay = null;
+ try {
+ cluster1HAManager.getHAGroupStoreRecord(haGroupName);
+ cluster2HAManager.getHAGroupStoreRecord(haGroupName);
+ waitForEffectiveHAGroupState(cluster1HAManager, haGroupName,
+ HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ waitForEffectiveHAGroupState(cluster2HAManager, haGroupName,
+ HAGroupStoreRecord.HAGroupState.STANDBY);
+
+ HRegionServer regionServer =
CLUSTERS.getHBaseCluster1().getHBaseCluster().getRegionServer(0);
+ replicationLogGroup = ReplicationLogGroup.get(cluster1Conf,
regionServer.getServerName(),
+ haGroupName, cluster1HAManager);
+ replicationLogReplay = ReplicationLogReplay.get(cluster2Conf,
haGroupName);
+ replicationLogReplay.startReplay();
+
+ assertEquals(0L,
cluster1HAManager.setHAGroupStatusToStoreAndForward(haGroupName));
+ waitForHAGroupState(cluster1HAManager, haGroupName,
+ HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC);
+ waitForHAGroupState(cluster2HAManager, haGroupName,
+ HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY);
+
+ assertEquals(0L,
cluster1HAManager.initiateFailoverOnActiveCluster(haGroupName));
+ waitForHAGroupState(cluster1HAManager, haGroupName,
+ HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY);
+ waitForHAGroupState(cluster2HAManager, haGroupName,
+ HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE);
+ replicationLogGroup.close();
+ replicationLogGroup = null;
+
+ waitForHAGroupState(cluster2HAManager, haGroupName,
+ HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
+ waitForHAGroupState(cluster1HAManager, haGroupName,
HAGroupStoreRecord.HAGroupState.STANDBY);
+
+ ClusterRoleRecord cluster1Role =
cluster1HAManager.getClusterRoleRecord(haGroupName);
+ ClusterRoleRecord cluster2Role =
cluster2HAManager.getClusterRoleRecord(haGroupName);
+ assertEquals(ClusterRoleRecord.ClusterRole.STANDBY,
+ cluster1Role.getRole(CLUSTERS.getMasterAddress1()));
+ assertEquals(ClusterRoleRecord.ClusterRole.ACTIVE,
+ cluster2Role.getRole(CLUSTERS.getMasterAddress2()));
+ } finally {
+ if (replicationLogReplay != null) {
+ replicationLogReplay.close();
+ }
+ if (replicationLogGroup != null) {
+ replicationLogGroup.close();
+ }
+ }
+ }
+
@Test
public void testE2EFailoverWithAutomaticStateTransitions() throws Exception {
String haGroupName = testName.getMethodName();
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java
new file mode 100644
index 0000000000..127f793415
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.replication;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.phoenix.util.TestUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Shared cross-cluster helpers for replication ITs (extracted from
ReplicationLogGroupIT). */
+public final class CrossClusterReplicationTestUtil {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CrossClusterReplicationTestUtil.class);
+
+ private CrossClusterReplicationTestUtil() {
+ }
+
+ /** Recursively collect every ".plog" file under {@code dir}; empty list if
{@code dir} absent. */
+ public static List<Path> findLogFiles(Path dir, FileSystem fs) throws
IOException {
+ List<Path> files = new ArrayList<>();
+ findLogFilesRecursive(dir, fs, files);
+ return files;
+ }
+
+ private static void findLogFilesRecursive(Path dir, FileSystem fs,
List<Path> files)
+ throws IOException {
+ if (!fs.exists(dir)) {
+ return;
+ }
+ for (FileStatus status : fs.listStatus(dir)) {
+ if (status.isDirectory()) {
+ findLogFilesRecursive(status.getPath(), fs, files);
+ } else if (status.getPath().getName().endsWith(".plog")) {
+ files.add(status.getPath());
+ }
+ }
+ }
+
+ /**
+ * Assert the given HBase table has cell-identical rows (all versions) on
cluster 1
+ * ({@code conf1}) and cluster 2 ({@code conf2}). Dumps both tables on the
first mismatch before
+ * failing.
+ */
+ public static void assertTablesEqualAcrossClusters(Configuration conf1,
Configuration conf2,
+ String hbaseTableName) throws Exception {
+ TableName tn = TableName.valueOf(hbaseTableName);
+ try (Connection hconn1 = ConnectionFactory.createConnection(conf1);
+ Connection hconn2 = ConnectionFactory.createConnection(conf2);
+ Table table1 = hconn1.getTable(tn); Table table2 = hconn2.getTable(tn)) {
+
+ Scan scan = new Scan();
+ scan.readAllVersions();
+
+ try (ResultScanner scanner1 = table1.getScanner(scan);
+ ResultScanner scanner2 = table2.getScanner(scan)) {
+ int rowCount = 0;
+ while (true) {
+ Result r1 = scanner1.next();
+ Result r2 = scanner2.next();
+ if (r1 == null && r2 == null) {
+ break;
+ }
+ assertNotNull(
+ String.format("Table %s: cluster 2 has fewer rows at row %d",
hbaseTableName, rowCount),
+ r2);
+ assertNotNull(
+ String.format("Table %s: cluster 1 has fewer rows at row %d",
hbaseTableName, rowCount),
+ r1);
+ try {
+ Result.compareResults(r1, r2, true);
+ } catch (Exception e) {
+ LOG.error("Table {} row {} mismatch. Dumping both tables:",
hbaseTableName, rowCount);
+ LOG.error("--- Cluster 1 ---");
+ TestUtil.dumpTable(table1);
+ LOG.error("--- Cluster 2 ---");
+ TestUtil.dumpTable(table2);
+ fail(String.format("Table %s row %d mismatch: %s", hbaseTableName,
rowCount,
+ e.getMessage()));
+ }
+ rowCount++;
+ }
+ LOG.info("Table {} matches across clusters: {} rows verified",
hbaseTableName, rowCount);
+ }
+ }
+ }
+}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupBaseIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupBaseIT.java
index fa3bcc565f..5ac326789e 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupBaseIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupBaseIT.java
@@ -28,12 +28,10 @@ import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -41,7 +39,6 @@ import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
@@ -269,7 +266,7 @@ public abstract class ReplicationLogGroupBaseIT extends
HABaseIT {
// Replay replication log on cluster 2
FileSystem fs = standByLogDir.getFileSystem(conf2);
- List<Path> logFiles = findLogFiles(standByLogDir, fs);
+ List<Path> logFiles =
CrossClusterReplicationTestUtil.findLogFiles(standByLogDir, fs);
assertTrue("Should have at least one log file", !logFiles.isEmpty());
ReplicationLogProcessor processor = ReplicationLogProcessor.get(conf2,
haGroupName);
try {
@@ -359,70 +356,14 @@ public abstract class ReplicationLogGroupBaseIT extends
HABaseIT {
return patchVersion >= 9;
}
- protected List<Path> findLogFiles(Path dir, FileSystem fs) throws
IOException {
- List<Path> files = new ArrayList<>();
- findLogFilesRecursive(dir, fs, files);
- return files;
- }
-
- private void findLogFilesRecursive(Path dir, FileSystem fs, List<Path>
files) throws IOException {
- if (!fs.exists(dir)) {
- return;
- }
- for (FileStatus status : fs.listStatus(dir)) {
- if (status.isDirectory()) {
- findLogFilesRecursive(status.getPath(), fs, files);
- } else if (status.getPath().getName().endsWith(".plog")) {
- files.add(status.getPath());
- }
- }
- }
-
/**
* Compares an HBase table between the two clusters using {@link
Result#compareResults}. Scans
- * both tables with all versions and asserts that every row matches at the
cell level.
+ * both tables with all versions and asserts that every row matches at the
cell level. Delegates
+ * to the shared {@link
CrossClusterReplicationTestUtil#assertTablesEqualAcrossClusters} with this
+ * IT's two cluster configurations.
*/
protected void assertTablesEqualAcrossClusters(String hbaseTableName) throws
Exception {
- TableName tn = TableName.valueOf(hbaseTableName);
- try (
- org.apache.hadoop.hbase.client.Connection hconn1 =
ConnectionFactory.createConnection(conf1);
- org.apache.hadoop.hbase.client.Connection hconn2 =
ConnectionFactory.createConnection(conf2);
- Table table1 = hconn1.getTable(tn); Table table2 = hconn2.getTable(tn)) {
-
- Scan scan = new Scan();
- scan.readAllVersions();
-
- try (ResultScanner scanner1 = table1.getScanner(scan);
- ResultScanner scanner2 = table2.getScanner(scan)) {
- int rowCount = 0;
- while (true) {
- Result r1 = scanner1.next();
- Result r2 = scanner2.next();
- if (r1 == null && r2 == null) {
- break;
- }
- assertNotNull(
- String.format("Table %s: cluster 2 has fewer rows at row %d",
hbaseTableName, rowCount),
- r2);
- assertNotNull(
- String.format("Table %s: cluster 1 has fewer rows at row %d",
hbaseTableName, rowCount),
- r1);
- try {
- Result.compareResults(r1, r2, true);
- } catch (Exception e) {
- LOG.error("Table {} row {} mismatch. Dumping both tables:",
hbaseTableName, rowCount);
- LOG.error("--- Cluster 1 ---");
- TestUtil.dumpTable(table1);
- LOG.error("--- Cluster 2 ---");
- TestUtil.dumpTable(table2);
- fail(String.format("Table %s row %d mismatch: %s", hbaseTableName,
rowCount,
- e.getMessage()));
- }
- rowCount++;
- }
- LOG.info("Table {} matches across clusters: {} rows verified",
hbaseTableName, rowCount);
- }
- }
+ CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1,
conf2, hbaseTableName);
}
/**
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
index 23cbc0996b..525ebb400a 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java
@@ -1021,7 +1021,7 @@ public class ReplicationLogGroupIT extends
ReplicationLogGroupBaseIT {
// Replay the replication log on cluster 2
FileSystem fs = standByLogDir.getFileSystem(conf2);
- List<Path> logFiles = findLogFiles(standByLogDir, fs);
+ List<Path> logFiles =
CrossClusterReplicationTestUtil.findLogFiles(standByLogDir, fs);
LOG.info("Found {} log files to replay", logFiles.size());
assertTrue("Should have at least one log file", logFiles.size() > 0);
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java
new file mode 100644
index 0000000000..1b363c8b0e
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.replication;
+
+import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.STORE_AND_FORWARD;
+import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.SYNC;
+
+import java.io.IOException;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * Integration-test-only access shim for the package-visible mode controls on
+ * {@link ReplicationLogGroup}. Lets an IT in another package flip a live
writer SYNC ->
+ * STORE_AND_FORWARD deterministically, exercising the REAL {@link
StoreAndForwardModeImpl} (local
+ * 'out' log + forwarder + periodic ACTIVE_NOT_IN_SYNC persistence) without
having to injure a real
+ * peer HDFS to provoke the sync failure that triggers store-and-forward in
production. Lives in the
+ * test source tree, so no production code is modified.
+ */
+public final class ReplicationLogGroupTestAccess {
+
+ private ReplicationLogGroupTestAccess() {
+ }
+
+ /**
+ * Flip a live writer from SYNC to STORE_AND_FORWARD and fence the swap
through sync boundaries.
+ * The event handler reads the mode field in processPendingSyncs *after*
completing a sync's
+ * future, so we issue two syncs: the first lets the handler observe the new
mode and run
+ * StoreAndForwardModeImpl.onEnter (open the local 'out' log, start the
forwarder); the second
+ * runs entirely under the store-and-forward mode impl, guaranteeing any
subsequent append is
+ * buffered locally rather than written straight to the peer.
+ * @return true if the mode was SYNC and is now STORE_AND_FORWARD; false if
it was not SYNC.
+ */
+ public static boolean forceStoreAndForward(ReplicationLogGroup logGroup)
throws IOException {
+ boolean swapped = logGroup.checkAndSetMode(SYNC, STORE_AND_FORWARD);
+ logGroup.sync();
+ logGroup.sync();
+ return swapped;
+ }
+
+ /** True if the writer's current mode is STORE_AND_FORWARD. */
+ public static boolean isStoreAndForward(ReplicationLogGroup logGroup) {
+ return logGroup.getMode() == STORE_AND_FORWARD;
+ }
+
+ /**
+ * The peer (standby) 'in' directory this writer forwards to — the exact
path a replay instance on
+ * the peer cluster must read.
+ */
+ public static Path peerStandbyDir(ReplicationLogGroup logGroup) throws
IOException {
+ return logGroup.getOrCreatePeerShardManager().getRootDirectoryPath();
+ }
+}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java
index 7652db997b..98209280f4 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java
@@ -350,6 +350,32 @@ public class ReplicationLogDiscoveryReplayTestIT extends
HABaseIT {
false);
}
+ @Test
+ public void
testInitializeLastRoundProcessed_DegradedStateWithZeroLastSyncTime()
+ throws IOException {
+ long newFileTimestamp = 1704240060000L;
+ long lastSyncStateTime = 0L; // never synced / unset - must NOT rewind to
the epoch
+ long currentTime = 1704240900000L;
+ long roundTimeMills = 60000L; // 1 minute
+
+ // lastRoundProcessed uses the minimum of new files and current time (the
file frontier).
+ long expectedEndTime =
+ (newFileTimestamp / TimeUnit.MINUTES.toMillis(1)) *
TimeUnit.MINUTES.toMillis(1);
+ ReplicationRound expectedLastRoundProcessed =
+ new ReplicationRound(expectedEndTime - roundTimeMills, expectedEndTime);
+
+ // The zero-lastSyncStateTimeInMs guard is shared by the DEGRADED_STANDBY
and STANDBY_TO_ACTIVE
+ // init paths. On the DEGRADED path too, lastRoundInSync must collapse
onto the file frontier
+ // (== lastRoundProcessed) instead of ReplicationRound(0, 0), which would
rewind all the way to
+ // the epoch and drive the consistency point to 0 (retain-everything).
+ ReplicationRound expectedLastRoundInSync = expectedLastRoundProcessed;
+
+ testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime,
newFileTimestamp, null,
+ HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY,
expectedLastRoundProcessed,
+ expectedLastRoundInSync,
ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED,
+ false);
+ }
+
@Test
public void testInitializeLastRoundProcessed_SyncStateWithInProgressFiles()
throws IOException {
long currentTime = 1704412800000L;
@@ -440,8 +466,10 @@ public class ReplicationLogDiscoveryReplayTestIT extends
HABaseIT {
}
/**
- * Tests initializeLastRoundProcessed in STANDBY_TO_ACTIVE state. Validates
that failoverPending
- * is set to true when HA group state is STANDBY_TO_ACTIVE.
+ * Tests initializeLastRoundProcessed in STANDBY_TO_ACTIVE state with no
files. lastRoundInSync
+ * collapses onto lastRoundProcessed (nothing to rewind), so the replay
initializes directly to
+ * SYNC (not SYNCED_RECOVERY) and promotion is not delayed by a round
window. failoverPending is
+ * still armed because the HA group state is STANDBY_TO_ACTIVE.
*/
@Test
public void testInitializeLastRoundProcessed_StandbyToActiveState() throws
IOException {
@@ -464,6 +492,217 @@ public class ReplicationLogDiscoveryReplayTestIT extends
HABaseIT {
expectedLastRoundInSync,
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, true);
}
+ @Test
+ public void
testInitializeLastRoundProcessed_StandbyToActiveStateWithLastSyncStateAsMin()
+ throws IOException {
+ long newFileTimestamp = 1704240060000L;
+ long lastSyncStateTime = 1704240030000L;
+ long currentTime = 1704240900000L;
+ long roundTimeMills = 60000L; // 1 minute
+
+ // lastRoundProcessed uses the minimum of new files and current time (the
file frontier).
+ long expectedEndTime =
+ (newFileTimestamp / TimeUnit.MINUTES.toMillis(1)) *
TimeUnit.MINUTES.toMillis(1);
+ ReplicationRound expectedLastRoundProcessed =
+ new ReplicationRound(expectedEndTime - roundTimeMills, expectedEndTime);
+
+ // lastRoundInSync uses the minimum of lastSyncStateTime and file
timestamps, so on restart into
+ // STANDBY_TO_ACTIVE it stays one round BEHIND lastRoundProcessed (not
collapsed onto it) - this
+ // is the rewind target that guarantees zero RPO after a direct
DEGRADED_STANDBY transition.
+ long expectedSyncEndTime =
+ (lastSyncStateTime / TimeUnit.MINUTES.toMillis(1)) *
TimeUnit.MINUTES.toMillis(1);
+ ReplicationRound expectedLastRoundInSync =
+ new ReplicationRound(expectedSyncEndTime - roundTimeMills,
expectedSyncEndTime);
+
+ testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime,
newFileTimestamp, null,
+ HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE,
expectedLastRoundProcessed,
+ expectedLastRoundInSync,
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY,
+ true);
+ }
+
+ @Test
+ public void
testInitializeLastRoundProcessed_StandbyToActiveStateWithZeroLastSyncTime()
+ throws IOException {
+ long newFileTimestamp = 1704240060000L;
+ long lastSyncStateTime = 0L; // never synced / unset - must NOT rewind to
the epoch
+ long currentTime = 1704240900000L;
+ long roundTimeMills = 60000L; // 1 minute
+
+ // lastRoundProcessed uses the minimum of new files and current time (the
file frontier).
+ long expectedEndTime =
+ (newFileTimestamp / TimeUnit.MINUTES.toMillis(1)) *
TimeUnit.MINUTES.toMillis(1);
+ ReplicationRound expectedLastRoundProcessed =
+ new ReplicationRound(expectedEndTime - roundTimeMills, expectedEndTime);
+
+ // With lastSyncStateTimeInMs == 0 (unset), lastRoundInSync must collapse
onto the file frontier
+ // (== lastRoundProcessed) instead of ReplicationRound(0, 0), which would
rewind all the way to
+ // the epoch and drive the consistency point to 0 (retain-everything).
Because the two rounds
+ // are then equal (nothing to rewind), the replay initializes directly to
SYNC rather than
+ // SYNCED_RECOVERY, so promotion is not delayed by a round window.
+ ReplicationRound expectedLastRoundInSync = expectedLastRoundProcessed;
+
+ testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime,
newFileTimestamp, null,
+ HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE,
expectedLastRoundProcessed,
+ expectedLastRoundInSync,
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, true);
+ }
+
+ /**
+ * PHOENIX-7920 restart into STANDBY_TO_ACTIVE with nothing left to replay:
a fresh replay that
+ * reads a persisted STANDBY_TO_ACTIVE record (an RS bounced mid-failover
after the data was
+ * already replayed) must promote on the FIRST replay() tick. With no files,
lastRoundInSync
+ * collapses onto lastRoundProcessed, so init picks SYNC (not
SYNCED_RECOVERY), failoverPending is
+ * armed, and shouldTriggerFailover() passes immediately. Regression guard
for the init-branch
+ * latency fix: on the earlier unconditional-SYNCED_RECOVERY init, the SYNC
state gate would block
+ * promotion until a round became time-eligible (~one round window later),
so triggerFailover
+ * would not fire on this first tick.
+ */
+ @Test
+ public void
testReplay_RestartIntoStandbyToActive_NothingToReplay_PromotesOnFirstTick()
+ throws IOException {
+ long currentTime = 1704153600000L;
+
+ TestableReplicationLogTracker fileTracker =
+ createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri);
+ fileTracker.init();
+ try {
+ // Persisted STANDBY_TO_ACTIVE with no in / in-progress files: models an
RS that restarted
+ // after a direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition once
the data had already
+ // been replayed.
+ HAGroupStoreRecord record =
+ new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION,
haGroupName,
+ HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, currentTime,
+ HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl,
CLUSTERS.getMasterAddress1(),
+ CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(),
CLUSTERS.getHdfsUrl2(), 0L);
+
+ EnvironmentEdge edge = () -> currentTime;
+ EnvironmentEdgeManager.injectEdge(edge);
+ try {
+ TestableReplicationLogDiscoveryReplay discovery =
+ new TestableReplicationLogDiscoveryReplay(fileTracker, record);
+ // init() subscribes the LOCAL listeners (none fire here: this group's
LOCAL state is
+ // ACTIVE,
+ // not one of the subscribed failover/degrade targets, and the record
is never transitioned
+ // in this test), runs initializeLastRoundProcessed() over the
injected STANDBY_TO_ACTIVE
+ // record, and wires up the metrics source that replay() needs.
+ discovery.init();
+
+ // Nothing to rewind -> init directly to SYNC with failoverPending
armed.
+ assertEquals("no-rewind restart into STANDBY_TO_ACTIVE must init to
SYNC",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+ assertTrue("failoverPending must be armed on a STANDBY_TO_ACTIVE
restart",
+ discovery.getFailoverPending());
+
+ // First replay() tick must promote (shouldTriggerFailover passes
since state is SYNC).
+ discovery.replay();
+ assertEquals("promotion must fire on the first replay tick", 1,
+ discovery.getTriggerFailoverCallCount());
+ } finally {
+ EnvironmentEdgeManager.reset();
+ }
+ } finally {
+ fileTracker.close();
+ }
+ }
+
+ /**
+ * PHOENIX-7920 init-window race regression: a direct DEGRADED_STANDBY ->
STANDBY_TO_ACTIVE
+ * flip can fire triggerFailoverListner while initializeLastRoundProcessed()
still holds
+ * NOT_INITIALIZED -- its compareAndSet(DEGRADED, SYNCED_RECOVERY) no-ops
there and only
+ * failoverPending survives. The DEGRADED_STANDBY branch's
compareAndSet(NOT_INITIALIZED,
+ * DEGRADED) then lands, leaving state DEGRADED with a failover pending;
nothing re-fires (the
+ * record is already STANDBY_TO_ACTIVE and a direct failover never revisits
STANDBY), so
+ * shouldTriggerFailover() -- gated on SYNC -- would never promote and the
failover would stay
+ * silently stuck until an RS restart. The init-window reconcile must finish
the promotion the
+ * listener intended: observe the pending failover from DEGRADED and CAS to
SYNCED_RECOVERY.
+ * <p>
+ * Reproduced deterministically (no ZooKeeper, no real listeners): the
overridden
+ * getHAGroupRecord() arms failoverPending -- the surviving effect of the
listener firing at
+ * NOT_INITIALIZED -- as it returns the stale DEGRADED_STANDBY record,
landing the arm before the
+ * DEGRADED-branch CAS exactly as the real interleaving does. Before the
reconcile was added this
+ * test fails: state stays DEGRADED with a pending failover that never
promotes.
+ */
+ @Test
+ public void
testReplay_InitWindowRace_DegradedFlipToStandbyToActive_Reconciles()
+ throws IOException {
+ // Same file/sync-point layout as
+ // testInitializeLastRoundProcessed_DegradedStateWithLastSyncStateAsMin:
+ // a new file at the frontier with lastSyncStateTime one round earlier, so
lastRoundInSync
+ // initializes one round BEHIND lastRoundProcessed -- genuine rewind work
that SYNCED_RECOVERY
+ // must cover (the round forwarded-but-not-synced while degraded).
+ long newFileTimestamp = 1704240060000L;
+ long lastSyncStateTime = 1704240030000L;
+ long currentTime = 1704240900000L;
+ long roundTimeMills = 60000L; // 1 minute
+
+ long expectedProcEndTime =
+ (newFileTimestamp / TimeUnit.MINUTES.toMillis(1)) *
TimeUnit.MINUTES.toMillis(1);
+ ReplicationRound expectedLastRoundProcessed =
+ new ReplicationRound(expectedProcEndTime - roundTimeMills,
expectedProcEndTime);
+ long expectedSyncEndTime =
+ (lastSyncStateTime / TimeUnit.MINUTES.toMillis(1)) *
TimeUnit.MINUTES.toMillis(1);
+ ReplicationRound expectedLastRoundInSync =
+ new ReplicationRound(expectedSyncEndTime - roundTimeMills,
expectedSyncEndTime);
+
+ TestableReplicationLogTracker fileTracker =
+ createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri);
+ fileTracker.init();
+ try {
+ // New file at the frontier so lastRoundProcessed lands on it.
+ ReplicationRound newFileRound =
+ new ReplicationRound(newFileTimestamp - roundTimeMills,
newFileTimestamp);
+ Path shardPath = fileTracker.getReplicationShardDirectoryManager()
+ .getShardDirectory(newFileRound.getStartTime());
+ rootFs.mkdirs(shardPath);
+ Path newFile = new Path(shardPath, newFileTimestamp + "_rs-1.plog");
+ rootFs.create(newFile, true).close();
+
+ // Stale DEGRADED_STANDBY record (recordTime carries lastSyncStateTime,
the last good sync
+ // point) -- what getHAGroupRecord() reads before the LOCAL record flips
to STANDBY_TO_ACTIVE.
+ HAGroupStoreRecord degradedRecord =
+ new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION,
haGroupName,
+ HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY, lastSyncStateTime,
+ HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl,
CLUSTERS.getMasterAddress1(),
+ CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(),
CLUSTERS.getHdfsUrl2(), 0L);
+
+ EnvironmentEdge edge = () -> currentTime;
+ EnvironmentEdgeManager.injectEdge(edge);
+ try {
+ TestableReplicationLogDiscoveryReplay discovery =
+ new TestableReplicationLogDiscoveryReplay(fileTracker,
degradedRecord) {
+ @Override
+ protected HAGroupStoreRecord getHAGroupRecord() throws IOException
{
+ // Arm failoverPending as the stale record is read: the
surviving effect of
+ // triggerFailoverListner firing at NOT_INITIALIZED (its CAS
no-ops there). This
+ // lands the arm before the DEGRADED-branch CAS, as the real
interleaving does.
+ setFailoverPending(true);
+ return super.getHAGroupRecord();
+ }
+ };
+
+ discovery.initializeLastRoundProcessed();
+
+ // The reconcile must finish the promotion: DEGRADED + pending
failover -> SYNCED_RECOVERY.
+ assertEquals(
+ "init-window race must reconcile DEGRADED + pending failover to
SYNCED_RECOVERY",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY,
+ discovery.getReplicationReplayState());
+ assertTrue("failoverPending must remain armed through the reconcile",
+ discovery.getFailoverPending());
+ // Rewind target preserved: lastRoundInSync stays one round behind
lastRoundProcessed, so
+ // SYNCED_RECOVERY has genuine work (re-sync the
forwarded-but-unsynced round).
+ assertEquals("lastRoundProcessed must land on the file frontier",
+ expectedLastRoundProcessed, discovery.getLastRoundProcessed());
+ assertEquals("lastRoundInSync must stay one round behind (the rewind
target)",
+ expectedLastRoundInSync, discovery.getLastRoundInSync());
+ } finally {
+ EnvironmentEdgeManager.reset();
+ }
+ } finally {
+ fileTracker.close();
+ }
+ }
+
/**
* Helper method to test initializeLastRoundProcessed with various file and
state configurations.
* Handles file creation, state setup, and validation of lastRoundProcessed
and lastRoundInSync.
@@ -1319,6 +1558,277 @@ public class ReplicationLogDiscoveryReplayTestIT
extends HABaseIT {
}
}
+ /**
+ * PHOENIX-7920: drives the real LOCAL listeners through the DIRECT
transition DEGRADED_STANDBY
+ * -> STANDBY_TO_ACTIVE (skipping STANDBY) via ZooKeeper, asserting that
triggerFailoverListner
+ * promotes the replay state DEGRADED -> SYNCED_RECOVERY (so replay()
will rewind) in addition
+ * to arming failoverPending. Before the fix the state stayed DEGRADED and
failover was blocked
+ * forever.
+ */
+ @Test
+ public void testReplay_RuntimeListeners_DirectDegradedToStandbyToActive()
throws Exception {
+ PhoenixHAAdmin haAdmin = CLUSTERS.getHaAdmin1();
+ // Base state: local STANDBY, written before init() so the discovery's
client picks it up.
+ writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY);
+
+ TestableReplicationLogTracker fileTracker =
+ createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri);
+ fileTracker.init();
+ try {
+ // Null injected record: read the real effective record and subscribe
the real LOCAL
+ // listeners to HAGroupStoreManager.
+ TestableReplicationLogDiscoveryReplay discovery =
+ new TestableReplicationLogDiscoveryReplay(fileTracker, null);
+ discovery.init();
+ // Let the initial STANDBY load settle so it cannot race the degrade
below.
+ Thread.sleep(5000L);
+
+ // degrade: LOCAL -> DEGRADED_STANDBY must drive the degradedListener to
DEGRADED.
+ writeLocalRecord(haAdmin,
HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY);
+ awaitCondition(
+ () -> discovery.getReplicationReplayState()
+ == ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED,
+ "degradedListener should drive replay state to DEGRADED");
+
+ // direct failover: LOCAL -> STANDBY_TO_ACTIVE (no STANDBY hop) must
both arm failoverPending
+ // AND move DEGRADED -> SYNCED_RECOVERY so a rewind is scheduled.
+ writeLocalRecord(haAdmin,
HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE);
+ awaitCondition(
+ () -> discovery.getReplicationReplayState()
+ ==
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY,
+ "direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE must move replay state
to SYNCED_RECOVERY");
+ assertTrue("failoverPending must be set after STANDBY_TO_ACTIVE",
+ discovery.getFailoverPending());
+ } finally {
+ fileTracker.close();
+ haAdmin.getCurator().delete().quietly().forPath(toPath(haGroupName));
+ }
+ }
+
+ /**
+ * PHOENIX-7920: the healthy failover path STANDBY -> STANDBY_TO_ACTIVE
(no prior degrade) must
+ * NOT change the replay state. init from STANDBY leaves state SYNC, and the
trigger listener's
+ * compareAndSet(DEGRADED, SYNCED_RECOVERY) is a no-op from SYNC, arming
only failoverPending.
+ * This guards against a naive unconditional set(SYNCED_RECOVERY) that would
force a needless
+ * rewind on every healthy failover.
+ */
+ @Test
+ public void testReplay_RuntimeListeners_HealthyStandbyToActiveKeepsSync()
throws Exception {
+ PhoenixHAAdmin haAdmin = CLUSTERS.getHaAdmin1();
+ writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY);
+
+ TestableReplicationLogTracker fileTracker =
+ createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri);
+ fileTracker.init();
+ try {
+ TestableReplicationLogDiscoveryReplay discovery =
+ new TestableReplicationLogDiscoveryReplay(fileTracker, null);
+ discovery.init();
+ // Let the initial STANDBY load settle so init has resolved to SYNC
before we proceed.
+ Thread.sleep(5000L);
+ assertEquals("init from STANDBY should leave replay state SYNC",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+
+ // healthy failover: LOCAL -> STANDBY_TO_ACTIVE from SYNC must arm
failoverPending without
+ // changing the replay state. triggerFailoverListner arms
failoverPending first, then does
+ // compareAndSet(DEGRADED, SYNCED_RECOVERY) -- a no-op from SYNC. So the
state stays SYNC
+ // before, during, and after the callback no matter where the CAS lands
relative to our
+ // read; the "state unchanged" assertion below is order-independent, not
a race on a sleep.
+ writeLocalRecord(haAdmin,
HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE);
+ awaitCondition(discovery::getFailoverPending,
+ "triggerFailoverListner should set failoverPending");
+ // The CAS is a no-op from SYNC, so the state must be SYNC here (not
SYNCED_RECOVERY). A
+ // (wrong) unconditional set(SYNCED_RECOVERY) would already be visible
alongside
+ // failoverPending.
+ assertEquals("STANDBY_TO_ACTIVE from SYNC must not change replay state",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+ // Settle briefly and re-check both invariants to catch a late or
redelivered event wrongly
+ // flipping the state or clearing the pending flag.
+ Thread.sleep(500L);
+ assertTrue("failoverPending must remain armed after settling",
+ discovery.getFailoverPending());
+ assertEquals("replay state must remain SYNC after settling",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+ } finally {
+ fileTracker.close();
+ haAdmin.getCurator().delete().quietly().forPath(toPath(haGroupName));
+ }
+ }
+
+ /**
+ * PHOENIX-7920 (C2 regression guard): recoveryListener uses
compareAndSet(DEGRADED,
+ * SYNCED_RECOVERY), NOT an unconditional set(). This drives the two
behaviours that distinguish
+ * the CAS from a naive set through the real LOCAL listeners over ZooKeeper.
+ * <p>
+ * Phase 1 (must NOT over-fire): a LOCAL -> STANDBY event that arrives
while the replay state
+ * is already SYNC must leave it SYNC. STANDBY is reached without ever
passing through
+ * DEGRADED_STANDBY, via an aborted-failover round trip (STANDBY ->
STANDBY_TO_ACTIVE ->
+ * ABORT_TO_STANDBY -> STANDBY). recoveryListener has no observable side
effect from SYNC, so
+ * determinism is manufactured in two steps: first we wait for the client
cache to surface STANDBY
+ * (proving that event was delivered, not coalesced into the next write);
then a trailing
+ * STANDBY_TO_ACTIVE acts as a barrier - its failoverPending write runs on
the same single
+ * cache-dispatcher thread strictly after the STANDBY event's
recoveryListener callback, so once
+ * we observe failoverPending that callback has already completed. Its
compareAndSet(DEGRADED,
+ * SYNCED_RECOVERY) is a no-op from SYNC (and from SYNCED_RECOVERY), so it
does not disturb
+ * whatever recoveryListener left. With a wrong unconditional set() the
state would read
+ * SYNCED_RECOVERY here.
+ * <p>
+ * Phase 2 (must still fire from DEGRADED): DEGRADED_STANDBY -> STANDBY
must reach
+ * SYNCED_RECOVERY, proving the CAS still schedules the rewind on a genuine
recovery.
+ */
+ @Test
+ public void
testReplay_RuntimeListeners_SpuriousStandbyKeepsSyncButDegradedRecovers()
+ throws Exception {
+ PhoenixHAAdmin haAdmin = CLUSTERS.getHaAdmin1();
+ // Base state: local STANDBY, written before init() so the discovery's
client picks it up.
+ writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY);
+
+ TestableReplicationLogTracker fileTracker =
+ createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri);
+ fileTracker.init();
+ try {
+ TestableReplicationLogDiscoveryReplay discovery =
+ new TestableReplicationLogDiscoveryReplay(fileTracker, null);
+ discovery.init();
+ // Let the initial STANDBY load settle so init has resolved to SYNC
before we proceed.
+ Thread.sleep(5000L);
+ assertEquals("init from STANDBY should leave replay state SYNC",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+
+ // ---- Phase 1: spurious LOCAL -> STANDBY while SYNC must keep SYNC ----
+ // Aborted-failover round trip that never touches DEGRADED_STANDBY. Each
step before the
+ // spurious STANDBY is gated on its own observable effect so the cache
cannot coalesce events.
+ writeLocalRecord(haAdmin,
HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE);
+ awaitCondition(discovery::getFailoverPending,
+ "triggerFailoverListner should set failoverPending");
+ writeLocalRecord(haAdmin,
HAGroupStoreRecord.HAGroupState.ABORT_TO_STANDBY);
+ awaitCondition(() -> !discovery.getFailoverPending(),
+ "abortFailoverListner should clear failoverPending");
+
+ // The spurious event. Confirm the client cache actually surfaced
STANDBY (so it was not
+ // coalesced into the barrier write below) before proceeding.
+ writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY);
+ awaitCondition(() ->
effectiveLocalStateIs(HAGroupStoreRecord.HAGroupState.STANDBY),
+ "client cache should surface the spurious LOCAL STANDBY");
+
+ // Barrier: a trailing STANDBY_TO_ACTIVE. Observing its failoverPending
guarantees the STANDBY
+ // event's recoveryListener callback (dispatched earlier on the same
thread) has completed.
+ writeLocalRecord(haAdmin,
HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE);
+ awaitCondition(discovery::getFailoverPending,
+ "trailing STANDBY_TO_ACTIVE should re-arm failoverPending
(recovery-callback barrier)");
+ assertEquals(
+ "spurious LOCAL -> STANDBY while SYNC must not flip replay state to
SYNCED_RECOVERY",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+
+ // ---- Phase 2: genuine DEGRADED_STANDBY -> STANDBY must reach
SYNCED_RECOVERY ----
+ // Return to a plain STANDBY, then degrade, then recover. Each hop is
gated on its effect.
+ writeLocalRecord(haAdmin,
HAGroupStoreRecord.HAGroupState.ABORT_TO_STANDBY);
+ awaitCondition(() -> !discovery.getFailoverPending(),
+ "abortFailoverListner should clear failoverPending again");
+ writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY);
+ awaitCondition(() ->
effectiveLocalStateIs(HAGroupStoreRecord.HAGroupState.STANDBY),
+ "client cache should surface STANDBY before degrade");
+ writeLocalRecord(haAdmin,
HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY);
+ awaitCondition(
+ () -> discovery.getReplicationReplayState()
+ == ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED,
+ "degradedListener should drive replay state to DEGRADED");
+ writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY);
+ awaitCondition(
+ () -> discovery.getReplicationReplayState()
+ ==
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY,
+ "recoveryListener CAS must drive DEGRADED -> SYNCED_RECOVERY on a
genuine recovery");
+ } finally {
+ fileTracker.close();
+ haAdmin.getCurator().delete().quietly().forPath(toPath(haGroupName));
+ }
+ }
+
+ /**
+ * PHOENIX-7920 direct-path outcome: with a failover pending during
degradation, once the rewind
+ * (SYNCED_RECOVERY) completes and replay returns to SYNC with everything
caught up, failover is
+ * triggered exactly once. Complements
testReplay_StateTransition_DegradedToAbortToRecover (the
+ * aborted case, where triggerFailover must NOT fire) and
testReplay_StateTransition_
+ * DegradedToSyncedRecovery (rewind with no pending failover, where it also
must NOT fire).
+ */
+ @Test
+ public void testReplay_DirectFailover_RewindsThenTriggersFailoverOnce()
throws IOException {
+ TestableReplicationLogTracker fileTracker =
+ createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri);
+
+ try {
+ long initialEndTime = 1704585600000L; // 2024-01-07 00:00:00
+ long roundTimeMills =
+
fileTracker.getReplicationShardDirectoryManager().getReplicationRoundDurationSeconds()
+ * 1000L;
+ long bufferMillis = (long) (roundTimeMills * 0.15);
+
+ // DEGRADED record (peer-blind degraded standby).
+ HAGroupStoreRecord mockRecord =
+ new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION,
haGroupName,
+ HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY, initialEndTime,
+ HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl,
CLUSTERS.getMasterAddress1(),
+ CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(),
CLUSTERS.getHdfsUrl2(), 0L);
+
+ // Allow processing up to 5 rounds.
+ long currentTime = initialEndTime + (5 * roundTimeMills) + bufferMillis;
+ EnvironmentEdge edge = () -> currentTime;
+ EnvironmentEdgeManager.injectEdge(edge);
+
+ try {
+ TestableReplicationLogDiscoveryReplay discovery =
+ new TestableReplicationLogDiscoveryReplay(fileTracker, mockRecord);
+ discovery.init();
+
+ // Large rewind span: lastRoundProcessed is well ahead of the frozen
lastRoundInSync.
+ ReplicationRound lastInSyncRound =
+ new ReplicationRound(initialEndTime - roundTimeMills,
initialEndTime);
+ discovery.setLastRoundProcessed(new ReplicationRound(initialEndTime +
roundTimeMills,
+ initialEndTime + (2 * roundTimeMills)));
+ discovery.setLastRoundInSync(lastInSyncRound);
+ discovery
+
.setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED);
+ // Failover requested during degradation, then the direct transition
schedules the rewind
+ // (SYNCED_RECOVERY) after round 2. failoverPending is never cleared -
it must survive.
+ discovery.setFailoverPending(true);
+ discovery.setStateChangeAfterRounds(2,
+
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY);
+
+ discovery.replay();
+
+ // 2 rounds in DEGRADED, rewind to lastRoundInSync, then 5 rounds
re-replayed in SYNC.
+ assertEquals("processRound should be called 7 times", 7,
+ discovery.getProcessRoundCallCount());
+
+ // Recovery completed: back to SYNC, both pointers caught up to the
final round.
+ assertEquals("State should be SYNC after recovery",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+ ReplicationRound expectedFinalRound = new ReplicationRound(
+ initialEndTime + (4 * roundTimeMills), initialEndTime + (5 *
roundTimeMills));
+ assertEquals("Last round processed should be the final round",
expectedFinalRound,
+ discovery.getLastRoundProcessed());
+ assertEquals("Last round in sync should match last round processed
after recovery",
+ expectedFinalRound, discovery.getLastRoundInSync());
+
+ // The pending failover survived the rewind and fired exactly once,
from SYNC.
+ assertEquals("Failover should be triggered exactly once after rewind
completes", 1,
+ discovery.getTriggerFailoverCallCount());
+ assertFalse("failoverPending should be cleared after failover is
triggered",
+ discovery.getFailoverPending());
+ } finally {
+ EnvironmentEdgeManager.reset();
+ }
+ } finally {
+ fileTracker.close();
+ }
+ }
+
/** Write (create, or version-checked update) the LOCAL HA record for this
group on ZooKeeper. */
private void writeLocalRecord(PhoenixHAAdmin haAdmin,
HAGroupStoreRecord.HAGroupState state)
throws Exception {
@@ -1337,6 +1847,17 @@ public class ReplicationLogDiscoveryReplayTestIT extends
HABaseIT {
}
}
+ /** True if the discovery's LOCAL client cache currently surfaces {@code
state} as effective. */
+ private boolean effectiveLocalStateIs(HAGroupStoreRecord.HAGroupState state)
{
+ try {
+ Optional<HAGroupStoreRecord> rec =
+
HAGroupStoreManager.getInstance(conf1).getEffectiveHAGroupStoreRecord(haGroupName);
+ return rec.isPresent() && rec.get().getHAGroupState() == state;
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
/** Poll until the condition holds or a fixed timeout elapses, then assert
it. */
private static void awaitCondition(java.util.function.BooleanSupplier
condition, String message)
throws InterruptedException {
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java
new file mode 100644
index 0000000000..b03a018f6d
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java
@@ -0,0 +1,395 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.replication.reader;
+
+import static
org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR;
+import static
org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.getHighAvailibilityGroup;
+import static org.apache.phoenix.query.BaseTest.generateUniqueName;
+import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY;
+import static
org.apache.phoenix.replication.reader.ReplicationLogReplayService.PHOENIX_REPLICATION_REPLAY_ENABLED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.function.BooleanSupplier;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.jdbc.FailoverPhoenixConnection;
+import org.apache.phoenix.jdbc.HABaseIT;
+import org.apache.phoenix.jdbc.HAGroupStoreClient;
+import org.apache.phoenix.jdbc.HAGroupStoreManager;
+import org.apache.phoenix.jdbc.HAGroupStoreRecord;
+import org.apache.phoenix.jdbc.HighAvailabilityGroup;
+import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
+import org.apache.phoenix.jdbc.HighAvailabilityTestingUtility;
+import org.apache.phoenix.replication.CrossClusterReplicationTestUtil;
+import org.apache.phoenix.replication.ReplicationLogGroup;
+import org.apache.phoenix.replication.ReplicationLogGroupTestAccess;
+import org.apache.phoenix.replication.ReplicationLogTracker;
+import org.apache.phoenix.replication.ReplicationShardDirectoryManager;
+import
org.apache.phoenix.replication.metrics.MetricsReplicationLogTrackerReplayImpl;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * End-to-end zero-RPO test for a direct DEGRADED_STANDBY ->
STANDBY_TO_ACTIVE failover in
+ * store-and-forward mode (PHOENIX-7920). Cluster1 runs the real writer;
Cluster2 runs a
+ * manually-driven real ReplicationLogDiscoveryReplay (the auto-scheduler is
disabled).
+ */
+@Category(NeedsOwnMiniClusterTest.class)
+public class StoreAndForwardFailoverIT extends HABaseIT {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(StoreAndForwardFailoverIT.class);
+
+ private static final int A_START = 0;
+ private static final int A_COUNT = 50;
+ private static final int B_START = 50;
+ private static final int B_COUNT = 50;
+
+ // Cluster2's peer-aware reaction to Cluster1's ACTIVE_NOT_IN_SYNC drives
its LOCAL record to
+ // DEGRADED_STANDBY asynchronously; bound how long transitionCluster2 waits
for that reaction to
+ // land (and how often it polls) before attempting its own transition.
+ private static final long PEER_REACTION_SETTLE_MILLIS = 10000L;
+ private static final long PEER_REACTION_POLL_MILLIS = 500L;
+
+ @Rule
+ public TestName name = new TestName();
+
+ private String haGroupName;
+ private Properties clientProps;
+ private HighAvailabilityGroup haGroup;
+ private ReplicationLogGroup logGroup;
+ private ReplicationLogDiscoveryReplay discovery;
+ private ReplicationLogTracker replayTracker;
+ private FileSystem standbyFs;
+ private Path standbyInDir;
+ private String tableName;
+ private long syncPointAfterA;
+ private int logFileCountAfterA;
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ // Short rounds so replay-round eligibility is a few seconds, not the 60s
production default.
+ conf1.setInt(PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY, 2);
+ conf2.setInt(PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY, 2);
+ // Disable the auto replay scheduler so our manually-driven instance is
the only replayer.
+ // (HABaseIT.doBaseSetup enables it; the writer/forwarder are gated on
SYNCHRONOUS_REPLICATION,
+ // not on this flag, so they are unaffected.)
+ // This flag has a second consumer: CompactionScanner also gates the
replication
+ // consistency-point compaction guard on it, so setting it false disables
that guard during
+ // major compactions too. Both effects share this one flag and cannot be
separated. It is
+ // immaterial here: this test triggers no major compaction and asserts
cross-cluster
+ // cell-equality directly, so the guard's absence does not affect what it
verifies.
+ conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false);
+ conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false);
+ CLUSTERS.start();
+ }
+
+ @Before
+ public void beforeTest() throws Exception {
+ LOG.info("Starting test {}", name.getMethodName());
+ haGroupName = name.getMethodName();
+ clientProps = HighAvailabilityTestingUtility.getHATestProperties();
+ clientProps.setProperty(PHOENIX_HA_GROUP_ATTR, haGroupName);
+ // Cluster1 = ACTIVE_IN_SYNC, Cluster2 = STANDBY, with hdfs urls populated
on both records.
+ CLUSTERS.initClusterRole(haGroupName, HighAvailabilityPolicy.FAILOVER);
+ haGroup = getHighAvailibilityGroup(CLUSTERS.getJdbcHAUrl(), clientProps);
+
+ // The live Cluster1 writer (RS0). Same singleton the RS write path uses,
so flipping its mode
+ // affects real writes. Cluster1 is ACTIVE_IN_SYNC => this starts in SYNC
mode.
+ HRegionServer rs =
CLUSTERS.getHBaseCluster1().getHBaseCluster().getRegionServer(0);
+ logGroup = ReplicationLogGroup.get(conf1, rs.getServerName(), haGroupName);
+
+ // Replicated table (REPLICATION_SCOPE=1 + LOCAL INDEX) on BOTH clusters.
+ tableName = "T_" + generateUniqueName();
+ CLUSTERS.createTableOnClusterPair(haGroup, tableName);
+
+ // Ensure Cluster2 sees its peer, so its effective HA record is STANDBY
(not the peer-blind
+ // DEGRADED_STANDBY). This guarantees the replay initializes in SYNC
(else-branch).
+ HAGroupStoreManager c2Manager = HAGroupStoreManager.getInstance(conf2);
+ awaitCondition(() -> {
+ try {
+ Optional<HAGroupStoreRecord> eff =
c2Manager.getEffectiveHAGroupStoreRecord(haGroupName);
+ return eff.isPresent()
+ && eff.get().getHAGroupState() ==
HAGroupStoreRecord.HAGroupState.STANDBY;
+ } catch (IOException e) {
+ return false;
+ }
+ }, 60000L, "cluster2 effective state should settle to STANDBY before
constructing replay");
+
+ // Build the REAL replay over the exact standby 'in' directory the writer
forwards to, while
+ // Cluster2 is still STANDBY, so init() subscribes its LOCAL listeners
BEFORE any degrade /
+ // failover write and starts in SYNC.
+ standbyInDir = ReplicationLogGroupTestAccess.peerStandbyDir(logGroup);
+ standbyFs = standbyInDir.getFileSystem(conf2);
+ ReplicationShardDirectoryManager shardMgr =
+ new ReplicationShardDirectoryManager(conf2, standbyFs, standbyInDir);
+ replayTracker = new ReplicationLogTracker(conf2, haGroupName, shardMgr,
+ new MetricsReplicationLogTrackerReplayImpl(haGroupName));
+ replayTracker.init();
+ discovery = new ReplicationLogDiscoveryReplay(replayTracker);
+ discovery.init();
+ }
+
+ @After
+ public void afterTest() throws Exception {
+ LOG.info("Cleaning up test {}", name.getMethodName());
+ if (replayTracker != null) {
+ replayTracker.close();
+ }
+ if (logGroup != null) {
+ logGroup.close();
+ }
+ }
+
+ @Test
+ public void testDirectFailoverInStoreAndForwardModeIsZeroRPO() throws
Exception {
+ // Stage 0: the replay initialized in SYNC while Cluster2 is STANDBY.
+ assertEquals("replay must initialize in SYNC while cluster2 is STANDBY",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+ // Stage 1: SYNC batch A replicates to cluster2 and advances the sync
point.
+ stageWriteAndReplayBatchA();
+ // Stage 2: enter store-and-forward, write batch B, await it forwarded to
cluster2 'in'.
+ stageEnterStoreAndForwardAndForwardBatchB();
+ // Stage 3: degrade, direct failover, drive replay to promotion, assert
zero RPO.
+ stageDegradeDirectFailoverAndAssertZeroRPO();
+ }
+
+ /** Poll until {@code condition} holds or {@code timeoutMs} elapses, then
assert it. */
+ private static void awaitCondition(BooleanSupplier condition, long
timeoutMs, String message)
+ throws InterruptedException {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (!condition.getAsBoolean() && System.currentTimeMillis() < deadline)
{
+ Thread.sleep(250L);
+ }
+ assertTrue(message, condition.getAsBoolean());
+ }
+
+ /**
+ * Repeatedly call {@code discovery.replay()} (short real-clock rounds)
until {@code done} holds
+ * or {@code timeoutMs} elapses, then assert {@code done}. This is how the
test advances the
+ * manually driven replayer in the absence of the auto-scheduler.
+ */
+ private void driveReplayUntil(BooleanSupplier done, long timeoutMs, String
message)
+ throws Exception {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (System.currentTimeMillis() < deadline) {
+ discovery.replay();
+ if (done.getAsBoolean()) {
+ return;
+ }
+ Thread.sleep(500L);
+ }
+ discovery.replay();
+ assertTrue(message, done.getAsBoolean());
+ }
+
+ /**
+ * Upsert {@code count} rows [startId, startId+count) into the replicated
table via the HA URL.
+ */
+ private void upsertRows(int startId, int count) throws SQLException {
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " +
tableName + " VALUES (?, ?)");
+ for (int i = 0; i < count; i++) {
+ stmt.setInt(1, startId + i);
+ stmt.setInt(2, startId + i);
+ stmt.executeUpdate();
+ }
+ conn.commit();
+ }
+ }
+
+ /**
+ * Stage 1: write batch A on Cluster1 in SYNC mode (.plog goes straight to
Cluster2's 'in' dir),
+ * then drive replay until every A row is present on Cluster2 and
lastRoundInSync has advanced to
+ * cover A. Records the sync point so Task 5 can assert it later advances to
cover B.
+ */
+ private void stageWriteAndReplayBatchA() throws Exception {
+ upsertRows(A_START, A_COUNT);
+
+ // Drive replay until A has been applied to Cluster2 and the sync point
advanced past epoch.
+ driveReplayUntil(() -> {
+ try {
+ CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1,
conf2, tableName);
+ return discovery.getLastRoundInSync() != null
+ && discovery.getLastRoundInSync().getEndTime() > 0L;
+ } catch (AssertionError notYet) {
+ return false;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 90000L, "batch A must replicate to cluster2 and advance
lastRoundInSync");
+
+ syncPointAfterA = discovery.getLastRoundInSync().getEndTime();
+ assertTrue("lastRoundInSync must be set after batch A", syncPointAfterA >
0L);
+ LOG.info("Stage 1 complete: batch A in sync, syncPointAfterA={}",
syncPointAfterA);
+ }
+
+ /**
+ * Stage 2: flip the live Cluster1 writer SYNC -> STORE_AND_FORWARD (real
+ * StoreAndForwardModeImpl onEnter: local 'out' log + forwarder + periodic
ACTIVE_NOT_IN_SYNC
+ * persistence on Cluster1), write batch B (buffered to Cluster1 'out'), and
wait for the real
+ * forwarder to copy B's .plog files into Cluster2's 'in' directory. Does
NOT drive replay here,
+ * so B stays unreplayed.
+ */
+ private void stageEnterStoreAndForwardAndForwardBatchB() throws Exception {
+ logFileCountAfterA =
+ CrossClusterReplicationTestUtil.findLogFiles(standbyInDir,
standbyFs).size();
+
+ boolean swapped =
ReplicationLogGroupTestAccess.forceStoreAndForward(logGroup);
+ assertTrue("writer must have been in SYNC and flipped to
STORE_AND_FORWARD", swapped);
+ assertTrue("writer must now report STORE_AND_FORWARD",
+ ReplicationLogGroupTestAccess.isStoreAndForward(logGroup));
+
+ upsertRows(B_START, B_COUNT);
+
+ // The forwarder copies out->in on its own executor; poll (do not race)
for B's new files.
+ awaitCondition(() -> {
+ try {
+ return CrossClusterReplicationTestUtil.findLogFiles(standbyInDir,
standbyFs).size()
+ > logFileCountAfterA;
+ } catch (IOException e) {
+ return false;
+ }
+ }, 60000L, "forwarder must copy batch B's .plog files into cluster2 'in'");
+ LOG.info("Stage 2 complete: store-and-forward active, batch B forwarded to
cluster2 'in'");
+ }
+
+ /**
+ * Drive Cluster2's LOCAL HA record to {@code target} through the same
cached client the real
+ * triggerFailover uses. setHAGroupStatusIfNeeded returns a positive
dwell-time when the
+ * transition is throttled; retry until it applies (returns 0) or the
deadline elapses.
+ * <p>
+ * Cluster2 is peer-aware: when Cluster1 persists {@code ACTIVE_NOT_IN_SYNC}
on entering
+ * store-and-forward, Cluster2 auto-reacts and drives its own LOCAL record to
+ * {@code DEGRADED_STANDBY}. That reaction may still be in flight when this
helper runs, so for
+ * the {@code DEGRADED_STANDBY} target only we first poll a bounded window
for the record to
+ * settle and treat "already at target" as a no-op — otherwise a redundant
transition (e.g.
+ * DEGRADED_STANDBY -> DEGRADED_STANDBY) would throw
InvalidClusterRoleTransitionException on a
+ * slow box. Other targets (e.g. {@code STANDBY_TO_ACTIVE}) are never
peer-driven, so the
+ * settle-poll is skipped for them to avoid dead time.
+ */
+ private void transitionCluster2(HAGroupStoreRecord.HAGroupState target)
throws Exception {
+ // Only DEGRADED_STANDBY is reached by the peer-aware auto-reaction; let
any in-flight reaction
+ // settle so the no-op check below fires deterministically. Skipping this
for non-peer-driven
+ // targets avoids waiting on a condition that can never become true.
+ if (target == HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY) {
+ long settleDeadline = System.currentTimeMillis() +
PEER_REACTION_SETTLE_MILLIS;
+ while (System.currentTimeMillis() < settleDeadline &&
!cluster2StateIs(target)) {
+ Thread.sleep(PEER_REACTION_POLL_MILLIS);
+ }
+ }
+ // Check if already at target - no-op if so (avoid redundant transition
exception).
+ if (cluster2StateIs(target)) {
+ return;
+ }
+ HAGroupStoreClient client = HAGroupStoreClient.getInstance(conf2,
haGroupName);
+ long deadline = System.currentTimeMillis() + 60000L;
+ long lastWait = -1L;
+ while (System.currentTimeMillis() < deadline) {
+ lastWait = client.setHAGroupStatusIfNeeded(target);
+ if (lastWait == 0L) {
+ return;
+ }
+ Thread.sleep(Math.min(lastWait, 2000L));
+ }
+ throw new AssertionError("cluster2 transition to " + target
+ + " was still throttled at deadline (lastWait=" + lastWait + ")");
+ }
+
+ /** True if Cluster2's persisted HA record is currently {@code state}. */
+ private boolean cluster2StateIs(HAGroupStoreRecord.HAGroupState state) {
+ try {
+ Optional<HAGroupStoreRecord> rec =
+
HAGroupStoreManager.getInstance(conf2).getHAGroupStoreRecord(haGroupName);
+ return rec.isPresent() && rec.get().getHAGroupState() == state;
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Stage 3: degrade Cluster2 (freezes lastRoundInSync at A), perform the
real direct
+ * DEGRADED_STANDBY -> STANDBY_TO_ACTIVE, then drive replay until
Cluster2 promotes to
+ * ACTIVE_IN_SYNC. Assert zero RPO: every A and B row is present
cell-for-cell on Cluster2, the
+ * replay state is back to SYNC, and lastRoundInSync advanced past its
post-A value to cover B.
+ * <p>
+ * Regression guard: without PHOENIX-7920's triggerFailoverListner
CAS(DEGRADED, SYNCED_RECOVERY),
+ * the direct transition leaves the replay state at DEGRADED forever,
shouldTriggerFailover()
+ * never passes, Cluster2 never reaches ACTIVE_IN_SYNC, and the promotion
poll below times out
+ * (red).
+ */
+ private void stageDegradeDirectFailoverAndAssertZeroRPO() throws Exception {
+ // Degrade: LOCAL -> DEGRADED_STANDBY drives the degradedListener to
DEGRADED and freezes
+ // lastRoundInSync at the batch-A sync point (B is forwarded but not yet
replayed).
+ transitionCluster2(HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY);
+ awaitCondition(
+ () -> discovery.getReplicationReplayState()
+ == ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED,
+ 30000L, "degradedListener should drive replay state to DEGRADED");
+ assertEquals("lastRoundInSync must stay frozen at the batch-A sync point
during DEGRADED",
+ syncPointAfterA, discovery.getLastRoundInSync().getEndTime());
+
+ // Direct failover: LOCAL -> STANDBY_TO_ACTIVE (no STANDBY hop).
triggerFailoverListner must
+ // CAS DEGRADED -> SYNCED_RECOVERY and arm failoverPending.
+ transitionCluster2(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE);
+ awaitCondition(
+ () -> discovery.getReplicationReplayState()
+ ==
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY,
+ 30000L, "direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE must move replay
to SYNCED_RECOVERY");
+ assertTrue("failoverPending must be armed after STANDBY_TO_ACTIVE",
+ discovery.getFailoverPending());
+
+ // Make sure the client cache reflects STANDBY_TO_ACTIVE before
triggerFailover reads it.
+ awaitCondition(() ->
cluster2StateIs(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE), 30000L,
+ "cluster2 record should reflect STANDBY_TO_ACTIVE before driving
failover");
+
+ logGroup.close();
+
+ // Drive replay: SYNCED_RECOVERY rewinds to lastRoundInSync (A),
re-replays B in SYNC. With no
+ // new files arriving, the replay drains, shouldTriggerFailover() passes,
and triggerFailover()
+ // sets ACTIVE_IN_SYNC on Cluster2.
+ driveReplayUntil(() ->
cluster2StateIs(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC), 120000L,
+ "cluster2 must promote to ACTIVE_IN_SYNC after the direct failover");
+
+ // Zero-RPO assertions.
+ CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1,
conf2, tableName);
+ assertEquals("replay state must be SYNC after promotion",
+ ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+ assertTrue("lastRoundInSync must advance past the batch-A sync point to
cover batch B",
+ discovery.getLastRoundInSync().getEndTime() > syncPointAfterA);
+ LOG.info("Stage 3 complete: cluster2 promoted to ACTIVE_IN_SYNC with zero
RPO");
+ }
+}
