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 91db1394c6 PHOENIX-7566 HAGroupStore control plane: znode-recreation
events, client retention, admin contracts (#2587)
91db1394c6 is described below
commit 91db1394c69803c57a645bb2ca199dc1bf84d0c6
Author: ritegarg <[email protected]>
AuthorDate: Tue Jul 28 09:20:18 2026 -0700
PHOENIX-7566 HAGroupStore control plane: znode-recreation events, client
retention, admin contracts (#2587)
Six fixes found while qualifying the consistent-HA control plane on a
two-cluster
setup.
Deliver events for recreated znodes. Both caches de-duplicated on
ZooKeeper's data
version alone, which resets to 0 when a znode is deleted and recreated, so
the new
incarnation looked stale and was dropped. Track (czxid, version) instead: a
changed
czxid is a new incarnation and is accepted at version 0. A recreated local
record is
redelivered once even when its state is unchanged, so ABORT_TO_STANDBY,
ABORT_TO_ACTIVE_IN_SYNC, and ABORT_TO_ACTIVE_NOT_IN_SYNC still drive their
completion transitions. The peer watcher's forced reconnect redelivery is
unchanged,
and closeConnection() clears the watermark so a new peer ensemble starts
clean.
Keep the cached client through a transient local ZooKeeper outage.
SUSPENDED/LOST
marked the client unhealthy, and the next getInstanceForZkUrl() built a
replacement
and closed the old one. Subscriptions live per instance and failover
management wires
its listeners once per HA group, so every internal and application listener
stayed on
the discarded object while the replacement came up with no subscribers.
Only create a
client when none is mapped; unhealthy instances still fail closed and
recover on
CONNECTION_RECONNECTED.
Validate the admin tool's HA policy against HighAvailabilityPolicy before
dry-run,
reads, or writes, and store the canonical enum name; unsupported values
reached
persistence and only failed later at valueOf. Make create strictly
create-only:
an existing HA-group name now returns RET_VALIDATION_ERROR instead of
reporting
success for a conflicting request.
Gate failover and abort success on both clusters converging to stable local
STANDBY
and peer ACTIVE_IN_SYNC or ACTIVE_NOT_IN_SYNC. Polling accepted the local
state alone,
so abort reported success while the peer was still in
ABORT_TO_ACTIVE_IN_SYNC. Roles
alone are insufficient because the abort states already map to the final
roles.
Correct CLI guidance: malformed ZooKeeper URLs no longer suggest --force,
which cannot
bypass identity-key validation; top-level help/-h/--help exit 0; and a
version-only
update explains that a configuration field is also required.
Validate that the client initialization timeout and system-table sync
interval are
positive before any admin, cache, executor, or cache-map entry is created,
naming the
key and value. Peer-cache retry and legacy CRR reconciliation keep <= 0 as
disable.
HAGroupStoreManager translates the resulting IllegalArgumentException to
IOException
so a misconfiguration cannot abort a RegionServer through a coprocessor
hook.
Co-authored-by: Cursor
Co-authored-by: Cursor <[email protected]>
---
.../apache/phoenix/jdbc/HAGroupStoreCacheUtil.java | 5 ++
.../apache/phoenix/jdbc/HAGroupStoreClient.java | 63 +++++++++-----
.../apache/phoenix/jdbc/HAGroupStoreManager.java | 8 +-
.../apache/phoenix/jdbc/PeerClusterWatcher.java | 15 ++--
.../apache/phoenix/jdbc/PhoenixHAAdminTool.java | 97 ++++++++++++++--------
.../apache/phoenix/jdbc/HAGroupStoreClientIT.java | 42 ++++++++++
.../apache/phoenix/jdbc/HAGroupStoreMetricsIT.java | 7 +-
.../apache/phoenix/jdbc/PhoenixHAAdminToolIT.java | 55 +++++++++---
.../jdbc/HAGroupStoreClientConfigurationTest.java | 55 ++++++++++++
.../phoenix/jdbc/PeerClusterWatcherTest.java | 45 ++++++++++
.../phoenix/jdbc/PhoenixHAAdminToolTest.java | 85 +++++++++++++++++++
11 files changed, 399 insertions(+), 78 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreCacheUtil.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreCacheUtil.java
index 2fad2392d8..8080c1974d 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreCacheUtil.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreCacheUtil.java
@@ -56,6 +56,11 @@ final class HAGroupStoreCacheUtil {
return cache == null ? Pair.of(null, null) :
recordAndStat(cache.getCurrentData(path));
}
+ /** A different creation zxid is a new znode incarnation; versions order
updates within it. */
+ static boolean isNewerRevision(Stat candidate, long lastCzxid, int
lastVersion) {
+ return candidate.getCzxid() != lastCzxid || candidate.getVersion() >
lastVersion;
+ }
+
/**
* Build and start a cache, waiting up to {@code timeoutMs} for the initial
load. The supplied
* {@code listener} receives every cache event; this method releases its
initial-load latch on the
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 82c8460f78..0e118f6a10 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
@@ -141,6 +141,9 @@ public class HAGroupStoreClient implements Closeable {
private final long rotationTimeMs;
// State tracking for transition detection
private volatile HAGroupState lastKnownLocalState;
+ // ZooKeeper revision of the last local state processed.
+ private long lastDeliveredLocalCzxid = -1L;
+ private int lastDeliveredLocalVersion = -1;
// Last peer state delivered to subscribers; used only to populate the
"from" state in PEER
// notifications. Written from the peer watcher's single delivery thread.
private volatile HAGroupState lastKnownPeerState;
@@ -173,8 +176,9 @@ public class HAGroupStoreClient implements Closeable {
}
/**
- * Creates/gets an instance of HAGroupStoreClient. Can return null instance
if unable to
- * initialize.
+ * Creates/gets an instance of HAGroupStoreClient. Can return null if
initial construction fails.
+ * A registered instance remains cached while temporarily unhealthy and
fails operations closed
+ * until Curator reconnects; explicit {@link #close()} removes it.
* @param conf configuration
* @param haGroupName name of the HA group. Only specified HA group is
tracked.
* @param zkUrl zkUrl to use for the client. Prefer providing this
parameter to avoid the
@@ -184,17 +188,16 @@ public class HAGroupStoreClient implements Closeable {
public static HAGroupStoreClient getInstanceForZkUrl(Configuration conf,
String haGroupName,
String zkUrl) {
Preconditions.checkNotNull(haGroupName, "haGroupName cannot be null");
+ validateConfiguration(conf);
String localZkUrl = Objects.toString(zkUrl, getLocalZkUrl(conf));
Preconditions.checkNotNull(localZkUrl, "zkUrl cannot be null");
HAGroupStoreClient result =
instances.getOrDefault(localZkUrl, new
ConcurrentHashMap<>()).getOrDefault(haGroupName, null);
- if (result == null || !result.isHealthy) {
- HAGroupStoreClient replaced = null;
+ if (result == null) {
synchronized (HAGroupStoreClient.class) {
result = instances.getOrDefault(localZkUrl, new ConcurrentHashMap<>())
.getOrDefault(haGroupName, null);
- if (result == null || !result.isHealthy) {
- HAGroupStoreClient stale = result;
+ if (result == null) {
result = new HAGroupStoreClient(conf, null, haGroupName, zkUrl);
if (!result.isHealthy) {
result.close();
@@ -209,20 +212,29 @@ public class HAGroupStoreClient implements Closeable {
v.put(haGroupName, created);
return v;
});
- replaced = stale;
}
}
}
- // Reclaim the replaced unhealthy client's threads and caches. Outside
the monitor because
- // close() can block; safe because deregisterFromInstances() is
value-based and will not
- // remove the replacement we just registered.
- if (replaced != null) {
- replaced.close();
- }
}
return result;
}
+ @VisibleForTesting
+ static void validateConfiguration(Configuration conf) {
+ requirePositive(conf,
PHOENIX_HA_GROUP_STORE_CLIENT_INITIALIZATION_TIMEOUT_MS,
+ DEFAULT_HA_GROUP_STORE_CLIENT_INITIALIZATION_TIMEOUT_MS);
+ requirePositive(conf, HA_GROUP_STORE_SYNC_INTERVAL_SECONDS,
+ DEFAULT_HA_GROUP_STORE_SYNC_INTERVAL_SECONDS);
+ }
+
+ private static void requirePositive(Configuration conf, String key, long
defaultValue) {
+ long value = conf.getLong(key, defaultValue);
+ if (value <= 0) {
+ throw new IllegalArgumentException(
+ key + " must be greater than 0; configured value=" + value);
+ }
+ }
+
/**
* Get the list of HAGroupNames from system table. We can also get the list
of HAGroupNames from
* the system table by providing the zkUrl in where clause, but we need to
match the formatted
@@ -702,8 +714,8 @@ public class HAGroupStoreClient implements Closeable {
return true;
} catch (Exception e) {
metricsSource.incrementSystemTableSyncFailedCount();
- LOGGER.error("Failed to update system table on best effort basis for HA
group {}", haGroupName,
- e);
+ LOGGER.error("Failed to update system table on best effort basis for HA
group {}",
+ haGroupName, e);
return false;
}
}
@@ -1149,11 +1161,10 @@ public class HAGroupStoreClient implements Closeable {
}
/**
- * Remove this instance from the static {@link #instances} map. Idempotent.
Uses value-based
- * remove so that, if a concurrent {@link #getInstanceForZkUrl} has already
swapped in a fresh
- * replacement, the replacement is preserved. The outer-CHM {@code
computeIfPresent} below pairs
- * with the {@code compute} in {@link #getInstanceForZkUrl} to serialize
bucket creation/removal
- * on the same key.
+ * Remove this instance from the static {@link #instances} map. Idempotent.
Value-based removal
+ * preserves a fresh instance created concurrently after this one was
deregistered. The outer-CHM
+ * {@code computeIfPresent} below pairs with the {@code compute} in {@link
#getInstanceForZkUrl}
+ * to serialize bucket creation/removal on the same key.
*/
private void deregisterFromInstances() {
final String key = (this.zkUrl != null) ? this.zkUrl :
getLocalZkUrl(this.conf);
@@ -1485,6 +1496,16 @@ public class HAGroupStoreClient implements Closeable {
private void handleLocalStateChange(HAGroupStoreRecord newRecord, Stat
newStat) {
HAGroupState newState = newRecord.getHAGroupState();
synchronized (localTransitionLock) {
+ if (
+ !HAGroupStoreCacheUtil.isNewerRevision(newStat,
lastDeliveredLocalCzxid,
+ lastDeliveredLocalVersion)
+ ) {
+ return;
+ }
+ boolean recreated =
+ lastDeliveredLocalCzxid >= 0 && newStat.getCzxid() !=
lastDeliveredLocalCzxid;
+ lastDeliveredLocalCzxid = newStat.getCzxid();
+ lastDeliveredLocalVersion = newStat.getVersion();
HAGroupState oldState = lastKnownLocalState;
lastKnownLocalState = newState;
metricsSource.setCurrentLocalState(newState);
@@ -1509,7 +1530,7 @@ public class HAGroupStoreClient implements Closeable {
return;
}
}
- if (oldState == null || !oldState.equals(newState)) {
+ if (recreated || oldState == null || !oldState.equals(newState)) {
// from = the effective state subscribers last saw, derived from the
prior raw state and the
// overlay flag (unchanged on this path) -- not the bare raw oldState.
This is what makes a
// failover while degraded read DEGRADED_STANDBY -> STANDBY_TO_ACTIVE.
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 4240748fc0..e579d109c1 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
@@ -575,8 +575,12 @@ public class HAGroupStoreManager {
* @throws IOException when HAGroupStoreClient is not initialized
*/
private HAGroupStoreClient getHAGroupStoreClient(final String haGroupName)
throws IOException {
- HAGroupStoreClient haGroupStoreClient =
- HAGroupStoreClient.getInstanceForZkUrl(conf, haGroupName, zkUrl);
+ HAGroupStoreClient haGroupStoreClient;
+ try {
+ haGroupStoreClient = HAGroupStoreClient.getInstanceForZkUrl(conf,
haGroupName, zkUrl);
+ } catch (IllegalArgumentException e) {
+ throw new IOException("Invalid HAGroupStoreClient configuration: " +
e.getMessage(), e);
+ }
if (haGroupStoreClient == null) {
throw new IOException("HAGroupStoreClient is not initialized for HA
group: " + haGroupName);
}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PeerClusterWatcher.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PeerClusterWatcher.java
index 8be827d31c..ef21b570d3 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PeerClusterWatcher.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PeerClusterWatcher.java
@@ -102,6 +102,7 @@ final class PeerClusterWatcher implements Closeable {
private String peerZkUrl; // desired peer; blank = none
private PhoenixHAAdmin admin;
private PathChildrenCache cache;
+ private long lastDeliveredCzxid = -1L;
private int lastDeliveredVersion = -1;
private long peerCacheRetryAttempts = 0L;
private volatile boolean watcherClosed = false;
@@ -342,25 +343,28 @@ final class PeerClusterWatcher implements Closeable {
// no-ops if empty.
Pair<HAGroupStoreRecord, Stat> snapshot;
synchronized (stateLock) {
- lastDeliveredVersion = -1; // bypass the de-dup check so the forced
redelivery is not skipped
snapshot = HAGroupStoreCacheUtil.recordAndStatAt(cache,
toPath(haGroupName));
}
setVisible();
deliver(snapshot, true);
}
- private void deliver(Pair<HAGroupStoreRecord, Stat> recordAndStat, boolean
forced) {
+ @VisibleForTesting
+ void deliver(Pair<HAGroupStoreRecord, Stat> recordAndStat, boolean forced) {
HAGroupStoreRecord record = recordAndStat.getLeft();
if (record == null || !Objects.equals(record.getHaGroupName(),
haGroupName)) {
return;
}
Stat stat = recordAndStat.getRight();
synchronized (stateLock) {
- int version = stat != null ? stat.getVersion() : -1;
- if (!forced && version <= lastDeliveredVersion) {
+ if (
+ !forced
+ && !HAGroupStoreCacheUtil.isNewerRevision(stat, lastDeliveredCzxid,
lastDeliveredVersion)
+ ) {
return; // duplicate or stale peer event
}
- lastDeliveredVersion = version;
+ lastDeliveredCzxid = stat.getCzxid();
+ lastDeliveredVersion = stat.getVersion();
}
listener.onPeerStateChanged(record, stat);
}
@@ -407,6 +411,7 @@ final class PeerClusterWatcher implements Closeable {
cache = null;
a = admin;
admin = null;
+ lastDeliveredCzxid = -1L;
lastDeliveredVersion = -1;
}
closeCacheQuietly(c);
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdminTool.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdminTool.java
index 7fbd9d5db5..f295117af8 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdminTool.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixHAAdminTool.java
@@ -42,6 +42,7 @@ import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
+import java.util.Locale;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -176,6 +177,13 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
printUsage();
return RET_ARGUMENT_ERROR;
}
+ if (
+ args.length == 1
+ && ("help".equalsIgnoreCase(args[0]) || "-h".equals(args[0]) ||
"--help".equals(args[0]))
+ ) {
+ printUsage();
+ return RET_SUCCESS;
+ }
String command = args[0].toLowerCase();
String[] commandArgs = Arrays.copyOfRange(args, 1, args.length);
@@ -211,7 +219,7 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
// Required: --ha-group
return executeGetClusterRoleRecord(commandArgs);
case CMD_CREATE:
- // Creates a new HA group entry in SYSTEM.HA_GROUP (idempotent)
+ // Creates a new HA group entry in SYSTEM.HA_GROUP
// Required: --ha-group, --policy, --zk-url-1, --cluster-url-1,
--cluster-role-1,
// --zk-url-2, --cluster-url-2, --cluster-role-2
// Optional: --hdfs-url-1, --hdfs-url-2, --admin-version, --dry-run
@@ -300,7 +308,7 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
}
// Parse configuration fields
- String policy = cmdLine.getOptionValue(POLICY_OPT.getOpt());
+ String policy = parsePolicy(cmdLine.getOptionValue(POLICY_OPT.getOpt()));
String state = cmdLine.getOptionValue(STATE_OPT.getOpt());
String clusterUrl = cmdLine.getOptionValue(CLUSTER_URL_OPT.getOpt());
String peerClusterUrl =
cmdLine.getOptionValue(PEER_CLUSTER_URL_OPT.getOpt());
@@ -326,7 +334,8 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
&& peerZkUrl == null && protocolVersion == null && lastSyncTimeStr
== null
&& hdfsUrl == null && peerHdfsUrl == null
) {
- throw new IllegalArgumentException("Must specify at least one field to
update");
+ throw new IllegalArgumentException("Must specify at least one
configuration field in "
+ + "addition to --admin-version/--auto-increment-version");
}
// Validate URLs against the registry type the read path normalizes them
with (ZK quorum vs
@@ -541,7 +550,10 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
if (invalid.length() == 0) {
return;
}
- if (force) {
+ if (registryType == ClusterRoleRecord.RegistryType.ZK) {
+ throw new IllegalArgumentException("Malformed ZooKeeper URL(s): " +
invalid
+ + ". ZooKeeper URLs are identity keys and must be valid; --force
cannot bypass this validation.");
+ } else if (force) {
LOG.warn("Storing malformed URL(s) due to --force: {}", invalid);
} else {
throw new IllegalArgumentException(
@@ -676,8 +688,7 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
// Poll for state transitions
System.out
.println("\n[Step 4] Monitoring state transitions (timeout: " +
timeoutSeconds + "s)...");
- boolean transitionComplete =
- pollForStateTransition(manager, haGroupName, finalExpectedState,
timeoutSeconds);
+ boolean transitionComplete = pollForStateTransition(manager,
haGroupName, timeoutSeconds);
if (transitionComplete) {
System.out.println("\n✓ Failover completed successfully");
@@ -689,8 +700,8 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
return RET_SUCCESS;
} else {
System.err.println("\n⚠ Failover transition incomplete");
- System.err.println(" The failover was initiated but did not
complete within "
- + timeoutSeconds + " seconds.");
+ System.err.println(" The local failover may have completed, but the
cluster pair did "
+ + "not converge within " + timeoutSeconds + " seconds.");
printFailoverRecoveryGuidance(haGroupName);
return RET_UPDATE_ERROR;
}
@@ -736,8 +747,8 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
* </li>
* <li>Poll for completion via pollForStateTransition():
* <ul>
- * <li>Polls every 2 seconds, waiting for STANDBY state</li>
- * <li>Returns true if reached within timeout, false otherwise</li>
+ * <li>Polls every 2 seconds for local STANDBY and a stable ACTIVE peer</li>
+ * <li>Returns true if the pair converges within timeout, false
otherwise</li>
* </ul>
* </li>
* <li>Return RET_SUCCESS if completed, RET_UPDATE_ERROR if timeout</li>
@@ -809,8 +820,7 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
// Poll for state transitions
System.out
.println("\n[Step 4] Monitoring state transitions (timeout: " +
timeoutSeconds + "s)...");
- boolean transitionComplete =
- pollForStateTransition(manager, haGroupName, HAGroupState.STANDBY,
timeoutSeconds);
+ boolean transitionComplete = pollForStateTransition(manager,
haGroupName, timeoutSeconds);
if (transitionComplete) {
System.out.println("\n✓ Failover abort completed successfully");
@@ -822,9 +832,10 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
return RET_SUCCESS;
} else {
System.err.println("\n⚠ Abort transition incomplete");
- System.err.println(" The abort was initiated but did not complete
within "
- + timeoutSeconds + " seconds.");
- printFailoverRecoveryGuidance(haGroupName);
+ System.err.println(" The local abort may have completed, but the
cluster pair did not "
+ + "converge within " + timeoutSeconds + " seconds.");
+ System.err.println(" Inspect the pair with get-cluster-role-record
-g " + haGroupName
+ + " before retrying or forcing state.");
return RET_UPDATE_ERROR;
}
}
@@ -899,10 +910,8 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
}
/**
- * Creates a new HA group entry in SYSTEM.HA_GROUP. Idempotent: if the group
already exists,
- * prints a skip message and returns success without modifying the existing
row. The ZK znode is
- * initialized automatically on first access by HAGroupStoreClient. Run the
same command on both
- * clusters.
+ * Creates a new HA group entry in SYSTEM.HA_GROUP. Existing groups are
rejected and must be
+ * inspected with get and changed through update where supported.
*/
private int executeCreate(String[] args) throws Exception {
try {
@@ -914,7 +923,7 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
}
String haGroupName = getRequiredOption(cmdLine, HA_GROUP_OPT, "HA group
name");
- String policy = getRequiredOption(cmdLine, POLICY_OPT, "policy");
+ String policy = parsePolicy(getRequiredOption(cmdLine, POLICY_OPT,
"policy"));
String zkUrl1 = getRequiredOption(cmdLine, ZK_URL_1_OPT, "ZK URL for
cluster 1");
String clusterUrl1 =
getRequiredOption(cmdLine, CLUSTER_URL_1_OPT, "cluster URL for cluster
1");
@@ -950,9 +959,9 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
String localZkUrl = getLocalZkUrl(getConf());
if (haGroupExistsInSystemTable(haGroupName, localZkUrl)) {
- System.out.println("HA group '" + haGroupName
- + "' already exists in SYSTEM.HA_GROUP. Skipping creation. Use
update command to modify it.");
- return RET_SUCCESS;
+ System.err.println("HA group '" + haGroupName
+ + "' already exists in SYSTEM.HA_GROUP. Use get to inspect it and
update for supported changes.");
+ return RET_VALIDATION_ERROR;
}
System.out.println("\n=== HA Group to Create ===\n");
@@ -1132,21 +1141,21 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
}
/**
- * Poll for state transition completion with timeout
+ * Poll for stable local STANDBY and peer ACTIVE convergence with timeout.
* @param manager HAGroupStoreManager instance
* @param haGroupName HA group name
- * @param finalState Expected final state
* @param timeoutSeconds Timeout in seconds
* @return true if transition completed, false if timed out
*/
private boolean pollForStateTransition(HAGroupStoreManager manager, String
haGroupName,
- HAGroupState finalState, int timeoutSeconds) {
+ int timeoutSeconds) {
long startTime = System.currentTimeMillis();
long timeoutMillis = timeoutSeconds * 1000L;
long pollIntervalMillis = 2000; // Poll every 2 seconds
HAGroupState lastSeenState = null;
+ HAGroupState lastSeenPeerState = null;
int dots = 0;
try {
@@ -1162,13 +1171,14 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
HAGroupState currentState = recordOpt.get().getHAGroupState();
// Print state change
- if (lastSeenState != currentState) {
+ if (lastSeenState != currentState || lastSeenPeerState != peerState)
{
if (lastSeenState != null) {
System.out.println(); // New line after dots
}
System.out.println(" Current State: " + currentState);
System.out.println(" Peer State: " + peerState);
lastSeenState = currentState;
+ lastSeenPeerState = peerState;
dots = 0;
} else {
// Print progress dots
@@ -1180,8 +1190,7 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
}
}
- // Check if we reached the final state
- if (currentState == finalState) {
+ if (isStableFailoverPair(currentState, peerState)) {
if (dots > 0) {
System.out.println(); // New line after dots
}
@@ -1200,7 +1209,8 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
}
System.err.println(" ✗ Timeout after " + timeoutSeconds + " seconds");
if (lastSeenState != null) {
- System.err.println(" Last seen state: " + lastSeenState);
+ System.err.println(" Last seen local state: " + lastSeenState);
+ System.err.println(" Last seen peer state: " + lastSeenPeerState);
}
return false;
@@ -1213,6 +1223,12 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
}
}
+ @VisibleForTesting
+ static boolean isStableFailoverPair(HAGroupState localState, HAGroupState
peerState) {
+ return localState == HAGroupState.STANDBY
+ && (peerState == HAGroupState.ACTIVE_IN_SYNC || peerState ==
HAGroupState.ACTIVE_NOT_IN_SYNC);
+ }
+
/**
* Print operator guidance shown when a failover / abort does not converge
within the timeout.
* Post-partition states (e.g. active ACTIVE_IN_SYNC_TO_STANDBY with peer
STANDBY) are ambiguous
@@ -1668,6 +1684,19 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
}
}
+ private static String parsePolicy(String policy) {
+ if (policy == null) {
+ return null;
+ }
+ try {
+ return
HighAvailabilityPolicy.valueOf(policy.toUpperCase(Locale.ROOT)).name();
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Invalid HA policy: " + policy +
"\nSupported policies: "
+ + Arrays.stream(HighAvailabilityPolicy.values()).map(Enum::name)
+ .collect(Collectors.joining(", ")));
+ }
+ }
+
/**
* Parse long from string
*/
@@ -1694,8 +1723,7 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
System.out.println("Usage: phoenix-consistentha-admin-tool <command>
[options]");
System.out.println();
System.out.println("Commands:");
- System.out
- .println(" create Create a new HA group in
SYSTEM.HA_GROUP (idempotent)");
+ System.out.println(" create Create a new HA group in
SYSTEM.HA_GROUP");
System.out.println(" update Update HA group
configuration");
System.out.println(" get Show HA group
configuration");
System.out.println(" list List all HA groups");
@@ -1730,9 +1758,8 @@ public class PhoenixHAAdminTool extends Configured
implements Tool {
System.out.println();
System.out.println("Description:");
System.out.println(" Creates a new HA group entry in SYSTEM.HA_GROUP. The
ZK znode is");
- System.out.println(" initialized automatically on first access. This
command is idempotent:");
- System.out.println(" running it again on an existing HA group prints a
skip message and");
- System.out.println(" returns success. Run the same command on both
clusters.");
+ System.out.println(" initialized automatically on first access. Existing
names are rejected;");
+ System.out.println(" use get to inspect them and update for supported
changes.");
System.out.println();
System.out.println("REQUIRED:");
System.out.println(" -g, --ha-group <name> HA group name");
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreClientIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreClientIT.java
index 7b8f90dba3..4ea96ed0ba 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreClientIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreClientIT.java
@@ -32,6 +32,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -557,6 +558,10 @@ public class HAGroupStoreClientIT extends HABaseIT {
createOrUpdateHAGroupStoreRecordOnZookeeper(haAdmin, haGroupName, record);
HAGroupStoreClient haGroupStoreClient = HAGroupStoreClient
.getInstanceForZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration(),
haGroupName, zkUrl);
+ AtomicInteger deliveries = new AtomicInteger();
+
haGroupStoreClient.subscribeToTargetState(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC,
+ ClusterType.LOCAL,
+ (group, from, to, mtime, clusterType, lastSync) ->
deliveries.incrementAndGet());
Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
HAGroupStoreRecord currentRecord =
haGroupStoreClient.getHAGroupStoreRecord();
@@ -573,10 +578,14 @@ public class HAGroupStoreClientIT extends HABaseIT {
Field isHealthyField =
HAGroupStoreClient.class.getDeclaredField("isHealthy");
isHealthyField.setAccessible(true);
assertFalse((boolean) isHealthyField.get(haGroupStoreClient));
+ HAGroupStoreClient duringOutage = HAGroupStoreClient
+ .getInstanceForZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration(),
haGroupName, zkUrl);
// Start ZK on the same port to simulate CONNECTION_RECONNECTED event
CLUSTERS.getHBaseCluster1().startMiniZKCluster(1, Integer.parseInt(
CLUSTERS.getHBaseCluster1().getConfiguration().get("hbase.zookeeper.property.clientPort")));
+ assertSame("factory must retain the subscribed client while local ZK is
unavailable",
+ haGroupStoreClient, duringOutage);
Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
// Check that HAGroupStoreClient instance is back to healthy and provides
correct response
@@ -585,6 +594,39 @@ public class HAGroupStoreClientIT extends HABaseIT {
&& currentRecord.getHAGroupState() ==
HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC;
// Check that the HAGroupStoreClient instance is healthy via reflection
assertTrue((boolean) isHealthyField.get(haGroupStoreClient));
+ record =
record.withHAGroupState(HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC);
+ createOrUpdateHAGroupStoreRecordOnZookeeper(haAdmin, haGroupName, record);
+ long deadline = System.currentTimeMillis() +
ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS * 3;
+ while (deliveries.get() < 1 && System.currentTimeMillis() < deadline) {
+ Thread.sleep(100);
+ }
+ assertEquals("listener registered before the outage must survive
reconnect", 1,
+ deliveries.get());
+ }
+
+ @Test
+ public void testRecreatedLocalZnodeRedeliversSameAbortState() throws
Exception {
+ String haGroupName = testName.getMethodName();
+ HAGroupStoreRecord record = new HAGroupStoreRecord("v1.0", haGroupName,
+ HAGroupStoreRecord.HAGroupState.ABORT_TO_ACTIVE_IN_SYNC, 0L,
+ HighAvailabilityPolicy.FAILOVER.toString(), this.peerZKUrl,
this.masterUrl,
+ this.peerMasterUrl, CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L);
+ createOrUpdateHAGroupStoreRecordOnZookeeper(haAdmin, haGroupName, record);
+ HAGroupStoreClient client = HAGroupStoreClient
+ .getInstanceForZkUrl(CLUSTERS.getHBaseCluster1().getConfiguration(),
haGroupName, zkUrl);
+ AtomicInteger deliveries = new AtomicInteger();
+
client.subscribeToTargetState(HAGroupStoreRecord.HAGroupState.ABORT_TO_ACTIVE_IN_SYNC,
+ ClusterType.LOCAL,
+ (group, from, to, mtime, clusterType, lastSync) ->
deliveries.incrementAndGet());
+
+ createOrUpdateHAGroupStoreRecordOnZookeeper(haAdmin, haGroupName, record);
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+ assertEquals("same znode incarnation should not redeliver the same state",
0, deliveries.get());
+
+ haAdmin.deleteHAGroupStoreRecordInZooKeeper(haGroupName);
+ haAdmin.createHAGroupStoreRecordInZooKeeper(record);
+ Thread.sleep(ZK_CURATOR_EVENT_PROPAGATION_TIMEOUT_MS);
+ assertEquals("recreated znode should redeliver the same abort state once",
1, deliveries.get());
}
/**
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java
index 80e1f7781a..3c37d83b09 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/HAGroupStoreMetricsIT.java
@@ -23,7 +23,6 @@ import static org.apache.phoenix.jdbc.PhoenixHAAdmin.toPath;
import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STORE_SYNC_INTERVAL_SECONDS;
import static
org.apache.phoenix.replication.reader.ReplicationLogReplayService.PHOENIX_REPLICATION_REPLAY_ENABLED;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
@@ -148,11 +147,11 @@ public class HAGroupStoreMetricsIT extends HABaseIT {
@Test
public void testLateInitializationFailureResetsGauges() throws Exception {
writeLocal(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
- Configuration invalidConf =
- new Configuration(CLUSTERS.getHBaseCluster1().getConfiguration());
+ Configuration invalidConf = new
Configuration(CLUSTERS.getHBaseCluster1().getConfiguration());
invalidConf.setLong(HA_GROUP_STORE_SYNC_INTERVAL_SECONDS, 0L);
- assertNull(HAGroupStoreClient.getInstanceForZkUrl(invalidConf, group(),
zkUrl));
+ HAGroupStoreClient failed = new HAGroupStoreClient(invalidConf, null,
group(), zkUrl);
+ assertThrows(IOException.class, failed::getHAGroupStoreRecord);
HAGroupStoreMetricValues values = metrics();
assertEquals(1L, values.getLocalCacheHealthStatus());
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/PhoenixHAAdminToolIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/PhoenixHAAdminToolIT.java
index c6d110f55e..9fb69457ea 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/PhoenixHAAdminToolIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/PhoenixHAAdminToolIT.java
@@ -34,6 +34,7 @@ import static
org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ZK_URL_2;
import static org.apache.phoenix.jdbc.PhoenixHAAdmin.toPath;
import static org.apache.phoenix.jdbc.PhoenixHAAdminTool.RET_ARGUMENT_ERROR;
import static org.apache.phoenix.jdbc.PhoenixHAAdminTool.RET_SUCCESS;
+import static org.apache.phoenix.jdbc.PhoenixHAAdminTool.RET_VALIDATION_ERROR;
import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_ZK;
import static org.junit.Assert.assertEquals;
@@ -984,10 +985,7 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
}
}
- /**
- * Test that the create command is idempotent: if the HA group already
exists in SYSTEM.HA_GROUP,
- * it returns success without modifying the existing row.
- */
+ /** Existing HA groups are rejected without modifying the stored row. */
@Test(timeout = 180000)
public void testCreateCommandAlreadyExists() throws Exception {
// haGroupName is pre-inserted by @Before
@@ -995,6 +993,7 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
assertNotNull("Record should exist before create command", originalRecord);
System.setOut(new PrintStream(STDOUT_CAPTURE));
+ System.setErr(new PrintStream(STDOUT_CAPTURE));
// Run create with different values - these should NOT overwrite the
existing row
int ret = ToolRunner.run(adminTool,
@@ -1003,14 +1002,13 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
CLUSTERS.getMasterAddress2(), "-cr2", "ACTIVE", "-hdfs1",
CLUSTERS.getHdfsUrl1(), "-hdfs2",
CLUSTERS.getHdfsUrl2() });
- assertEquals("create command should return success even when group already
exists", RET_SUCCESS,
- ret);
+ assertEquals("create command should reject an existing group",
RET_VALIDATION_ERROR, ret);
String output = STDOUT_CAPTURE.toString();
LOG.info("Got stdout from create (existing group):
\n++++++++\n{}++++++++\n", output);
- assertTrue("Output should indicate group already exists and was skipped",
- output.contains("already exists") && output.contains("Skipping"));
+ assertTrue("Output should direct the operator to inspect and update the
existing group",
+ output.contains("already exists") && output.contains("get") &&
output.contains("update"));
// Verify the system table row was NOT modified
SystemTableRecord afterRecord = querySystemTable(haGroupName,
CLUSTERS.getZkUrl1());
@@ -1025,6 +1023,37 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
afterRecord.clusterUrl2);
}
+ @Test(timeout = 180000)
+ public void testPolicyIsValidatedBeforeDryRunOrWrite() throws Exception {
+ SystemTableRecord originalRecord = querySystemTable(haGroupName,
CLUSTERS.getZkUrl1());
+ new HAGroupStoreManager(CLUSTERS.getHBaseCluster1().getConfiguration())
+ .getHAGroupStoreRecord(haGroupName);
+ assertEquals(RET_ARGUMENT_ERROR, ToolRunner.run(adminTool,
+ new String[] { "update", "-g", haGroupName, "-p", "BOGUS", "-av", "-d"
}));
+
+ String newGroup = haGroupName + "_new";
+ assertEquals(RET_ARGUMENT_ERROR,
+ ToolRunner.run(adminTool,
+ new String[] { "create", "-g", newGroup, "-p", "BOGUS", "-zk1",
CLUSTERS.getZkUrl1(), "-c1",
+ CLUSTERS.getMasterAddress1(), "-cr1", "ACTIVE", "-zk2",
CLUSTERS.getZkUrl2(), "-c2",
+ CLUSTERS.getMasterAddress2(), "-cr2", "STANDBY", "-hdfs1",
CLUSTERS.getHdfsUrl1(),
+ "-hdfs2", CLUSTERS.getHdfsUrl2(), "-d" }));
+
+ assertEquals(originalRecord.version,
+ querySystemTable(haGroupName, CLUSTERS.getZkUrl1()).version);
+ assertTrue(querySystemTable(newGroup, CLUSTERS.getZkUrl1()) == null);
+
+ STDOUT_CAPTURE.reset();
+ System.setOut(new PrintStream(STDOUT_CAPTURE));
+ assertEquals(RET_SUCCESS,
+ ToolRunner.run(adminTool,
+ new String[] { "create", "-g", newGroup, "-p", "failover", "-zk1",
CLUSTERS.getZkUrl1(),
+ "-c1", CLUSTERS.getMasterAddress1(), "-cr1", "ACTIVE", "-zk2",
CLUSTERS.getZkUrl2(),
+ "-c2", CLUSTERS.getMasterAddress2(), "-cr2", "STANDBY", "-hdfs1",
CLUSTERS.getHdfsUrl1(),
+ "-hdfs2", CLUSTERS.getHdfsUrl2(), "-d" }));
+ assertTrue(STDOUT_CAPTURE.toString().contains("FAILOVER"));
+ }
+
/**
* Test that the create command dry-run mode shows what would be created
without writing to
* SYSTEM.HA_GROUP.
@@ -1129,16 +1158,18 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
new HAGroupStoreManager(CLUSTERS.getHBaseCluster1().getConfiguration())
.getHAGroupStoreRecord(haGroupName);
- System.setOut(new PrintStream(STDOUT_CAPTURE));
+ System.setErr(new PrintStream(STDOUT_CAPTURE));
int ret = ToolRunner.run(adminTool,
new String[] { "update", "-g", haGroupName, "-pz", MALFORMED_URL, "-av"
});
assertEquals("Malformed peer ZK URL should be rejected",
RET_ARGUMENT_ERROR, ret);
+ assertTrue(STDOUT_CAPTURE.toString().contains("--force cannot bypass"));
STDOUT_CAPTURE.reset();
int forcedRet = ToolRunner.run(adminTool,
new String[] { "update", "-g", haGroupName, "-pz", MALFORMED_URL, "-av",
"-F" });
assertEquals("Malformed peer ZK URL should be rejected even with --force",
RET_ARGUMENT_ERROR,
forcedRet);
+ assertTrue(STDOUT_CAPTURE.toString().contains("--force cannot bypass"));
}
/**
@@ -1147,7 +1178,7 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
@Test(timeout = 180000)
public void testCreateRejectsMalformedUrl() throws Exception {
String createGroup = "testCreateBadUrl_" + System.currentTimeMillis();
- System.setOut(new PrintStream(STDOUT_CAPTURE));
+ System.setErr(new PrintStream(STDOUT_CAPTURE));
int ret = ToolRunner.run(adminTool,
new String[] { "create", "-g", createGroup, "-p", "FAILOVER", "-zk1",
CLUSTERS.getZkUrl1(),
@@ -1155,6 +1186,7 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
CLUSTERS.getMasterAddress2(), "-cr2", "STANDBY", "-hdfs1",
CLUSTERS.getHdfsUrl1(), "-hdfs2",
CLUSTERS.getHdfsUrl2() });
assertEquals("Malformed cluster-url-1 should be rejected on create",
RET_ARGUMENT_ERROR, ret);
+ assertTrue(STDOUT_CAPTURE.toString().contains("pass --force"));
assertTrue("No row should be created for a rejected create",
querySystemTable(createGroup, CLUSTERS.getZkUrl1()) == null);
@@ -1166,7 +1198,7 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
@Test(timeout = 180000)
public void testCreateRejectsMalformedZkUrlEvenWhenForced() throws Exception
{
String createGroup = "testCreateBadZkUrl_" + System.currentTimeMillis();
- System.setOut(new PrintStream(STDOUT_CAPTURE));
+ System.setErr(new PrintStream(STDOUT_CAPTURE));
int ret = ToolRunner.run(adminTool,
new String[] { "create", "-g", createGroup, "-p", "FAILOVER", "-zk1",
MALFORMED_URL, "-c1",
@@ -1175,6 +1207,7 @@ public class PhoenixHAAdminToolIT extends HABaseIT {
CLUSTERS.getHdfsUrl2(), "-F" });
assertEquals("Malformed zk-url-1 should be rejected even with --force",
RET_ARGUMENT_ERROR,
ret);
+ assertTrue(STDOUT_CAPTURE.toString().contains("--force cannot bypass"));
assertTrue("No row should be created for a rejected create",
querySystemTable(createGroup, CLUSTERS.getZkUrl1()) == null);
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreClientConfigurationTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreClientConfigurationTest.java
new file mode 100644
index 0000000000..bd6276ba0b
--- /dev/null
+++
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HAGroupStoreClientConfigurationTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.jdbc;
+
+import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STORE_PEER_CACHE_RETRY_INTERVAL_SECONDS;
+import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STORE_SYNC_INTERVAL_SECONDS;
+import static
org.apache.phoenix.query.QueryServices.PHOENIX_HA_LEGACY_CRR_RECONCILIATION_INTERVAL_SECONDS;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+
+public class HAGroupStoreClientConfigurationTest {
+
+ @Test
+ public void testRequiredIntervalsMustBePositive() {
+
assertInvalid(HAGroupStoreClient.PHOENIX_HA_GROUP_STORE_CLIENT_INITIALIZATION_TIMEOUT_MS,
0L);
+
assertInvalid(HAGroupStoreClient.PHOENIX_HA_GROUP_STORE_CLIENT_INITIALIZATION_TIMEOUT_MS,
-1L);
+ assertInvalid(HA_GROUP_STORE_SYNC_INTERVAL_SECONDS, 0L);
+ assertInvalid(HA_GROUP_STORE_SYNC_INTERVAL_SECONDS, -1L);
+ }
+
+ @Test
+ public void testOptionalIntervalsMayDisableWork() {
+ Configuration conf = new Configuration(false);
+ conf.setLong(HA_GROUP_STORE_PEER_CACHE_RETRY_INTERVAL_SECONDS, 0L);
+ conf.setLong(PHOENIX_HA_LEGACY_CRR_RECONCILIATION_INTERVAL_SECONDS, -1L);
+ HAGroupStoreClient.validateConfiguration(conf);
+ }
+
+ private static void assertInvalid(String key, long value) {
+ Configuration conf = new Configuration(false);
+ conf.setLong(key, value);
+ IllegalArgumentException error =
assertThrows(IllegalArgumentException.class,
+ () -> HAGroupStoreClient.validateConfiguration(conf));
+ assertTrue(error.getMessage().contains(key));
+ assertTrue(error.getMessage().contains(String.valueOf(value)));
+ }
+}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PeerClusterWatcherTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PeerClusterWatcherTest.java
index f111ff0a58..37ebdc7314 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PeerClusterWatcherTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PeerClusterWatcherTest.java
@@ -18,11 +18,14 @@
package org.apache.phoenix.jdbc;
import static
org.apache.phoenix.query.QueryServices.HA_GROUP_STORE_PEER_CACHE_RETRY_INTERVAL_SECONDS;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;
@@ -192,4 +195,46 @@ public class PeerClusterWatcherTest {
watcher.close();
}
}
+
+ @Test
+ public void testRecreatedPeerZnodeVersionZeroIsDeliveredOnce() throws
Exception {
+ Configuration conf = new Configuration();
+ conf.setLong(HA_GROUP_STORE_PEER_CACHE_RETRY_INTERVAL_SECONDS, 0L);
+ AtomicInteger deliveries = new AtomicInteger();
+ PeerClusterWatcher watcher =
+ new PeerClusterWatcher(conf, "g", "ns", new
PeerClusterWatcher.PeerStateListener() {
+ @Override
+ public void onPeerStateChanged(HAGroupStoreRecord peerRecord, Stat
stat) {
+ deliveries.incrementAndGet();
+ }
+
+ @Override
+ public void onPeerVisible() {
+ }
+
+ @Override
+ public void onPeerBlind() {
+ }
+ });
+ try {
+ watcher.deliver(recordAndStat(10L, 3), false);
+ watcher.deliver(recordAndStat(20L, 0), false);
+ watcher.deliver(recordAndStat(20L, 0), false);
+ watcher.deliver(recordAndStat(20L, 1), false);
+ watcher.deliver(recordAndStat(5L, 0), false);
+ assertEquals("new znode incarnations and newer versions should be
delivered exactly once", 4,
+ deliveries.get());
+ } finally {
+ watcher.close();
+ }
+ }
+
+ private static Pair<HAGroupStoreRecord, Stat> recordAndStat(long czxid, int
version) {
+ HAGroupStoreRecord record = new HAGroupStoreRecord("v1.0", "g",
+ HAGroupStoreRecord.HAGroupState.STANDBY, 0L, "FAILOVER", null, null,
null, null, null, 1L);
+ Stat stat = new Stat();
+ stat.setCzxid(czxid);
+ stat.setVersion(version);
+ return Pair.of(record, stat);
+ }
}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixHAAdminToolTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixHAAdminToolTest.java
new file mode 100644
index 0000000000..d223a64254
--- /dev/null
+++
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixHAAdminToolTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.jdbc;
+
+import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.ABORT_TO_ACTIVE_IN_SYNC;
+import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.ABORT_TO_STANDBY;
+import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC;
+import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC_TO_STANDBY;
+import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC;
+import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER;
+import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.ACTIVE_WITH_OFFLINE_PEER;
+import static
org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY;
+import static org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.OFFLINE;
+import static org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.STANDBY;
+import static org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState.UNKNOWN;
+import static org.apache.phoenix.jdbc.PhoenixHAAdminTool.RET_ARGUMENT_ERROR;
+import static org.apache.phoenix.jdbc.PhoenixHAAdminTool.RET_SUCCESS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import org.junit.Test;
+
+public class PhoenixHAAdminToolTest {
+
+ @Test
+ public void testStableFailoverPairRejectsTransitionalRoles() {
+ assertTrue(PhoenixHAAdminTool.isStableFailoverPair(STANDBY,
ACTIVE_IN_SYNC));
+ assertTrue(PhoenixHAAdminTool.isStableFailoverPair(STANDBY,
ACTIVE_NOT_IN_SYNC));
+
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(ABORT_TO_STANDBY,
ACTIVE_IN_SYNC));
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(STANDBY,
ABORT_TO_ACTIVE_IN_SYNC));
+
assertFalse(PhoenixHAAdminTool.isStableFailoverPair(ACTIVE_IN_SYNC_TO_STANDBY,
STANDBY));
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(DEGRADED_STANDBY,
ACTIVE_IN_SYNC));
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(STANDBY,
DEGRADED_STANDBY));
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(STANDBY, OFFLINE));
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(STANDBY, UNKNOWN));
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(STANDBY,
ACTIVE_WITH_OFFLINE_PEER));
+ assertFalse(
+ PhoenixHAAdminTool.isStableFailoverPair(STANDBY,
ACTIVE_NOT_IN_SYNC_WITH_OFFLINE_PEER));
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(null, ACTIVE_IN_SYNC));
+ assertFalse(PhoenixHAAdminTool.isStableFailoverPair(STANDBY, null));
+ }
+
+ @Test
+ public void testTopLevelHelpReturnsSuccess() throws Exception {
+ PhoenixHAAdminTool tool = new PhoenixHAAdminTool();
+ assertEquals(RET_SUCCESS, tool.run(new String[] { "help" }));
+ assertEquals(RET_SUCCESS, tool.run(new String[] { "-h" }));
+ assertEquals(RET_SUCCESS, tool.run(new String[] { "--help" }));
+ assertEquals(RET_ARGUMENT_ERROR, tool.run(new String[0]));
+ assertEquals(RET_ARGUMENT_ERROR, tool.run(new String[] { "unknown" }));
+ }
+
+ @Test
+ public void testVersionOnlyUpdateExplainsThatAConfigurationFieldIsRequired()
throws Exception {
+ PrintStream originalErr = System.err;
+ ByteArrayOutputStream error = new ByteArrayOutputStream();
+ try {
+ System.setErr(new PrintStream(error));
+ PhoenixHAAdminTool tool = new PhoenixHAAdminTool();
+ assertEquals(RET_ARGUMENT_ERROR, tool.run(new String[] { "update", "-g",
"g", "-v", "1" }));
+ } finally {
+ System.setErr(originalErr);
+ }
+ assertTrue(error.toString().contains("in addition to"));
+ }
+}
