From 182597aa16f7e74eedc4cf291a36a75a5955e45b Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Tue, 29 Sep 2026 10:13:10 +0200 Subject: [PATCH] #2057 FetcherBolt: extract the response and failure handling Signed-off-by: Davide Polato --- .../stormcrawler/bolt/FetchOutcomes.java | 193 ++++++++++++++ .../apache/stormcrawler/bolt/FetcherBolt.java | 179 +++---------- .../stormcrawler/bolt/FetchOutcomesTest.java | 246 ++++++++++++++++++ 3 files changed, 477 insertions(+), 141 deletions(-) create mode 100644 core/src/main/java/org/apache/stormcrawler/bolt/FetchOutcomes.java create mode 100644 core/src/test/java/org/apache/stormcrawler/bolt/FetchOutcomesTest.java diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchOutcomes.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetchOutcomes.java new file mode 100644 index 000000000..3c0b61924 --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetchOutcomes.java @@ -0,0 +1,193 @@ +/* + * 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.stormcrawler.bolt; + +import java.io.InterruptedIOException; +import java.net.UnknownHostException; +import org.apache.commons.lang3.StringUtils; +import org.apache.http.HttpHeaders; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.metrics.ScopedCounter; +import org.apache.stormcrawler.persistence.Status; +import org.apache.stormcrawler.protocol.AbstractHttpProtocol; +import org.apache.stormcrawler.protocol.FetchTimeoutException; +import org.apache.stormcrawler.protocol.ProtocolResponse; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Turns the response of a protocol for a URL, or any exception thrown while fetching it, into what + * {@link FetcherBolt} reports for the URL. Counts each response in status_{code}, each failure in + * exception, timeouts also in fetch.timeout and the fetcher.thread.timeout deadline also in + * fetch.deadline. The bolt does the emitting. + */ +final class FetchOutcomes { + + // the bolt's category: log configurations for FetcherBolt keep covering these lines + private static final Logger LOG = LoggerFactory.getLogger(FetcherBolt.class); + + private static final String[] NO_KEY_VALUES = {}; + + /** + * What to report for a URL. + * + * @param metadata the metadata to emit with the URL + * @param parse whether the content goes to the default stream for parsing; otherwise the status + * goes to the status stream + * @param redirectTarget the Location of a redirect, as sent and not resolved; null for anything + * but a redirect with a non-blank Location + * @param redirectKeyValues custom key/values for the outlink to the redirect target, as {@link + * StatusEmitterBolt#emitOutlink} takes them + */ + record Outcome( + Status status, + Metadata metadata, + boolean parse, + String redirectTarget, + String[] redirectKeyValues) {} + + private final String protocolMetadataPrefix; + private final ScopedCounter eventCounter; + + FetchOutcomes(String protocolMetadataPrefix, ScopedCounter eventCounter) { + this.protocolMetadataPrefix = protocolMetadataPrefix; + this.eventCounter = eventCounter; + } + + /** + * @param metadata the metadata the URL came with, left unchanged: the outcome has a merged copy + * @param robotsCrawlDelaySecs the robots crawl delay to report, or null + * @param timeFetching milliseconds spent fetching + * @param timeInQueues milliseconds the URL waited in the fetch queues + */ + Outcome ofResponse( + ProtocolResponse response, + Metadata metadata, + String robotsCrawlDelaySecs, + long timeFetching, + long timeInQueues) { + // merges the original MD and the ones returned by the + // protocol + Metadata mergedMetadata = new Metadata(); + mergedMetadata.putAll(metadata); + + // add a prefix to avoid confusion, preserve protocol + // metadata persisted or transferred from previous fetches + mergedMetadata.putAll(response.getMetadata(), protocolMetadataPrefix); + + // Only the locally parsed robots.txt value may populate this control signal. + // A colliding protocol prefix/header must not pace an unrelated queue. + mergedMetadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY); + + // Request shaping comes from the configuration, never from a fetched page: a + // response header named set-header lands on exactly the key the protocols read + // to add headers to an outgoing request. Any value configured for this URL is + // put back after the response metadata has been dropped. + final String setHeaderKey = + protocolMetadataPrefix + AbstractHttpProtocol.SET_HEADER_BY_REQUEST; + mergedMetadata.remove(setHeaderKey); + mergedMetadata.setValues(setHeaderKey, metadata.getValues(setHeaderKey)); + if (robotsCrawlDelaySecs != null) { + mergedMetadata.setValue(Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs); + } + + mergedMetadata.setValue("fetch.statusCode", Integer.toString(response.getStatusCode())); + mergedMetadata.setValue("fetch.byteLength", Integer.toString(response.getContent().length)); + mergedMetadata.setValue("fetch.loadingTime", Long.toString(timeFetching)); + mergedMetadata.setValue("fetch.timeInQueues", Long.toString(timeInQueues)); + + final Status status = Status.fromHTTPCode(response.getStatusCode()); + eventCounter.scope("status_" + response.getStatusCode()).incrBy(1); + + if (status == Status.FETCHED) { + // a 304 is marked as fetched so that it gets rescheduled, but there is nothing to + // parse or index + boolean parse = response.getStatusCode() != 304; + return new Outcome(status, mergedMetadata, parse, null, NO_KEY_VALUES); + } + + String redirection = + status == Status.REDIRECTION + ? response.getMetadata().getFirstValue(HttpHeaders.LOCATION) + : null; + if (StringUtils.isBlank(redirection)) { + return new Outcome(status, mergedMetadata, false, null, NO_KEY_VALUES); + } + + // stores the URL it redirects to + // used for debugging mainly - do not resolve the target + // URL + mergedMetadata.setValue("_redirTo", redirection); + + // a sitemap which redirects (e.g. /sitemap.xml to + // /sitemap_index.xml) must stay a sitemap: the key + // is persisted, not transferred to outlinks by + // default, so carry it onto the redirect target + String[] keyValues = + Boolean.parseBoolean(mergedMetadata.getFirstValue(SiteMapParserBolt.isSitemapKey)) + ? new String[] {SiteMapParserBolt.isSitemapKey, "true"} + : NO_KEY_VALUES; + return new Outcome(status, mergedMetadata, false, redirection, keyValues); + } + + /** + * Records the failure as fetch.exception: "Socket timeout fetching", "Unknown host", or the + * class name of the exception. + * + * @param url the URL, for the log + * @param metadata the metadata the URL came with; the failure is written to it, or to a new + * instance when it is empty, since it may then be the read-only {@link Metadata#empty} + */ + Outcome ofFailure(Exception e, String url, Metadata metadata) { + String message = e.getMessage(); + if (message == null) { + message = ""; + } + + // common exceptions for which we log only a short message + if (e instanceof InterruptedIOException || message.contains(" timed out")) { + LOG.info("Socket timeout fetching {}", url); + message = "Socket timeout fetching"; + eventCounter.scope("fetch.timeout").incrBy(1); + if (e instanceof FetchTimeoutException) { + // the hard deadline, as opposed to the protocol's socket timeouts + eventCounter.scope("fetch.deadline").incrBy(1); + } + } else if (e.getCause() instanceof UnknownHostException + || e instanceof UnknownHostException) { + LOG.info("Unknown host {}", url); + message = "Unknown host"; + } else { + message = e.getClass().getName(); + if (LOG.isDebugEnabled()) { + LOG.debug("Exception while fetching {}", url, e); + } else { + LOG.info("Exception while fetching {} -> {}", url, message); + } + } + + if (metadata.size() == 0) { + metadata = new Metadata(); + } + // add the reason of the failure in the metadata + metadata.setValue("fetch.exception", message); + eventCounter.scope("exception").incrBy(1); + return new Outcome(Status.FETCH_ERROR, metadata, false, null, NO_KEY_VALUES); + } +} diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java index 5763692bd..61ca63b7c 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -27,7 +27,6 @@ import java.util.Map.Entry; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.lang3.StringUtils; -import org.apache.http.HttpHeaders; import org.apache.storm.Config; import org.apache.storm.task.OutputCollector; import org.apache.storm.task.TopologyContext; @@ -44,8 +43,6 @@ import org.apache.stormcrawler.metrics.ScopedCounter; import org.apache.stormcrawler.metrics.ScopedReducedMetric; import org.apache.stormcrawler.persistence.Status; -import org.apache.stormcrawler.protocol.AbstractHttpProtocol; -import org.apache.stormcrawler.protocol.FetchTimeoutException; import org.apache.stormcrawler.protocol.Protocol; import org.apache.stormcrawler.protocol.ProtocolFactory; import org.apache.stormcrawler.protocol.ProtocolResponse; @@ -105,6 +102,8 @@ public class FetcherBolt extends StatusEmitterBolt { private CrawlDelayPolicy crawlDelayPolicy; + private FetchOutcomes fetchOutcomes; + /** Largest number of helper threads ever alive; for tests. */ int helperPoolSize() { return fetchHelpers == null ? 0 : fetchHelpers.largestPoolSize(); @@ -125,20 +124,12 @@ private class FetcherThread extends Thread { private long timeoutInQueues = -1; - // by default remains as is-pre 1.17 - private String protocolMetadataPrefix = ""; - public FetcherThread(Config conf, int num) { this.setDaemon(true); // don't hang JVM on exit this.setName("FetcherThread #" + num); // use an informative name this.threadNum = num; timeoutInQueues = ConfUtils.getLong(conf, QUEUED_TIMEOUT_PARAM_KEY, timeoutInQueues); - protocolMetadataPrefix = - ConfUtils.getString( - conf, - ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, - protocolMetadataPrefix); } @Override @@ -294,100 +285,36 @@ public void run() { response.getStatusCode(), timeFetching); - // merges the original MD and the ones returned by the - // protocol - Metadata mergedMetadata = new Metadata(); - mergedMetadata.putAll(metadata); - - // add a prefix to avoid confusion, preserve protocol - // metadata persisted or transferred from previous fetches - mergedMetadata.putAll(response.getMetadata(), protocolMetadataPrefix); - - // Only the locally parsed robots.txt value may populate this control signal. - // A colliding protocol prefix/header must not pace an unrelated queue. - mergedMetadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY); - - // Request shaping comes from the configuration, never from a fetched page: a - // response header named set-header lands on exactly the key the protocols read - // to add headers to an outgoing request. Any value configured for this URL is - // put back after the response metadata has been dropped. - final String setHeaderKey = - protocolMetadataPrefix + AbstractHttpProtocol.SET_HEADER_BY_REQUEST; - mergedMetadata.remove(setHeaderKey); - mergedMetadata.setValues(setHeaderKey, metadata.getValues(setHeaderKey)); - if (robotsCrawlDelaySecs != null) { - mergedMetadata.setValue( - Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs); - } - - mergedMetadata.setValue( - "fetch.statusCode", Integer.toString(response.getStatusCode())); - - mergedMetadata.setValue("fetch.byteLength", Integer.toString(byteLength)); - - mergedMetadata.setValue("fetch.loadingTime", Long.toString(timeFetching)); - - mergedMetadata.setValue("fetch.timeInQueues", Long.toString(timeInQueues)); - - // determine the status based on the status code - final Status status = Status.fromHTTPCode(response.getStatusCode()); - - eventCounter.scope("status_" + response.getStatusCode()).incrBy(1); - - final Values tupleToSend = new Values(fit.url, mergedMetadata, status); - - // if the status is OK emit on default stream - if (status.equals(Status.FETCHED)) { - if (response.getStatusCode() == 304) { - // mark this URL as fetched so that it gets - // rescheduled - // but do not try to parse or index - collector.emit(Constants.StatusStreamName, fit.tuple, tupleToSend); - } else { - // send content for parsing - collector.emit( - Utils.DEFAULT_STREAM_ID, + FetchOutcomes.Outcome outcome = + fetchOutcomes.ofResponse( + response, + metadata, + robotsCrawlDelaySecs, + timeFetching, + timeInQueues); + if (outcome.parse()) { + // send content for parsing + collector.emit( + Utils.DEFAULT_STREAM_ID, + fit.tuple, + new Values(fit.url, response.getContent(), outcome.metadata())); + } else { + // allowRedirs() can be overridden: it is asked for every redirect and + // nothing else (#954) + if (outcome.status() == Status.REDIRECTION + && allowRedirs() + && outcome.redirectTarget() != null) { + emitOutlink( fit.tuple, - new Values(fit.url, response.getContent(), mergedMetadata)); - } - } else if (status.equals(Status.REDIRECTION)) { - - // find the URL it redirects to - String redirection = - response.getMetadata().getFirstValue(HttpHeaders.LOCATION); - - // stores the URL it redirects to - // used for debugging mainly - do not resolve the target - // URL - if (StringUtils.isNotBlank(redirection)) { - mergedMetadata.setValue("_redirTo", redirection); + url, + outcome.redirectTarget(), + outcome.metadata(), + outcome.redirectKeyValues()); } - - // https://github.com/apache/stormcrawler/issues/954 - if (allowRedirs() && StringUtils.isNotBlank(redirection)) { - // a sitemap which redirects (e.g. /sitemap.xml to - // /sitemap_index.xml) must stay a sitemap: the key - // is persisted, not transferred to outlinks by - // default, so carry it onto the redirect target - if (Boolean.parseBoolean( - mergedMetadata.getFirstValue(SiteMapParserBolt.isSitemapKey))) { - emitOutlink( - fit.tuple, - url, - redirection, - mergedMetadata, - SiteMapParserBolt.isSitemapKey, - "true"); - } else { - emitOutlink(fit.tuple, url, redirection, mergedMetadata); - } - } - - // mark this URL as redirected - collector.emit(Constants.StatusStreamName, fit.tuple, tupleToSend); - } else { - // error - collector.emit(Constants.StatusStreamName, fit.tuple, tupleToSend); + collector.emit( + Constants.StatusStreamName, + fit.tuple, + new Values(fit.url, outcome.metadata(), outcome.status())); } } catch (FetchTimeoutHelpers.SaturatedException e) { @@ -405,47 +332,12 @@ public void run() { fetchHelpers.maxHelpers()); saturated = true; } catch (Exception exece) { - String message = exece.getMessage(); - if (message == null) { - message = ""; - } - - // common exceptions for which we log only a short message - if (exece instanceof java.io.InterruptedIOException - || message.contains(" timed out")) { - LOG.info("Socket timeout fetching {}", fit.url); - message = "Socket timeout fetching"; - eventCounter.scope("fetch.timeout").incrBy(1); - if (exece instanceof FetchTimeoutException) { - // the hard deadline, as opposed to the protocol's socket timeouts - eventCounter.scope("fetch.deadline").incrBy(1); - } - } else if (exece.getCause() instanceof java.net.UnknownHostException - || exece instanceof java.net.UnknownHostException) { - LOG.info("Unknown host {}", fit.url); - message = "Unknown host"; - } else { - message = exece.getClass().getName(); - if (LOG.isDebugEnabled()) { - LOG.debug("Exception while fetching {}", fit.url, exece); - } else { - LOG.info("Exception while fetching {} -> {}", fit.url, message); - } - } - - if (metadata.size() == 0) { - metadata = new Metadata(); - } - // add the reason of the failure in the metadata - metadata.setValue("fetch.exception", message); - - // send to status stream + FetchOutcomes.Outcome outcome = + fetchOutcomes.ofFailure(exece, fit.url, metadata); collector.emit( Constants.StatusStreamName, fit.tuple, - new Values(fit.url, metadata, Status.FETCH_ERROR)); - - eventCounter.scope("exception").incrBy(1); + new Values(fit.url, outcome.metadata(), outcome.status())); } finally { if (saturated) { long delay = fetchQueues.backOffFetchItem(fit); @@ -573,6 +465,11 @@ public void prepare( this.fetchQueues = new FetchItemQueues(conf); + // by default remains as is-pre 1.17 + String protocolMetadataPrefix = + ConfUtils.getString(conf, ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, ""); + fetchOutcomes = new FetchOutcomes(protocolMetadataPrefix, eventCounter); + crawlDelayPolicy = new CrawlDelayPolicy( ConfUtils.getInt(conf, "fetcher.max.crawl.delay", 30) * 1000L, diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchOutcomesTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchOutcomesTest.java new file mode 100644 index 000000000..c019dcba6 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchOutcomesTest.java @@ -0,0 +1,246 @@ +/* + * 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.stormcrawler.bolt; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.net.SocketTimeoutException; +import java.net.UnknownHostException; +import java.util.HashMap; +import java.util.Map; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.metrics.ScopedCounter; +import org.apache.stormcrawler.persistence.Status; +import org.apache.stormcrawler.protocol.AbstractHttpProtocol; +import org.apache.stormcrawler.protocol.FetchTimeoutException; +import org.apache.stormcrawler.protocol.ProtocolResponse; +import org.junit.jupiter.api.Test; + +class FetchOutcomesTest { + + private static final String URL = "http://a.test/page"; + + private static final String PREFIX = "protocol."; + + private final Map counts = new HashMap<>(); + + private final ScopedCounter counter = name -> n -> counts.merge(name, n, Long::sum); + + private final FetchOutcomes outcomes = new FetchOutcomes(PREFIX, counter); + + private static ProtocolResponse response(int code, String... headers) { + Metadata metadata = new Metadata(); + for (int i = 0; i < headers.length; i += 2) { + metadata.setValue(headers[i], headers[i + 1]); + } + return new ProtocolResponse(new byte[] {1, 2, 3}, code, metadata); + } + + private FetchOutcomes.Outcome ofResponse(ProtocolResponse response, Metadata metadata) { + return outcomes.ofResponse(response, metadata, null, 7, 11); + } + + @Test + void pageIsSentForParsingWithTheMergedMetadata() { + Metadata source = new Metadata(); + source.setValue("depth", "1"); + + FetchOutcomes.Outcome outcome = + ofResponse(response(200, "Content-Type", "text/html"), source); + assertEquals(Status.FETCHED, outcome.status()); + assertTrue(outcome.parse()); + assertNull(outcome.redirectTarget()); + Metadata md = outcome.metadata(); + assertNotSame(source, md); + assertEquals(1, source.size()); + assertEquals("1", md.getFirstValue("depth")); + assertEquals("text/html", md.getFirstValue(PREFIX + "Content-Type")); + assertEquals("200", md.getFirstValue("fetch.statusCode")); + assertEquals("3", md.getFirstValue("fetch.byteLength")); + assertEquals("7", md.getFirstValue("fetch.loadingTime")); + assertEquals("11", md.getFirstValue("fetch.timeInQueues")); + assertEquals(Map.of("status_200", 1L), counts); + } + + @Test + void notModifiedIsNotParsed() { + FetchOutcomes.Outcome outcome = ofResponse(response(304), new Metadata()); + assertEquals(Status.FETCHED, outcome.status()); + assertFalse(outcome.parse()); + } + + @Test + void redirectCarriesItsTarget() { + FetchOutcomes.Outcome outcome = + ofResponse(response(301, "Location", "http://a.test/new"), new Metadata()); + assertEquals(Status.REDIRECTION, outcome.status()); + assertFalse(outcome.parse()); + assertEquals("http://a.test/new", outcome.redirectTarget()); + assertEquals("http://a.test/new", outcome.metadata().getFirstValue("_redirTo")); + assertArrayEquals(new String[0], outcome.redirectKeyValues()); + } + + @Test + void redirectWithoutLocationHasNoTarget() { + FetchOutcomes.Outcome outcome = ofResponse(response(302), new Metadata()); + assertEquals(Status.REDIRECTION, outcome.status()); + assertNull(outcome.redirectTarget()); + assertNull(outcome.metadata().getFirstValue("_redirTo")); + } + + @Test + void locationOutsideARedirectIsIgnored() { + FetchOutcomes.Outcome outcome = + ofResponse(response(201, "Location", "http://a.test/new"), new Metadata()); + assertNull(outcome.redirectTarget()); + assertNull(outcome.metadata().getFirstValue("_redirTo")); + } + + @Test + void blankLocationIsNoTarget() { + FetchOutcomes.Outcome outcome = ofResponse(response(301, "Location", " "), new Metadata()); + assertNull(outcome.redirectTarget()); + assertNull(outcome.metadata().getFirstValue("_redirTo")); + } + + @Test + void redirectedSitemapStaysASitemap() { + Metadata source = new Metadata(); + source.setValue(SiteMapParserBolt.isSitemapKey, "true"); + + FetchOutcomes.Outcome outcome = + ofResponse(response(301, "Location", "http://a.test/sitemap_index.xml"), source); + assertArrayEquals( + new String[] {SiteMapParserBolt.isSitemapKey, "true"}, outcome.redirectKeyValues()); + } + + @Test + void errorStatusIsAFetchError() { + FetchOutcomes.Outcome outcome = ofResponse(response(404), new Metadata()); + assertEquals(Status.FETCH_ERROR, outcome.status()); + assertFalse(outcome.parse()); + assertEquals(Map.of("status_404", 1L), counts); + } + + /** + * The robots crawl delay key only ever carries the value parsed from robots.txt, and request + * headers only ever come from the configuration, whatever the response says. + */ + @Test + void responseCannotSetControlKeys() { + FetchOutcomes unprefixed = new FetchOutcomes("", counter); + Metadata source = new Metadata(); + source.setValue(AbstractHttpProtocol.SET_HEADER_BY_REQUEST, "X-Configured=yes"); + ProtocolResponse response = + response( + 200, + Constants.ROBOTS_CRAWL_DELAY_KEY, + "120", + AbstractHttpProtocol.SET_HEADER_BY_REQUEST, + "X-Injected=yes"); + + Metadata md = unprefixed.ofResponse(response, source, null, 0, 0).metadata(); + assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + assertArrayEquals( + new String[] {"X-Configured=yes"}, + md.getValues(AbstractHttpProtocol.SET_HEADER_BY_REQUEST)); + + md = unprefixed.ofResponse(response, source, "60", 0, 0).metadata(); + assertEquals("60", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } + + @Test + void fetchDeadlineIsASocketTimeout() { + FetchOutcomes.Outcome outcome = + outcomes.ofFailure(new FetchTimeoutException(URL, 5), URL, metadata()); + assertEquals(Status.FETCH_ERROR, outcome.status()); + assertEquals( + "Socket timeout fetching", outcome.metadata().getFirstValue("fetch.exception")); + assertEquals(Map.of("fetch.timeout", 1L, "fetch.deadline", 1L, "exception", 1L), counts); + } + + @Test + void socketTimeoutIsNotADeadline() { + FetchOutcomes.Outcome outcome = + outcomes.ofFailure(new SocketTimeoutException("read"), URL, metadata()); + assertEquals( + "Socket timeout fetching", outcome.metadata().getFirstValue("fetch.exception")); + assertEquals(Map.of("fetch.timeout", 1L, "exception", 1L), counts); + } + + /** The protocols report some timeouts only in the message. */ + @Test + void timedOutMessageIsASocketTimeout() { + FetchOutcomes.Outcome outcome = + outcomes.ofFailure(new IOException("connect timed out"), URL, metadata()); + assertEquals( + "Socket timeout fetching", outcome.metadata().getFirstValue("fetch.exception")); + assertEquals(Map.of("fetch.timeout", 1L, "exception", 1L), counts); + } + + @Test + void unknownHostIsRecognisedDirectlyOrAsCause() { + FetchOutcomes.Outcome direct = + outcomes.ofFailure(new UnknownHostException("a.test"), URL, metadata()); + FetchOutcomes.Outcome wrapped = + outcomes.ofFailure( + new IOException(new UnknownHostException("a.test")), URL, metadata()); + assertEquals("Unknown host", direct.metadata().getFirstValue("fetch.exception")); + assertEquals("Unknown host", wrapped.metadata().getFirstValue("fetch.exception")); + assertEquals(Map.of("exception", 2L), counts); + } + + @Test + void otherFailureIsNamedByItsClass() { + Metadata source = metadata(); + + FetchOutcomes.Outcome outcome = + outcomes.ofFailure(new IllegalArgumentException("bad proxy"), URL, source); + assertSame(source, outcome.metadata()); + assertEquals( + IllegalArgumentException.class.getName(), + outcome.metadata().getFirstValue("fetch.exception")); + assertEquals(Map.of("exception", 1L), counts); + } + + /** Metadata.empty is read-only and shared, so the failure goes to a new instance. */ + @Test + void failureNeverWritesToTheSharedEmptyMetadata() { + FetchOutcomes.Outcome outcome = + outcomes.ofFailure(new IllegalStateException(), URL, Metadata.empty); + assertNotSame(Metadata.empty, outcome.metadata()); + assertEquals( + IllegalStateException.class.getName(), + outcome.metadata().getFirstValue("fetch.exception")); + assertEquals(0, Metadata.empty.size()); + } + + private static Metadata metadata() { + Metadata metadata = new Metadata(); + metadata.setValue("depth", "1"); + return metadata; + } +}