This is an automated email from the ASF dual-hosted git repository.
lewismc pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nutch.git
The following commit(s) were added to refs/heads/master by this push:
new 101328833 NUTCH-3162 Fetcher and Parser latency metrics to properly
merge data from all threads and tasks (#906)
101328833 is described below
commit 101328833f02e0ec59d1fb3772065a4c5fcf5ed8
Author: Lewis John McGibbney <[email protected]>
AuthorDate: Tue Jul 21 19:16:42 2026 -0700
NUTCH-3162 Fetcher and Parser latency metrics to properly merge data from
all threads and tasks (#906)
---
src/java/org/apache/nutch/fetcher/Fetcher.java | 68 ++++-
.../org/apache/nutch/fetcher/FetcherThread.java | 10 +-
.../org/apache/nutch/metrics/ErrorTracker.java | 29 ++-
.../org/apache/nutch/metrics/LatencyTracker.java | 145 +++++++++--
.../org/apache/nutch/metrics/NutchMetrics.java | 7 +
src/java/org/apache/nutch/parse/ParseSegment.java | 81 +++++-
.../apache/nutch/fetcher/TestFetcherReducer.java | 105 ++++++++
.../org/apache/nutch/metrics/LatencyTestUtil.java | 142 +++++++++++
.../apache/nutch/metrics/TestLatencyTracker.java | 284 +++++++++++++++++++++
.../org/apache/nutch/parse/TestParseSegment.java | 64 ++++-
.../apache/nutch/util/ReducerContextWrapper.java | 10 +
11 files changed, 893 insertions(+), 52 deletions(-)
diff --git a/src/java/org/apache/nutch/fetcher/Fetcher.java
b/src/java/org/apache/nutch/fetcher/Fetcher.java
index cffd024a0..acf8a5728 100644
--- a/src/java/org/apache/nutch/fetcher/Fetcher.java
+++ b/src/java/org/apache/nutch/fetcher/Fetcher.java
@@ -34,12 +34,14 @@ import org.apache.commons.lang3.time.StopWatch;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
@@ -47,9 +49,12 @@ import
org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
+
+import com.tdunning.math.stats.MergingDigest;
import org.apache.nutch.crawl.CrawlDatum;
import org.apache.nutch.crawl.NutchWritable;
import org.apache.nutch.metadata.Nutch;
+import org.apache.nutch.metrics.LatencyTracker;
import org.apache.nutch.metrics.NutchMetrics;
import org.apache.nutch.util.MimeUtil;
import org.apache.nutch.util.NutchConfiguration;
@@ -223,9 +228,9 @@ public class Fetcher extends NutchTool implements Tool {
setup(innerContext);
initCounters(innerContext);
+ LinkedList<FetcherThread> fetcherThreads = new LinkedList<>();
try {
Configuration conf = innerContext.getConfiguration();
- LinkedList<FetcherThread> fetcherThreads = new LinkedList<>();
FetchItemQueues fetchQueues = new FetchItemQueues(conf);
QueueFeeder feeder;
@@ -499,11 +504,70 @@ public class Fetcher extends NutchTool implements Tool {
} while (activeThreads.get() > 0);
LOG.info("-activeThreads={}", activeThreads);
} finally {
+ // Merge all thread latency trackers and emit once; emit TDigest for
reducer
+ LatencyTracker mergedLatencyTracker = new LatencyTracker(
+ NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_LATENCY);
+ for (FetcherThread fetcherThread : fetcherThreads) {
+ mergedLatencyTracker.merge(fetcherThread.getFetchLatencyTracker());
+ }
+ mergedLatencyTracker.emitCountAndSumOnly(innerContext);
+ byte[] digestBytes = mergedLatencyTracker.toBytes();
+ if (digestBytes.length > 0) {
+ innerContext.write(new Text(NutchMetrics.LATENCY_KEY),
+ new NutchWritable(new BytesWritable(digestBytes)));
+ }
cleanup(innerContext);
}
}
}
+ /**
+ * Reducer that passes through (url, datum) records and merges TDigests from
+ * map tasks to set job-level latency percentile counters.
+ */
+ public static class FetcherReducer extends
+ Reducer<Text, NutchWritable, Text, NutchWritable> {
+
+ private static final Text LATENCY_KEY = new Text(NutchMetrics.LATENCY_KEY);
+
+ @Override
+ public void reduce(Text key, Iterable<NutchWritable> values,
+ Context context) throws IOException, InterruptedException {
+ if (key.equals(LATENCY_KEY)) {
+ MergingDigest mergedDigest = null;
+ for (NutchWritable nutchWritable : values) {
+ if (nutchWritable.get() instanceof BytesWritable) {
+ BytesWritable digestBytesWritable = (BytesWritable)
nutchWritable.get();
+ byte[] digestBytes = digestBytesWritable.copyBytes();
+ if (digestBytes != null && digestBytes.length > 0) {
+ MergingDigest digest = LatencyTracker.fromBytes(digestBytes);
+ if (digest != null) {
+ if (mergedDigest == null) {
+ mergedDigest = digest;
+ } else {
+ mergedDigest.add(digest);
+ }
+ }
+ }
+ }
+ }
+ // Set only percentile counters; count_total and sum_ms are already
correct from task aggregation
+ if (mergedDigest != null) {
+ context.getCounter(NutchMetrics.GROUP_FETCHER,
+ NutchMetrics.FETCHER_LATENCY +
LatencyTracker.SUFFIX_P50_MS).setValue((long) mergedDigest.quantile(0.50));
+ context.getCounter(NutchMetrics.GROUP_FETCHER,
+ NutchMetrics.FETCHER_LATENCY +
LatencyTracker.SUFFIX_P95_MS).setValue((long) mergedDigest.quantile(0.95));
+ context.getCounter(NutchMetrics.GROUP_FETCHER,
+ NutchMetrics.FETCHER_LATENCY +
LatencyTracker.SUFFIX_P99_MS).setValue((long) mergedDigest.quantile(0.99));
+ }
+ return;
+ }
+ for (NutchWritable value : values) {
+ context.write(key, value);
+ }
+ }
+ }
+
public void fetch(Path segment, int threads) throws IOException,
InterruptedException, ClassNotFoundException {
@@ -566,6 +630,8 @@ public class Fetcher extends NutchTool implements Tool {
job.setInputFormatClass(InputFormat.class);
job.setJarByClass(Fetcher.class);
job.setMapperClass(Fetcher.FetcherRun.class);
+ job.setReducerClass(Fetcher.FetcherReducer.class);
+ job.setNumReduceTasks(1);
FileOutputFormat.setOutputPath(job, segment);
job.setOutputFormatClass(FetcherOutputFormat.class);
diff --git a/src/java/org/apache/nutch/fetcher/FetcherThread.java
b/src/java/org/apache/nutch/fetcher/FetcherThread.java
index 4a4810d8e..12712aeca 100644
--- a/src/java/org/apache/nutch/fetcher/FetcherThread.java
+++ b/src/java/org/apache/nutch/fetcher/FetcherThread.java
@@ -551,8 +551,6 @@ public class FetcherThread extends Thread {
if (fit != null) {
fetchQueues.finishFetchItem(fit);
}
- // Emit fetch latency metrics
- fetchLatencyTracker.emitCounters(context);
// Emit error metrics
errorTracker.emitCounters(context);
activeThreads.decrementAndGet(); // count threads
@@ -561,6 +559,14 @@ public class FetcherThread extends Thread {
}
}
+ /**
+ * Returns the fetch latency tracker for this thread so the mapper can merge
+ * all thread trackers and emit job-level percentiles.
+ */
+ public LatencyTracker getFetchLatencyTracker() {
+ return fetchLatencyTracker;
+ }
+
private Text handleRedirect(FetchItem fit, String newUrl,
boolean temp, String redirType)
throws MalformedURLException, URLFilterException, InterruptedException {
diff --git a/src/java/org/apache/nutch/metrics/ErrorTracker.java
b/src/java/org/apache/nutch/metrics/ErrorTracker.java
index 192107160..efc97a09f 100644
--- a/src/java/org/apache/nutch/metrics/ErrorTracker.java
+++ b/src/java/org/apache/nutch/metrics/ErrorTracker.java
@@ -34,10 +34,12 @@ import org.apache.hadoop.mapreduce.TaskInputOutputContext;
* based on exception type. It uses a bounded set of error categories to stay
within
* Hadoop's counter limits (~120 counters).
*
- * <p>Usage:
+ * <p><b>Usage in mapper/reducer or task threads:</b>
* <pre>
* // In mapper/reducer setup or thread initialization
* errorTracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER);
+ * // or with context for cached counters:
+ * errorTracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER, context);
*
* // When catching exceptions
* try {
@@ -49,11 +51,20 @@ import org.apache.hadoop.mapreduce.TaskInputOutputContext;
* // Or with manual categorization
* errorTracker.recordError(ErrorTracker.ErrorType.NETWORK);
*
- * // In cleanup - emit all error counters
+ * // In cleanup - emit all error counters to the job
* errorTracker.emitCounters(context);
* </pre>
*
- * <p>Emits the following counters:
+ * <p><b>Usage in driver/client code (no task context):</b>
+ * When used in a job driver or other code that does not run inside a
mapper/reducer,
+ * create an ErrorTracker with the single-argument constructor (counter group
only).
+ * Call {@link #recordError(Throwable)} or {@link
#recordError(ErrorTracker.ErrorType)}
+ * for consistent error categorization. Do <em>not</em> call {@link
#emitCounters(TaskInputOutputContext)};
+ * Hadoop counters can only be written from within a task, so counts remain
in-memory only.
+ * This allows the same categorization and logging pattern (e.g. with
LOG.error) as in
+ * tasks, without emitting to job counters.
+ *
+ * <p>Emits the following counters (when used inside a task and emitCounters
is called):
* <ul>
* <li>errors_total - total number of errors across all categories</li>
* <li>errors_network_total - network-related errors</li>
@@ -104,9 +115,15 @@ public class ErrorTracker {
/**
* Creates a new ErrorTracker for the specified counter group.
*
- * <p>This constructor creates an ErrorTracker without cached counters.
- * Call {@link #initCounters(TaskInputOutputContext)} in setup() to cache
- * counter references for better performance.
+ * <p>Use in mapper/reducer setup or thread initialization: call
+ * {@link #initCounters(TaskInputOutputContext)} in setup() to cache counter
+ * references, then {@link #emitCounters(TaskInputOutputContext)} in cleanup
to
+ * emit counts to the job.
+ *
+ * <p>Use in driver/client code (no task context): do not call initCounters
or
+ * emitCounters. Only {@link #recordError(Throwable)} and
+ * {@link #recordError(ErrorTracker.ErrorType)} are used; counts stay
in-memory
+ * for consistent categorization and logging (e.g. with LOG.error).
*
* @param group the Hadoop counter group name (e.g.,
NutchMetrics.GROUP_FETCHER)
*/
diff --git a/src/java/org/apache/nutch/metrics/LatencyTracker.java
b/src/java/org/apache/nutch/metrics/LatencyTracker.java
index 3777bb29e..5725b1ddb 100644
--- a/src/java/org/apache/nutch/metrics/LatencyTracker.java
+++ b/src/java/org/apache/nutch/metrics/LatencyTracker.java
@@ -16,31 +16,36 @@
*/
package org.apache.nutch.metrics;
+import java.nio.ByteBuffer;
+
import org.apache.hadoop.mapreduce.TaskInputOutputContext;
+import com.tdunning.math.stats.MergingDigest;
import com.tdunning.math.stats.TDigest;
/**
* A utility class for tracking latency metrics using TDigest for percentile
* calculation.
- *
- * <p>This class wraps a TDigest data structure to collect latency samples and
+ *
+ * <p>This class wraps a MergingDigest data structure to collect latency
samples and
* emit Hadoop counters with count, sum, and percentile values (p50, p95, p99).
- *
+ * MergingDigest supports merging digests from multiple tasks for job-level
percentile
+ * computation.
+ *
* <p>Usage:
* <pre>
* // In mapper/reducer setup
* latencyTracker = new LatencyTracker(NutchMetrics.GROUP_FETCHER,
NutchMetrics.FETCHER_LATENCY);
- *
+ *
* // During processing
* long start = System.currentTimeMillis();
* // ... operation ...
* latencyTracker.record(System.currentTimeMillis() - start);
- *
+ *
* // In cleanup
* latencyTracker.emitCounters(context);
* </pre>
- *
+ *
* <p>Emits the following counters:
* <ul>
* <li>{prefix}_count_total - total number of samples</li>
@@ -49,7 +54,7 @@ import com.tdunning.math.stats.TDigest;
* <li>{prefix}_p95_ms - 95th percentile latency</li>
* <li>{prefix}_p99_ms - 99th percentile latency</li>
* </ul>
- *
+ *
* @since 1.22
*/
public class LatencyTracker {
@@ -57,7 +62,18 @@ public class LatencyTracker {
/** Default compression factor for TDigest (controls accuracy vs memory). */
private static final double DEFAULT_COMPRESSION = 100.0;
- private final TDigest digest;
+ /** Counter name suffix for total sample count. */
+ public static final String SUFFIX_COUNT_TOTAL = "_count_total";
+ /** Counter name suffix for sum of latencies in milliseconds. */
+ public static final String SUFFIX_SUM_MS = "_sum_ms";
+ /** Counter name suffix for 50th percentile latency in milliseconds. */
+ public static final String SUFFIX_P50_MS = "_p50_ms";
+ /** Counter name suffix for 95th percentile latency in milliseconds. */
+ public static final String SUFFIX_P95_MS = "_p95_ms";
+ /** Counter name suffix for 99th percentile latency in milliseconds. */
+ public static final String SUFFIX_P99_MS = "_p99_ms";
+
+ private final MergingDigest digest;
private final String group;
private final String prefix;
private long count = 0;
@@ -65,19 +81,19 @@ public class LatencyTracker {
/**
* Creates a new LatencyTracker.
- *
+ *
* @param group the Hadoop counter group name
* @param prefix the prefix for counter names (e.g., "fetch_latency")
*/
public LatencyTracker(String group, String prefix) {
- this.digest = TDigest.createDigest(DEFAULT_COMPRESSION);
+ this.digest = (MergingDigest)
TDigest.createMergingDigest(DEFAULT_COMPRESSION);
this.group = group;
this.prefix = prefix;
}
/**
* Records a latency sample.
- *
+ *
* @param latencyMs the latency in milliseconds
*/
public void record(long latencyMs) {
@@ -86,9 +102,24 @@ public class LatencyTracker {
sum += latencyMs;
}
+ /**
+ * Merges another LatencyTracker's digest and aggregates count/sum into this
one.
+ * Used to combine per-thread or per-task metrics before emitting or
serializing.
+ *
+ * @param other the other tracker to merge in (not modified)
+ */
+ public void merge(LatencyTracker other) {
+ if (other == null || other.count == 0) {
+ return;
+ }
+ digest.add(other.digest);
+ count += other.count;
+ sum += other.sum;
+ }
+
/**
* Returns the number of recorded samples.
- *
+ *
* @return the count of recorded latency samples
*/
public long getCount() {
@@ -97,7 +128,7 @@ public class LatencyTracker {
/**
* Returns the sum of all recorded latencies.
- *
+ *
* @return the sum of latencies in milliseconds
*/
public long getSum() {
@@ -106,7 +137,7 @@ public class LatencyTracker {
/**
* Returns the percentile value for the given quantile.
- *
+ *
* @param quantile the quantile (0.0 to 1.0)
* @return the percentile value in milliseconds
*/
@@ -117,28 +148,88 @@ public class LatencyTracker {
return (long) digest.quantile(quantile);
}
+ /**
+ * Serializes the digest to bytes for transmission to a reducer or side file.
+ * Returns an empty array if no samples have been recorded.
+ *
+ * @return serialized digest bytes, or empty array if count is 0
+ */
+ public byte[] toBytes() {
+ if (count == 0) {
+ return new byte[0];
+ }
+ ByteBuffer buf = ByteBuffer.allocate(digest.smallByteSize());
+ digest.asSmallBytes(buf);
+ return buf.array();
+ }
+
+ /**
+ * Deserializes a MergingDigest from bytes (as produced by {@link
#toBytes()}).
+ *
+ * @param bytes serialized digest bytes
+ * @return MergingDigest instance, or null if bytes is null or empty
+ */
+ public static MergingDigest fromBytes(byte[] bytes) {
+ if (bytes == null || bytes.length == 0) {
+ return null;
+ }
+ return MergingDigest.fromBytes(ByteBuffer.wrap(bytes));
+ }
+
+ /**
+ * Emits only count and sum counters (not percentiles). Use in mappers when
+ * a reducer will merge TDigests and set job-level percentile counters.
+ */
+ public void emitCountAndSumOnly(TaskInputOutputContext<?, ?, ?, ?> context) {
+ context.getCounter(group, prefix + SUFFIX_COUNT_TOTAL).setValue(count);
+ context.getCounter(group, prefix + SUFFIX_SUM_MS).setValue(sum);
+ }
+
/**
* Emits all latency counters to the Hadoop context.
- *
+ *
* <p>Should be called once during cleanup to emit aggregated metrics.
- *
+ *
* @param context the Hadoop task context
*/
public void emitCounters(TaskInputOutputContext<?, ?, ?, ?> context) {
- context.getCounter(group, prefix + "_count_total").setValue(count);
- context.getCounter(group, prefix + "_sum_ms").setValue(sum);
-
+ context.getCounter(group, prefix + SUFFIX_COUNT_TOTAL).setValue(count);
+ context.getCounter(group, prefix + SUFFIX_SUM_MS).setValue(sum);
+
if (count > 0) {
- context.getCounter(group, prefix + "_p50_ms").setValue((long)
digest.quantile(0.50));
- context.getCounter(group, prefix + "_p95_ms").setValue((long)
digest.quantile(0.95));
- context.getCounter(group, prefix + "_p99_ms").setValue((long)
digest.quantile(0.99));
+ context.getCounter(group, prefix + SUFFIX_P50_MS).setValue((long)
digest.quantile(0.50));
+ context.getCounter(group, prefix + SUFFIX_P95_MS).setValue((long)
digest.quantile(0.95));
+ context.getCounter(group, prefix + SUFFIX_P99_MS).setValue((long)
digest.quantile(0.99));
} else {
// Set to 0 if no samples recorded
- context.getCounter(group, prefix + "_p50_ms").setValue(0);
- context.getCounter(group, prefix + "_p95_ms").setValue(0);
- context.getCounter(group, prefix + "_p99_ms").setValue(0);
+ context.getCounter(group, prefix + SUFFIX_P50_MS).setValue(0);
+ context.getCounter(group, prefix + SUFFIX_P95_MS).setValue(0);
+ context.getCounter(group, prefix + SUFFIX_P99_MS).setValue(0);
}
}
-}
-
+ /**
+ * Sets job-level percentile counters from a merged digest (e.g. in a reducer
+ * that merged TDigests from all tasks). Uses the same counter names as
+ * {@link #emitCounters(TaskInputOutputContext)}.
+ *
+ * @param context the Hadoop task context
+ * @param mergedCount total count from merged digest
+ * @param mergedSum total sum from merged digest
+ * @param mergedDigest the merged MergingDigest (may be null if mergedCount
is 0)
+ */
+ public static void setJobLevelCounters(TaskInputOutputContext<?, ?, ?, ?>
context,
+ String group, String prefix, long mergedCount, long mergedSum,
MergingDigest mergedDigest) {
+ context.getCounter(group, prefix +
SUFFIX_COUNT_TOTAL).setValue(mergedCount);
+ context.getCounter(group, prefix + SUFFIX_SUM_MS).setValue(mergedSum);
+ if (mergedCount > 0 && mergedDigest != null) {
+ context.getCounter(group, prefix + SUFFIX_P50_MS).setValue((long)
mergedDigest.quantile(0.50));
+ context.getCounter(group, prefix + SUFFIX_P95_MS).setValue((long)
mergedDigest.quantile(0.95));
+ context.getCounter(group, prefix + SUFFIX_P99_MS).setValue((long)
mergedDigest.quantile(0.99));
+ } else {
+ context.getCounter(group, prefix + SUFFIX_P50_MS).setValue(0);
+ context.getCounter(group, prefix + SUFFIX_P95_MS).setValue(0);
+ context.getCounter(group, prefix + SUFFIX_P99_MS).setValue(0);
+ }
+ }
+}
diff --git a/src/java/org/apache/nutch/metrics/NutchMetrics.java
b/src/java/org/apache/nutch/metrics/NutchMetrics.java
index 32767c04e..620b65b6b 100644
--- a/src/java/org/apache/nutch/metrics/NutchMetrics.java
+++ b/src/java/org/apache/nutch/metrics/NutchMetrics.java
@@ -375,6 +375,13 @@ public final class NutchMetrics {
*/
public static final String INDEXER_LATENCY = "index_latency";
+ /**
+ * Special key used in map output to send serialized TDigest bytes to the
+ * reducer for job-level percentile merge. Reducers detect this key and merge
+ * digests instead of writing to output.
+ */
+ public static final String LATENCY_KEY = "__LATENCY__";
+
// =========================================================================
// Common Error Counter Names (used with component-specific groups)
// These constants are shared across all components for consistent error
diff --git a/src/java/org/apache/nutch/parse/ParseSegment.java
b/src/java/org/apache/nutch/parse/ParseSegment.java
index e31830dea..944d1c073 100644
--- a/src/java/org/apache/nutch/parse/ParseSegment.java
+++ b/src/java/org/apache/nutch/parse/ParseSegment.java
@@ -20,6 +20,7 @@ import org.apache.commons.lang3.time.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.nutch.crawl.CrawlDatum;
+import org.apache.nutch.crawl.NutchWritable;
import org.apache.nutch.crawl.SignatureFactory;
import org.apache.nutch.segment.SegmentChecker;
import org.apache.nutch.util.NutchConfiguration;
@@ -46,9 +47,10 @@ import org.apache.nutch.scoring.ScoringFilterException;
import org.apache.nutch.scoring.ScoringFilters;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.Partitioner;
import java.io.File;
import java.io.IOException;
@@ -83,7 +85,7 @@ public class ParseSegment extends NutchTool implements Tool {
}
public static class ParseSegmentMapper extends
- Mapper<WritableComparable<?>, Content, Text, ParseImpl> {
+ Mapper<WritableComparable<?>, Content, Text, NutchWritable> {
private ParseUtil parseUtil;
private Text newKey = new Text();
@@ -94,7 +96,7 @@ public class ParseSegment extends NutchTool implements Tool {
private ErrorTracker errorTracker;
@Override
- public void setup(Mapper<WritableComparable<?>, Content, Text,
ParseImpl>.Context context) {
+ public void setup(Mapper<WritableComparable<?>, Content, Text,
NutchWritable>.Context context) {
Configuration conf = context.getConfiguration();
scfilters = new ScoringFilters(conf);
skipTruncated = conf.getBoolean(SKIP_TRUNCATED, true);
@@ -106,15 +108,19 @@ public class ParseSegment extends NutchTool implements
Tool {
}
@Override
- public void cleanup(Mapper<WritableComparable<?>, Content, Text,
ParseImpl>.Context context)
+ public void cleanup(Mapper<WritableComparable<?>, Content, Text,
NutchWritable>.Context context)
throws IOException, InterruptedException {
- // Emit parse latency metrics
- parseLatencyTracker.emitCounters(context);
+ parseLatencyTracker.emitCountAndSumOnly(context);
+ byte[] digestBytes = parseLatencyTracker.toBytes();
+ if (digestBytes.length > 0) {
+ context.write(new Text(NutchMetrics.LATENCY_KEY),
+ new NutchWritable(new BytesWritable(digestBytes)));
+ }
}
@Override
public void map(WritableComparable<?> key, Content content,
- Context context)
+ Mapper<WritableComparable<?>, Content, Text, NutchWritable>.Context
context)
throws IOException, InterruptedException {
// convert on the fly from old UTF8 keys
if (key instanceof Text) {
@@ -191,8 +197,8 @@ public class ParseSegment extends NutchTool implements Tool
{
context.write(
url,
- new ParseImpl(new ParseText(parse.getText()), parse.getData(),
parse
- .isCanonical()));
+ new NutchWritable(new ParseImpl(new ParseText(parse.getText()),
+ parse.getData(), parse.isCanonical())));
}
}
}
@@ -247,15 +253,60 @@ public class ParseSegment extends NutchTool implements
Tool {
return false;
}
+ /** Sends LATENCY_KEY to partition 0 so one reducer merges all TDigests. */
+ public static class ParseSegmentPartitioner extends Partitioner<Text,
NutchWritable> {
+ @Override
+ public int getPartition(Text key, NutchWritable value, int numPartitions) {
+ if (numPartitions <= 1) {
+ return 0;
+ }
+ if (key.toString().equals(NutchMetrics.LATENCY_KEY)) {
+ return 0;
+ }
+ return (key.hashCode() & Integer.MAX_VALUE) % numPartitions;
+ }
+ }
+
public static class ParseSegmentReducer extends
- Reducer<Text, Writable, Text, Writable> {
+ Reducer<Text, NutchWritable, Text, ParseImpl> {
+
+ private static final Text LATENCY_KEY = new Text(NutchMetrics.LATENCY_KEY);
@Override
- public void reduce(Text key, Iterable<Writable> values,
- Context context)
+ public void reduce(Text key, Iterable<NutchWritable> values,
+ Reducer<Text, NutchWritable, Text, ParseImpl>.Context context)
throws IOException, InterruptedException {
- Iterator<Writable> valuesIter = values.iterator();
- context.write(key, valuesIter.next()); // collect first value
+ if (key.equals(LATENCY_KEY)) {
+ com.tdunning.math.stats.MergingDigest merged = null;
+ for (NutchWritable w : values) {
+ if (w.get() instanceof BytesWritable) {
+ byte[] bytes = ((BytesWritable) w.get()).copyBytes();
+ if (bytes != null && bytes.length > 0) {
+ com.tdunning.math.stats.MergingDigest d =
LatencyTracker.fromBytes(bytes);
+ if (d != null) {
+ if (merged == null) {
+ merged = d;
+ } else {
+ merged.add(d);
+ }
+ }
+ }
+ }
+ }
+ if (merged != null) {
+ context.getCounter(NutchMetrics.GROUP_PARSER,
+ NutchMetrics.PARSER_LATENCY +
LatencyTracker.SUFFIX_P50_MS).setValue((long) merged.quantile(0.50));
+ context.getCounter(NutchMetrics.GROUP_PARSER,
+ NutchMetrics.PARSER_LATENCY +
LatencyTracker.SUFFIX_P95_MS).setValue((long) merged.quantile(0.95));
+ context.getCounter(NutchMetrics.GROUP_PARSER,
+ NutchMetrics.PARSER_LATENCY +
LatencyTracker.SUFFIX_P99_MS).setValue((long) merged.quantile(0.99));
+ }
+ return;
+ }
+ Iterator<NutchWritable> valuesIter = values.iterator();
+ if (valuesIter.hasNext()) {
+ context.write(key, (ParseImpl) valuesIter.next().get());
+ }
}
}
@@ -280,6 +331,8 @@ public class ParseSegment extends NutchTool implements Tool
{
job.setJarByClass(ParseSegment.class);
job.setMapperClass(ParseSegment.ParseSegmentMapper.class);
job.setReducerClass(ParseSegment.ParseSegmentReducer.class);
+ job.setPartitionerClass(ParseSegment.ParseSegmentPartitioner.class);
+ job.setMapOutputValueClass(NutchWritable.class);
FileOutputFormat.setOutputPath(job, segment);
job.setOutputFormatClass(ParseOutputFormat.class);
diff --git a/src/test/org/apache/nutch/fetcher/TestFetcherReducer.java
b/src/test/org/apache/nutch/fetcher/TestFetcherReducer.java
new file mode 100644
index 000000000..a1aa5f0ac
--- /dev/null
+++ b/src/test/org/apache/nutch/fetcher/TestFetcherReducer.java
@@ -0,0 +1,105 @@
+/*
+ * 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.nutch.fetcher;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.nutch.crawl.CrawlDatum;
+import org.apache.nutch.crawl.NutchWritable;
+import org.apache.nutch.metrics.NutchMetrics;
+import org.apache.nutch.metrics.LatencyTestUtil;
+import org.apache.nutch.util.NutchConfiguration;
+import org.apache.nutch.util.ReducerContextWrapper;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link Fetcher.FetcherReducer}: latency key branch (merge
TDigests,
+ * set job-level counters) and pass-through branch (write url, datum).
+ */
+class TestFetcherReducer {
+
+ @Test
+ void testReduceLatencyKeyMergesDigestsAndSetsCounters() throws IOException,
InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, NutchWritable> out = new HashMap<>();
+ Fetcher.FetcherReducer reducer = new Fetcher.FetcherReducer();
+ ReducerContextWrapper<Text, NutchWritable, Text, NutchWritable> wrapper =
+ new ReducerContextWrapper<>(reducer, conf, out);
+
+ byte[] digestBytes = LatencyTestUtil.createDigestBytes(100, 200, 300);
+ List<NutchWritable> values = new ArrayList<>();
+ values.add(new NutchWritable(new BytesWritable(digestBytes)));
+
+ reducer.reduce(new Text(NutchMetrics.LATENCY_KEY), values,
wrapper.getContext());
+
+ LatencyTestUtil.assertPercentilesInRange(wrapper.getCounters(),
+ NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_LATENCY, 100, 300);
+ assertEquals(0, out.size());
+ }
+
+ @Test
+ void testReduceLatencyKeyWithMultipleDigestsMergesAndSetsCounters() throws
IOException, InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, NutchWritable> out = new HashMap<>();
+ Fetcher.FetcherReducer reducer = new Fetcher.FetcherReducer();
+ ReducerContextWrapper<Text, NutchWritable, Text, NutchWritable> wrapper =
+ new ReducerContextWrapper<>(reducer, conf, out);
+
+ List<BytesWritable> digestWritables =
LatencyTestUtil.createDigestBytesWritables(
+ new long[] { 10 }, new long[] { 90 });
+ List<NutchWritable> values = new ArrayList<>();
+ for (BytesWritable bw : digestWritables) {
+ values.add(new NutchWritable(bw));
+ }
+
+ reducer.reduce(new Text(NutchMetrics.LATENCY_KEY), values,
wrapper.getContext());
+
+ LatencyTestUtil.assertPercentilesInRange(wrapper.getCounters(),
+ NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_LATENCY, 10, 90);
+ assertEquals(0, out.size());
+ }
+
+ @Test
+ void testReducePassThroughWritesKeyValue() throws IOException,
InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, NutchWritable> out = new HashMap<>();
+ Fetcher.FetcherReducer reducer = new Fetcher.FetcherReducer();
+ ReducerContextWrapper<Text, NutchWritable, Text, NutchWritable> wrapper =
+ new ReducerContextWrapper<>(reducer, conf, out);
+
+ Text url = new Text("http://example.com/";);
+ CrawlDatum datum = new CrawlDatum(CrawlDatum.STATUS_FETCH_SUCCESS, 0,
0.0f);
+ List<NutchWritable> values = Collections.singletonList(new
NutchWritable(datum));
+
+ reducer.reduce(url, values, wrapper.getContext());
+
+ assertEquals(1, out.size());
+ assertTrue(out.containsKey(url));
+ assertEquals(datum, out.get(url).get());
+ }
+}
diff --git a/src/test/org/apache/nutch/metrics/LatencyTestUtil.java
b/src/test/org/apache/nutch/metrics/LatencyTestUtil.java
new file mode 100644
index 000000000..093c8396e
--- /dev/null
+++ b/src/test/org/apache/nutch/metrics/LatencyTestUtil.java
@@ -0,0 +1,142 @@
+/*
+ * 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.nutch.metrics;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.mapreduce.Counters;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Test utility for latency-tracking tests. Reduces boilerplate when testing
+ * Fetcher, ParseSegment, and Indexer reducers that merge TDigests and set
+ * job-level percentile counters.
+ *
+ * <p>Use with the real Hadoop {@link Counters} from {@link
org.apache.nutch.util.ReducerContextWrapper#getCounters()}
+ * (no mocks).
+ */
+public final class LatencyTestUtil {
+
+ private static final String DUMMY_GROUP = "test";
+ private static final String DUMMY_PREFIX = "latency";
+
+ private LatencyTestUtil() {}
+
+ /**
+ * Builds serialized TDigest bytes from the given samples. Uses a temporary
+ * LatencyTracker with dummy group/prefix. Callers wrap the result as needed
+ * (e.g. {@code new BytesWritable(bytes)} or {@code new NutchWritable(new
BytesWritable(bytes))}).
+ *
+ * @param samples latency values in milliseconds to record
+ * @return serialized digest as from {@link LatencyTracker#toBytes()}
+ */
+ public static byte[] createDigestBytes(long... samples) {
+ LatencyTracker tracker = new LatencyTracker(DUMMY_GROUP, DUMMY_PREFIX);
+ for (long sample : samples) {
+ tracker.record(sample);
+ }
+ return tracker.toBytes();
+ }
+
+ /**
+ * Builds one BytesWritable per array of samples (e.g. one per map task).
+ * Useful for reducer tests that merge multiple digests.
+ *
+ * @param sampleArrays each array is recorded into one tracker and
serialized to one BytesWritable
+ * @return list of digest BytesWritable, in order
+ */
+ public static List<BytesWritable> createDigestBytesWritables(long[]...
sampleArrays) {
+ List<BytesWritable> list = new ArrayList<>(sampleArrays.length);
+ for (long[] samples : sampleArrays) {
+ list.add(new BytesWritable(createDigestBytes(samples)));
+ }
+ return list;
+ }
+
+ /**
+ * Asserts that the job-level percentile counters (p50, p95, p99) for the
+ * given group and prefix are in the range [minMs, maxMs]. Uses
+ * {@link LatencyTracker#SUFFIX_P50_MS} etc.
+ *
+ * @param counters counters from {@link
org.apache.nutch.util.ReducerContextWrapper#getCounters()}
+ * @param group counter group (e.g. {@link NutchMetrics#GROUP_FETCHER})
+ * @param prefix counter name prefix (e.g. {@link
NutchMetrics#FETCHER_LATENCY})
+ * @param minMs inclusive lower bound for all percentiles (ms)
+ * @param maxMs inclusive upper bound for all percentiles (ms)
+ */
+ public static void assertPercentilesInRange(Counters counters, String group,
String prefix,
+ long minMs, long maxMs) {
+ long p50 = counters.findCounter(group, prefix +
LatencyTracker.SUFFIX_P50_MS).getValue();
+ long p95 = counters.findCounter(group, prefix +
LatencyTracker.SUFFIX_P95_MS).getValue();
+ long p99 = counters.findCounter(group, prefix +
LatencyTracker.SUFFIX_P99_MS).getValue();
+ assertTrue(p50 >= minMs && p50 <= maxMs,
+ "p50=" + p50 + " not in [" + minMs + "," + maxMs + "]");
+ assertTrue(p95 >= minMs && p95 <= maxMs,
+ "p95=" + p95 + " not in [" + minMs + "," + maxMs + "]");
+ assertTrue(p99 >= minMs && p99 <= maxMs,
+ "p99=" + p99 + " not in [" + minMs + "," + maxMs + "]");
+ }
+
+ /**
+ * Asserts that the count and sum counters for the given group and prefix
+ * match the expected values. Uses {@link LatencyTracker#SUFFIX_COUNT_TOTAL}
+ * and {@link LatencyTracker#SUFFIX_SUM_MS}.
+ *
+ * @param counters counters from {@link
org.apache.nutch.util.ReducerContextWrapper#getCounters()}
+ * @param group counter group
+ * @param prefix counter name prefix
+ * @param expectedCount expected _count_total value
+ * @param expectedSumMs expected _sum_ms value
+ */
+ public static void assertCountAndSum(Counters counters, String group, String
prefix,
+ long expectedCount, long expectedSumMs) {
+ assertEquals(expectedCount,
+ counters.findCounter(group, prefix +
LatencyTracker.SUFFIX_COUNT_TOTAL).getValue());
+ assertEquals(expectedSumMs,
+ counters.findCounter(group, prefix +
LatencyTracker.SUFFIX_SUM_MS).getValue());
+ }
+
+ /**
+ * Asserts that the percentile counters (p50, p95, p99) for the given group
+ * and prefix are all zero. Useful for tests that emit with zero samples.
+ *
+ * @param counters counters from {@link
org.apache.nutch.util.ReducerContextWrapper#getCounters()}
+ * @param group counter group
+ * @param prefix counter name prefix
+ */
+ public static void assertPercentilesZero(Counters counters, String group,
String prefix) {
+ assertEquals(0, counters.findCounter(group, prefix +
LatencyTracker.SUFFIX_P50_MS).getValue());
+ assertEquals(0, counters.findCounter(group, prefix +
LatencyTracker.SUFFIX_P95_MS).getValue());
+ assertEquals(0, counters.findCounter(group, prefix +
LatencyTracker.SUFFIX_P99_MS).getValue());
+ }
+
+ /**
+ * Asserts that count, sum, and all percentile counters are zero.
+ *
+ * @param counters counters from {@link
org.apache.nutch.util.ReducerContextWrapper#getCounters()}
+ * @param group counter group
+ * @param prefix counter name prefix
+ */
+ public static void assertCountSumAndPercentilesZero(Counters counters,
String group, String prefix) {
+ assertCountAndSum(counters, group, prefix, 0, 0);
+ assertPercentilesZero(counters, group, prefix);
+ }
+}
diff --git a/src/test/org/apache/nutch/metrics/TestLatencyTracker.java
b/src/test/org/apache/nutch/metrics/TestLatencyTracker.java
new file mode 100644
index 000000000..863dc3ff3
--- /dev/null
+++ b/src/test/org/apache/nutch/metrics/TestLatencyTracker.java
@@ -0,0 +1,284 @@
+/*
+ * 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.nutch.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.nutch.util.NutchConfiguration;
+import org.apache.nutch.util.ReducerContextWrapper;
+import org.junit.jupiter.api.Test;
+
+import com.tdunning.math.stats.MergingDigest;
+import com.tdunning.math.stats.TDigest;
+
+/**
+ * Unit tests for {@link LatencyTracker} merge, serialization
(toBytes/fromBytes),
+ * percentile behavior, and counter emission. Counter-emitting tests use the
real
+ * Hadoop Context and Counters via {@link ReducerContextWrapper} (no mocks of
Hadoop).
+ */
+class TestLatencyTracker {
+
+ private static final String GROUP = "test";
+ private static final String PREFIX = "test_latency";
+
+ @Test
+ void testMergeAggregatesCountAndSum() {
+ LatencyTracker a = new LatencyTracker(GROUP, PREFIX);
+ a.record(10);
+ a.record(20);
+ LatencyTracker b = new LatencyTracker(GROUP, PREFIX);
+ b.record(30);
+ a.merge(b);
+ assertEquals(3, a.getCount());
+ assertEquals(60, a.getSum());
+ }
+
+ @Test
+ void testMergeNullOrEmptyIsNoOp() {
+ LatencyTracker a = new LatencyTracker(GROUP, PREFIX);
+ a.record(5);
+ long countBefore = a.getCount();
+ a.merge(null);
+ assertEquals(countBefore, a.getCount());
+ LatencyTracker empty = new LatencyTracker(GROUP, PREFIX);
+ a.merge(empty);
+ assertEquals(countBefore, a.getCount());
+ }
+
+ @Test
+ void testToBytesFromBytesRoundTrip() {
+ LatencyTracker tracker = new LatencyTracker(GROUP, PREFIX);
+ tracker.record(100);
+ tracker.record(200);
+ tracker.record(300);
+ byte[] bytes = tracker.toBytes();
+ assertNotNull(bytes);
+ assertEquals(3, tracker.getCount());
+ com.tdunning.math.stats.MergingDigest restored =
LatencyTracker.fromBytes(bytes);
+ assertNotNull(restored);
+ double q50 = restored.quantile(0.50);
+ assertTrue(q50 >= 100 && q50 <= 300);
+ }
+
+ @Test
+ void testToBytesEmptyReturnsEmptyArray() {
+ LatencyTracker tracker = new LatencyTracker(GROUP, PREFIX);
+ byte[] bytes = tracker.toBytes();
+ assertNotNull(bytes);
+ assertEquals(0, bytes.length);
+ }
+
+ @Test
+ void testFromBytesNullOrEmptyReturnsNull() {
+ assertNull(LatencyTracker.fromBytes(null));
+ assertNull(LatencyTracker.fromBytes(new byte[0]));
+ }
+
+ @Test
+ void testGetPercentileReturnsValueInRange() {
+ LatencyTracker tracker = new LatencyTracker(GROUP, PREFIX);
+ tracker.record(100);
+ tracker.record(200);
+ tracker.record(300);
+ long p50 = tracker.getPercentile(0.50);
+ long p95 = tracker.getPercentile(0.95);
+ long p99 = tracker.getPercentile(0.99);
+ assertTrue(p50 >= 100 && p50 <= 300);
+ assertTrue(p95 >= 100 && p95 <= 300);
+ assertTrue(p99 >= 100 && p99 <= 300);
+ }
+
+ @Test
+ void testGetPercentileWithZeroSamplesReturnsZero() {
+ LatencyTracker tracker = new LatencyTracker(GROUP, PREFIX);
+ assertEquals(0, tracker.getPercentile(0.50));
+ assertEquals(0, tracker.getPercentile(0.95));
+ }
+
+ // Integration-style tests: real Hadoop Context and Counters (no mocks).
+ // Uses ReducerContextWrapper to drive a reducer that emits latency counters.
+
+ @Test
+ void testEmitCountAndSumOnlyUpdatesJobCounters() throws IOException,
InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, Text> out = new HashMap<>();
+ EmitCountAndSumOnlyReducer reducer = new EmitCountAndSumOnlyReducer(GROUP,
PREFIX);
+ ReducerContextWrapper<Text, Text, Text, Text> wrapper =
+ new ReducerContextWrapper<>(reducer, conf, out);
+ reducer.reduce(new Text("k"), Collections.singletonList(new Text("v")),
wrapper.getContext());
+
+ LatencyTestUtil.assertCountAndSum(wrapper.getCounters(), GROUP, PREFIX, 2,
30);
+ }
+
+ @Test
+ void testEmitCountersUpdatesJobCounters() throws IOException,
InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, Text> out = new HashMap<>();
+ EmitCountersReducer reducer = new EmitCountersReducer(GROUP, PREFIX);
+ ReducerContextWrapper<Text, Text, Text, Text> wrapper =
+ new ReducerContextWrapper<>(reducer, conf, out);
+ reducer.reduce(new Text("k"), Collections.singletonList(new Text("v")),
wrapper.getContext());
+
+ LatencyTestUtil.assertCountAndSum(wrapper.getCounters(), GROUP, PREFIX, 3,
600);
+ LatencyTestUtil.assertPercentilesInRange(wrapper.getCounters(), GROUP,
PREFIX, 100, 300);
+ }
+
+ @Test
+ void testEmitCountersWithZeroSamplesSetsPercentilesToZero() throws
IOException, InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, Text> out = new HashMap<>();
+ EmitCountersZeroReducer reducer = new EmitCountersZeroReducer(GROUP,
PREFIX);
+ ReducerContextWrapper<Text, Text, Text, Text> wrapper =
+ new ReducerContextWrapper<>(reducer, conf, out);
+ reducer.reduce(new Text("k"), Collections.emptyList(),
wrapper.getContext());
+
+ LatencyTestUtil.assertCountSumAndPercentilesZero(wrapper.getCounters(),
GROUP, PREFIX);
+ }
+
+ @Test
+ void testSetJobLevelCountersUpdatesJobCounters() throws IOException,
InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, Text> out = new HashMap<>();
+ SetJobLevelCountersReducer reducer = new SetJobLevelCountersReducer(GROUP,
PREFIX);
+ ReducerContextWrapper<Text, Text, Text, Text> wrapper =
+ new ReducerContextWrapper<>(reducer, conf, out);
+ reducer.reduce(new Text("k"), Collections.singletonList(new Text("v")),
wrapper.getContext());
+
+ LatencyTestUtil.assertCountAndSum(wrapper.getCounters(), GROUP, PREFIX, 3,
600);
+ LatencyTestUtil.assertPercentilesInRange(wrapper.getCounters(), GROUP,
PREFIX, 100, 300);
+ }
+
+ @Test
+ void testSetJobLevelCountersWithZeroCountSetsPercentilesToZero() throws
IOException, InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, Text> out = new HashMap<>();
+ SetJobLevelCountersZeroReducer reducer = new
SetJobLevelCountersZeroReducer(GROUP, PREFIX);
+ ReducerContextWrapper<Text, Text, Text, Text> wrapper =
+ new ReducerContextWrapper<>(reducer, conf, out);
+ reducer.reduce(new Text("k"), Collections.emptyList(),
wrapper.getContext());
+
+ LatencyTestUtil.assertCountSumAndPercentilesZero(wrapper.getCounters(),
GROUP, PREFIX);
+ }
+
+ /** Reducer that emits only count and sum via LatencyTracker (real Context,
no mocks). */
+ private static final class EmitCountAndSumOnlyReducer extends Reducer<Text,
Text, Text, Text> {
+ private final String group;
+ private final String prefix;
+
+ EmitCountAndSumOnlyReducer(String group, String prefix) {
+ this.group = group;
+ this.prefix = prefix;
+ }
+
+ @Override
+ protected void reduce(Text key, Iterable<Text> values, Context context)
+ throws IOException, InterruptedException {
+ LatencyTracker tracker = new LatencyTracker(group, prefix);
+ tracker.record(10);
+ tracker.record(20);
+ tracker.emitCountAndSumOnly(context);
+ }
+ }
+
+ /** Reducer that emits count, sum, and percentiles via LatencyTracker (real
Context, no mocks). */
+ private static final class EmitCountersReducer extends Reducer<Text, Text,
Text, Text> {
+ private final String group;
+ private final String prefix;
+
+ EmitCountersReducer(String group, String prefix) {
+ this.group = group;
+ this.prefix = prefix;
+ }
+
+ @Override
+ protected void reduce(Text key, Iterable<Text> values, Context context)
+ throws IOException, InterruptedException {
+ LatencyTracker tracker = new LatencyTracker(group, prefix);
+ tracker.record(100);
+ tracker.record(200);
+ tracker.record(300);
+ tracker.emitCounters(context);
+ }
+ }
+
+ /** Reducer that emits counters with zero samples (percentiles set to 0). */
+ private static final class EmitCountersZeroReducer extends Reducer<Text,
Text, Text, Text> {
+ private final String group;
+ private final String prefix;
+
+ EmitCountersZeroReducer(String group, String prefix) {
+ this.group = group;
+ this.prefix = prefix;
+ }
+
+ @Override
+ protected void reduce(Text key, Iterable<Text> values, Context context)
+ throws IOException, InterruptedException {
+ LatencyTracker tracker = new LatencyTracker(group, prefix);
+ tracker.emitCounters(context);
+ }
+ }
+
+ /** Reducer that calls setJobLevelCounters with a merged digest (real
Context, no mocks). */
+ private static final class SetJobLevelCountersReducer extends Reducer<Text,
Text, Text, Text> {
+ private final String group;
+ private final String prefix;
+
+ SetJobLevelCountersReducer(String group, String prefix) {
+ this.group = group;
+ this.prefix = prefix;
+ }
+
+ @Override
+ protected void reduce(Text key, Iterable<Text> values, Context context)
+ throws IOException, InterruptedException {
+ MergingDigest digest = (MergingDigest)
TDigest.createMergingDigest(100.0);
+ digest.add(100);
+ digest.add(200);
+ digest.add(300);
+ LatencyTracker.setJobLevelCounters(context, group, prefix, 3, 600,
digest);
+ }
+ }
+
+ /** Reducer that calls setJobLevelCounters with zero count (percentiles set
to 0). */
+ private static final class SetJobLevelCountersZeroReducer extends
Reducer<Text, Text, Text, Text> {
+ private final String group;
+ private final String prefix;
+
+ SetJobLevelCountersZeroReducer(String group, String prefix) {
+ this.group = group;
+ this.prefix = prefix;
+ }
+
+ @Override
+ protected void reduce(Text key, Iterable<Text> values, Context context)
+ throws IOException, InterruptedException {
+ LatencyTracker.setJobLevelCounters(context, group, prefix, 0, 0, null);
+ }
+ }
+}
diff --git a/src/test/org/apache/nutch/parse/TestParseSegment.java
b/src/test/org/apache/nutch/parse/TestParseSegment.java
index d989c7a0b..e166bc891 100644
--- a/src/test/org/apache/nutch/parse/TestParseSegment.java
+++ b/src/test/org/apache/nutch/parse/TestParseSegment.java
@@ -16,14 +16,29 @@
*/
package org.apache.nutch.parse;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.nutch.crawl.NutchWritable;
import org.apache.nutch.metadata.Metadata;
+import org.apache.nutch.metrics.LatencyTestUtil;
+import org.apache.nutch.metrics.NutchMetrics;
import org.apache.nutch.net.protocols.Response;
import org.apache.nutch.protocol.Content;
+import org.apache.nutch.util.NutchConfiguration;
+import org.apache.nutch.util.ReducerContextWrapper;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestParseSegment {
private static byte[] BYTES = "the quick brown
fox".getBytes(StandardCharsets.UTF_8);
@@ -77,4 +92,49 @@ public class TestParseSegment {
content.setContent(BYTES);
assertFalse(ParseSegment.isTruncated(content));
}
+
+ /**
+ * ParseSegmentReducer latency branch: when key is LATENCY_KEY, merges
+ * BytesWritable TDigests and sets job-level parser latency counters.
+ */
+ @Test
+ void testParseSegmentReducerLatencyKeySetsCounters() throws IOException,
InterruptedException {
+ Configuration conf = NutchConfiguration.create();
+ Map<Text, ParseImpl> out = new HashMap<>();
+ ParseSegment.ParseSegmentReducer reducer = new
ParseSegment.ParseSegmentReducer();
+ ReducerContextWrapper<Text, NutchWritable, Text, ParseImpl> wrapper = new
ReducerContextWrapper<>(
+ reducer, conf, out);
+
+ byte[] digestBytes = LatencyTestUtil.createDigestBytes(100, 200);
+ List<NutchWritable> values = new ArrayList<>();
+ values.add(new NutchWritable(new BytesWritable(digestBytes)));
+
+ reducer.reduce(new Text(NutchMetrics.LATENCY_KEY), values,
+ wrapper.getContext());
+
+ LatencyTestUtil.assertPercentilesInRange(wrapper.getCounters(),
+ NutchMetrics.GROUP_PARSER, NutchMetrics.PARSER_LATENCY, 100, 200);
+ assertEquals(0, out.size());
+ }
+
+ /**
+ * ParseSegmentPartitioner sends LATENCY_KEY to partition 0 so one reducer
+ * merges all TDigests.
+ */
+ @Test
+ void testParseSegmentPartitionerSendsLatencyKeyToPartitionZero() {
+ ParseSegment.ParseSegmentPartitioner partitioner = new
ParseSegment.ParseSegmentPartitioner();
+ int numPartitions = 4;
+ assertEquals(0, partitioner.getPartition(new
Text(NutchMetrics.LATENCY_KEY),
+ new NutchWritable(new BytesWritable()), numPartitions));
+ }
+
+ @Test
+ void testParseSegmentPartitionerWithSinglePartition() {
+ ParseSegment.ParseSegmentPartitioner partitioner = new
ParseSegment.ParseSegmentPartitioner();
+ assertEquals(0, partitioner.getPartition(new
Text(NutchMetrics.LATENCY_KEY),
+ new NutchWritable(new BytesWritable()), 1));
+ assertEquals(0, partitioner.getPartition(new Text("http://example.com/";),
+ new NutchWritable(new BytesWritable()), 1));
+ }
}
diff --git a/src/test/org/apache/nutch/util/ReducerContextWrapper.java
b/src/test/org/apache/nutch/util/ReducerContextWrapper.java
index ec683e466..8fddf5857 100644
--- a/src/test/org/apache/nutch/util/ReducerContextWrapper.java
+++ b/src/test/org/apache/nutch/util/ReducerContextWrapper.java
@@ -71,6 +71,16 @@ public class ReducerContextWrapper<KEYIN, VALUEIN, KEYOUT,
VALUEOUT> {
return context;
}
+ /**
+ * Return the underlying counters updated by the context, for assertions in
tests.
+ * Uses the real Hadoop Counters API (no mocks).
+ *
+ * @return the counters instance
+ */
+ public Counters getCounters() {
+ return counters;
+ }
+
@SuppressWarnings("unchecked")
private void initContext() {
context = Mockito.mock(Reducer.Context.class,
