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 523a58b7f7 PHOENIX-7995 Tag client HA metrics with the HA group and
add missing HA/CRR/failover client metrics (#2609)
523a58b7f7 is described below
commit 523a58b7f7144535a3b17cc974a2b8af582d0a5b
Author: Lokesh Khurana <[email protected]>
AuthorDate: Tue Sep 1 09:58:06 2026 -0700
PHOENIX-7995 Tag client HA metrics with the HA group and add missing
HA/CRR/failover client metrics (#2609)
* PHOENIX-7872 Addendum record HA failover duration on the CRR-write path
and add a connection-failed counter
The HA failover observability metrics added under PHOENIX-7872 recorded
HA_FAILOVER_DURATION_MS inside FailoverPhoenixConnection.failover(long).
That
method is only reached through wrapActionDuringFailover -> FailoverPolicy
.shouldFailover(), which returns false under the default
ExplicitFailoverPolicy,
or through the explicit static failover(Connection, long) helper. Neither
runs
during an autonomous, CRR-driven failover, so the duration metric never
moved in
production.
Move the duration measurement to the path that actually drives failovers:
refreshClusterRoleRecord, where the cluster-role transition is dispatched
and
where HA_FAILOVER_COUNT is already gated by shouldCountFailover. The
dispatch
block is wrapped in a try/finally so the duration is recorded on every exit
(success, timeout, policy failure, or interrupt), avoiding a silent metric
miss
if a future exit path is added. The now-dead timing in failover(long) is
removed.
Add HA_FAILOVER_CONNECTION_FAILED_COUNTER, incremented at the single
SQLException
throw funnel in connectActive (no active cluster, cluster demoted
mid-connect, or
the underlying connect threw). This tracks real active-cluster connection
failures
regardless of the configured failover policy.
Tests: three unit tests in HighAvailabilityGroupTest -- a counted role-flip
transition records both HA_FAILOVER_COUNT and an HA_FAILOVER_DURATION_MS
sample on
the CRR-write path; a failed connectActive increments the connection-failed
counter; a successful connectActive leaves it unchanged.
HighAvailabilityGroupTest
16/16, FailoverPhoenixConnectionTest 8/8.
Generated-by: Claude Code (Opus 4.8)
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
* PHOENIX-7995 Tag client HA metrics with the HA group and add missing
HA/CRR/failover client metrics
The ZK-less HA client emits all HA/CRR/failover metrics as JVM-global
GLOBAL_HA_* counters, so a JVM connected to more than one HA group cannot
attribute a failover, stale-CRR detection, or poller failure to a specific
group. It also lacks client counters for several HA/CRR/failover events that
are relevant on the ZK-less path.
Per-HA-group tagging
--------------------
Add HAGroupClientMetricsSource, a per-group Hadoop Metrics2 source (one per
HA
group name) that stamps a "ha_group" tag carrying the group name and
appends the
quoted group name to its JMX context so each group registers as a distinct
source/MBean. HAGroupMetricsManager is the process-wide registry: it lazily
creates a source per group (only when global client metrics are enabled),
routes per-group increment/update, and detaches the source on HA-group
close so
the same group can re-register later. Creating a source registers it with
the
metrics system and can throw; getOrCreate swallows that failure (logs and
returns null) so metrics registration never breaks a caller such as
HighAvailabilityGroup.init(). Emission is dual: every group-attributable
GLOBAL_HA_* increment is mirrored to the group's source, and the JVM-global
counters continue to emit unchanged.
The tag key lives in a new module-neutral MetricConstants.HA_GROUP_TAG_NAME
("ha_group") referenced by both this client source and the server-side
HAGroupStoreMetricsSource, so both sides tag with the same key and can be
filtered together downstream.
New client counters
-------------------
- HA_FAILOVER_CONNECTION_CREATED_COUNTER: FailoverPhoenixConnection
instances
successfully created against the active cluster; one increment per
constructed
connection. This is not a matched pair with the connection-failed counter,
which increments on every failed active-connect attempt across all callers
(including each failover retry), so failed/(created+failed) is not a rate.
- HA_ROLE_TRANSITION_FAILED_COUNTER: cluster-role-transition dispatch
failures
(execution error or timeout) on the CRR-write path.
- CRR_TRANSITION_COUNT: cluster-role-record transitions applied per HA
policy,
including transitions into a no-active state (distinct from
HA_FAILOVER_COUNT,
which counts only transitions that establish/move an ACTIVE cluster).
Group-attributable emission is wired at the existing sites in
HighAvailabilityGroup (failover count/duration, CRR refresh,
connect-failed, and
the two new transition metrics), FailoverPhoenixConnection (connection
created,
stale-CRR, mutation-blocked), HighAvailabilityPolicy (parallel fallback),
ParallelPhoenixContext/ParallelPhoenixUtil (parallel connection
created/error,
task timeout), and GetClusterRoleRecordUtil (poller tick count/failures).
The
JVM-shared parallel-executor pool metrics and the HA_CRR_CACHE_AGE_MS gauge
are
deliberately excluded from the per-group source (JVM-shared / not a
counter).
Tests: HAGroupClientMetricsSourceTest (6) and HAGroupMetricsManagerTest (7)
cover the ha_group tag/quoting, per-counter increment/update, per-group
isolation, distinct registered sources, and detach/re-create.
HighAvailabilityGroupTest additionally asserts the per-group
(ha_group-tagged)
counters mirror the JVM-global counters on a real counted transition
(HA_FAILOVER_COUNT + CRR_TRANSITION_COUNT) and on a failed / successful
connectActive (HA_FAILOVER_CONNECTION_FAILED_COUNTER), and that init
pre-registers the group's source while close detaches it. 30/30 pass across
the
three.
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../phoenix/jdbc/FailoverPhoenixConnection.java | 106 +++++-----
.../apache/phoenix/jdbc/HighAvailabilityGroup.java | 28 +++
.../phoenix/jdbc/HighAvailabilityPolicy.java | 51 ++++-
.../phoenix/jdbc/ParallelPhoenixContext.java | 5 +
.../apache/phoenix/jdbc/ParallelPhoenixUtil.java | 5 +
.../phoenix/monitoring/GlobalClientMetrics.java | 12 +-
.../monitoring/HAGroupClientMetricsSource.java | 133 ++++++++++++
.../phoenix/monitoring/HAGroupMetricsManager.java | 135 +++++++++++++
.../org/apache/phoenix/monitoring/MetricType.java | 26 ++-
.../phoenix/util/GetClusterRoleRecordUtil.java | 4 +
.../org/apache/phoenix/jdbc/HAGroupMetricsIT.java | 10 +-
.../phoenix/jdbc/HighAvailabilityGroupTest.java | 223 +++++++++++++++++++++
.../monitoring/HAGroupClientMetricsSourceTest.java | 130 ++++++++++++
.../monitoring/HAGroupMetricsManagerTest.java | 164 +++++++++++++++
14 files changed, 962 insertions(+), 70 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java
index 5916ea86c4..8541423269 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/FailoverPhoenixConnection.java
@@ -19,7 +19,7 @@ package org.apache.phoenix.jdbc;
import static
org.apache.phoenix.jdbc.HighAvailabilityUtil.isMutationBlockedIOExceptionExistsInThrowable;
import static
org.apache.phoenix.jdbc.HighAvailabilityUtil.isStaleClusterRoleRecordExceptionExistsInThrowable;
-import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_DURATION_MS;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_CREATED_COUNTER;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_MUTATION_BLOCKED_COUNT;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_STALE_CRR_DETECTED_COUNT;
@@ -44,6 +44,7 @@ import java.util.concurrent.Executor;
import org.apache.phoenix.exception.FailoverSQLException;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.monitoring.HAGroupMetricsManager;
import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.slf4j.Logger;
@@ -111,6 +112,13 @@ public class FailoverPhoenixConnection implements
PhoenixMonitoredConnection {
this.isClosed = false;
this.connection =
context.getHAGroup().connectActive(context.getProperties(),
context.getHAURLInfo());
+ // Count one successfully constructed FailoverPhoenixConnection. This is
not a matched pair with
+ // HA_FAILOVER_CONNECTION_FAILED_COUNTER: that counter increments on every
failed active-connect
+ // attempt across all callers (including each failover() retry), so
failed/(created+failed) is
+ // not a valid rate.
+ GLOBAL_HA_FAILOVER_CONNECTION_CREATED_COUNTER.increment();
+ HAGroupMetricsManager.increment(context.getHAGroup().getName(),
+ MetricType.HA_FAILOVER_CONNECTION_CREATED_COUNTER);
}
/**
@@ -175,63 +183,55 @@ public class FailoverPhoenixConnection implements
PhoenixMonitoredConnection {
return;
}
- final long failoverStartMs = EnvironmentEdgeManager.currentTimeMillis();
- try {
- PhoenixConnection newConn = null;
- SQLException cause = null;
- final long startTime = EnvironmentEdgeManager.currentTimeMillis();
- while (
- newConn == null && EnvironmentEdgeManager.currentTimeMillis() <
startTime + timeoutMs
- ) {
+ PhoenixConnection newConn = null;
+ SQLException cause = null;
+ final long startTime = EnvironmentEdgeManager.currentTimeMillis();
+ while (newConn == null && EnvironmentEdgeManager.currentTimeMillis() <
startTime + timeoutMs) {
+ try {
+ newConn =
+ context.getHAGroup().connectActive(context.getProperties(),
context.getHAURLInfo());
+ } catch (SQLException e) {
+ cause = e;
+ LOG.info("Got exception when trying to connect to active cluster.", e);
try {
- newConn =
- context.getHAGroup().connectActive(context.getProperties(),
context.getHAURLInfo());
- } catch (SQLException e) {
- cause = e;
- LOG.info("Got exception when trying to connect to active cluster.",
e);
- try {
- Thread.sleep(100); // TODO: be smart than this
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- throw new SQLException("Got interrupted waiting for connection
failover", e);
- }
+ Thread.sleep(100); // TODO: be smart than this
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new SQLException("Got interrupted waiting for connection
failover", e);
}
}
- if (newConn == null) {
- throw new FailoverSQLException("Can not failover connection",
- context.getHAGroup().getGroupInfo().toString(), cause);
- }
+ }
+ if (newConn == null) {
+ throw new FailoverSQLException("Can not failover connection",
+ context.getHAGroup().getGroupInfo().toString(), cause);
+ }
- final PhoenixConnection oldConn = connection;
- connection = newConn;
- if (oldConn != null) {
- // aggregate metrics
- previousMutationMetrics = oldConn.getMutationMetrics();
- previousReadMetrics = oldConn.getReadMetrics();
- oldConn.clearMetrics();
-
- // close old connection
- if (!oldConn.isClosed()) {
- // TODO: what happens to in-flight edits/mutations?
- // Can we copy into the new connection we do not allow this failover?
- // MutationState state = oldConn.getMutationState();
- try {
- oldConn.close(new
SQLExceptionInfo.Builder(SQLExceptionCode.HA_CLOSED_AFTER_FAILOVER)
- .setMessage("Phoenix connection got closed due to failover")
-
.setHaGroupInfo(context.getHAGroup().getGroupInfo().toString()).build()
- .buildException());
- } catch (SQLException e) {
- LOG.error("Failed to close old connection after failover: {}",
e.getMessage());
- LOG.info("Full stack when closing old connection after failover",
e);
- }
+ final PhoenixConnection oldConn = connection;
+ connection = newConn;
+ if (oldConn != null) {
+ // aggregate metrics
+ previousMutationMetrics = oldConn.getMutationMetrics();
+ previousReadMetrics = oldConn.getReadMetrics();
+ oldConn.clearMetrics();
+
+ // close old connection
+ if (!oldConn.isClosed()) {
+ // TODO: what happens to in-flight edits/mutations?
+ // Can we copy into the new connection we do not allow this failover?
+ // MutationState state = oldConn.getMutationState();
+ try {
+ oldConn.close(new
SQLExceptionInfo.Builder(SQLExceptionCode.HA_CLOSED_AFTER_FAILOVER)
+ .setMessage("Phoenix connection got closed due to failover")
+
.setHaGroupInfo(context.getHAGroup().getGroupInfo().toString()).build()
+ .buildException());
+ } catch (SQLException e) {
+ LOG.error("Failed to close old connection after failover: {}",
e.getMessage());
+ LOG.info("Full stack when closing old connection after failover", e);
}
}
- LOG.info("Connection {} failed over to {}",
context.getHAGroup().getGroupInfo(),
- connection.getURL());
- } finally {
- GLOBAL_HA_FAILOVER_DURATION_MS
- .update(EnvironmentEdgeManager.currentTimeMillis() - failoverStartMs);
}
+ LOG.info("Connection {} failed over to {}",
context.getHAGroup().getGroupInfo(),
+ connection.getURL());
}
/**
@@ -337,6 +337,8 @@ public class FailoverPhoenixConnection implements
PhoenixMonitoredConnection {
} catch (Exception e) {
if (isStaleClusterRoleRecordExceptionExistsInThrowable(e)) {
GLOBAL_HA_STALE_CRR_DETECTED_COUNT.increment();
+ HAGroupMetricsManager.increment(context.getHAGroup().getName(),
+ MetricType.HA_STALE_CRR_DETECTED_COUNT);
// If we receive StaleClusterRoleRecordException, that means
Operation was
// supposed to be executed on Active Cluster but was in reality was
sent to
// STANDBY Cluster, that can happen only when Failover is in
Progress, So we
@@ -363,6 +365,8 @@ public class FailoverPhoenixConnection implements
PhoenixMonitoredConnection {
}
if (isMutationBlockedIOExceptionExistsInThrowable(e)) {
GLOBAL_HA_MUTATION_BLOCKED_COUNT.increment();
+ HAGroupMetricsManager.increment(context.getHAGroup().getName(),
+ MetricType.HA_MUTATION_BLOCKED_COUNT);
}
if (policy.shouldFailover(e, ++failoverCount)) {
failover(timeoutMs);
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java
index 05ce278764..90669e5f8d 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityGroup.java
@@ -19,7 +19,9 @@ package org.apache.phoenix.jdbc;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_CRR_CACHE_AGE_MS;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_CRR_REFRESH_COUNT;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_ROLE_TRANSITION_FAILED_COUNTER;
import static
org.apache.phoenix.query.QueryServicesOptions.DEFAULT_CLIENT_CONNECTION_CACHE_MAX_DURATION;
import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
@@ -61,6 +63,8 @@ import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole;
import org.apache.phoenix.jdbc.ClusterRoleRecord.RegistryType;
+import org.apache.phoenix.monitoring.HAGroupMetricsManager;
+import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.query.HBaseFactoryProvider;
import org.apache.phoenix.util.GetClusterRoleRecordUtil;
import org.apache.phoenix.util.JDBCUtil;
@@ -635,6 +639,11 @@ public class HighAvailabilityGroup {
roleRecord = roleRecordFromEndpoint;
lastClusterRoleRecordRefreshTime = System.currentTimeMillis();
state = State.READY;
+ // Pre-register the per-group metrics2 source so the ha_group-tagged
series exists as soon as
+ // the
+ // group is ready, rather than only after the first HA metric fires. No-op
when global client
+ // metrics are disabled.
+ HAGroupMetricsManager.getOrCreate(getName());
}
/**
@@ -700,6 +709,11 @@ public class HighAvailabilityGroup {
} catch (SQLException e) {
LOG.error("Failed to connect to active cluster in HA group {}, record:
{}", info, roleRecord,
e);
+ // Single throw funnel for a failed active-cluster connection (no active
cluster, cluster
+ // demoted mid-connect, or the underlying connect threw). Counted here
so it tracks real
+ // production failures regardless of failover policy.
+ GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.increment();
+ HAGroupMetricsManager.increment(getName(),
MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER);
throw new
SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION)
.setMessage("Failed to connect to active cluster in HA group")
.setHaGroupInfo(info.toString()).setRootCause(e).build().buildException();
@@ -789,6 +803,8 @@ public class HighAvailabilityGroup {
*/
void close() {
state = State.CLOSED;
+ // Detach the per-group metrics2 source so its JMX-context name is freed
for possible re-create.
+ HAGroupMetricsManager.remove(getName());
}
@Override
@@ -1160,6 +1176,7 @@ public class HighAvailabilityGroup {
// otherwise inflate this counter against its name (a "refresh" with no
fetch is a no-op
// from a CRR-state perspective).
GLOBAL_HA_CRR_REFRESH_COUNT.increment();
+ HAGroupMetricsManager.increment(getName(),
MetricType.HA_CRR_REFRESH_COUNT);
if (roleRecord == null) {
// First-load init path: no prior cache state to compare against, so
this is not a
// failover transition and HA_FAILOVER_COUNT is intentionally NOT
incremented here.
@@ -1197,6 +1214,11 @@ public class HighAvailabilityGroup {
return true;
}
+ // The detected CRR change is dispatched to the HA policy, which decides
whether it is a real
+ // transition (a url or role change) versus a pure version bump, and
counts and times it
+ // there (see HighAvailabilityPolicy#transitClusterRoleRecord). The
equals() and
+ // shouldApplyRefreshedRecord() guards above only establish that this is
a strictly newer
+ // record for the same HA group.
final ClusterRoleRecord oldRecord = roleRecord;
state = State.IN_TRANSITION;
LOG.info("HA group {} is in {} to set V{} record", info, state,
newRoleRecord.getVersion());
@@ -1225,6 +1247,11 @@ public class HighAvailabilityGroup {
} catch (ExecutionException | TimeoutException e) {
LOG.error("HA group {} failed to transit cluster roles per policy {}
to new " + "record {}",
info, roleRecord.getPolicy(), newRoleRecord, e);
+ // Dispatch of the policy-side cluster-role transition failed
(execution error or timed
+ // out). Count every such failure, including the
HA_ROLE_TRANSITION_NOT_ALLOWED case
+ // rethrown just below.
+ GLOBAL_HA_ROLE_TRANSITION_FAILED_COUNTER.increment();
+ HAGroupMetricsManager.increment(getName(),
MetricType.HA_ROLE_TRANSITION_FAILED_COUNTER);
// Rethrow the Role transitions not allowed exceptions
if (e.getCause() != null && e.getCause().getCause() != null) {
if (
@@ -1252,6 +1279,7 @@ public class HighAvailabilityGroup {
// unit-tested directly without driving a full mini-cluster transition.
if (shouldCountFailover(transitionSucceeded, oldRecord, newRoleRecord)) {
GLOBAL_HA_FAILOVER_COUNT.increment();
+ HAGroupMetricsManager.increment(getName(),
MetricType.HA_FAILOVER_COUNT);
}
// Update the role record and the last refresh time
roleRecord = newRoleRecord;
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java
index 6b898ebeb5..267171dcb3 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HighAvailabilityPolicy.java
@@ -30,6 +30,8 @@ import java.util.Properties;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.monitoring.GlobalClientMetrics;
+import org.apache.phoenix.monitoring.HAGroupMetricsManager;
+import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -194,6 +196,8 @@ public enum HighAvailabilityPolicy {
// Give regular connection or a failover connection?
LOG.warn("Falling back to single phoenix connection due to resource
constraints");
GlobalClientMetrics.GLOBAL_HA_PARALLEL_CONNECTION_FALLBACK_COUNTER.increment();
+ HAGroupMetricsManager.increment(haGroup.getName(),
+ MetricType.HA_PARALLEL_CONNECTION_FALLBACK_COUNTER);
return haGroup.connectActive(info, haURLInfo);
}
}
@@ -322,15 +326,44 @@ public enum HighAvailabilityPolicy {
*/
public void transitClusterRoleRecord(HighAvailabilityGroup haGroup,
ClusterRoleRecord oldRecord,
ClusterRoleRecord newRecord) throws SQLException {
- if (
- !oldRecord.getUrl1().equals(newRecord.getUrl1())
- || !oldRecord.getUrl2().equals(newRecord.getUrl2())
- ) {
- // If there is a change in url then we need to handle the transitions.
- transitClusterUrl(haGroup, oldRecord, newRecord);
- } else {
- // If url is not changing then we need to check if there is a role
transition.
- transitClusterRole(haGroup, oldRecord, newRecord);
+ boolean registryChanged = oldRecord.getRegistryType() !=
newRecord.getRegistryType();
+ boolean urlChanged = !oldRecord.getUrl1().equals(newRecord.getUrl1())
+ || !oldRecord.getUrl2().equals(newRecord.getUrl2());
+ boolean roleChanged =
+ oldRecord.getRole1() != newRecord.getRole1() || oldRecord.getRole2() !=
newRecord.getRole2();
+
+ // "Transition" here means the CRR changed in a field this dispatch reacts
to: registry, url, or
+ // role. A strictly newer CRR version can arrive with all three identical
(a pure version bump)
+ // after clearing the isNewerThan + hasSameInfo guards in
refreshClusterRoleRecord; that is not
+ // a transition, so skip it rather than inflate CRR_TRANSITION_COUNT or
push a sample into
+ // CRR_TRANSITION_DURATION_MS.
+ if (!registryChanged && !urlChanged && !roleChanged) {
+ LOG.info("HA group {} applied a newer CRR version with no
registry/url/role change; no "
+ + "transition work to do", haGroup.getGroupInfo());
+ return;
+ }
+
+ // Count and time the transition here on the CRR-write path, the
autonomous-failover path for
+ // the ZK-less client: refreshClusterRoleRecord detects a newer CRR and
dispatches here to run
+ // the close/invalidate work per HA policy. Duration is recorded in
finally so a transit that
+ // fails partway still contributes its time-to-failure;
HA_ROLE_TRANSITION_FAILED_COUNTER
+ // (recorded by the caller on dispatch failure/timeout) distinguishes
failed transits.
+ GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_COUNT.increment();
+ HAGroupMetricsManager.increment(haGroup.getName(),
MetricType.CRR_TRANSITION_COUNT);
+ long startTime = System.currentTimeMillis();
+ try {
+ if (urlChanged) {
+ // A change in url requires handling the connection transitions.
+ transitClusterUrl(haGroup, oldRecord, newRecord);
+ } else {
+ // If url is not changing then handle any role transition.
+ transitClusterRole(haGroup, oldRecord, newRecord);
+ }
+ } finally {
+ long durationMs = System.currentTimeMillis() - startTime;
+
GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_DURATION_MS.update(durationMs);
+ HAGroupMetricsManager.update(haGroup.getName(),
MetricType.CRR_TRANSITION_DURATION_MS,
+ durationMs);
}
}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixContext.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixContext.java
index 928ad2923b..f6269e7066 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixContext.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixContext.java
@@ -32,6 +32,7 @@ import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.monitoring.HAGroupMetricsManager;
import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
@@ -80,6 +81,8 @@ public class ParallelPhoenixContext {
Preconditions.checkArgument(executors.size() >= 2,
"Expected 2 executor pairs, one for each connection with a normal/close
executor");
GLOBAL_HA_PARALLEL_CONNECTION_CREATED_COUNTER.increment();
+ HAGroupMetricsManager.increment(haGroup != null ? haGroup.getName() : null,
+ MetricType.HA_PARALLEL_CONNECTION_CREATED_COUNTER);
this.properties = properties;
this.haGroup = haGroup;
this.haurlInfo = haurlInfo;
@@ -176,6 +179,8 @@ public class ParallelPhoenixContext {
isClosed = true;
if (isErrored) {
GLOBAL_HA_PARALLEL_CONNECTION_ERROR_COUNTER.increment();
+ HAGroupMetricsManager.increment(this.haGroup != null ?
this.haGroup.getName() : null,
+ MetricType.HA_PARALLEL_CONNECTION_ERROR_COUNTER);
}
}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixUtil.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixUtil.java
index b223e3d9fb..c8f842345a 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixUtil.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixUtil.java
@@ -34,7 +34,9 @@ import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.PairOfSameType;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.monitoring.HAGroupMetricsManager;
import org.apache.phoenix.monitoring.Metric;
+import org.apache.phoenix.monitoring.MetricType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -126,6 +128,9 @@ public class ParallelPhoenixUtil {
if (timedout) {
GLOBAL_HA_PARALLEL_TASK_TIMEOUT_COUNTER.increment();
+ HAGroupMetricsManager.increment(
+ context.getHaGroup() != null ? context.getHaGroup().getName() : null,
+ MetricType.HA_PARALLEL_TASK_TIMEOUT_COUNTER);
if (futures.isEmpty()) {
LOGGER.warn("Unexpected race between timeout and failure occurred.");
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java
index bbd23bcbc5..b988730f14 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java
@@ -35,10 +35,13 @@ import static
org.apache.phoenix.monitoring.MetricType.COUNT_ROWS_SCANNED;
import static org.apache.phoenix.monitoring.MetricType.COUNT_RPC_CALLS;
import static org.apache.phoenix.monitoring.MetricType.COUNT_RPC_RETRIES;
import static org.apache.phoenix.monitoring.MetricType.COUNT_SCANNED_REGIONS;
+import static org.apache.phoenix.monitoring.MetricType.CRR_TRANSITION_COUNT;
+import static
org.apache.phoenix.monitoring.MetricType.CRR_TRANSITION_DURATION_MS;
import static org.apache.phoenix.monitoring.MetricType.HA_CRR_CACHE_AGE_MS;
import static org.apache.phoenix.monitoring.MetricType.HA_CRR_REFRESH_COUNT;
+import static
org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_CONNECTION_CREATED_COUNTER;
+import static
org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_COUNT;
-import static org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_DURATION_MS;
import static
org.apache.phoenix.monitoring.MetricType.HA_MUTATION_BLOCKED_COUNT;
import static
org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_CONNECTION_CREATED_COUNTER;
import static
org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_CONNECTION_ERROR_COUNTER;
@@ -56,6 +59,7 @@ import static
org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_POOL2_TASK_RE
import static
org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_TASK_TIMEOUT_COUNTER;
import static org.apache.phoenix.monitoring.MetricType.HA_POLLER_TICK_COUNT;
import static org.apache.phoenix.monitoring.MetricType.HA_POLLER_TICK_FAILURES;
+import static
org.apache.phoenix.monitoring.MetricType.HA_ROLE_TRANSITION_FAILED_COUNTER;
import static
org.apache.phoenix.monitoring.MetricType.HA_STALE_CRR_DETECTED_COUNT;
import static org.apache.phoenix.monitoring.MetricType.HCONNECTIONS_COUNTER;
import static
org.apache.phoenix.monitoring.MetricType.INDEX_COMMIT_FAILURE_SIZE;
@@ -172,7 +176,11 @@ public enum GlobalClientMetrics {
GLOBAL_HA_PARALLEL_CONNECTION_CREATED_COUNTER(HA_PARALLEL_CONNECTION_CREATED_COUNTER),
GLOBAL_HA_FAILOVER_COUNT(HA_FAILOVER_COUNT),
- GLOBAL_HA_FAILOVER_DURATION_MS(HA_FAILOVER_DURATION_MS),
+
GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER(HA_FAILOVER_CONNECTION_FAILED_COUNTER),
+
GLOBAL_HA_FAILOVER_CONNECTION_CREATED_COUNTER(HA_FAILOVER_CONNECTION_CREATED_COUNTER),
+ GLOBAL_HA_ROLE_TRANSITION_FAILED_COUNTER(HA_ROLE_TRANSITION_FAILED_COUNTER),
+ GLOBAL_HA_CRR_TRANSITION_COUNT(CRR_TRANSITION_COUNT),
+ GLOBAL_HA_CRR_TRANSITION_DURATION_MS(CRR_TRANSITION_DURATION_MS),
GLOBAL_HA_MUTATION_BLOCKED_COUNT(HA_MUTATION_BLOCKED_COUNT),
GLOBAL_HA_STALE_CRR_DETECTED_COUNT(HA_STALE_CRR_DETECTED_COUNT),
GLOBAL_HA_CRR_REFRESH_COUNT(HA_CRR_REFRESH_COUNT),
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSource.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSource.java
new file mode 100644
index 0000000000..7b6b127059
--- /dev/null
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSource.java
@@ -0,0 +1,133 @@
+/*
+ * 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.monitoring;
+
+import static org.apache.phoenix.monitoring.MetricType.CRR_TRANSITION_COUNT;
+import static
org.apache.phoenix.monitoring.MetricType.CRR_TRANSITION_DURATION_MS;
+import static org.apache.phoenix.monitoring.MetricType.HA_CRR_REFRESH_COUNT;
+import static
org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_CONNECTION_CREATED_COUNTER;
+import static
org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.HA_FAILOVER_COUNT;
+import static
org.apache.phoenix.monitoring.MetricType.HA_MUTATION_BLOCKED_COUNT;
+import static
org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_CONNECTION_CREATED_COUNTER;
+import static
org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_CONNECTION_ERROR_COUNTER;
+import static
org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_CONNECTION_FALLBACK_COUNTER;
+import static
org.apache.phoenix.monitoring.MetricType.HA_PARALLEL_TASK_TIMEOUT_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.HA_POLLER_TICK_COUNT;
+import static org.apache.phoenix.monitoring.MetricType.HA_POLLER_TICK_FAILURES;
+import static
org.apache.phoenix.monitoring.MetricType.HA_ROLE_TRANSITION_FAILED_COUNTER;
+import static
org.apache.phoenix.monitoring.MetricType.HA_STALE_CRR_DETECTED_COUNT;
+
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+import javax.management.ObjectName;
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.hadoop.metrics2.lib.Interns;
+import org.apache.hadoop.metrics2.lib.MutableFastCounter;
+import org.apache.phoenix.metrics.MetricConstants;
+
+import
org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
+import org.apache.phoenix.thirdparty.com.google.common.collect.ImmutableList;
+
+/**
+ * Per-HA-group Hadoop Metrics2 source for the ZK-less HA client's
HA/CRR/failover metrics.
+ * <p>
+ * All groups share one metrics name/context; each instance appends {@code
,haGroup=<quoted-name>}
+ * to the JMX context so it registers as a distinct metrics2 source / MBean,
and stamps a
+ * {@code ha_group} tag ({@link MetricConstants#HA_GROUP_TAG_NAME}) carrying
the (unquoted) group
+ * name so the series can be sliced per HA group downstream. This mirrors the
server-side
+ * {@code HAGroupStoreMetricsSourceImpl} tagging pattern; both reference the
same tag key.
+ * <p>
+ * Each metric in {@link #METRIC_TYPES} is a monotonic {@link
MutableFastCounter}. Accumulating
+ * metrics such as {@code CRR_TRANSITION_DURATION_MS} add total milliseconds
via {@link #update},
+ * which matches the semantics of the JVM-global {@code GLOBAL_HA_*} counters;
those continue to
+ * emit unchanged alongside this per-group source (dual emit).
+ * <p>
+ * The set is intentionally limited to metrics attributable to a single HA
group (each emission site
+ * has a {@code HighAvailabilityGroup} in scope). The shared parallel-executor
pool metrics
+ * ({@code HA_PARALLEL_POOL1_*}/{@code HA_PARALLEL_POOL2_*}) and the {@code
HA_CRR_CACHE_AGE_MS}
+ * gauge are excluded: the pools are JVM-shared and the gauge is not a counter.
+ */
+public class HAGroupClientMetricsSource extends BaseSourceImpl {
+
+ static final String METRICS_NAME = "HAGroupClient";
+ static final String METRICS_DESC = "Phoenix HA Group Client Metrics";
+ static final String METRICS_CONTEXT = "phoenix";
+ static final String METRICS_JMX_CONTEXT = "Phoenix,sub=" + METRICS_NAME;
+
+ /**
+ * The HA metrics attributable to a single HA group; each emission site has a
+ * {@code HighAvailabilityGroup} in scope.
+ */
+ private static final List<MetricType> METRIC_TYPES =
ImmutableList.of(HA_FAILOVER_COUNT,
+ CRR_TRANSITION_DURATION_MS, HA_FAILOVER_CONNECTION_CREATED_COUNTER,
+ HA_FAILOVER_CONNECTION_FAILED_COUNTER, HA_STALE_CRR_DETECTED_COUNT,
HA_MUTATION_BLOCKED_COUNT,
+ HA_CRR_REFRESH_COUNT, HA_ROLE_TRANSITION_FAILED_COUNTER,
CRR_TRANSITION_COUNT,
+ HA_PARALLEL_CONNECTION_FALLBACK_COUNTER,
HA_PARALLEL_CONNECTION_CREATED_COUNTER,
+ HA_PARALLEL_CONNECTION_ERROR_COUNTER, HA_PARALLEL_TASK_TIMEOUT_COUNTER,
HA_POLLER_TICK_COUNT,
+ HA_POLLER_TICK_FAILURES);
+
+ private final Map<MetricType, MutableFastCounter> counters = new
EnumMap<>(MetricType.class);
+
+ public HAGroupClientMetricsSource(String haGroupName) {
+ super(METRICS_NAME, METRICS_DESC, METRICS_CONTEXT,
+ METRICS_JMX_CONTEXT + ",haGroup=" + ObjectName.quote(haGroupName));
+ getMetricsRegistry().tag(
+ Interns.info(MetricConstants.HA_GROUP_TAG_NAME,
MetricConstants.HA_GROUP_TAG_DESC),
+ haGroupName);
+ for (MetricType type : METRIC_TYPES) {
+ counters.put(type,
+ getMetricsRegistry().newCounter(type.columnName(), type.description(),
0L));
+ }
+ }
+
+ /** Increment the group's counter for the given HA metric type. */
+ public void increment(MetricType type) {
+ MutableFastCounter counter = counters.get(type);
+ if (counter != null) {
+ counter.incr();
+ }
+ }
+
+ /**
+ * Add {@code value} to the group's accumulating counter for the given HA
metric type (used for
+ * {@code CRR_TRANSITION_DURATION_MS}).
+ */
+ public void update(MetricType type, long value) {
+ MutableFastCounter counter = counters.get(type);
+ if (counter != null) {
+ counter.incr(value);
+ }
+ }
+
+ /**
+ * Detach this source from the metrics system on HA-group close, freeing its
JMX-context name in
+ * {@link DefaultMetricsSystem} so the same group can register again if
later re-created.
+ */
+ public void unregister() {
+ DefaultMetricsSystem.instance().unregisterSource(metricsJmxContext);
+ }
+
+ @VisibleForTesting
+ public long getCounterValue(MetricType type) {
+ MutableFastCounter counter = counters.get(type);
+ return counter == null ? -1L : counter.value();
+ }
+}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupMetricsManager.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupMetricsManager.java
new file mode 100644
index 0000000000..ee32f706ad
--- /dev/null
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/HAGroupMetricsManager.java
@@ -0,0 +1,135 @@
+/*
+ * 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.monitoring;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import
org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
+
+/**
+ * Central registry of {@link HAGroupClientMetricsSource}, one per HA group
name. Each group's
+ * source registers with Hadoop's metrics2 {@code DefaultMetricsSystem} tagged
+ * {@code haGroup=<name>} so the HA/CRR/failover client metrics can be sliced
per HA group alongside
+ * the JVM-global {@code GLOBAL_HA_*} counters (which continue to emit
unchanged).
+ * <p>
+ * Sources are created only when global client metrics are enabled
(constructing a source registers
+ * it with the metrics system). {@link #remove(String)} detaches a group's
source on HA-group close.
+ */
+public class HAGroupMetricsManager {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(HAGroupMetricsManager.class);
+
+ private static final boolean IS_GLOBAL_METRICS_ENABLED =
+ QueryServicesOptions.withDefaults().isGlobalMetricsEnabled();
+
+ private static final Map<String, HAGroupClientMetricsSource> GROUP_SOURCES =
+ new ConcurrentHashMap<>();
+
+ private HAGroupMetricsManager() {
+ }
+
+ /**
+ * Get (registering a metrics2 source on first sight) the metrics source for
an HA group. Safe to
+ * call repeatedly for the same name. Returns {@code null} for a null/empty
group name, when
+ * global client metrics are disabled, or when registering the source fails.
+ */
+ public static HAGroupClientMetricsSource getOrCreate(String haGroupName) {
+ if (StringUtils.isEmpty(haGroupName) || !IS_GLOBAL_METRICS_ENABLED) {
+ return null;
+ }
+ HAGroupClientMetricsSource source = GROUP_SOURCES.get(haGroupName);
+ if (source != null) {
+ return source;
+ }
+ synchronized (HAGroupMetricsManager.class) {
+ source = GROUP_SOURCES.get(haGroupName);
+ if (source == null) {
+ try {
+ source = new HAGroupClientMetricsSource(haGroupName);
+ } catch (Exception e) {
+ // Constructing a source registers it with DefaultMetricsSystem and
can throw (e.g. a
+ // MetricsException on a duplicate JMX name). Metrics registration
is best-effort and must
+ // never break a caller such as HighAvailabilityGroup.init();
swallow and skip this group.
+ LOGGER.error("Failed creating HA-group client metrics source for
group '{}'", haGroupName,
+ e);
+ return null;
+ }
+ GROUP_SOURCES.put(haGroupName, source);
+ LOGGER.info("Created HA-group client metrics source for group '{}'",
haGroupName);
+ }
+ }
+ return source;
+ }
+
+ /**
+ * Increment the per-group counter for an HA metric type. Registers the
group's source if needed.
+ * Metric emission is best-effort and never propagates an exception to the
caller's request path.
+ */
+ public static void increment(String haGroupName, MetricType type) {
+ try {
+ HAGroupClientMetricsSource source = getOrCreate(haGroupName);
+ if (source != null) {
+ source.increment(type);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed incrementing HA-group metric {} for group '{}'",
type, haGroupName, e);
+ }
+ }
+
+ /**
+ * Add a sample to a per-group accumulating HA metric (e.g. {@code
CRR_TRANSITION_DURATION_MS}).
+ * Registers the group's source if needed. Best-effort; never propagates an
exception.
+ */
+ public static void update(String haGroupName, MetricType type, long value) {
+ try {
+ HAGroupClientMetricsSource source = getOrCreate(haGroupName);
+ if (source != null) {
+ source.update(type, value);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Failed updating HA-group metric {} for group '{}'", type,
haGroupName, e);
+ }
+ }
+
+ /**
+ * Tear down a group's metrics source on HA-group close, freeing its
metrics2 source name so the
+ * same group can be re-created later.
+ */
+ public static void remove(String haGroupName) {
+ if (StringUtils.isEmpty(haGroupName)) {
+ return;
+ }
+ synchronized (HAGroupMetricsManager.class) {
+ HAGroupClientMetricsSource source = GROUP_SOURCES.remove(haGroupName);
+ if (source != null) {
+ source.unregister();
+ LOGGER.info("Removed HA-group client metrics source for group '{}'",
haGroupName);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ public static HAGroupClientMetricsSource getIfPresent(String haGroupName) {
+ return GROUP_SOURCES.get(haGroupName);
+ }
+}
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java
index 22a9ecbdcf..3b8f0597f7 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java
@@ -358,9 +358,29 @@ public enum MetricType {
"Counter for cluster-level failover transitions (active URL flips between
peers or recovers "
+ "from no-active state) recorded at the CRR write site",
LogLevel.DEBUG, PLong.INSTANCE),
- HA_FAILOVER_DURATION_MS("hafd",
- "Total time in milliseconds spent in connection-level failover
transitions, summed across "
- + "all observing connections (per-connection observation, not
per-cluster-event)",
+ CRR_TRANSITION_DURATION_MS("crtdms",
+ "Total time in milliseconds spent dispatching cluster role record
transitions per HA policy "
+ + "(HighAvailabilityPolicy#transitClusterRoleRecord), recorded on each
applied "
+ + "registry/url/role change; pairs with CRR_TRANSITION_COUNT",
+ LogLevel.DEBUG, PLong.INSTANCE),
+ HA_FAILOVER_CONNECTION_FAILED_COUNTER("hafcf",
+ "Counter for failed attempts to connect to the active cluster in an HA
group, recorded at the "
+ + "connectActive throw site (no active cluster, demoted mid-connect, or
connect error); "
+ + "incremented on every failed attempt across all callers, including
each failover retry",
+ LogLevel.DEBUG, PLong.INSTANCE),
+ HA_FAILOVER_CONNECTION_CREATED_COUNTER("hafcc",
+ "Counter for FailoverPhoenixConnection instances successfully created
(connected to the active "
+ + "cluster); one increment per constructed connection. Not a matched
pair with "
+ + "HA_FAILOVER_CONNECTION_FAILED_COUNTER, which counts every failed
active-connect attempt "
+ + "across all callers (including per failover retry)",
+ LogLevel.DEBUG, PLong.INSTANCE),
+ HA_ROLE_TRANSITION_FAILED_COUNTER("hrtf",
+ "Counter for cluster-role-transition dispatch failures
(ExecutionException/TimeoutException) "
+ + "when applying a new CRR per HA policy at the CRR-write path",
+ LogLevel.DEBUG, PLong.INSTANCE),
+ CRR_TRANSITION_COUNT("crtc",
+ "Counter for cluster role record transitions applied per HA policy (a
registry/url/role change "
+ + "that triggers a transit), including transitions to a no-active state",
LogLevel.DEBUG, PLong.INSTANCE),
HA_MUTATION_BLOCKED_COUNT("hambc",
"Counter for MutationBlockedIOException surfaces caught by
wrapActionDuringFailover",
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java
index b25d5f46f2..1bc202d87e 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/util/GetClusterRoleRecordUtil.java
@@ -43,6 +43,8 @@ import org.apache.phoenix.jdbc.HighAvailabilityGroup;
import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.monitoring.GlobalClientMetrics;
+import org.apache.phoenix.monitoring.HAGroupMetricsManager;
+import org.apache.phoenix.monitoring.MetricType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -223,6 +225,7 @@ public class GetClusterRoleRecordUtil {
// Increment unconditionally so a failed tick still alternates next
iteration.
long tick = tickCount.getAndIncrement();
GlobalClientMetrics.GLOBAL_HA_POLLER_TICK_COUNT.increment();
+ HAGroupMetricsManager.increment(haGroupName,
MetricType.HA_POLLER_TICK_COUNT);
// Sample current CRR cache age into the gauge each tick. Without
this, the
// HA_CRR_CACHE_AGE_MS counter-backed gauge is only updated on
connect() and would
// not advance during idle periods between connects, making it look
fresher than it
@@ -271,6 +274,7 @@ public class GetClusterRoleRecordUtil {
}
} catch (SQLException e) {
GlobalClientMetrics.GLOBAL_HA_POLLER_TICK_FAILURES.increment();
+ HAGroupMetricsManager.increment(haGroupName,
MetricType.HA_POLLER_TICK_FAILURES);
LOGGER.error(
"Exception found while polling for ClusterRoleRecord on {} for HA
group" + " {}: {}",
tickUrl, haGroupName, e.getMessage());
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 7f02288cc6..efe9371972 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
@@ -21,8 +21,8 @@ import static
org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATT
import static
org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.getHighAvailibilityGroup;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_CRR_CACHE_AGE_MS;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_CRR_REFRESH_COUNT;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_DURATION_MS;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT;
-import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_FAILOVER_DURATION_MS;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_MUTATION_BLOCKED_COUNT;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_POLLER_TICK_COUNT;
import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_POLLER_TICK_FAILURES;
@@ -109,7 +109,7 @@ public class HAGroupMetricsIT extends HABaseIT {
@Test(timeout = 300000)
public void testFailoverCountAndDuration() throws Exception {
long countBefore = GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue();
- long durationBefore =
GLOBAL_HA_FAILOVER_DURATION_MS.getMetric().getValue();
+ long durationBefore =
GLOBAL_HA_CRR_TRANSITION_DURATION_MS.getMetric().getValue();
try (Connection conn =
DriverManager.getConnection(CLUSTERS.getJdbcHAUrl(), clientProperties)) {
conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES (1,
1)");
@@ -118,10 +118,10 @@ public class HAGroupMetricsIT extends HABaseIT {
}
long countAfter = GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue();
- long durationAfter = GLOBAL_HA_FAILOVER_DURATION_MS.getMetric().getValue();
+ long durationAfter =
GLOBAL_HA_CRR_TRANSITION_DURATION_MS.getMetric().getValue();
assertTrue("HA_FAILOVER_COUNT should increment on cluster role transition",
countAfter > countBefore);
- assertTrue("HA_FAILOVER_DURATION_MS sum should grow on transition",
+ assertTrue("CRR_TRANSITION_DURATION_MS sum should grow on transition",
durationAfter >= durationBefore);
}
@@ -281,7 +281,7 @@ public class HAGroupMetricsIT extends HABaseIT {
private void resetAllHaMetrics() {
GLOBAL_HA_FAILOVER_COUNT.getMetric().reset();
- GLOBAL_HA_FAILOVER_DURATION_MS.getMetric().reset();
+ GLOBAL_HA_CRR_TRANSITION_DURATION_MS.getMetric().reset();
GLOBAL_HA_MUTATION_BLOCKED_COUNT.getMetric().reset();
GLOBAL_HA_STALE_CRR_DETECTED_COUNT.getMetric().reset();
GLOBAL_HA_CRR_REFRESH_COUNT.getMetric().reset();
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java
index 5bf956fa88..5d275f3da7 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/HighAvailabilityGroupTest.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.jdbc;
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.assertTrue;
@@ -27,12 +28,16 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.sql.SQLException;
+import java.util.HashSet;
import java.util.Properties;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole;
import org.apache.phoenix.jdbc.HighAvailabilityGroup.HAGroupInfo;
import org.apache.phoenix.jdbc.HighAvailabilityGroup.State;
import org.apache.phoenix.monitoring.GlobalClientMetrics;
+import org.apache.phoenix.monitoring.HAGroupClientMetricsSource;
+import org.apache.phoenix.monitoring.HAGroupMetricsManager;
+import org.apache.phoenix.monitoring.MetricType;
import org.junit.Test;
import org.mockito.Mockito;
import org.slf4j.Logger;
@@ -618,4 +623,222 @@ public class HighAvailabilityGroupTest {
Thread.interrupted();
}
}
+
+ /**
+ * A real counted role-flip transition driven through {@code
refreshClusterRoleRecord} must both
+ * increment {@code HA_FAILOVER_COUNT} and record a sample on {@code
CRR_TRANSITION_DURATION_MS}.
+ * This pins the duration metric to the CRR-write transition path (the path
that autonomous
+ * failovers actually take) rather than the connection-level
+ * {@code FailoverPhoenixConnection.failover()} path, which is never
auto-invoked under the
+ * default {@code ExplicitFailoverPolicy}. Active URL flips from url1 to
url2 so
+ * {@code shouldCountFailover} returns true; the {@code URLS} entry is
seeded empty so the policy
+ * transition is a clean no-op (no real connections needed).
+ */
+ @Test
+ public void testCountedTransitionRecordsFailoverCountAndDuration() throws
Exception {
+ String haGroupName =
"testCountedTransitionRecordsFailoverCountAndDuration";
+ String url1 = "host1\\:60010";
+ String url2 = "host2\\:60010";
+ ClusterRoleRecord aActiveBStandby = new ClusterRoleRecord(haGroupName,
+ HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2,
ClusterRole.STANDBY, 10L);
+ ClusterRoleRecord aStandbyBActive = new ClusterRoleRecord(haGroupName,
+ HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2,
ClusterRole.ACTIVE, 11L);
+
+ HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2);
+ // Seed an empty URL set so the policy-side transition iterates nothing
and is a clean no-op.
+ HighAvailabilityGroup.URLS.put(info, new HashSet<>());
+ try {
+ HighAvailabilityGroup group = Mockito
+ .spy(new HighAvailabilityGroup(info, new Properties(),
aActiveBStandby, State.READY));
+
Mockito.doReturn(aStandbyBActive).when(group).getClusterRoleRecordFromEndpoint();
+
+ long countBefore =
GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue();
+ long durationSamplesBefore =
+
GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_DURATION_MS.getMetric().getNumberOfSamples();
+
+ assertTrue("A real role-flip transition must apply and return true",
+ group.refreshClusterRoleRecord(true));
+ assertSame("The new record must be applied after the transition",
aStandbyBActive,
+ group.getRoleRecord());
+
+ assertEquals("An active-URL flip must increment HA_FAILOVER_COUNT",
countBefore + 1,
+ GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue());
+ assertEquals(
+ "The transition must record a CRR_TRANSITION_DURATION_MS sample on the
CRR-write " + "path",
+ durationSamplesBefore + 1,
+
GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_DURATION_MS.getMetric().getNumberOfSamples());
+
+ // Per-group (ha_group-tagged) emission mirrors the JVM-global counters.
This group name is
+ // unique to this test, so its source starts empty and this single
transition yields exactly 1
+ // on both the failover and the applied-transition counters.
+ HAGroupClientMetricsSource source =
HAGroupMetricsManager.getIfPresent(haGroupName);
+ assertNotNull("The transition must register the per-group metrics
source", source);
+ assertEquals("Per-group HA_FAILOVER_COUNT must count the active-URL
flip", 1L,
+ source.getCounterValue(MetricType.HA_FAILOVER_COUNT));
+ assertEquals("Per-group CRR_TRANSITION_COUNT must count the applied
transition", 1L,
+ source.getCounterValue(MetricType.CRR_TRANSITION_COUNT));
+ } finally {
+ HighAvailabilityGroup.URLS.remove(info);
+ HAGroupMetricsManager.remove(haGroupName);
+ }
+ }
+
+ /**
+ * A strictly newer CRR version whose registry/url/role fields are all
identical is a pure version
+ * bump, not a transition. It clears the {@code equals()} and {@code
shouldApplyRefreshedRecord()}
+ * guards in {@code refreshClusterRoleRecord} (so the newer record is still
adopted), but the HA
+ * policy's change guard in {@code transitClusterRoleRecord} must skip it: no
+ * {@code CRR_TRANSITION_COUNT} increment, no {@code
CRR_TRANSITION_DURATION_MS} sample, and no
+ * failover. This pins the guard against a regression that would re-inflate
the transition
+ * counters on version bumps.
+ */
+ @Test
+ public void testPureVersionBumpDoesNotCountOrTimeTransition() throws
Exception {
+ String haGroupName = "testPureVersionBumpDoesNotCountOrTimeTransition";
+ String url1 = "host1\\:60010";
+ String url2 = "host2\\:60010";
+ ClusterRoleRecord recordV10 = new ClusterRoleRecord(haGroupName,
+ HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2,
ClusterRole.STANDBY, 10L);
+ // Same registry/url/role, only the admin version advances.
+ ClusterRoleRecord recordV11 = new ClusterRoleRecord(haGroupName,
+ HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2,
ClusterRole.STANDBY, 11L);
+
+ HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2);
+ HighAvailabilityGroup.URLS.put(info, new HashSet<>());
+ try {
+ HighAvailabilityGroup group =
+ Mockito.spy(new HighAvailabilityGroup(info, new Properties(),
recordV10, State.READY));
+
Mockito.doReturn(recordV11).when(group).getClusterRoleRecordFromEndpoint();
+
+ long transitionCountBefore =
+
GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_COUNT.getMetric().getValue();
+ long durationSamplesBefore =
+
GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_DURATION_MS.getMetric().getNumberOfSamples();
+ long failoverCountBefore =
+ GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue();
+
+ assertTrue("A strictly newer version must still be applied",
+ group.refreshClusterRoleRecord(true));
+ assertSame("The newer-version record must be adopted", recordV11,
group.getRoleRecord());
+
+ assertEquals("A pure version bump must not increment
CRR_TRANSITION_COUNT",
+ transitionCountBefore,
+
GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_COUNT.getMetric().getValue());
+ assertEquals("A pure version bump must not record a
CRR_TRANSITION_DURATION_MS sample",
+ durationSamplesBefore,
+
GlobalClientMetrics.GLOBAL_HA_CRR_TRANSITION_DURATION_MS.getMetric().getNumberOfSamples());
+ assertEquals("A pure version bump is not a failover",
failoverCountBefore,
+ GlobalClientMetrics.GLOBAL_HA_FAILOVER_COUNT.getMetric().getValue());
+
+ // Nothing was emitted for this group, so if a per-group source exists
its transition counter
+ // must still be zero.
+ HAGroupClientMetricsSource source =
HAGroupMetricsManager.getIfPresent(haGroupName);
+ if (source != null) {
+ assertEquals("Per-group CRR_TRANSITION_COUNT must stay zero on a
version bump", 0L,
+ source.getCounterValue(MetricType.CRR_TRANSITION_COUNT));
+ }
+ } finally {
+ HighAvailabilityGroup.URLS.remove(info);
+ HAGroupMetricsManager.remove(haGroupName);
+ }
+ }
+
+ /**
+ * A failed {@code connectActive} (no active cluster in the record) must
increment
+ * {@code HA_FAILOVER_CONNECTION_FAILED_COUNTER} on its single SQLException
throw funnel.
+ */
+ @Test
+ public void testConnectActiveFailureIncrementsFailedCounter() {
+ String haGroupName = "testConnectActiveFailureIncrementsFailedCounter";
+ String url1 = "host1\\:60010";
+ String url2 = "host2\\:60010";
+ // Both STANDBY → no active URL → connectActive takes the
HA_NO_ACTIVE_CLUSTER throw path.
+ ClusterRoleRecord bothStandby = new ClusterRoleRecord(haGroupName,
+ HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.STANDBY, url2,
ClusterRole.STANDBY, 10L);
+ HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2);
+ HighAvailabilityGroup group =
+ new HighAvailabilityGroup(info, new Properties(), bothStandby,
State.READY);
+
+ long failedBefore =
+
GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.getMetric().getValue();
+ try {
+ group.connectActive(new Properties(), new HAURLInfo(haGroupName));
+ fail("connectActive must throw when the HA group has no active cluster");
+ } catch (SQLException e) {
+
assertEquals(SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION.getErrorCode(),
e.getErrorCode());
+ }
+ assertEquals("A failed connectActive must increment the failed counter",
failedBefore + 1,
+
GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.getMetric().getValue());
+ // Per-group mirror: unique group name → its FAILED counter reflects
exactly this one failure.
+ HAGroupClientMetricsSource source =
HAGroupMetricsManager.getIfPresent(haGroupName);
+ assertNotNull("A failed connectActive must register the per-group metrics
source", source);
+ assertEquals("Per-group HA_FAILOVER_CONNECTION_FAILED_COUNTER must count
the failed connect",
+ 1L,
source.getCounterValue(MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER));
+ HAGroupMetricsManager.remove(haGroupName);
+ }
+
+ /**
+ * A successful {@code connectActive} must NOT increment
+ * {@code HA_FAILOVER_CONNECTION_FAILED_COUNTER}. Guards against the counter
being placed on a
+ * path that also runs on success (a non-vacuous negative assertion).
+ */
+ @Test
+ public void testConnectActiveSuccessLeavesFailedCounterUnchanged() throws
Exception {
+ String haGroupName =
"testConnectActiveSuccessLeavesFailedCounterUnchanged";
+ String url1 = "host1\\:60010";
+ String url2 = "host2\\:60010";
+ ClusterRoleRecord aActiveBStandby = new ClusterRoleRecord(haGroupName,
+ HighAvailabilityPolicy.FAILOVER, url1, ClusterRole.ACTIVE, url2,
ClusterRole.STANDBY, 10L);
+ HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2);
+ HighAvailabilityGroup group =
+ Mockito.spy(new HighAvailabilityGroup(info, new Properties(),
aActiveBStandby, State.READY));
+
+ PhoenixConnection conn = Mockito.mock(PhoenixConnection.class);
+
Mockito.doReturn(conn).when(group).connectToOneCluster(Mockito.any(String.class),
+ Mockito.any(Properties.class), Mockito.any(HAURLInfo.class));
+ Mockito.doReturn(true).when(group).isActive(conn);
+
+ long failedBefore =
+
GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.getMetric().getValue();
+ assertSame("connectActive must return the established connection", conn,
+ group.connectActive(new Properties(), new HAURLInfo(haGroupName)));
+ assertEquals("A successful connectActive must not increment the failed
counter", failedBefore,
+
GlobalClientMetrics.GLOBAL_HA_FAILOVER_CONNECTION_FAILED_COUNTER.getMetric().getValue());
+ // Per-group negative check: a successful connect emits no FAILED for this
group. Either the
+ // source was never created (no emission touched it) or its FAILED counter
is still zero.
+ HAGroupClientMetricsSource source =
HAGroupMetricsManager.getIfPresent(haGroupName);
+ assertTrue("A successful connectActive must not emit a per-group failed
count", source == null
+ ||
source.getCounterValue(MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER) == 0L);
+ HAGroupMetricsManager.remove(haGroupName);
+ }
+
+ /**
+ * The per-group metrics source lifecycle is bound to the HA group: {@code
init} pre-registers it
+ * (so the {@code ha_group}-tagged series exists as soon as the group is
ready) and {@code close}
+ * detaches it (freeing the metrics2 source name for a later re-create).
+ */
+ @Test
+ public void testInitRegistersPerGroupSourceAndCloseRemovesIt() throws
Exception {
+ String haGroupName = "testInitRegistersPerGroupSourceAndCloseRemovesIt";
+ String url1 = "host1\\:60010";
+ String url2 = "host2\\:60010";
+ ClusterRoleRecord record = new ClusterRoleRecord(haGroupName,
HighAvailabilityPolicy.FAILOVER,
+ url1, ClusterRole.ACTIVE, url2, ClusterRole.STANDBY, 10L);
+ HAGroupInfo info = new HAGroupInfo(haGroupName, url1, url2);
+ HighAvailabilityGroup group =
+ Mockito.spy(new HighAvailabilityGroup(info, new Properties(), record,
State.UNINITIALIZED));
+ Mockito.doReturn(record).when(group).getClusterRoleRecordFromEndpoint();
+ try {
+ assertNull("No per-group source should exist before init",
+ HAGroupMetricsManager.getIfPresent(haGroupName));
+ group.init();
+ assertNotNull("init must pre-register the per-group metrics source",
+ HAGroupMetricsManager.getIfPresent(haGroupName));
+ group.close();
+ assertNull("close must detach the per-group metrics source",
+ HAGroupMetricsManager.getIfPresent(haGroupName));
+ } finally {
+ HAGroupMetricsManager.remove(haGroupName);
+ }
+ }
}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSourceTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSourceTest.java
new file mode 100644
index 0000000000..552788577d
--- /dev/null
+++
b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupClientMetricsSourceTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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.monitoring;
+
+import static org.apache.phoenix.metrics.MetricConstants.HA_GROUP_TAG_NAME;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import javax.management.ObjectName;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.junit.After;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link HAGroupClientMetricsSource}: the per-HA-group
metrics2 source stamps a
+ * {@code ha_group} tag carrying the group name, keeps per-group counters, and
detaches cleanly on
+ * {@link HAGroupClientMetricsSource#unregister()}.
+ * <p>
+ * Constructing a source registers it with the process-global {@link
DefaultMetricsSystem}, so each
+ * test uses a unique group name and {@link #tearDown()} unregisters every
source it created.
+ */
+public class HAGroupClientMetricsSourceTest {
+
+ private final List<HAGroupClientMetricsSource> created = new ArrayList<>();
+
+ private HAGroupClientMetricsSource newSource(String group) {
+ HAGroupClientMetricsSource source = new HAGroupClientMetricsSource(group);
+ created.add(source);
+ return source;
+ }
+
+ @After
+ public void tearDown() {
+ for (HAGroupClientMetricsSource source : created) {
+ source.unregister();
+ }
+ created.clear();
+ }
+
+ @Test
+ public void testTagCarriesUnquotedGroupName() {
+ String group = "srcTag";
+ HAGroupClientMetricsSource source = newSource(group);
+ MetricsTag tag = source.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME);
+ assertNotNull("the ha_group tag must be present on the source", tag);
+ assertEquals("ha_group tag must carry the unquoted group name", group,
tag.value());
+ }
+
+ @Test
+ public void testJmxContextIsQuotedPerGroup() {
+ String group = "group,one=weird";
+ HAGroupClientMetricsSource source = newSource(group);
+ assertEquals(group,
source.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value());
+ assertTrue(source.getMetricsJmxContext().endsWith(",haGroup=" +
ObjectName.quote(group)));
+ }
+
+ @Test
+ public void testIncrementAndUpdateArePerCounter() {
+ HAGroupClientMetricsSource source = newSource("srcCounters");
+ source.increment(MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER);
+ source.increment(MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER);
+ // The duration metric accumulates via update(); other counters are
untouched by it.
+ source.update(MetricType.CRR_TRANSITION_DURATION_MS, 40L);
+ source.update(MetricType.CRR_TRANSITION_DURATION_MS, 60L);
+
+ assertEquals(2L,
source.getCounterValue(MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER));
+ assertEquals(100L,
source.getCounterValue(MetricType.CRR_TRANSITION_DURATION_MS));
+ assertEquals("an untouched counter stays at zero", 0L,
+ source.getCounterValue(MetricType.HA_STALE_CRR_DETECTED_COUNT));
+ }
+
+ @Test
+ public void testUnknownMetricTypeIsIgnored() {
+ HAGroupClientMetricsSource source = newSource("srcUnknown");
+ // A type that is not in METRIC_TYPES has no backing counter:
increment/update are no-ops and
+ // getCounterValue reports -1 to distinguish "absent" from a
present-but-zero counter.
+ source.increment(MetricType.MUTATION_BATCH_SIZE);
+ source.update(MetricType.MUTATION_BATCH_SIZE, 5L);
+ assertEquals(-1L, source.getCounterValue(MetricType.MUTATION_BATCH_SIZE));
+ }
+
+ @Test
+ public void testEachGroupIsADistinctRegisteredSource() {
+ HAGroupClientMetricsSource a = newSource("srcDistinctA");
+ HAGroupClientMetricsSource b = newSource("srcDistinctB");
+ assertNotEquals("distinct groups must get distinct JMX contexts",
a.getMetricsJmxContext(),
+ b.getMetricsJmxContext());
+
assertNotNull(DefaultMetricsSystem.instance().getSource(a.getMetricsJmxContext()));
+
assertNotNull(DefaultMetricsSystem.instance().getSource(b.getMetricsJmxContext()));
+ }
+
+ @Test
+ public void testUnregisterFreesTheSourceNameForReuse() {
+ String group = "srcReuse";
+ HAGroupClientMetricsSource first = new HAGroupClientMetricsSource(group);
+ String jmxContext = first.getMetricsJmxContext();
+ assertNotNull(DefaultMetricsSystem.instance().getSource(jmxContext));
+
+ first.unregister();
+ assertNull("unregister must free the source name in DefaultMetricsSystem",
+ DefaultMetricsSystem.instance().getSource(jmxContext));
+
+ // The same group name must be registrable again (would throw "source
already exists" if the
+ // prior source were still attached).
+ HAGroupClientMetricsSource second = newSource(group);
+
assertNotNull(DefaultMetricsSystem.instance().getSource(second.getMetricsJmxContext()));
+ assertEquals(jmxContext, second.getMetricsJmxContext());
+ }
+}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupMetricsManagerTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupMetricsManagerTest.java
new file mode 100644
index 0000000000..3d4ed86b04
--- /dev/null
+++
b/phoenix-core/src/test/java/org/apache/phoenix/monitoring/HAGroupMetricsManagerTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.monitoring;
+
+import static org.apache.phoenix.metrics.MetricConstants.HA_GROUP_TAG_NAME;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link HAGroupMetricsManager}: the per-HA-group source
registry. Covers per-group
+ * isolation (two groups never cross-count), that each group registers as its
own
+ * {@code ha_group}-tagged metrics2 source, and that {@link
HAGroupMetricsManager#remove(String)}
+ * detaches the source so the same group can be re-created fresh.
+ * <p>
+ * These tests mutate process-global state ({@link DefaultMetricsSystem} and
the manager's static
+ * map), so each group name is unique per test and {@link #tearDown()} removes
every group this
+ * class created to avoid leaking sources into sibling tests.
+ */
+public class HAGroupMetricsManagerTest {
+
+ private final List<String> createdGroups = new ArrayList<>();
+
+ @BeforeClass
+ public static void assumeGlobalMetricsEnabled() {
+ // The manager creates sources only when global metrics are enabled (the
default). If a
+ // surrounding config disabled them, getOrCreate would return null and
these assertions would
+ // not apply.
+ assertTrue("these tests assume the default global-metrics-enabled=true",
+ QueryServicesOptions.withDefaults().isGlobalMetricsEnabled());
+ }
+
+ @Before
+ public void setUp() {
+ createdGroups.clear();
+ }
+
+ @After
+ public void tearDown() {
+ for (String group : createdGroups) {
+ HAGroupMetricsManager.remove(group);
+ }
+ }
+
+ private String track(String group) {
+ createdGroups.add(group);
+ return group;
+ }
+
+ @Test
+ public void testGetOrCreateIsIdempotentAndRegistersSource() {
+ String group = track("mgrCreate");
+ HAGroupClientMetricsSource first =
HAGroupMetricsManager.getOrCreate(group);
+ HAGroupClientMetricsSource second =
HAGroupMetricsManager.getOrCreate(group);
+ assertNotNull(first);
+ assertSame("getOrCreate must be idempotent for a group name", first,
second);
+ assertNotNull("the group's source must be registered with
DefaultMetricsSystem",
+ DefaultMetricsSystem.instance().getSource(first.getMetricsJmxContext()));
+ assertEquals("the source must carry the ha_group tag", group,
+ first.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value());
+ }
+
+ @Test
+ public void testNullAndEmptyGroupNameIsNoOp() {
+ assertNull(HAGroupMetricsManager.getOrCreate(null));
+ assertNull(HAGroupMetricsManager.getOrCreate(""));
+ // remove must tolerate null/empty without throwing.
+ HAGroupMetricsManager.remove(null);
+ HAGroupMetricsManager.remove("");
+ }
+
+ @Test
+ public void testTwoGroupsDoNotCrossCount() {
+ String groupA = track("mgrIsoA");
+ String groupB = track("mgrIsoB");
+ MetricType type = MetricType.HA_FAILOVER_CONNECTION_FAILED_COUNTER;
+
+ HAGroupMetricsManager.increment(groupA, type);
+ HAGroupMetricsManager.increment(groupA, type);
+ HAGroupMetricsManager.increment(groupB, type);
+
+ assertEquals(2L,
HAGroupMetricsManager.getIfPresent(groupA).getCounterValue(type));
+ assertEquals("each HA group must maintain independent counters", 1L,
+ HAGroupMetricsManager.getIfPresent(groupB).getCounterValue(type));
+ }
+
+ @Test
+ public void testEachGroupGetsADistinctTaggedSource() {
+ String groupA = track("mgrDistinctA");
+ String groupB = track("mgrDistinctB");
+ HAGroupClientMetricsSource regA =
HAGroupMetricsManager.getOrCreate(groupA);
+ HAGroupClientMetricsSource regB =
HAGroupMetricsManager.getOrCreate(groupB);
+
+ assertNotEquals("distinct groups must map to distinct JMX contexts",
+ regA.getMetricsJmxContext(), regB.getMetricsJmxContext());
+ assertEquals(groupA,
regA.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value());
+ assertEquals(groupB,
regB.getMetricsRegistry().getTag(HA_GROUP_TAG_NAME).value());
+ }
+
+ @Test
+ public void testUpdateAccumulatesPerGroupDuration() {
+ String group = track("mgrDuration");
+ MetricType type = MetricType.CRR_TRANSITION_DURATION_MS;
+ HAGroupMetricsManager.update(group, type, 40L);
+ HAGroupMetricsManager.update(group, type, 60L);
+ assertEquals(100L,
HAGroupMetricsManager.getIfPresent(group).getCounterValue(type));
+ }
+
+ @Test
+ public void testRemoveDetachesSource() {
+ String group = "mgrRemove"; // not tracked: this test removes it itself
+ HAGroupClientMetricsSource source =
HAGroupMetricsManager.getOrCreate(group);
+ String jmxContext = source.getMetricsJmxContext();
+ assertNotNull(HAGroupMetricsManager.getIfPresent(group));
+ assertNotNull(DefaultMetricsSystem.instance().getSource(jmxContext));
+
+ HAGroupMetricsManager.remove(group);
+
+ assertNull("holder must be gone after remove",
HAGroupMetricsManager.getIfPresent(group));
+ assertNull("source must be detached from DefaultMetricsSystem after
remove",
+ DefaultMetricsSystem.instance().getSource(jmxContext));
+ }
+
+ @Test
+ public void testGetOrCreateAfterRemoveRebuildsFresh() {
+ String group = track("mgrRebuild");
+ MetricType type = MetricType.HA_STALE_CRR_DETECTED_COUNT;
+ HAGroupMetricsManager.increment(group, type);
+ assertEquals(1L,
HAGroupMetricsManager.getIfPresent(group).getCounterValue(type));
+
+ HAGroupMetricsManager.remove(group);
+ // A re-create after remove must yield a fresh source with zeroed counters.
+ HAGroupClientMetricsSource rebuilt =
HAGroupMetricsManager.getOrCreate(group);
+ assertNotNull(rebuilt);
+ assertEquals("re-created group must start from zero", 0L,
rebuilt.getCounterValue(type));
+ }
+}
