cachedCounters;
+
+ /**
+ * Creates a new ErrorTracker for the specified counter group.
+ *
+ * This constructor creates an ErrorTracker without cached counters.
+ * Call {@link #initCounters(TaskInputOutputContext)} in setup() to cache
+ * counter references for better performance.
+ *
+ * @param group the Hadoop counter group name (e.g., NutchMetrics.GROUP_FETCHER)
+ */
+ public ErrorTracker(String group) {
+ this.group = group;
+ this.counts = new EnumMap<>(ErrorType.class);
+ this.cachedCounters = new EnumMap<>(ErrorType.class);
+ this.totalCount = new AtomicLong(0);
+
+ // Initialize all counts to 0
+ for (ErrorType type : ErrorType.values()) {
+ counts.put(type, new AtomicLong(0));
+ }
+ }
+
+ /**
+ * Creates a new ErrorTracker with cached counter references.
+ *
+ *
This constructor caches all counter references at creation time,
+ * avoiding repeated counter lookups in hot paths.
+ *
+ * @param group the Hadoop counter group name
+ * @param context the Hadoop task context for caching counters
+ */
+ public ErrorTracker(String group, TaskInputOutputContext context) {
+ this(group);
+ initCounters(context);
+ }
+
+ /**
+ * Initializes cached counter references from the Hadoop context.
+ *
+ *
Call this method in the mapper/reducer setup() method to cache
+ * counter references and avoid repeated lookups during processing.
+ *
+ * @param context the Hadoop task context
+ */
+ public void initCounters(TaskInputOutputContext context) {
+ cachedTotalCounter = context.getCounter(group, NutchMetrics.ERROR_TOTAL);
+ for (ErrorType type : ErrorType.values()) {
+ cachedCounters.put(type, context.getCounter(group, getCounterName(type)));
+ }
+ }
+
+ /**
+ * Records an error with automatic categorization based on the throwable type.
+ *
+ * @param t the throwable to categorize and record
+ */
+ public void recordError(Throwable t) {
+ recordError(categorize(t));
+ }
+
+ /**
+ * Records an error with explicit category.
+ *
+ * @param type the error type category
+ */
+ public void recordError(ErrorType type) {
+ counts.get(type).incrementAndGet();
+ totalCount.incrementAndGet();
+ }
+
+ /**
+ * Returns the count for a specific error type.
+ *
+ * @param type the error type
+ * @return the count for that error type
+ */
+ public long getCount(ErrorType type) {
+ return counts.get(type).get();
+ }
+
+ /**
+ * Returns the total count of all errors.
+ *
+ * @return the total error count
+ */
+ public long getTotalCount() {
+ return totalCount.get();
+ }
+
+ /**
+ * Emits all error counters to the Hadoop context.
+ *
+ *
Should be called once during cleanup to emit aggregated metrics.
+ * Only emits counters for error types that have non-zero counts.
+ *
+ *
If counters were cached via {@link #initCounters(TaskInputOutputContext)},
+ * uses the cached references for better performance.
+ *
+ * @param context the Hadoop task context
+ */
+ public void emitCounters(TaskInputOutputContext context) {
+ // Use cached counters if available, otherwise look up
+ if (cachedTotalCounter != null) {
+ cachedTotalCounter.increment(totalCount.get());
+ for (ErrorType type : ErrorType.values()) {
+ long count = counts.get(type).get();
+ if (count > 0) {
+ cachedCounters.get(type).increment(count);
+ }
+ }
+ } else {
+ // Fallback to direct lookup
+ context.getCounter(group, NutchMetrics.ERROR_TOTAL).increment(totalCount.get());
+ for (ErrorType type : ErrorType.values()) {
+ long count = counts.get(type).get();
+ if (count > 0) {
+ context.getCounter(group, getCounterName(type)).increment(count);
+ }
+ }
+ }
+ }
+
+ /**
+ * Directly increments cached error counters without local accumulation.
+ *
+ *
Use this method when you want to immediately update Hadoop counters
+ * rather than accumulating locally and emitting in cleanup.
+ * Requires {@link #initCounters(TaskInputOutputContext)} to have been called.
+ *
+ * @param t the throwable to categorize and count
+ * @throws IllegalStateException if counters have not been initialized
+ */
+ public void incrementCounters(Throwable t) {
+ incrementCounters(categorize(t));
+ }
+
+ /**
+ * Directly increments cached error counters without local accumulation.
+ *
+ *
Use this method when you want to immediately update Hadoop counters
+ * rather than accumulating locally and emitting in cleanup.
+ * Requires {@link #initCounters(TaskInputOutputContext)} to have been called.
+ *
+ * @param type the error type to count
+ * @throws IllegalStateException if counters have not been initialized
+ */
+ public void incrementCounters(ErrorType type) {
+ if (cachedTotalCounter == null) {
+ throw new IllegalStateException(
+ "Counters not initialized. Call initCounters() first.");
+ }
+ cachedTotalCounter.increment(1);
+ cachedCounters.get(type).increment(1);
+ }
+
+ /**
+ * Categorizes a throwable into an error type.
+ *
+ *
The categorization checks the exception class hierarchy to determine
+ * the most appropriate category. Timeout exceptions are checked first as
+ * they are a subclass of IOException.
+ *
+ * @param t the throwable to categorize
+ * @return the appropriate ErrorType for the throwable
+ */
+ public static ErrorType categorize(Throwable t) {
+ if (t == null) {
+ return ErrorType.OTHER;
+ }
+
+ String className = t.getClass().getName();
+
+ // Check for timeout first (before general IOException)
+ if (t instanceof SocketTimeoutException
+ || className.contains("TimeoutException")
+ || className.contains("Timeout")) {
+ return ErrorType.TIMEOUT;
+ }
+
+ // Network errors
+ if (t instanceof SocketException
+ || t instanceof UnknownHostException
+ || className.contains("ConnectException")
+ || className.contains("NoRouteToHostException")
+ || className.contains("ConnectionRefusedException")) {
+ return ErrorType.NETWORK;
+ }
+
+ // URL errors (check before general IOException since MalformedURLException extends IOException)
+ if (t instanceof MalformedURLException
+ || className.contains("URLFilterException")
+ || className.contains("URISyntaxException")) {
+ return ErrorType.URL;
+ }
+
+ // General IOException (but not the specific subtypes above)
+ if (t instanceof IOException) {
+ return ErrorType.NETWORK;
+ }
+
+ // Protocol errors
+ if (className.contains("ProtocolException")
+ || className.contains("ProtocolNotFound")) {
+ return ErrorType.PROTOCOL;
+ }
+
+ // Parsing errors
+ if (className.contains("ParseException")
+ || className.contains("ParserNotFound")
+ || className.contains("SAXException")
+ || className.contains("ParserConfigurationException")) {
+ return ErrorType.PARSING;
+ }
+
+ // Scoring errors
+ if (className.contains("ScoringFilterException")) {
+ return ErrorType.SCORING;
+ }
+
+ // Indexing errors
+ if (className.contains("IndexingException")) {
+ return ErrorType.INDEXING;
+ }
+
+ // Check cause chain for more specific categorization
+ Throwable cause = t.getCause();
+ if (cause != null && cause != t) {
+ ErrorType causeType = categorize(cause);
+ if (causeType != ErrorType.OTHER) {
+ return causeType;
+ }
+ }
+
+ return ErrorType.OTHER;
+ }
+
+ /**
+ * Gets the counter name constant for a given error type.
+ *
+ * @param type the error type
+ * @return the counter name constant from NutchMetrics
+ */
+ public static String getCounterName(ErrorType type) {
+ switch (type) {
+ case NETWORK:
+ return NutchMetrics.ERROR_NETWORK_TOTAL;
+ case PROTOCOL:
+ return NutchMetrics.ERROR_PROTOCOL_TOTAL;
+ case PARSING:
+ return NutchMetrics.ERROR_PARSING_TOTAL;
+ case URL:
+ return NutchMetrics.ERROR_URL_TOTAL;
+ case SCORING:
+ return NutchMetrics.ERROR_SCORING_TOTAL;
+ case INDEXING:
+ return NutchMetrics.ERROR_INDEXING_TOTAL;
+ case TIMEOUT:
+ return NutchMetrics.ERROR_TIMEOUT_TOTAL;
+ case OTHER:
+ default:
+ return NutchMetrics.ERROR_OTHER_TOTAL;
+ }
+ }
+
+ /**
+ * Gets the counter name for a throwable based on its categorization.
+ *
+ *
This is a convenience method for direct use in catch blocks:
+ *
+ * } catch (Exception e) {
+ * context.getCounter(group, ErrorTracker.getCounterName(e)).increment(1);
+ * }
+ *
+ *
+ * @param t the throwable to get the counter name for
+ * @return the counter name constant from NutchMetrics
+ */
+ public static String getCounterName(Throwable t) {
+ return getCounterName(categorize(t));
+ }
+}
diff --git a/src/java/org/apache/nutch/metrics/NutchMetrics.java b/src/java/org/apache/nutch/metrics/NutchMetrics.java
index dea34be7f6..c65a4f0ceb 100644
--- a/src/java/org/apache/nutch/metrics/NutchMetrics.java
+++ b/src/java/org/apache/nutch/metrics/NutchMetrics.java
@@ -145,9 +145,6 @@ private NutchMetrics() {
/** URLs rejected by URL filters. */
public static final String GENERATOR_URL_FILTERS_REJECTED_TOTAL = "url_filters_rejected_total";
- /** URL filter exceptions. */
- public static final String GENERATOR_URL_FILTER_EXCEPTION_TOTAL = "url_filter_exception_total";
-
/** URLs rejected by fetch schedule. */
public static final String GENERATOR_SCHEDULE_REJECTED_TOTAL = "schedule_rejected_total";
@@ -166,9 +163,6 @@ private NutchMetrics() {
/** URLs rejected due to fetch interval exceeding threshold. */
public static final String GENERATOR_INTERVAL_REJECTED_TOTAL = "interval_rejected_total";
- /** Malformed URLs encountered. */
- public static final String GENERATOR_MALFORMED_URL_TOTAL = "malformed_url_total";
-
/** URLs skipped due to per-host overflow. */
public static final String GENERATOR_URLS_SKIPPED_PER_HOST_OVERFLOW_TOTAL = "urls_skipped_per_host_overflow_total";
@@ -200,12 +194,6 @@ private NutchMetrics() {
/** Documents skipped by indexing filter. */
public static final String INDEXER_SKIPPED_BY_INDEXING_FILTER_TOTAL = "skipped_by_indexing_filter_total";
- /** Scoring filter errors. */
- public static final String INDEXER_ERRORS_SCORING_FILTER_TOTAL = "errors_scoring_filter_total";
-
- /** Indexing filter errors. */
- public static final String INDEXER_ERRORS_INDEXING_FILTER_TOTAL = "errors_indexing_filter_total";
-
/** Documents indexed (added or updated). */
public static final String INDEXER_INDEXED_TOTAL = "indexed_total";
@@ -248,9 +236,6 @@ private NutchMetrics() {
// HostDb Counters
// =========================================================================
- /** Malformed URLs in HostDb. */
- public static final String HOSTDB_MALFORMED_URL_TOTAL = "malformed_url_total";
-
/** Records filtered in HostDb. */
public static final String HOSTDB_FILTERED_RECORDS_TOTAL = "filtered_records_total";
@@ -346,15 +331,9 @@ private NutchMetrics() {
/** Omitted empty responses in WARC export. */
public static final String WARC_OMITTED_EMPTY_RESPONSE_TOTAL = "omitted_empty_response_total";
- /** Invalid URIs in WARC export. */
- public static final String WARC_INVALID_URI_TOTAL = "invalid_uri_total";
-
/** WARC records generated. */
public static final String WARC_RECORDS_GENERATED_TOTAL = "records_generated_total";
- /** Exceptions during WARC export. */
- public static final String WARC_EXCEPTION_TOTAL = "exception_total";
-
// =========================================================================
// Domain Statistics Counters (enum-based, kept for compatibility)
// =========================================================================
@@ -389,5 +368,65 @@ private NutchMetrics() {
* Used with {@link LatencyTracker} to emit indexing timing counters.
*/
public static final String INDEXER_LATENCY = "index_latency";
+
+ // =========================================================================
+ // Common Error Counter Names (used with component-specific groups)
+ // These constants are shared across all components for consistent error
+ // categorization. Use with ErrorTracker for automatic classification.
+ // =========================================================================
+
+ /**
+ * Total errors across all categories.
+ * This is incremented alongside any category-specific error counter.
+ */
+ public static final String ERROR_TOTAL = "errors_total";
+
+ /**
+ * Network-related errors.
+ * Includes: IOException, SocketException, ConnectException, UnknownHostException
+ */
+ public static final String ERROR_NETWORK_TOTAL = "errors_network_total";
+
+ /**
+ * Protocol errors.
+ * Includes: ProtocolException, ProtocolNotFound
+ */
+ public static final String ERROR_PROTOCOL_TOTAL = "errors_protocol_total";
+
+ /**
+ * Parsing errors.
+ * Includes: ParseException, ParserNotFound
+ */
+ public static final String ERROR_PARSING_TOTAL = "errors_parsing_total";
+
+ /**
+ * URL-related errors.
+ * Includes: MalformedURLException, URLFilterException
+ */
+ public static final String ERROR_URL_TOTAL = "errors_url_total";
+
+ /**
+ * Scoring filter errors.
+ * Includes: ScoringFilterException
+ */
+ public static final String ERROR_SCORING_TOTAL = "errors_scoring_total";
+
+ /**
+ * Indexing filter errors.
+ * Includes: IndexingException
+ */
+ public static final String ERROR_INDEXING_TOTAL = "errors_indexing_total";
+
+ /**
+ * Timeout errors.
+ * Includes: SocketTimeoutException, connection timeouts
+ */
+ public static final String ERROR_TIMEOUT_TOTAL = "errors_timeout_total";
+
+ /**
+ * Other uncategorized errors.
+ * Used as fallback for exceptions not matching any specific category.
+ */
+ public static final String ERROR_OTHER_TOTAL = "errors_other_total";
}
diff --git a/src/java/org/apache/nutch/parse/ParseSegment.java b/src/java/org/apache/nutch/parse/ParseSegment.java
index a7fbe066ce..0b2a6f2290 100644
--- a/src/java/org/apache/nutch/parse/ParseSegment.java
+++ b/src/java/org/apache/nutch/parse/ParseSegment.java
@@ -37,6 +37,7 @@
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.nutch.metadata.Metadata;
import org.apache.nutch.metadata.Nutch;
+import org.apache.nutch.metrics.ErrorTracker;
import org.apache.nutch.metrics.LatencyTracker;
import org.apache.nutch.metrics.NutchMetrics;
import org.apache.nutch.net.protocols.Response;
@@ -83,6 +84,7 @@ public static class ParseSegmentMapper extends
private ScoringFilters scfilters;
private boolean skipTruncated;
private LatencyTracker parseLatencyTracker;
+ private ErrorTracker errorTracker;
@Override
public void setup(Mapper, Content, Text, ParseImpl>.Context context) {
@@ -91,6 +93,8 @@ public void setup(Mapper, Content, Text, ParseImpl>.Contex
skipTruncated = conf.getBoolean(SKIP_TRUNCATED, true);
parseLatencyTracker = new LatencyTracker(
NutchMetrics.GROUP_PARSER, NutchMetrics.PARSER_LATENCY);
+ // Initialize error tracker with cached counters
+ errorTracker = new ErrorTracker(NutchMetrics.GROUP_PARSER, context);
}
@Override
@@ -133,6 +137,7 @@ public void map(WritableComparable key, Content content,
parseResult = parseUtil.parse(content);
} catch (Exception e) {
LOG.warn("Error parsing: {}: {}", key, StringUtils.stringifyException(e));
+ errorTracker.incrementCounters(e);
return;
}
@@ -164,6 +169,7 @@ public void map(WritableComparable key, Content content,
scfilters.passScoreAfterParsing(url, content, parse);
} catch (ScoringFilterException e) {
LOG.warn("Error passing score: {}: {}", url, e.getMessage());
+ errorTracker.incrementCounters(ErrorTracker.ErrorType.SCORING);
}
long end = System.currentTimeMillis();
diff --git a/src/java/org/apache/nutch/tools/warc/WARCExporter.java b/src/java/org/apache/nutch/tools/warc/WARCExporter.java
index 96e8c5a974..f271adfe94 100644
--- a/src/java/org/apache/nutch/tools/warc/WARCExporter.java
+++ b/src/java/org/apache/nutch/tools/warc/WARCExporter.java
@@ -58,6 +58,7 @@
import org.apache.nutch.parse.ParseText;
import org.apache.nutch.protocol.Content;
import org.apache.nutch.tools.WARCUtils;
+import org.apache.nutch.metrics.ErrorTracker;
import org.apache.nutch.metrics.NutchMetrics;
import org.apache.nutch.util.HadoopFSUtil;
import org.apache.nutch.util.NutchConfiguration;
@@ -117,9 +118,8 @@ public static class WARCReducer
private Counter missingContentCounter;
private Counter missingMetadataCounter;
private Counter omittedEmptyResponseCounter;
- private Counter invalidUriCounter;
private Counter recordsGeneratedCounter;
- private Counter exceptionCounter;
+ private ErrorTracker errorTracker;
@Override
public void setup(Context context) {
@@ -130,12 +130,10 @@ public void setup(Context context) {
NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_MISSING_METADATA_TOTAL);
omittedEmptyResponseCounter = context.getCounter(
NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_OMITTED_EMPTY_RESPONSE_TOTAL);
- invalidUriCounter = context.getCounter(
- NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_INVALID_URI_TOTAL);
recordsGeneratedCounter = context.getCounter(
NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_RECORDS_GENERATED_TOTAL);
- exceptionCounter = context.getCounter(
- NutchMetrics.GROUP_WARC_EXPORTER, NutchMetrics.WARC_EXCEPTION_TOTAL);
+ // Initialize error tracker with cached counters
+ errorTracker = new ErrorTracker(NutchMetrics.GROUP_WARC_EXPORTER, context);
}
@Override
@@ -263,7 +261,7 @@ public void reduce(Text key, Iterable values,
.append(uri.toASCIIString()).append(CRLF);
} catch (Exception e) {
LOG.error("Invalid URI {} ", key);
- invalidUriCounter.increment(1);
+ errorTracker.incrementCounters(e);
return;
}
@@ -300,7 +298,7 @@ public void reduce(Text key, Iterable values,
LOG.error(
"Exception when generating WARC resource record for {} : {}", key,
exception.getMessage());
- exceptionCounter.increment(1);
+ errorTracker.incrementCounters(exception);
}
// Do we need to emit a metadata record too?
@@ -342,7 +340,7 @@ public void reduce(Text key, Iterable values,
.append(uri.toASCIIString()).append(CRLF);
} catch (Exception e) {
LOG.error("Invalid URI {} ", key);
- invalidUriCounter.increment(1);
+ errorTracker.incrementCounters(e);
return;
}
@@ -363,7 +361,7 @@ public void reduce(Text key, Iterable values,
LOG.error(
"Exception when generating WARC metadata record for {} : {}",
key, exception.getMessage(), exception);
- exceptionCounter.increment(1);
+ errorTracker.incrementCounters(exception);
}
}
@@ -401,7 +399,7 @@ public void reduce(Text key, Iterable values,
.append(uri.toASCIIString()).append(CRLF);
} catch (Exception e) {
LOG.error("Invalid URI {} ", key);
- invalidUriCounter.increment(1);
+ errorTracker.incrementCounters(e);
return;
}
@@ -422,7 +420,7 @@ public void reduce(Text key, Iterable values,
LOG.error(
"Exception when generating WARC metadata record for {} : {}",
key, exception.getMessage(), exception);
- exceptionCounter.increment(1);
+ errorTracker.incrementCounters(exception);
}
}
}
diff --git a/src/java/org/apache/nutch/util/SitemapProcessor.java b/src/java/org/apache/nutch/util/SitemapProcessor.java
index a0378ec63d..4b55a72ebb 100644
--- a/src/java/org/apache/nutch/util/SitemapProcessor.java
+++ b/src/java/org/apache/nutch/util/SitemapProcessor.java
@@ -46,6 +46,7 @@
import org.apache.hadoop.util.ToolRunner;
import org.apache.nutch.crawl.CrawlDatum;
import org.apache.nutch.hostdb.HostDatum;
+import org.apache.nutch.metrics.ErrorTracker;
import org.apache.nutch.metrics.NutchMetrics;
import org.apache.nutch.net.URLFilters;
import org.apache.nutch.net.URLNormalizers;
@@ -121,6 +122,7 @@ private static class SitemapMapper extends Mapper mockContext;
+
+ @Mock
+ private Counter mockCounter;
+
+ @BeforeEach
+ void setUp() {
+ // Configure mock context to return mock counter for any counter request
+ lenient().when(mockContext.getCounter(anyString(), anyString())).thenReturn(mockCounter);
+ }
+
+ // =========================================================================
+ // Network Error Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeNetworkErrors() {
+ // Test IOException
+ assertEquals(ErrorType.NETWORK,
+ ErrorTracker.categorize(new IOException("Connection failed")));
+
+ // Test SocketException
+ assertEquals(ErrorType.NETWORK,
+ ErrorTracker.categorize(new SocketException("Socket closed")));
+
+ // Test UnknownHostException
+ assertEquals(ErrorType.NETWORK,
+ ErrorTracker.categorize(new UnknownHostException("example.com")));
+
+ // Test ConnectException
+ assertEquals(ErrorType.NETWORK,
+ ErrorTracker.categorize(new ConnectException("Connection refused")));
+ }
+
+ // =========================================================================
+ // Timeout Error Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeTimeoutErrors() {
+ // Test SocketTimeoutException
+ assertEquals(ErrorType.TIMEOUT,
+ ErrorTracker.categorize(new SocketTimeoutException("Read timed out")));
+ }
+
+ @Test
+ public void testCategorizeTimeoutByClassName() {
+ // Test custom exception with "Timeout" in class name
+ // The categorize method checks className.contains("Timeout")
+ Exception customTimeout = new CustomTimeoutException("Custom timeout");
+ assertEquals(ErrorType.TIMEOUT, ErrorTracker.categorize(customTimeout));
+ }
+
+ // Custom exception class for testing class name-based detection
+ private static class CustomTimeoutException extends Exception {
+ CustomTimeoutException(String message) {
+ super(message);
+ }
+ }
+
+ // =========================================================================
+ // URL Error Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeUrlErrors() {
+ // Test MalformedURLException
+ assertEquals(ErrorType.URL,
+ ErrorTracker.categorize(new MalformedURLException("Invalid URL")));
+
+ // Test URISyntaxException
+ assertEquals(ErrorType.URL,
+ ErrorTracker.categorize(new URISyntaxException("bad uri", "Invalid syntax")));
+ }
+
+ @Test
+ public void testCategorizeUrlFilterException() {
+ // Test URLFilterException (Nutch-specific)
+ assertEquals(ErrorType.URL,
+ ErrorTracker.categorize(new URLFilterException("URL filtered")));
+ }
+
+ // =========================================================================
+ // Protocol Error Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeProtocolErrors() {
+ // Test ProtocolException (Nutch-specific)
+ assertEquals(ErrorType.PROTOCOL,
+ ErrorTracker.categorize(new ProtocolException("Protocol error")));
+
+ // Test ProtocolNotFound (Nutch-specific)
+ assertEquals(ErrorType.PROTOCOL,
+ ErrorTracker.categorize(new ProtocolNotFound("ftp")));
+ }
+
+ // =========================================================================
+ // Parsing Error Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeParsingErrors() {
+ // Test ParseException (Nutch-specific)
+ assertEquals(ErrorType.PARSING,
+ ErrorTracker.categorize(new ParseException("Parse failed")));
+
+ // Test ParserNotFound (Nutch-specific)
+ assertEquals(ErrorType.PARSING,
+ ErrorTracker.categorize(new ParserNotFound("text/unknown")));
+
+ // Test SAXException
+ assertEquals(ErrorType.PARSING,
+ ErrorTracker.categorize(new SAXException("XML parse error")));
+ }
+
+ // =========================================================================
+ // Scoring Error Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeScoringErrors() {
+ // Test ScoringFilterException (Nutch-specific)
+ assertEquals(ErrorType.SCORING,
+ ErrorTracker.categorize(new ScoringFilterException("Scoring failed")));
+ }
+
+ // =========================================================================
+ // Indexing Error Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeIndexingErrors() {
+ // Test IndexingException (Nutch-specific)
+ assertEquals(ErrorType.INDEXING,
+ ErrorTracker.categorize(new IndexingException("Indexing failed")));
+ }
+
+ // =========================================================================
+ // Other/Fallback Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeNullThrowable() {
+ // Null should return OTHER
+ assertEquals(ErrorType.OTHER, ErrorTracker.categorize(null));
+ }
+
+ @Test
+ public void testCategorizeGenericException() {
+ // Generic Exception should return OTHER
+ assertEquals(ErrorType.OTHER,
+ ErrorTracker.categorize(new Exception("Generic error")));
+
+ // RuntimeException should return OTHER
+ assertEquals(ErrorType.OTHER,
+ ErrorTracker.categorize(new RuntimeException("Runtime error")));
+ }
+
+ // =========================================================================
+ // Cause Chain Categorization Tests
+ // =========================================================================
+
+ @Test
+ public void testCategorizeCauseChain() {
+ // Exception with a network cause should be categorized as NETWORK
+ IOException cause = new IOException("Root cause");
+ Exception wrapper = new Exception("Wrapper", cause);
+ assertEquals(ErrorType.NETWORK, ErrorTracker.categorize(wrapper));
+
+ // Exception with a timeout cause should be categorized as TIMEOUT
+ SocketTimeoutException timeoutCause = new SocketTimeoutException("Timeout");
+ Exception timeoutWrapper = new Exception("Wrapper", timeoutCause);
+ assertEquals(ErrorType.TIMEOUT, ErrorTracker.categorize(timeoutWrapper));
+ }
+
+ @Test
+ public void testCategorizeNestedCauseChain() {
+ // Deep nested cause chain: RuntimeException -> Exception -> IOException
+ IOException rootCause = new IOException("Root cause");
+ Exception middleWrapper = new Exception("Middle", rootCause);
+ RuntimeException outerWrapper = new RuntimeException("Outer", middleWrapper);
+ assertEquals(ErrorType.NETWORK, ErrorTracker.categorize(outerWrapper));
+
+ // Deep nested with Nutch-specific exception
+ ScoringFilterException scoringCause = new ScoringFilterException("Scoring error");
+ Exception wrapper1 = new Exception("Wrapper 1", scoringCause);
+ Exception wrapper2 = new Exception("Wrapper 2", wrapper1);
+ assertEquals(ErrorType.SCORING, ErrorTracker.categorize(wrapper2));
+ }
+
+ // =========================================================================
+ // Record Error Tests (Local Accumulation)
+ // =========================================================================
+
+ @Test
+ public void testRecordErrorByType() {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER);
+
+ // Initially all counts should be 0
+ assertEquals(0, tracker.getTotalCount());
+ assertEquals(0, tracker.getCount(ErrorType.NETWORK));
+
+ // Record a NETWORK error
+ tracker.recordError(ErrorType.NETWORK);
+ assertEquals(1, tracker.getTotalCount());
+ assertEquals(1, tracker.getCount(ErrorType.NETWORK));
+ assertEquals(0, tracker.getCount(ErrorType.TIMEOUT));
+
+ // Record another NETWORK error
+ tracker.recordError(ErrorType.NETWORK);
+ assertEquals(2, tracker.getTotalCount());
+ assertEquals(2, tracker.getCount(ErrorType.NETWORK));
+
+ // Record a TIMEOUT error
+ tracker.recordError(ErrorType.TIMEOUT);
+ assertEquals(3, tracker.getTotalCount());
+ assertEquals(2, tracker.getCount(ErrorType.NETWORK));
+ assertEquals(1, tracker.getCount(ErrorType.TIMEOUT));
+ }
+
+ @Test
+ public void testRecordErrorByThrowable() {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER);
+
+ // Record an IOException (should be categorized as NETWORK)
+ tracker.recordError(new IOException("Test"));
+ assertEquals(1, tracker.getTotalCount());
+ assertEquals(1, tracker.getCount(ErrorType.NETWORK));
+
+ // Record a SocketTimeoutException (should be categorized as TIMEOUT)
+ tracker.recordError(new SocketTimeoutException("Test"));
+ assertEquals(2, tracker.getTotalCount());
+ assertEquals(1, tracker.getCount(ErrorType.TIMEOUT));
+
+ // Record a MalformedURLException (should be categorized as URL)
+ tracker.recordError(new MalformedURLException("Test"));
+ assertEquals(3, tracker.getTotalCount());
+ assertEquals(1, tracker.getCount(ErrorType.URL));
+ }
+
+ // =========================================================================
+ // Counter Name Mapping Tests
+ // =========================================================================
+
+ @Test
+ public void testGetCounterName() {
+ // Test counter name mapping
+ assertEquals(NutchMetrics.ERROR_NETWORK_TOTAL,
+ ErrorTracker.getCounterName(ErrorType.NETWORK));
+ assertEquals(NutchMetrics.ERROR_PROTOCOL_TOTAL,
+ ErrorTracker.getCounterName(ErrorType.PROTOCOL));
+ assertEquals(NutchMetrics.ERROR_PARSING_TOTAL,
+ ErrorTracker.getCounterName(ErrorType.PARSING));
+ assertEquals(NutchMetrics.ERROR_URL_TOTAL,
+ ErrorTracker.getCounterName(ErrorType.URL));
+ assertEquals(NutchMetrics.ERROR_SCORING_TOTAL,
+ ErrorTracker.getCounterName(ErrorType.SCORING));
+ assertEquals(NutchMetrics.ERROR_INDEXING_TOTAL,
+ ErrorTracker.getCounterName(ErrorType.INDEXING));
+ assertEquals(NutchMetrics.ERROR_TIMEOUT_TOTAL,
+ ErrorTracker.getCounterName(ErrorType.TIMEOUT));
+ assertEquals(NutchMetrics.ERROR_OTHER_TOTAL,
+ ErrorTracker.getCounterName(ErrorType.OTHER));
+ }
+
+ @Test
+ public void testGetCounterNameForThrowable() {
+ // Test getting counter name directly from throwable
+ assertEquals(NutchMetrics.ERROR_NETWORK_TOTAL,
+ ErrorTracker.getCounterName(new IOException("Test")));
+ assertEquals(NutchMetrics.ERROR_TIMEOUT_TOTAL,
+ ErrorTracker.getCounterName(new SocketTimeoutException("Test")));
+ assertEquals(NutchMetrics.ERROR_URL_TOTAL,
+ ErrorTracker.getCounterName(new MalformedURLException("Test")));
+ assertEquals(NutchMetrics.ERROR_OTHER_TOTAL,
+ ErrorTracker.getCounterName(new RuntimeException("Test")));
+
+ // Test Nutch-specific exceptions
+ assertEquals(NutchMetrics.ERROR_PROTOCOL_TOTAL,
+ ErrorTracker.getCounterName(new ProtocolException("Test")));
+ assertEquals(NutchMetrics.ERROR_PARSING_TOTAL,
+ ErrorTracker.getCounterName(new ParseException("Test")));
+ assertEquals(NutchMetrics.ERROR_SCORING_TOTAL,
+ ErrorTracker.getCounterName(new ScoringFilterException("Test")));
+ assertEquals(NutchMetrics.ERROR_INDEXING_TOTAL,
+ ErrorTracker.getCounterName(new IndexingException("Test")));
+ }
+
+ // =========================================================================
+ // Hadoop Context Integration Tests (Using Mocks)
+ // =========================================================================
+
+ @Test
+ public void testConstructorWithContext() {
+ // Create ErrorTracker with context - should initialize counters
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER, mockContext);
+
+ // Verify counters were requested from context
+ // Total counter + 8 error type counters = 9 calls
+ verify(mockContext, atLeast(9)).getCounter(anyString(), anyString());
+ }
+
+ @Test
+ public void testInitCounters() {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER);
+
+ // Initialize counters
+ tracker.initCounters(mockContext);
+
+ // Verify counters were requested
+ verify(mockContext).getCounter(NutchMetrics.GROUP_FETCHER, NutchMetrics.ERROR_TOTAL);
+ verify(mockContext).getCounter(NutchMetrics.GROUP_FETCHER, NutchMetrics.ERROR_NETWORK_TOTAL);
+ verify(mockContext).getCounter(NutchMetrics.GROUP_FETCHER, NutchMetrics.ERROR_TIMEOUT_TOTAL);
+ }
+
+ @Test
+ public void testIncrementCountersWithType() {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER, mockContext);
+
+ // Increment counters directly
+ tracker.incrementCounters(ErrorType.NETWORK);
+
+ // Verify counter was incremented (total + specific type)
+ verify(mockCounter, times(2)).increment(1);
+ }
+
+ @Test
+ public void testIncrementCountersWithThrowable() {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER, mockContext);
+
+ // Increment counters with throwable
+ tracker.incrementCounters(new IOException("Test"));
+
+ // Verify counter was incremented (total + NETWORK type)
+ verify(mockCounter, times(2)).increment(1);
+ }
+
+ @Test
+ public void testIncrementCountersWithoutInit() {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER);
+
+ // Should throw IllegalStateException when counters not initialized
+ assertThrows(IllegalStateException.class, () -> {
+ tracker.incrementCounters(ErrorType.NETWORK);
+ });
+ }
+
+ @Test
+ public void testEmitCounters() {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER);
+
+ // Record some errors locally
+ tracker.recordError(ErrorType.NETWORK);
+ tracker.recordError(ErrorType.NETWORK);
+ tracker.recordError(ErrorType.TIMEOUT);
+
+ // Emit counters (without cached counters - uses fallback)
+ tracker.emitCounters(mockContext);
+
+ // Verify counters were requested and incremented
+ verify(mockContext).getCounter(NutchMetrics.GROUP_FETCHER, NutchMetrics.ERROR_TOTAL);
+ verify(mockContext).getCounter(NutchMetrics.GROUP_FETCHER, NutchMetrics.ERROR_NETWORK_TOTAL);
+ verify(mockContext).getCounter(NutchMetrics.GROUP_FETCHER, NutchMetrics.ERROR_TIMEOUT_TOTAL);
+ }
+
+ @Test
+ public void testEmitCountersWithCachedCounters() {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER, mockContext);
+
+ // Reset mock to clear constructor calls
+ reset(mockCounter);
+
+ // Record some errors locally
+ tracker.recordError(ErrorType.NETWORK);
+ tracker.recordError(ErrorType.NETWORK);
+ tracker.recordError(ErrorType.TIMEOUT);
+
+ // Emit counters (with cached counters)
+ tracker.emitCounters(mockContext);
+
+ // Verify cached counters were used (increment called with accumulated values)
+ verify(mockCounter).increment(3L); // total count
+ verify(mockCounter).increment(2L); // NETWORK count
+ verify(mockCounter).increment(1L); // TIMEOUT count
+ }
+
+ // =========================================================================
+ // Thread Safety Tests
+ // =========================================================================
+
+ @Test
+ public void testThreadSafety() throws InterruptedException {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER);
+
+ // Create multiple threads that record errors concurrently
+ Thread[] threads = new Thread[10];
+ for (int i = 0; i < threads.length; i++) {
+ threads[i] = new Thread(() -> {
+ for (int j = 0; j < 100; j++) {
+ tracker.recordError(ErrorType.NETWORK);
+ }
+ });
+ }
+
+ // Start all threads
+ for (Thread thread : threads) {
+ thread.start();
+ }
+
+ // Wait for all threads to complete
+ for (Thread thread : threads) {
+ thread.join();
+ }
+
+ // Verify counts
+ assertEquals(1000, tracker.getTotalCount());
+ assertEquals(1000, tracker.getCount(ErrorType.NETWORK));
+ }
+
+ @Test
+ public void testThreadSafetyMixedErrorTypes() throws InterruptedException {
+ ErrorTracker tracker = new ErrorTracker(NutchMetrics.GROUP_FETCHER);
+
+ // Create threads that record different error types concurrently
+ Thread networkThread = new Thread(() -> {
+ for (int i = 0; i < 500; i++) {
+ tracker.recordError(ErrorType.NETWORK);
+ }
+ });
+
+ Thread timeoutThread = new Thread(() -> {
+ for (int i = 0; i < 300; i++) {
+ tracker.recordError(ErrorType.TIMEOUT);
+ }
+ });
+
+ Thread urlThread = new Thread(() -> {
+ for (int i = 0; i < 200; i++) {
+ tracker.recordError(ErrorType.URL);
+ }
+ });
+
+ networkThread.start();
+ timeoutThread.start();
+ urlThread.start();
+
+ networkThread.join();
+ timeoutThread.join();
+ urlThread.join();
+
+ // Verify counts
+ assertEquals(1000, tracker.getTotalCount());
+ assertEquals(500, tracker.getCount(ErrorType.NETWORK));
+ assertEquals(300, tracker.getCount(ErrorType.TIMEOUT));
+ assertEquals(200, tracker.getCount(ErrorType.URL));
+ }
+}