This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new 90f648b86d PHOENIX-7992 Add replay-side performance metrics for HA
replication (#2607)
90f648b86d is described below
commit 90f648b86d857ee958c702adf3943d961234dad0
Author: Himanshu Gwalani <[email protected]>
AuthorDate: Thu Sep 10 21:12:33 2026 +0530
PHOENIX-7992 Add replay-side performance metrics for HA replication (#2607)
---
.../replication/ReplicationLogDiscovery.java | 60 ++++--
.../ReplicationLogDiscoveryForwarder.java | 8 +-
.../phoenix/replication/ReplicationLogTracker.java | 5 +
.../metrics/MetricsReplicationLogDiscovery.java | 11 +
.../MetricsReplicationLogDiscoveryImpl.java | 10 +-
.../MetricsReplicationLogDiscoveryReplay.java | 35 +++
.../MetricsReplicationLogDiscoveryReplayImpl.java | 28 +++
.../metrics/MetricsReplicationLogProcessor.java | 23 ++
.../MetricsReplicationLogProcessorImpl.java | 19 +-
.../metrics/MetricsReplicationLogTracker.java | 15 ++
.../metrics/MetricsReplicationLogTrackerImpl.java | 12 +-
.../ReplicationLogDiscoveryMetricValues.java | 8 +-
.../ReplicationLogProcessorMetricValues.java | 14 +-
.../metrics/ReplicationLogTrackerMetricValues.java | 9 +-
.../reader/ReplicationLogDiscoveryReplay.java | 77 ++++++-
.../reader/ReplicationLogProcessor.java | 9 +
.../reader/ReplicationLogProcessorTestIT.java | 75 +++++++
.../ReplicationLogDiscoveryForwarderTest.java | 12 +-
.../replication/ReplicationLogDiscoveryTest.java | 172 +++++++++++----
.../replication/ReplicationLogTrackerTest.java | 56 +++++
.../metrics/ReplicationReplayPerfMetricsTest.java | 119 +++++++++++
...plicationLogDiscoveryReplayProcessFileTest.java | 236 +++++++++++++++++++++
22 files changed, 932 insertions(+), 81 deletions(-)
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 21e1117b87..e499d16db4 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
@@ -47,7 +47,7 @@ import
org.apache.phoenix.thirdparty.com.google.common.util.concurrent.ThreadFac
* service Round-based Processing: - Files are organized into replication
rounds based on timestamps
* - Each round represents a time window (e.g., 1 minute) of replication
activity - Processing waits
* for round completion + buffer time before processing to ensure all files
are available Subclasses
- * must implement: - processFile(Path): Defines how individual replication log
files are processed -
+ * must implement: - processFile(Path, boolean): Defines how individual
replication log files are processed -
* createMetricsSource(): Provides metrics tracking for monitoring -
Configuration methods: Thread
* counts, intervals, probabilities, etc. File Processing Flow: 1. Discover
new files for the
* current round 2. Mark files as in-progress (move to in-progress directory)
3. Process each file
@@ -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 @@ public abstract class ReplicationLogDiscovery {
* 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 @@ public abstract class ReplicationLogDiscovery {
List<Path> 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 @@ public abstract class ReplicationLogDiscovery {
replicationLogTracker.getInProgressLogSubDirectoryName(),
renameTimestampThreshold,
files.size(), haGroupName);
while (!files.isEmpty() && isRunning()) {
- Optional<Path> failedFile = processOneRandomFile(files);
+ Optional<Path> 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 @@ public abstract class ReplicationLogDiscovery {
/**
* 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<Path> processOneRandomFile(final List<Path> files) throws
IOException {
+ private Optional<Path> processOneRandomFile(final List<Path> files, final
boolean firstClaim)
+ throws IOException {
// Pick a random file and process it
Path file = files.get(ThreadLocalRandom.current().nextInt(files.size()));
Optional<Path> 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 @@ public abstract class ReplicationLogDiscovery {
/**
* 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 abstract class ReplicationLogDiscovery {
/**
* 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
- * <em>before</em> 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.
+ * <em>before</em> 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 abstract class ReplicationLogDiscovery {
} 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 097760bb8f..82a5347d93 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 class ReplicationLogDiscoveryForwarder extends
ReplicationLogDiscovery {
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 00bfa4ed68..0a13a13ac9 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 @@ public class ReplicationLogTracker {
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 73371bb55b..2480c69910 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,14 @@ public interface MetricsReplicationLogDiscovery extends
BaseSource {
*/
void updateTimeToProcessInProgressFiles(long timeMs);
+ /**
+ * Increments the counter for rounds whose new-file processing time exceeded
the round time. This
+ * counter lives on the shared discovery base, so it is emitted by both the
forwarder-side and
+ * replay-side discovery sources; a rising rate signals that that side's
new-file processing 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 35183d92b3..ccd9a1bd11 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 class MetricsReplicationLogDiscoveryImpl extends
BaseSourceImpl
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 class MetricsReplicationLogDiscoveryImpl extends
BaseSourceImpl
timeToProcessInProgressFiles.add(timeMs);
}
+ @Override
+ public void incrementRoundsExceedingRoundTime() {
+ roundsExceedingRoundTime.incr();
+ }
+
@Override
public void close() {
// Unregister this metrics source
@@ -75,7 +83,7 @@ public class MetricsReplicationLogDiscoveryImpl extends
BaseSourceImpl
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 f78b762146..cf64c045bc 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,16 @@ 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. Sampled only on
successful replay, "
+ + "so a file that keeps failing contributes no sample";
+ 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. Sampled only "
+ + "on a file's first claim";
/**
* Updates the consistency point metric. The consistency point represents
the timestamp up to
@@ -34,4 +44,29 @@ 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.
+ * <p>
+ * Recorded only after a file replays successfully, so it is
survivorship-biased: a file that
+ * keeps failing on every reclaim contributes no sample even though that is
the clearest replay
+ * stall. Read it alongside {@code logFileReplayFailureCount}, {@code
roundsExceedingRoundTime},
+ * and {@code consistencyPoint} staleness, which a healthy-looking lag
histogram would otherwise
+ * mask.
+ * @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).
+ * <p>
+ * Recorded only on a file's first claim (never on a reclaim), so like
+ * {@link #updateEndToEndReplayLag(long)} it cannot see a file that is
claimed once and then never
+ * replays; read it alongside the failure and staleness signals noted there.
+ * @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 dc51fdae26..339078242f 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
@@ -22,6 +22,9 @@ import static
org.apache.phoenix.metrics.MetricConstants.HA_GROUP_TAG_NAME;
import org.apache.hadoop.metrics2.lib.Interns;
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
@@ -33,6 +36,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,
@@ -41,10 +46,33 @@ public class MetricsReplicationLogDiscoveryReplayImpl
extends MetricsReplication
MetricsReplicationLogDiscoveryReplayImpl.METRICS_JMX_CONTEXT +
",haGroup=" + haGroupName);
getMetricsRegistry().tag(Interns.info(HA_GROUP_TAG_NAME,
HA_GROUP_TAG_DESC), 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 5ed7bd8e6f..f6169e459e 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 1d487efc4b..47c1fa3f7a 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
@@ -37,6 +37,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,
@@ -59,6 +61,10 @@ public class MetricsReplicationLogProcessorImpl extends
BaseSourceImpl
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
@@ -91,6 +97,16 @@ public class MetricsReplicationLogProcessorImpl extends
BaseSourceImpl
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
@@ -101,7 +117,8 @@ public class MetricsReplicationLogProcessorImpl extends
BaseSourceImpl
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 dcc2af2a5d..6bc7ce785e 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,12 @@ 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. Counts only the boolean rename()
== false outcome -- an "
+ + "IOException thrown during the rename is not counted here (that path
only logs an error). 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 +58,15 @@ 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. An
IOException thrown during the
+ * rename is not counted here (that path only logs an error and returns
empty). 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 b938a504d4..f83b181e0a 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 class MetricsReplicationLogTrackerImpl extends
BaseSourceImpl
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 class MetricsReplicationLogTrackerImpl extends
BaseSourceImpl
markFileInProgressRequestCount.incr();
}
+ @Override
+ public void incrementMarkFileInProgressRenameFailedCount() {
+ markFileInProgressRenameFailedCount.incr();
+ }
+
@Override
public void incrementMarkFileCompletedRequestCount() {
markFileCompletedRequestCount.incr();
@@ -100,7 +109,8 @@ public class MetricsReplicationLogTrackerImpl extends
BaseSourceImpl
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 dd380a3608..f6242257bd 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 class ReplicationLogDiscoveryMetricValues {
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 4911ac4bd7..8a0b276a0b 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 class ReplicationLogProcessorMetricValues {
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 76678db000..ddaa570536 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 class ReplicationLogTrackerMetricValues {
this.markFileInProgressTimeMs = markFileInProgressTimeMs;
this.markFileCompletedTimeMs = markFileCompletedTimeMs;
this.markFileFailedTimeMs = markFileFailedTimeMs;
+ this.markFileInProgressRenameFailedCount =
markFileInProgressRenameFailedCount;
}
public long getMarkFileInProgressRequestCount() {
@@ -71,4 +74,8 @@ public class ReplicationLogTrackerMetricValues {
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 a3ae1ca1f7..833326b6b2 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,85 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
}
@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();
+ final long fileTimestamp;
+ try {
+ fileTimestamp = tracker.getFileTimestamp(path);
+ } catch (NumberFormatException e) {
+ // A malformed file name cannot be anchored to a round. getFileTimestamp
is validated on the
+ // new-files path (getNewFilesForRound skips names that fail to parse)
but not on the reclaim
+ // path, so convert the unchecked parse failure into the IOException
that every other per-file
+ // failure uses. That keeps a single bad name isolated to
processOneRandomFile's catch (marked
+ // failed and retry-counted) instead of escaping as a RuntimeException
that aborts the whole
+ // in-progress sweep for the cycle.
+ throw new IOException("Cannot extract timestamp from replication log
file name: " + path, e);
+ }
+ long roundEligibleTime = getRoundEligibleTime(fileTimestamp);
+ // 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<Long> renameTs = tracker.getRenameTimestamp(path);
+ if (renameTs.isPresent()) {
+ getReplayMetrics().updatePickupLag(Math.max(0L, renameTs.get() -
roundEligibleTime));
+ } else {
+ // A first-claim file was just renamed into the in-progress directory
by markInProgress,
+ // which always stamps a rename timestamp, so this should not happen.
Drop the pickup-lag
+ // sample rather than record a bogus one, but leave a breadcrumb --
getRenameTimestamp only
+ // logs on a malformed 4-part name, not on an unexpectedly short name.
+ LOG.warn("No rename timestamp on first-claim file {}; skipping
pickup-lag sample", path);
+ }
+ }
+ 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.
Every successfully
+ // replayed file is sampled here -- including rotation-only /
zero-mutation files, which the
+ // mutationsPerFile histogram deliberately excludes -- so the two
distributions cover different
+ // file populations.
+ 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.
+ * <p>
+ * Rounds are half-open {@code [start, end)}: a file at {@code creationTs}
belongs to the round
+ * {@code [roundStart, roundStart + roundTimeMills)}, where {@code
roundStart} floors
+ * {@code creationTs} down to a round boundary. This matches the on-disk
layout, since
+ * getShardDirectory also floors the timestamp: a file that lands exactly on
a boundary is stored
+ * in, and replayed by, the round that starts there -- not the one that ends
there. The owning
+ * round's end is therefore always {@code roundStart + roundTimeMills},
including on a boundary.
+ * @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 =
replicationLogTracker.getReplicationShardDirectoryManager()
+ .getNearestRoundStartTimestamp(creationTs);
+ return roundStart + roundTimeMills + bufferMillis;
+ }
+
/**
* Initializes lastRoundProcessed and lastRoundInSync based on the persisted
HA group state.
* <ul>
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 414bd1bb63..423b2202bc 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 class ReplicationLogProcessor implements Closeable {
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/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
index 089bde4d47..6168d2692b 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
@@ -628,6 +628,18 @@ public class ReplicationLogProcessorTestIT extends
ParallelStatsDisabledIT {
metricValues.getFailedMutationsCount());
assertEquals("There must not be any failed files", 0,
metricValues.getLogFileReplayFailureCount());
+ // Recording-site guard (PHOENIX-7992):
incrementSuccessfulFileMutationsReplayedCount is
+ // written inside processLogFile before the catch, not by its callers,
so a real replay is the
+ // only thing that exercises it. The exact replayed-mutation count is
intentionally not
+ // asserted: generateHBaseMutations builds its Puts from a Phoenix
mutation state that is never
+ // committed or cleared between rows, so the replayed total is an
artifact of that helper rather
+ // than the "2 + 5" rows requested (testProcessLogFileBatchSizeLogic
pins an exact count on a
+ // 1:1 record model). We also do not assert the mutationsPerFile
histogram here:
+ // MutableHistogram.getMax() reads FastLongHistogram bins that the
metrics system resets to 0 on
+ // its periodic snapshot(), so a positive max is not stable to read
after a multi-second real
+ // replay -- only the counter (a MutableFastCounter, never reset by
snapshot) is reliable.
+ assertTrue("A successful multi-mutation file must record positive replay
throughput",
+ metricValues.getSuccessfulFileMutationsReplayedCount() > 0);
} finally {
replicationLogProcessor.close();
@@ -698,6 +710,13 @@ public class ReplicationLogProcessorTestIT extends
ParallelStatsDisabledIT {
metricValues.getFailedMutationsCount());
assertEquals("There must not be any failed files", 0,
metricValues.getLogFileReplayFailureCount());
+ // A rotation-only / zero-mutation file adds 0 to the throughput counter
and, because of the
+ // `totalProcessed > 0` guard, must not populate the per-file histogram
(its max stays 0
+ // rather than skewing the distribution toward zero).
+ assertEquals("Zero-mutation file must not add to the throughput
counter", 0,
+ metricValues.getSuccessfulFileMutationsReplayedCount());
+ assertEquals("Zero-mutation file must not populate the mutationsPerFile
histogram", 0,
+ metricValues.getMutationsPerFile());
} catch (Exception e) {
fail("Processing empty log file should not throw exception: " +
e.getMessage());
} finally {
@@ -970,11 +989,67 @@ public class ReplicationLogProcessorTestIT extends
ParallelStatsDisabledIT {
metricValues.getFailedMutationsCount());
assertEquals("There must not be any failed files", 0,
metricValues.getLogFileReplayFailureCount());
+ // Accumulation guard: totalProcessed sums across all three batch flushes
(4 + 5 + 2 = 11), and
+ // is recorded on the same line as updateMutationsPerFile, so the exact
counter value also pins
+ // the histogram's input. We assert the counter (a MutableFastCounter)
rather than
+ // getMutationsPerFile(): the histogram max is read from FastLongHistogram
bins that the metrics
+ // system resets to 0 on its periodic snapshot(), so a positive max is not
a stable value to
+ // assert against.
+ assertEquals("successfulFileMutationsReplayedCount must sum across
batches", 11,
+ metricValues.getSuccessfulFileMutationsReplayedCount());
// Clean up
spyProcessor.close();
}
+ /**
+ * Guards the placement of the throughput / per-file metric writes: they sit
after a whole-file
+ * success and before the catch, so a file that applies some batches and
then throws mid-replay
+ * must count zero mutations. The second batch flush is stubbed to throw
after the first has
+ * "applied", and the file must then book purely as a failure with no
throughput recorded.
+ */
+ @Test
+ public void testProcessLogFileMidReplayFailureDoesNotCountMutations() throws
Exception {
+ final Path filePath = new
Path(testFolder.newFile("testProcessLogFileMidReplayFailure").toURI());
+ final String tableNameString = "T_" + generateUniqueName();
+ LogFileWriter writer = initLogFileWriter(filePath);
+ // Six single-mutation records with batch size 3 -> two full flushes; the
second one throws.
+ for (int i = 0; i < 6; i++) {
+ Put put = new Put(Bytes.toBytes("row" + i));
+ put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("qual"),
Bytes.toBytes("v" + i));
+ writer.append(tableNameString, i, LogFileTestUtil.cellsOf(put));
+ }
+ writer.close();
+
+ Configuration testConf = new Configuration(conf);
+
testConf.setInt(ReplicationLogProcessor.REPLICATION_STANDBY_LOG_REPLAY_BATCH_SIZE,
3);
+ ReplicationLogProcessor spyProcessor =
+ Mockito.spy(new ReplicationLogProcessor(testConf, testHAGroupName));
+ // First flush applies (no-op); the second throws mid-replay after the
first already "applied".
+ Mockito.doNothing().doThrow(new IOException("injected mid-replay
failure")).when(spyProcessor)
+ .processReplicationLogBatch(Mockito.any(Map.class));
+ try {
+ spyProcessor.processLogFile(localFs, filePath);
+ fail("processLogFile must propagate the mid-replay failure");
+ } catch (IOException expected) {
+ // expected
+ }
+
+ ReplicationLogProcessorMetricValues metricValues =
+ spyProcessor.getMetrics().getCurrentMetricValues();
+ assertEquals("A mid-replay failure books the file as a failure", 1,
+ metricValues.getLogFileReplayFailureCount());
+ assertEquals("No success is recorded for a file that failed mid-replay", 0,
+ metricValues.getLogFileReplaySuccessCount());
+ // The first batch's mutations were applied, but the throughput write sits
after whole-file
+ // success and before the catch, so nothing is counted.
+ assertEquals("Mutations applied before a mid-replay failure must not be
counted", 0,
+ metricValues.getSuccessfulFileMutationsReplayedCount());
+ assertEquals("A failed file must not populate the mutationsPerFile
histogram", 0,
+ metricValues.getMutationsPerFile());
+ spyProcessor.close();
+ }
+
/**
* A single record's body is a flat cell stream that reconstructs into
multiple mutations of mixed
* type (e.g. a batch that upserts some rows and, on one of those rows, also
nulls a column -- a
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 725a2f0662..c49b1b0f58 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 class ReplicationLogDiscoveryForwarderTest extends
ReplicationLogBaseTest
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 class ReplicationLogDiscoveryForwarderTest extends
ReplicationLogBaseTest
}
}).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 class ReplicationLogDiscoveryForwarderTest extends
ReplicationLogBaseTest
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 class ReplicationLogDiscoveryForwarderTest extends
ReplicationLogBaseTest
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 class ReplicationLogDiscoveryForwarderTest extends
ReplicationLogBaseTest
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 db9ae36fc2..dfe2aa66d9 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 class ReplicationLogDiscoveryTest {
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 class ReplicationLogDiscoveryTest {
}
}
+ /**
+ * 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 class ReplicationLogDiscoveryTest {
// 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);
@@ -1105,15 +1162,20 @@ public class ReplicationLogDiscoveryTest {
Mockito.argThat(path ->
path.getName().startsWith(expectedFile.getName().split("\\.")[0])));
}
- // Verify that processFile was called for each file in the round
- Mockito.verify(discovery,
Mockito.times(5)).processFile(Mockito.any(Path.class));
+ // Verify that processFile was called for each file in the round, pinning
firstClaim=true. The
+ // new-files path is the first claim; pickup lag is recorded only there
(see
+ // ReplicationLogDiscoveryReplay#processFile). eq(true) guards the literal
in
+ // ReplicationLogDiscovery#processNewFilesForRound against a future swap.
+ Mockito.verify(discovery,
Mockito.times(5)).processFile(Mockito.any(Path.class),
+ Mockito.eq(true));
// 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 +1238,16 @@ public class ReplicationLogDiscoveryTest {
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 +1262,9 @@ public class ReplicationLogDiscoveryTest {
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 +1442,13 @@ public class ReplicationLogDiscoveryTest {
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
@@ -1406,8 +1473,11 @@ public class ReplicationLogDiscoveryTest {
}
// 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));
+ // failed once that would succeed in next retry), pinning
firstClaim=false. The in-progress
+ // path is a reclaim, so pickup lag is deliberately not recorded there;
eq(false) guards the
+ // literal in ReplicationLogDiscovery#processInProgressDirectory against
a future swap.
+ Mockito.verify(discovery,
Mockito.times(7)).processFile(Mockito.any(Path.class),
+ Mockito.eq(false));
// 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 +1487,8 @@ public class ReplicationLogDiscoveryTest {
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 +1542,9 @@ public class ReplicationLogDiscoveryTest {
// 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 +1555,8 @@ public class ReplicationLogDiscoveryTest {
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 +1565,19 @@ public class ReplicationLogDiscoveryTest {
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 +1598,26 @@ public class ReplicationLogDiscoveryTest {
// 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 +1668,8 @@ public class ReplicationLogDiscoveryTest {
// 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 +1683,15 @@ public class ReplicationLogDiscoveryTest {
.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 +1749,7 @@ public class ReplicationLogDiscoveryTest {
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 +2769,7 @@ public class ReplicationLogDiscoveryTest {
}
@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 1f2fff3158..81258ddc2e 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 class ReplicationLogTrackerTest {
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<Path> 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<Path> 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 0000000000..1b6fdf48b5
--- /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:
+ * <ul>
+ * <li>{@link MetricsReplicationLogDiscoveryReplayImpl}: endToEndReplayLag,
pickupLag (histograms)
+ * and roundsExceedingRoundTime (counter)</li>
+ * <li>{@link MetricsReplicationLogProcessorImpl}:
successfulFileMutationsReplayedCount (counter)
+ * and mutationsPerFile (histogram)</li>
+ * <li>{@link MetricsReplicationLogTrackerReplayImpl}:
markFileInProgressRenameFailedCount
+ * (counter)</li>
+ * </ul>
+ * 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 0000000000..1cefdf9340
--- /dev/null
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayProcessFileTest.java
@@ -0,0 +1,236 @@
+/*
+ * 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.assertFalse;
+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.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+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:
+ * <ul>
+ * <li>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 round
that starts there);</li>
+ * <li>first-claim gating: pickup lag is recorded on the new-files path
(firstClaim=true) and not on
+ * an in-progress reclaim (firstClaim=false);</li>
+ * <li>failure handling: a file that throws mid-replay still records pickup
lag but never records an
+ * end-to-end lag sample.</li>
+ * </ul>
+ * The round duration is configured to 60s and the waiting buffer to 15%,
giving a round of 60000ms
+ * and a buffer of 9000ms. Rounds are half-open [start, end): a file created
at 100000ms belongs to
+ * the round [60000, 120000) and becomes eligible at 120000 + 9000 = 129000ms;
a file created
+ * exactly on the 120000ms boundary belongs to the next round [120000, 180000)
and is eligible at
+ * 180000 + 9000 = 189000ms.
+ */
+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 belongs to the round that
starts at that boundary,
+ * not the one that ends there: rounds are half-open [start, end) and
getShardDirectory floors the
+ * timestamp, so the boundary file is stored in and replayed by the later
round. Its owning round
+ * is [120000, 180000), so it becomes eligible at 180000 + 9000 = 189000ms
-- a full round later
+ * than a file created just inside the previous round. Regression guard
against anchoring a
+ * boundary file to the round that ends there, which would under-count
eligibility by one round
+ * and over-report its lag.
+ */
+ @Test
+ public void testFirstClaimOnRoundBoundaryAnchorsToStartingRound() 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(190000L));
+ injectNow(200000L);
+
+ newReplay(tracker, metrics).processFile(FILE, true);
+
+ // Owning round [120000, 180000); eligible = 180000 + 9000 buffer = 189000.
+ long eligible = 189000L;
+ verify(metrics).updatePickupLag(190000L - eligible);
+ verify(metrics).updateEndToEndReplayLag(200000L - eligible);
+ }
+
+ /**
+ * 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());
+ }
+
+ /**
+ * A malformed in-progress file name (getFileTimestamp throws
NumberFormatException) must surface
+ * as an IOException -- the type processOneRandomFile catches and routes
through markFailed --
+ * rather than an unchecked exception that would escape the per-file handler
and abort the whole
+ * sweep. Nothing is replayed and no lag sample is recorded.
+ */
+ @Test
+ public void testMalformedFileNameSurfacesAsIOException() throws IOException {
+ ReplicationLogTracker tracker = mock(ReplicationLogTracker.class);
+ MetricsReplicationLogDiscoveryReplay metrics =
mock(MetricsReplicationLogDiscoveryReplay.class);
+ when(tracker.getFileTimestamp(any())).thenThrow(new
NumberFormatException("bad name"));
+ TestReplay replay = newReplay(tracker, metrics);
+
+ assertThrows(IOException.class, () -> replay.processFile(FILE, true));
+
+ assertFalse("replay must not run when the file name cannot be parsed",
replay.replayInvoked);
+ verify(metrics, never()).updatePickupLag(anyLong());
+ 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) {
+ // Use a real ReplicationShardDirectoryManager, not a mock: its
constructor only reads config
+ // and never touches the file system, so getRoundEligibleTime's round math
uses the real
+ // getNearestRoundStartTimestamp instead of a stubbed constant that would
otherwise re-encode
+ // the same floor formula this test verifies.
+ Configuration conf = new Configuration(false);
+
conf.setInt(ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY,
+ (int) ROUND_SECONDS);
+ ReplicationShardDirectoryManager shardManager =
+ new ReplicationShardDirectoryManager(conf, mock(FileSystem.class), new
Path("/replication"));
+
when(tracker.getReplicationShardDirectoryManager()).thenReturn(shardManager);
+ 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;
+ private boolean replayInvoked;
+
+ 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 {
+ replayInvoked = true;
+ if (failReplay) {
+ throw new IOException("injected replay failure");
+ }
+ }
+ }
+}
