This is an automated email from the ASF dual-hosted git repository.
lokiore pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new 123f94c6c6 PHOENIX-8038 :- Remove
CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED and always block mutations in the
ACTIVE_TO_STANDBY role (#2649)
123f94c6c6 is described below
commit 123f94c6c634028e87675e50737160f5ebf0b4fc
Author: Lokesh Khurana <[email protected]>
AuthorDate: Thu Nov 5 09:51:25 2026 -0800
PHOENIX-8038 :- Remove CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED and always
block mutations in the ACTIVE_TO_STANDBY role (#2649)
* PHOENIX-8038 :- Remove CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED and
always block mutations in the ACTIVE_TO_STANDBY role
Consistent failover is not correct without the ACTIVE_TO_STANDBY mutation
block, so remove
the property and make HAGroupStoreManager#isMutationBlocked always check
the cluster role.
The block stays gated on SYNCHRONOUS_REPLICATION_ENABLED and applies only
to HA client
batches. Drop the property from ITs and remove testMutationBlockDisabled.
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../exception/MutationBlockedIOException.java | 4 +-
.../apache/phoenix/jdbc/HAGroupStoreClient.java | 42 ++++-----
.../apache/phoenix/jdbc/HAGroupStoreManager.java | 101 ++++++++++++---------
.../org/apache/phoenix/query/QueryServices.java | 3 -
.../apache/phoenix/query/QueryServicesOptions.java | 4 -
.../org/apache/phoenix/schema/MetaDataClient.java | 6 +-
.../phoenix/hbase/index/IndexRegionObserver.java | 19 +---
...cReplicationRolloutFeatureOnMissingGroupIT.java | 4 +-
.../end2end/SyncReplicationRolloutStep1IT.java | 14 ++-
.../index/BypassedMutationBlockMetricsIT.java | 5 -
.../IndexRegionObserverMutationBlockingIT.java | 45 ++++-----
.../org/apache/phoenix/jdbc/HAGroupMetricsIT.java | 5 -
.../apache/phoenix/jdbc/HAGroupStoreManagerIT.java | 74 ++++++---------
.../phoenix/jdbc/ParallelPhoenixConnectionIT.java | 5 -
14 files changed, 141 insertions(+), 190 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/exception/MutationBlockedIOException.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/exception/MutationBlockedIOException.java
index 4c2663cefa..9f37225245 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/exception/MutationBlockedIOException.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/exception/MutationBlockedIOException.java
@@ -20,8 +20,8 @@ package org.apache.phoenix.exception;
import org.apache.hadoop.hbase.DoNotRetryIOException;
/**
- * Exception thrown when CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED is set and
the current cluster
- * role is ACTIVE_TO_STANDBY.
+ * Exception thrown when an HA client mutation reaches a cluster whose role
for the mutation's HA
+ * group is ACTIVE_TO_STANDBY.
* <p>
* Extends {@link DoNotRetryIOException} so HBase's RPC retry layers
* ({@code AsyncRequestFutureImpl}, {@code RpcRetryingCallerImpl}) fail-fast
on the first hit
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
index 79f6409eba..e3e722a2b2 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java
@@ -370,34 +370,32 @@ public class HAGroupStoreClient implements Closeable {
}
/**
- * Returns whether the underlying ZooKeeper connection for this client is
currently live.
- * When {@code false}, the Curator connection has been lost/suspended and
any call that
- * reads shared state ({@link #getHAGroupStoreRecord()}, {@link
#subscribeToTargetState},
- * etc.) will throw an {@link IOException}.
+ * Returns whether the underlying ZooKeeper connection for this client is
currently live. When
+ * {@code false}, the Curator connection has been lost/suspended and any
call that reads shared
+ * state ({@link #getHAGroupStoreRecord()}, {@link #subscribeToTargetState},
etc.) will throw an
+ * {@link IOException}.
* <p>
- * Consumers that subscribe an {@link HAGroupStateListener} and cache the
resulting events
- * in memory (i.e. treat the listener view as the source of truth) should
consult this
- * before trusting their cached snapshot — a {@code false} return
means the client
- * has stopped receiving events and the cached view may be arbitrarily stale.
+ * Consumers that subscribe an {@link HAGroupStateListener} and cache the
resulting events in
+ * memory (i.e. treat the listener view as the source of truth) should
consult this before
+ * trusting their cached snapshot — a {@code false} return means the
client has stopped
+ * receiving events and the cached view may be arbitrarily stale.
* <p>
* Flag transitions:
* <ul>
- * <li>Set to {@code true} on successful initial construction, after the
local
- * {@code PathChildrenCache} has been started and warmed.</li>
- * <li>Set to {@code false} by the Curator connection-state listener on
- * {@code CONNECTION_LOST} / {@code CONNECTION_SUSPENDED}.</li>
- * <li>Set back to {@code true} on {@code CONNECTION_RECONNECTED}. This
tracks the
- * Curator reconnect event, <em>not</em> cache resynchronization
— Curator
- * resyncs the {@code PathChildrenCache} asynchronously after
reconnect, so a brief
- * window exists where {@code isHealthy()} returns {@code true} but
the cache has
- * not yet observed post-outage state changes. Consumers requiring
absolute
- * freshness after a reconnect should re-read via {@link
#getHAGroupStoreRecord()}
- * rather than trust a listener-cached snapshot.</li>
+ * <li>Set to {@code true} on successful initial construction, after the
local
+ * {@code PathChildrenCache} has been started and warmed.</li>
+ * <li>Set to {@code false} by the Curator connection-state listener on
{@code CONNECTION_LOST} /
+ * {@code CONNECTION_SUSPENDED}.</li>
+ * <li>Set back to {@code true} on {@code CONNECTION_RECONNECTED}. This
tracks the Curator
+ * reconnect event, <em>not</em> cache resynchronization — Curator
resyncs the
+ * {@code PathChildrenCache} asynchronously after reconnect, so a brief
window exists where
+ * {@code isHealthy()} returns {@code true} but the cache has not yet
observed post-outage state
+ * changes. Consumers requiring absolute freshness after a reconnect should
re-read via
+ * {@link #getHAGroupStoreRecord()} rather than trust a listener-cached
snapshot.</li>
* </ul>
* The field is {@code volatile}; cross-thread reads are safe.
- *
- * @return {@code true} if the ZK connection is live; {@code false} if the
connection has
- * been lost or suspended
+ * @return {@code true} if the ZK connection is live; {@code false} if the
connection has been
+ * lost or suspended
*/
public boolean isHealthy() {
return isHealthy;
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java
index 6a4d5aeb30..13f6f400ad 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreManager.java
@@ -23,9 +23,7 @@ import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.ACTIVE_NOT
import static org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.STANDBY;
import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE;
import static org.apache.phoenix.jdbc.PhoenixHAAdmin.getLocalZkUrl;
-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.QueryServicesOptions.DEFAULT_CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
import static
org.apache.phoenix.query.QueryServicesOptions.DEFAULT_HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED;
import java.io.IOException;
@@ -40,8 +38,10 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.phoenix.exception.InvalidClusterRoleTransitionException;
+import org.apache.phoenix.exception.MutationBlockedIOException;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.exception.StaleClusterRoleRecordException;
import org.apache.phoenix.exception.StaleHAGroupStoreRecordVersionException;
import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
import org.slf4j.Logger;
@@ -61,7 +61,6 @@ public class HAGroupStoreManager {
// Can revert this but will fail for tests with one cluster down
private static Map<String, HAGroupStoreManager> instances = new
ConcurrentHashMap<>();
- private final boolean mutationBlockEnabled;
private final boolean checkStaleCRRForEveryMutation;
private final String zkUrl;
private final Configuration conf;
@@ -182,8 +181,6 @@ public class HAGroupStoreManager {
}
private HAGroupStoreManager(final Configuration conf, final String zkUrl) {
- this.mutationBlockEnabled =
conf.getBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED,
- DEFAULT_CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED);
this.checkStaleCRRForEveryMutation =
conf.getBoolean(HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED,
DEFAULT_HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED);
this.zkUrl = zkUrl;
@@ -202,40 +199,47 @@ public class HAGroupStoreManager {
}
/**
- * Checks whether mutation is blocked or not for a specific HA group.
- * @param haGroupName name of the HA group, null for default HA group which
tracks all HA groups.
- * @return true if mutation is blocked, false otherwise.
- * @throws IOException when HAGroupStoreClient is not healthy.
+ * Checks an HA client mutation for {@code haGroupName} with one read of the
HA group record.
+ * Throws {@link StaleClusterRoleRecordException} if the record is stale for
the client (see
+ * {@link #isHAGroupOnClientStale}), then {@link MutationBlockedIOException}
if the local cluster
+ * role is ACTIVE_TO_STANDBY. A missing record passes both checks.
+ * @param haGroupName name of the HA group
+ * @throws IOException when the mutation is rejected or HAGroupStoreClient
is not healthy
*/
- public boolean isMutationBlocked(String haGroupName) throws IOException {
- if (mutationBlockEnabled) {
- HAGroupStoreClient haGroupStoreClient =
- getHAGroupStoreClientAndSetupFailoverManagement(haGroupName);
- HAGroupStoreRecord recordWithMetadata =
haGroupStoreClient.getHAGroupStoreRecord();
- return recordWithMetadata != null && recordWithMetadata.getClusterRole()
!= null
- && recordWithMetadata.getClusterRole().isMutationBlocked();
+ public void checkMutationAllowed(String haGroupName) throws IOException {
+ HAGroupStoreRecord record =
+
getHAGroupStoreClientAndSetupFailoverManagement(haGroupName).getHAGroupStoreRecord();
+ if (!hasClusterRole(record)) {
+ return;
+ }
+ if (checkStaleCRRForEveryMutation && isStaleOnClient(record)) {
+ throw new StaleClusterRoleRecordException(
+ String.format("HAGroupStoreRecord is stale for haGroup %s on client",
haGroupName));
+ }
+ if (record.getClusterRole().isMutationBlocked()) {
+ throw new MutationBlockedIOException(String.format(
+ "Blocking mutation because the cluster role for HA group %s is
ACTIVE_TO_STANDBY",
+ haGroupName));
}
- return false;
}
/**
- * Returns whether the {@link HAGroupStoreClient} backing {@code
haGroupName} has a live
- * ZooKeeper connection. See {@link HAGroupStoreClient#isHealthy()} for the
full state
- * transitions (including the brief post-reconnect window where the flag
returns
- * {@code true} before Curator has resynchronized the local cache).
+ * Returns whether the {@link HAGroupStoreClient} backing {@code
haGroupName} has a live ZooKeeper
+ * connection. See {@link HAGroupStoreClient#isHealthy()} for the full state
transitions
+ * (including the brief post-reconnect window where the flag returns {@code
true} before Curator
+ * has resynchronized the local cache).
* <p>
* Consumers that subscribe via {@link #subscribeToTargetState} and cache
the resulting
- * {@link HAGroupStateListener} events in memory should consult this before
trusting the
- * cached snapshot: a {@code false} return means the client stopped
receiving events and
- * the cached view may be stale until Curator reconnects. This gives
listener-based
- * consumers a first-class alternative to matching on the {@code
"HAGroupStoreClient is
+ * {@link HAGroupStateListener} events in memory should consult this before
trusting the cached
+ * snapshot: a {@code false} return means the client stopped receiving
events and the cached view
+ * may be stale until Curator reconnects. This gives listener-based
consumers a first-class
+ * alternative to matching on the {@code "HAGroupStoreClient is
* not healthy"} {@link IOException} thrown by the state-reading methods.
- *
* @param haGroupName name of the HA group
- * @return {@code true} if the underlying ZK connection is live; {@code
false} if it has
- * been lost or suspended
- * @throws IOException when the {@link HAGroupStoreClient} for {@code
haGroupName} cannot
- * be resolved or set up
+ * @return {@code true} if the underlying ZK connection is live; {@code
false} if it has been lost
+ * or suspended
+ * @throws IOException when the {@link HAGroupStoreClient} for {@code
haGroupName} cannot be
+ * resolved or set up
*/
public boolean isHealthy(String haGroupName) throws IOException {
HAGroupStoreClient haGroupStoreClient =
@@ -245,25 +249,32 @@ public class HAGroupStoreManager {
public boolean isHAGroupOnClientStale(String haGroupName) throws IOException
{
if (checkStaleCRRForEveryMutation) {
- HAGroupStoreClient haGroupStoreClient =
- getHAGroupStoreClientAndSetupFailoverManagement(haGroupName);
- // If local cluster is not ACTIVE/ACTIVE_TO_STANDBY, it means the
Failover CRR is stale
- // on client as they are trying to write/read from a STANDBY cluster.
- // If we plan to use Standby clusters for scan operations, depends on
whether the
- // connection is an SCN connection or not, If a connection is not an SCN
connection then
- // Standby cluster can't be used as most recent data is only available
through ACTIVE
- // cluster. For SCN connections, future client functionality might allow
attempts to
- // open connections to the standby cluster, which would then accept or
reject the query
- // based on whether the SCN falls within its consistency point, and will
require a change
- // in the logic her, with much bigger change.
- HAGroupStoreRecord hagroupStoreRecord =
haGroupStoreClient.getHAGroupStoreRecord();
- return HighAvailabilityPolicy.valueOf(hagroupStoreRecord.getPolicy())
- == HighAvailabilityPolicy.FAILOVER
- && !(hagroupStoreRecord.getHAGroupState().getClusterRole().isActive());
+ HAGroupStoreRecord record =
+
getHAGroupStoreClientAndSetupFailoverManagement(haGroupName).getHAGroupStoreRecord();
+ return hasClusterRole(record) && isStaleOnClient(record);
}
return false;
}
+ private static boolean hasClusterRole(HAGroupStoreRecord record) {
+ return record != null && record.getClusterRole() != null;
+ }
+
+ /** Caller must check {@link #hasClusterRole} first. */
+ private static boolean isStaleOnClient(HAGroupStoreRecord record) {
+ // If local cluster is not ACTIVE/ACTIVE_TO_STANDBY, it means the Failover
CRR is stale
+ // on client as they are trying to write/read from a STANDBY cluster.
+ // If we plan to use Standby clusters for scan operations, depends on
whether the
+ // connection is an SCN connection or not, If a connection is not an SCN
connection then
+ // Standby cluster can't be used as most recent data is only available
through ACTIVE
+ // cluster. For SCN connections, future client functionality might allow
attempts to
+ // open connections to the standby cluster, which would then accept or
reject the query
+ // based on whether the SCN falls within its consistency point, and will
require a change
+ // in the logic her, with much bigger change.
+ return HighAvailabilityPolicy.valueOf(record.getPolicy()) ==
HighAvailabilityPolicy.FAILOVER
+ && !record.getClusterRole().isActive();
+ }
+
/**
* Force rebuilds the HAGroupStoreClient instance for all HA groups. If any
HAGroupStoreClient
* instance is not created, it will be created.
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
index 93eda74fd4..82b0244a8f 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -452,9 +452,6 @@ public interface QueryServices extends SQLCloseable {
public static final String PHOENIX_VIEW_TTL_ENABLED =
"phoenix.view.ttl.enabled";
public static final String PHOENIX_VIEW_TTL_TENANT_VIEWS_PER_SCAN_LIMIT =
"phoenix.view.ttl.tenant_views_per_scan.limit";
- // Block mutations based on cluster role record
- public static final String CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED =
- "phoenix.cluster.role.based.mutation.block.enabled";
// Check HAGroup is Stale for mutations
public static final String HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED =
"phoenix.ha.group.stale.for.mutation.check.enabled";
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index 8cca59aabc..940e9b4c14 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@ -30,7 +30,6 @@ import static
org.apache.phoenix.query.QueryServices.CDC_TTL_SHARED_CACHE_EXPIRY
import static
org.apache.phoenix.query.QueryServices.CLIENT_INDEX_ASYNC_THRESHOLD;
import static org.apache.phoenix.query.QueryServices.CLIENT_METRICS_TAG;
import static
org.apache.phoenix.query.QueryServices.CLIENT_SPOOL_THRESHOLD_BYTES_ATTRIB;
-import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
import static
org.apache.phoenix.query.QueryServices.COLLECT_REQUEST_LEVEL_METRICS;
import static org.apache.phoenix.query.QueryServices.COMMIT_STATS_ASYNC;
import static
org.apache.phoenix.query.QueryServices.CONNECTION_ACTIVITY_LOGGING_ENABLED;
@@ -502,7 +501,6 @@ public class QueryServicesOptions {
public static final int DEFAULT_PHOENIX_STREAMS_GET_TABLE_REGIONS_TIMEOUT =
300000; // 5 minutes
public static final boolean DEFAULT_SYSTEM_CATALOG_INDEXES_ENABLED = false;
- public static final Boolean
DEFAULT_CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED = false;
public static final Boolean
DEFAULT_HA_GROUP_STALE_FOR_MUTATION_CHECK_ENABLED = true;
public static final Boolean DEFAULT_HA_GROUP_STORE_CLIENT_PREWARM_ENABLED =
false;
public static final Boolean DEFAULT_CQSI_THREAD_POOL_ENABLED = false;
@@ -647,8 +645,6 @@ public class QueryServicesOptions {
DEFAULT_CONNECTION_EXPLAIN_PLAN_LOGGING_ENABLED)
.setIfUnset(CONNECTION_ACTIVITY_LOGGING_INTERVAL,
DEFAULT_CONNECTION_ACTIVITY_LOGGING_INTERVAL_IN_MINS)
- .setIfUnset(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED,
- DEFAULT_CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED)
.setIfUnset(CQSI_THREAD_POOL_ENABLED, DEFAULT_CQSI_THREAD_POOL_ENABLED)
.setIfUnset(CQSI_THREAD_POOL_KEEP_ALIVE_SECONDS,
DEFAULT_CQSI_THREAD_POOL_KEEP_ALIVE_SECONDS)
.setIfUnset(CQSI_THREAD_POOL_CORE_POOL_SIZE,
DEFAULT_CQSI_THREAD_POOL_CORE_POOL_SIZE)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index b9d917cd3b..4b6e656268 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -4926,9 +4926,9 @@ public class MetaDataClient {
/**
* To check if TTL is defined at any of the child below we are
checking it at
* {@link
org.apache.phoenix.coprocessor.MetaDataEndpointImpl#mutateColumn(List,
ColumnMutator, int, PTable, PTable, boolean)}
- * level where in function
- * {@link org.apache.phoenix.coprocessor.MetaDataEndpointImpl#
validateIfMutationAllowedOnParent(PTable, List, PTableType, long, byte[],
byte[], byte[], List, int)}
- * we are already traversing through allDescendantViews.
+ * level where in function {@link
org.apache.phoenix.coprocessor.MetaDataEndpointImpl#
+ * validateIfMutationAllowedOnParent(PTable, List, PTableType, long,
byte[], byte[],
+ * byte[], List, int)} we are already traversing through
allDescendantViews.
*/
}
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 6cbd8e7a54..0109d2f77e 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
@@ -105,8 +105,6 @@ import
org.apache.phoenix.coprocessor.generated.IndexMutationsProtos;
import org.apache.phoenix.coprocessor.generated.PTableProtos;
import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
import org.apache.phoenix.exception.DataExceedsCapacityException;
-import org.apache.phoenix.exception.MutationBlockedIOException;
-import org.apache.phoenix.exception.StaleClusterRoleRecordException;
import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.expression.CaseExpression;
import org.apache.phoenix.expression.Expression;
@@ -912,7 +910,8 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
}
/*
- * Also checks for mutationBlockEnabled if
CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED is enabled.
+ * When sync replication is enabled, also rejects HA client batches whose HA
group is stale on the
+ * client or whose local cluster role is ACTIVE_TO_STANDBY.
*/
@Override
public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
@@ -972,17 +971,9 @@ public class IndexRegionObserver implements
RegionCoprocessor, RegionObserver {
// 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));
- }
-
- if (StringUtils.isNotBlank(name) &&
haGroupStoreManager.isMutationBlocked(name)) {
- throw new MutationBlockedIOException(
- "Blocking Mutation as Some CRRs " + "are in ACTIVE_TO_STANDBY
state and "
- + "CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED is true");
+ // Throws StaleClusterRoleRecordException, then
MutationBlockedIOException, from one read.
+ if (StringUtils.isNotBlank(name)) {
+ haGroupStoreManager.checkMutationAllowed(name);
}
}
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
index 78944df564..aac9608288 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutFeatureOnMissingGroupIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutFeatureOnMissingGroupIT.java
@@ -21,7 +21,6 @@ import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverCons
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;
@@ -68,10 +67,9 @@ public class SyncReplicationRolloutFeatureOnMissingGroupIT
extends BaseTest {
@BeforeClass
public static synchronized void doSetup() throws Exception {
- Map<String, String> props = Maps.newHashMapWithExpectedSize(3);
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(2);
// 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()));
}
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
index b6c50b9249..3a7c49a0bc 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutStep1IT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SyncReplicationRolloutStep1IT.java
@@ -21,7 +21,6 @@ import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverCons
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;
@@ -51,24 +50,23 @@ import
org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
* = 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.
+ * The mutation block is always on and the stale-CRR check defaults on, so
without the master switch
+ * {@code IndexRegionObserver.preBatchMutate} would reach the
cluster-role-based mutation-block /
+ * staleness gate in this window — which keys off {@code _HAGroupName}
presence — and resolve 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.
+ * write must succeed regardless of the stale-check sub-flag 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);
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(2);
// 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()));
}
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 aad94f4d40..e6230fd997 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
@@ -18,7 +18,6 @@
package org.apache.phoenix.end2end.index;
import static org.apache.phoenix.query.BaseTest.generateUniqueName;
-import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
import static org.junit.Assert.assertTrue;
import java.sql.Connection;
@@ -47,10 +46,6 @@ public class BypassedMutationBlockMetricsIT extends HABaseIT
{
@BeforeClass
public static synchronized void doSetup() throws Exception {
- CLUSTERS.getHBaseCluster1().getConfiguration()
- .setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
- CLUSTERS.getHBaseCluster2().getConfiguration()
- .setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexRegionObserverMutationBlockingIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexRegionObserverMutationBlockingIT.java
index f1966e9342..406c13acdc 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexRegionObserverMutationBlockingIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexRegionObserverMutationBlockingIT.java
@@ -20,7 +20,6 @@ package org.apache.phoenix.end2end.index;
import static
org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR;
import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_HA_GROUP_NAME;
import static org.apache.phoenix.query.BaseTest.generateUniqueName;
-import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -35,6 +34,7 @@ import
org.apache.phoenix.exception.MutationBlockedIOException;
import org.apache.phoenix.execute.CommitException;
import org.apache.phoenix.jdbc.FailoverPhoenixConnection;
import org.apache.phoenix.jdbc.HABaseIT;
+import org.apache.phoenix.jdbc.HAGroupStoreManager;
import org.apache.phoenix.jdbc.HAGroupStoreRecord;
import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
import org.apache.phoenix.jdbc.HighAvailabilityTestingUtility;
@@ -49,8 +49,8 @@ import org.junit.rules.TestName;
/**
* Integration test for mutation blocking functionality in
IndexRegionObserver.preBatchMutate()
- * Tests that MutationBlockedIOException is properly thrown when cluster
role-based mutation
- * blocking is enabled and CRRs are in ACTIVE_TO_STANDBY state.
+ * Tests that MutationBlockedIOException is properly thrown when CRRs are in
ACTIVE_TO_STANDBY
+ * state.
*/
@Category(NeedsOwnMiniClusterTest.class)
public class IndexRegionObserverMutationBlockingIT extends HABaseIT {
@@ -67,10 +67,6 @@ public class IndexRegionObserverMutationBlockingIT extends
HABaseIT {
@BeforeClass
public static synchronized void doSetup() throws Exception {
- CLUSTERS.getHBaseCluster1().getConfiguration()
- .setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
- CLUSTERS.getHBaseCluster2().getConfiguration()
- .setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
}
@@ -348,7 +344,8 @@ public class IndexRegionObserverMutationBlockingIT extends
HABaseIT {
HighAvailabilityPolicy.FAILOVER.toString(), this.peerZkUrl,
CLUSTERS.getMasterAddress1(),
CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(),
CLUSTERS.getHdfsUrl2(), 0L);
haAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName,
haGroupStoreRecord, -1);
- awaitZkPropagation(haGroupName,
HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY);
+ awaitServerCachePropagation(haGroupName,
+ HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY);
long startMs = System.currentTimeMillis();
try {
@@ -431,7 +428,8 @@ public class IndexRegionObserverMutationBlockingIT extends
HABaseIT {
HighAvailabilityPolicy.FAILOVER.toString(), this.peerZkUrl,
CLUSTERS.getMasterAddress1(),
CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(),
CLUSTERS.getHdfsUrl2(), 0L);
haAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName,
haGroupStoreRecord, -1);
- awaitZkPropagation(haGroupName,
HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY);
+ awaitServerCachePropagation(haGroupName,
+ HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY);
long startMs = System.currentTimeMillis();
try {
@@ -456,36 +454,29 @@ public class IndexRegionObserverMutationBlockingIT
extends HABaseIT {
}
/**
- * Polls ZooKeeper for the HAGroupStoreRecord to reflect the {@code
expectedState} after a
- * {@code haAdmin.updateHAGroupStoreRecordInZooKeeper} write. Returns as
soon as the cached state
- * matches the expected, or after a timeout. Replaces the unconditional
- * {@code Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS)} pattern
with a bounded polling
- * loop (with a short between-poll park) so the wait is condition-driven
rather than
- * fixed-duration.
- * <p>
- * The poll interval is short ({@link #ZK_PROPAGATION_POLL_INTERVAL_MS} ms)
so the typical wait is
- * well under the propagation budget; the deadline gives generous CI
headroom. Throws
- * {@code AssertionError} via {@code fail()} if the state never converges,
which surfaces the same
- * failure shape a stale read would (the subsequent UPSERT would observe the
old state).
- * @param haGroupName the HA group whose state to await
- * @param expectedState the {@link HAGroupStoreRecord.HAGroupState} the
cached record must reach
- * @throws Exception if reading the cached record throws
+ * Waits until the cluster 1 RegionServer's cached HAGroupStoreRecord (the
one
+ * {@code preBatchMutate} reads) reaches {@code expectedState}. A direct ZK
read can return the
+ * new state before the Curator cache event arrives, so polling ZK alone
lets the next UPSERT race
+ * the cache. The mini-cluster shares this JVM, so {@link
HAGroupStoreManager#getInstance} returns
+ * the RegionServer's instance.
*/
// java:S2925 (Thread.sleep in tests) suppressed: the sleep here is bounded
between condition
// checks inside a deadline-bounded polling loop, not an unconditional wait.
The rule's flake
// concern (fixed-duration sleeps) does not apply.
@SuppressWarnings("java:S2925")
- private void awaitZkPropagation(String haGroupName,
HAGroupStoreRecord.HAGroupState expectedState)
- throws Exception {
+ private void awaitServerCachePropagation(String haGroupName,
+ HAGroupStoreRecord.HAGroupState expectedState) throws Exception {
+ HAGroupStoreManager manager =
+
HAGroupStoreManager.getInstance(CLUSTERS.getHBaseCluster1().getConfiguration());
long deadlineMs = System.currentTimeMillis() +
ZK_PROPAGATION_AWAIT_DEADLINE_MS;
while (System.currentTimeMillis() < deadlineMs) {
- HAGroupStoreRecord cached =
haAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName).getLeft();
+ HAGroupStoreRecord cached =
manager.getHAGroupStoreRecord(haGroupName).orElse(null);
if (cached != null && cached.getHAGroupState() == expectedState) {
return;
}
Thread.sleep(ZK_PROPAGATION_POLL_INTERVAL_MS);
}
- fail("Timed out waiting for ZK to propagate HAGroupStoreRecord state " +
expectedState
+ fail("Timed out waiting for the RegionServer HA group cache to reach state
" + expectedState
+ " for HA group " + haGroupName + " within " +
ZK_PROPAGATION_AWAIT_DEADLINE_MS + "ms");
}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupMetricsIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupMetricsIT.java
index efe9371972..f4ae3c7817 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupMetricsIT.java
@@ -27,7 +27,6 @@ import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_MUTATI
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_POLLER_TICK_COUNT;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_POLLER_TICK_FAILURES;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_STALE_CRR_DETECTED_COUNT;
-import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -71,10 +70,6 @@ public class HAGroupMetricsIT extends HABaseIT {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
- CLUSTERS.getHBaseCluster1().getConfiguration()
- .setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
- CLUSTERS.getHBaseCluster2().getConfiguration()
- .setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
}
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 85d29ed872..b3a08ba72d 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
@@ -22,7 +22,6 @@ import static org.apache.hadoop.test.GenericTestUtils.waitFor;
import static
org.apache.phoenix.jdbc.HAGroupStoreClient.ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE;
import static org.apache.phoenix.jdbc.PhoenixHAAdmin.getLocalZkUrl;
import static org.apache.phoenix.jdbc.PhoenixHAAdmin.toPath;
-import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STORE_CLIENT_PREWARM_ENABLED;
import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.DEFAULT_REPLICATION_ROUND_DURATION_SECONDS;
import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY;
@@ -31,21 +30,21 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.IOException;
-import java.lang.reflect.Field;
import java.util.List;
import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.exception.InvalidClusterRoleTransitionException;
+import org.apache.phoenix.exception.MutationBlockedIOException;
+import org.apache.phoenix.exception.StaleClusterRoleRecordException;
import org.apache.phoenix.replication.ReplicationLogGroup;
import org.apache.phoenix.replication.reader.ReplicationLogReplay;
import org.apache.phoenix.util.HAGroupStoreTestUtil;
@@ -76,9 +75,7 @@ public class HAGroupStoreManagerIT extends HABaseIT {
@BeforeClass
public static synchronized void doSetup() throws Exception {
- conf1.setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
conf1.setLong(ZK_SESSION_TIMEOUT, 30 * 1000);
- conf2.setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
conf2.setLong(ZK_SESSION_TIMEOUT, 30 * 1000);
// Both mini-clusters run in one JVM. Keep background initialization from
claiming shared
// replay and log-group singletons before each test binds them to the
intended cluster.
@@ -121,7 +118,7 @@ public class HAGroupStoreManagerIT extends HABaseIT {
HAGroupStoreManager haGroupStoreManager =
HAGroupStoreManager.getInstance(conf1);
// Initially no mutation should be blocked
- assertFalse(haGroupStoreManager.isMutationBlocked(haGroupName));
+ haGroupStoreManager.checkMutationAllowed(haGroupName);
// Update to ACTIVE_TO_STANDBY role (should block mutations)
HAGroupStoreRecord transitionRecord = new HAGroupStoreRecord("1.0",
haGroupName,
@@ -132,7 +129,27 @@ public class HAGroupStoreManagerIT extends HABaseIT {
Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
// Now mutations should be blocked
- assertTrue(haGroupStoreManager.isMutationBlocked(haGroupName));
+ assertThrows(MutationBlockedIOException.class,
+ () -> haGroupStoreManager.checkMutationAllowed(haGroupName));
+ }
+
+ @Test
+ public void testMutationRejectedAsStaleOnStandby() throws Exception {
+ String haGroupName = testName.getMethodName();
+ HAGroupStoreManager haGroupStoreManager =
HAGroupStoreManager.getInstance(conf1);
+ haGroupStoreManager.checkMutationAllowed(haGroupName);
+
+ // A FAILOVER group whose local role is STANDBY means the client CRR is
stale.
+ HAGroupStoreRecord standbyRecord =
+ new HAGroupStoreRecord("1.0", haGroupName,
HAGroupStoreRecord.HAGroupState.STANDBY, 0L,
+ HighAvailabilityPolicy.FAILOVER.toString(), this.peerZKUrl,
CLUSTERS.getMasterAddress1(),
+ CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(),
CLUSTERS.getHdfsUrl2(), 0L);
+ haAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName, standbyRecord, 0);
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+
+ assertTrue(haGroupStoreManager.isHAGroupOnClientStale(haGroupName));
+ assertThrows(StaleClusterRoleRecordException.class,
+ () -> haGroupStoreManager.checkMutationAllowed(haGroupName));
}
@Test
@@ -158,8 +175,8 @@ public class HAGroupStoreManagerIT extends HABaseIT {
CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2());
// No mutations should be blocked
- assertFalse(haGroupStoreManager.isMutationBlocked(haGroupName1));
- assertFalse(haGroupStoreManager.isMutationBlocked(haGroupName2));
+ haGroupStoreManager.checkMutationAllowed(haGroupName1);
+ haGroupStoreManager.checkMutationAllowed(haGroupName2);
// Move only the second group into a mutation-blocking drain state
// (ACTIVE_IN_SYNC_TO_STANDBY maps to ACTIVE_TO_STANDBY, which blocks
mutations).
@@ -172,8 +189,9 @@ public class HAGroupStoreManagerIT extends HABaseIT {
Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
// Blocking is per-group: only the second group is blocked; the first
stays writable.
- assertFalse(haGroupStoreManager.isMutationBlocked(haGroupName1));
- assertTrue(haGroupStoreManager.isMutationBlocked(haGroupName2));
+ haGroupStoreManager.checkMutationAllowed(haGroupName1);
+ assertThrows(MutationBlockedIOException.class,
+ () -> haGroupStoreManager.checkMutationAllowed(haGroupName2));
}
@Test
@@ -336,38 +354,6 @@ public class HAGroupStoreManagerIT extends HABaseIT {
assertTrue(recordOpt.isPresent());
}
- @Test
- public void testMutationBlockDisabled() throws Exception {
- String haGroupName = testName.getMethodName();
-
- // Create configuration with mutation block disabled
- Configuration conf = new Configuration();
- conf.set(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, "false");
- conf.set(HConstants.ZOOKEEPER_QUORUM, getLocalZkUrl(conf1));
-
- // Set the HAGroupStoreManager instance to null via reflection to force
recreation
- Field field = HAGroupStoreManager.class.getDeclaredField("instances");
- field.setAccessible(true);
- field.set(null, new ConcurrentHashMap<>());
-
- HAGroupStoreManager haGroupStoreManager =
HAGroupStoreManager.getInstance(conf);
- // Create HAGroupStoreRecord with ACTIVE_IN_SYNC_TO_STANDBY role
- HAGroupStoreRecord transitionRecord = new HAGroupStoreRecord("1.0",
haGroupName,
- HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY, 0L,
- HighAvailabilityPolicy.FAILOVER.toString(), this.peerZKUrl,
CLUSTERS.getMasterAddress1(),
- CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(),
CLUSTERS.getHdfsUrl2(), 0L);
-
- haAdmin.updateHAGroupStoreRecordInZooKeeper(haGroupName, transitionRecord,
-1);
- Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
-
- // Mutations should not be blocked even with ACTIVE_TO_STANDBY role
- assertFalse(haGroupStoreManager.isMutationBlocked(haGroupName));
-
- // Set the HAGroupStoreManager instance back to null via reflection to
force recreation for
- // other tests
- field.set(null, new ConcurrentHashMap<>());
- }
-
@Test
public void testSetHAGroupStatusToStoreAndForward() throws Exception {
String haGroupName = testName.getMethodName();
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java
index 84ff5199d4..e4d29c1160 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java
@@ -35,7 +35,6 @@ import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_PARALL
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_PARALLEL_POOL2_TASK_REJECTED_COUNTER;
import static
org.apache.phoenix.query.BaseTest.extractThreadPoolExecutorFromCQSI;
import static org.apache.phoenix.query.QueryServices.AUTO_COMMIT_ATTRIB;
-import static
org.apache.phoenix.query.QueryServices.CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED;
import static
org.apache.phoenix.query.QueryServices.CQSI_THREAD_POOL_ALLOW_CORE_THREAD_TIMEOUT;
import static
org.apache.phoenix.query.QueryServices.CQSI_THREAD_POOL_CORE_POOL_SIZE;
import static org.apache.phoenix.query.QueryServices.CQSI_THREAD_POOL_ENABLED;
@@ -126,10 +125,6 @@ public class ParallelPhoenixConnectionIT extends HABaseIT {
@BeforeClass
public static void setUpBeforeClass() throws Exception {
- CLUSTERS.getHBaseCluster1().getConfiguration()
- .setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
- CLUSTERS.getHBaseCluster2().getConfiguration()
- .setBoolean(CLUSTER_ROLE_BASED_MUTATION_BLOCK_ENABLED, true);
CLUSTERS.start();
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
DriverManager.registerDriver(new PhoenixTestDriver());
