From 52a1c657f2cd5d26a927fbcd6614f1ec3891c3c4 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Sat, 22 Aug 2026 16:40:34 +0530 Subject: [PATCH 1/2] PHOENIX-7992 Add replay-side performance metrics for HA replication Adds six standby replay-side performance metrics, each inheriting the per-haGroup dimension from its metric source. ReplicationLogDiscoveryReplay: - endToEndReplayLagMs (histogram): round-eligible -> replay finished - pickupLagMs (histogram): round-eligible -> file claimed (rename into the in-progress directory) ReplicationLogDiscovery (shared base): - roundsExceedingRoundTime (counter): rounds whose new-file processing duration exceeded roundTimeMills ReplicationLogProcessor: - mutationsReplayedCount (counter): replay throughput across all files - mutationsPerFile (histogram): per-file mutation count, recorded only for files that replayed at least one mutation so rotation-only files do not skew the distribution toward zero ReplicationLogTracker: - markFileInProgressCollisionCount (counter): claim renames lost to a competing process, a strict subset of the request count The lag metrics use the round-eligible instant (round end + waiting buffer) as the zero reference, so they exclude the fixed built-in wait rather than measuring raw file age. Tests: metric-source plumbing for all six metrics, plus recording-site tests for the collision metric (rename-loses vs rename-wins) and the slow-round metric (fast round vs slow round). --- .../replication/ReplicationLogDiscovery.java | 33 ++--- .../replication/ReplicationLogTracker.java | 3 + .../MetricsReplicationLogDiscovery.java | 10 ++ .../MetricsReplicationLogDiscoveryImpl.java | 10 +- .../MetricsReplicationLogDiscoveryReplay.java | 23 ++++ ...ricsReplicationLogDiscoveryReplayImpl.java | 28 +++++ .../MetricsReplicationLogProcessor.java | 20 +++ .../MetricsReplicationLogProcessorImpl.java | 19 ++- .../metrics/MetricsReplicationLogTracker.java | 10 ++ .../MetricsReplicationLogTrackerImpl.java | 11 +- .../ReplicationLogDiscoveryMetricValues.java | 8 +- .../ReplicationLogProcessorMetricValues.java | 14 ++- .../ReplicationLogTrackerMetricValues.java | 9 +- .../reader/ReplicationLogDiscoveryReplay.java | 30 ++++- .../reader/ReplicationLogProcessor.java | 6 + .../ReplicationLogDiscoveryTest.java | 57 ++++++++- .../ReplicationLogTrackerTest.java | 53 ++++++++ .../ReplicationReplayPerfMetricsTest.java | 114 ++++++++++++++++++ 18 files changed, 435 insertions(+), 23 deletions(-) create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java index 21e1117b87d..81091670b74 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java @@ -145,8 +145,9 @@ public abstract class ReplicationLogDiscovery { @GuardedBy("this") protected long lastAlignedTargetMillis = Long.MIN_VALUE; /** - * One-shot guard so a misconfigured {@link #REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY} logs its - * fall-back WARN once rather than every scheduling cycle (the epsilon is read live per cycle). + * One-shot guard so a misconfigured {@link #REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY} logs + * its fall-back WARN once rather than every scheduling cycle (the epsilon is read live per + * cycle). */ private final AtomicBoolean warnedInvalidEpsilon = new AtomicBoolean(false); @@ -279,11 +280,11 @@ protected void scheduleNextReplay() { * Runs one replay pass and, unless the service has been stopped, schedules the next aligned pass. * A recoverable {@link Exception} from {@link #replay()} is logged and the chain continues, so a * single failed round does not break it. A fatal {@link Error} (OOM, stack overflow, linkage) is - * logged, tears the chain down (marking the service not-running so the supervisor can rebuild it), - * and is rethrown -- never rescheduled onto a potentially corrupted JVM. - * The reschedule is guarded by the same lock stop() uses; if stop() shut the scheduler down - * first, {@link #isRunning} is false and we do not reschedule (and a concurrent shutdown that - * rejects the submission is caught and treated as "stop the chain"). + * logged, tears the chain down (marking the service not-running so the supervisor can rebuild + * it), and is rethrown -- never rescheduled onto a potentially corrupted JVM. The reschedule is + * guarded by the same lock stop() uses; if stop() shut the scheduler down first, + * {@link #isRunning} is false and we do not reschedule (and a concurrent shutdown that rejects + * the submission is caught and treated as "stop the chain"). * @param owner the scheduler this cycle was launched on. If a stop()->start() restart has since * swapped in a new scheduler, {@code owner} no longer equals {@link #scheduler} and * this stale cycle must not reschedule onto the new generation (which would create a @@ -445,6 +446,9 @@ protected void processNewFilesForRound(ReplicationRound replicationRound) throws LOG.info("Finished new files processing for round: {} in {}ms for haGroup: {}", replicationRound, duration, haGroupName); getMetrics().updateTimeToProcessNewFiles(duration); + if (duration > roundTimeMills) { + getMetrics().incrementRoundsExceedingRoundTime(); + } } /** @@ -696,11 +700,12 @@ public int getInProgressFileMinAgeSeconds() { /** * Returns the epsilon margin (milliseconds) added to the aligned scheduler wake instant. Guards * against a misconfigured value: a non-numeric string, a negative value (which would move wakes - * before eligibility and reintroduce missed rounds), or a value {@code >= roundTimeMills} - * (which wraps through {@link Math#floorMod} and no longer represents the documented "epsilon - * after the boundary") all fall back to {@link #DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS} with a - * one-shot WARN. Riding over the misconfiguration keeps replay polling on a safe default rather - * than wedging the group, while the WARN makes the bad config visible. + * before eligibility and reintroduce missed rounds), or a value + * {@code >= roundTimeMills} (which wraps through {@link Math#floorMod} and no longer represents + * the documented "epsilon after the boundary") all fall back to + * {@link #DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS} with a one-shot WARN. Riding over the + * misconfiguration keeps replay polling on a safe default rather than wedging the group, while + * the WARN makes the bad config visible. * @return the epsilon margin in milliseconds, always within {@code [0, roundTimeMills)}. */ public long getAlignedDelayEpsilonMillis() { @@ -711,8 +716,8 @@ public long getAlignedDelayEpsilonMillis() { } catch (NumberFormatException e) { // Hadoop's getLong throws (rather than returning the default) when the key is present but // not parseable as a number. - warnInvalidEpsilon("non-numeric value \"" - + conf.get(REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY) + "\""); + warnInvalidEpsilon( + "non-numeric value \"" + conf.get(REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY) + "\""); return DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS; } if (epsilon < 0 || epsilon >= roundTimeMills) { diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java index 00bfa4ed685..90a1c58db96 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java @@ -384,6 +384,9 @@ protected Optional markInProgress(final Path file) { return Optional.of(newPath); } else { LOG.warn("Failed to rename file for in-progress marking: {}", file); + // Rename lost to another process claiming the same file: a collision, a strict subset of + // the request count incremented in the finally block below. + getMetrics().incrementMarkFileInProgressCollisionCount(); return Optional.empty(); } } catch (IOException e) { diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscovery.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscovery.java index 73371bb55b7..4ce66b743e2 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscovery.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscovery.java @@ -37,6 +37,9 @@ public interface MetricsReplicationLogDiscovery extends BaseSource { String TIME_TO_PROCESS_IN_PROGRESS_FILES = "timeToProcessInProgressFilesMs"; String TIME_TO_PROCESS_IN_PROGRESS_FILES_DESC = "Histogram of time taken to process in progress files in milliseconds"; + String ROUNDS_EXCEEDING_ROUND_TIME = "roundsExceedingRoundTime"; + String ROUNDS_EXCEEDING_ROUND_TIME_DESC = + "Number of rounds whose new-file processing time exceeded the round time"; /** * Increments the counter for rounds processed. This counter tracks the number of rounds processed @@ -62,6 +65,13 @@ public interface MetricsReplicationLogDiscovery extends BaseSource { */ void updateTimeToProcessInProgressFiles(long timeMs); + /** + * Increments the counter for rounds whose new-file processing time exceeded the round time. A + * rising rate of such rounds signals that replay is falling behind the cadence at which new + * rounds become eligible. + */ + void incrementRoundsExceedingRoundTime(); + /** * Unregister this metrics source. */ diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryImpl.java index 35183d92b31..ccd9a1bd118 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryImpl.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryImpl.java @@ -31,6 +31,7 @@ public class MetricsReplicationLogDiscoveryImpl extends BaseSourceImpl protected final MutableFastCounter numInProgressDirectoryProcessed; protected final MutableHistogram timeToProcessNewFiles; protected final MutableHistogram timeToProcessInProgressFiles; + protected final MutableFastCounter roundsExceedingRoundTime; public MetricsReplicationLogDiscoveryImpl(String metricsName, String metricsDescription, String metricsContext, String metricsJmxContext) { @@ -43,6 +44,8 @@ public MetricsReplicationLogDiscoveryImpl(String metricsName, String metricsDesc getMetricsRegistry().newHistogram(TIME_TO_PROCESS_NEW_FILES, TIME_TO_PROCESS_NEW_FILES_DESC); timeToProcessInProgressFiles = getMetricsRegistry() .newHistogram(TIME_TO_PROCESS_IN_PROGRESS_FILES, TIME_TO_PROCESS_IN_PROGRESS_FILES_DESC); + roundsExceedingRoundTime = getMetricsRegistry().newCounter(ROUNDS_EXCEEDING_ROUND_TIME, + ROUNDS_EXCEEDING_ROUND_TIME_DESC, 0L); } @Override @@ -65,6 +68,11 @@ public void updateTimeToProcessInProgressFiles(long timeMs) { timeToProcessInProgressFiles.add(timeMs); } + @Override + public void incrementRoundsExceedingRoundTime() { + roundsExceedingRoundTime.incr(); + } + @Override public void close() { // Unregister this metrics source @@ -75,7 +83,7 @@ public void close() { public ReplicationLogDiscoveryMetricValues getCurrentMetricValues() { return new ReplicationLogDiscoveryMetricValues(numRoundsProcessed.value(), numInProgressDirectoryProcessed.value(), timeToProcessNewFiles.getMax(), - timeToProcessInProgressFiles.getMax()); + timeToProcessInProgressFiles.getMax(), roundsExceedingRoundTime.value()); } @Override diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplay.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplay.java index f78b7621466..185fce977aa 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplay.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplay.java @@ -26,6 +26,14 @@ public interface MetricsReplicationLogDiscoveryReplay extends MetricsReplication String CONSISTENCY_POINT = "consistencyPoint"; String CONSISTENCY_POINT_DESC = "Consistency point timestamp in milliseconds for the HA Group during replay"; + String END_TO_END_REPLAY_LAG = "endToEndReplayLagMs"; + String END_TO_END_REPLAY_LAG_DESC = + "Histogram of end-to-end replay lag, from when a file's round became eligible for processing " + + "to when the file finished replaying, in milliseconds"; + String PICKUP_LAG = "pickupLagMs"; + String PICKUP_LAG_DESC = + "Histogram of pickup lag, from when a file's round became eligible for processing to when the " + + "file was claimed (renamed into the in-progress directory), in milliseconds"; /** * Updates the consistency point metric. The consistency point represents the timestamp up to @@ -34,4 +42,19 @@ public interface MetricsReplicationLogDiscoveryReplay extends MetricsReplication * @param consistencyPointMs The consistency point timestamp in milliseconds */ void updateConsistencyPoint(long consistencyPointMs); + + /** + * Records a sample into the end-to-end replay lag histogram: the elapsed time from when a file's + * round became eligible for processing to when the file finished replaying. + * @param lagMs The end-to-end replay lag in milliseconds + */ + void updateEndToEndReplayLag(long lagMs); + + /** + * Records a sample into the pickup lag histogram: the elapsed time from when a file's round + * became eligible for processing to when the file was claimed (renamed into the in-progress + * directory). + * @param lagMs The pickup lag in milliseconds + */ + void updatePickupLag(long lagMs); } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplayImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplayImpl.java index 16e65287d98..ee7548e53db 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplayImpl.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogDiscoveryReplayImpl.java @@ -18,6 +18,9 @@ package org.apache.phoenix.replication.metrics; import org.apache.hadoop.metrics2.lib.MutableGaugeLong; +import org.apache.hadoop.metrics2.lib.MutableHistogram; + +import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting; /** Implementation of metrics source for ReplicationLogDiscoveryReplay operations. */ public class MetricsReplicationLogDiscoveryReplayImpl extends MetricsReplicationLogDiscoveryImpl @@ -29,6 +32,8 @@ public class MetricsReplicationLogDiscoveryReplayImpl extends MetricsReplication private static final String METRICS_JMX_CONTEXT = "RegionServer,sub=" + METRICS_NAME; private final MutableGaugeLong consistencyPoint; + private final MutableHistogram endToEndReplayLag; + private final MutableHistogram pickupLag; public MetricsReplicationLogDiscoveryReplayImpl(final String haGroupName) { super(MetricsReplicationLogDiscoveryReplayImpl.METRICS_NAME, @@ -36,10 +41,33 @@ public MetricsReplicationLogDiscoveryReplayImpl(final String haGroupName) { MetricsReplicationLogDiscoveryImpl.METRICS_CONTEXT, MetricsReplicationLogDiscoveryReplayImpl.METRICS_JMX_CONTEXT + ",haGroup=" + haGroupName); consistencyPoint = getMetricsRegistry().newGauge(CONSISTENCY_POINT, CONSISTENCY_POINT_DESC, 0L); + endToEndReplayLag = + getMetricsRegistry().newHistogram(END_TO_END_REPLAY_LAG, END_TO_END_REPLAY_LAG_DESC); + pickupLag = getMetricsRegistry().newHistogram(PICKUP_LAG, PICKUP_LAG_DESC); } @Override public void updateConsistencyPoint(long consistencyPointMs) { consistencyPoint.set(consistencyPointMs); } + + @Override + public void updateEndToEndReplayLag(long lagMs) { + endToEndReplayLag.add(lagMs); + } + + @Override + public void updatePickupLag(long lagMs) { + pickupLag.add(lagMs); + } + + @VisibleForTesting + MutableHistogram getEndToEndReplayLagHistogram() { + return endToEndReplayLag; + } + + @VisibleForTesting + MutableHistogram getPickupLagHistogram() { + return pickupLag; + } } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java index 5ed7bd8e6f3..0281385bdcf 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java @@ -41,6 +41,11 @@ public interface MetricsReplicationLogProcessor extends BaseSource { String LOG_FILE_REPLAY_TIME = "logFileReplayTimeMs"; String LOG_FILE_REPLAY_TIME_DESC = "Histogram of time taken for replaying a log file in milliseconds"; + String MUTATIONS_REPLAYED_COUNT = "mutationsReplayedCount"; + String MUTATIONS_REPLAYED_COUNT_DESC = "Total number of mutations replayed across all log files"; + String MUTATIONS_PER_FILE = "mutationsPerFile"; + String MUTATIONS_PER_FILE_DESC = + "Histogram of the number of mutations replayed per log file (files with mutations only)"; /** * Increments the counter for failed mutations. This counter tracks the number of mutations that @@ -78,6 +83,21 @@ public interface MetricsReplicationLogProcessor extends BaseSource { */ void updateLogFileReplayTime(long timeMs); + /** + * Increments the total number of mutations replayed by the given amount. This counter tracks + * replay throughput across all log files. + * @param delta the number of mutations replayed for a log file + */ + void incrementMutationsReplayedCount(long delta); + + /** + * Records a sample into the per-file mutation-count histogram. Called only for files that + * replayed at least one mutation, so rotation-only (zero-mutation) files do not skew the + * distribution toward zero. + * @param mutationsCount the number of mutations replayed for a single log file + */ + void updateMutationsPerFile(long mutationsCount); + /** * Unregister this metrics source. */ diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessorImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessorImpl.java index be8f7715c96..2643eaef162 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessorImpl.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessorImpl.java @@ -33,6 +33,8 @@ public class MetricsReplicationLogProcessorImpl extends BaseSourceImpl private final MutableFastCounter logFileReplaySuccessCount; private final MutableHistogram batchReplayTime; private final MutableHistogram logFileReplayTime; + private final MutableFastCounter mutationsReplayedCount; + private final MutableHistogram mutationsPerFile; public MetricsReplicationLogProcessorImpl(final String haGroupName) { this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, @@ -54,6 +56,10 @@ public MetricsReplicationLogProcessorImpl(String metricsName, String metricsDesc batchReplayTime = getMetricsRegistry().newHistogram(BATCH_REPLAY_TIME, BATCH_REPLAY_TIME_DESC); logFileReplayTime = getMetricsRegistry().newHistogram(LOG_FILE_REPLAY_TIME, LOG_FILE_REPLAY_TIME_DESC); + mutationsReplayedCount = + getMetricsRegistry().newCounter(MUTATIONS_REPLAYED_COUNT, MUTATIONS_REPLAYED_COUNT_DESC, 0L); + mutationsPerFile = + getMetricsRegistry().newHistogram(MUTATIONS_PER_FILE, MUTATIONS_PER_FILE_DESC); } @Override @@ -86,6 +92,16 @@ public void updateLogFileReplayTime(long timeMs) { logFileReplayTime.add(timeMs); } + @Override + public void incrementMutationsReplayedCount(long delta) { + mutationsReplayedCount.incr(delta); + } + + @Override + public void updateMutationsPerFile(long mutationsCount) { + mutationsPerFile.add(mutationsCount); + } + @Override public void close() { // Unregister this metrics source @@ -96,7 +112,8 @@ public void close() { public ReplicationLogProcessorMetricValues getCurrentMetricValues() { return new ReplicationLogProcessorMetricValues(failedMutationsCount.value(), logFileReplayFailureCount.value(), logFileReplaySuccessCount.value(), - logFileReplayTime.getMax(), batchReplayTime.getMax()); + logFileReplayTime.getMax(), batchReplayTime.getMax(), mutationsReplayedCount.value(), + mutationsPerFile.getMax()); } @Override diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTracker.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTracker.java index dcc2af2a5db..47d8ff24e1a 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTracker.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTracker.java @@ -29,6 +29,9 @@ public interface MetricsReplicationLogTracker extends BaseSource { String MARK_FILE_IN_PROGRESS_REQUEST_COUNT = "markFileInProgressRequestCount"; String MARK_FILE_IN_PROGRESS_REQUEST_COUNT_DESC = "Number of requests made to mark file in progress"; + String MARK_FILE_IN_PROGRESS_COLLISION_COUNT = "markFileInProgressCollisionCount"; + String MARK_FILE_IN_PROGRESS_COLLISION_COUNT_DESC = + "Number of mark-file-in-progress requests that lost the claim rename to another process"; String MARK_FILE_COMPLETED_REQUEST_COUNT = "markFileCompletedRequestCount"; String MARK_FILE_COMPLETED_REQUEST_COUNT_DESC = "Number of requests made to mark file completed"; String MARK_FILE_FAILED_REQUEST_COUNT = "markFileFailedRequestCount"; @@ -52,6 +55,13 @@ public interface MetricsReplicationLogTracker extends BaseSource { */ void incrementMarkFileInProgressRequestCount(); + /** + * Increments the counter for mark-file-in-progress collisions. This counter tracks the number of + * claim attempts whose rename lost to another process (the decentralized, no-coordination claim + * design), a strict subset of {@link #incrementMarkFileInProgressRequestCount()}. + */ + void incrementMarkFileInProgressCollisionCount(); + /** * Increments the counter for mark file completed requests. This counter tracks the number of * requests made to mark files completed. diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTrackerImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTrackerImpl.java index b938a504d49..c503c476c80 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTrackerImpl.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTrackerImpl.java @@ -28,6 +28,7 @@ public class MetricsReplicationLogTrackerImpl extends BaseSourceImpl protected String groupMetricsContext; protected final MutableFastCounter markFileInProgressRequestCount; + protected final MutableFastCounter markFileInProgressCollisionCount; protected final MutableFastCounter markFileCompletedRequestCount; protected final MutableFastCounter markFileFailedRequestCount; protected final MutableFastCounter markFileCompletedRequestFailedCount; @@ -40,6 +41,8 @@ public MetricsReplicationLogTrackerImpl(String metricsName, String metricsDescri super(metricsName, metricsDescription, metricsContext, metricsJmxContext); markFileInProgressRequestCount = getMetricsRegistry().newCounter( MARK_FILE_IN_PROGRESS_REQUEST_COUNT, MARK_FILE_IN_PROGRESS_REQUEST_COUNT_DESC, 0L); + markFileInProgressCollisionCount = getMetricsRegistry().newCounter( + MARK_FILE_IN_PROGRESS_COLLISION_COUNT, MARK_FILE_IN_PROGRESS_COLLISION_COUNT_DESC, 0L); markFileCompletedRequestCount = getMetricsRegistry() .newCounter(MARK_FILE_COMPLETED_REQUEST_COUNT, MARK_FILE_COMPLETED_REQUEST_COUNT_DESC, 0L); markFileFailedRequestCount = getMetricsRegistry().newCounter(MARK_FILE_FAILED_REQUEST_COUNT, @@ -59,6 +62,11 @@ public void incrementMarkFileInProgressRequestCount() { markFileInProgressRequestCount.incr(); } + @Override + public void incrementMarkFileInProgressCollisionCount() { + markFileInProgressCollisionCount.incr(); + } + @Override public void incrementMarkFileCompletedRequestCount() { markFileCompletedRequestCount.incr(); @@ -100,7 +108,8 @@ public ReplicationLogTrackerMetricValues getCurrentMetricValues() { return new ReplicationLogTrackerMetricValues(markFileInProgressRequestCount.value(), markFileCompletedRequestCount.value(), markFileFailedRequestCount.value(), markFileCompletedRequestFailedCount.value(), markFileInProgressTime.getMax(), - markFileCompletedTime.getMax(), markFileFailedTime.getMax()); + markFileCompletedTime.getMax(), markFileFailedTime.getMax(), + markFileInProgressCollisionCount.value()); } @Override diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogDiscoveryMetricValues.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogDiscoveryMetricValues.java index dd380a36080..f6242257bd8 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogDiscoveryMetricValues.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogDiscoveryMetricValues.java @@ -26,14 +26,16 @@ public class ReplicationLogDiscoveryMetricValues { private final long numInProgressDirectoryProcessed; private final long timeToProcessNewFilesMs; private final long timeToProcessInProgressFilesMs; + private final long roundsExceedingRoundTime; public ReplicationLogDiscoveryMetricValues(long numRoundsProcessed, long numInProgressDirectoryProcessed, long timeToProcessNewFilesMs, - long timeToProcessInProgressFilesMs) { + long timeToProcessInProgressFilesMs, long roundsExceedingRoundTime) { this.numRoundsProcessed = numRoundsProcessed; this.numInProgressDirectoryProcessed = numInProgressDirectoryProcessed; this.timeToProcessNewFilesMs = timeToProcessNewFilesMs; this.timeToProcessInProgressFilesMs = timeToProcessInProgressFilesMs; + this.roundsExceedingRoundTime = roundsExceedingRoundTime; } public long getNumRoundsProcessed() { @@ -52,4 +54,8 @@ public long getTimeToProcessInProgressFilesMs() { return timeToProcessInProgressFilesMs; } + public long getRoundsExceedingRoundTime() { + return roundsExceedingRoundTime; + } + } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogProcessorMetricValues.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogProcessorMetricValues.java index 4911ac4bd75..cc09283e237 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogProcessorMetricValues.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogProcessorMetricValues.java @@ -25,15 +25,19 @@ public class ReplicationLogProcessorMetricValues { private final long logFileReplaySuccessCount; private final long logFileReplayTime; private final long logFileBatchReplayTime; + private final long mutationsReplayedCount; + private final long mutationsPerFile; public ReplicationLogProcessorMetricValues(long failedMutationsCount, long logFileReplayFailureCount, long logFileReplaySuccessCount, long logFileReplayTime, - long logFileBatchReplayTime) { + long logFileBatchReplayTime, long mutationsReplayedCount, long mutationsPerFile) { this.failedMutationsCount = failedMutationsCount; this.logFileReplayFailureCount = logFileReplayFailureCount; this.logFileReplaySuccessCount = logFileReplaySuccessCount; this.logFileReplayTime = logFileReplayTime; this.logFileBatchReplayTime = logFileBatchReplayTime; + this.mutationsReplayedCount = mutationsReplayedCount; + this.mutationsPerFile = mutationsPerFile; } public long getFailedMutationsCount() { @@ -55,4 +59,12 @@ public long getLogFileReplayTime() { public long getLogFileBatchReplayTime() { return logFileBatchReplayTime; } + + public long getMutationsReplayedCount() { + return mutationsReplayedCount; + } + + public long getMutationsPerFile() { + return mutationsPerFile; + } } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogTrackerMetricValues.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogTrackerMetricValues.java index 76678db000f..b575b622219 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogTrackerMetricValues.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogTrackerMetricValues.java @@ -29,11 +29,13 @@ public class ReplicationLogTrackerMetricValues { private final long markFileInProgressTimeMs; private final long markFileCompletedTimeMs; private final long markFileFailedTimeMs; + private final long markFileInProgressCollisionCount; public ReplicationLogTrackerMetricValues(long markFileInProgressRequestCount, long markFileCompletedRequestCount, long markFileFailedRequestCount, long markFileCompletedRequestFailedCount, long markFileInProgressTimeMs, - long markFileCompletedTimeMs, long markFileFailedTimeMs) { + long markFileCompletedTimeMs, long markFileFailedTimeMs, + long markFileInProgressCollisionCount) { this.markFileInProgressRequestCount = markFileInProgressRequestCount; this.markFileCompletedRequestCount = markFileCompletedRequestCount; this.markFileFailedRequestCount = markFileFailedRequestCount; @@ -41,6 +43,7 @@ public ReplicationLogTrackerMetricValues(long markFileInProgressRequestCount, this.markFileInProgressTimeMs = markFileInProgressTimeMs; this.markFileCompletedTimeMs = markFileCompletedTimeMs; this.markFileFailedTimeMs = markFileFailedTimeMs; + this.markFileInProgressCollisionCount = markFileInProgressCollisionCount; } public long getMarkFileInProgressRequestCount() { @@ -71,4 +74,8 @@ public long getMarkFileFailedTimeMs() { return markFileFailedTimeMs; } + public long getMarkFileInProgressCollisionCount() { + return markFileInProgressCollisionCount; + } + } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java index a3ae1ca1f79..667d1ac9477 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java @@ -230,8 +230,34 @@ public void init() throws IOException { @Override protected void processFile(Path path) throws IOException { LOG.info("Starting to process file {}", path); - ReplicationLogProcessor.get(getConf(), getHaGroupName()) - .processLogFile(getReplicationLogFileTracker().getFileSystem(), path); + ReplicationLogTracker tracker = getReplicationLogFileTracker(); + long roundEligibleTime = getRoundEligibleTime(tracker.getFileTimestamp(path)); + // Pickup lag (eligible -> claimed): the claim (rename into the in-progress directory) already + // happened in markInProgress, so record it at entry. Skip if the in-progress name carries no + // rename timestamp (should not happen for a file that reached processFile). + Optional renameTs = tracker.getRenameTimestamp(path); + renameTs + .ifPresent(ts -> getReplayMetrics().updatePickupLag(Math.max(0L, ts - roundEligibleTime))); + ReplicationLogProcessor.get(getConf(), getHaGroupName()).processLogFile(tracker.getFileSystem(), + path); + // End-to-end lag (eligible -> replay done): recorded only after a successful processLogFile + // return. On failure processLogFile throws and logFileReplayFailureCount already fires, so an + // end-to-end lag sample for an unfinished replay would be misleading. + getReplayMetrics().updateEndToEndReplayLag( + Math.max(0L, EnvironmentEdgeManager.currentTime() - roundEligibleTime)); + } + + /** + * Returns the wall-clock instant (ms) at which the round owning a file with the given creation + * timestamp became eligible for processing: the round's end boundary plus the waiting buffer. + * This mirrors the eligibility gate used by {@link #getNextRoundToProcess()} (a round is eligible + * once {@code currentTime >= roundEnd + bufferMillis}) and is the zero-reference for the + * replay-lag metrics, so they exclude the fixed built-in wait rather than measuring raw file age. + * @param creationTs the file creation timestamp (first component of the log file name) + * @return the round-eligible wall-clock instant in milliseconds + */ + private long getRoundEligibleTime(long creationTs) { + return (creationTs / roundTimeMills) * roundTimeMills + roundTimeMills + bufferMillis; } /** diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java index 414bd1bb63a..d085c635c25 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java @@ -278,6 +278,12 @@ public void processLogFile(FileSystem fs, Path filePath) throws IOException { LOG.info("Completed processing log file {}. Total mutations processed: {}", logFileReader.getContext().getFilePath(), totalProcessed); getMetrics().incrementLogFileReplaySuccessCount(); + // Replay throughput: unconditional (+= 0 is a no-op for rotation-only files). The per-file + // distribution excludes zero-mutation files so they do not skew it toward zero. + getMetrics().incrementMutationsReplayedCount(totalProcessed); + if (totalProcessed > 0) { + getMetrics().updateMutationsPerFile(totalProcessed); + } } catch (Exception e) { LOG.error("Error while processing replication log file", e); getMetrics().incrementLogFileReplayFailureCount(); diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java index db9ae36fc23..6db1c8dcafe 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java @@ -329,7 +329,8 @@ public void testScheduleNextReplayAdvancesPastLastTargetedGrid() { discovery.setScheduler(mockScheduler); long roundTimeMs = discovery.roundTimeMills; // Pin wall-clock exactly on an epsilon-shifted grid tick so computeAlignedInitialDelay == 0. - long onTick = 5 * roundTimeMs + discovery.bufferMillis + discovery.getAlignedDelayEpsilonMillis(); + long onTick = + 5 * roundTimeMs + discovery.bufferMillis + discovery.getAlignedDelayEpsilonMillis(); AtomicLong mockTime = new AtomicLong(onTick); EnvironmentEdgeManager.injectEdge(new EnvironmentEdge() { @Override @@ -1070,6 +1071,60 @@ public void testProcessNewFilesForRound() throws IOException { } } + /** + * Metric #6 (roundsExceedingRoundTime) recording site: a round whose new-file processing + * completes within the round time must NOT increment the counter. Freezing the clock makes the + * measured duration exactly 0, which is well under {@code roundTimeMills}. + */ + @Test + public void testProcessNewFilesForRoundDoesNotCountFastRound() throws IOException { + long baseSlowRounds = + metricsLogDiscovery.getCurrentMetricValues().getRoundsExceedingRoundTime(); + + // Freeze the clock so startTime == endTime, giving duration == 0 (<= roundTimeMills). + EnvironmentEdge frozenEdge = () -> 1704153600000L; + EnvironmentEdgeManager.injectEdge(frozenEdge); + try { + // Empty round: no files created for it, so processing returns without entering the loop. + ReplicationRound emptyRound = new ReplicationRound(1704153600000L, 1704153660000L); + discovery.processNewFilesForRound(emptyRound); + } finally { + EnvironmentEdgeManager.reset(); + } + + assertEquals("A fast round must not increment roundsExceedingRoundTime", baseSlowRounds, + metricsLogDiscovery.getCurrentMetricValues().getRoundsExceedingRoundTime()); + } + + /** + * Metric #6 (roundsExceedingRoundTime) recording site: a round whose new-file processing exceeds + * the round time must increment the counter exactly once. An advancing clock whose step exceeds a + * full round guarantees the measured duration (end - start) strictly exceeds + * {@code roundTimeMills} regardless of how many times the wall clock is read during processing. + */ + @Test + public void testProcessNewFilesForRoundCountsSlowRound() throws IOException { + long baseSlowRounds = + metricsLogDiscovery.getCurrentMetricValues().getRoundsExceedingRoundTime(); + + long step = discovery.getRoundTimeMills() + 1L; + AtomicLong clock = new AtomicLong(1704153600000L); + EnvironmentEdge advancingEdge = () -> clock.getAndAdd(step); + EnvironmentEdgeManager.injectEdge(advancingEdge); + try { + // Empty round: the start/end clock reads alone span more than a full round, so duration + // strictly exceeds roundTimeMills. + ReplicationRound emptyRound = new ReplicationRound(1704153600000L, 1704153660000L); + discovery.processNewFilesForRound(emptyRound); + } finally { + EnvironmentEdgeManager.reset(); + } + + assertEquals("A slow round must increment roundsExceedingRoundTime by exactly one", + baseSlowRounds + 1, + metricsLogDiscovery.getCurrentMetricValues().getRoundsExceedingRoundTime()); + } + /** * Tests partial failure handling during new file processing. Validates that successful files are * marked completed while failed files are marked failed. diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java index 1f2fff31585..8b0f52fa6fb 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java @@ -844,6 +844,59 @@ public void testMarkInProgressForNonExistentFile() throws IOException { assertFalse("markInProgress should return false for non-existent file", result.isPresent()); } + @Test + public void testMarkInProgressCollisionIncrementsMetricOnRenameFailure() throws IOException { + // Initialize tracker + tracker.init(); + + // The static metrics source accumulates across tests in this class (e.g. + // testMarkInProgressForNonExistentFile also trips the rename-false path), so assert deltas. + long baseCollisions = metrics.getCurrentMetricValues().getMarkFileInProgressCollisionCount(); + long baseRequests = metrics.getCurrentMetricValues().getMarkFileInProgressRequestCount(); + + // Create a real file so the outcome is driven purely by the claim rename, not a missing source. + ReplicationShardDirectoryManager shardManager = tracker.getReplicationShardDirectoryManager(); + Path shardPath = shardManager.getAllShardPaths().get(0); + localFs.mkdirs(shardPath); + Path originalFile = new Path(shardPath, "1704153600000_rs1.plog"); + localFs.create(originalFile, true).close(); + + // Force the claim rename to lose, as if another replayer claimed the same file first. + Mockito.doReturn(false).when(mockFs).rename(Mockito.eq(originalFile), Mockito.any(Path.class)); + + Optional result = tracker.markInProgress(originalFile); + + assertFalse("markInProgress must fail when the claim rename loses", result.isPresent()); + assertEquals("Collision count must increment by exactly one on rename failure", + baseCollisions + 1, metrics.getCurrentMetricValues().getMarkFileInProgressCollisionCount()); + assertEquals("Request count must also increment (collision is a strict subset of requests)", + baseRequests + 1, metrics.getCurrentMetricValues().getMarkFileInProgressRequestCount()); + } + + @Test + public void testMarkInProgressNoCollisionOnRenameSuccess() throws IOException { + // Initialize tracker + tracker.init(); + + long baseCollisions = metrics.getCurrentMetricValues().getMarkFileInProgressCollisionCount(); + long baseRequests = metrics.getCurrentMetricValues().getMarkFileInProgressRequestCount(); + + // Create a real file that the real rename will successfully claim. + ReplicationShardDirectoryManager shardManager = tracker.getReplicationShardDirectoryManager(); + Path shardPath = shardManager.getAllShardPaths().get(0); + localFs.mkdirs(shardPath); + Path originalFile = new Path(shardPath, "1704153600000_rs1.plog"); + localFs.create(originalFile, true).close(); + + Optional result = tracker.markInProgress(originalFile); + + assertTrue("markInProgress must succeed when the claim rename lands", result.isPresent()); + assertEquals("Collision count must not change on a successful claim", baseCollisions, + metrics.getCurrentMetricValues().getMarkFileInProgressCollisionCount()); + assertEquals("Request count still increments on a successful claim", baseRequests + 1, + metrics.getCurrentMetricValues().getMarkFileInProgressRequestCount()); + } + @Test public void testMarkCompletedSuccessfulDeletion() throws IOException { // Initialize tracker diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java new file mode 100644 index 00000000000..cc5dc90aaf2 --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java @@ -0,0 +1,114 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.replication.metrics; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; + +/** + * Unit tests for the replay performance metrics added under PHOENIX-7992. These verify the + * metric-source plumbing (registration, recording, and read-back) for the six metrics across the + * three replay metric sources: + *
    + *
  • {@link MetricsReplicationLogDiscoveryReplayImpl}: endToEndReplayLag, pickupLag (histograms) + * and roundsExceedingRoundTime (counter)
    • + *
  • {@link MetricsReplicationLogProcessorImpl}: mutationsReplayedCount (counter) and + * mutationsPerFile (histogram)
    • + *
  • {@link MetricsReplicationLogTrackerReplayImpl}: markFileInProgressCollisionCount + * (counter)
    • + *
+ * The sources are {@code static} and constructed once (mirroring + * {@code ReplicationLogDiscoveryTest}) because Hadoop's {@code DefaultMetricsSystem} throws on + * duplicate source registration and these sources do not all unregister cleanly on close(). Counter + * assertions are therefore delta-based; each histogram is written by exactly one test method, so + * its {@code getMax()} is deterministic. + */ +public class ReplicationReplayPerfMetricsTest { + + private static final String HA_GROUP = "replayPerfMetricsGroup"; + + private static final MetricsReplicationLogDiscoveryReplayImpl DISCOVERY = + new MetricsReplicationLogDiscoveryReplayImpl(HA_GROUP); + private static final MetricsReplicationLogProcessorImpl PROCESSOR = + new MetricsReplicationLogProcessorImpl(HA_GROUP); + private static final MetricsReplicationLogTrackerReplayImpl TRACKER = + new MetricsReplicationLogTrackerReplayImpl(HA_GROUP); + + /** + * Metrics #1 (endToEndReplayLag), #2 (pickupLag), and #6 (roundsExceedingRoundTime) live on the + * replay discovery source. The lag histograms report their max across samples; the slow-round + * counter increments once per call. + */ + @Test + public void testDiscoveryReplayLagAndSlowRoundMetrics() { + // #1 endToEndReplayLag histogram: getMax reflects the largest recorded sample. + DISCOVERY.updateEndToEndReplayLag(1200L); + DISCOVERY.updateEndToEndReplayLag(800L); + assertEquals("endToEndReplayLag should report the max sample", 1200L, + DISCOVERY.getEndToEndReplayLagHistogram().getMax()); + + // #2 pickupLag histogram. + DISCOVERY.updatePickupLag(300L); + DISCOVERY.updatePickupLag(900L); + assertEquals("pickupLag should report the max sample", 900L, + DISCOVERY.getPickupLagHistogram().getMax()); + + // #6 roundsExceedingRoundTime counter. + long baseRounds = DISCOVERY.getCurrentMetricValues().getRoundsExceedingRoundTime(); + DISCOVERY.incrementRoundsExceedingRoundTime(); + DISCOVERY.incrementRoundsExceedingRoundTime(); + assertEquals("roundsExceedingRoundTime should increment once per call", baseRounds + 2, + DISCOVERY.getCurrentMetricValues().getRoundsExceedingRoundTime()); + } + + /** + * Metrics #3 (mutationsReplayedCount) and #4 (mutationsPerFile) live on the log processor source. + * The count accumulates the supplied delta; the per-file histogram reports its max. + */ + @Test + public void testProcessorMutationMetrics() { + // #3 mutationsReplayedCount counter accumulates by the supplied delta. + long baseMutations = PROCESSOR.getCurrentMetricValues().getMutationsReplayedCount(); + PROCESSOR.incrementMutationsReplayedCount(150L); + PROCESSOR.incrementMutationsReplayedCount(50L); + assertEquals("mutationsReplayedCount should accumulate the supplied deltas", + baseMutations + 200, PROCESSOR.getCurrentMetricValues().getMutationsReplayedCount()); + + // #4 mutationsPerFile histogram reports the max per-file count. + PROCESSOR.updateMutationsPerFile(150L); + PROCESSOR.updateMutationsPerFile(400L); + PROCESSOR.updateMutationsPerFile(30L); + assertEquals("mutationsPerFile should report the max per-file count", 400L, + PROCESSOR.getCurrentMetricValues().getMutationsPerFile()); + } + + /** + * Metric #5 (markFileInProgressCollisionCount) lives on the tracker source and increments once + * per call. + */ + @Test + public void testTrackerCollisionMetric() { + long baseCollisions = TRACKER.getCurrentMetricValues().getMarkFileInProgressCollisionCount(); + TRACKER.incrementMarkFileInProgressCollisionCount(); + TRACKER.incrementMarkFileInProgressCollisionCount(); + TRACKER.incrementMarkFileInProgressCollisionCount(); + assertEquals("markFileInProgressCollisionCount should increment once per call", + baseCollisions + 3, TRACKER.getCurrentMetricValues().getMarkFileInProgressCollisionCount()); + } +} From 02ad6ec60244b42f2b19aa124a6c3f7dfbc33b3c Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Sat, 22 Aug 2026 19:45:59 +0530 Subject: [PATCH 2/2] PHOENIX-7992 Refine replay perf metrics and add recording-site tests Record pickupLag only on a file's first claim, skipping in-progress reclaims that re-stamp the rename timestamp; anchor round-boundary files to the earlier round in getRoundEligibleTime; and rename two counters to match what they actually measure: mutationsReplayedCount -> successfulFileMutationsReplayedCount and markFileInProgressCollisionCount -> markFileInProgressRenameFailedCount. Adds ReplicationLogDiscoveryReplayProcessFileTest covering the eligibility math, first-claim-only pickup lag, and mid-replay failure handling. --- .../replication/ReplicationLogDiscovery.java | 25 ++- .../ReplicationLogDiscoveryForwarder.java | 8 +- .../replication/ReplicationLogTracker.java | 8 +- .../MetricsReplicationLogProcessor.java | 15 +- .../MetricsReplicationLogProcessorImpl.java | 14 +- .../metrics/MetricsReplicationLogTracker.java | 17 +- .../MetricsReplicationLogTrackerImpl.java | 13 +- .../ReplicationLogProcessorMetricValues.java | 10 +- .../ReplicationLogTrackerMetricValues.java | 10 +- .../reader/ReplicationLogDiscoveryReplay.java | 52 +++-- .../reader/ReplicationLogProcessor.java | 9 +- .../ReplicationLogDiscoveryForwarderTest.java | 12 +- .../ReplicationLogDiscoveryTest.java | 106 ++++++---- .../ReplicationLogTrackerTest.java | 15 +- .../ReplicationReplayPerfMetricsTest.java | 43 ++-- ...tionLogDiscoveryReplayProcessFileTest.java | 198 ++++++++++++++++++ 16 files changed, 420 insertions(+), 135 deletions(-) create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayProcessFileTest.java diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java index 81091670b74..957aec9d347 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java @@ -439,7 +439,7 @@ protected void processNewFilesForRound(ReplicationRound replicationRound) throws List files = replicationLogTracker.getNewFilesForRound(replicationRound); LOG.info("Number of new files for round {} is {}", replicationRound, files.size()); while (!files.isEmpty() && isRunning()) { - processOneRandomFile(files); + processOneRandomFile(files, true); files = replicationLogTracker.getNewFilesForRound(replicationRound); } long duration = EnvironmentEdgeManager.currentTime() - startTime; @@ -472,7 +472,7 @@ protected void processInProgressDirectory() throws IOException { replicationLogTracker.getInProgressLogSubDirectoryName(), renameTimestampThreshold, files.size(), haGroupName); while (!files.isEmpty() && isRunning()) { - Optional failedFile = processOneRandomFile(files); + Optional failedFile = processOneRandomFile(files, false); if (failedFile.isPresent()) { String prefix = replicationLogTracker.getFilePrefix(failedFile.get()); int count = failureCount.merge(prefix, 1, Integer::sum); @@ -498,17 +498,23 @@ protected void processInProgressDirectory() throws IOException { /** * Processes a single random file from the provided list. Marks the file as in-progress, processes * it, and marks it as completed or failed. - * @param files - List of files from which to select and process one randomly + * @param files - List of files from which to select and process one randomly + * @param firstClaim - true when {@code files} are new files being claimed for the first time + * (from {@link #processNewFilesForRound}); false when they are already in the + * in-progress directory and are being reprocessed (from + * {@link #processInProgressDirectory}). Forwarded to + * {@link #processFile(Path, boolean)}. * @return the original path of the file that failed, or empty if processing succeeded */ - private Optional processOneRandomFile(final List files) throws IOException { + private Optional processOneRandomFile(final List files, final boolean firstClaim) + throws IOException { // Pick a random file and process it Path file = files.get(ThreadLocalRandom.current().nextInt(files.size())); Optional optionalInProgressFilePath = Optional.empty(); try { optionalInProgressFilePath = replicationLogTracker.markInProgress(file); if (optionalInProgressFilePath.isPresent()) { - processFile(optionalInProgressFilePath.get()); + processFile(optionalInProgressFilePath.get(), firstClaim); replicationLogTracker.markCompleted(optionalInProgressFilePath.get()); } } catch (IOException exception) { @@ -522,10 +528,15 @@ private Optional processOneRandomFile(final List files) throws IOExc /** * Handles the processing of a single file. - * @param path - The file to be processed + * @param path - The file to be processed + * @param firstClaim - true when this invocation is the file's first claim (it was just moved into + * the in-progress directory from the new-files path); false when reprocessing a + * file that was already in the in-progress directory. Lets subclasses record + * first-claim-only signals (e.g. pickup lag) without double counting on + * retries. * @throws IOException if there's an error during file processing */ - protected abstract void processFile(Path path) throws IOException; + protected abstract void processFile(Path path, boolean firstClaim) throws IOException; /** Creates a new metrics source for monitoring operations. */ protected abstract MetricsReplicationLogDiscovery createMetricsSource(); diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java index 097760bb8f3..82a5347d93d 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java @@ -104,8 +104,14 @@ public void init() throws IOException { super.init(); } + /** + * Forwards a single log file to the peer cluster. The {@code firstClaim} flag (true when the file + * was just claimed from the new-files directory, false on an in-progress reclaim) is unused here: + * forwarding is idempotent on the stable (ts, originServerName) destination and records no + * claim-latency metrics, so first claims and reclaims are handled identically. + */ @Override - protected void processFile(Path src) throws IOException { + protected void processFile(Path src, boolean firstClaim) throws IOException { FileSystem srcFS = replicationLogTracker.getFileSystem(); FileStatus srcStat = srcFS.getFileStatus(src); long ts = replicationLogTracker.getFileTimestamp(srcStat.getPath()); diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java index 90a1c58db96..0a13a13ac9f 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java @@ -384,9 +384,11 @@ protected Optional markInProgress(final Path file) { return Optional.of(newPath); } else { LOG.warn("Failed to rename file for in-progress marking: {}", file); - // Rename lost to another process claiming the same file: a collision, a strict subset of - // the request count incremented in the finally block below. - getMetrics().incrementMarkFileInProgressCollisionCount(); + // Claim rename returned false. Usually another process claimed the same file first (the + // common collision case), but any other rename() == false outcome lands here too, so the + // counter is named for the observable condition (rename failed), not the presumed cause. A + // strict subset of the request count incremented in the finally block below. + getMetrics().incrementMarkFileInProgressRenameFailedCount(); return Optional.empty(); } } catch (IOException e) { diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java index 0281385bdcf..f6169e459eb 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessor.java @@ -41,8 +41,10 @@ public interface MetricsReplicationLogProcessor extends BaseSource { String LOG_FILE_REPLAY_TIME = "logFileReplayTimeMs"; String LOG_FILE_REPLAY_TIME_DESC = "Histogram of time taken for replaying a log file in milliseconds"; - String MUTATIONS_REPLAYED_COUNT = "mutationsReplayedCount"; - String MUTATIONS_REPLAYED_COUNT_DESC = "Total number of mutations replayed across all log files"; + String SUCCESSFUL_FILE_MUTATIONS_REPLAYED_COUNT = "successfulFileMutationsReplayedCount"; + String SUCCESSFUL_FILE_MUTATIONS_REPLAYED_COUNT_DESC = + "Total number of mutations replayed across log files that completed successfully; mutations " + + "already applied by a file that then fails mid-replay are not counted"; String MUTATIONS_PER_FILE = "mutationsPerFile"; String MUTATIONS_PER_FILE_DESC = "Histogram of the number of mutations replayed per log file (files with mutations only)"; @@ -84,11 +86,12 @@ public interface MetricsReplicationLogProcessor extends BaseSource { void updateLogFileReplayTime(long timeMs); /** - * Increments the total number of mutations replayed by the given amount. This counter tracks - * replay throughput across all log files. - * @param delta the number of mutations replayed for a log file + * Increments the count of mutations replayed by files that completed successfully. Called once + * per file after its replay returns without error, so mutations already applied by a file that + * then fails mid-replay are not counted. Tracks replay throughput across fully-successful files. + * @param delta the number of mutations replayed for a successfully completed log file */ - void incrementMutationsReplayedCount(long delta); + void incrementSuccessfulFileMutationsReplayedCount(long delta); /** * Records a sample into the per-file mutation-count histogram. Called only for files that diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessorImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessorImpl.java index 2643eaef162..71aa5d5fbb7 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessorImpl.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogProcessorImpl.java @@ -33,7 +33,7 @@ public class MetricsReplicationLogProcessorImpl extends BaseSourceImpl private final MutableFastCounter logFileReplaySuccessCount; private final MutableHistogram batchReplayTime; private final MutableHistogram logFileReplayTime; - private final MutableFastCounter mutationsReplayedCount; + private final MutableFastCounter successfulFileMutationsReplayedCount; private final MutableHistogram mutationsPerFile; public MetricsReplicationLogProcessorImpl(final String haGroupName) { @@ -56,8 +56,8 @@ public MetricsReplicationLogProcessorImpl(String metricsName, String metricsDesc batchReplayTime = getMetricsRegistry().newHistogram(BATCH_REPLAY_TIME, BATCH_REPLAY_TIME_DESC); logFileReplayTime = getMetricsRegistry().newHistogram(LOG_FILE_REPLAY_TIME, LOG_FILE_REPLAY_TIME_DESC); - mutationsReplayedCount = - getMetricsRegistry().newCounter(MUTATIONS_REPLAYED_COUNT, MUTATIONS_REPLAYED_COUNT_DESC, 0L); + successfulFileMutationsReplayedCount = getMetricsRegistry().newCounter( + SUCCESSFUL_FILE_MUTATIONS_REPLAYED_COUNT, SUCCESSFUL_FILE_MUTATIONS_REPLAYED_COUNT_DESC, 0L); mutationsPerFile = getMetricsRegistry().newHistogram(MUTATIONS_PER_FILE, MUTATIONS_PER_FILE_DESC); } @@ -93,8 +93,8 @@ public void updateLogFileReplayTime(long timeMs) { } @Override - public void incrementMutationsReplayedCount(long delta) { - mutationsReplayedCount.incr(delta); + public void incrementSuccessfulFileMutationsReplayedCount(long delta) { + successfulFileMutationsReplayedCount.incr(delta); } @Override @@ -112,8 +112,8 @@ public void close() { public ReplicationLogProcessorMetricValues getCurrentMetricValues() { return new ReplicationLogProcessorMetricValues(failedMutationsCount.value(), logFileReplayFailureCount.value(), logFileReplaySuccessCount.value(), - logFileReplayTime.getMax(), batchReplayTime.getMax(), mutationsReplayedCount.value(), - mutationsPerFile.getMax()); + logFileReplayTime.getMax(), batchReplayTime.getMax(), + successfulFileMutationsReplayedCount.value(), mutationsPerFile.getMax()); } @Override diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTracker.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTracker.java index 47d8ff24e1a..951388a125e 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTracker.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTracker.java @@ -29,9 +29,11 @@ public interface MetricsReplicationLogTracker extends BaseSource { String MARK_FILE_IN_PROGRESS_REQUEST_COUNT = "markFileInProgressRequestCount"; String MARK_FILE_IN_PROGRESS_REQUEST_COUNT_DESC = "Number of requests made to mark file in progress"; - String MARK_FILE_IN_PROGRESS_COLLISION_COUNT = "markFileInProgressCollisionCount"; - String MARK_FILE_IN_PROGRESS_COLLISION_COUNT_DESC = - "Number of mark-file-in-progress requests that lost the claim rename to another process"; + String MARK_FILE_IN_PROGRESS_RENAME_FAILED_COUNT = "markFileInProgressRenameFailedCount"; + String MARK_FILE_IN_PROGRESS_RENAME_FAILED_COUNT_DESC = + "Number of mark-file-in-progress attempts whose claim rename returned false; typically the " + + "claim was lost to another process, but any other rename failure is also counted. A strict " + + "subset of the request count."; String MARK_FILE_COMPLETED_REQUEST_COUNT = "markFileCompletedRequestCount"; String MARK_FILE_COMPLETED_REQUEST_COUNT_DESC = "Number of requests made to mark file completed"; String MARK_FILE_FAILED_REQUEST_COUNT = "markFileFailedRequestCount"; @@ -56,11 +58,12 @@ public interface MetricsReplicationLogTracker extends BaseSource { void incrementMarkFileInProgressRequestCount(); /** - * Increments the counter for mark-file-in-progress collisions. This counter tracks the number of - * claim attempts whose rename lost to another process (the decentralized, no-coordination claim - * design), a strict subset of {@link #incrementMarkFileInProgressRequestCount()}. + * Increments the counter for mark-file-in-progress claim renames that returned false. Typically + * the claim was lost to another process (the decentralized, no-coordination claim design), but + * any other {@code rename() == false} outcome is also counted. A strict subset of + * {@link #incrementMarkFileInProgressRequestCount()}. */ - void incrementMarkFileInProgressCollisionCount(); + void incrementMarkFileInProgressRenameFailedCount(); /** * Increments the counter for mark file completed requests. This counter tracks the number of diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTrackerImpl.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTrackerImpl.java index c503c476c80..f83b181e0a9 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTrackerImpl.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/MetricsReplicationLogTrackerImpl.java @@ -28,7 +28,7 @@ public class MetricsReplicationLogTrackerImpl extends BaseSourceImpl protected String groupMetricsContext; protected final MutableFastCounter markFileInProgressRequestCount; - protected final MutableFastCounter markFileInProgressCollisionCount; + protected final MutableFastCounter markFileInProgressRenameFailedCount; protected final MutableFastCounter markFileCompletedRequestCount; protected final MutableFastCounter markFileFailedRequestCount; protected final MutableFastCounter markFileCompletedRequestFailedCount; @@ -41,8 +41,9 @@ public MetricsReplicationLogTrackerImpl(String metricsName, String metricsDescri super(metricsName, metricsDescription, metricsContext, metricsJmxContext); markFileInProgressRequestCount = getMetricsRegistry().newCounter( MARK_FILE_IN_PROGRESS_REQUEST_COUNT, MARK_FILE_IN_PROGRESS_REQUEST_COUNT_DESC, 0L); - markFileInProgressCollisionCount = getMetricsRegistry().newCounter( - MARK_FILE_IN_PROGRESS_COLLISION_COUNT, MARK_FILE_IN_PROGRESS_COLLISION_COUNT_DESC, 0L); + markFileInProgressRenameFailedCount = + getMetricsRegistry().newCounter(MARK_FILE_IN_PROGRESS_RENAME_FAILED_COUNT, + MARK_FILE_IN_PROGRESS_RENAME_FAILED_COUNT_DESC, 0L); markFileCompletedRequestCount = getMetricsRegistry() .newCounter(MARK_FILE_COMPLETED_REQUEST_COUNT, MARK_FILE_COMPLETED_REQUEST_COUNT_DESC, 0L); markFileFailedRequestCount = getMetricsRegistry().newCounter(MARK_FILE_FAILED_REQUEST_COUNT, @@ -63,8 +64,8 @@ public void incrementMarkFileInProgressRequestCount() { } @Override - public void incrementMarkFileInProgressCollisionCount() { - markFileInProgressCollisionCount.incr(); + public void incrementMarkFileInProgressRenameFailedCount() { + markFileInProgressRenameFailedCount.incr(); } @Override @@ -109,7 +110,7 @@ public ReplicationLogTrackerMetricValues getCurrentMetricValues() { markFileCompletedRequestCount.value(), markFileFailedRequestCount.value(), markFileCompletedRequestFailedCount.value(), markFileInProgressTime.getMax(), markFileCompletedTime.getMax(), markFileFailedTime.getMax(), - markFileInProgressCollisionCount.value()); + markFileInProgressRenameFailedCount.value()); } @Override diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogProcessorMetricValues.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogProcessorMetricValues.java index cc09283e237..8a0b276a0b2 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogProcessorMetricValues.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogProcessorMetricValues.java @@ -25,18 +25,18 @@ public class ReplicationLogProcessorMetricValues { private final long logFileReplaySuccessCount; private final long logFileReplayTime; private final long logFileBatchReplayTime; - private final long mutationsReplayedCount; + private final long successfulFileMutationsReplayedCount; private final long mutationsPerFile; public ReplicationLogProcessorMetricValues(long failedMutationsCount, long logFileReplayFailureCount, long logFileReplaySuccessCount, long logFileReplayTime, - long logFileBatchReplayTime, long mutationsReplayedCount, long mutationsPerFile) { + long logFileBatchReplayTime, long successfulFileMutationsReplayedCount, long mutationsPerFile) { this.failedMutationsCount = failedMutationsCount; this.logFileReplayFailureCount = logFileReplayFailureCount; this.logFileReplaySuccessCount = logFileReplaySuccessCount; this.logFileReplayTime = logFileReplayTime; this.logFileBatchReplayTime = logFileBatchReplayTime; - this.mutationsReplayedCount = mutationsReplayedCount; + this.successfulFileMutationsReplayedCount = successfulFileMutationsReplayedCount; this.mutationsPerFile = mutationsPerFile; } @@ -60,8 +60,8 @@ public long getLogFileBatchReplayTime() { return logFileBatchReplayTime; } - public long getMutationsReplayedCount() { - return mutationsReplayedCount; + public long getSuccessfulFileMutationsReplayedCount() { + return successfulFileMutationsReplayedCount; } public long getMutationsPerFile() { diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogTrackerMetricValues.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogTrackerMetricValues.java index b575b622219..ddaa5705369 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogTrackerMetricValues.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/metrics/ReplicationLogTrackerMetricValues.java @@ -29,13 +29,13 @@ public class ReplicationLogTrackerMetricValues { private final long markFileInProgressTimeMs; private final long markFileCompletedTimeMs; private final long markFileFailedTimeMs; - private final long markFileInProgressCollisionCount; + private final long markFileInProgressRenameFailedCount; public ReplicationLogTrackerMetricValues(long markFileInProgressRequestCount, long markFileCompletedRequestCount, long markFileFailedRequestCount, long markFileCompletedRequestFailedCount, long markFileInProgressTimeMs, long markFileCompletedTimeMs, long markFileFailedTimeMs, - long markFileInProgressCollisionCount) { + long markFileInProgressRenameFailedCount) { this.markFileInProgressRequestCount = markFileInProgressRequestCount; this.markFileCompletedRequestCount = markFileCompletedRequestCount; this.markFileFailedRequestCount = markFileFailedRequestCount; @@ -43,7 +43,7 @@ public ReplicationLogTrackerMetricValues(long markFileInProgressRequestCount, this.markFileInProgressTimeMs = markFileInProgressTimeMs; this.markFileCompletedTimeMs = markFileCompletedTimeMs; this.markFileFailedTimeMs = markFileFailedTimeMs; - this.markFileInProgressCollisionCount = markFileInProgressCollisionCount; + this.markFileInProgressRenameFailedCount = markFileInProgressRenameFailedCount; } public long getMarkFileInProgressRequestCount() { @@ -74,8 +74,8 @@ public long getMarkFileFailedTimeMs() { return markFileFailedTimeMs; } - public long getMarkFileInProgressCollisionCount() { - return markFileInProgressCollisionCount; + public long getMarkFileInProgressRenameFailedCount() { + return markFileInProgressRenameFailedCount; } } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java index 667d1ac9477..a8244092619 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java @@ -228,36 +228,62 @@ public void init() throws IOException { } @Override - protected void processFile(Path path) throws IOException { - LOG.info("Starting to process file {}", path); + protected void processFile(Path path, boolean firstClaim) throws IOException { + LOG.info("Starting to process file {} (firstClaim={})", path, firstClaim); ReplicationLogTracker tracker = getReplicationLogFileTracker(); long roundEligibleTime = getRoundEligibleTime(tracker.getFileTimestamp(path)); - // Pickup lag (eligible -> claimed): the claim (rename into the in-progress directory) already - // happened in markInProgress, so record it at entry. Skip if the in-progress name carries no - // rename timestamp (should not happen for a file that reached processFile). - Optional renameTs = tracker.getRenameTimestamp(path); - renameTs - .ifPresent(ts -> getReplayMetrics().updatePickupLag(Math.max(0L, ts - roundEligibleTime))); - ReplicationLogProcessor.get(getConf(), getHaGroupName()).processLogFile(tracker.getFileSystem(), - path); - // End-to-end lag (eligible -> replay done): recorded only after a successful processLogFile - // return. On failure processLogFile throws and logFileReplayFailureCount already fires, so an + // Pickup lag (eligible -> claimed) is recorded only on the first claim (the new-files path). + // markInProgress re-stamps a fresh rename timestamp on every reclaim of an already-in-progress + // file, so recording on the in-progress path (firstClaim=false) would sample eligible -> + // latest-reclaim, not eligible -> first-pickup, and multi-count the same file across sweeps. + // Skip if the in-progress name carries no rename timestamp (should not happen for a file that + // reached processFile). + if (firstClaim) { + Optional renameTs = tracker.getRenameTimestamp(path); + renameTs + .ifPresent(ts -> getReplayMetrics().updatePickupLag(Math.max(0L, ts - roundEligibleTime))); + } + replayLogFile(path); + // End-to-end lag (eligible -> replay done): recorded only after a successful replayLogFile + // return. On failure replayLogFile throws and logFileReplayFailureCount already fires, so an // end-to-end lag sample for an unfinished replay would be misleading. getReplayMetrics().updateEndToEndReplayLag( Math.max(0L, EnvironmentEdgeManager.currentTime() - roundEligibleTime)); } + /** + * Replays a single log file through the {@link ReplicationLogProcessor}. Extracted as a seam so + * the lag-recording logic in {@link #processFile(Path, boolean)} can be unit-tested without a + * live processor or file system. + * @param path the in-progress log file to replay + * @throws IOException if replay fails; the caller then skips the end-to-end lag sample + */ + protected void replayLogFile(Path path) throws IOException { + ReplicationLogProcessor.get(getConf(), getHaGroupName()) + .processLogFile(getReplicationLogFileTracker().getFileSystem(), path); + } + /** * Returns the wall-clock instant (ms) at which the round owning a file with the given creation * timestamp became eligible for processing: the round's end boundary plus the waiting buffer. * This mirrors the eligibility gate used by {@link #getNextRoundToProcess()} (a round is eligible * once {@code currentTime >= roundEnd + bufferMillis}) and is the zero-reference for the * replay-lag metrics, so they exclude the fixed built-in wait rather than measuring raw file age. + *

+ * Round bounds are inclusive at both ends (see {@code getNewFilesForRound}), so consecutive + * rounds share a boundary and a file whose creation timestamp lands exactly on a round boundary + * ({@code creationTs % roundTimeMills == 0}) matches both the earlier round (as its end) and the + * later round (as its start). The earlier round runs first and claims the file, so the owning + * round's end is {@code creationTs} itself in that case, not {@code creationTs + roundTimeMills}. + * Anchoring to the later round would over-count the eligibility by one full round and clamp the + * resulting lag to zero. * @param creationTs the file creation timestamp (first component of the log file name) * @return the round-eligible wall-clock instant in milliseconds */ private long getRoundEligibleTime(long creationTs) { - return (creationTs / roundTimeMills) * roundTimeMills + roundTimeMills + bufferMillis; + long roundStart = (creationTs / roundTimeMills) * roundTimeMills; + long owningRoundEnd = (creationTs == roundStart) ? creationTs : roundStart + roundTimeMills; + return owningRoundEnd + bufferMillis; } /** diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java index d085c635c25..423b2202bc0 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java @@ -278,9 +278,12 @@ public void processLogFile(FileSystem fs, Path filePath) throws IOException { LOG.info("Completed processing log file {}. Total mutations processed: {}", logFileReader.getContext().getFilePath(), totalProcessed); getMetrics().incrementLogFileReplaySuccessCount(); - // Replay throughput: unconditional (+= 0 is a no-op for rotation-only files). The per-file - // distribution excludes zero-mutation files so they do not skew it toward zero. - getMetrics().incrementMutationsReplayedCount(totalProcessed); + // Replay throughput for files that completed successfully. This runs only after the whole + // file replayed without error (before the catch), so mutations already applied by a file that + // then fails mid-replay are not counted (+= 0 is a no-op for rotation-only files). The + // per-file distribution below excludes zero-mutation files so they do not skew it toward + // zero. + getMetrics().incrementSuccessfulFileMutationsReplayedCount(totalProcessed); if (totalProcessed > 0) { getMetrics().updateMutationsPerFile(totalProcessed); } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java index 725a2f06628..c49b1b0f589 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java @@ -195,8 +195,8 @@ public void testForwardPreservesOriginServerIdentity() throws Exception { Path srcB = markInProgressSource(localTracker, new Path(shardDir, ts + "_" + originB + ".plog")); - forwarder.processFile(srcA); - forwarder.processFile(srcB); + forwarder.processFile(srcA, true); + forwarder.processFile(srcB, true); // Both files must land on the peer under their own origin identity -- no collision. Path peerShardDir = peerShardManager.getShardDirectory(ts); @@ -314,7 +314,7 @@ public Boolean answer(InvocationOnMock invocation) throws Throwable { } }).when(peerFs).rename(eq(staging), eq(dst)); - logGroup.getLogForwarder().processFile(src); + logGroup.getLogForwarder().processFile(src, true); assertTrue("rename interceptor should have run", invariantHeld[0]); assertTrue(".plog must exist after rename", peerFs.exists(dst)); @@ -346,7 +346,7 @@ public void testForwardRetryOntoExistingDestinationSucceeds() throws Exception { out.write(existing); } - logGroup.getLogForwarder().processFile(src); + logGroup.getLogForwarder().processFile(src, true); assertTrue("dst should still exist", peerFs.exists(dst)); assertFalse("staging file should be cleaned up", peerFs.exists(staging)); @@ -371,7 +371,7 @@ public void testForwardReclaimsOrphanStagingFile() throws Exception { out.write("stale-garbage-bytes".getBytes()); } - logGroup.getLogForwarder().processFile(src); + logGroup.getLogForwarder().processFile(src, true); assertTrue("dst should be published", peerFs.exists(dst)); assertFalse("staging file should be gone", peerFs.exists(staging)); @@ -395,7 +395,7 @@ public void testForwardRenameFailureLeavesSourceForRetry() throws Exception { doReturn(false).when(peerFs).rename(eq(staging), eq(dst)); try { - logGroup.getLogForwarder().processFile(src); + logGroup.getLogForwarder().processFile(src, true); fail("processFile should have thrown on rename failure"); } catch (IOException expected) { // expected diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java index 6db1c8dcafe..e0a72cfa20d 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java @@ -1141,12 +1141,14 @@ public void testProcessNewFilesForRoundWithPartialFailure() throws IOException { // matching String file1Prefix = newFilesForRound.get(1).getName().substring(0, newFilesForRound.get(1).getName().lastIndexOf(".")); - Mockito.doThrow(new IOException("Processing failed for file 1")).when(discovery) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix))); + Mockito.doThrow(new IOException("Processing failed for file 1")).when(discovery).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix)), + Mockito.anyBoolean()); String file3Prefix = newFilesForRound.get(3).getName().substring(0, newFilesForRound.get(3).getName().lastIndexOf(".")); - Mockito.doThrow(new IOException("Processing failed for file 3")).when(discovery) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file3Prefix))); + Mockito.doThrow(new IOException("Processing failed for file 3")).when(discovery).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file3Prefix)), + Mockito.anyBoolean()); // Process new files for the round discovery.processNewFilesForRound(replicationRound); @@ -1161,14 +1163,16 @@ public void testProcessNewFilesForRoundWithPartialFailure() throws IOException { } // Verify that processFile was called for each file in the round - Mockito.verify(discovery, Mockito.times(5)).processFile(Mockito.any(Path.class)); + Mockito.verify(discovery, Mockito.times(5)).processFile(Mockito.any(Path.class), + Mockito.anyBoolean()); // Verify that processFile was called for each specific file (using prefix matching) for (Path expectedFile : newFilesForRound) { String expectedPrefix = expectedFile.getName().substring(0, expectedFile.getName().lastIndexOf(".")); - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(expectedPrefix))); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(expectedPrefix)), + Mockito.anyBoolean()); } // Verify that markCompleted was called for each successfully processed file @@ -1231,14 +1235,16 @@ public void testProcessNewFilesForRoundWithAllFailures() throws IOException { String filePrefix = file.getName().substring(0, file.getName().lastIndexOf(".")); Mockito.doThrow(new IOException("Processing failed for file: " + file.getName())) .when(discovery) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(filePrefix))); + .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(filePrefix)), + Mockito.anyBoolean()); } // Process new files for the round discovery.processNewFilesForRound(replicationRound); // Verify that processFile was called for each file in the round - Mockito.verify(discovery, Mockito.times(5)).processFile(Mockito.any(Path.class)); + Mockito.verify(discovery, Mockito.times(5)).processFile(Mockito.any(Path.class), + Mockito.anyBoolean()); // Verify that markInProgress was called 5 times (before processing fails) Mockito.verify(fileTracker, Mockito.times(5)).markInProgress(Mockito.any(Path.class)); @@ -1253,8 +1259,9 @@ public void testProcessNewFilesForRoundWithAllFailures() throws IOException { for (Path expectedFile : newFilesForRound) { String expectedPrefix = expectedFile.getName().substring(0, expectedFile.getName().lastIndexOf(".")); - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(expectedPrefix))); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(expectedPrefix)), + Mockito.anyBoolean()); } // Verify that markCompleted was NOT called for any file (all failed) @@ -1432,11 +1439,13 @@ public void testProcessInProgressDirectoryWithIntermittentFailure() throws IOExc String file1Prefix = extractPrefix(allInProgressFiles.get(1).getName()); Mockito.doThrow(new IOException("Processing failed for file 1")).doCallRealMethod() .when(discovery) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix))); + .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix)), + Mockito.anyBoolean()); String file3Prefix = extractPrefix(allInProgressFiles.get(3).getName()); Mockito.doThrow(new IOException("Processing failed for file 3")).doCallRealMethod() .when(discovery) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file3Prefix))); + .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file3Prefix)), + Mockito.anyBoolean()); // Inject an advancing clock so that rename timestamps from markInProgress are always // strictly before the renameTimestampThreshold computed on the next loop iteration @@ -1462,7 +1471,8 @@ public void testProcessInProgressDirectoryWithIntermittentFailure() throws IOExc // Verify that processFile was called for each file in the directory (i.e. 5 + 2 times for // failed once that would succeed in next retry) - Mockito.verify(discovery, Mockito.times(7)).processFile(Mockito.any(Path.class)); + Mockito.verify(discovery, Mockito.times(7)).processFile(Mockito.any(Path.class), + Mockito.anyBoolean()); // Verify that processFile was called for each specific file (using prefix matching) // Files 1 and 3 should be called twice (fail once, succeed on retry), others once @@ -1472,7 +1482,8 @@ public void testProcessInProgressDirectoryWithIntermittentFailure() throws IOExc int expectedTimes = (i == 1 || i == 3) ? 2 : 1; // Files 1 and 3 are called twice (fail + // retry success) Mockito.verify(discovery, Mockito.times(expectedTimes)).processFile( - Mockito.argThat(path -> extractPrefix(path.getName()).equals(expectedPrefix))); + Mockito.argThat(path -> extractPrefix(path.getName()).equals(expectedPrefix)), + Mockito.anyBoolean()); } // Verify that markCompleted was called for each successfully processed file @@ -1526,8 +1537,9 @@ public void testProcessInProgressDirectorySkipsFailedFiles() throws IOException // Mock processFile to always throw for file 1 (persistent failure) String file1Prefix = extractPrefix(inProgressFiles.get(1).getName()); - Mockito.doThrow(new IOException("Persistent failure for file 1")).when(discovery) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix))); + Mockito.doThrow(new IOException("Persistent failure for file 1")).when(discovery).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix)), + Mockito.anyBoolean()); // Process in-progress directory discovery.processInProgressDirectory(); @@ -1538,7 +1550,8 @@ public void testProcessInProgressDirectorySkipsFailedFiles() throws IOException Mockito.verify(fileTracker, Mockito.times(3)).markInProgress(Mockito.any(Path.class)); // processFile called 3 times (all files attempted once) - Mockito.verify(discovery, Mockito.times(3)).processFile(Mockito.any(Path.class)); + Mockito.verify(discovery, Mockito.times(3)).processFile(Mockito.any(Path.class), + Mockito.anyBoolean()); // markCompleted called only for the 2 successful files Mockito.verify(fileTracker, Mockito.times(2)).markCompleted(Mockito.any(Path.class)); @@ -1547,16 +1560,19 @@ public void testProcessInProgressDirectorySkipsFailedFiles() throws IOException Mockito.verify(fileTracker, Mockito.times(1)).markFailed(Mockito.any(Path.class)); // Verify the failed file was NOT retried (processFile called exactly once for file1) - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix))); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix)), + Mockito.anyBoolean()); // Verify the successful files were each called exactly once String file0Prefix = extractPrefix(inProgressFiles.get(0).getName()); - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file0Prefix))); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file0Prefix)), + Mockito.anyBoolean()); String file2Prefix = extractPrefix(inProgressFiles.get(2).getName()); - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file2Prefix))); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file2Prefix)), + Mockito.anyBoolean()); } /** @@ -1577,22 +1593,26 @@ public void testProcessInProgressDirectoryAllFilesFail() throws IOException { // Mock processFile to always throw for all files Mockito.doThrow(new IOException("Persistent failure")).when(discovery) - .processFile(Mockito.any(Path.class)); + .processFile(Mockito.any(Path.class), Mockito.anyBoolean()); // Process in-progress directory - should terminate without infinite loop discovery.processInProgressDirectory(); // Each file attempted exactly once Mockito.verify(fileTracker, Mockito.times(3)).markInProgress(Mockito.any(Path.class)); - Mockito.verify(discovery, Mockito.times(3)).processFile(Mockito.any(Path.class)); + Mockito.verify(discovery, Mockito.times(3)).processFile(Mockito.any(Path.class), + Mockito.anyBoolean()); // Verify per-prefix: each file attempted once - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file0Prefix))); - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix))); - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file2Prefix))); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file0Prefix)), + Mockito.anyBoolean()); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix)), + Mockito.anyBoolean()); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file2Prefix)), + Mockito.anyBoolean()); // All files marked as failed Mockito.verify(fileTracker, Mockito.times(3)).markFailed(Mockito.any(Path.class)); @@ -1643,7 +1663,8 @@ public void testProcessInProgressDirectoryFailedFilesSucceedOnNextRound() throws // Mock processFile to fail for file 1 only on the first call, succeed on subsequent calls Mockito.doThrow(new IOException("Transient failure for file 1")).doCallRealMethod() .when(discovery) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix))); + .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix)), + Mockito.anyBoolean()); // --- First invocation: file 1 fails, files 0 and 2 succeed --- discovery.processInProgressDirectory(); @@ -1657,12 +1678,15 @@ public void testProcessInProgressDirectoryFailedFilesSucceedOnNextRound() throws .markInProgress(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file2Prefix))); // Verify processFile called once per file - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file0Prefix))); - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix))); - Mockito.verify(discovery, Mockito.times(1)) - .processFile(Mockito.argThat(path -> extractPrefix(path.getName()).equals(file2Prefix))); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file0Prefix)), + Mockito.anyBoolean()); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file1Prefix)), + Mockito.anyBoolean()); + Mockito.verify(discovery, Mockito.times(1)).processFile( + Mockito.argThat(path -> extractPrefix(path.getName()).equals(file2Prefix)), + Mockito.anyBoolean()); // Verify markCompleted called for files 0 and 2 (successful) Mockito.verify(fileTracker, Mockito.times(1)) @@ -1720,7 +1744,7 @@ public void testProcessInProgressDirectoryFailedFilesSucceedOnNextRound() throws assertTrue("Second round rename timestamp should be newer than first round's", ts.get() > firstRoundRenameTs); return true; - })); + }), Mockito.anyBoolean()); // Verify markCompleted called for file 1 Mockito.verify(fileTracker, Mockito.times(1)) @@ -2740,7 +2764,7 @@ public long getBufferMillis() { } @Override - protected void processFile(Path path) throws IOException { + protected void processFile(Path path, boolean firstClaim) throws IOException { // Simulate file processing processedFiles.add(path); } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java index 8b0f52fa6fb..81258ddc2e7 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java @@ -851,7 +851,8 @@ public void testMarkInProgressCollisionIncrementsMetricOnRenameFailure() throws // The static metrics source accumulates across tests in this class (e.g. // testMarkInProgressForNonExistentFile also trips the rename-false path), so assert deltas. - long baseCollisions = metrics.getCurrentMetricValues().getMarkFileInProgressCollisionCount(); + long baseRenameFailures = + metrics.getCurrentMetricValues().getMarkFileInProgressRenameFailedCount(); long baseRequests = metrics.getCurrentMetricValues().getMarkFileInProgressRequestCount(); // Create a real file so the outcome is driven purely by the claim rename, not a missing source. @@ -867,8 +868,9 @@ public void testMarkInProgressCollisionIncrementsMetricOnRenameFailure() throws Optional result = tracker.markInProgress(originalFile); assertFalse("markInProgress must fail when the claim rename loses", result.isPresent()); - assertEquals("Collision count must increment by exactly one on rename failure", - baseCollisions + 1, metrics.getCurrentMetricValues().getMarkFileInProgressCollisionCount()); + assertEquals("Rename-failed count must increment by exactly one on rename failure", + baseRenameFailures + 1, + metrics.getCurrentMetricValues().getMarkFileInProgressRenameFailedCount()); assertEquals("Request count must also increment (collision is a strict subset of requests)", baseRequests + 1, metrics.getCurrentMetricValues().getMarkFileInProgressRequestCount()); } @@ -878,7 +880,8 @@ public void testMarkInProgressNoCollisionOnRenameSuccess() throws IOException { // Initialize tracker tracker.init(); - long baseCollisions = metrics.getCurrentMetricValues().getMarkFileInProgressCollisionCount(); + long baseRenameFailures = + metrics.getCurrentMetricValues().getMarkFileInProgressRenameFailedCount(); long baseRequests = metrics.getCurrentMetricValues().getMarkFileInProgressRequestCount(); // Create a real file that the real rename will successfully claim. @@ -891,8 +894,8 @@ public void testMarkInProgressNoCollisionOnRenameSuccess() throws IOException { Optional result = tracker.markInProgress(originalFile); assertTrue("markInProgress must succeed when the claim rename lands", result.isPresent()); - assertEquals("Collision count must not change on a successful claim", baseCollisions, - metrics.getCurrentMetricValues().getMarkFileInProgressCollisionCount()); + assertEquals("Rename-failed count must not change on a successful claim", baseRenameFailures, + metrics.getCurrentMetricValues().getMarkFileInProgressRenameFailedCount()); assertEquals("Request count still increments on a successful claim", baseRequests + 1, metrics.getCurrentMetricValues().getMarkFileInProgressRequestCount()); } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java index cc5dc90aaf2..1b6fdf48b5f 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java @@ -28,9 +28,9 @@ *

    *
  • {@link MetricsReplicationLogDiscoveryReplayImpl}: endToEndReplayLag, pickupLag (histograms) * and roundsExceedingRoundTime (counter)
    • - *
  • {@link MetricsReplicationLogProcessorImpl}: mutationsReplayedCount (counter) and - * mutationsPerFile (histogram)
    • - *
  • {@link MetricsReplicationLogTrackerReplayImpl}: markFileInProgressCollisionCount + *
  • {@link MetricsReplicationLogProcessorImpl}: successfulFileMutationsReplayedCount (counter) + * and mutationsPerFile (histogram)
    • + *
  • {@link MetricsReplicationLogTrackerReplayImpl}: markFileInProgressRenameFailedCount * (counter)
    • *
* The sources are {@code static} and constructed once (mirroring @@ -78,17 +78,20 @@ public void testDiscoveryReplayLagAndSlowRoundMetrics() { } /** - * Metrics #3 (mutationsReplayedCount) and #4 (mutationsPerFile) live on the log processor source. - * The count accumulates the supplied delta; the per-file histogram reports its max. + * Metrics #3 (successfulFileMutationsReplayedCount) and #4 (mutationsPerFile) live on the log + * processor source. The count accumulates the supplied delta; the per-file histogram reports its + * max. */ @Test public void testProcessorMutationMetrics() { - // #3 mutationsReplayedCount counter accumulates by the supplied delta. - long baseMutations = PROCESSOR.getCurrentMetricValues().getMutationsReplayedCount(); - PROCESSOR.incrementMutationsReplayedCount(150L); - PROCESSOR.incrementMutationsReplayedCount(50L); - assertEquals("mutationsReplayedCount should accumulate the supplied deltas", - baseMutations + 200, PROCESSOR.getCurrentMetricValues().getMutationsReplayedCount()); + // #3 successfulFileMutationsReplayedCount counter accumulates by the supplied delta. + long baseMutations = + PROCESSOR.getCurrentMetricValues().getSuccessfulFileMutationsReplayedCount(); + PROCESSOR.incrementSuccessfulFileMutationsReplayedCount(150L); + PROCESSOR.incrementSuccessfulFileMutationsReplayedCount(50L); + assertEquals("successfulFileMutationsReplayedCount should accumulate the supplied deltas", + baseMutations + 200, + PROCESSOR.getCurrentMetricValues().getSuccessfulFileMutationsReplayedCount()); // #4 mutationsPerFile histogram reports the max per-file count. PROCESSOR.updateMutationsPerFile(150L); @@ -99,16 +102,18 @@ public void testProcessorMutationMetrics() { } /** - * Metric #5 (markFileInProgressCollisionCount) lives on the tracker source and increments once + * Metric #5 (markFileInProgressRenameFailedCount) lives on the tracker source and increments once * per call. */ @Test - public void testTrackerCollisionMetric() { - long baseCollisions = TRACKER.getCurrentMetricValues().getMarkFileInProgressCollisionCount(); - TRACKER.incrementMarkFileInProgressCollisionCount(); - TRACKER.incrementMarkFileInProgressCollisionCount(); - TRACKER.incrementMarkFileInProgressCollisionCount(); - assertEquals("markFileInProgressCollisionCount should increment once per call", - baseCollisions + 3, TRACKER.getCurrentMetricValues().getMarkFileInProgressCollisionCount()); + public void testTrackerRenameFailedMetric() { + long baseRenameFailures = + TRACKER.getCurrentMetricValues().getMarkFileInProgressRenameFailedCount(); + TRACKER.incrementMarkFileInProgressRenameFailedCount(); + TRACKER.incrementMarkFileInProgressRenameFailedCount(); + TRACKER.incrementMarkFileInProgressRenameFailedCount(); + assertEquals("markFileInProgressRenameFailedCount should increment once per call", + baseRenameFailures + 3, + TRACKER.getCurrentMetricValues().getMarkFileInProgressRenameFailedCount()); } } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayProcessFileTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayProcessFileTest.java new file mode 100644 index 00000000000..1896d85005c --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayProcessFileTest.java @@ -0,0 +1,198 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.replication.reader; + +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.Optional; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.util.EnvironmentEdge; +import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +import org.apache.phoenix.replication.ReplicationLogTracker; +import org.apache.phoenix.replication.ReplicationShardDirectoryManager; +import org.apache.phoenix.replication.metrics.MetricsReplicationLogDiscoveryReplay; +import org.junit.After; +import org.junit.Test; + +/** + * Recording-site tests for the replay lag metrics that drive the real + * {@link ReplicationLogDiscoveryReplay#processFile(Path, boolean)} rather than exercising the + * metric source in isolation. They cover: + *
    + *
  • the {@code getRoundEligibleTime} math the lag samples are anchored to, including a file whose + * creation timestamp lands exactly on a round boundary (owned by the earlier round);
    • + *
  • first-claim gating: pickup lag is recorded on the new-files path (firstClaim=true) and not on + * an in-progress reclaim (firstClaim=false);
    • + *
  • failure handling: a file that throws mid-replay still records pickup lag but never records an + * end-to-end lag sample.
    • + *
+ * The round duration is stubbed to 60s and the waiting buffer to 15%, giving a round of 60000ms and + * a buffer of 9000ms. A file created at 100000ms is owned by the round ending at 120000ms and thus + * becomes eligible at 120000 + 9000 = 129000ms; a file created exactly on the 120000ms boundary is + * owned by that same earlier round and is also eligible at 129000ms. + */ +public class ReplicationLogDiscoveryReplayProcessFileTest { + + private static final long ROUND_SECONDS = 60L; + /** Eligible instant for a file whose owning round ends at 120000ms: 120000 + 9000 buffer. */ + private static final long ELIGIBLE_MS = 129000L; + private static final Path FILE = new Path("/replication/inprogress/1_rs_uuid_135000.plog"); + + @After + public void tearDown() { + EnvironmentEdgeManager.reset(); + } + + /** + * First claim of a file whose creation timestamp is strictly inside a round: both pickup lag + * (rename - eligible) and end-to-end lag (now - eligible) are recorded. + */ + @Test + public void testFirstClaimRecordsPickupAndEndToEndLag() throws IOException { + ReplicationLogTracker tracker = mock(ReplicationLogTracker.class); + MetricsReplicationLogDiscoveryReplay metrics = mock(MetricsReplicationLogDiscoveryReplay.class); + when(tracker.getFileTimestamp(any())).thenReturn(100000L); + when(tracker.getRenameTimestamp(any())).thenReturn(Optional.of(135000L)); + injectNow(140000L); + + newReplay(tracker, metrics).processFile(FILE, true); + + // eligible = 129000; pickup = 135000 - 129000; endToEnd = 140000 - 129000. + verify(metrics).updatePickupLag(135000L - ELIGIBLE_MS); + verify(metrics).updateEndToEndReplayLag(140000L - ELIGIBLE_MS); + } + + /** + * A file created exactly on a round boundary is owned by the earlier round (inclusive round + * bounds), so its eligibility is the boundary itself plus the buffer, not a full round later. + * Regression guard for the off-by-one: the old formula anchored to 189000ms, which clamped this + * pickup lag to 0; the corrected formula anchors to 129000ms and records the real 1000ms. + */ + @Test + public void testFirstClaimOnRoundBoundaryAnchorsToOwningRound() throws IOException { + ReplicationLogTracker tracker = mock(ReplicationLogTracker.class); + MetricsReplicationLogDiscoveryReplay metrics = mock(MetricsReplicationLogDiscoveryReplay.class); + when(tracker.getFileTimestamp(any())).thenReturn(120000L); + when(tracker.getRenameTimestamp(any())).thenReturn(Optional.of(130000L)); + injectNow(140000L); + + newReplay(tracker, metrics).processFile(FILE, true); + + verify(metrics).updatePickupLag(130000L - ELIGIBLE_MS); + verify(metrics).updateEndToEndReplayLag(140000L - ELIGIBLE_MS); + } + + /** + * An in-progress reclaim (firstClaim=false) must not record pickup lag -- markInProgress + * re-stamps a fresh rename timestamp on every reclaim, so a pickup sample here would measure + * eligible -> latest-reclaim, not eligible -> first-pickup -- but end-to-end lag is still + * recorded. + */ + @Test + public void testReclaimDoesNotRecordPickupLag() throws IOException { + ReplicationLogTracker tracker = mock(ReplicationLogTracker.class); + MetricsReplicationLogDiscoveryReplay metrics = mock(MetricsReplicationLogDiscoveryReplay.class); + when(tracker.getFileTimestamp(any())).thenReturn(100000L); + injectNow(140000L); + + newReplay(tracker, metrics).processFile(FILE, false); + + verify(metrics, never()).updatePickupLag(anyLong()); + verify(metrics).updateEndToEndReplayLag(140000L - ELIGIBLE_MS); + } + + /** + * When replay throws, pickup lag has already been recorded (before the replay call) but the + * end-to-end lag sample -- which represents a completed replay -- is never recorded, and the + * failure propagates. + */ + @Test + public void testReplayFailureRecordsPickupButNotEndToEndLag() throws IOException { + ReplicationLogTracker tracker = mock(ReplicationLogTracker.class); + MetricsReplicationLogDiscoveryReplay metrics = mock(MetricsReplicationLogDiscoveryReplay.class); + when(tracker.getFileTimestamp(any())).thenReturn(100000L); + when(tracker.getRenameTimestamp(any())).thenReturn(Optional.of(135000L)); + injectNow(140000L); + TestReplay replay = newReplay(tracker, metrics); + replay.failReplay = true; + + assertThrows(IOException.class, () -> replay.processFile(FILE, true)); + + verify(metrics).updatePickupLag(135000L - ELIGIBLE_MS); + verify(metrics, never()).updateEndToEndReplayLag(anyLong()); + } + + private static void injectNow(long now) { + EnvironmentEdgeManager.injectEdge(new EnvironmentEdge() { + @Override + public long currentTime() { + return now; + } + }); + } + + private static TestReplay newReplay(ReplicationLogTracker tracker, + MetricsReplicationLogDiscoveryReplay metrics) { + ReplicationShardDirectoryManager shardManager = mock(ReplicationShardDirectoryManager.class); + when(tracker.getReplicationShardDirectoryManager()).thenReturn(shardManager); + when(shardManager.getReplicationRoundDurationSeconds()).thenReturn((int) ROUND_SECONDS); + return new TestReplay(tracker, metrics); + } + + /** + * Minimal subclass that stands in a mocked metrics source and short-circuits the actual log-file + * replay, so only the lag-recording logic in {@code processFile} is under test. Overriding + * {@link #getWaitingBufferPercentage()} to a constant keeps the base constructor off the (unset) + * Configuration, fixing the buffer at 15% of the round. + */ + private static final class TestReplay extends ReplicationLogDiscoveryReplay { + + private final MetricsReplicationLogDiscoveryReplay replayMetricsMock; + private boolean failReplay; + + TestReplay(ReplicationLogTracker tracker, + MetricsReplicationLogDiscoveryReplay replayMetricsMock) { + super(tracker); + this.replayMetricsMock = replayMetricsMock; + } + + @Override + public double getWaitingBufferPercentage() { + return 15.0; + } + + @Override + protected MetricsReplicationLogDiscoveryReplay getReplayMetrics() { + return replayMetricsMock; + } + + @Override + protected void replayLogFile(Path path) throws IOException { + if (failReplay) { + throw new IOException("injected replay failure"); + } + } + } +}