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 dc5719ae93 PHOENIX-7962 Fix ReplicationLogGroup mode initialization
and handle failover deadlock (#2581)
dc5719ae93 is described below
commit dc5719ae935305741467aeb0370c93318fe6449a
Author: tkhurana <[email protected]>
AuthorDate: Mon Jul 27 13:01:27 2026 -0700
PHOENIX-7962 Fix ReplicationLogGroup mode initialization and handle
failover deadlock (#2581)
A ReplicationLogGroup is a writer and must exist only where the local
cluster
is active for the HA group. Previously it was created unconditionally
(lazily
on first mutation, and eagerly at RS startup), which let non-active clusters
start writers and re-ship, and let the cutover writer deadlock standby
failover.
Changes:
- Role-gated creation: ReplicationLogGroup.get(...) now returns Optional and
creates a group only when the local role is active; a non-active role
returns
empty and is not cached, so the next call after promotion re-checks. Live
write path (preBatchMutate) throws on empty (split-brain guard); replay
path
(preWALRestore) skips re-ship on empty.
- Skip eager init on non-active clusters: PhoenixRegionServerEndpoint only
eager-inits where the local role is active; others init lazily on
promotion.
- Failover deadlock fix: a new failoverPending flag (set on entry to the
ACTIVE_IN_SYNC_TO_STANDBY cutover gate, cleared on abort) suspends log
rotation so no new files land in the peer's shard directory, and turns a
SYNC-write failure into an RS abort instead of the illegal cutover->SAF
transition. init() seeds the flag on restart-in-cutover.
- Consolidated all LOCAL state listeners into subscribeToStateChanges(),
each
paired with its unsubscribe so close() leaks no watchers; handles cutover
suspend, STANDBY/DEGRADED_STANDBY close, and SYNC/SYNC_AND_FORWARD flips.
- New constructor overload lets init() reuse the record the role gate read.
Tests: 9 new unit tests in ReplicationLogGroupTest and 2 integration tests
(testWriterClosesOnDemotionToStandby,
testCutoverSuspendsAndResumesRotation).
---
.../coprocessor/PhoenixRegionServerEndpoint.java | 40 ++-
.../phoenix/hbase/index/IndexRegionObserver.java | 111 ++++++--
.../apache/phoenix/replication/ReplicationLog.java | 17 ++
.../ReplicationLogDiscoveryForwarder.java | 62 +----
.../phoenix/replication/ReplicationLogGroup.java | 285 +++++++++++++++++----
.../apache/phoenix/replication/SyncModeImpl.java | 11 +
.../replication/ReplicationLogGroupBaseIT.java | 3 +-
.../phoenix/replication/ReplicationLogGroupIT.java | 128 +++++++++
.../replication/ReplicationLogGroupTest.java | 272 +++++++++++++++++++-
9 files changed, 786 insertions(+), 143 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java
index f821e6995a..c68ee31ddc 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java
@@ -29,6 +29,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.hadoop.conf.Configuration;
@@ -51,6 +52,7 @@ import
org.apache.phoenix.hbase.index.parallel.ThreadPoolManager;
import org.apache.phoenix.hbase.index.parallel.WaitForCompletionTaskRunner;
import org.apache.phoenix.jdbc.ClusterRoleRecord;
import org.apache.phoenix.jdbc.HAGroupStoreManager;
+import org.apache.phoenix.jdbc.HAGroupStoreRecord;
import org.apache.phoenix.protobuf.ProtobufUtil;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
@@ -306,8 +308,6 @@ public class PhoenixRegionServerEndpoint extends
// Phase 2: Prewarm individual HAGroupStoreClients with retry
// and eagerly initialize ReplicationLogGroup instances
- boolean shouldInitReplicationLogGroup = serverName != null && conf
- .getBoolean(SYNCHRONOUS_REPLICATION_ENABLED,
DEFAULT_SYNCHRONOUS_REPLICATION_ENABLED);
try {
while (!pending.isEmpty()) {
Iterator<String> iterator = pending.iterator();
@@ -315,7 +315,7 @@ public class PhoenixRegionServerEndpoint extends
String haGroup = iterator.next();
try {
manager.getClusterRoleRecord(haGroup);
- if (shouldInitReplicationLogGroup) {
+ if (shouldEagerlyInit(manager, haGroup)) {
try {
ReplicationLogGroup.get(conf, serverName, haGroup,
abortable);
LOGGER.info("Eagerly initialized ReplicationLogGroup {} on
server {}", haGroup,
@@ -324,6 +324,11 @@ public class PhoenixRegionServerEndpoint extends
LOGGER.warn("Failed to eagerly initialize
ReplicationLogGroup for HA group: {}."
+ " Will be lazily initialized on first mutation.",
haGroup, e);
}
+ } else {
+ LOGGER.info(
+ "Skipping eager ReplicationLogGroup init for HA group: {} on
server {};"
+ + " the writer will be created lazily if this cluster
becomes active.",
+ haGroup, serverName);
}
iterator.remove();
LOGGER.info("Prewarmed HAGroupStoreClient: {} ({} remaining)",
haGroup,
@@ -346,4 +351,33 @@ public class PhoenixRegionServerEndpoint extends
});
}
+ /**
+ * Eagerly initialize ReplicationLogGroup only on clusters whose local role
is currently active
+ * (ACTIVE or ACTIVE_TO_STANDBY). On STANDBY/OFFLINE/UNKNOWN clusters the
group will be created
+ * lazily on first mutation if the role transitions.
+ */
+ private boolean shouldEagerlyInit(HAGroupStoreManager manager, String
haGroup) {
+ if (serverName == null) {
+ return false;
+ }
+ if (
+ !conf.getBoolean(SYNCHRONOUS_REPLICATION_ENABLED,
DEFAULT_SYNCHRONOUS_REPLICATION_ENABLED)
+ ) {
+ return false;
+ }
+ try {
+ Optional<HAGroupStoreRecord> record =
manager.getHAGroupStoreRecord(haGroup);
+ if (!record.isPresent()) {
+ LOGGER.warn("Skipping eager ReplicationLogGroup init for {}: no local
HAGroupStoreRecord",
+ haGroup);
+ return false;
+ }
+ return record.get().getHAGroupState().getClusterRole().isActive();
+ } catch (Exception e) {
+ LOGGER.warn("Skipping eager ReplicationLogGroup init for {}: unable to
read local role",
+ haGroup, e);
+ return false;
+ }
+ }
+
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
index b06d7f8931..715683e979 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
@@ -136,6 +136,8 @@ import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.index.PhoenixIndexMetaData;
import org.apache.phoenix.index.PhoenixIndexMetaDataBuilder;
import org.apache.phoenix.jdbc.HAGroupStoreManager;
+import org.apache.phoenix.jdbc.HAGroupStoreRecord;
+import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
import org.apache.phoenix.query.KeyRange;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServicesOptions;
@@ -849,8 +851,10 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
throw new IOException(
"HAGroupStoreManager is null " + "for current cluster, check
configuration");
}
- // Extract HAGroupName from the mutations
- Optional<ReplicationLogGroup> logGroup =
getHAGroupFromBatch(c.getEnvironment(), miniBatchOp);
+ // Extract the HA group name the batch carries (if any). The
mutation-block and staleness
+ // gates below key off this name, not off writer presence, so they apply
on a standby or
+ // demoted cluster too — not only where this cluster is active.
+ Optional<String> haGroupName = getHAGroupNameFromBatch(miniBatchOp);
// Path-coverage counter: increments whenever a mutation batch reaches
preBatchMutate
// without a resolvable HA group attribute, so the cluster-role-based
mutation-block gate
@@ -859,10 +863,10 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// feature is disabled or no block window is active, there is no
property to breach).
// Tracked globally rather than per-table so operators can compare
baseline vs.
// post-deploy delta to spot new write paths that forgot to attach
_HAGroupName.
- // Intentionally scoped to !logGroup.isPresent() regardless of
dataTableName —
+ // Intentionally scoped to !haGroupName.isPresent() regardless of
dataTableName —
// system-HA-group writes WITH a haGroup are an intended gate exemption
(state writes
// must proceed during a block window) and are not counted here.
- if (!logGroup.isPresent()) {
+ if (shouldReplicate && !haGroupName.isPresent()) {
try {
MetricsHaBypassSourceFactory.getInstance().incrementBypassedMutationBlockCount();
} catch (Throwable t) {
@@ -871,30 +875,34 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
// We don't want to check for mutation blocking for the system ha group
table
- if (!dataTableName.equals(SYSTEM_HA_GROUP_NAME) && logGroup.isPresent())
{
- // Check if mutation is blocked for the HA Group
- String haGroupName = logGroup.get().getHAGroupName();
+ if (!dataTableName.equals(SYSTEM_HA_GROUP_NAME) &&
haGroupName.isPresent()) {
+ String name = haGroupName.get();
// TODO: Below approach might be slow need to figure out faster way,
// slower part is getting haGroupStoreClient We can also cache
// roleRecord (I tried it and still it's slow due to haGroupStoreClient
// initialization) and caching will give us old result in case one
cluster
// is unreachable instead of UNKNOWN.
- boolean isHAGroupOnClientStale =
haGroupStoreManager.isHAGroupOnClientStale(haGroupName);
- if (StringUtils.isNotBlank(haGroupName) && isHAGroupOnClientStale) {
- throw new StaleClusterRoleRecordException(String
- .format("HAGroupStoreRecord is stale for haGroup %s on " +
"client", haGroupName));
+ boolean isHAGroupOnClientStale =
haGroupStoreManager.isHAGroupOnClientStale(name);
+ if (StringUtils.isNotBlank(name) && isHAGroupOnClientStale) {
+ throw new StaleClusterRoleRecordException(
+ String.format("HAGroupStoreRecord is stale for haGroup %s on " +
"client", name));
}
// Check if mutation's haGroup is stale
- if (
- StringUtils.isNotBlank(haGroupName) &&
haGroupStoreManager.isMutationBlocked(haGroupName)
- ) {
+ if (StringUtils.isNotBlank(name) &&
haGroupStoreManager.isMutationBlocked(name)) {
throw new MutationBlockedIOException(
"Blocking Mutation as Some CRRs " + "are in ACTIVE_TO_STANDBY
state and "
+ "CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED is true");
}
}
+
+ // Resolve the replication writer for this batch (present only where
this cluster is active
+ // for the HA group). This also enforces the split-brain guard for
non-PARALLEL live writes
+ // to a non-active cluster.
+ Optional<ReplicationLogGroup> logGroup = haGroupName.isPresent()
+ ? getReplicationLogGroup(c.getEnvironment(), haGroupStoreManager,
haGroupName.get())
+ : Optional.empty();
preBatchMutateWithExceptions(c, miniBatchOp, logGroup);
return;
} catch (Throwable t) {
@@ -905,24 +913,71 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
/**
- * Get the HA group associated with the batch. We assume that all the
mutations in the batch will
- * have the same HA group.
- * @return HA group if present or empty if missing
+ * Resolve the replication writer for the named HA group. Returns the writer
when this cluster is
+ * active for the group, empty when replication is disabled or a non-active
PARALLEL cluster
+ * legitimately receives a double-write, and throws for a non-active
non-PARALLEL (split-brain)
+ * live write.
+ * @return the HA group writer, or empty if none should be used
*/
- private Optional<ReplicationLogGroup>
getHAGroupFromBatch(RegionCoprocessorEnvironment env,
- MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
+ private Optional<ReplicationLogGroup>
getReplicationLogGroup(RegionCoprocessorEnvironment env,
+ HAGroupStoreManager haGroupStoreManager, String haGroup) throws
IOException {
+ if (!shouldReplicate) {
+ return Optional.empty();
+ }
+ Optional<ReplicationLogGroup> logGroup =
ReplicationLogGroup.get(env.getConfiguration(),
+ env.getServerName(), haGroup, haGroupStoreManager);
+ if (logGroup.isPresent()) {
+ return logGroup;
+ }
+ // get() returned empty: this cluster is not active for the HA group. Its
meaning depends on
+ // the HA policy of the write that just arrived:
+ // - PARALLEL: the client deliberately double-writes to both clusters, so
a live write on
+ // the standby is expected. Commit locally and skip re-ship (a non-active
cluster has no
+ // writer) — the peer gets its copy from the client's other leg. Same as
the replay path
+ // in getHAGroupFromWALKey.
+ // - FAILOVER: only the active cluster takes writes, so a live write here
is a stray/stale
+ // sync-path write to a demoted cluster. Reject it rather than committing
locally
+ // unreplicated (split brain).
+ if (isParallelPolicy(haGroupStoreManager, haGroup)) {
+ return Optional.empty();
+ }
+ throw new IOException(String
+ .format("HAGroup %s cannot accept a sync-path write: this cluster is not
active", haGroup));
+ }
+
+ /**
+ * Extract the HA group name carried by the batch, if any. All mutations in
a batch share the same
+ * HA group, so the first mutation's {@code _HAGroupName} attribute is
authoritative. Pure
+ * extraction: no configuration, role, or policy lookups.
+ * @return the HA group name, or empty if the batch carries no HA group
attribute
+ */
+ private Optional<String>
+ getHAGroupNameFromBatch(MiniBatchOperationInProgress<Mutation>
miniBatchOp) {
if (miniBatchOp.size() > 0) {
Mutation m = miniBatchOp.getOperation(0);
byte[] haGroupName =
m.getAttribute(BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB);
if (haGroupName != null) {
- ReplicationLogGroup logGroup =
ReplicationLogGroup.get(env.getConfiguration(),
- env.getServerName(), Bytes.toString(haGroupName), abortable);
- return Optional.of(logGroup);
+ return Optional.of(Bytes.toString(haGroupName));
}
}
return Optional.empty();
}
+ /**
+ * True when the HA group uses the PARALLEL policy, under which the client
double-writes to both
+ * clusters. On a non-active cluster this distinguishes a legitimate
PARALLEL standby write (which
+ * should commit locally and skip re-ship) from a stray FAILOVER write
(split brain, to reject). A
+ * missing record is treated as not-PARALLEL so the caller keeps the
conservative split-brain
+ * guard.
+ */
+ private boolean isParallelPolicy(HAGroupStoreManager haGroupStoreManager,
String haGroupName)
+ throws IOException {
+ Optional<HAGroupStoreRecord> record =
+ haGroupStoreManager.getEffectiveHAGroupStoreRecord(haGroupName);
+ return record.isPresent()
+ &&
HighAvailabilityPolicy.PARALLEL.name().equalsIgnoreCase(record.get().getPolicy());
+ }
+
/**
* Get the HA group associated with the WAL key. A batch of mutations is
recorded in a single WAL
* edit.
@@ -932,9 +987,15 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
Map<String, byte[]> walKeyAttrs) throws IOException {
byte[] haGroupName =
walKeyAttrs.get(BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB);
if (haGroupName != null) {
- ReplicationLogGroup logGroup =
ReplicationLogGroup.get(env.getConfiguration(),
- env.getServerName(), Bytes.toString(haGroupName), abortable);
- return Optional.of(logGroup);
+ // get() returns empty when this cluster is not active. On the replay
path this means a
+ // demoted
+ // cluster is replaying its own formerly-ACTIVE edits after a crash: the
peer is ACTIVE now,
+ // so
+ // this cluster must not start a replication writer or re-ship.
Returning empty lets the
+ // caller
+ // skip the re-ship; HBase still replays the edits into the local region.
+ return ReplicationLogGroup.get(env.getConfiguration(),
env.getServerName(),
+ Bytes.toString(haGroupName), abortable);
}
return Optional.empty();
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java
index 9621966ed1..694bd31061 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java
@@ -244,8 +244,19 @@ public class ReplicationLog {
* Submits a {@link LogRotationTask} to the executor. The CAS gate ensures
only one rotation can
* be queued or in flight at a time — if the flag is already set, a task is
already pending or
* running. Both scheduled ticks and on-demand size-triggered callers go
through this method.
+ * <p>
+ * Suspends rotation during the in-sync cutover gate: while {@code
failoverPending} is set we skip
+ * before the CAS so a tick never engages it. Minting a new file each round
would keep dropping
+ * .plog files into the peer's shard directory and deadlock the standby's
failover check; the
+ * current writer stays open so in-flight writes still land. Skipping ahead
of the CAS (rather
+ * than inside {@link LogRotationTask#run()}) keeps the gate clear, so a
later tick resumes
+ * rotation as soon as the flag clears on abort.
*/
private void requestRotation() {
+ if (logGroup.isFailoverPending()) {
+ LOG.info("HAGroup {} rotation suspended: failover pending", logGroup);
+ return;
+ }
if (rotationRequested.compareAndSet(false, true)) {
try {
rotationExecutor.execute(new LogRotationTask());
@@ -411,6 +422,12 @@ public class ReplicationLog {
@VisibleForTesting
protected void forceRotation() {
+ // Model the same suspension policy as a scheduled tick: rotation is a
no-op while the group is
+ // in the in-sync cutover gate (see requestRotation()).
+ if (logGroup.isFailoverPending()) {
+ LOG.info("HAGroup {} rotation suspended: failover pending", logGroup);
+ return;
+ }
new LogRotationTask().run();
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
index 1757421599..c1a710362e 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
@@ -18,7 +18,6 @@
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 static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.SYNC_AND_FORWARD;
import java.io.IOException;
@@ -26,11 +25,6 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
-import org.apache.phoenix.jdbc.ClusterType;
-import org.apache.phoenix.jdbc.HAGroupStateListener;
-import org.apache.phoenix.jdbc.HAGroupStoreManager;
-import org.apache.phoenix.jdbc.HAGroupStoreRecord;
-import org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode;
import org.apache.phoenix.replication.metrics.MetricsReplicationLogDiscovery;
import
org.apache.phoenix.replication.metrics.MetricsReplicationLogForwarderSourceFactory;
import org.apache.phoenix.util.EnvironmentEdgeManager;
@@ -97,42 +91,6 @@ public class ReplicationLogDiscoveryForwarder extends
ReplicationLogDiscovery {
// Initialize the discovery only. Forwarding begins only when we switch to
the
// STORE_AND_FORWARD mode or SYNC_AND_FORWARD mode.
super.init();
-
- // Set up a listener to the ACTIVE_NOT_IN_SYNC state. This is needed
because whenever any
- // RS switches to STORE_AND_FORWARD mode, other RS's in the cluster must
move out of SYNC
- // mode.
- HAGroupStateListener activeNotInSync =
- (groupName, fromState, toState, modifiedTime, clusterType,
lastSyncStateTimeInMs) -> {
- if (
- clusterType == ClusterType.LOCAL
- &&
HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC.equals(toState)
- ) {
- LOG.info("Received ACTIVE_NOT_IN_SYNC event for {}", logGroup);
- // If the current mode is SYNC only then switch to SYNC_AND_FORWARD
mode
- checkAndSetModeAndNotify(SYNC, SYNC_AND_FORWARD);
- }
- };
-
- // Set up a listener to the ACTIVE_IN_SYNC state. This is needed because
when the RS
- // switches back to SYNC mode, the other RS's in the cluster must move out
of
- // SYNC_AND_FORWARD mode to SYNC mode.
- HAGroupStateListener activeInSync =
- (groupName, fromState, toState, modifiedTime, clusterType,
lastSyncStateTimeInMs) -> {
- if (
- clusterType == ClusterType.LOCAL
- && HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC.equals(toState)
- ) {
- LOG.info("Received ACTIVE_IN_SYNC event for {}", logGroup);
- // Set the current mode to SYNC
- checkAndSetModeAndNotify(SYNC_AND_FORWARD, SYNC);
- }
- };
-
- HAGroupStoreManager haGroupStoreManager =
logGroup.getHAGroupStoreManager();
- haGroupStoreManager.subscribeToTargetState(logGroup.getHAGroupName(),
- HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC, ClusterType.LOCAL,
activeNotInSync);
- haGroupStoreManager.subscribeToTargetState(logGroup.getHAGroupName(),
- HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC, ClusterType.LOCAL,
activeInSync);
}
@Override
@@ -153,7 +111,7 @@ public class ReplicationLogDiscoveryForwarder extends
ReplicationLogDiscovery {
&& isLogCopyThroughputAboveThreshold(srcStat.getLen(), copyTime)
) {
// start recovery by switching to SYNC_AND_FORWARD
- checkAndSetModeAndNotify(STORE_AND_FORWARD, SYNC_AND_FORWARD);
+ logGroup.checkAndSetModeAndNotify(STORE_AND_FORWARD, SYNC_AND_FORWARD);
}
}
@@ -170,7 +128,7 @@ public class ReplicationLogDiscoveryForwarder extends
ReplicationLogDiscovery {
LOG.info("Processed all the replication log files for {}", logGroup);
// if this RS is still in STORE_AND_FORWARD mode like when it didn't
process any file
// move this RS to SYNC_AND_FORWARD
- checkAndSetModeAndNotify(STORE_AND_FORWARD, SYNC_AND_FORWARD);
+ logGroup.checkAndSetModeAndNotify(STORE_AND_FORWARD, SYNC_AND_FORWARD);
if (syncUpdateTS <= EnvironmentEdgeManager.currentTimeMillis()) {
try {
@@ -212,22 +170,6 @@ public class ReplicationLogDiscoveryForwarder extends
ReplicationLogDiscovery {
return replicationLogTracker;
}
- /**
- * Helper API to check and set the replication mode and then notify the
disruptor
- */
- private boolean checkAndSetModeAndNotify(ReplicationMode expectedMode,
ReplicationMode newMode) {
- boolean ret = logGroup.checkAndSetMode(expectedMode, newMode);
- if (ret) {
- // replication mode switched, notify the event handler
- try {
- logGroup.sync();
- } catch (IOException e) {
- LOG.info("Failed to notify event handler for {}", logGroup, e);
- }
- }
- return ret;
- }
-
@Override
public double getWaitingBufferPercentage() {
return
getConf().getDouble(REPLICATION_FORWARDER_WAITING_BUFFER_PERCENTAGE_KEY,
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
index aee7143761..3d760af01a 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
@@ -66,6 +66,9 @@ import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.phoenix.execute.MutationState;
+import org.apache.phoenix.jdbc.ClusterRoleRecord;
+import org.apache.phoenix.jdbc.ClusterType;
+import org.apache.phoenix.jdbc.HAGroupStateListener;
import org.apache.phoenix.jdbc.HAGroupStoreManager;
import org.apache.phoenix.jdbc.HAGroupStoreRecord;
import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
@@ -234,8 +237,18 @@ public class ReplicationLogGroup {
protected long syncTimeoutMs;
protected long peerInitTimeoutMs;
protected final AtomicBoolean closed = new AtomicBoolean(false);
+ // Set while the local cluster is in the in-sync cutover gate
(ACTIVE_IN_SYNC_TO_STANDBY).
+ // Suspends
+ // log rotation (LogRotationTask.run() no-ops) so no new files are dropped
into the peer's shard
+ // directory, and turns a SYNC-write failure into an RS abort instead of the
illegal SYNC->SAF
+ // transition. Cleared when the state returns to ACTIVE_IN_SYNC (cutover
abort).
+ private final AtomicBoolean failoverPending = new AtomicBoolean(false);
protected final Abortable abortable;
protected long shutdownTimeoutMs;
+ // Undo actions for every LOCAL state subscription registered in init(), run
on close() to remove
+ // them. Registering the unsubscribe alongside the subscribe (see
subscribeLocal) keeps the two in
+ // lockstep so a closed group leaks no ZK watchers.
+ private final List<Runnable> stateUnsubscribers = new ArrayList<>();
/**
* The replication mode determines how mutations are handled. Mode
transitions occur automatically
@@ -383,44 +396,33 @@ public class ReplicationLogGroup {
protected LogEventHandler eventHandler;
/**
- * Get or create a ReplicationLogGroup instance for the given HA Group.
+ * Get or create a ReplicationLogGroup instance for the given HA Group.
Returns empty when this
+ * cluster is not active for the HA group (see {@link #getOrCreate}).
* @param conf Configuration object
* @param serverName The server name
* @param haGroupName The HA Group name
- * @return ReplicationLogGroup instance
+ * @return ReplicationLogGroup instance, or empty if this cluster is not
active
* @throws IOException if initialization fails
*/
- public static ReplicationLogGroup get(Configuration conf, ServerName
serverName,
+ public static Optional<ReplicationLogGroup> get(Configuration conf,
ServerName serverName,
String haGroupName) throws IOException {
return get(conf, serverName, haGroupName, (Abortable) null);
}
/**
- * Get or create a ReplicationLogGroup instance for the given HA Group.
+ * Get or create a ReplicationLogGroup instance for the given HA Group.
Returns empty when this
+ * cluster is not active for the HA group (see {@link #getOrCreate}).
* @param conf Configuration object
* @param serverName The server name
* @param haGroupName The HA Group name
* @param abortable Abortable to invoke on fatal errors (typically
RegionServerServices)
- * @return ReplicationLogGroup instance
+ * @return ReplicationLogGroup instance, or empty if this cluster is not
active
* @throws IOException if initialization fails
*/
- public static ReplicationLogGroup get(Configuration conf, ServerName
serverName,
+ public static Optional<ReplicationLogGroup> get(Configuration conf,
ServerName serverName,
String haGroupName, Abortable abortable) throws IOException {
- try {
- return INSTANCES.computeIfAbsent(instanceKey(serverName, haGroupName), k
-> {
- try {
- ReplicationLogGroup group = new ReplicationLogGroup(conf,
serverName, haGroupName,
- HAGroupStoreManager.getInstance(conf), abortable);
- group.init();
- return group;
- } catch (IOException e) {
- LOG.error("Failed to create ReplicationLogGroup for HA Group: {}",
haGroupName, e);
- throw new UncheckedIOException(e);
- }
- });
- } catch (UncheckedIOException e) {
- throw e.getCause();
- }
+ return getOrCreate(conf, serverName, haGroupName,
HAGroupStoreManager.getInstance(conf),
+ abortable);
}
/**
@@ -429,23 +431,65 @@ public class ReplicationLogGroup {
* @param serverName The server name
* @param haGroupName The HA Group name
* @param haGroupStoreManager HA Group Store Manager instance
- * @return ReplicationLogGroup instance
+ * @return ReplicationLogGroup instance, or empty if this cluster is not
active
* @throws IOException if initialization fails
*/
- public static ReplicationLogGroup get(Configuration conf, ServerName
serverName,
+ public static Optional<ReplicationLogGroup> get(Configuration conf,
ServerName serverName,
String haGroupName, HAGroupStoreManager haGroupStoreManager) throws
IOException {
+ return getOrCreate(conf, serverName, haGroupName, haGroupStoreManager,
null);
+ }
+
+ /**
+ * Get an existing, or create and initialize a new, ReplicationLogGroup —
but only when this
+ * cluster is active for the HA group. A replication log group is a WRITER;
it must exist only
+ * where this cluster is the active side. When the local role is not active
this returns
+ * {@link Optional#empty()} and does NOT cache, so a later call re-checks
the (now-active) role
+ * once this cluster is promoted.
+ */
+ private static Optional<ReplicationLogGroup> getOrCreate(Configuration conf,
+ ServerName serverName, String haGroupName, HAGroupStoreManager
haGroupStoreManager,
+ Abortable abortable) throws IOException {
try {
- return INSTANCES.computeIfAbsent(instanceKey(serverName, haGroupName), k
-> {
- try {
- ReplicationLogGroup group =
- new ReplicationLogGroup(conf, serverName, haGroupName,
haGroupStoreManager);
- group.init();
- return group;
- } catch (IOException e) {
- LOG.error("Failed to create ReplicationLogGroup for HA Group: {}",
haGroupName, e);
- throw new UncheckedIOException(e);
- }
- });
+ ReplicationLogGroup group =
+ INSTANCES.computeIfAbsent(instanceKey(serverName, haGroupName), k -> {
+ try {
+ // Role check BEFORE constructing any writer resource. Only
mutations carrying
+ // HA_GROUP_NAME reach get() (the standby replay-apply path does
not annotate that
+ // attribute), via either the live write path or preWALRestore
recovery. On a non-active
+ // role we must not start a writer: on the live path it would be a
stray/split-brain
+ // sync-path write, and on the replay path it is a demoted cluster
replaying its own
+ // formerly-ACTIVE edits after a crash while the peer is now
ACTIVE. Returning null
+ // keeps
+ // computeIfAbsent from caching, so the first call after this
cluster is promoted
+ // re-runs
+ // the check against the now-active role. The mutation-blocked
role (ACTIVE_TO_STANDBY,
+ // the ACTIVE_IN_SYNC_TO_STANDBY cutover gate) is active and
deliberately allowed: it is
+ // reached from preWALRestore recovery on an RS restart during
cutover, and the writer
+ // starts in SYNC with rotation suspended via failoverPending (see
init()).
+ Optional<HAGroupStoreRecord> haRecord =
+ haGroupStoreManager.getEffectiveHAGroupStoreRecord(haGroupName);
+ if (!haRecord.isPresent()) {
+ throw new UncheckedIOException(new IOException(
+ String.format("HAGroup %s got an empty group store record",
haGroupName)));
+ }
+ ClusterRoleRecord.ClusterRole role =
haRecord.get().getClusterRole();
+ if (!role.isActive()) {
+ LOG.info(
+ "Not creating a ReplicationLogGroup for HA Group {}: local
role {} is not active"
+ + " (state {})",
+ haGroupName, role, haRecord.get().getHAGroupState());
+ return null;
+ }
+ ReplicationLogGroup created = new ReplicationLogGroup(conf,
serverName, haGroupName,
+ haGroupStoreManager, abortable, haRecord.get());
+ created.init();
+ return created;
+ } catch (IOException e) {
+ LOG.error("Failed to create ReplicationLogGroup for HA Group: {}",
haGroupName, e);
+ throw new UncheckedIOException(e);
+ }
+ });
+ return Optional.ofNullable(group);
} catch (UncheckedIOException e) {
throw e.getCause();
}
@@ -473,6 +517,22 @@ public class ReplicationLogGroup {
*/
protected ReplicationLogGroup(Configuration conf, ServerName serverName,
String haGroupName,
HAGroupStoreManager haGroupStoreManager, Abortable abortable) {
+ this(conf, serverName, haGroupName, haGroupStoreManager, abortable, null);
+ }
+
+ /**
+ * Protected constructor for ReplicationLogGroup.
+ * @param conf Configuration object
+ * @param serverName The server name
+ * @param haGroupName The HA Group name
+ * @param haGroupStoreManager HA Group Store Manager instance
+ * @param abortable Abortable to invoke on fatal errors (may be
null)
+ * @param haGroupStoreRecord the effective record already read by the
caller (may be null); when
+ * provided, init() uses it instead of re-reading
the record
+ */
+ protected ReplicationLogGroup(Configuration conf, ServerName serverName,
String haGroupName,
+ HAGroupStoreManager haGroupStoreManager, Abortable abortable,
+ HAGroupStoreRecord haGroupStoreRecord) {
// conf object from coprocessor is instance of
// org.apache.hadoop.hbase.coprocessor.ReadOnlyConfiguration and we need
to modify it when
// we send rpc to namenode so copying it
@@ -487,6 +547,7 @@ public class ReplicationLogGroup {
this.haGroupName = haGroupName;
this.haGroupStoreManager = haGroupStoreManager;
this.abortable = abortable;
+ this.haGroupStoreRecord = haGroupStoreRecord;
this.shutdownTimeoutMs =
clonedConf.getLong(REPLICATION_LOG_GROUP_SHUTDOWN_TIMEOUT_MS_KEY,
DEFAULT_REPLICATION_LOG_GROUP_SHUTDOWN_TIMEOUT_MS);
this.metrics = createMetricsSource();
@@ -499,16 +560,23 @@ public class ReplicationLogGroup {
*/
protected void init() throws IOException {
LOG.info("Initializing ReplicationLogGroup {}", haGroupName);
- Optional<HAGroupStoreRecord> haRecord =
- haGroupStoreManager.getEffectiveHAGroupStoreRecord(haGroupName);
- if (!haRecord.isPresent()) {
- String message =
- String.format("HAGroup %s got an empty group store record while
initializing mode", this);
- LOG.error(message);
- throw new IOException(message);
+ // getOrCreate() reads the effective record for the role gate and seeds it
via the constructor,
+ // so by the time init() runs the record is present and the local role is
known active (possibly
+ // the ACTIVE_TO_STANDBY cutover gate, which starts in SYNC with rotation
suspended — see
+ // below).
+ // The record may be null only when init() is invoked directly (tests);
fall back to a read
+ // then.
+ if (haGroupStoreRecord == null) {
+ Optional<HAGroupStoreRecord> haRecord =
+ haGroupStoreManager.getEffectiveHAGroupStoreRecord(haGroupName);
+ if (!haRecord.isPresent()) {
+ String message =
+ String.format("HAGroup %s got an empty group store record while
initializing mode", this);
+ LOG.error(message);
+ throw new IOException(message);
+ }
+ this.haGroupStoreRecord = haRecord.get();
}
- HAGroupStoreRecord record = haRecord.get();
- this.haGroupStoreRecord = record;
this.localShardManager = createLocalShardManager();
this.peerInitTimeoutMs =
conf.getLong(REPLICATION_LOG_PEER_INIT_TIMEOUT_MS_KEY,
DEFAULT_REPLICATION_LOG_PEER_INIT_TIMEOUT_MS);
@@ -517,14 +585,114 @@ public class ReplicationLogGroup {
this.logForwarder = new ReplicationLogDiscoveryForwarder(this);
this.logForwarder.init();
// Initialize the replication mode based on the HAGroupStore state
- initializeReplicationMode(record);
+ initializeReplicationMode(haGroupStoreRecord);
// Use the override value if provided in the config, else use a derived
value
this.syncTimeoutMs = conf.getLong(REPLICATION_LOG_SYNC_TIMEOUT_KEY,
calculateSyncTimeout());
// Initialize the disruptor so that we start processing events
initializeDisruptor();
+ // Register all LOCAL state listeners that react to this cluster's role
changes (demotion,
+ // cutover suspend/resume, and mode flips). init() itself fails fast on a
non-active role, so a
+ // group only ever exists while active and these listeners drive it from
there.
+ subscribeToStateChanges();
+ // Restart-in-cutover: when init() runs (via preWALRestore recovery) while
the persisted state
+ // is already ACTIVE_IN_SYNC_TO_STANDBY, the cutover listener won't fire —
the state was read,
+ // not transitioned into — so seed failoverPending here. This keeps
rotation suspended from the
+ // start so the fresh writer does not re-arm the standby's failover
deadlock.
+ if
(haGroupStoreRecord.getHAGroupState().equals(HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY))
{
+ setFailoverPending(true);
+ }
LOG.info("HAGroup {} started with mode={}", this, mode);
}
+ /**
+ * Subscribe every LOCAL state listener that reacts to this cluster's role
changes. Dispatch is
+ * already keyed on {@code (clusterType, targetState)} (see
+ * {@code HAGroupStoreClient#notifySubscribers}), so each action fires only
for its exact target
+ * state on the local cluster — no in-listener guards needed.
+ * <ul>
+ * <li>{@code ACTIVE_IN_SYNC_TO_STANDBY} — the in-sync cutover gate (role
ACTIVE_TO_STANDBY,
+ * mutations blocked). Do <b>not</b> close here: a batch that captured this
group before the
+ * mutation block, committed locally, and reaches append/sync after a close
would hit
+ * IOException("Closed") — a committed mutation the peer never receives.
Instead set
+ * {@code failoverPending}, which suspends rotation (no new files into the
peer's shard directory,
+ * so the standby's failover-completion check can proceed) while keeping the
writer open so those
+ * in-flight writes land. The terminal {@code STANDBY} demotion performs the
actual close.</li>
+ * <li>{@code STANDBY} and {@code DEGRADED_STANDBY} — the terminal demotion.
When the peer is not
+ * visible a demotion to STANDBY surfaces as DEGRADED_STANDBY (peer-blind),
so listening only for
+ * STANDBY would leave the writer alive through the degraded window. The
listener contract forbids
+ * blocking on the cache-event thread, so the (blocking) {@link #close()} is
handed off to a
+ * short-lived daemon thread.</li>
+ * <li>{@code ACTIVE_IN_SYNC} — a cutover abort (ACTIVE_IN_SYNC_TO_STANDBY →
+ * ABORT_TO_ACTIVE_IN_SYNC → ACTIVE_IN_SYNC) clears {@code failoverPending}
to resume rotation on
+ * the same live group, and flips any RS still in SYNC_AND_FORWARD back to
SYNC.</li>
+ * <li>{@code ACTIVE_NOT_IN_SYNC} — when any RS drops to STORE_AND_FORWARD
the others must leave
+ * SYNC, so flip SYNC to SYNC_AND_FORWARD.</li>
+ * </ul>
+ * Idempotent: {@code close()} is CAS-guarded, and closing removes this
instance from the cache so
+ * a later write re-runs {@link #init()} against the current role — which
the tightened role gate
+ * rejects until this cluster returns to pure ACTIVE (promotion, or a
cutover abort back to
+ * ACTIVE_IN_SYNC).
+ */
+ protected void subscribeToStateChanges() throws IOException {
+ subscribeLocal(HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY, () -> {
+ LOG.info("HAGroup {} entered cutover gate; suspending rotation", this);
+ setFailoverPending(true);
+ });
+ Runnable closeOnDemotion = () -> {
+ LOG.info("HAGroup {} demoted; closing replication log writer", this);
+ Thread t = new Thread(this::close, "ReplicationLogGroup-demote-" +
getHAGroupName());
+ t.setDaemon(true);
+ t.start();
+ };
+ subscribeLocal(HAGroupState.STANDBY, closeOnDemotion);
+ subscribeLocal(HAGroupState.DEGRADED_STANDBY, closeOnDemotion);
+ subscribeLocal(HAGroupState.ACTIVE_IN_SYNC, () -> {
+ LOG.info("HAGroup {} returned to ACTIVE_IN_SYNC; resuming rotation and
SYNC mode", this);
+ // A cutover abort resumes rotation on this live group.
+ setFailoverPending(false);
+ // The group returned to in-sync: if this RS was recovering in
SYNC_AND_FORWARD, drop the
+ // forward leg and go back to pure SYNC.
+ checkAndSetModeAndNotify(ReplicationMode.SYNC_AND_FORWARD,
ReplicationMode.SYNC);
+ });
+ subscribeLocal(HAGroupState.ACTIVE_NOT_IN_SYNC, () -> {
+ LOG.info("HAGroup {} received ACTIVE_NOT_IN_SYNC", this);
+ // When any RS drops to STORE_AND_FORWARD, others must leave SYNC.
+ checkAndSetModeAndNotify(ReplicationMode.SYNC,
ReplicationMode.SYNC_AND_FORWARD);
+ });
+ }
+
+ /**
+ * Subscribe {@code action} to a LOCAL transition into {@code targetState}
and record the matching
+ * unsubscribe so {@link #close()} removes it. Keeping the subscribe and its
undo in one call
+ * prevents the two sets from drifting apart. The action runs on the
cache-event thread, so it
+ * must be fast and non-blocking (hand off blocking work to a separate
thread).
+ */
+ private void subscribeLocal(HAGroupState targetState, Runnable action)
throws IOException {
+ HAGroupStateListener listener = (groupName, fromState, toState,
modifiedTime, clusterType,
+ lastSyncStateTimeInMs) -> action.run();
+ haGroupStoreManager.subscribeToTargetState(haGroupName, targetState,
ClusterType.LOCAL,
+ listener);
+ stateUnsubscribers.add(() ->
haGroupStoreManager.unsubscribeFromTargetState(haGroupName,
+ targetState, ClusterType.LOCAL, listener));
+ }
+
+ /**
+ * Conditionally switch replication mode and, if it changed, notify the
event handler with a sync
+ * so the disruptor picks up the new mode.
+ */
+ protected boolean checkAndSetModeAndNotify(ReplicationMode expectedMode,
+ ReplicationMode newMode) {
+ boolean switched = checkAndSetMode(expectedMode, newMode);
+ if (switched) {
+ try {
+ sync();
+ } catch (IOException e) {
+ LOG.info("Failed to notify event handler for {}", this, e);
+ }
+ }
+ return switched;
+ }
+
/**
* Calculate how long the application thread should wait for a sync to
finish. Uses the SAF sync
* timeout as the base — the consumer thread must have time to attempt SYNC
retries, flip to SAF,
@@ -539,12 +707,19 @@ public class ReplicationLogGroup {
}
/**
- * Initialize the replication mode based on the HAGroupStore state
+ * Initialize the replication mode based on the HAGroupStore state. {@link
#init()} fails fast on
+ * a non-active role, so this is only reached for active states:
ACTIVE_IN_SYNC and the
+ * ACTIVE_IN_SYNC_TO_STANDBY cutover gate both start SYNC (the gate is
reached via preWALRestore
+ * recovery on a restart-in-cutover); every other active state
(ACTIVE_NOT_IN_SYNC,
+ * ACTIVE_WITH_OFFLINE_PEER, ACTIVE_NOT_IN_SYNC_TO_STANDBY, ...) starts
STORE_AND_FORWARD.
*/
protected void initializeReplicationMode(HAGroupStoreRecord record) throws
IOException {
HAGroupState haGroupState = record.getHAGroupState();
LOG.info("HAGroup {} initializing mode from state {}", this, haGroupState);
- if (haGroupState.equals(HAGroupState.ACTIVE_IN_SYNC)) {
+ if (
+ haGroupState.equals(HAGroupState.ACTIVE_IN_SYNC)
+ || haGroupState.equals(HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY)
+ ) {
setMode(SYNC);
} else {
setMode(STORE_AND_FORWARD);
@@ -750,6 +925,20 @@ public class ReplicationLogGroup {
return closed.get();
}
+ /**
+ * Set or clear the failover-pending flag. Set on entry to the in-sync
cutover gate
+ * (ACTIVE_IN_SYNC_TO_STANDBY); cleared when the state returns to
ACTIVE_IN_SYNC (cutover abort).
+ * While set, log rotation is suspended and a SYNC-write failure aborts the
RS.
+ */
+ public void setFailoverPending(boolean pending) {
+ failoverPending.set(pending);
+ }
+
+ /** Returns true while the local cluster is in the in-sync cutover gate. */
+ public boolean isFailoverPending() {
+ return failoverPending.get();
+ }
+
/**
* Close the ReplicationLogGroup. Drains pending events from the Disruptor
with a bounded timeout
* (which triggers onShutdown → modeImpl.onExit → ReplicationLog.close),
then cleans up all
@@ -776,6 +965,10 @@ public class ReplicationLogGroup {
+ "proceeding with shutdown without the start gate", this, e);
Thread.currentThread().interrupt();
}
+ // Remove every LOCAL state subscription registered in
subscribeToStateChanges().
+ stateUnsubscribers.forEach(Runnable::run);
+ stateUnsubscribers.clear();
+
try {
disruptor.shutdown(shutdownTimeoutMs, TimeUnit.MILLISECONDS);
} catch (com.lmax.disruptor.TimeoutException e) {
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/SyncModeImpl.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/SyncModeImpl.java
index ca258a5b76..d05b9687cb 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/SyncModeImpl.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/SyncModeImpl.java
@@ -56,6 +56,17 @@ public class SyncModeImpl extends ReplicationModeImpl {
@Override
ReplicationMode onFailure(Throwable e) throws IOException {
LOG.info("HAGroup {} mode={} got error", logGroup, this, e);
+ if (logGroup.isFailoverPending()) {
+ // During the in-sync cutover gate we cannot fall back to
STORE_AND_FORWARD: that transition
+ // targets ACTIVE_NOT_IN_SYNC, which is not allowed from
ACTIVE_IN_SYNC_TO_STANDBY, and the
+ // active has no self-abort lever. Fail-stop instead — the thrown
exception propagates to the
+ // event handler, which fails the pending syncs so the blocked producer
aborts the RS. This
+ // preserves the invariant that a locally-committed mutation is never
silently lost.
+ String message = String
+ .format("HAGroup %s SYNC write failed during cutover; aborting region
server", logGroup);
+ LOG.error(message, e);
+ throw ReplicationLogGroup.asIOException(message, e);
+ }
return transitionToStoreAndForward();
}
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 95df33a5d3..fa3bcc565f 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
@@ -138,7 +138,8 @@ public abstract class ReplicationLogGroupBaseIT extends
HABaseIT {
private ReplicationLogGroup getReplicationLogGroup() throws IOException {
HRegionServer rs =
CLUSTERS.getHBaseCluster1().getHBaseCluster().getRegionServer(0);
- return ReplicationLogGroup.get(conf1, rs.getServerName(), haGroupName);
+ return ReplicationLogGroup.get(conf1, rs.getServerName(), haGroupName)
+ .orElseThrow(() -> new IOException("No active ReplicationLogGroup for "
+ haGroupName));
}
protected Map<String, List<Mutation>> groupLogsByTable() throws Exception {
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 fc4cc9f563..23cbc0996b 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
@@ -55,6 +55,7 @@ import org.apache.hadoop.hbase.util.JVMClusterUtil;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.hbase.index.IndexRegionObserver;
+import org.apache.phoenix.jdbc.ClusterRoleRecord;
import org.apache.phoenix.jdbc.FailoverPhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixResultSet;
import org.apache.phoenix.query.PhoenixTestBuilder;
@@ -1213,6 +1214,133 @@ public class ReplicationLogGroupIT extends
ReplicationLogGroupBaseIT {
}
}
+ /**
+ * A replication log group is a writer and must not outlive the active role.
This drives a real
+ * LOCAL demotion (cluster 1 ACTIVE -> STANDBY, cluster 2 STANDBY -> ACTIVE)
through the actual
+ * HAGroupStoreClient/ZooKeeper path -- the delivery that the unit tests
mock out -- and asserts
+ * the writer created for the active role is torn down. A subsequent get()
must fail fast against
+ * the now-STANDBY role rather than re-create a writer.
+ */
+ @Test
+ public void testWriterClosesOnDemotionToStandby() throws Exception {
+ final String tableName = "T_" + generateUniqueName();
+ String createTableDdl = String.format(
+ "create table if not exists %s (id integer not null primary key, val
varchar)", tableName);
+
+ // Write on the active so the lazily-created writer is real and running.
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(createTableDdl);
+ conn.commit();
+ PreparedStatement stmt = conn.prepareStatement("upsert into " +
tableName + " VALUES(?, ?)");
+ for (int i = 0; i < 10; i++) {
+ stmt.setInt(1, i);
+ stmt.setString(2, "v" + i);
+ stmt.executeUpdate();
+ }
+ conn.commit();
+ }
+ assertFalse("Writer should be open while the local cluster is active",
logGroup.isClosed());
+
+ // Demote cluster 1 to STANDBY (cluster 2 becomes ACTIVE). The RS's
HAGroupStoreClient observes
+ // the ZK role change and fires the LOCAL demotion listener registered by
the log group.
+ ServerName sn =
+
CLUSTERS.getHBaseCluster1().getHBaseCluster().getRegionServer(0).getServerName();
+ CLUSTERS.transitClusterRole(haGroup, ClusterRoleRecord.ClusterRole.STANDBY,
+ ClusterRoleRecord.ClusterRole.ACTIVE);
+
+ // close() is handed to a daemon thread by the listener, so poll for
completion.
+ long deadline = System.currentTimeMillis() + 30000;
+ while (!logGroup.isClosed() && System.currentTimeMillis() < deadline) {
+ Threads.sleep(200);
+ }
+ assertTrue("Writer must be closed after demotion to STANDBY",
logGroup.isClosed());
+
+ // The role gate must now refuse to re-create a writer on the demoted
(STANDBY) cluster: get()
+ // returns empty rather than caching a writer for a non-active role.
+ assertFalse("get() should return empty on a STANDBY cluster",
+ ReplicationLogGroup.get(conf1, sn, haGroupName).isPresent());
+ }
+
+ /**
+ * The graceful-failover deadlock fix: while the local cluster sits in the
in-sync cutover gate
+ * (ACTIVE_IN_SYNC_TO_STANDBY), time-based log rotation must be suspended so
no new .plog files
+ * are dropped into the peer's shard directory — otherwise the standby's
failover-completion check
+ * (getNewFiles(...).isEmpty()) never holds and both clusters spin forever.
This drives the real
+ * ZK role change through the RS's HAGroupStoreClient (the delivery the unit
tests mock) and
+ * asserts: (1) entering the cutover gate suspends rotation — the peer .plog
count is unchanged
+ * across several rounds while the writer stays open; (2) aborting back to
ACTIVE_IN_SYNC resumes
+ * rotation — new files appear again.
+ */
+ @Test
+ public void testCutoverSuspendsAndResumesRotation() throws Exception {
+ final String tableName = "T_" + generateUniqueName();
+ String createTableDdl = String.format(
+ "create table if not exists %s (id integer not null primary key, val
varchar)", tableName);
+
+ // Write on the active so a real SYNC-mode writer is open and dropping
files into the peer dir.
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(createTableDdl);
+ conn.commit();
+ PreparedStatement stmt = conn.prepareStatement("upsert into " +
tableName + " VALUES(?, ?)");
+ for (int i = 0; i < 10; i++) {
+ stmt.setInt(1, i);
+ stmt.setString(2, "v" + i);
+ stmt.executeUpdate();
+ }
+ conn.commit();
+ }
+ assertFalse("Writer should be open while the local cluster is active",
logGroup.isClosed());
+
+ Path peerDir =
logGroup.getOrCreatePeerShardManager().getRootDirectoryPath();
+ FileSystem peerFs = peerDir.getFileSystem(conf2);
+ long roundMs = TimeUnit.SECONDS.toMillis(5); //
PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY
+
+ // Enter the in-sync cutover gate (cluster 1 ACTIVE_TO_STANDBY ->
ACTIVE_IN_SYNC_TO_STANDBY).
+ CLUSTERS.transitClusterRole(haGroup,
ClusterRoleRecord.ClusterRole.ACTIVE_TO_STANDBY,
+ ClusterRoleRecord.ClusterRole.STANDBY_TO_ACTIVE);
+
+ // The RS's HAGroupStoreClient observes the ZK role change and fires the
LOCAL listener.
+ long deadline = System.currentTimeMillis() + 30000;
+ while (!logGroup.isFailoverPending() && System.currentTimeMillis() <
deadline) {
+ Threads.sleep(200);
+ }
+ assertTrue("Cutover must set failover-pending",
logGroup.isFailoverPending());
+ assertFalse("Group must stay open during cutover", logGroup.isClosed());
+
+ // Snapshot after suspension is active, then wait several rounds; the
count must not grow. The
+ // base class disables the replay reader on both clusters
(PHOENIX_REPLICATION_REPLAY_ENABLED =
+ // false); replay only runs via the explicit replayAndVerifyAcrossClusters
helper, which this
+ // test never calls. So nothing consumes or deletes files from the peer
dir — the count only
+ // ever grows via the active's rotation, making both assertions below
deterministic.
+ int filesAtCutover = findLogFiles(peerDir, peerFs).size();
+ Threads.sleep(3 * roundMs);
+ assertEquals("Rotation must be suspended during cutover: no new peer .plog
files",
+ filesAtCutover, findLogFiles(peerDir, peerFs).size());
+ assertFalse("Writer must stay open through the suspended window",
logGroup.isClosed());
+
+ // Abort the failover back to ACTIVE_IN_SYNC; rotation must resume on the
same live group.
+ CLUSTERS.transitClusterRole(haGroup, ClusterRoleRecord.ClusterRole.ACTIVE,
+ ClusterRoleRecord.ClusterRole.STANDBY);
+ deadline = System.currentTimeMillis() + 30000;
+ while (logGroup.isFailoverPending() && System.currentTimeMillis() <
deadline) {
+ Threads.sleep(200);
+ }
+ assertFalse("Cutover abort must clear failover-pending",
logGroup.isFailoverPending());
+
+ // With rotation resumed, subsequent rounds mint new files again.
+ deadline = System.currentTimeMillis() + 3 * roundMs + 10000;
+ while (
+ findLogFiles(peerDir, peerFs).size() <= filesAtCutover
+ && System.currentTimeMillis() < deadline
+ ) {
+ Threads.sleep(200);
+ }
+ assertTrue("Rotation must resume after the abort clears failover-pending",
+ findLogFiles(peerDir, peerFs).size() > filesAtCutover);
+ }
+
private PhoenixTestBuilder.SchemaBuilder createViewHierarchy() throws
Exception {
// Define the test schema.
// 1. Table with columns => (ORG_ID, KP, COL1, COL2, COL3), PK => (ORG_ID,
KP)
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
index 7a7ae208a0..22db791b73 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
@@ -19,11 +19,13 @@ package org.apache.phoenix.replication;
import static java.lang.Thread.sleep;
import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.STORE_AND_FORWARD;
+import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.SYNC;
import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
@@ -34,6 +36,7 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
@@ -60,6 +63,8 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.phoenix.jdbc.ClusterType;
+import org.apache.phoenix.jdbc.HAGroupStateListener;
import org.apache.phoenix.jdbc.HAGroupStoreManager;
import org.apache.phoenix.jdbc.HAGroupStoreRecord;
import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
@@ -73,6 +78,7 @@ import
org.apache.phoenix.replication.log.LogFileWriterContext;
import org.apache.phoenix.replication.metrics.ReplicationLogMetricValues;
import org.junit.Assume;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
@@ -1312,9 +1318,9 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Get instances for the first HA group
ReplicationLogGroup g1_1 =
- ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager);
+ ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager).get();
ReplicationLogGroup g1_2 =
- ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager);
+ ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager).get();
// Verify same instance is returned for same haGroupId
assertNotNull("ReplicationLogGroup should not be null", g1_1);
@@ -1324,16 +1330,16 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Get instance for a different HA group
ReplicationLogGroup g2_1 =
- ReplicationLogGroup.get(conf, serverName, haGroupId2,
haGroupStoreManager);
+ ReplicationLogGroup.get(conf, serverName, haGroupId2,
haGroupStoreManager).get();
assertNotNull("ReplicationLogGroup should not be null", g2_1);
assertTrue("Different instance should be returned for different
haGroupId", g2_1 != g1_1);
assertEquals("HA Group name should match", haGroupId2,
g2_1.getHAGroupName());
// Verify multiple calls still return cached instances
ReplicationLogGroup g1_3 =
- ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager);
+ ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager).get();
ReplicationLogGroup g2_2 =
- ReplicationLogGroup.get(conf, serverName, haGroupId2,
haGroupStoreManager);
+ ReplicationLogGroup.get(conf, serverName, haGroupId2,
haGroupStoreManager).get();
assertTrue("Cached instance should be returned", g1_3 == g1_1);
assertTrue("Cached instance should be returned", g2_2 == g2_1);
@@ -1352,13 +1358,13 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Get initial instance
ReplicationLogGroup g1_1 =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
assertNotNull("ReplicationLogGroup should not be null", g1_1);
assertFalse("Group should not be closed initially", g1_1.isClosed());
// Verify cached instance is returned
ReplicationLogGroup g1_2 =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
assertTrue("Same instance should be returned before close", g1_2 == g1_1);
// Close the group
@@ -1367,7 +1373,7 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Get instance after close - should be a new instance
ReplicationLogGroup g1_3 =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
assertNotNull("ReplicationLogGroup should not be null after close", g1_3);
assertFalse("New group should not be closed", g1_3.isClosed());
assertTrue("New instance should be created after close", g1_1 != g1_3);
@@ -2467,4 +2473,254 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
assertTrue("fsSyncTime should be > 0, got " + values.getFsSyncTimeMax(),
values.getFsSyncTimeMax() > 0);
}
+
+ /**
+ * A replication log group is a writer and must exist only where the local
cluster is active.
+ * get() must return empty (and not cache) on a non-active role (here
STANDBY), before any writer
+ * resource is constructed — no group instance, no forwarder, no demotion
subscription.
+ */
+ @Test
+ public void testGetReturnsEmptyOnStandby() throws Exception {
+ // Use a fresh HA group + its own mock manager so the never()
verifications observe only this
+ // get() attempt, independent of the active logGroup created by
setUpBase().
+ final String standbyGroup = haGroupName + "-standby";
+ HAGroupStoreManager standbyManager =
Mockito.mock(HAGroupStoreManager.class);
+ HAGroupStoreRecord standbyRecord = new HAGroupStoreRecord(null,
standbyGroup,
+ HAGroupState.STANDBY, 0, HighAvailabilityPolicy.FAILOVER.toString(),
"peerZKUrl",
+ "clusterUrl", "peerClusterUrl", localUri.toString(), peerUri.toString(),
0L);
+ doReturn(Optional.of(standbyRecord)).when(standbyManager)
+ .getEffectiveHAGroupStoreRecord(anyString());
+
+ Optional<ReplicationLogGroup> group =
+ ReplicationLogGroup.get(conf, serverName, standbyGroup, standbyManager);
+ assertFalse("get() must return empty on a non-active (STANDBY) role",
group.isPresent());
+
+ // No writer resources should have been created: the role gate runs before
construction, so no
+ // group exists to subscribe any state listener.
+ verify(standbyManager, never()).subscribeToTargetState(eq(standbyGroup),
+ any(HAGroupState.class), eq(ClusterType.LOCAL),
any(HAGroupStateListener.class));
+ }
+
+ /**
+ * ACTIVE_IN_SYNC_TO_STANDBY is the in-sync cutover gate (role
ACTIVE_TO_STANDBY, mutations
+ * blocked). init() reaches it not from the live write path (blocked) but
from preWALRestore
+ * recovery on an RS restart / region reassignment during cutover. It must
NOT fail fast: throwing
+ * would abort WAL replay and silently fail to re-ship already-committed
edits to the peer.
+ * Instead the writer starts in SYNC with failoverPending already set, so
recovery ships its edits
+ * while rotation stays suspended (no new files that would re-arm the
standby deadlock).
+ */
+ @Test
+ public void testInitInCutoverStartsSyncWithFailoverPending() throws
Exception {
+ final String cutoverGroup = haGroupName + "-cutover";
+ HAGroupStoreManager cutoverManager =
Mockito.mock(HAGroupStoreManager.class);
+ HAGroupStoreRecord cutoverRecord = new HAGroupStoreRecord(null,
cutoverGroup,
+ HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY, 0,
HighAvailabilityPolicy.FAILOVER.toString(),
+ "peerZKUrl", "clusterUrl", "peerClusterUrl", localUri.toString(),
peerUri.toString(), 0L);
+ doReturn(Optional.of(cutoverRecord)).when(cutoverManager)
+ .getEffectiveHAGroupStoreRecord(anyString());
+
+ ReplicationLogGroup group =
+ new TestableLogGroup(conf, serverName, cutoverGroup, cutoverManager,
useAlignedRotation());
+ group.init();
+ try {
+ assertEquals("Cutover init must start in SYNC", SYNC, group.getMode());
+ assertTrue("Cutover init must seed failover-pending so rotation stays
suspended",
+ group.isFailoverPending());
+ } finally {
+ group.close();
+ }
+ }
+
+ /**
+ * On the live demotion path (an existing writer created while
ACTIVE_IN_SYNC), entering the
+ * in-sync cutover gate (ACTIVE_IN_SYNC_TO_STANDBY) must set failoverPending
— which suspends
+ * rotation — but must NOT close the group: a straggler write that committed
locally before the
+ * mutation block still needs the open writer to reach the peer.
+ */
+ @Test
+ public void testCutoverSetsFailoverPending() throws Exception {
+ ArgumentCaptor<HAGroupStateListener> captor =
+ ArgumentCaptor.forClass(HAGroupStateListener.class);
+ verify(haGroupStoreManager).subscribeToTargetState(eq(haGroupName),
+ eq(HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY), eq(ClusterType.LOCAL),
captor.capture());
+ HAGroupStateListener demotionListener = captor.getValue();
+ assertNotNull("A LOCAL ACTIVE_IN_SYNC_TO_STANDBY listener should be
registered",
+ demotionListener);
+ assertFalse("Group should not be failover-pending before cutover",
+ logGroup.isFailoverPending());
+
+ demotionListener.onStateChange(haGroupName, HAGroupState.ACTIVE_IN_SYNC,
+ HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY, 0L, ClusterType.LOCAL, 0L);
+
+ assertTrue("Cutover must set failover-pending",
logGroup.isFailoverPending());
+ // Rotation is suspended: forceRotation() stages no new writer, and the
current writer stays
+ // open.
+ ReplicationLog activeLog = logGroup.getActiveLog();
+ LogFileWriter writerBefore = activeLog.getWriter();
+ activeLog.forceRotation();
+ assertEquals("Rotation must be suppressed while failover-pending",
writerBefore,
+ activeLog.getWriter());
+ assertFalse("Group must stay open during cutover", logGroup.isClosed());
+ }
+
+ /**
+ * With failoverPending set, rotation is suspended: even an explicit
forceRotation() must be a
+ * no-op, leaving the same writer open so an in-flight append + sync still
lands on it (the
+ * straggler mutation reaches the peer instead of hitting
IOException("Closed") or a fresh file).
+ */
+ @Test
+ public void testCutoverKeepsWriterOpenForInflight() throws Exception {
+ final String tableName = "TBLCUT";
+ final Mutation put = LogFileTestUtil.newPut("row", 1, 1);
+
+ ReplicationLog activeLog = logGroup.getActiveLog();
+ LogFileWriter writerBefore = activeLog.getWriter();
+ logGroup.setFailoverPending(true);
+
+ // Rotation is suspended: forceRotation() stages no new writer.
+ activeLog.forceRotation();
+ assertSame("Cutover must keep the same writer open, not rotate to a new
one", writerBefore,
+ activeLog.getWriter());
+
+ // The in-flight write still lands on that open writer.
+ logGroup.append(tableName, 1L, put);
+ logGroup.sync();
+
+ assertFalse("Group must stay open during cutover", logGroup.isClosed());
+ assertSame("Append + sync must not have rotated the writer", writerBefore,
+ activeLog.getWriter());
+ verify(writerBefore, times(1)).sync();
+ }
+
+ /**
+ * A cutover abort (ACTIVE_IN_SYNC_TO_STANDBY → ABORT_TO_ACTIVE_IN_SYNC →
ACTIVE_IN_SYNC) must
+ * clear failoverPending on the same live group, resuming rotation.
+ */
+ @Test
+ public void testAbortClearsFailoverPending() throws Exception {
+ ArgumentCaptor<HAGroupStateListener> captor =
+ ArgumentCaptor.forClass(HAGroupStateListener.class);
+ verify(haGroupStoreManager).subscribeToTargetState(eq(haGroupName),
+ eq(HAGroupState.ACTIVE_IN_SYNC), eq(ClusterType.LOCAL),
captor.capture());
+ HAGroupStateListener abortResumeListener = captor.getValue();
+
+ logGroup.setFailoverPending(true);
+ abortResumeListener.onStateChange(haGroupName,
HAGroupState.ABORT_TO_ACTIVE_IN_SYNC,
+ HAGroupState.ACTIVE_IN_SYNC, 0L, ClusterType.LOCAL, 0L);
+
+ assertFalse("Cutover abort must clear failover-pending",
logGroup.isFailoverPending());
+ // Rotation resumes: forceRotation() now stages a new writer.
+ ReplicationLog activeLog = logGroup.getActiveLog();
+ LogFileWriter writerBefore = activeLog.getWriter();
+ activeLog.forceRotation();
+ logGroup.append("TBLRES", 1L, LogFileTestUtil.newPut("row", 1, 1));
+ logGroup.sync();
+ assertNotEquals("Rotation must resume after abort clears the flag",
writerBefore,
+ activeLog.getWriter());
+ }
+
+ /**
+ * A SYNC write that fails while failoverPending must NOT transition to
STORE_AND_FORWARD (that
+ * targets ACTIVE_NOT_IN_SYNC, an illegal transition from the cutover gate).
Instead it
+ * fail-stops: the sync throws and the group aborts, never calling
+ * setHAGroupStatusToStoreAndForward.
+ */
+ @Test
+ public void testCutoverSyncFailureAborts() throws Exception {
+ final String tableName = "TBLCUTFAIL";
+ final Mutation put = LogFileTestUtil.newPut("row", 1, 1);
+
+ ReplicationLog activeLog = logGroup.getActiveLog();
+ LogFileWriter initialWriter = activeLog.getWriter();
+ doThrow(new IOException("Simulated sync
failure")).when(initialWriter).sync();
+
+ logGroup.setFailoverPending(true);
+ logGroup.append(tableName, 1L, put);
+ try {
+ logGroup.sync();
+ fail("Sync failure during cutover should abort, not fall back to
STORE_AND_FORWARD");
+ } catch (IOException expected) {
+ // expected — the fail-stop path
+ }
+
+ verify(haGroupStoreManager,
never()).setHAGroupStatusToStoreAndForward(haGroupName);
+ assertNotEquals("Must not transition to STORE_AND_FORWARD during cutover",
STORE_AND_FORWARD,
+ logGroup.getMode());
+ }
+
+ /**
+ * A group created while active must not outlive the active role. When the
LOCAL cluster is
+ * demoted to STANDBY, the subscribed listener closes the writer group (on a
background thread,
+ * per the non-blocking listener contract).
+ */
+ @Test
+ public void testDemotionToStandbyClosesGroup() throws Exception {
+ // Capture the LOCAL STANDBY listener that init() registered for the
default (active) logGroup.
+ ArgumentCaptor<HAGroupStateListener> captor =
+ ArgumentCaptor.forClass(HAGroupStateListener.class);
+ verify(haGroupStoreManager).subscribeToTargetState(eq(haGroupName),
eq(HAGroupState.STANDBY),
+ eq(ClusterType.LOCAL), captor.capture());
+ HAGroupStateListener demotionListener = captor.getValue();
+ assertNotNull("A LOCAL STANDBY listener should have been registered",
demotionListener);
+ assertFalse("Group should be open before demotion", logGroup.isClosed());
+
+ // Fire the demotion event. close() is handed to a daemon thread, so poll
for completion.
+ demotionListener.onStateChange(haGroupName,
HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY,
+ HAGroupState.STANDBY, 0L, ClusterType.LOCAL, 0L);
+
+ long deadline = System.currentTimeMillis() + 5000;
+ while (!logGroup.isClosed() && System.currentTimeMillis() < deadline) {
+ sleep(20);
+ }
+ assertTrue("Group should be closed after demotion to STANDBY",
logGroup.isClosed());
+ }
+
+ /**
+ * A peer-blind demotion surfaces as LOCAL DEGRADED_STANDBY (the bare
STANDBY notification is
+ * suppressed while the peer is not visible). The same listener must tear
the writer down so it
+ * does not outlive the active role through the degraded window.
+ */
+ @Test
+ public void testDemotionToDegradedStandbyClosesGroup() throws Exception {
+ ArgumentCaptor<HAGroupStateListener> captor =
+ ArgumentCaptor.forClass(HAGroupStateListener.class);
+ verify(haGroupStoreManager).subscribeToTargetState(eq(haGroupName),
+ eq(HAGroupState.DEGRADED_STANDBY), eq(ClusterType.LOCAL),
captor.capture());
+ HAGroupStateListener demotionListener = captor.getValue();
+ assertNotNull("A LOCAL DEGRADED_STANDBY listener should have been
registered",
+ demotionListener);
+ assertFalse("Group should be open before demotion", logGroup.isClosed());
+
+ demotionListener.onStateChange(haGroupName,
HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY,
+ HAGroupState.DEGRADED_STANDBY, 0L, ClusterType.LOCAL, 0L);
+
+ long deadline = System.currentTimeMillis() + 5000;
+ while (!logGroup.isClosed() && System.currentTimeMillis() < deadline) {
+ sleep(20);
+ }
+ assertTrue("Group should be closed after demotion to DEGRADED_STANDBY",
logGroup.isClosed());
+ }
+
+ /**
+ * close() must unsubscribe every LOCAL state listener registered during
init() —
+ * ACTIVE_IN_SYNC_TO_STANDBY / STANDBY / DEGRADED_STANDBY (demotion),
ACTIVE_IN_SYNC (abort resume
+ * + SYNC mode flip), and ACTIVE_NOT_IN_SYNC (mode flip) — so a group closed
on demotion leaks no
+ * listeners holding a reference to the dead group.
+ */
+ @Test
+ public void testCloseUnsubscribesListeners() throws Exception {
+ logGroup.close();
+
+ verify(haGroupStoreManager).unsubscribeFromTargetState(eq(haGroupName),
+ eq(HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY), eq(ClusterType.LOCAL),
+ any(HAGroupStateListener.class));
+ verify(haGroupStoreManager).unsubscribeFromTargetState(eq(haGroupName),
+ eq(HAGroupState.STANDBY), eq(ClusterType.LOCAL),
any(HAGroupStateListener.class));
+ verify(haGroupStoreManager).unsubscribeFromTargetState(eq(haGroupName),
+ eq(HAGroupState.DEGRADED_STANDBY), eq(ClusterType.LOCAL),
any(HAGroupStateListener.class));
+ verify(haGroupStoreManager).unsubscribeFromTargetState(eq(haGroupName),
+ eq(HAGroupState.ACTIVE_NOT_IN_SYNC), eq(ClusterType.LOCAL),
any(HAGroupStateListener.class));
+ verify(haGroupStoreManager).unsubscribeFromTargetState(eq(haGroupName),
+ eq(HAGroupState.ACTIVE_IN_SYNC), eq(ClusterType.LOCAL),
any(HAGroupStateListener.class));
+ }
}
