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 5916ea86c4d..387c85e40b1 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,6 @@ 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_MUTATION_BLOCKED_COUNT; import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HA_STALE_CRR_DETECTED_COUNT; @@ -175,63 +174,55 @@ void failover(long timeoutMs) throws SQLException { 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()); } /** 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 05ce2787644..db4408fdbf2 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 @@ 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_FAILOVER_DURATION_MS; import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_CLIENT_CONNECTION_CACHE_MAX_DURATION; import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR; @@ -700,6 +702,10 @@ PhoenixConnection connectActive(final Properties properties, final HAURLInfo hau } 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(); 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(); @@ -1213,53 +1219,65 @@ public boolean refreshClusterRoleRecord(boolean forceRefresh) throws SQLExceptio long maxTransitionTimeMs = StringUtils.isNotEmpty(transitionTimeoutProp) ? Long.parseLong(transitionTimeoutProp) : PHOENIX_HA_TRANSITION_TIMEOUT_MS_DEFAULT; - boolean transitionSucceeded = false; + // Time the cluster-transition dispatch on this CRR-write path, which is where autonomous + // failovers are actually driven. The duration is recorded on every exit (success, timeout, + // policy failure, or interrupt) via the finally block below so it tracks time spent handling + // detected CRR transitions rather than the connection-level failover() path, which is never + // auto-invoked under the default ExplicitFailoverPolicy. + final long transitionStartMs = System.currentTimeMillis(); try { - future.get(maxTransitionTimeMs, TimeUnit.MILLISECONDS); - transitionSucceeded = true; - } catch (InterruptedException ie) { - LOG.error("Got interrupted when transiting cluster roles for HA group {}", info, ie); - future.cancel(true); - Thread.currentThread().interrupt(); - return false; - } catch (ExecutionException | TimeoutException e) { - LOG.error("HA group {} failed to transit cluster roles per policy {} to new " + "record {}", - info, roleRecord.getPolicy(), newRoleRecord, e); - // Rethrow the Role transitions not allowed exceptions - if (e.getCause() != null && e.getCause().getCause() != null) { - if ( - e.getCause().getCause() instanceof SQLException - && ((SQLException) e.getCause().getCause()).getErrorCode() - == SQLExceptionCode.HA_ROLE_TRANSITION_NOT_ALLOWED.getErrorCode() - ) { - state = State.READY; - throw (SQLException) e.getCause().getCause(); + boolean transitionSucceeded = false; + try { + future.get(maxTransitionTimeMs, TimeUnit.MILLISECONDS); + transitionSucceeded = true; + } catch (InterruptedException ie) { + LOG.error("Got interrupted when transiting cluster roles for HA group {}", info, ie); + future.cancel(true); + Thread.currentThread().interrupt(); + return false; + } catch (ExecutionException | TimeoutException e) { + LOG.error( + "HA group {} failed to transit cluster roles per policy {} to new " + "record {}", info, + roleRecord.getPolicy(), newRoleRecord, e); + // Rethrow the Role transitions not allowed exceptions + if (e.getCause() != null && e.getCause().getCause() != null) { + if ( + e.getCause().getCause() instanceof SQLException + && ((SQLException) e.getCause().getCause()).getErrorCode() + == SQLExceptionCode.HA_ROLE_TRANSITION_NOT_ALLOWED.getErrorCode() + ) { + state = State.READY; + throw (SQLException) e.getCause().getCause(); + } } + // Calling back HA policy function for cluster switch is conducted with best effort. + // HA group continues transition when its HA policy fails to deal with context switch + // (e.g. to close existing connections) + // The goal here is to gain higher availability even though existing resources against + // previous ACTIVE cluster may have not been closed cleanly. } - // Calling back HA policy function for cluster switch is conducted with best effort. - // HA group continues transition when its HA policy fails to deal with context switch - // (e.g. to close existing connections) - // The goal here is to gain higher availability even though existing resources against - // previous ACTIVE cluster may have not been closed cleanly. - } - // Count the transition as a failover only when the policy-side transition actually - // succeeded AND an active cluster is established or moves between peers. Operator-driven - // transitions to a no-active state (both clusters STANDBY) are not counted as failovers; - // recovery from no-active back to having an ACTIVE peer is counted. Transitions where - // future.get() failed (ExecutionException/TimeoutException) are best-effort fall-through - // per the comment above, but they are NOT counted as successful failovers. Gate decision - // factored into the package-private static {@link #shouldCountFailover} so it can be - // unit-tested directly without driving a full mini-cluster transition. - if (shouldCountFailover(transitionSucceeded, oldRecord, newRoleRecord)) { - GLOBAL_HA_FAILOVER_COUNT.increment(); + // Count the transition as a failover only when the policy-side transition actually + // succeeded AND an active cluster is established or moves between peers. Operator-driven + // transitions to a no-active state (both clusters STANDBY) are not counted as failovers; + // recovery from no-active back to having an ACTIVE peer is counted. Transitions where + // future.get() failed (ExecutionException/TimeoutException) are best-effort fall-through + // per the comment above, but they are NOT counted as successful failovers. Gate decision + // factored into the package-private static {@link #shouldCountFailover} so it can be + // unit-tested directly without driving a full mini-cluster transition. + if (shouldCountFailover(transitionSucceeded, oldRecord, newRoleRecord)) { + GLOBAL_HA_FAILOVER_COUNT.increment(); + } + // Update the role record and the last refresh time + roleRecord = newRoleRecord; + lastClusterRoleRecordRefreshTime = System.currentTimeMillis(); + state = State.READY; + LOG.info("HA group {} is in {} state, Old: {}, new: {}", info, state, oldRecord, + roleRecord); + LOG.debug("HA group is ready: {}", this); + return true; + } finally { + GLOBAL_HA_FAILOVER_DURATION_MS.update(System.currentTimeMillis() - transitionStartMs); } - // Update the role record and the last refresh time - roleRecord = newRoleRecord; - lastClusterRoleRecordRefreshTime = System.currentTimeMillis(); - state = State.READY; - LOG.info("HA group {} is in {} state, Old: {}, new: {}", info, state, oldRecord, roleRecord); - LOG.debug("HA group is ready: {}", this); - return true; } finally { writeLock.unlock(); } 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 bbd23bcbc52..816a142cfa9 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 @@ -37,6 +37,7 @@ import static org.apache.phoenix.monitoring.MetricType.COUNT_SCANNED_REGIONS; 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_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; @@ -173,6 +174,7 @@ public enum GlobalClientMetrics { 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_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/MetricType.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java index 22a9ecbdcf3..ecb3d314471 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 @@ -359,8 +359,12 @@ public enum MetricType { + "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)", + "Total time in milliseconds spent dispatching cluster-role transitions, recorded at the CRR " + + "write site (refreshClusterRoleRecord) on every transition exit", + 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)", LogLevel.DEBUG, PLong.INSTANCE), HA_MUTATION_BLOCKED_COUNT("hambc", "Counter for MutationBlockedIOException surfaces caught by wrapActionDuringFailover", 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 5bf956fa88e..f323f183576 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 @@ -27,6 +27,7 @@ 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; @@ -618,4 +619,109 @@ public void testEndpointRestoresInterruptStatusOnFallback() throws Exception { 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 HA_FAILOVER_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_FAILOVER_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 HA_FAILOVER_DURATION_MS sample on the CRR-write " + "path", + durationSamplesBefore + 1, + GlobalClientMetrics.GLOBAL_HA_FAILOVER_DURATION_MS.getMetric().getNumberOfSamples()); + } finally { + HighAvailabilityGroup.URLS.remove(info); + } + } + + /** + * 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()); + } + + /** + * 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()); + } }