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 b2411f57ca PHOENIX-8011 Classify batch origin and gate HA write-path
on the sync-replication master switch (#2627)
b2411f57ca is described below
commit b2411f57ca3d0620aba4613177321567986a3a27
Author: tkhurana <[email protected]>
AuthorDate: Thu Oct 1 11:02:40 2026 -0700
PHOENIX-8011 Classify batch origin and gate HA write-path on the
sync-replication master switch (#2627)
* PHOENIX-8011 Classify batch origin and gate HA write-path on the
sync-replication master switch
Replace the scattered per-batch re-derivation of a batch's HA disposition
with a
single BatchOrigin classification, gate all HA write-path semantics on the
SYNCHRONOUS_REPLICATION_ENABLED master switch, and add replication-log
commit-id
observability.
- BatchOrigin (CLIENT_HA, CLIENT_NON_HA, PHX_REPLAY, NATIVE_IN): classified
once
per batch and carried on BatchMutateContext as the single source of truth;
isReplay() is derived from it (no separate flag).
- Split shouldReplicate into syncReplicationEnabled (master switch) and
replicateTable (switch AND per-table eligibility). Gate the cluster-role
mutation-block / staleness check on the master switch so it never runs, or
resolves a not-yet-created HA group, while the feature is off -- makes the
rollout window safe.
- bypassedMutationBlockCount increments only for CLIENT_NON_HA (replay and
native-replicated-in excluded). getReplicationLogGroup owns the full
"which
writer" decision; isReplicatedBatch reduces to log-group presence, with a
checkState enforcing present => CLIENT_HA.
- Replication-log observability: carry the WAL seq id as each record's
commitId;
advance lastCommitId through a single appendRecord() path (normal and
post-failover replay) and log it at mode switches; log recordsRead on
standby
replay; drop the batchSize histogram.
- Correct the operator-facing bypassedMutationBlockCount JMX/javadoc to the
CLIENT_NON_HA-on-eligible-table definition.
- Add HA_GROUP_NAME_ATTRIB to ReplicationLogGroup.REPLICATION_ATTR_KEYS so
the HA
group name rides the replication envelope like schema/table/tenant; drop
the
special-case WAL-key stamp in IndexRegionObserver (the envelope loop now
carries
it, and the WAL-restore path reads it back via getHAGroupFromWALKey).
- Gate BaseScannerRegionObserver's scan-path HAGroup staleness check on the
SYNCHRONOUS_REPLICATION_ENABLED master switch, mirroring the
IndexRegionObserver
write-path gate, so it never runs while the feature is off (including the
rollout
window where clients send _HAGroupName before the HA group exists).
- Log a per-table WARN (once per table per RegionServer, unbounded set, no
TTL) when
a CLIENT_NON_HA batch reaches a replication-eligible table without
_HAGroupName, to
identify the write path that is missing the attribute.
---
.../coprocessor/BaseScannerRegionObserver.java | 11 +-
.../phoenix/hbase/index/IndexRegionObserver.java | 269 +++++++++++++++------
.../hbase/index/metrics/MetricsHaBypassSource.java | 50 ++--
.../phoenix/replication/ReplicationLogGroup.java | 46 ++--
.../replication/log/LogFileFormatWriter.java | 7 +-
.../metrics/MetricsReplicationLogGroupSource.java | 9 -
.../MetricsReplicationLogGroupSourceImpl.java | 11 +-
.../metrics/ReplicationLogMetricValues.java | 12 -
.../reader/ReplicationLogProcessor.java | 7 +-
...cReplicationRolloutFeatureOnMissingGroupIT.java | 117 +++++++++
.../end2end/SyncReplicationRolloutStep1IT.java | 108 +++++++++
.../index/BypassedMutationBlockMetricsIT.java | 9 +-
.../replication/ReplicationLogGroupBaseIT.java | 13 +-
.../phoenix/replication/ReplicationLogGroupIT.java | 81 +++++++
.../replication/ReplicationLogGroupParallelIT.java | 97 ++++++++
.../hbase/index/IndexRegionObserverReplayTest.java | 114 +++++++++
.../replication/ReplicationLogGroupTest.java | 12 +-
.../phoenix/replication/log/LogFileCodecTest.java | 3 +
.../MetricsReplicationLogGroupSourceImplTest.java | 4 +-
19 files changed, 811 insertions(+), 169 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
index 6d09ce9a90..4977548412 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
@@ -190,10 +190,19 @@ abstract public class BaseScannerRegionObserver
implements RegionObserver {
}
byte[] haGroupNameBytes =
scan.getAttribute(BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB);
- // Check if scan's HAGroup is stale on client. If yes, throw an exception.
+ // Check if scan's HAGroup is stale on client. If yes, throw an exception.
Gated on the
+ // sync-replication master switch (mirrors IndexRegionObserver's
write-path gate) so the check
+ // never runs while the feature is off — including the rollout window
where clients already send
+ // _HAGroupName but the HA group does not exist yet.
if (haGroupNameBytes != null) {
String haGroupName = Bytes.toString(haGroupNameBytes);
final Configuration conf = c.getEnvironment().getConfiguration();
+ if (
+ !conf.getBoolean(QueryServices.SYNCHRONOUS_REPLICATION_ENABLED,
+ QueryServicesOptions.DEFAULT_SYNCHRONOUS_REPLICATION_ENABLED)
+ ) {
+ return;
+ }
final HAGroupStoreManager haGroupStoreManager =
HAGroupStoreManager.getInstance(conf);
if (haGroupStoreManager == null) {
throw new IOException(
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 3cc6310196..f80da44167 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
@@ -48,6 +48,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
@@ -196,6 +197,11 @@ import
org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
public class IndexRegionObserver implements RegionCoprocessor, RegionObserver {
private static final Logger LOG =
LoggerFactory.getLogger(IndexRegionObserver.class);
+ // Tables already warned about for a CLIENT_NON_HA write on a
replication-eligible table (a writer
+ // not attaching the _HAGroupName attribute). Each table is logged at most
once per RegionServer
+ // lifetime so a high-volume untagged writer cannot flood the log. Shared
across all per-region
+ // instances; bounded in practice by the finite set of tables on the
RegionServer.
+ private static final Set<String> BYPASS_WARNED_TABLES =
ConcurrentHashMap.newKeySet();
private static final OperationStatus IGNORE = new OperationStatus(SUCCESS);
private static final OperationStatus NOWRITE = new OperationStatus(SUCCESS);
public static final String PHOENIX_APPEND_METADATA_TO_WAL =
"phoenix.append.metadata.to.wal";
@@ -263,9 +269,10 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
public static final boolean DEFAULT_PHOENIX_INDEX_CDC_MUTATION_SERIALIZE =
false;
// Generic marker attribute set on every mutation produced by the standby
reader from a
// replication-log record. Value is opaque (presence is the signal).
Detected in
- // preBatchMutateWithExceptions to set context.isReplication, which gates
the primary-side clock
- // work (getBatchTimestamp/setTimestamps) off so the cell timestamps shipped
from the active
- // cluster are preserved. Never set on primary-side mutations.
+ // preBatchMutateWithExceptions to classify the batch as PHX_REPLAY (see
context.isReplay()),
+ // which
+ // gates the primary-side clock work (getBatchTimestamp/setTimestamps) off
so the cell timestamps
+ // shipped from the active cluster are preserved. Never set on primary-side
mutations.
public static final String REPLICATED_MUTATION = "_ReplicatedMutation";
// Per-row mutation attribute that the standby reader synthesizes from the
pre-image cell and
// attaches to each reconstructed mutation when its row had a pre-image
entry. Value is the
@@ -491,19 +498,17 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
private boolean returnOldRow;
private boolean hasConditionalTTL; // table has Conditional TTL
private boolean immutableRows;
- // True when this batch was produced by the standby reader from a
replication-log record (i.e.
- // every mutation carries the {@link
IndexRegionObserver#REPLICATED_MUTATION} marker). Batch-
- // uniform by construction: the standby reader stamps every reconstructed
mutation, so checking
- // the first one is sufficient. The standby uses this to skip the
data-table scan in the PRE
- // phase (pre-image cells are carried as the per-row {@link
IndexRegionObserver#PRE_IMAGE}
- // attribute when the table has a global/uncovered/transform index — same
schema on both
- // clusters, so when we're inside that branch on the standby we always
have pre-images).
- private boolean isReplication;
+ // Where this batch came from, classified once from the first mutation and
carried as the single
+ // source of truth for this batch's HA disposition. The counter, the
cluster-role gate, and
+ // log-group resolution key off it in preBatchMutate; the replay predicate
{@link #isReplay()}
+ // is
+ // 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();
// 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 (isReplication).
+ // first built; only populated on the standby replay path (isReplay).
private List<ReplicatedRowGroup> replicatedRowGroups;
public BatchMutateContext() {
@@ -514,6 +519,20 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
this.clientVersion = clientVersion;
}
+ /**
+ * True when this batch was produced by the standby reader from a
replication-log record (batch
+ * origin {@link BatchOrigin#PHX_REPLAY} — every mutation carries the
+ * {@link IndexRegionObserver#REPLICATED_MUTATION} marker). Batch-uniform
by construction: the
+ * standby reader stamps every reconstructed mutation, so the origin
classified from the first
+ * one holds for the whole batch. The standby uses this to skip the
data-table scan in the PRE
+ * phase (pre-image cells are carried as the per-row {@link
IndexRegionObserver#PRE_IMAGE}
+ * attribute when the table has a global/uncovered/transform index — same
schema on both
+ * clusters, so when we're inside that branch on the standby we always
have pre-images).
+ */
+ boolean isReplay() {
+ return origin == BatchOrigin.PHX_REPLAY;
+ }
+
public void
populateOriginalMutations(MiniBatchOperationInProgress<Mutation> miniBatchOp) {
originalMutations = new ArrayList<Mutation>(miniBatchOp.size());
for (int k = 0; k < miniBatchOp.size(); k++) {
@@ -622,7 +641,14 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
private static final int DEFAULT_ROWLOCK_WAIT_DURATION = 30000;
private static final int DEFAULT_CONCURRENT_MUTATION_WAIT_DURATION_IN_MS =
100;
private byte[] encodedRegionName;
- private boolean shouldReplicate;
+ // Raw SYNCHRONOUS_REPLICATION_ENABLED value: the master switch for all HA
semantics. Unlike
+ // replicateTable it is not narrowed by per-table replication eligibility,
so the cluster-role
+ // gates key off this — applying across all tables when the feature is on,
and fully skipped when
+ // it is off (e.g. the rollout window before the HA group exists).
+ private boolean syncReplicationEnabled;
+ // Sync replication is enabled AND this table is replication-eligible: i.e.
this table's writes
+ // should ride the replication log on this cluster. Set once per coproc load.
+ private boolean replicateTable;
private Abortable abortable;
// Don't replicate the mutation if this attribute is set
@@ -732,13 +758,12 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// when the table descriptor changes, the coproc is reloaded
this.useBloomFilter = bloomFilterType == BloomType.ROW;
byte[] tableName = env.getRegionInfo().getTable().getName();
- this.shouldReplicate =
env.getConfiguration().getBoolean(SYNCHRONOUS_REPLICATION_ENABLED,
- DEFAULT_SYNCHRONOUS_REPLICATION_ENABLED);
- if (this.shouldReplicate) {
- // replication feature is enabled, check if it is enabled for the table
- this.shouldReplicate = SchemaUtil.shouldReplicateTable(tableName);
- }
- if (this.shouldReplicate) {
+ this.syncReplicationEnabled = env.getConfiguration()
+ .getBoolean(SYNCHRONOUS_REPLICATION_ENABLED,
DEFAULT_SYNCHRONOUS_REPLICATION_ENABLED);
+ // replication feature is enabled, check if it is enabled for the table
+ this.replicateTable =
+ this.syncReplicationEnabled &&
SchemaUtil.shouldReplicateTable(tableName);
+ if (this.replicateTable) {
this.ignoreReplicationFilter =
getSynchronousReplicationFilter(tableName);
}
// @CoreCoprocessor guarantees HasRegionServerServices, but guard for
testability
@@ -839,6 +864,24 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
}
+ /**
+ * Where a mutation batch reaching the write path originated. Classified
once (see
+ * {@link #classifyBatch}) and then used by every HA decision instead of
re-deriving from raw
+ * attributes.
+ */
+ enum BatchOrigin {
+ /** Originating client write carrying the {@code _HAGroupName} attribute.
*/
+ CLIENT_HA,
+ /** Originating write with no {@code _HAGroupName} (non-HA client, or one
that omitted it). */
+ CLIENT_NON_HA,
+ /** Standby replay: reconstructed by the log reader, carries {@link
#REPLICATED_MUTATION}. */
+ PHX_REPLAY,
+ /**
+ * Arrived via HBase-native replication: carries source cluster ids, not
{@code _HAGroupName}.
+ */
+ NATIVE_IN
+ }
+
/*
* Also checks for mutationBlockEnabled if
CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED is enabled.
*/
@@ -859,41 +902,54 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// 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
- // has no haGroupName to evaluate against and is skipped. This counts
the code path being
- // short-circuited — it does NOT imply any safety property was breached
(when the block
- // 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 !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 (shouldReplicate && !haGroupName.isPresent()) {
+ BatchOrigin origin = classifyBatch(miniBatchOp, haGroupName);
+
+ // Path-coverage counter: increments only for a genuine client write
that reached the write
+ // path without attaching _HAGroupName (CLIENT_NON_HA), on a
replication-eligible table. The
+ // signal is "a client missed attaching _HAGroupName" — it does NOT
imply any safety property
+ // was breached. Tracked globally rather than per-table so operators can
compare baseline vs.
+ // post-deploy delta to spot new write paths that forgot the attribute.
Standby replay
+ // (PHX_REPLAY) and native-replicated-in (NATIVE_IN) batches are
distinct origins and so are
+ // excluded, as are system-HA-group state writes (which carry a haGroup,
i.e. CLIENT_HA).
+ if (replicateTable && origin == BatchOrigin.CLIENT_NON_HA) {
try {
MetricsHaBypassSourceFactory.getInstance().incrementBypassedMutationBlockCount();
} catch (Throwable t) {
LOG.warn("Failed to increment bypassed mutation block count metric;
continuing", t);
}
+ // Per-table breadcrumb for triage: identify which table has a writer
not attaching
+ // _HAGroupName. Logged once per table per RegionServer lifetime
(add() is true only on
+ // first insertion) so it names the offender without flooding on
sustained writes.
+ if (BYPASS_WARNED_TABLES.add(dataTableName)) {
+ LOG.warn(
+ "Table {} received a mutation batch without an _HAGroupName
attribute while sync "
+ + "replication is enabled (origin CLIENT_NON_HA); a writer to
this table is not "
+ + "attaching the HA group attribute. Investigate and fix that
write path.",
+ dataTableName);
+ }
}
- // We don't want to check for mutation blocking for the system ha group
table
- if (!dataTableName.equals(SYSTEM_HA_GROUP_NAME) &&
haGroupName.isPresent()) {
+ // Cluster-role mutation-block / staleness gate. All HA semantics are
gated on the master
+ // switch (syncReplicationEnabled): with the feature off this never runs
— including the
+ // rollout window where clients already send _HAGroupName but the HA
group does not exist yet.
+ // The system ha group table is exempt so its state writes proceed
during a block window.
+ if (
+ syncReplicationEnabled && origin == BatchOrigin.CLIENT_HA
+ && !dataTableName.equals(SYSTEM_HA_GROUP_NAME)
+ ) {
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.
-
+ // Check if mutation's haGroup is stale
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(name) &&
haGroupStoreManager.isMutationBlocked(name)) {
throw new MutationBlockedIOException(
"Blocking Mutation as Some CRRs " + "are in ACTIVE_TO_STANDBY
state and "
@@ -901,13 +957,11 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
}
- // 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);
+ // 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 =
+ getReplicationLogGroup(c.getEnvironment(), haGroupStoreManager,
origin, haGroupName);
+ preBatchMutateWithExceptions(c, miniBatchOp, logGroup, origin);
return;
} catch (Throwable t) {
rethrowIndexingException(t);
@@ -917,17 +971,22 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
/**
- * 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.
+ * Resolve the replication writer for this batch, or empty when none should
be used. Returns empty
+ * 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
*/
private Optional<ReplicationLogGroup>
getReplicationLogGroup(RegionCoprocessorEnvironment env,
- HAGroupStoreManager haGroupStoreManager, String haGroup) throws
IOException {
- if (!shouldReplicate) {
+ HAGroupStoreManager haGroupStoreManager, BatchOrigin origin,
Optional<String> haGroupName)
+ throws IOException {
+ if (!replicateTable || origin != BatchOrigin.CLIENT_HA) {
return Optional.empty();
}
+ // origin == CLIENT_HA guarantees the HA group name is present.
+ String haGroup = haGroupName.get();
Optional<ReplicationLogGroup> logGroup =
ReplicationLogGroup.get(env.getConfiguration(),
env.getServerName(), haGroup, haGroupStoreManager, abortable);
if (logGroup.isPresent()) {
@@ -945,8 +1004,9 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
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));
+ throw new IOException(
+ String.format("HAGroup %s cannot accept a HA write on %s: this cluster
is not active",
+ haGroup, dataTableName));
}
/**
@@ -967,6 +1027,40 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
return Optional.empty();
}
+ /**
+ * A replay batch is one the standby reader reconstructed; every mutation
carries the
+ * {@link #REPLICATED_MUTATION} marker. The marker is batch-uniform by
construction, so the first
+ * mutation is authoritative.
+ * @return true if the batch is a standby replay batch
+ */
+ boolean isReplayBatch(MiniBatchOperationInProgress<Mutation> miniBatchOp) {
+ return miniBatchOp.size() > 0
+ && miniBatchOp.getOperation(0).getAttribute(REPLICATED_MUTATION) != null;
+ }
+
+ /**
+ * Classify where a batch originated, from the first mutation (all mutations
in a batch are
+ * uniform by construction). The order is significant: the standby-replay
marker and the
+ * native-replication cluster ids are authoritative and mutually exclusive
with an originating
+ * client write, so they are checked before falling back to {@code
_HAGroupName} presence.
+ * @param haGroupName the batch's HA group name as already extracted by
+ * {@link #getHAGroupNameFromBatch}
+ * @return the batch origin
+ */
+ BatchOrigin classifyBatch(MiniBatchOperationInProgress<Mutation> miniBatchOp,
+ Optional<String> haGroupName) {
+ if (isReplayBatch(miniBatchOp)) {
+ return BatchOrigin.PHX_REPLAY;
+ }
+ if (miniBatchOp.size() > 0) {
+ List<UUID> clusterIds = miniBatchOp.getOperation(0).getClusterIds();
+ if (clusterIds != null && !clusterIds.isEmpty()) {
+ return BatchOrigin.NATIVE_IN;
+ }
+ }
+ 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
@@ -1012,7 +1106,7 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
if (this.disabled) {
return;
}
- if (!shouldReplicate) {
+ if (!replicateTable) {
return;
}
Map<String, byte[]> walKeyAttrs = getAttributeValuesFromWALKey(logKey);
@@ -1063,7 +1157,9 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// The stamped value is always empty, so re-stamping empty is equivalent
to copying it through.
Map<String, byte[]> replicationAttrs =
MutationCellGrouper.buildReplicationAttributes(
walKeyAttrs, walKeyAttrs.containsKey(PhoenixIndexCodec.INDEX_UUID));
- logGroup.append(tableName, -1, replicable, replicationAttrs);
+ // Cold restore path: all cells in a WAL edit share one sequence id, so
use the WAL key's
+ // sequence id as the per-record commit id.
+ logGroup.append(tableName, logKey.getSequenceId(), replicable,
replicationAttrs);
logGroup.sync();
}
@@ -1144,7 +1240,7 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
miniBatchOp.setOperationStatus(i, new OperationStatus(SUCCESS,
result));
// since this mutation is ignored by setting it's status to success
in the coproc
// it shouldn't be synchronously replicated
- if (this.shouldReplicate) {
+ if (this.replicateTable) {
m.setAttribute(IGNORE_REPLICATION_ATTRIB,
IGNORE_REPLICATION_ATTRIB_VAL);
}
}
@@ -1729,8 +1825,9 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
/**
- * True when this batch is replicated on an active, replication-configured
cluster: replication is
- * on and an HA log group is present. Gates the active-side pre-image
capture — the pre-image
+ * 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}:
@@ -1739,7 +1836,12 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
* WAL stamp, and the replay path intact so crash-recovery re-ship is
faithfully exercised.
*/
private boolean isReplicatedBatch(BatchMutateContext context) {
- return shouldReplicate && context.logGroup.isPresent();
+ // 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
+ // preBatchMutateWithExceptions.
+ return context.logGroup.isPresent();
}
/**
@@ -2187,7 +2289,7 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// The rest of this method is for handling global index updates
context.indexUpdates =
ArrayListMultimap.<HTableInterfaceReference, Pair<Mutation, byte[]>>
create();
- if (context.isReplication) {
+ if (context.isReplay()) {
// Replicated batches carry per-row pre-images and per-cell timestamps
from the active.
// Group by (row, ts) so each active-side batch's mutations are
processed against their own
// pre-image — recovers the active-batch boundary the reader's
coalescing can erase.
@@ -2381,7 +2483,7 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
* read it from the mutation.
*/
private static long indexMutationTimestamp(BatchMutateContext context,
Mutation m) {
- return context.isReplication ? IndexUtil.getMaxTimestamp(m) :
context.batchTimestamp;
+ return context.isReplay() ? IndexUtil.getMaxTimestamp(m) :
context.batchTimestamp;
}
private void preparePostIndexMutations(BatchMutateContext context,
@@ -2621,23 +2723,27 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
public void
preBatchMutateWithExceptions(ObserverContext<RegionCoprocessorEnvironment> c,
- MiniBatchOperationInProgress<Mutation> miniBatchOp,
Optional<ReplicationLogGroup> logGroup)
- throws Throwable {
+ MiniBatchOperationInProgress<Mutation> miniBatchOp,
Optional<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);
setBatchMutateContext(c, context);
identifyIndexMaintainerTypes(indexMetaData, context);
identifyMutationTypes(miniBatchOp, context);
context.populateOriginalMutations(miniBatchOp);
- // The standby reader stamps every reconstructed mutation with
REPLICATED_MUTATION; checking
- // the first one is sufficient since the marker is batch-uniform by
construction.
- context.isReplication = !context.getOriginalMutations().isEmpty()
- &&
context.getOriginalMutations().get(0).getAttribute(REPLICATED_MUTATION) != null;
// Replicated batches must not carry active-side resolution flags. These
are resolved on the
// active cluster before replication, so cells reach the standby already
in their final form.
Preconditions.checkState(
- !context.isReplication
+ !context.isReplay()
|| (!context.hasAtomic && !context.returnResult &&
!context.hasConditionalTTL),
"replicated batch must not carry active-side resolution flags");
@@ -2662,7 +2768,7 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// per-row PRE_IMAGE the active shipped, never reading prior state from
the data-table region.
// Routing it here keeps every active-only step (row-state scans,
timestamp assignment, atomic
// resolution, replication capture) out of a batch that must not run any
of them.
- if (context.isReplication) {
+ if (context.isReplay()) {
preBatchMutateReplication(c, miniBatchOp, context, indexMetaData);
return;
}
@@ -2893,21 +2999,21 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
/**
- * Stamp the WAL key with everything the WAL-restore re-ship path consumes:
the HA group name it
- * keys on ({@link #getHAGroupFromWALKey}) plus the replication attribute
envelope
- * (schema/table/tenant, and an empty INDEX_UUID for indexed tables) it
reads back in
- * {@link #replicateEditOnWALRestore}. 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.
+ * Stamp the WAL key with everything the WAL-restore re-ship path consumes:
the replication
+ * attribute envelope (schema/table/tenant, the HA group name it keys on via
+ * {@link #getHAGroupFromWALKey}, and an empty INDEX_UUID for indexed
tables) it reads back in
+ * {@link #replicateEditOnWALRestore}. The HA group name rides the envelope
like the other origin
+ * 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.
*/
private void appendReplicationAttributesToWALKey(WALKey key,
IndexRegionObserver.BatchMutateContext context) {
if (context == null || !isReplicatedBatch(context)) {
return;
}
- IndexRegionObserver.appendToWALKey(key,
BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB,
- Bytes.toBytes(context.logGroup.get().getHAGroupName()));
for (Map.Entry<String, byte[]> e :
buildReplicationAttributes(context).entrySet()) {
IndexRegionObserver.appendToWALKey(key, e.getKey(), e.getValue());
}
@@ -3574,8 +3680,10 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
private void replicateMutations(ReplicationLogGroup logGroup,
MiniBatchOperationInProgress<Mutation> miniBatchOp, BatchMutateContext
context)
throws IOException {
- // Replicated batches on the standby never re-replicate.
- if (context.isReplication) {
+ // Replicated batches on the standby never re-replicate. Unreachable in
practice — the only
+ // caller gates on isReplicatedBatch (origin == CLIENT_HA), which excludes
PHX_REPLAY — but kept
+ // as a defensive invariant.
+ if (context.isReplay()) {
return;
}
if (context.getOriginalMutations().isEmpty()) {
@@ -3615,9 +3723,12 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
if (flattened.isEmpty()) {
return;
}
+ // All cells in a miniBatch share one WAL sequence id; capture it from the
first contributing
+ // data cell (data cells precede any appended pre-image cell) as the
per-record commit id.
+ long commitId = flattened.get(0).getSequenceId();
Map<String, byte[]> replicationAttributes =
buildReplicationAttributes(context);
try {
- logGroup.append(dataTableName, -1, flattened, replicationAttributes);
+ logGroup.append(dataTableName, commitId, flattened,
replicationAttributes);
logGroup.sync();
} finally {
metricSource.updateReplicationSyncTime(dataTableName,
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsHaBypassSource.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsHaBypassSource.java
index 276f1f2bb6..0f8dc0590f 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsHaBypassSource.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/metrics/MetricsHaBypassSource.java
@@ -20,19 +20,21 @@ package org.apache.phoenix.hbase.index.metrics;
import org.apache.hadoop.hbase.metrics.BaseSource;
/**
- * Server-side JMX metrics source that counts how many mutation batches pass
through
- * {@code IndexRegionObserver.preBatchMutate} <em>without</em> a resolvable HA
group attribute, so
- * the cluster-role-based mutation-block gate has no haGroupName to evaluate
against and is skipped
- * for that batch.
+ * Server-side JMX metrics source that counts how many client-originated
mutation batches pass
+ * through {@code IndexRegionObserver.preBatchMutate} <em>without</em> an
{@code _HAGroupName}
+ * attribute, so the cluster-role-based mutation-block gate has no haGroupName
to evaluate against
+ * and is skipped for that batch.
* <p>
* This is a <strong>path-coverage detector</strong>, not a safety violation
alarm. The counter is
- * incremented in {@code IndexRegionObserver.preBatchMutate} for
<em>every</em> mutation batch whose
- * {@code _HAGroupName} attribute cannot be resolved — regardless of whether
the cluster-role-based
- * mutation-block feature is enabled, regardless of whether a mutation-block
window is currently
- * active for any HA group, and regardless of whether the table being written
to is HA-replicated.
- * The "bypass" terminology refers strictly to the gate-evaluation code path
being short-circuited;
- * it does <strong>not</strong> imply that any safety property was breached —
when the block feature
- * is disabled or no block window is active, there is no property to breach in
the first place.
+ * incremented in {@code IndexRegionObserver.preBatchMutate} only for a
genuine client write that
+ * reached the write path without attaching {@code _HAGroupName} (batch origin
+ * {@code CLIENT_NON_HA}) on a replication-eligible table (the
sync-replication master switch is on
+ * AND the table is eligible). Standby replay ({@code PHX_REPLAY}) and
native-replicated-in
+ * ({@code NATIVE_IN}) batches also lack {@code _HAGroupName} but are distinct
origins and are
+ * <em>excluded</em>, as are system-HA-group state writes (which carry a
haGroup, i.e.
+ * {@code CLIENT_HA}). The "bypass" terminology refers strictly to the
gate-evaluation code path
+ * being short-circuited; it does <strong>not</strong> imply that any safety
property was breached —
+ * the signal is simply "a client missed attaching {@code _HAGroupName}."
* <p>
* Intended operator use:
* <ul>
@@ -58,20 +60,22 @@ public interface MetricsHaBypassSource extends BaseSource {
String BYPASSED_MUTATION_BLOCK_COUNT = "bypassedMutationBlockCount";
String BYPASSED_MUTATION_BLOCK_COUNT_DESC =
- "Path-coverage counter: number of mutation batches that reached
preBatchMutate without a "
- + "resolvable _HAGroupName attribute (so the cluster-role-based
mutation-block gate had "
- + "nothing to evaluate against and was skipped). Counts the code path
being skipped, not "
- + "a safety breach — when the block feature is disabled or no block
window is active, "
- + "there is no property to breach. Actionable signal is
delta-against-baseline (e.g., a "
- + "spike after a deploy introducing a new mutation path), not absolute
value";
+ "Path-coverage counter: number of client-originated mutation batches that
reached "
+ + "preBatchMutate on a replication-eligible table without an
_HAGroupName attribute "
+ + "(origin CLIENT_NON_HA), so the cluster-role-based mutation-block gate
had nothing to "
+ + "evaluate against and was skipped. Standby-replay and
native-replicated-in batches are "
+ + "excluded. Counts the code path being skipped, not a safety breach —
the signal is that "
+ + "a client missed attaching _HAGroupName. Actionable signal is
delta-against-baseline "
+ + "(e.g., a spike after a deploy introducing a new mutation path), not
absolute value";
/**
- * Increments the gate-skipped-path counter. Called unconditionally from
- * {@code IndexRegionObserver.preBatchMutate} whenever the resolved
- * {@code Optional<ReplicationLogGroup>} is empty — i.e., the mutation batch
carries no
- * {@code _HAGroupName} attribute and the cluster-role-based mutation-block
gate cannot be
- * evaluated for it. Independent of whether the mutation-block feature is
enabled or any block
- * window is active.
+ * Increments the gate-skipped-path counter. Called from
+ * {@code IndexRegionObserver.preBatchMutate} only for a {@code
CLIENT_NON_HA} batch on a
+ * replication-eligible table — i.e., a genuine client write that reached
the write path without
+ * an {@code _HAGroupName} attribute while the sync-replication master
switch is on and the table
+ * is eligible, so the cluster-role-based mutation-block gate cannot be
evaluated for it.
+ * Standby-replay ({@code PHX_REPLAY}), native-replicated-in ({@code
NATIVE_IN}), and
+ * {@code CLIENT_HA} batches are excluded.
*/
void incrementBypassedMutationBlockCount();
}
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 8f73a8c886..dbf75c98b3 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
@@ -65,6 +65,7 @@ import org.apache.hadoop.hbase.Abortable;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.jdbc.ClusterRoleRecord;
import org.apache.phoenix.jdbc.ClusterType;
@@ -205,11 +206,18 @@ public class ReplicationLogGroup {
* stamps an empty UUID onto the envelope (see {@code IndexRegionObserver}).
Keying off the
* client-set attribute here would make the standby's index regeneration
depend on client behavior
* rather than on whether the table actually has indexes.
+ * <p>
+ * {@code HA_GROUP_NAME_ATTRIB} ({@code _HAGroupName}) IS in this list: it
is a batch-uniform,
+ * client-set attribute that identifies the originating HA group, so it
belongs with the other
+ * origin-metadata keys. Carrying it through the single envelope makes both
the log record and the
+ * WAL key (for the WAL-restore path) self-describing, and lets the
WAL-append path stamp it
+ * uniformly rather than as a special case.
*/
public static final List<String> REPLICATION_ATTR_KEYS = Collections
.unmodifiableList(Arrays.asList(MutationState.MutationMetadataType.SCHEMA_NAME.toString(),
MutationState.MutationMetadataType.LOGICAL_TABLE_NAME.toString(),
- MutationState.MutationMetadataType.TENANT_ID.toString()));
+ MutationState.MutationMetadataType.TENANT_ID.toString(),
+ BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB));
public static final String STANDBY_DIR = "in";
public static final String FALLBACK_DIR = "out";
@@ -1356,8 +1364,11 @@ public class ReplicationLogGroup {
private final List<PendingSync> pendingSyncs = new ArrayList<>();
private ReplicationModeImpl currentModeImpl;
private volatile IOException fatalException;
- // Counts events drained per Disruptor batch. Single-threaded access from
onEvent.
- private int batchEventCount;
+ // commitId (the HBase WAL sequence id) of the last data record handed to
a mode via
+ // appendRecord -- covers both the normal event path and the post-failover
replay path, so it
+ // stays current across a failure-driven mode switch. Logged at mode
switches so a transition is
+ // correlatable with the WAL position. -1 until the first data record.
+ private long lastCommitId = -1;
// Counted down by onStart() once the consumer thread has entered its run
loop (past
// BatchEventProcessor.run()'s opening clearAlert()). close() awaits this
before halting so
// halt()'s alert cannot be swallowed by a clearAlert() that races an
immediate close.
@@ -1451,7 +1462,8 @@ public class ReplicationLogGroup {
ReplicationMode newMode = getMode();
if (newMode != currentModeImpl.getMode()) {
// some other thread switched the mode on the replication group
- LOG.info("Mode switched at sequence {} from {} to {}", sequence,
currentModeImpl, newMode);
+ LOG.info("Mode switched at sequence {} (lastCommitId {}) from {} to
{}", sequence,
+ lastCommitId, currentModeImpl, newMode);
// call exit on the last mode here since we can guarantee that the
lastMode
// is not processing any event like append/sync because this is the
only thread
// that is consuming the events from the ring buffer and handing them
off to the
@@ -1501,10 +1513,22 @@ public class ReplicationLogGroup {
replayFailedEvent(failedEvent, sequence);
}
+ /**
+ * Hand a data record to the current mode and advance {@link
#lastCommitId}. The single append
+ * path for every source (normal events and post-failover replays) so the
last-seen commitId
+ * stays current across a failure-driven mode switch.
+ */
+ private void appendRecord(Record record) throws IOException {
+ currentModeImpl.append(record);
+ lastCommitId = record.commitId;
+ }
+
/** Replay all append events which were not yet synced */
private void replayBatch(List<Record> unsyncedAppends) throws IOException {
+ LOG.info("Replaying {} unsynced records into new mode {}",
unsyncedAppends.size(),
+ currentModeImpl.getMode());
for (Record r : unsyncedAppends) {
- currentModeImpl.append(r);
+ appendRecord(r);
}
}
@@ -1515,7 +1539,7 @@ public class ReplicationLogGroup {
// sync event future to the pending future list before the sync event
can potentially
// fail.
if (failedEvent.type == EVENT_TYPE_DATA) {
- currentModeImpl.append(failedEvent.record);
+ appendRecord(failedEvent.record);
}
processPendingSyncs(sequence);
}
@@ -1553,7 +1577,6 @@ public class ReplicationLogGroup {
if (event.type == EVENT_TYPE_SYNC) {
metrics.updateRingBufferTime(currentTimeNs - event.timestampNs);
}
- batchEventCount++;
if (fatalException != null) {
// Append events are ignored; sync futures are failed immediately
// so producer threads unblock without waiting for the sync timeout.
@@ -1565,7 +1588,7 @@ public class ReplicationLogGroup {
try {
switch (event.type) {
case EVENT_TYPE_DATA:
- currentModeImpl.append(event.record);
+ appendRecord(event.record);
break;
case EVENT_TYPE_SYNC:
pendingSyncs.add(new PendingSync(event.syncFuture, currentTimeNs));
@@ -1597,13 +1620,6 @@ public class ReplicationLogGroup {
new IOException("Unexpected error in event handler at sequence " +
sequence, t);
setFatalException(wrapped);
failPendingSyncs(sequence, wrapped);
- } finally {
- // Reset on endOfBatch regardless of success/failure so the counter
never leaks
- // across batches when an exception path bypasses processPendingSyncs.
- if (endOfBatch) {
- metrics.updateBatchSize(batchEventCount);
- batchEventCount = 0;
- }
}
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileFormatWriter.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileFormatWriter.java
index 942e9fa9f3..b52889829b 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileFormatWriter.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/log/LogFileFormatWriter.java
@@ -42,6 +42,9 @@ public class LogFileFormatWriter implements Closeable {
private DataOutputStream blockDataStream;
private long recordCount = 0;
private long blockCount = 0;
+ // commitId (WAL seqId) of the last appended record, so a writer switch is
correlatable across
+ // files. -1 until the first append.
+ private long lastCommitId = -1;
private long blocksStartOffset = -1;
private CRC64 crc = new CRC64(); // Indirect this when we have more than one
type
// Cached buffer for compression for performance
@@ -80,6 +83,7 @@ public class LogFileFormatWriter implements Closeable {
}
encoder.write(record);
recordCount++;
+ lastCommitId = record.getCommitId();
// Check if the current block size exceeds the limit AFTER writing the
record
if (currentBlockBytes.size() >= context.getMaxBlockSize()) {
@@ -220,7 +224,8 @@ public class LogFileFormatWriter implements Closeable {
@Override
public String toString() {
return "LogFileFormatWriter [writerContext=" + context + ", recordCount="
+ recordCount
- + ", blockCount=" + blockCount + ", blocksStartOffset=" +
blocksStartOffset + "]";
+ + ", lastCommitId=" + lastCommitId + ", blockCount=" + blockCount + ",
blocksStartOffset="
+ + blocksStartOffset + "]";
}
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSource.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSource.java
index 13dd680998..56d9f64184 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSource.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSource.java
@@ -51,9 +51,6 @@ public interface MetricsReplicationLogGroupSource extends
BaseSource {
String FS_SYNC_TIME_DESC =
"Histogram of time taken for the underlying filesystem sync (fsync) in
milliseconds";
- String BATCH_SIZE = "phoenixWALBatchSize";
- String BATCH_SIZE_DESC = "Histogram of number of events drained per
Disruptor batch";
-
String PENDING_SYNC_COUNT = "phoenixWALPendingSyncCount";
String PENDING_SYNC_COUNT_DESC = "Histogram of pending sync futures
coalesced into one fsync";
@@ -102,12 +99,6 @@ public interface MetricsReplicationLogGroupSource extends
BaseSource {
*/
void updateFsSyncTime(long timeNs);
- /**
- * Update the number of events drained in a single Disruptor batch.
- * @param size Number of events in the batch
- */
- void updateBatchSize(long size);
-
/**
* Update the number of pending sync futures coalesced into one fsync.
* @param count Number of sync futures
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSourceImpl.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSourceImpl.java
index 845cb19184..c605eaa449 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSourceImpl.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSourceImpl.java
@@ -44,7 +44,6 @@ public class MetricsReplicationLogGroupSourceImpl extends
BaseSourceImpl
private final MutableTimeHistogram rotationTimeMs;
private final MutableHistogram ringBufferTimeNs;
private final MutableTimeHistogram fsSyncTimeMs;
- private final MutableSizeHistogram batchSize;
private final MutableSizeHistogram pendingSyncCount;
private final MutableHistogram pendingSyncWaitTimeNs;
@@ -67,7 +66,6 @@ public class MetricsReplicationLogGroupSourceImpl extends
BaseSourceImpl
rotationTimeMs = getMetricsRegistry().newTimeHistogram(ROTATION_TIME,
ROTATION_TIME_DESC);
ringBufferTimeNs = getMetricsRegistry().newHistogram(RING_BUFFER_TIME,
RING_BUFFER_TIME_DESC);
fsSyncTimeMs = getMetricsRegistry().newTimeHistogram(FS_SYNC_TIME,
FS_SYNC_TIME_DESC);
- batchSize = getMetricsRegistry().newSizeHistogram(BATCH_SIZE,
BATCH_SIZE_DESC);
pendingSyncCount =
getMetricsRegistry().newSizeHistogram(PENDING_SYNC_COUNT,
PENDING_SYNC_COUNT_DESC);
pendingSyncWaitTimeNs =
@@ -119,11 +117,6 @@ public class MetricsReplicationLogGroupSourceImpl extends
BaseSourceImpl
fsSyncTimeMs.add(TimeUnit.NANOSECONDS.toMillis(timeNs));
}
- @Override
- public void updateBatchSize(long size) {
- batchSize.add(size);
- }
-
@Override
public void updatePendingSyncCount(long count) {
pendingSyncCount.add(count);
@@ -156,8 +149,8 @@ public class MetricsReplicationLogGroupSourceImpl extends
BaseSourceImpl
.ringBufferTimeMax(ringSnap.getMax()).ringBufferTimeP50(ringSnap.getMedian())
.ringBufferTimeP99(ringSnap.get99thPercentile()).fsSyncTimeMax(fsSnap.getMax())
.fsSyncTimeP50(fsSnap.getMedian()).fsSyncTimeP99(fsSnap.get99thPercentile())
-
.batchSizeMax(batchSize.getMax()).pendingSyncCountMax(pendingSyncCount.getMax())
-
.pendingSyncWaitTimeMax(pendSnap.getMax()).pendingSyncWaitTimeP50(pendSnap.getMedian())
+
.pendingSyncCountMax(pendingSyncCount.getMax()).pendingSyncWaitTimeMax(pendSnap.getMax())
+ .pendingSyncWaitTimeP50(pendSnap.getMedian())
.pendingSyncWaitTimeP99(pendSnap.get99thPercentile()).build();
}
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogMetricValues.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogMetricValues.java
index c177743d6b..abf93952d3 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogMetricValues.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogMetricValues.java
@@ -39,7 +39,6 @@ public class ReplicationLogMetricValues {
private final long fsSyncTimeMs;
private final long fsSyncTimeP50Ms;
private final long fsSyncTimeP99Ms;
- private final long batchSize;
private final long pendingSyncCount;
private final long pendingSyncWaitTimeNs;
private final long pendingSyncWaitTimeP50Ns;
@@ -60,7 +59,6 @@ public class ReplicationLogMetricValues {
this.fsSyncTimeMs = b.fsSyncTimeMs;
this.fsSyncTimeP50Ms = b.fsSyncTimeP50Ms;
this.fsSyncTimeP99Ms = b.fsSyncTimeP99Ms;
- this.batchSize = b.batchSize;
this.pendingSyncCount = b.pendingSyncCount;
this.pendingSyncWaitTimeNs = b.pendingSyncWaitTimeNs;
this.pendingSyncWaitTimeP50Ns = b.pendingSyncWaitTimeP50Ns;
@@ -127,10 +125,6 @@ public class ReplicationLogMetricValues {
return fsSyncTimeP99Ms;
}
- public long getBatchSizeMax() {
- return batchSize;
- }
-
public long getPendingSyncCountMax() {
return pendingSyncCount;
}
@@ -162,7 +156,6 @@ public class ReplicationLogMetricValues {
private long fsSyncTimeMs;
private long fsSyncTimeP50Ms;
private long fsSyncTimeP99Ms;
- private long batchSize;
private long pendingSyncCount;
private long pendingSyncWaitTimeNs;
private long pendingSyncWaitTimeP50Ns;
@@ -238,11 +231,6 @@ public class ReplicationLogMetricValues {
return this;
}
- public Builder batchSizeMax(long v) {
- this.batchSize = v;
- return this;
- }
-
public Builder pendingSyncCountMax(long v) {
this.pendingSyncCount = v;
return this;
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java
index 423b2202bc..d754f49f8e 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java
@@ -219,7 +219,7 @@ public class ReplicationLogProcessor implements Closeable {
// Map from Table Name to List of Mutations
Map<TableName, List<Mutation>> tableToMutationsMap = new HashMap<>();
- // Track the total number of processed records from input log file
+ // Track the total number of mutations replayed from the input log file
long totalProcessed = 0;
// Track the current batch size as records will be processed in batch size
of
@@ -275,8 +275,9 @@ public class ReplicationLogProcessor implements Closeable {
totalProcessed += currentBatchSize;
}
- LOG.info("Completed processing log file {}. Total mutations processed:
{}",
- logFileReader.getContext().getFilePath(), totalProcessed);
+ LOG.info("Completed processing log file {}. Records read: {}, mutations
replayed: {}",
+ logFileReader.getContext().getFilePath(),
logFileReader.getContext().getRecordsRead(),
+ totalProcessed);
getMetrics().incrementLogFileReplaySuccessCount();
// Replay throughput for files that completed successfully. This runs
only after the whole
// file replayed without error (before the catch), so mutations already
applied by a file that
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutFeatureOnMissingGroupIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutFeatureOnMissingGroupIT.java
new file mode 100644
index 0000000000..78944df564
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutFeatureOnMissingGroupIT.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.end2end;
+
+import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB;
+import static
org.apache.phoenix.query.QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES;
+import static org.apache.phoenix.query.QueryConstants.EMPTY_COLUMN_BYTES;
+import static org.apache.phoenix.query.QueryConstants.EMPTY_COLUMN_VALUE_BYTES;
+import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
+import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED;
+import static
org.apache.phoenix.query.QueryServices.SYNCHRONOUS_REPLICATION_ENABLED;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.query.BaseTest;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+
+/**
+ * Complement of {@link SyncReplicationRolloutStep1IT}: the same
production-representative HA guard
+ * flags, but with the master switch {@code SYNCHRONOUS_REPLICATION_ENABLED =
true} while the HA
+ * group has not been created yet — an out-of-order rollout (the feature
enabled before the group
+ * exists).
+ * <p>
+ * With the feature on, {@code IndexRegionObserver.preBatchMutate} runs the
cluster-role-based
+ * mutation-block / staleness gate for a {@code CLIENT_HA} write and tries to
resolve the (absent)
+ * HA group, so the write fails closed. This pins that fail-closed contract,
ensuring the master
+ * switch must not be flipped on until the HA group is created (rollout Step 2
before Step 3).
+ */
+@Category(NeedsOwnMiniClusterTest.class)
+public class SyncReplicationRolloutFeatureOnMissingGroupIT extends BaseTest {
+
+ private static final Logger LOG =
+
LoggerFactory.getLogger(SyncReplicationRolloutFeatureOnMissingGroupIT.class);
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(3);
+ // Feature on, but the HA group does not exist yet;
production-representative HA guard flags on.
+ props.put(SYNCHRONOUS_REPLICATION_ENABLED, Boolean.TRUE.toString());
+ props.put(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED,
Boolean.TRUE.toString());
+ props.put(HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED,
Boolean.TRUE.toString());
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @Test
+ public void
testWriteWithHAGroupAttributeFailsWhenFeatureEnabledAndGroupMissing()
+ throws Exception {
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ // COLUMN_ENCODED_BYTES=0 so the raw Put below can use literal column
qualifiers.
+ conn.createStatement().execute("CREATE TABLE " + tableName
+ + " (id VARCHAR PRIMARY KEY, v VARCHAR) COLUMN_ENCODED_BYTES=0");
+ }
+
+ // Emulate an upgraded client: a well-formed Phoenix row whose mutation
carries _HAGroupName for
+ // a group that has no server-side HAGroupStoreRecord (the feature was
enabled before Step 2).
+ Put put = new Put(Bytes.toBytes("row1"));
+ put.addColumn(DEFAULT_COLUMN_FAMILY_BYTES, EMPTY_COLUMN_BYTES,
EMPTY_COLUMN_VALUE_BYTES);
+ put.addColumn(DEFAULT_COLUMN_FAMILY_BYTES, Bytes.toBytes("V"),
Bytes.toBytes("hello"));
+ put.setAttribute(HA_GROUP_NAME_ATTRIB, Bytes.toBytes("group_" +
tableName));
+
+ // Bound client retries so a retriable failure surfaces promptly rather
than storming.
+ Configuration conf = new Configuration(getUtility().getConfiguration());
+ conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);
+ conf.setLong(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 60000);
+
+ try (org.apache.hadoop.hbase.client.Connection hcon =
ConnectionFactory.createConnection(conf);
+ Table htable = hcon.getTable(TableName.valueOf(tableName))) {
+ // Feature on -> the HA gate runs, resolves the absent HA group, and
fails the write.
+ htable.put(put);
+ fail("write carrying _HAGroupName must fail when the feature is enabled
but the HA group "
+ + "does not exist yet");
+ } catch (IOException expected) {
+ LOG.info("Write failed as expected for missing HA group", expected);
+ }
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ ResultSet rs =
+ conn.createStatement().executeQuery("SELECT v FROM " + tableName + "
WHERE id = 'row1'");
+ assertFalse("a write that failed closed must not be visible", rs.next());
+ }
+ }
+}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutStep1IT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutStep1IT.java
new file mode 100644
index 0000000000..b6c50b9249
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutStep1IT.java
@@ -0,0 +1,108 @@
+/*
+ * 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.end2end;
+
+import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB;
+import static
org.apache.phoenix.query.QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES;
+import static org.apache.phoenix.query.QueryConstants.EMPTY_COLUMN_BYTES;
+import static org.apache.phoenix.query.QueryConstants.EMPTY_COLUMN_VALUE_BYTES;
+import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
+import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED;
+import static
org.apache.phoenix.query.QueryServices.SYNCHRONOUS_REPLICATION_ENABLED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.util.Map;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.query.BaseTest;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+
+/**
+ * Reproduces the Step-1 window of the synchronous-replication rollout:
servers are upgraded and the
+ * new write-path code is live, but the feature is still dark
+ * ({@code SYNCHRONOUS_REPLICATION_ENABLED
+ * = false}) and the HA group has not been created yet, while clients —
upgraded first — already
+ * attach the {@code _HAGroupName} attribute to every mutation.
+ * <p>
+ * Production runs {@code phoenix.cluster.role.based.mutation.block.enabled =
true} and the
+ * stale-CRR check defaults on, so in this window {@code
IndexRegionObserver.preBatchMutate} reaches
+ * the cluster-role-based mutation-block / staleness gate — which keys off
{@code _HAGroupName}
+ * presence, not the feature flag — and resolves the (absent) HA group,
failing the write.
+ * <p>
+ * The contract this test asserts (design decision D1): {@code
SYNCHRONOUS_REPLICATION_ENABLED} is
+ * the master switch for all HA semantics, so with the feature off the gate
must not run and the
+ * write must succeed regardless of the block/stale sub-flags or whether the
HA group exists.
+ */
+@Category(NeedsOwnMiniClusterTest.class)
+public class SyncReplicationRolloutStep1IT extends BaseTest {
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(3);
+ // Step-1 state: feature dark, but production-representative HA guard
flags on.
+ props.put(SYNCHRONOUS_REPLICATION_ENABLED, Boolean.FALSE.toString());
+ props.put(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED,
Boolean.TRUE.toString());
+ props.put(HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED,
Boolean.TRUE.toString());
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @Test
+ public void
testWriteWithHAGroupAttributeSucceedsWhenFeatureDisabledAndGroupMissing()
+ throws Exception {
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ // COLUMN_ENCODED_BYTES=0 so the raw Put below can use literal column
qualifiers.
+ conn.createStatement().execute("CREATE TABLE " + tableName
+ + " (id VARCHAR PRIMARY KEY, v VARCHAR) COLUMN_ENCODED_BYTES=0");
+ }
+
+ // Emulate an upgraded client: a well-formed Phoenix row whose mutation
carries _HAGroupName for
+ // a group that has no server-side HAGroupStoreRecord (not created until
rollout Step 2).
+ Put put = new Put(Bytes.toBytes("row1"));
+ put.addColumn(DEFAULT_COLUMN_FAMILY_BYTES, EMPTY_COLUMN_BYTES,
EMPTY_COLUMN_VALUE_BYTES);
+ put.addColumn(DEFAULT_COLUMN_FAMILY_BYTES, Bytes.toBytes("V"),
Bytes.toBytes("hello"));
+ put.setAttribute(HA_GROUP_NAME_ATTRIB, Bytes.toBytes("group_" +
tableName));
+
+ try (
+ org.apache.hadoop.hbase.client.Connection hcon =
+ ConnectionFactory.createConnection(getUtility().getConfiguration());
+ Table htable = hcon.getTable(TableName.valueOf(tableName))) {
+ // With the feature off this must not evaluate the HA gate for the
missing group.
+ htable.put(put);
+ }
+
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ ResultSet rs =
+ conn.createStatement().executeQuery("SELECT v FROM " + tableName + "
WHERE id = 'row1'");
+ assertTrue("row written with the _HAGroupName attribute must be
visible", rs.next());
+ assertEquals("hello", rs.getString(1));
+ }
+ }
+}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/BypassedMutationBlockMetricsIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/BypassedMutationBlockMetricsIT.java
index 4db777d287..aad94f4d40 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/BypassedMutationBlockMetricsIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/BypassedMutationBlockMetricsIT.java
@@ -35,9 +35,12 @@ import org.junit.experimental.categories.Category;
/**
* Integration test for the server-side {@code bypassedMutationBlockCount} JMX
counter. A "bypass"
- * is a mutation batch that reaches {@code IndexRegionObserver.preBatchMutate}
without an associated
- * HA group attribute, causing the cluster-role-based mutation-block gate to
be skipped. This IT
- * drives a write that does not carry {@code _HAGroupName} and asserts the
counter increments.
+ * is a client-originated mutation batch (origin {@code CLIENT_NON_HA}) that
reaches
+ * {@code IndexRegionObserver.preBatchMutate} on a replication-eligible table
without an
+ * {@code _HAGroupName} attribute, causing the cluster-role-based
mutation-block gate to be skipped;
+ * standby-replay and native-replicated-in batches are excluded. This IT
relies on {@link HABaseIT}
+ * enabling the sync-replication master switch on both clusters, then drives a
write that does not
+ * carry {@code _HAGroupName} and asserts the counter increments.
*/
@Category(NeedsOwnMiniClusterTest.class)
public class BypassedMutationBlockMetricsIT extends HABaseIT {
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 5ac326789e..e00a046d0a 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
@@ -21,7 +21,6 @@ import static
org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_C
import static
org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR;
import static
org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.getHighAvailibilityGroup;
import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME;
-import static
org.apache.phoenix.query.QueryServices.SYNCHRONOUS_REPLICATION_ENABLED;
import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY;
import static
org.apache.phoenix.replication.reader.ReplicationLogReplayService.PHOENIX_REPLICATION_REPLAY_ENABLED;
import static org.junit.Assert.assertArrayEquals;
@@ -102,8 +101,6 @@ public abstract class ReplicationLogGroupBaseIT extends
HABaseIT {
conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false);
// Disable replay on cluster 2, we will explicitly replay the log files
conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false);
- // Disable writer on cluster 2
- conf2.setBoolean(SYNCHRONOUS_REPLICATION_ENABLED, false);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
}
@@ -120,7 +117,7 @@ public abstract class ReplicationLogGroupBaseIT extends
HABaseIT {
haGroupName = name.getMethodName();
clientProps = HighAvailabilityTestingUtility.getHATestProperties();
clientProps.setProperty(PHOENIX_HA_GROUP_ATTR, haGroupName);
- CLUSTERS.initClusterRole(haGroupName, HighAvailabilityPolicy.FAILOVER);
+ CLUSTERS.initClusterRole(haGroupName, getHAPolicy());
haGroup = getHighAvailibilityGroup(CLUSTERS.getJdbcHAUrl(), clientProps);
LOG.info("Initialized haGroup {} with URL {}", haGroup,
CLUSTERS.getJdbcHAUrl());
logGroup = getReplicationLogGroup();
@@ -133,6 +130,14 @@ public abstract class ReplicationLogGroupBaseIT extends
HABaseIT {
LOG.info("Ending cleanup for test {}", name.getMethodName());
}
+ /**
+ * HA policy for the per-test group. Defaults to FAILOVER; a subclass
overrides this to exercise
+ * the PARALLEL role/policy path (e.g. a double-write reaching a non-active
PARALLEL cluster).
+ */
+ protected HighAvailabilityPolicy getHAPolicy() {
+ return HighAvailabilityPolicy.FAILOVER;
+ }
+
private ReplicationLogGroup getReplicationLogGroup() throws IOException {
HRegionServer rs =
CLUSTERS.getHBaseCluster1().getHBaseCluster().getRegionServer(0);
return ReplicationLogGroup.get(conf1, rs.getServerName(), haGroupName)
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 4b5bd6c1d3..a512b79389 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
@@ -56,6 +56,8 @@ 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.hbase.index.metrics.MetricsHaBypassSourceFactory;
+import org.apache.phoenix.hbase.index.metrics.MetricsHaBypassSourceImpl;
import org.apache.phoenix.jdbc.ClusterRoleRecord;
import org.apache.phoenix.jdbc.FailoverPhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixResultSet;
@@ -305,6 +307,85 @@ public class ReplicationLogGroupIT extends
ReplicationLogGroupBaseIT {
}
}
+ /**
+ * Negative counter case for PHOENIX-8011: standby replay batches (origin
{@code PHX_REPLAY})
+ * carry no {@code _HAGroupName} and previously inflated {@code
bypassedMutationBlockCount}, which
+ * is meant to flag only live client writes that forgot the attribute. This
drives an HA write on
+ * cluster 1, then replays its log onto cluster 2 and asserts the counter
does not move across the
+ * replay -- snapshotting tightly around the replay loop only, since the
cluster-2 schema DDL runs
+ * on a non-HA connection ({@code CLIENT_NON_HA}) and would itself tick the
counter. Cross-cluster
+ * equality proves the replay actually pushed batches through the standby
IRO, so the assertion is
+ * not vacuous. Complements {@code BypassedMutationBlockMetricsIT}, which
covers the positive
+ * ({@code CLIENT_NON_HA} increments) case.
+ */
+ @Test
+ public void testReplayBatchDoesNotIncrementBypassCounter() throws Exception {
+ final String tableName = "T_" + generateUniqueName();
+ final String indexName = "I_" + generateUniqueName();
+ final String createTableDdl = String.format(
+ "create table if not exists %s (id integer not null primary key, val1
varchar, val2 varchar)",
+ tableName);
+ final String createIndexDdl =
+ String.format("create index if not exists %s on %s (val1)", indexName,
tableName);
+
+ try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection)
DriverManager
+ .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) {
+ conn.createStatement().execute(createTableDdl);
+ conn.createStatement().execute(createIndexDdl);
+ conn.commit();
+ PreparedStatement stmt =
+ conn.prepareStatement("upsert into " + tableName + " VALUES(?, ?, ?)");
+ for (int i = 0; i < 10; ++i) {
+ stmt.setInt(1, i);
+ stmt.setString(2, "val1_" + i);
+ stmt.setString(3, "val2_" + i);
+ stmt.executeUpdate();
+ }
+ conn.commit();
+ }
+
+ // Quiesce the group so every log file has a durable trailer before the
reader recovers its
+ // lease (see replayAndVerifyAcrossClusters).
+ Path standByLogDir =
logGroup.getOrCreatePeerShardManager().getRootDirectoryPath();
+ logGroup.close();
+
+ // Create the same schema on cluster 2 before replay. This uses a non-HA
connection, so its
+ // catalog writes are CLIENT_NON_HA and may tick the counter -- hence the
snapshot is taken
+ // after this step, not before.
+ try (Connection conn2 = CLUSTERS.getCluster2Connection(haGroup)) {
+ conn2.createStatement().execute(createTableDdl);
+ conn2.createStatement().execute(createIndexDdl);
+ conn2.commit();
+ }
+
+ FileSystem fs = standByLogDir.getFileSystem(conf2);
+ List<Path> logFiles = findLogFiles(standByLogDir, fs);
+ assertFalse("Should have at least one log file", logFiles.isEmpty());
+
+ MetricsHaBypassSourceImpl source =
+ (MetricsHaBypassSourceImpl) MetricsHaBypassSourceFactory.getInstance();
+ long before = source.getBypassedMutationBlockCountForTesting();
+
+ ReplicationLogProcessor processor = ReplicationLogProcessor.get(conf2,
haGroupName);
+ try {
+ for (Path logFile : logFiles) {
+ processor.processLogFile(fs, logFile);
+ }
+ } finally {
+ processor.close();
+ }
+
+ long after = source.getBypassedMutationBlockCountForTesting();
+ assertEquals(
+ "standby replay (PHX_REPLAY) batches must not increment
bypassedMutationBlockCount", before,
+ after);
+
+ // Prove the replay actually drove batches through the standby IRO, so the
counter assertion is
+ // meaningful and not passing on a no-op replay.
+ assertTablesEqualAcrossClusters(tableName);
+ assertTablesEqualAcrossClusters(indexName);
+ }
+
/**
* Local-index-only variant of {@link #testAppendAndSync}: the table has a
local index but no
* global/uncovered/transform index. Such a table never enters the
global-index branch that
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
new file mode 100644
index 0000000000..ae3b4047bf
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupParallelIT.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.replication;
+
+import static org.apache.phoenix.query.BaseTest.generateUniqueName;
+import static org.junit.Assert.assertEquals;
+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.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
+import org.apache.phoenix.jdbc.ParallelPhoenixConnection;
+import org.junit.BeforeClass;
+import org.junit.Test;
+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.
+ */
+@Category(NeedsOwnMiniClusterTest.class)
+public class ReplicationLogGroupParallelIT extends ReplicationLogGroupBaseIT {
+
+ @BeforeClass
+ public static void doSetup() throws Exception {
+ setupClusters();
+ }
+
+ @Override
+ protected HighAvailabilityPolicy getHAPolicy() {
+ return HighAvailabilityPolicy.PARALLEL;
+ }
+
+ @Test
+ public void
testParallelDoubleWriteCommitsOnNonActiveClusterWithoutSplitBrainThrow()
+ throws Exception {
+ final String tableName = "T_" + generateUniqueName();
+ final int rowCount = 10;
+
+ // Create the table on both clusters (schema: id INTEGER PK, v INTEGER).
replicationScope=false
+ // disables HBase-native replication, so cluster2's rows come only from
the parallel
+ // double-write's standby leg -- isolating the PARALLEL branch under test.
+ 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.
+ 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();
+ }
+
+ // Rows present on BOTH clusters => the double-write committed on the
non-active cluster too,
+ // i.e. its PARALLEL leg took the commit-locally path rather than a
split-brain rejection.
+ assertRowCount(CLUSTERS.getCluster1Connection(haGroup), tableName,
rowCount);
+ assertRowCount(CLUSTERS.getCluster2Connection(haGroup), tableName,
rowCount);
+ }
+
+ private static void assertRowCount(Connection conn, String tableName, int
expected)
+ throws Exception {
+ try (Connection c = conn;
+ ResultSet rs = c.createStatement().executeQuery("select count(*) from "
+ tableName)) {
+ assertTrue(rs.next());
+ assertEquals(expected, rs.getInt(1));
+ }
+ }
+}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexRegionObserverReplayTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexRegionObserverReplayTest.java
index 651eb54dde..008f4d68b0 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexRegionObserverReplayTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexRegionObserverReplayTest.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.hbase.index;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -27,13 +28,18 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
+import org.apache.hadoop.hbase.regionserver.OperationStatus;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.wal.WALEdit;
import org.junit.Test;
/**
@@ -72,6 +78,114 @@ public class IndexRegionObserverReplayTest {
return m;
}
+ /** Wraps the given mutations in a MiniBatchOperationInProgress spanning the
whole array. */
+ private static MiniBatchOperationInProgress<Mutation> batchOf(Mutation...
mutations) {
+ return new MiniBatchOperationInProgress<>(mutations, new
OperationStatus[mutations.length],
+ new WALEdit[mutations.length], 0, mutations.length, mutations.length);
+ }
+
+ // ---- isReplayBatch ----
+
+ @Test
+ public void testIsReplayBatchTrueWhenFirstMutationCarriesMarker() {
+ Put replayed = putRowTs(ROW, TS, Q1);
+ replayed.setAttribute(IndexRegionObserver.REPLICATED_MUTATION,
Bytes.toBytes(true));
+ assertTrue("a batch whose first mutation carries REPLICATED_MUTATION is a
replay batch",
+ iro.isReplayBatch(batchOf(replayed)));
+ }
+
+ @Test
+ public void testIsReplayBatchFalseForLiveWrite() {
+ // A live client write with no REPLICATED_MUTATION marker (and, as here,
no _HAGroupName)
+ // must not be treated as replay -- this is exactly the case the bypass
counter must count.
+ assertFalse("a batch without the marker is not a replay batch",
+ iro.isReplayBatch(batchOf(putRowTs(ROW, TS, Q1))));
+ }
+
+ @Test
+ public void testIsReplayBatchFalseForEmptyBatch() {
+ assertFalse("an empty batch is not a replay batch",
iro.isReplayBatch(batchOf()));
+ }
+
+ // ---- classifyBatch ----
+
+ @Test
+ public void testClassifyBatchClientHaWhenHAGroupPresent() {
+ assertEquals(IndexRegionObserver.BatchOrigin.CLIENT_HA,
+ iro.classifyBatch(batchOf(putRowTs(ROW, TS, Q1)), Optional.of("g")));
+ }
+
+ @Test
+ public void testClassifyBatchClientNonHaWhenNoHAGroup() {
+ assertEquals(IndexRegionObserver.BatchOrigin.CLIENT_NON_HA,
+ iro.classifyBatch(batchOf(putRowTs(ROW, TS, Q1)), Optional.empty()));
+ }
+
+ @Test
+ public void testClassifyBatchNativeInWhenClusterIdsPresent() {
+ // A write replicated in via HBase-native replication carries source
cluster ids and no
+ // _HAGroupName -- it must not be mistaken for a client write that forgot
the attribute.
+ Put p = putRowTs(ROW, TS, Q1);
+ p.setClusterIds(Collections.singletonList(UUID.randomUUID()));
+ assertEquals(IndexRegionObserver.BatchOrigin.NATIVE_IN,
+ iro.classifyBatch(batchOf(p), Optional.empty()));
+ }
+
+ @Test
+ public void testClassifyBatchReplayMarkerWinsOverClusterIds() {
+ // The replay marker is authoritative and checked first.
+ Put p = putRowTs(ROW, TS, Q1);
+ p.setClusterIds(Collections.singletonList(UUID.randomUUID()));
+ p.setAttribute(IndexRegionObserver.REPLICATED_MUTATION,
Bytes.toBytes(true));
+ assertEquals(IndexRegionObserver.BatchOrigin.PHX_REPLAY,
+ iro.classifyBatch(batchOf(p), Optional.empty()));
+ }
+
+ @Test
+ public void testClassifyBatchClusterIdsWinOverHAGroup() {
+ // Origin precedence: native-replicated-in (cluster ids) before
_HAGroupName presence.
+ Put p = putRowTs(ROW, TS, Q1);
+ p.setClusterIds(Collections.singletonList(UUID.randomUUID()));
+ assertEquals(IndexRegionObserver.BatchOrigin.NATIVE_IN,
+ iro.classifyBatch(batchOf(p), Optional.of("g")));
+ }
+
+ @Test
+ public void testClassifyBatchEmptyIsClientNonHa() {
+ assertEquals(IndexRegionObserver.BatchOrigin.CLIENT_NON_HA,
+ iro.classifyBatch(batchOf(), Optional.empty()));
+ }
+
+ @Test
+ public void
testClassifyBatchFirstMutationIsAuthoritativeForMultiMutationBatch() {
+ // A batch is uniform in origin by construction (the standby reader stamps
every reconstructed
+ // mutation), so classifyBatch/isReplayBatch inspect only the first
mutation. Prove that
+ // contract holds for a multi-mutation batch: the replay marker on the
FIRST mutation classifies
+ // the whole batch as PHX_REPLAY even though the trailing mutations carry
no marker.
+ Put replayed = putRowTs(ROW, TS, Q1);
+ replayed.setAttribute(IndexRegionObserver.REPLICATED_MUTATION,
Bytes.toBytes(true));
+ Put unmarked = putRowTs(R2, TS, Q2);
+ assertEquals("first mutation's marker classifies the whole batch",
+ IndexRegionObserver.BatchOrigin.PHX_REPLAY,
+ iro.classifyBatch(batchOf(replayed, unmarked), Optional.empty()));
+ assertTrue("isReplayBatch keys off the first mutation for a multi-mutation
batch",
+ iro.isReplayBatch(batchOf(replayed, unmarked)));
+ }
+
+ @Test
+ public void testReplayBatchIsNotBypassCounterCandidate() {
+ // The bypassedMutationBlockCount gate fires only for CLIENT_NON_HA. A
standby replay batch
+ // carries REPLICATED_MUTATION and classifies as PHX_REPLAY -- never
CLIENT_NON_HA -- so the
+ // counter is not incremented for replayed mutations.
+ Put replayed = putRowTs(ROW, TS, Q1);
+ replayed.setAttribute(IndexRegionObserver.REPLICATED_MUTATION,
Bytes.toBytes(true));
+ IndexRegionObserver.BatchOrigin origin =
iro.classifyBatch(batchOf(replayed), Optional.empty());
+ assertEquals("a replay batch classifies as PHX_REPLAY",
+ IndexRegionObserver.BatchOrigin.PHX_REPLAY, origin);
+ assertNotEquals("a replay batch must not be the counter-triggering origin",
+ IndexRegionObserver.BatchOrigin.CLIENT_NON_HA, origin);
+ }
+
// ---- decodePreImage ----
@Test
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 3dfde73c26..2b337cbd6a 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
@@ -2844,12 +2844,12 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
long componentSumNs = ringBufferTimeNs + pendingSyncWaitTimeNs +
fsSyncTimeNs;
long truncationToleranceNs = TimeUnit.MILLISECONDS.toNanos(1);
LOG.info(
- "Metrics snapshot: maxBatchSize={} maxPendingSyncCount={}"
- + " syncTime[p50={}ms p99={}ms max={}ms]" + " ringBuffer[p50={}ns
p99={}ns max={}ns]"
- + " fsSync[p50={}ms p99={}ms max={}ms]" + " pendingSyncWait[p50={}ns
p99={}ns max={}ns]"
+ "Metrics snapshot: maxPendingSyncCount={}" + " syncTime[p50={}ms
p99={}ms max={}ms]"
+ + " ringBuffer[p50={}ns p99={}ns max={}ns]" + " fsSync[p50={}ms
p99={}ms max={}ms]"
+ + " pendingSyncWait[p50={}ns p99={}ns max={}ns]"
+ " maxComponentSumNs={} syncTimeWithinBound={}",
- metricValues.getBatchSizeMax(), metricValues.getPendingSyncCountMax(),
- metricValues.getSyncTimeP50(), metricValues.getSyncTimeP99(),
metricValues.getSyncTimeMax(),
+ metricValues.getPendingSyncCountMax(), metricValues.getSyncTimeP50(),
+ metricValues.getSyncTimeP99(), metricValues.getSyncTimeMax(),
metricValues.getRingBufferTimeP50(),
metricValues.getRingBufferTimeP99(),
metricValues.getRingBufferTimeMax(), metricValues.getFsSyncTimeP50(),
metricValues.getFsSyncTimeP99(), metricValues.getFsSyncTimeMax(),
@@ -2919,8 +2919,6 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
assertTrue("pendingSyncWaitTime should be >= 0, got " +
values.getPendingSyncWaitTimeMax(),
values.getPendingSyncWaitTimeMax() >= 0);
// Counts.
- assertTrue("batchSize should be > 0, got " + values.getBatchSizeMax(),
- values.getBatchSizeMax() > 0);
assertTrue("pendingSyncCount should be > 0, got " +
values.getPendingSyncCountMax(),
values.getPendingSyncCountMax() > 0);
// Millisecond-resolution timers: the injected fsync delay clears the
truncation floor.
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java
index c3dfb162d3..6e5862ca63 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/log/LogFileCodecTest.java
@@ -41,6 +41,7 @@ import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.hbase.index.IndexRegionObserver;
import org.apache.phoenix.index.PhoenixIndexCodec;
@@ -760,6 +761,8 @@ public class LogFileCodecTest {
Bytes.toBytes("MY_TABLE"));
put.setAttribute(MutationState.MutationMetadataType.TENANT_ID.toString(),
Bytes.toBytes("tenant1"));
+ put.setAttribute(BaseScannerRegionObserverConstants.HA_GROUP_NAME_ATTRIB,
+ Bytes.toBytes("MY_HA_GROUP"));
put.setAttribute(IndexRegionObserver.REPLICATED_MUTATION,
HConstants.EMPTY_BYTE_ARRAY);
LogFile.Record original =
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSourceImplTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSourceImplTest.java
index a53c54280e..2b6699df50 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSourceImplTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogGroupSourceImplTest.java
@@ -95,11 +95,9 @@ public class MetricsReplicationLogGroupSourceImplTest {
}
@Test
- public void testBatchSizeAndPendingSyncCount() {
- source.updateBatchSize(100L);
+ public void testPendingSyncCount() {
source.updatePendingSyncCount(50L);
ReplicationLogMetricValues v = source.getCurrentMetricValues();
- assertEquals(100L, v.getBatchSizeMax());
assertEquals(50L, v.getPendingSyncCountMax());
}
}
