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..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 @@ -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 @@ -438,13 +439,16 @@ 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; LOG.info("Finished new files processing for round: {} in {}ms for haGroup: {}", replicationRound, duration, haGroupName); getMetrics().updateTimeToProcessNewFiles(duration); + if (duration > roundTimeMills) { + getMetrics().incrementRoundsExceedingRoundTime(); + } } /** @@ -468,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); @@ -494,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) { @@ -518,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(); @@ -696,11 +711,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 +727,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/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 00bfa4ed685..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,6 +384,11 @@ protected Optional markInProgress(final Path file) { return Optional.of(newPath); } else { LOG.warn("Failed to rename file for in-progress marking: {}", file); + // 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/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..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,6 +41,13 @@ 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 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)"; /** * Increments the counter for failed mutations. This counter tracks the number of mutations that @@ -78,6 +85,22 @@ public interface MetricsReplicationLogProcessor extends BaseSource { */ void updateLogFileReplayTime(long timeMs); + /** + * 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 incrementSuccessfulFileMutationsReplayedCount(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..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,6 +33,8 @@ public class MetricsReplicationLogProcessorImpl extends BaseSourceImpl private final MutableFastCounter logFileReplaySuccessCount; private final MutableHistogram batchReplayTime; private final MutableHistogram logFileReplayTime; + private final MutableFastCounter successfulFileMutationsReplayedCount; + 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); + 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); } @Override @@ -86,6 +92,16 @@ public void updateLogFileReplayTime(long timeMs) { logFileReplayTime.add(timeMs); } + @Override + public void incrementSuccessfulFileMutationsReplayedCount(long delta) { + successfulFileMutationsReplayedCount.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(), + 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 dcc2af2a5db..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,6 +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_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"; @@ -52,6 +57,14 @@ public interface MetricsReplicationLogTracker extends BaseSource { */ void 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 incrementMarkFileInProgressRenameFailedCount(); + /** * 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..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,6 +28,7 @@ public class MetricsReplicationLogTrackerImpl extends BaseSourceImpl protected String groupMetricsContext; protected final MutableFastCounter markFileInProgressRequestCount; + protected final MutableFastCounter markFileInProgressRenameFailedCount; protected final MutableFastCounter markFileCompletedRequestCount; protected final MutableFastCounter markFileFailedRequestCount; protected final MutableFastCounter markFileCompletedRequestFailedCount; @@ -40,6 +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); + 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, @@ -59,6 +63,11 @@ public void incrementMarkFileInProgressRequestCount() { markFileInProgressRequestCount.incr(); } + @Override + public void incrementMarkFileInProgressRenameFailedCount() { + markFileInProgressRenameFailedCount.incr(); + } + @Override public void incrementMarkFileCompletedRequestCount() { markFileCompletedRequestCount.incr(); @@ -100,7 +109,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(), + markFileInProgressRenameFailedCount.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..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,15 +25,19 @@ public class ReplicationLogProcessorMetricValues { private final long logFileReplaySuccessCount; private final long logFileReplayTime; private final long logFileBatchReplayTime; + private final long successfulFileMutationsReplayedCount; + private final long mutationsPerFile; public ReplicationLogProcessorMetricValues(long failedMutationsCount, long logFileReplayFailureCount, long logFileReplaySuccessCount, long logFileReplayTime, - long logFileBatchReplayTime) { + long logFileBatchReplayTime, long successfulFileMutationsReplayedCount, long mutationsPerFile) { this.failedMutationsCount = failedMutationsCount; this.logFileReplayFailureCount = logFileReplayFailureCount; this.logFileReplaySuccessCount = logFileReplaySuccessCount; this.logFileReplayTime = logFileReplayTime; this.logFileBatchReplayTime = logFileBatchReplayTime; + this.successfulFileMutationsReplayedCount = successfulFileMutationsReplayedCount; + this.mutationsPerFile = mutationsPerFile; } public long getFailedMutationsCount() { @@ -55,4 +59,12 @@ public long getLogFileReplayTime() { public long getLogFileBatchReplayTime() { return logFileBatchReplayTime; } + + public long getSuccessfulFileMutationsReplayedCount() { + return successfulFileMutationsReplayedCount; + } + + 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..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,11 +29,13 @@ public class ReplicationLogTrackerMetricValues { private final long markFileInProgressTimeMs; private final long markFileCompletedTimeMs; private final long markFileFailedTimeMs; + private final long markFileInProgressRenameFailedCount; public ReplicationLogTrackerMetricValues(long markFileInProgressRequestCount, long markFileCompletedRequestCount, long markFileFailedRequestCount, long markFileCompletedRequestFailedCount, long markFileInProgressTimeMs, - long markFileCompletedTimeMs, long markFileFailedTimeMs) { + long markFileCompletedTimeMs, long markFileFailedTimeMs, + long markFileInProgressRenameFailedCount) { 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.markFileInProgressRenameFailedCount = markFileInProgressRenameFailedCount; } public long getMarkFileInProgressRequestCount() { @@ -71,4 +74,8 @@ public long getMarkFileFailedTimeMs() { return markFileFailedTimeMs; } + 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 a3ae1ca1f79..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,12 +228,64 @@ 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) 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) { + long roundStart = (creationTs / roundTimeMills) * roundTimeMills; + long owningRoundEnd = (creationTs == roundStart) ? creationTs : roundStart + roundTimeMills; + return owningRoundEnd + bufferMillis; + } + /** * Initializes lastRoundProcessed and lastRoundInSync based on the persisted HA group state. *

    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..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,6 +278,15 @@ 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 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); + } } 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/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 db9ae36fc23..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 @@ -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. @@ -1086,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); @@ -1106,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 @@ -1176,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)); @@ -1198,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) @@ -1377,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 @@ -1407,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 @@ -1417,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 @@ -1471,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(); @@ -1483,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)); @@ -1492,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()); } /** @@ -1522,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)); @@ -1588,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(); @@ -1602,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)) @@ -1665,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)) @@ -2685,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 1f2fff31585..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 @@ -844,6 +844,62 @@ 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 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. + 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("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()); + } + + @Test + public void testMarkInProgressNoCollisionOnRenameSuccess() throws IOException { + // Initialize tracker + tracker.init(); + + long baseRenameFailures = + metrics.getCurrentMetricValues().getMarkFileInProgressRenameFailedCount(); + 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("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()); + } + @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..1b6fdf48b5f --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/metrics/ReplicationReplayPerfMetricsTest.java @@ -0,0 +1,119 @@ +/* + * 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}: successfulFileMutationsReplayedCount (counter) + * and mutationsPerFile (histogram)
      • + *
    • {@link MetricsReplicationLogTrackerReplayImpl}: markFileInProgressRenameFailedCount + * (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 (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 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); + PROCESSOR.updateMutationsPerFile(400L); + PROCESSOR.updateMutationsPerFile(30L); + assertEquals("mutationsPerFile should report the max per-file count", 400L, + PROCESSOR.getCurrentMetricValues().getMutationsPerFile()); + } + + /** + * Metric #5 (markFileInProgressRenameFailedCount) lives on the tracker source and increments once + * per call. + */ + @Test + 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"); + } + } + } +}