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 1f64927d19 PHOENIX-8025 Skip the consistent-failover replication log
for PARALLEL HA groups (#2648)
1f64927d19 is described below
commit 1f64927d19e696c46a9aebec0781b74fb1d6a6ba
Author: tkhurana <[email protected]>
AuthorDate: Wed Oct 28 12:05:16 2026 -0700
PHOENIX-8025 Skip the consistent-failover replication log for PARALLEL HA
groups (#2648)
A PARALLEL HA group does not need the replication log: the client writes
both
clusters itself. ReplicationLogGroup.get() now always returns a group whose
Status tells the caller what to do: REPLICATING (live writer), SKIP (no
replication log), or NOT_ACTIVE (FAILOVER group on a non-active cluster).
- A PARALLEL group gets a cached NoOpReplicationLogGroup on either role, so
it
never builds a writer (no peer-FS files, rotation, mode flips or abort
exposure), including at RS startup eager init, and later batches do not
re-read the HA group store. The policy is checked before the role.
- NOT_ACTIVE is a shared instance that is never cached, so the first get()
after promotion builds the writer. Non-replicated writes use the shared
NONE.
- getOrCreate checks INSTANCES before computeIfAbsent.
- IndexRegionObserver ships only when the group isReplicated(), throws
split-brain on NOT_ACTIVE, and drops isParallelPolicy. preWALRestore skips
any non-replicating group, which covers PARALLEL edits stamped before this
change.
Also close ReplicationLog when init() fails to create the first writer, so a
failed init() no longer leaves an orphaned rotation thread or staged writer.
Supersedes #2638; its IndexRegionObserver unit tests and PARALLEL IT case
are
carried over.
Generated-by: Claude Code (Opus 5.5)
---
.../coprocessor/PhoenixRegionServerEndpoint.java | 8 +-
.../phoenix/hbase/index/IndexRegionObserver.java | 142 +++++-------
.../apache/phoenix/replication/ReplicationLog.java | 10 +-
.../phoenix/replication/ReplicationLogGroup.java | 253 +++++++++++++++------
.../apache/phoenix/jdbc/HAGroupStoreManagerIT.java | 4 +-
.../replication/ReplicationLogGroupBaseIT.java | 8 +-
.../phoenix/replication/ReplicationLogGroupIT.java | 7 +-
.../replication/ReplicationLogGroupParallelIT.java | 61 ++++-
.../reader/StoreAndForwardFailoverIT.java | 2 +-
...IndexRegionObserverReplicationLogGroupTest.java | 232 +++++++++++++++++++
.../replication/ReplicationLogGroupTest.java | 168 +++++++++++---
11 files changed, 696 insertions(+), 199 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 c68ee31ddc..d271ec5fa0 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
@@ -325,10 +325,8 @@ public class PhoenixRegionServerEndpoint extends
+ " 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);
+ LOGGER.info("Skipping eager ReplicationLogGroup init for HA
group: {} on server {};"
+ + " the group will be created lazily on first mutation.",
haGroup, serverName);
}
iterator.remove();
LOGGER.info("Prewarmed HAGroupStoreClient: {} ({} remaining)",
haGroup,
@@ -354,7 +352,7 @@ 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.
+ * lazily on first mutation.
*/
private boolean shouldEagerlyInit(HAGroupStoreManager manager, String
haGroup) {
if (serverName == null) {
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 f80da44167..a11b5ec144 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,8 +136,6 @@ import org.apache.phoenix.index.PhoenixIndexBuilderHelper;
import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.index.PhoenixIndexMetaData;
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;
@@ -505,7 +503,7 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// derived from it. Defaults to CLIENT_NON_HA (benign: no HA handling).
private BatchOrigin origin = BatchOrigin.CLIENT_NON_HA;
// HAGroup associated with the batch
- private Optional<ReplicationLogGroup> logGroup = Optional.empty();
+ private ReplicationLogGroup logGroup = ReplicationLogGroup.NONE;
// Per-(row, ts) groups folded from this replicated batch, computed once
and shared by the
// global-index path (prepareReplicatedIndexMutations) and the local-index
path. Null until
// first built; only populated on the standby replay path (isReplay).
@@ -697,6 +695,12 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
this.dataTableName = dataTableName;
}
+ @VisibleForTesting
+ IndexRegionObserver(String dataTableName, boolean replicateTable) {
+ this(dataTableName);
+ this.replicateTable = replicateTable;
+ }
+
@Override
public Optional<RegionObserver> getRegionObserver() {
return Optional.of(this);
@@ -957,9 +961,9 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
}
- // Resolve the replication writer for this batch (empty if none should
be used). This also
- // enforces the split-brain guard for non-PARALLEL live writes to a
non-active cluster.
- Optional<ReplicationLogGroup> logGroup =
+ // Resolve the replication log group for this batch. This also enforces
the split-brain
+ // guard for non-PARALLEL live writes to a non-active cluster.
+ ReplicationLogGroup logGroup =
getReplicationLogGroup(c.getEnvironment(), haGroupStoreManager,
origin, haGroupName);
preBatchMutateWithExceptions(c, miniBatchOp, logGroup, origin);
return;
@@ -971,39 +975,32 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
/**
- * Resolve the replication writer for this batch, or empty when none should
be used. Returns empty
+ * Resolve the replication log group for this batch. Returns {@link
ReplicationLogGroup#NONE}
* unless this is a CLIENT_HA write on a replication-eligible cluster
({@code replicateTable} =
* feature on AND table eligible) — which covers the feature-off rollout
window and non-eligible
- * tables. For such a write it then answers the role/policy question: the
writer when this cluster
- * is active for the group, empty for a non-active PARALLEL cluster
(legitimate double-write, skip
- * re-ship), and throws for a non-active non-PARALLEL (split-brain) live
write.
- * @return the HA group writer, or empty if none should be used
+ * tables. For such a write the group answers the role/policy question: a
writer
+ * ({@code REPLICATING}) when this cluster is active for a non-PARALLEL
group, a no-op group
+ * ({@code SKIP}) for a PARALLEL group on either role (the client writes
both clusters itself),
+ * and a non-active non-PARALLEL (split-brain) live write throws.
+ * @return the HA group's replication log group, never null
*/
- private Optional<ReplicationLogGroup>
getReplicationLogGroup(RegionCoprocessorEnvironment env,
+ @VisibleForTesting
+ ReplicationLogGroup getReplicationLogGroup(RegionCoprocessorEnvironment env,
HAGroupStoreManager haGroupStoreManager, BatchOrigin origin,
Optional<String> haGroupName)
throws IOException {
if (!replicateTable || origin != BatchOrigin.CLIENT_HA) {
- return Optional.empty();
+ return ReplicationLogGroup.NONE;
}
// origin == CLIENT_HA guarantees the HA group name is present.
String haGroup = haGroupName.get();
- Optional<ReplicationLogGroup> logGroup =
ReplicationLogGroup.get(env.getConfiguration(),
+ ReplicationLogGroup logGroup =
ReplicationLogGroup.get(env.getConfiguration(),
env.getServerName(), haGroup, haGroupStoreManager, abortable);
- if (logGroup.isPresent()) {
+ if (logGroup.getStatus() != ReplicationLogGroup.Status.NOT_ACTIVE) {
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();
- }
+ // This cluster is not active for a non-PARALLEL HA group. Under FAILOVER
only the active
+ // cluster takes writes, so this is a stray/stale sync-path write to a
demoted cluster. Reject
+ // it rather than committing locally unreplicated (split brain).
throw new IOException(
String.format("HAGroup %s cannot accept a HA write on %s: this cluster
is not active",
haGroup, dataTableName));
@@ -1061,41 +1058,25 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
return haGroupName.isPresent() ? BatchOrigin.CLIENT_HA :
BatchOrigin.CLIENT_NON_HA;
}
- /**
- * 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.
- * @return HA group if present or empty if missing
+ * @return the HA group's replication log group, or {@link
ReplicationLogGroup#NONE} if the key
+ * carries no HA group
*/
- private Optional<ReplicationLogGroup>
getHAGroupFromWALKey(RegionCoprocessorEnvironment env,
+ private ReplicationLogGroup
getHAGroupFromWALKey(RegionCoprocessorEnvironment env,
Map<String, byte[]> walKeyAttrs) throws IOException {
byte[] haGroupName =
walKeyAttrs.get(BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB);
if (haGroupName != null) {
- // 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.
+ // get() returns a NOT_ACTIVE group 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. A PARALLEL
+ // group (including edits stamped before PARALLEL skipped the log)
returns SKIP. Either way
+ // the caller skips 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();
+ return ReplicationLogGroup.NONE;
}
@Override
@@ -1110,14 +1091,13 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
return;
}
Map<String, byte[]> walKeyAttrs = getAttributeValuesFromWALKey(logKey);
- Optional<ReplicationLogGroup> logGroup =
- getHAGroupFromWALKey(ctx.getEnvironment(), walKeyAttrs);
- if (!logGroup.isPresent()) {
+ ReplicationLogGroup logGroup = getHAGroupFromWALKey(ctx.getEnvironment(),
walKeyAttrs);
+ if (!logGroup.isReplicated()) {
return;
}
long start = EnvironmentEdgeManager.currentTimeMillis();
try {
- replicateEditOnWALRestore(logGroup.get(), logKey, walKeyAttrs, logEdit);
+ replicateEditOnWALRestore(logGroup, logKey, walKeyAttrs, logEdit);
} finally {
long duration = EnvironmentEdgeManager.currentTimeMillis() - start;
metricSource.updatePreWALRestoreTime(dataTableName, duration);
@@ -1825,23 +1805,24 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
/**
- * True when this batch must be replicated from this cluster: a resolved HA
log group is present,
- * which (per its sole resolution site) already means a CLIENT_HA write on a
replication-eligible
- * cluster that is active for the group. Gates the active-side pre-image
capture — the pre-image
- * exists only so the standby can regenerate its index, so capturing it on a
non-replicated batch
- * would be wasted work (and an unnecessary region scan on the local path) —
and the WAL-key
- * replication stamp. It intentionally does NOT gate on {@code
ignoreSyncReplicationForTesting}:
- * that flag suppresses only the synchronous ship (checked explicitly at the
ship site) to
- * simulate an RS crash after the WAL write but before the ship, leaving
pre-image capture, the
- * WAL stamp, and the replay path intact so crash-recovery re-ship is
faithfully exercised.
+ * True when this batch must be replicated from this cluster: the resolved
HA log group is
+ * replicating, which (per its sole resolution site) already means a
CLIENT_HA write on a
+ * replication-eligible cluster that is active for a non-PARALLEL group.
Gates the active-side
+ * pre-image capture — the pre-image exists only so the standby can
regenerate its index, so
+ * capturing it on a non-replicated batch would be wasted work (and an
unnecessary region scan on
+ * the local path) — and the WAL-key replication stamp. It intentionally
does NOT gate on
+ * {@code ignoreSyncReplicationForTesting}: that flag suppresses only the
synchronous ship
+ * (checked explicitly at the ship site) to simulate an RS crash after the
WAL write but before
+ * the ship, leaving pre-image capture, the WAL stamp, and the replay path
intact so
+ * crash-recovery re-ship is faithfully exercised.
*/
private boolean isReplicatedBatch(BatchMutateContext context) {
- // The log group is resolved (in preBatchMutate) only for a CLIENT_HA
write on a
- // replication-eligible cluster that is active for the group, and that
resolution site is its
- // sole assignment — so its presence alone is the full predicate for "ship
this batch". The
- // "present implies CLIENT_HA" half of that is asserted at context setup in
+ // A replicating log group is resolved (in preBatchMutate) only for a
CLIENT_HA write on a
+ // replication-eligible cluster that is active for a non-PARALLEL group,
and that resolution
+ // site is its sole assignment — so isReplicated() alone is the full
predicate for "ship this
+ // batch". The "replicated implies CLIENT_HA" half of that is asserted at
context setup in
// preBatchMutateWithExceptions.
- return context.logGroup.isPresent();
+ return context.logGroup.isReplicated();
}
/**
@@ -2723,19 +2704,18 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
public void
preBatchMutateWithExceptions(ObserverContext<RegionCoprocessorEnvironment> c,
- MiniBatchOperationInProgress<Mutation> miniBatchOp,
Optional<ReplicationLogGroup> logGroup,
+ MiniBatchOperationInProgress<Mutation> miniBatchOp, ReplicationLogGroup
logGroup,
BatchOrigin origin) throws Throwable {
PhoenixIndexMetaData indexMetaData = getPhoenixIndexMetaData(c,
miniBatchOp);
BatchMutateContext context = new
BatchMutateContext(indexMetaData.getClientVersion());
context.origin = origin;
context.logGroup = logGroup;
- // A resolved log group is produced (in preBatchMutate) only for a
CLIENT_HA write on a
- // replication-eligible cluster active for the group, so its presence must
imply CLIENT_HA. This
- // is the invariant isReplicatedBatch() relies on to key "ship this batch"
off log-group
- // presence
- // alone; assert it here so a future resolution change can't silently
break the ship predicate.
- Preconditions.checkState(!logGroup.isPresent() || origin ==
BatchOrigin.CLIENT_HA,
- "log group present but batch origin is %s, not CLIENT_HA", origin);
+ // A replicating log group is produced (in preBatchMutate) only for a
CLIENT_HA write on a
+ // replication-eligible cluster active for the group, so it must imply
CLIENT_HA. This is the
+ // invariant isReplicatedBatch() relies on to key "ship this batch" off
isReplicated() alone;
+ // assert it here so a future resolution change can't silently break the
ship predicate.
+ Preconditions.checkState(!logGroup.isReplicated() || origin ==
BatchOrigin.CLIENT_HA,
+ "log group replicating but batch origin is %s, not CLIENT_HA", origin);
setBatchMutateContext(c, context);
identifyIndexMaintainerTypes(indexMetaData, context);
identifyMutationTypes(miniBatchOp, context);
@@ -3006,8 +2986,8 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
* keys (it is in {@code ReplicationLogGroup.REPLICATION_ATTR_KEYS}), so it
needs no separate
* stamp. Gated on {@link #isReplicatedBatch} — the same predicate that
gates pre-image capture —
* so the WAL carries this stamp exactly when it carries the pre-images the
re-ship needs. On a
- * standby replay the log group is absent, so the standby's WAL never
carries replication metadata
- * it could never re-ship.
+ * standby replay the log group is NONE (not replicated), so the standby's
WAL never carries
+ * replication metadata it could never re-ship.
*/
private void appendReplicationAttributesToWALKey(WALKey key,
IndexRegionObserver.BatchMutateContext context) {
@@ -3095,7 +3075,7 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// ignoreSyncReplicationForTesting suppresses only this synchronous
ship, simulating an RS
// crash after the WAL write but before the ship; the WAL-restore path
re-ships on replay.
if (isReplicatedBatch(context) && !ignoreSyncReplicationForTesting) {
- replicateMutations(context.logGroup.get(), miniBatchOp, context);
+ replicateMutations(context.logGroup, miniBatchOp, context);
}
FutureUtils.get(postIndexFuture);
}
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 25fd8450e4..855f6f2c02 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
@@ -139,8 +139,16 @@ public class ReplicationLog {
/** Initialize the writer. */
public void init() throws IOException {
+ // Start rotation first so the first tick is aligned to the round in which
init began, even if
+ // creating the writer crosses a round boundary.
startRotationExecutor();
- currentWriter = createNewWriter();
+ try {
+ currentWriter = createNewWriter();
+ } catch (IOException | RuntimeException e) {
+ // Nothing owns a log whose init failed: stop rotation and close any
writer a tick staged.
+ close(false);
+ throw e;
+ }
generation = currentWriter.getGeneration();
}
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 b1751f11c5..a738066e72 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
@@ -75,6 +75,7 @@ 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;
+import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
import org.apache.phoenix.replication.metrics.MetricsReplicationLogGroupSource;
import
org.apache.phoenix.replication.metrics.MetricsReplicationLogGroupSourceFactory;
import org.slf4j.Logger;
@@ -240,6 +241,25 @@ public class ReplicationLogGroup {
protected static final ConcurrentHashMap<String, ReplicationLogGroup>
INSTANCES =
new ConcurrentHashMap<>();
+ /**
+ * What a write to this group should do with the replication log.
+ */
+ public enum Status {
+ /** A live writer: ship the batch through the replication log. */
+ REPLICATING,
+ /** No replication log for this write (PARALLEL group, or not a replicated
write). */
+ SKIP,
+ /** FAILOVER group whose local role is not active: this cluster must not
take the write. */
+ NOT_ACTIVE
+ }
+
+ /** Shared group for writes that are not replicated (non-CLIENT_HA origin,
ineligible table). */
+ public static final ReplicationLogGroup NONE = new
NoOpReplicationLogGroup(Status.SKIP);
+
+ /** Shared group returned for a FAILOVER group on a non-active cluster.
Never cached. */
+ private static final ReplicationLogGroup NOT_ACTIVE =
+ new NoOpReplicationLogGroup(Status.NOT_ACTIVE);
+
/**
* Canonical, DNS-independent identity of a RegionServer within this JVM's
{@link #INSTANCES}
* cache. Uses {@code (port, startcode)} and deliberately EXCLUDES the
hostname.
@@ -470,30 +490,30 @@ public class ReplicationLogGroup {
protected LogEventHandler eventHandler;
/**
- * 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}).
+ * Get or create a ReplicationLogGroup instance for the given HA Group.
Never null; check
+ * {@link #getStatus()} (see {@link #getOrCreate}).
* @param conf Configuration object
* @param serverName The server name
* @param haGroupName The HA Group name
- * @return ReplicationLogGroup instance, or empty if this cluster is not
active
+ * @return ReplicationLogGroup instance
* @throws IOException if initialization fails
*/
- public static Optional<ReplicationLogGroup> get(Configuration conf,
ServerName serverName,
+ public static 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.
Returns empty when this
- * cluster is not active for the HA group (see {@link #getOrCreate}).
+ * Get or create a ReplicationLogGroup instance for the given HA Group.
Never null; check
+ * {@link #getStatus()} (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, or empty if this cluster is not
active
+ * @return ReplicationLogGroup instance
* @throws IOException if initialization fails
*/
- public static Optional<ReplicationLogGroup> get(Configuration conf,
ServerName serverName,
+ public static ReplicationLogGroup get(Configuration conf, ServerName
serverName,
String haGroupName, Abortable abortable) throws IOException {
return getOrCreate(conf, serverName, haGroupName,
HAGroupStoreManager.getInstance(conf),
abortable);
@@ -505,83 +525,98 @@ public class ReplicationLogGroup {
* @param serverName The server name
* @param haGroupName The HA Group name
* @param haGroupStoreManager HA Group Store Manager instance
- * @return ReplicationLogGroup instance, or empty if this cluster is not
active
+ * @return ReplicationLogGroup instance
* @throws IOException if initialization fails
*/
- public static Optional<ReplicationLogGroup> get(Configuration conf,
ServerName serverName,
+ public static ReplicationLogGroup get(Configuration conf, ServerName
serverName,
String haGroupName, HAGroupStoreManager haGroupStoreManager) throws
IOException {
return getOrCreate(conf, serverName, haGroupName, haGroupStoreManager,
null);
}
/**
- * 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}). Used
by callers that already
- * hold the store manager and want fatal errors to abort via the supplied
{@link Abortable}.
+ * Get or create a ReplicationLogGroup instance for the given HA Group.
Never null; check
+ * {@link #getStatus()} (see {@link #getOrCreate}). Used by callers that
already hold the store
+ * manager and want fatal errors to abort via the supplied {@link Abortable}.
* @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 (typically
RegionServerServices)
- * @return ReplicationLogGroup instance, or empty if this cluster is not
active
+ * @return ReplicationLogGroup instance
* @throws IOException if initialization fails
*/
- public static Optional<ReplicationLogGroup> get(Configuration conf,
ServerName serverName,
+ public static ReplicationLogGroup get(Configuration conf, ServerName
serverName,
String haGroupName, HAGroupStoreManager haGroupStoreManager, Abortable
abortable)
throws IOException {
return getOrCreate(conf, serverName, haGroupName, haGroupStoreManager,
abortable);
}
/**
- * 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.
+ * Get an existing, or create and initialize a new, ReplicationLogGroup. A
PARALLEL group gets a
+ * cached {@link NoOpReplicationLogGroup} ({@link Status#SKIP}) on either
role: the client writes
+ * both clusters itself. Otherwise a replication log group is a WRITER and
must exist only where
+ * this cluster is the active side. When the local role is not active this
returns the shared
+ * {@link Status#NOT_ACTIVE} group 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 {
+ private static ReplicationLogGroup getOrCreate(Configuration conf,
ServerName serverName,
+ String haGroupName, HAGroupStoreManager haGroupStoreManager, Abortable
abortable)
+ throws IOException {
+ String key = instanceKey(serverName, haGroupName);
+ // Fast path: computeIfAbsent may lock the bucket even when the key is
present.
+ ReplicationLogGroup cached = INSTANCES.get(key);
+ if (cached != null) {
+ return cached;
+ }
try {
- 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);
+ ReplicationLogGroup group = INSTANCES.computeIfAbsent(key, k -> {
+ try {
+ // Policy, then role, 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. A PARALLEL group
+ // never uses the replication log, so it gets a cached no-op group
whatever the role. 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)));
}
- });
- return Optional.ofNullable(group);
+ if
(HighAvailabilityPolicy.PARALLEL.name().equalsIgnoreCase(haRecord.get().getPolicy()))
{
+ LOG.info("HA Group {} uses the PARALLEL policy; not creating a
replication log writer",
+ haGroupName);
+ return new NoOpReplicationLogGroup(serverName, 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, not NOT_ACTIVE: computeIfAbsent does not cache a
null, so the next
+ // get() re-reads the store and builds the writer once this
cluster is promoted. A
+ // cached NOT_ACTIVE would stick, because policy and role are only
read on a miss.
+ 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);
+ }
+ });
+ // null = not active and deliberately left uncached (see the lambda).
+ return group != null ? group : NOT_ACTIVE;
} catch (UncheckedIOException e) {
throw e.getCause();
}
@@ -657,6 +692,21 @@ public class ReplicationLogGroup {
this.mode = new AtomicReference<>(INIT);
}
+ /**
+ * Light constructor for {@link NoOpReplicationLogGroup}: no conf clone,
peer conf or metrics
+ * source, since a no-op group never writes.
+ */
+ private ReplicationLogGroup(ServerName serverName, String haGroupName) {
+ this.conf = null;
+ this.peerConf = null;
+ this.serverName = serverName;
+ this.haGroupName = haGroupName;
+ this.haGroupStoreManager = null;
+ this.abortable = null;
+ this.metrics = null;
+ this.mode = new AtomicReference<>(INIT);
+ }
+
/**
* Build the peer (standby) FS configuration: a clone of {@code base} with
the given overrides
* (prefix already stripped) applied on top. The applied override keys are
logged so an operator
@@ -717,8 +767,8 @@ public class ReplicationLogGroup {
// 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.
+ // cutover suspend/resume, and mode flips). getOrCreate() builds a writer
only on an active
+ // role, so 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,
@@ -830,10 +880,10 @@ public class ReplicationLogGroup {
}
/**
- * 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,
+ * Initialize the replication mode based on the HAGroupStore state. {@link
#getOrCreate} builds a
+ * writer only on an 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 {
@@ -1045,6 +1095,16 @@ public class ReplicationLogGroup {
}
}
+ /** What a write to this group should do with the replication log. */
+ public Status getStatus() {
+ return Status.REPLICATING;
+ }
+
+ /** True when writes to this group must be shipped through the replication
log. */
+ public boolean isReplicated() {
+ return getStatus() == Status.REPLICATING;
+ }
+
/**
* Publish a non-blocking mode-check marker so an idle consumer applies a
mode switch promptly.
* Uses {@code tryNext()} like {@link #publishSwapEvent()}: when the ring
buffer is full, the
@@ -1265,6 +1325,63 @@ public class ReplicationLogGroup {
return getHAGroupName();
}
+ /**
+ * A group with no replication log writer. A per-group instance ({@link
Status#SKIP}) is cached
+ * for a PARALLEL HA group on either role, so its writes never re-read the
store. The shared
+ * {@link #NONE} and {@link #NOT_ACTIVE} instances have no HA group and are
never cached. No
+ * disruptor, forwarder, shard managers or state subscriptions are created.
+ */
+ static final class NoOpReplicationLogGroup extends ReplicationLogGroup {
+ private final Status status;
+
+ /** Per-group PARALLEL instance, cached in {@link #INSTANCES}. */
+ NoOpReplicationLogGroup(ServerName serverName, String haGroupName) {
+ super(serverName, haGroupName);
+ this.status = Status.SKIP;
+ }
+
+ /** Shared instance with no server or HA group. */
+ private NoOpReplicationLogGroup(Status status) {
+ super(null, "<" + status + ">");
+ this.status = status;
+ }
+
+ @Override
+ public Status getStatus() {
+ return status;
+ }
+
+ @Override
+ protected void init() {
+ }
+
+ @Override
+ public void append(String tableName, long commitId, Mutation mutation) {
+ throw new IllegalStateException("HAGroup " + this + " has no replication
log writer");
+ }
+
+ @Override
+ public void append(String tableName, long commitId, List<Cell> cells,
+ Map<String, byte[]> attributes) {
+ throw new IllegalStateException("HAGroup " + this + " has no replication
log writer");
+ }
+
+ @Override
+ public void sync() {
+ throw new IllegalStateException("HAGroup " + this + " has no replication
log writer");
+ }
+
+ @Override
+ public void close() {
+ // The shared instances are never cached and must stay usable.
+ if (serverName == null || !closed.compareAndSet(false, true)) {
+ return;
+ }
+ INSTANCES.remove(instanceKey(serverName, haGroupName), this);
+ LOG.info("HAGroup {} closed (no replication log writer)", this);
+ }
+ }
+
/**
* Creates the top level directory on the cluster determined by the URI
* @param uri HDFS uri
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 cb9f790d51..85d29ed872 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
@@ -759,8 +759,8 @@ public class HAGroupStoreManagerIT extends HABaseIT {
// Start the forwarder only now; it finds no backlog and promotes the
group back to
// ACTIVE_IN_SYNC.
HRegionServer regionServer =
CLUSTERS.getHBaseCluster1().getHBaseCluster().getRegionServer(0);
- replicationLogGroup = ReplicationLogGroup
- .get(cluster1Conf, regionServer.getServerName(), haGroupName,
cluster1HAManager).get();
+ replicationLogGroup = ReplicationLogGroup.get(cluster1Conf,
regionServer.getServerName(),
+ haGroupName, cluster1HAManager);
waitForHAGroupState(cluster1HAManager, haGroupName,
HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
waitForHAGroupState(cluster2HAManager, haGroupName,
HAGroupStoreRecord.HAGroupState.STANDBY);
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 e00a046d0a..c363e4c1e9 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
@@ -140,8 +140,12 @@ 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)
- .orElseThrow(() -> new IOException("No active ReplicationLogGroup for "
+ haGroupName));
+ // A PARALLEL group returns its no-op SKIP group: there is no writer to
create.
+ ReplicationLogGroup group = ReplicationLogGroup.get(conf1,
rs.getServerName(), haGroupName);
+ if (group.getStatus() == ReplicationLogGroup.Status.NOT_ACTIVE) {
+ throw new IOException("No active ReplicationLogGroup for " +
haGroupName);
+ }
+ return group;
}
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 b72634458d..727af83a00 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
@@ -1687,9 +1687,10 @@ public class ReplicationLogGroupIT extends
ReplicationLogGroupBaseIT {
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());
+ // returns NOT_ACTIVE rather than caching a writer for a non-active role.
+ assertEquals("get() should return NOT_ACTIVE on a STANDBY cluster",
+ ReplicationLogGroup.Status.NOT_ACTIVE,
+ ReplicationLogGroup.get(conf1, sn, haGroupName).getStatus());
}
/**
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupParallelIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupParallelIT.java
index ae3b4047bf..14edc2ec0c 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupParallelIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupParallelIT.java
@@ -19,13 +19,19 @@ package org.apache.phoenix.replication;
import static org.apache.phoenix.query.BaseTest.generateUniqueName;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.jdbc.HAGroupStoreManager;
+import org.apache.phoenix.jdbc.HAGroupStoreRecord;
import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
import org.apache.phoenix.jdbc.ParallelPhoenixConnection;
import org.junit.BeforeClass;
@@ -35,11 +41,11 @@ import org.junit.experimental.categories.Category;
/**
* PARALLEL-policy coverage for the write-path role/policy branch in
* {@code IndexRegionObserver.getReplicationLogGroup}. Under PARALLEL the
client double-writes to
- * both clusters via a {@link ParallelPhoenixConnection}; the leg that lands
on the non-active
- * cluster must commit locally and skip re-ship (the {@code isParallelPolicy}
branch returns empty)
- * rather than reject the write as split-brain — which is the FAILOVER
behavior covered by
- * {@link ReplicationLogGroupIT}. Both clusters run with the replication
feature enabled, so the
- * standby actually reaches this branch.
+ * both clusters via a {@link ParallelPhoenixConnection}, so each leg commits
locally through the
+ * group's no-op {@code SKIP} log group: the non-active leg is not rejected as
split-brain (the
+ * FAILOVER behavior covered by {@link ReplicationLogGroupIT}) and the active
leg is not shipped to
+ * the peer. Both clusters run with the replication feature enabled, so both
legs actually reach
+ * this branch.
*/
@Category(NeedsOwnMiniClusterTest.class)
public class ReplicationLogGroupParallelIT extends ReplicationLogGroupBaseIT {
@@ -66,8 +72,7 @@ public class ReplicationLogGroupParallelIT extends
ReplicationLogGroupBaseIT {
CLUSTERS.createTableOnClusterPair(haGroup, tableName, false);
// PARALLEL policy => a ParallelPhoenixConnection that writes to both
clusters. The leg landing
- // on the non-active cluster must commit locally (isParallelPolicy
branch), not throw
- // split-brain.
+ // on the non-active cluster must commit locally (SKIP group), not throw
split-brain.
try (Connection conn =
DriverManager.getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
assertTrue("expected a ParallelPhoenixConnection under PARALLEL policy",
conn instanceof ParallelPhoenixConnection);
@@ -86,6 +91,48 @@ public class ReplicationLogGroupParallelIT extends
ReplicationLogGroupBaseIT {
assertRowCount(CLUSTERS.getCluster2Connection(haGroup), tableName,
rowCount);
}
+ @Test
+ public void testParallelWriteOnActiveClusterIsNotShippedToReplicationLog()
throws Exception {
+ final String tableName = "T_" + generateUniqueName();
+ final int rowCount = 10;
+ CLUSTERS.createTableOnClusterPair(haGroup, tableName, false);
+
+ try (Connection conn =
DriverManager.getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ assertTrue("expected a ParallelPhoenixConnection under PARALLEL policy",
+ conn instanceof ParallelPhoenixConnection);
+ PreparedStatement stmt = conn.prepareStatement("upsert into " +
tableName + " values (?, ?)");
+ for (int i = 0; i < rowCount; i++) {
+ stmt.setInt(1, i);
+ stmt.setInt(2, i);
+ stmt.executeUpdate();
+ }
+ conn.commit();
+ }
+ assertRowCount(CLUSTERS.getCluster1Connection(haGroup), tableName,
rowCount);
+ assertRowCount(CLUSTERS.getCluster2Connection(haGroup), tableName,
rowCount);
+
+ // Every RS on the active cluster holds the PARALLEL no-op group, never a
writer.
+ for (RegionServerThread rst : CLUSTERS.getHBaseCluster1().getHBaseCluster()
+ .getRegionServerThreads()) {
+ ReplicationLogGroup group =
+ ReplicationLogGroup.get(conf1, rst.getRegionServer().getServerName(),
haGroupName);
+ assertEquals(ReplicationLogGroup.Status.SKIP, group.getStatus());
+ }
+ // No writer was ever built for this group: an in-sync writer creates both
its local fallback
+ // (out) and peer (in) dirs, so neither may exist.
+ HAGroupStoreRecord record =
+
HAGroupStoreManager.getInstance(conf1).getHAGroupStoreRecord(haGroupName).get();
+ assertNoLogDir(conf1, record.getHdfsUrl(),
ReplicationLogGroup.FALLBACK_DIR);
+ assertNoLogDir(conf2, record.getPeerHdfsUrl(),
ReplicationLogGroup.STANDBY_DIR);
+ }
+
+ private void assertNoLogDir(Configuration conf, String hdfsUrl, String
logDirName)
+ throws Exception {
+ Path dir = new Path(new Path(hdfsUrl, haGroupName), logDirName);
+ assertFalse("PARALLEL group must not create replication log dir " + dir,
+ dir.getFileSystem(conf).exists(dir));
+ }
+
private static void assertRowCount(Connection conn, String tableName, int
expected)
throws Exception {
try (Connection c = conn;
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
index 1e45d29ae0..b03a018f6d 100644
---
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
@@ -127,7 +127,7 @@ public class StoreAndForwardFailoverIT extends HABaseIT {
// 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).get();
+ logGroup = ReplicationLogGroup.get(conf1, rs.getServerName(), haGroupName);
// Replicated table (REPLICATION_SCOPE=1 + LOCAL INDEX) on BOTH clusters.
tableName = "T_" + generateUniqueName();
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexRegionObserverReplicationLogGroupTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexRegionObserverReplicationLogGroupTest.java
new file mode 100644
index 0000000000..b816303976
--- /dev/null
+++
b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexRegionObserverReplicationLogGroupTest.java
@@ -0,0 +1,232 @@
+/*
+ * 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.hbase.index;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Optional;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.wal.WALEdit;
+import org.apache.hadoop.hbase.wal.WALKey;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
+import org.apache.phoenix.hbase.index.IndexRegionObserver.BatchOrigin;
+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;
+import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
+import org.apache.phoenix.replication.ReplicationLogGroup;
+import org.apache.phoenix.replication.ReplicationLogGroup.Status;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestName;
+
+/**
+ * Unit tests for the role/policy decision in {@link
IndexRegionObserver#getReplicationLogGroup}:
+ * which live write batches get a replication log writer, which commit locally
without one, and
+ * which are rejected as split-brain. Also covers the WAL-restore path for
PARALLEL groups.
+ */
+public class IndexRegionObserverReplicationLogGroupTest {
+
+ @Rule
+ public TemporaryFolder localFolder = new TemporaryFolder();
+ @Rule
+ public TemporaryFolder peerFolder = new TemporaryFolder();
+ @Rule
+ public TestName name = new TestName();
+
+ private Configuration conf;
+ private ServerName serverName;
+ private RegionCoprocessorEnvironment env;
+ private HAGroupStoreManager haGroupStoreManager;
+ private String haGroupName;
+
+ @Before
+ public void setUp() {
+ conf = HBaseConfiguration.create();
+ conf.setInt(ReplicationLogGroup.REPLICATION_LOG_RINGBUFFER_SIZE_KEY, 32);
+ conf.setLong(ReplicationLogGroup.REPLICATION_LOG_SYNC_TIMEOUT_KEY, 1000);
+ serverName = ServerName.valueOf("test", 60010,
EnvironmentEdgeManager.currentTimeMillis());
+ env = mock(RegionCoprocessorEnvironment.class);
+ when(env.getConfiguration()).thenReturn(conf);
+ when(env.getServerName()).thenReturn(serverName);
+ haGroupStoreManager = mock(HAGroupStoreManager.class);
+ haGroupName = name.getMethodName();
+ }
+
+ private void stubRecord(HAGroupState state, HighAvailabilityPolicy policy)
throws IOException {
+ HAGroupStoreRecord record = new HAGroupStoreRecord(null, haGroupName,
state, 0,
+ policy.toString(), "peerZKUrl", "clusterUrl", "peerClusterUrl",
+ new Path(localFolder.getRoot().toString()).toUri().toString(),
+ new Path(peerFolder.getRoot().toString()).toUri().toString(), 0L);
+ doReturn(Optional.of(record)).when(haGroupStoreManager)
+ .getEffectiveHAGroupStoreRecord(anyString());
+ }
+
+ private ReplicationLogGroup resolve(IndexRegionObserver iro, BatchOrigin
origin,
+ Optional<String> group) throws IOException {
+ return iro.getReplicationLogGroup(env, haGroupStoreManager, origin, group);
+ }
+
+ private ReplicationLogGroup resolveClientHa() throws IOException {
+ return resolve(new IndexRegionObserver("T", true), BatchOrigin.CLIENT_HA,
+ Optional.of(haGroupName));
+ }
+
+ /** No writer was constructed: a live group subscribes to LOCAL role changes
in init(). */
+ private void verifyNoWriterCreated() throws Exception {
+ verify(haGroupStoreManager, never()).subscribeToTargetState(anyString(),
+ any(HAGroupState.class), any(ClusterType.class),
any(HAGroupStateListener.class));
+ }
+
+ private void assertParallelSkips(HAGroupState state) throws Exception {
+ stubRecord(state, HighAvailabilityPolicy.PARALLEL);
+ ReplicationLogGroup logGroup = resolveClientHa();
+ try {
+ assertEquals("PARALLEL write must not ship through the replication log",
Status.SKIP,
+ logGroup.getStatus());
+ assertEquals(haGroupName, logGroup.getHAGroupName());
+ // Cached: later batches do not re-read the store.
+ assertSame(logGroup, resolveClientHa());
+ verify(haGroupStoreManager,
times(1)).getEffectiveHAGroupStoreRecord(haGroupName);
+ verifyNoWriterCreated();
+ } finally {
+ logGroup.close();
+ }
+ }
+
+ @Test
+ public void testParallelOnActiveSkips() throws Exception {
+ // PARALLEL clients write both clusters themselves, so the active must not
ship the batch
+ // through the replication log (no append, no pre-image, no WAL-key stamp).
+ assertParallelSkips(HAGroupState.ACTIVE_IN_SYNC);
+ }
+
+ @Test
+ public void testParallelOnStandbySkips() throws Exception {
+ assertParallelSkips(HAGroupState.STANDBY);
+ }
+
+ @Test
+ public void testFailoverOnActiveReplicates() throws Exception {
+ stubRecord(HAGroupState.ACTIVE_IN_SYNC, HighAvailabilityPolicy.FAILOVER);
+ ReplicationLogGroup logGroup = resolveClientHa();
+ try {
+ assertEquals(Status.REPLICATING, logGroup.getStatus());
+ assertEquals(haGroupName, logGroup.getHAGroupName());
+ } finally {
+ logGroup.close();
+ }
+ }
+
+ @Test
+ public void testFailoverOnStandbyThrows() throws Exception {
+ stubRecord(HAGroupState.STANDBY, HighAvailabilityPolicy.FAILOVER);
+ try {
+ resolveClientHa();
+ fail("FAILOVER write on a non-active cluster must be rejected as
split-brain");
+ } catch (IOException expected) {
+ assertTrue(expected.getMessage(), expected.getMessage().contains("not
active"));
+ }
+ verifyNoWriterCreated();
+ }
+
+ @Test
+ public void testMissingRecordFails() throws Exception {
+ // With no record the policy is unknown, so resolution fails rather than
committing.
+ doReturn(Optional.empty()).when(haGroupStoreManager)
+ .getEffectiveHAGroupStoreRecord(anyString());
+ try {
+ resolveClientHa();
+ fail("a missing HA group record must not silently skip replication");
+ } catch (IOException expected) {
+ assertTrue(expected.getMessage(), expected.getMessage().contains("empty
group store record"));
+ }
+ verifyNoWriterCreated();
+ }
+
+ @Test
+ public void testNonClientHaOriginsReturnNoneWithoutLookup() throws Exception
{
+ IndexRegionObserver iro = new IndexRegionObserver("T", true);
+ assertSame(ReplicationLogGroup.NONE, resolve(iro,
BatchOrigin.CLIENT_NON_HA, Optional.empty()));
+ assertSame(ReplicationLogGroup.NONE, resolve(iro, BatchOrigin.NATIVE_IN,
Optional.empty()));
+ assertSame(ReplicationLogGroup.NONE, resolve(iro, BatchOrigin.PHX_REPLAY,
Optional.empty()));
+ verifyNoInteractions(haGroupStoreManager);
+ }
+
+ @Test
+ public void testNonReplicatedTableReturnsNoneWithoutLookup() throws
Exception {
+ IndexRegionObserver iro = new IndexRegionObserver("T", false);
+ assertSame(ReplicationLogGroup.NONE,
+ resolve(iro, BatchOrigin.CLIENT_HA, Optional.of(haGroupName)));
+ verifyNoInteractions(haGroupStoreManager);
+ }
+
+ /**
+ * A WAL edit stamped with a PARALLEL group (e.g. written before PARALLEL
skipped the log) must
+ * not be re-shipped on WAL restore, and must not create a writer.
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreWALRestoreSkipsParallelGroup() throws Exception {
+ stubRecord(HAGroupState.ACTIVE_IN_SYNC, HighAvailabilityPolicy.PARALLEL);
+ // Prime the cache with the PARALLEL no-op group; preWALRestore resolves
through the same
+ // (server, HA group) cache entry.
+ ReplicationLogGroup logGroup = resolveClientHa();
+ try {
+ ObserverContext<RegionCoprocessorEnvironment> ctx =
mock(ObserverContext.class);
+ when(ctx.getEnvironment()).thenReturn(env);
+ WALKey logKey = mock(WALKey.class);
+ when(logKey.getExtendedAttributes()).thenReturn(Collections.singletonMap(
+ BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB,
Bytes.toBytes(haGroupName)));
+ WALEdit logEdit = mock(WALEdit.class);
+
+ new IndexRegionObserver("T", true).preWALRestore(ctx, null, logKey,
logEdit);
+
+ // A re-ship would read the edit's cells; no interaction proves
preWALRestore skipped it.
+ verifyNoInteractions(logEdit);
+ assertSame(logGroup, resolveClientHa());
+ } finally {
+ logGroup.close();
+ }
+ }
+}
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 25dbf4b45f..1e0854f2f3 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
@@ -1468,9 +1468,9 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Get instances for the first HA group
ReplicationLogGroup g1_1 =
- ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager);
ReplicationLogGroup g1_2 =
- ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager);
// Verify same instance is returned for same haGroupId
assertNotNull("ReplicationLogGroup should not be null", g1_1);
@@ -1480,16 +1480,16 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Get instance for a different HA group
ReplicationLogGroup g2_1 =
- ReplicationLogGroup.get(conf, serverName, haGroupId2,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId2,
haGroupStoreManager);
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).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId1,
haGroupStoreManager);
ReplicationLogGroup g2_2 =
- ReplicationLogGroup.get(conf, serverName, haGroupId2,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId2,
haGroupStoreManager);
assertTrue("Cached instance should be returned", g1_3 == g1_1);
assertTrue("Cached instance should be returned", g2_2 == g2_1);
@@ -1516,9 +1516,8 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
ServerName.valueOf("10.244.1.25", serverName.getPort(),
serverName.getStartcode());
ReplicationLogGroup viaFqdn =
- ReplicationLogGroup.get(conf, asFqdn, haGroupId,
haGroupStoreManager).get();
- ReplicationLogGroup viaIp =
- ReplicationLogGroup.get(conf, asIp, haGroupId,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, asFqdn, haGroupId, haGroupStoreManager);
+ ReplicationLogGroup viaIp = ReplicationLogGroup.get(conf, asIp, haGroupId,
haGroupStoreManager);
assertSame("Two host spellings of the same (port, startcode) must share
one instance", viaFqdn,
viaIp);
@@ -1528,7 +1527,7 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
ServerName otherServer =
ServerName.valueOf(asIp.getHostname(), serverName.getPort() + 1,
serverName.getStartcode());
ReplicationLogGroup viaOther =
- ReplicationLogGroup.get(conf, otherServer, haGroupId,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, otherServer, haGroupId,
haGroupStoreManager);
assertNotEquals("A distinct port must yield a distinct instance", viaFqdn,
viaOther);
viaFqdn.close();
@@ -1545,13 +1544,13 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Get initial instance
ReplicationLogGroup g1_1 =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
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).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
assertTrue("Same instance should be returned before close", g1_2 == g1_1);
// Close the group
@@ -1560,7 +1559,7 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Get instance after close - should be a new instance
ReplicationLogGroup g1_3 =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
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);
@@ -1613,7 +1612,7 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// A get() in this window must return the SAME (closing) instance, not a
fresh one.
ReplicationLogGroup duringClose =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
assertSame("A get() during close()'s teardown must not resurrect a
second instance", group,
duringClose);
@@ -1624,7 +1623,7 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// After close() fully completes, the entry is gone, so a later get()
creates a fresh
// instance.
ReplicationLogGroup afterClose =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
assertNotSame("A get() after close() completes must create a new
instance", group,
afterClose);
afterClose.close();
@@ -1670,7 +1669,7 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Despite the thrown teardown, the cache entry must be gone: a fresh
get() creates a NEW,
// usable instance rather than returning the stranded closed one.
ReplicationLogGroup afterClose =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
assertNotSame("A stranded closed instance must not be returned after a
throwing close()", group,
afterClose);
assertFalse("Re-created instance must be usable, not closed",
afterClose.isClosed());
@@ -1710,7 +1709,7 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
// Cache entry removed: a fresh get() creates a NEW instance.
ReplicationLogGroup afterClose =
- ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager).get();
+ ReplicationLogGroup.get(conf, serverName, haGroupId,
haGroupStoreManager);
assertNotSame("close() must remove the cache entry even when unsubscribe
throws", group,
afterClose);
afterClose.close();
@@ -3058,31 +3057,142 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
return result;
}
+ private HAGroupStoreManager mockManagerFor(String group, HAGroupState state,
+ HighAvailabilityPolicy policy) throws IOException {
+ HAGroupStoreManager manager = Mockito.mock(HAGroupStoreManager.class);
+ stubEffectiveRecord(manager, group, state, policy);
+ return manager;
+ }
+
+ private void stubEffectiveRecord(HAGroupStoreManager manager, String group,
HAGroupState state,
+ HighAvailabilityPolicy policy) throws IOException {
+ HAGroupStoreRecord record = new HAGroupStoreRecord(null, group, state, 0,
policy.toString(),
+ "peerZKUrl", "clusterUrl", "peerClusterUrl", localUri.toString(),
peerUri.toString(), 0L);
+
doReturn(Optional.of(record)).when(manager).getEffectiveHAGroupStoreRecord(anyString());
+ }
+
/**
* 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.
+ * get() must return NOT_ACTIVE (and not cache) on a non-active role (here
STANDBY), before any
+ * writer resource is constructed — no forwarder, no demotion subscription.
Once the role flips to
+ * active the next get() re-reads the store and builds the writer.
*/
@Test
- public void testGetReturnsEmptyOnStandby() throws Exception {
+ public void testGetReturnsNotActiveOnStandbyUntilPromoted() 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());
+ HAGroupStoreManager standbyManager =
+ mockManagerFor(standbyGroup, HAGroupState.STANDBY,
HighAvailabilityPolicy.FAILOVER);
+ for (int i = 0; i < 3; i++) {
+ ReplicationLogGroup group =
+ ReplicationLogGroup.get(conf, serverName, standbyGroup,
standbyManager);
+ assertEquals(ReplicationLogGroup.Status.NOT_ACTIVE, group.getStatus());
+ assertFalse(group.isReplicated());
+ }
+ // Not cached: every call re-read the store.
+ verify(standbyManager,
times(3)).getEffectiveHAGroupStoreRecord(standbyGroup);
// 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));
+
+ // Promotion: the next get() builds the writer.
+ stubEffectiveRecord(standbyManager, standbyGroup,
HAGroupState.ACTIVE_IN_SYNC,
+ HighAvailabilityPolicy.FAILOVER);
+ ReplicationLogGroup promoted =
+ ReplicationLogGroup.get(conf, serverName, standbyGroup, standbyManager);
+ try {
+ assertEquals(ReplicationLogGroup.Status.REPLICATING,
promoted.getStatus());
+ assertEquals(standbyGroup, promoted.getHAGroupName());
+ verify(standbyManager,
atLeastOnce()).subscribeToTargetState(eq(standbyGroup),
+ any(HAGroupState.class), eq(ClusterType.LOCAL),
any(HAGroupStateListener.class));
+ } finally {
+ promoted.close();
+ }
+ }
+
+ /**
+ * A PARALLEL group never uses the replication log, so get() returns a
cached no-op SKIP group on
+ * either role: one store read across many calls, and no writer resources
(no state
+ * subscriptions).
+ */
+ @Test
+ public void testParallelGetReturnsCachedSkipGroup() throws Exception {
+ for (HAGroupState state : new HAGroupState[] { HAGroupState.ACTIVE_IN_SYNC,
+ HAGroupState.STANDBY }) {
+ final String parallelGroup = haGroupName + "-parallel-" + state;
+ HAGroupStoreManager manager =
+ mockManagerFor(parallelGroup, state, HighAvailabilityPolicy.PARALLEL);
+
+ ReplicationLogGroup first = ReplicationLogGroup.get(conf, serverName,
parallelGroup, manager);
+ try {
+ assertEquals(state.toString(), ReplicationLogGroup.Status.SKIP,
first.getStatus());
+ assertFalse(first.isReplicated());
+ assertEquals(parallelGroup, first.getHAGroupName());
+ for (int i = 0; i < 5; i++) {
+ assertSame(first, ReplicationLogGroup.get(conf, serverName,
parallelGroup, manager));
+ }
+ verify(manager,
times(1)).getEffectiveHAGroupStoreRecord(parallelGroup);
+ verify(manager, never()).subscribeToTargetState(anyString(),
any(HAGroupState.class),
+ any(ClusterType.class), any(HAGroupStateListener.class));
+ try {
+ first.sync();
+ fail("a no-op group must not accept a sync");
+ } catch (IllegalStateException expected) {
+ }
+ } finally {
+ first.close();
+ }
+ }
+ }
+
+ /** close() and closeAll() on a PARALLEL no-op group evict it, so the next
get() re-reads. */
+ @Test
+ public void testParallelSkipGroupCloseAndCloseAll() throws Exception {
+ final String parallelGroup = haGroupName + "-parallel";
+ HAGroupStoreManager manager =
+ mockManagerFor(parallelGroup, HAGroupState.ACTIVE_IN_SYNC,
HighAvailabilityPolicy.PARALLEL);
+
+ ReplicationLogGroup first = ReplicationLogGroup.get(conf, serverName,
parallelGroup, manager);
+ first.close();
+ assertTrue(first.isClosed());
+ ReplicationLogGroup second = ReplicationLogGroup.get(conf, serverName,
parallelGroup, manager);
+ assertNotSame("close() must evict the no-op group", first, second);
+
+ ReplicationLogGroup.closeAll(serverName);
+ assertTrue(second.isClosed());
+ ReplicationLogGroup third = ReplicationLogGroup.get(conf, serverName,
parallelGroup, manager);
+ try {
+ assertNotSame("closeAll() must evict the no-op group", second, third);
+ verify(manager, times(3)).getEffectiveHAGroupStoreRecord(parallelGroup);
+ } finally {
+ third.close();
+ }
+ }
+
+ /** The shared NONE group is not tied to any HA group and survives close().
*/
+ @Test
+ public void testSharedNoneGroupIgnoresClose() {
+ ReplicationLogGroup.NONE.close();
+ assertFalse(ReplicationLogGroup.NONE.isClosed());
+ assertEquals(ReplicationLogGroup.Status.SKIP,
ReplicationLogGroup.NONE.getStatus());
+ }
+
+ /** A ReplicationLog whose first writer cannot be created must not leave
rotation running. */
+ @Test
+ public void testFailedLogInitStopsRotationExecutor() throws Exception {
+ ReplicationLog log =
logGroup.createReplicationLog(logGroup.getLocalShardManager());
+ doThrow(new IOException("Simulated writer creation
failure")).when(log).createNewWriter();
+ try {
+ log.init();
+ fail("init() must fail when the first writer cannot be created");
+ } catch (IOException expected) {
+ }
+ assertTrue("A failed init must close the log", log.isClosed());
+ assertTrue("Rotation executor must be stopped after a failed init",
+ log.rotationExecutor == null || log.rotationExecutor.isTerminated());
}
/**
