From 91ceaa905ed2f1f03ee19c03dffa37383218331d Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 6 Sep 2026 23:58:36 +0200 Subject: [PATCH 1/7] #2134 fetcher.thread.timeout: cancel the okhttp call, bounded helper pool for other protocols The bolt-level fetch timeout ran the protocol call on a single-thread executor owned by each FetcherThread and abandoned it with future.cancel(true) on expiry. okhttp does not honour the interrupt while connecting or reading, so the helper stayed blocked on the socket and the FetcherThread queued behind it at its next fetch: a host that dribbles bytes could take a thread out of service for as long as http.timeout. It also doubled the thread count of the bolt whenever the option was on, and the robots.txt lookup was not covered by the timeout at all. okhttp: fetcher.thread.timeout is applied as a deadline on the call (Call.timeout()), enforced by okio's shared watchdog: on expiry the call is cancelled, the socket closed and the fetching thread gets a FetchTimeoutException at once. The hops of a redirect chain share one deadline, each getting the time left. The robots.txt lookup and each redirect it follows are calls of their own with their own deadline, which is documented. With http.content.partial.as.trimmed the content received before the deadline is kept and flagged as trimmed for "time", as it already was for the call timeout. Protocol gains a default supportsFetchTimeout(url, metadata) (false); okhttp returns true when configured, DelegatorProtocol resolves the delegate per URL and requires the one the robots.txt lookup is routed to to support it as well. Bolts: the timeout machinery moves to a new package-private FetchTimeoutHelpers owned by FetcherBolt and SimpleFetcherBolt. Its call() runs both the robots.txt lookup and the fetch on the calling thread when the protocol enforces the timeout itself, otherwise on a helper thread from one bounded pool per bolt (fetcher.thread.timeout.helpers, default 2 x fetcher.threads.number, 2 for SimpleFetcherBolt, threads created on demand and released after a minute idle). With the default protocol no helper thread is ever created. A deadline on a helper throws the same FetchTimeoutException as okhttp. A robots.txt lookup which times out on a helper lets the page be fetched without rules, as HttpRobotRulesParser already does for a failed lookup with okhttp, so both paths behave the same. A full pool rejects at once: the URL never reached the network, so it is acked without a status, like a URL which waited too long in the queue, rather than reported as FETCH_ERROR and counted towards max.fetch.errors. FetchTimeout.secs(conf) reads and clamps the deadline to topology.message.timeout.secs for both the protocol and the helpers. FetchTimeoutException extends InterruptedIOException so that it is classified as a timeout; the new fetch.deadline counter counts only the deadline, fetch.timeout keeps counting every timeout, fetch.helper.rejected the saturations, robots.timeout the abandoned lookups; fetchhelpers gauge. The fetcher.thread.timeout key is defined in Constants; the alias FetcherBolt.FETCH_TIMEOUT_PARAM_KEY shipped in 3.6.0 is kept, deprecated for removal. Documentation for the parameter, which was missing, is added to configuration.adoc. Tests: HttpProtocolFetchTimeoutTest (cancellation at the deadline with the typed exception, socket timeout not reported as the deadline, redirect chain sharing the deadline, clamp, partial content), FetchTimeoutHelpersTest (every branch of call(), clamp), fetcher bolt tests with a protocol that hangs and ignores interruption (stuck fetch not blocking the following ones, bounded pool with the rejected URL acked without status, hanging robots.txt not failing the URL), slow robots.txt bounded with okhttp, no helper threads with okhttp, delegator capability per URL and for the robots.txt route. Fixes #2134. --- .../org/apache/stormcrawler/Constants.java | 9 + .../bolt/FetchTimeoutHelpers.java | 162 +++++++++++++ .../apache/stormcrawler/bolt/FetcherBolt.java | 124 +++++----- .../stormcrawler/bolt/SimpleFetcherBolt.java | 92 ++++---- .../protocol/DelegatorProtocol.java | 32 ++- .../stormcrawler/protocol/FetchTimeout.java | 55 +++++ .../protocol/FetchTimeoutException.java | 50 ++++ .../stormcrawler/protocol/Protocol.java | 14 ++ .../protocol/okhttp/HttpProtocol.java | 34 +++ core/src/main/resources/crawler-default.yaml | 15 +- .../bolt/AbstractFetcherBoltTest.java | 132 ++++++++++- .../bolt/FetchTimeoutHelpersTest.java | 219 ++++++++++++++++++ .../stormcrawler/bolt/FetcherBoltTest.java | 36 +++ .../protocol/DelegationProtocolTest.java | 74 ++++++ .../stormcrawler/protocol/StuckProtocol.java | 77 ++++++ .../okhttp/HttpProtocolFetchTimeoutTest.java | 198 ++++++++++++++++ docs/src/main/asciidoc/configuration.adoc | 2 + 17 files changed, 1208 insertions(+), 117 deletions(-) create mode 100644 core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java create mode 100644 core/src/main/java/org/apache/stormcrawler/protocol/FetchTimeout.java create mode 100644 core/src/main/java/org/apache/stormcrawler/protocol/FetchTimeoutException.java create mode 100644 core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java create mode 100644 core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java create mode 100644 core/src/test/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocolFetchTimeoutTest.java diff --git a/core/src/main/java/org/apache/stormcrawler/Constants.java b/core/src/main/java/org/apache/stormcrawler/Constants.java index 882f2d965..f925e79d9 100644 --- a/core/src/main/java/org/apache/stormcrawler/Constants.java +++ b/core/src/main/java/org/apache/stormcrawler/Constants.java @@ -60,4 +60,13 @@ public class Constants { public static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; private Constants() {} + + /** Hard deadline in seconds for a single fetch, -1 to disable. */ + public static final String FETCH_TIMEOUT_PARAM_KEY = "fetcher.thread.timeout"; + + /** + * Maximum number of helper threads on which fetches with a timeout are run for protocols that + * can not enforce the timeout themselves. Defaults to twice fetcher.threads.number. + */ + public static final String FETCH_TIMEOUT_HELPERS_PARAM_KEY = "fetcher.thread.timeout.helpers"; } diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java new file mode 100644 index 000000000..1a167f347 --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java @@ -0,0 +1,162 @@ +/* + * 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.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.storm.task.TopologyContext; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.metrics.CrawlerMetrics; +import org.apache.stormcrawler.protocol.FetchTimeout; +import org.apache.stormcrawler.protocol.FetchTimeoutException; +import org.apache.stormcrawler.protocol.Protocol; +import org.apache.stormcrawler.util.ConfUtils; + +/** + * Enforces {@code fetcher.thread.timeout} on protocol calls (robots.txt lookup and fetch) for the + * fetcher bolts. + * + *

When the timeout is off, or the protocol enforces it itself for the URL (see {@link + * Protocol#supportsFetchTimeout(String, Metadata)}, true for the default okhttp protocol), the call + * runs on the calling thread. Otherwise it runs on a helper thread and is abandoned there when the + * deadline passes: the helper stays busy until the protocol gives up on its own, so the pool is + * bounded ({@code fetcher.thread.timeout.helpers}) and a call that finds every helper busy fails at + * once with {@link SaturatedException}. Helper threads are created on demand and released after a + * minute of inactivity: with the default protocol none is ever created. + * + *

The deadline is the one of {@link FetchTimeout#secs(Map)}, clamped to the message timeout on + * both paths. + */ +final class FetchTimeoutHelpers { + + /** Thrown when no helper thread is available to run a call with a timeout. */ + static final class SaturatedException extends Exception { + SaturatedException(String url) { + super("No fetch helper available for " + url); + } + } + + private final long timeoutSecs; + private final ThreadPoolExecutor helpers; + + /** + * @param conf the bolt configuration + * @param defaultMaxHelpers pool bound used unless {@code fetcher.thread.timeout.helpers} is set + * @param threadNamePrefix prefix of the helper thread names + */ + FetchTimeoutHelpers(Map conf, int defaultMaxHelpers, String threadNamePrefix) { + this.timeoutSecs = FetchTimeout.secs(conf); + final int maxHelpers = + ConfUtils.getInt( + conf, Constants.FETCH_TIMEOUT_HELPERS_PARAM_KEY, defaultMaxHelpers); + final AtomicInteger helperNum = new AtomicInteger(); + this.helpers = + new ThreadPoolExecutor( + 0, + Math.max(1, maxHelpers), + 60L, + TimeUnit.SECONDS, + new SynchronousQueue<>(), + r -> { + Thread t = + new Thread(r, threadNamePrefix + helperNum.incrementAndGet()); + t.setDaemon(true); + return t; + }); + } + + /** Registers the {@code fetchhelpers} gauge: number of helper threads busy with a call. */ + void registerMetrics(TopologyContext context, Map conf, int bucketSecs) { + CrawlerMetrics.registerGauge( + context, conf, "fetchhelpers", helpers::getActiveCount, bucketSecs); + } + + /** Whether a timeout is configured at all. */ + boolean enabled() { + return timeoutSecs > 0; + } + + /** Timeout in seconds, -1 when disabled. */ + long timeoutSecs() { + return timeoutSecs; + } + + /** Pool bound. */ + int maxHelpers() { + return helpers.getMaximumPoolSize(); + } + + /** Largest number of helper threads ever alive. */ + int largestPoolSize() { + return helpers.getLargestPoolSize(); + } + + /** + * Runs a protocol call under the timeout. + * + * @param call the robots.txt lookup or the fetch + * @param protocol the protocol the call goes to + * @param url the URL being fetched, or whose robots.txt is looked up + * @param metadata the metadata of that URL, possibly null + * @throws SaturatedException when every helper is busy + * @throws FetchTimeoutException when the deadline passed + * @throws Exception the protocol's own exception + */ + T call(Callable call, Protocol protocol, String url, Metadata metadata) + throws Exception { + if (timeoutSecs <= 0 || protocol.supportsFetchTimeout(url, metadata)) { + return call.call(); + } + final Future future; + try { + future = helpers.submit(call); + } catch (RejectedExecutionException e) { + throw new SaturatedException(url); + } + try { + return future.get(timeoutSecs, TimeUnit.SECONDS); + } catch (TimeoutException e) { + // a courtesy for protocols which do honour interruption + future.cancel(true); + throw new FetchTimeoutException(url, timeoutSecs); + } catch (CancellationException e) { + throw new Exception("Fetch cancelled for " + url); + } catch (ExecutionException e) { + // unwrap the real cause so the bolts' classification sees it + Throwable cause = e.getCause(); + if (cause instanceof Exception) { + throw (Exception) cause; + } + throw new Exception(cause); + } + } + + void shutdown() { + helpers.shutdownNow(); + } +} 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 67ae0acfc..fb41b2081 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -31,17 +31,11 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Queue; -import java.util.concurrent.CancellationException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -63,10 +57,12 @@ import org.apache.stormcrawler.metrics.ScopedCounter; import org.apache.stormcrawler.metrics.ScopedReducedMetric; import org.apache.stormcrawler.persistence.Status; +import org.apache.stormcrawler.protocol.FetchTimeoutException; import org.apache.stormcrawler.protocol.Protocol; import org.apache.stormcrawler.protocol.ProtocolFactory; import org.apache.stormcrawler.protocol.ProtocolResponse; import org.apache.stormcrawler.protocol.RobotRules; +import org.apache.stormcrawler.protocol.RobotRulesParser; import org.apache.stormcrawler.util.ConfUtils; import org.apache.stormcrawler.util.URLUtil; import org.slf4j.LoggerFactory; @@ -88,12 +84,10 @@ public class FetcherBolt extends StatusEmitterBolt { public static final String QUEUED_TIMEOUT_PARAM_KEY = "fetcher.timeout.queue"; /** - * Hard timeout in seconds for a single call to {@link Protocol#getProtocolOutput}. If a fetch - * exceeds this duration the thread is interrupted, the URL is marked as FETCH_ERROR, and the - * thread moves on to the next item. A value of {@code -1} (the default) disables the bolt-level - * timeout, relying solely on the protocol-level socket timeouts. + * @deprecated since 3.8, use {@link Constants#FETCH_TIMEOUT_PARAM_KEY} */ - public static final String FETCH_TIMEOUT_PARAM_KEY = "fetcher.thread.timeout"; + @Deprecated(since = "3.8", forRemoval = true) + public static final String FETCH_TIMEOUT_PARAM_KEY = Constants.FETCH_TIMEOUT_PARAM_KEY; /** Key name of the custom crawl delay for a queue that may be present in the metadata. */ private static final String CRAWL_DELAY_KEY_NAME = "crawl.delay"; @@ -130,6 +124,14 @@ public class FetcherBolt extends StatusEmitterBolt { private String[] beingFetched; + /** Runs protocol calls under fetcher.thread.timeout, see {@link FetchTimeoutHelpers}. */ + private FetchTimeoutHelpers fetchHelpers; + + /** Largest number of helper threads ever alive; for tests. */ + int helperPoolSize() { + return fetchHelpers == null ? 0 : fetchHelpers.largestPoolSize(); + } + @Override public Map getComponentConfiguration() { Config conf = new Config(); @@ -635,15 +637,6 @@ private class FetcherThread extends Thread { private long timeoutInQueues = -1; - /** Hard timeout in seconds for a single protocol fetch. -1 means disabled. */ - private long fetchTimeout = -1; - - /** - * Single-thread executor used to run the protocol call so that it can be interrupted via - * {@link Future#cancel(boolean)} when the bolt-level timeout fires. - */ - private final ExecutorService fetchExecutor; - // by default remains as is-pre 1.17 private String protocolMetadataPrefix = ""; @@ -657,24 +650,11 @@ public FetcherThread(Config conf, int num) { this.crawlDelayForce = ConfUtils.getBoolean(conf, "fetcher.server.delay.force", false); this.threadNum = num; timeoutInQueues = ConfUtils.getLong(conf, QUEUED_TIMEOUT_PARAM_KEY, timeoutInQueues); - fetchTimeout = ConfUtils.getLong(conf, FETCH_TIMEOUT_PARAM_KEY, fetchTimeout); protocolMetadataPrefix = ConfUtils.getString( conf, ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, protocolMetadataPrefix); - - if (fetchTimeout > 0) { - fetchExecutor = - Executors.newSingleThreadExecutor( - r -> { - Thread t = new Thread(r, "FetcherTimeout #" + num); - t.setDaemon(true); - return t; - }); - } else { - fetchExecutor = null; - } } @Override @@ -735,7 +715,24 @@ public void run() { "No protocol implementation found for " + fit.url); } - BaseRobotRules rules = protocol.getRobotRules(fit.url); + BaseRobotRules rules; + try { + rules = + fetchHelpers.call( + () -> protocol.getRobotRules(fit.url), + protocol, + fit.url, + metadata); + } catch (FetchTimeoutException e) { + // same outcome as with okhttp, where HttpRobotRulesParser turns a + // failed lookup into empty rules: the page is fetched without rules + LOG.info( + "[Fetcher #{}] robots.txt lookup timed out for {}", + taskId, + fit.url); + eventCounter.scope("robots.timeout").incrBy(1); + rules = RobotRulesParser.EMPTY_RULES; + } boolean fromCache = false; if (rules instanceof RobotRules && ((RobotRules) rules).getContentLengthFetched().length == 0) { @@ -871,33 +868,12 @@ public void run() { final Metadata fetchMetadata = metadata; ProtocolResponse response; - if (fetchExecutor != null) { - Future future = - fetchExecutor.submit( - () -> protocol.getProtocolOutput(fit.url, fetchMetadata)); - try { - response = future.get(fetchTimeout, TimeUnit.SECONDS); - } catch (TimeoutException e) { - future.cancel(true); - throw new Exception( - "Fetch timed out after " - + fetchTimeout - + "s fetching " - + fit.url, - e); - } catch (CancellationException e) { - throw new Exception("Fetch cancelled for " + fit.url); - } catch (ExecutionException e) { - // unwrap the real cause so existing catch logic handles it - Throwable cause = e.getCause(); - if (cause instanceof Exception) { - throw (Exception) cause; - } - throw new Exception(cause); - } - } else { - response = protocol.getProtocolOutput(fit.url, metadata); - } + response = + fetchHelpers.call( + () -> protocol.getProtocolOutput(fit.url, fetchMetadata), + protocol, + fit.url, + fetchMetadata); long timeFetching = System.currentTimeMillis() - start; @@ -1017,6 +993,17 @@ public void run() { collector.emit(Constants.StatusStreamName, fit.tuple, tupleToSend); } + } catch (FetchTimeoutHelpers.SaturatedException e) { + // the URL never reached the network: like a URL which waited too long in + // the queue, it is acked without a status so that the spout retries it + // later, rather than taking a strike towards max.fetch.errors + eventCounter.scope("fetch.helper.rejected").incrBy(1); + LOG.warn( + "[Fetcher #{}] {}: all {} fetch helpers are busy", + taskId, + e.getMessage(), + fetchHelpers.maxHelpers()); + asap = true; } catch (Exception exece) { String message = exece.getMessage(); if (message == null) { @@ -1024,10 +1011,15 @@ public void run() { } // common exceptions for which we log only a short message - if (exece.getCause() instanceof java.util.concurrent.TimeoutException + 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); @@ -1137,6 +1129,11 @@ public void prepare( int threadCount = ConfUtils.getInt(conf, "fetcher.threads.number", 10); int startDelay = ConfUtils.getInt(conf, "fetcher.threads.start.delay", 10); + // helpers for protocols which can not cancel a fetch themselves; no thread until needed + fetchHelpers = + new FetchTimeoutHelpers(conf, Math.max(1, threadCount * 2), "FetcherTimeout-"); + fetchHelpers.registerMetrics(context, stormConf, metricsTimeBucketSecs); + for (int i = 0; i < threadCount; i++) { if (startDelay > 0 && i > 0) { // short delay to avoid that DNS or other resources are temporarily @@ -1183,6 +1180,9 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) { @Override public void cleanup() { super.cleanup(); + if (fetchHelpers != null) { + fetchHelpers.shutdown(); + } protocolFactory.cleanup(); } diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java index 612d2ede0..c4122f94d 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java @@ -28,13 +28,7 @@ import java.text.SimpleDateFormat; import java.util.Locale; import java.util.Map; -import java.util.concurrent.CancellationException; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.lang3.StringUtils; import org.apache.http.HttpHeaders; @@ -52,10 +46,12 @@ import org.apache.stormcrawler.metrics.ScopedCounter; import org.apache.stormcrawler.metrics.ScopedReducedMetric; import org.apache.stormcrawler.persistence.Status; +import org.apache.stormcrawler.protocol.FetchTimeoutException; import org.apache.stormcrawler.protocol.Protocol; import org.apache.stormcrawler.protocol.ProtocolFactory; import org.apache.stormcrawler.protocol.ProtocolResponse; import org.apache.stormcrawler.protocol.RobotRules; +import org.apache.stormcrawler.protocol.RobotRulesParser; import org.apache.stormcrawler.util.ConfUtils; import org.apache.stormcrawler.util.URLUtil; import org.slf4j.LoggerFactory; @@ -126,10 +122,8 @@ public class SimpleFetcherBolt extends StatusEmitterBolt { // by default remains as is-pre 1.17 private String protocolMetadataPrefix = ""; - /** Hard timeout in seconds for a single protocol fetch. -1 means disabled. */ - private long fetchTimeout = -1; - - private ExecutorService fetchExecutor; + /** Runs protocol calls under fetcher.thread.timeout, see {@link FetchTimeoutHelpers}. */ + private FetchTimeoutHelpers fetchHelpers; private void checkConfiguration() { @@ -222,17 +216,8 @@ public void prepare( ConfUtils.getString( conf, ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, protocolMetadataPrefix); - this.fetchTimeout = - ConfUtils.getLong(conf, FetcherBolt.FETCH_TIMEOUT_PARAM_KEY, fetchTimeout); - if (fetchTimeout > 0) { - fetchExecutor = - Executors.newSingleThreadExecutor( - r -> { - Thread t = new Thread(r, "SimpleFetcherTimeout #" + taskId); - t.setDaemon(true); - return t; - }); - } + fetchHelpers = new FetchTimeoutHelpers(conf, 2, "SimpleFetcherTimeout-" + taskId + "-"); + fetchHelpers.registerMetrics(context, conf, metricsTimeBucketSecs); } @Override @@ -246,8 +231,8 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) { public void cleanup() { super.cleanup(); protocolFactory.cleanup(); - if (fetchExecutor != null) { - fetchExecutor.shutdownNow(); + if (fetchHelpers != null) { + fetchHelpers.shutdown(); } } @@ -301,7 +286,21 @@ public void execute(Tuple input) { Protocol protocol = protocolFactory.getProtocol(url); - BaseRobotRules rules = protocol.getRobotRules(urlString); + BaseRobotRules rules; + try { + rules = + fetchHelpers.call( + () -> protocol.getRobotRules(urlString), + protocol, + urlString, + metadata); + } catch (FetchTimeoutException e) { + // same outcome as with okhttp, where HttpRobotRulesParser turns a failed + // lookup into empty rules: the page is fetched without rules + LOG.info("[Fetcher #{}] robots.txt lookup timed out for {}", taskId, urlString); + eventCounter.scope("robots.timeout").incrBy(1); + rules = RobotRulesParser.EMPTY_RULES; + } boolean fromCache = false; if (rules instanceof RobotRules && ((RobotRules) rules).getContentLengthFetched().length == 0) { @@ -446,28 +445,12 @@ public void execute(Tuple input) { final String fetchUrl = urlString; final Metadata fetchMetadata = metadata; ProtocolResponse response; - if (fetchExecutor != null) { - Future future = - fetchExecutor.submit( - () -> protocol.getProtocolOutput(fetchUrl, fetchMetadata)); - try { - response = future.get(fetchTimeout, TimeUnit.SECONDS); - } catch (TimeoutException e) { - future.cancel(true); - throw new Exception( - "Fetch timed out after " + fetchTimeout + "s fetching " + urlString, e); - } catch (CancellationException e) { - throw new Exception("Fetch cancelled for " + urlString); - } catch (ExecutionException e) { - Throwable cause = e.getCause(); - if (cause instanceof Exception) { - throw (Exception) cause; - } - throw new Exception(cause); - } - } else { - response = protocol.getProtocolOutput(urlString, metadata); - } + response = + fetchHelpers.call( + () -> protocol.getProtocolOutput(fetchUrl, fetchMetadata), + protocol, + urlString, + fetchMetadata); long timeFetching = System.currentTimeMillis() - start; final int byteLength = response.getContent().length; @@ -581,6 +564,15 @@ public void execute(Tuple input) { org.apache.stormcrawler.Constants.StatusStreamName, input, values4status); } + } catch (FetchTimeoutHelpers.SaturatedException e) { + // the URL never reached the network: acked without a status so that the spout + // retries it later, rather than taking a strike towards max.fetch.errors + eventCounter.scope("fetch.helper.rejected").incrBy(1); + LOG.warn( + "[Fetcher #{}] {}: all {} fetch helpers are busy", + taskId, + e.getMessage(), + fetchHelpers.maxHelpers()); } catch (Exception exece) { String message = exece.getMessage(); @@ -589,10 +581,14 @@ public void execute(Tuple input) { } // common exceptions for which we log only a short message - if (exece.getCause() instanceof java.util.concurrent.TimeoutException - || message.contains(" timed out")) { + if (exece instanceof java.io.InterruptedIOException || message.contains(" timed out")) { LOG.error("Socket timeout fetching {}", urlString); 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.error("Unknown host {}", urlString); diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java index b19760b93..db43d9a2b 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java @@ -74,7 +74,7 @@ public class DelegatorProtocol implements Protocol { protected static final org.slf4j.Logger LOG = LoggerFactory.getLogger(DelegatorProtocol.class); - private static final String ROBOTS = "robots.txt"; + static final String ROBOTS = "robots.txt"; static class Filter { @@ -157,6 +157,10 @@ public ProtocolResponse getProtocolOutput(String url, Metadata metadata) throws return protoInstance.getProtocolOutput(url, metadata); } + public boolean supportsFetchTimeout(String url, Metadata metadata) { + return protoInstance.supportsFetchTimeout(url, metadata); + } + public BaseRobotRules getRobotRules(String url) { return protoInstance.getRobotRules(url); } @@ -297,9 +301,7 @@ final FilteredProtocol getProtocolFor(String url, Metadata metadata) { @Override public @NotNull BaseRobotRules getRobotRules(@NotNull String url) { - final Metadata m = new Metadata(); - m.addValue(ROBOTS, "true"); - FilteredProtocol proto = getProtocolFor(url, m); + FilteredProtocol proto = getProtocolFor(url, robotsMetadata()); if (proto == null) { throw new RuntimeException("No sub protocol eligible to retrieve robots"); } @@ -320,6 +322,28 @@ final FilteredProtocol getProtocolFor(String url, Metadata metadata) { return proto.getProtocolOutput(url, metadata); } + /** + * Resolved per URL: true when the delegate the URL is routed to enforces the deadline itself. + * The robots.txt lookup of the URL may be routed differently (see {@link + * #getRobotRules(String)}), so that delegate must support it as well: a lookup run inline on a + * delegate which can not cancel it would hang the fetcher thread. + */ + @Override + public boolean supportsFetchTimeout(String url, Metadata metadata) { + final FilteredProtocol forFetch = getProtocolFor(url, metadata); + final FilteredProtocol forRobots = getProtocolFor(url, robotsMetadata()); + return forFetch != null + && forRobots != null + && forFetch.supportsFetchTimeout(url, metadata) + && forRobots.supportsFetchTimeout(url, metadata); + } + + private static Metadata robotsMetadata() { + final Metadata m = new Metadata(); + m.addValue(ROBOTS, "true"); + return m; + } + @Override public void cleanup() { for (FilteredProtocol p : protocols) { diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/FetchTimeout.java b/core/src/main/java/org/apache/stormcrawler/protocol/FetchTimeout.java new file mode 100644 index 000000000..e4b1d195d --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/protocol/FetchTimeout.java @@ -0,0 +1,55 @@ +/* + * 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.protocol; + +import java.util.Map; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.util.ConfUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Reads {@code fetcher.thread.timeout} from the configuration, the same way for every user. */ +public final class FetchTimeout { + + private static final Logger LOG = LoggerFactory.getLogger(FetchTimeout.class); + + private FetchTimeout() {} + + /** + * The deadline in seconds, or -1 when disabled. Never larger than {@code + * topology.message.timeout.secs}: Storm would fail the tuple first anyway, and with okhttp the + * per-call deadline replaces the client-level call timeout derived from the message timeout, + * which must not be loosened. + */ + public static long secs(Map conf) { + long timeout = ConfUtils.getLong(conf, Constants.FETCH_TIMEOUT_PARAM_KEY, -1); + if (timeout <= 0) { + return -1; + } + final long messageTimeout = ConfUtils.getLong(conf, "topology.message.timeout.secs", -1); + if (messageTimeout > 0 && timeout > messageTimeout) { + LOG.warn( + "{} ({}s) is larger than topology.message.timeout.secs ({}s): using the latter", + Constants.FETCH_TIMEOUT_PARAM_KEY, + timeout, + messageTimeout); + timeout = messageTimeout; + } + return timeout; + } +} diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/FetchTimeoutException.java b/core/src/main/java/org/apache/stormcrawler/protocol/FetchTimeoutException.java new file mode 100644 index 000000000..eb75c66a1 --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/protocol/FetchTimeoutException.java @@ -0,0 +1,50 @@ +/* + * 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.protocol; + +import java.io.InterruptedIOException; + +/** + * Thrown when a protocol call did not complete within {@code fetcher.thread.timeout}: by a {@link + * Protocol} which enforces the deadline itself (okhttp cancels the call), or by the fetcher bolts + * when the call ran on a helper thread and was abandoned there. + * + *

It is an {@link InterruptedIOException} so that it is classified like any other timeout, but + * can be told apart from the socket timeouts of the protocol ({@code http.timeout}). + */ +public class FetchTimeoutException extends InterruptedIOException { + + private final long timeoutSecs; + + public FetchTimeoutException(String url, long timeoutSecs) { + this(url, timeoutSecs, null); + } + + public FetchTimeoutException(String url, long timeoutSecs, Throwable cause) { + super("Fetch timed out after " + timeoutSecs + "s fetching " + url); + this.timeoutSecs = timeoutSecs; + if (cause != null) { + initCause(cause); + } + } + + /** The deadline that passed, in seconds. */ + public long getTimeoutSecs() { + return timeoutSecs; + } +} diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java index 242395235..3bbdde803 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java @@ -53,6 +53,20 @@ public interface Protocol { */ ProtocolResponse getProtocolOutput(String url, Metadata metadata) throws Exception; + /** + * Whether this protocol enforces {@code fetcher.thread.timeout} itself for the given URL, by + * cancelling the request when the deadline passes and throwing a {@link FetchTimeoutException}. + * When true the fetcher bolts call {@link #getProtocolOutput(String, Metadata)} and {@link + * #getRobotRules(String)} directly instead of running them on a helper thread that they abandon + * on timeout. Defaults to false. + * + * @param url the URL about to be fetched, or whose robots.txt is about to be looked up + * @param metadata the metadata of that URL, possibly null + */ + default boolean supportsFetchTimeout(String url, Metadata metadata) { + return false; + } + BaseRobotRules getRobotRules(String url); void cleanup(); diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java index cda61b47d..bf4e144b6 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java @@ -78,6 +78,8 @@ import org.apache.stormcrawler.Metadata; import org.apache.stormcrawler.filtering.URLFilters; import org.apache.stormcrawler.protocol.AbstractHttpProtocol; +import org.apache.stormcrawler.protocol.FetchTimeout; +import org.apache.stormcrawler.protocol.FetchTimeoutException; import org.apache.stormcrawler.protocol.IPFilterRules; import org.apache.stormcrawler.protocol.ProtocolResponse; import org.apache.stormcrawler.protocol.ProtocolResponse.TrimmedContentReason; @@ -107,6 +109,9 @@ public class HttpProtocol extends AbstractHttpProtocol { private int completionTimeout = -1; + /** Per-call deadline in seconds from fetcher.thread.timeout, -1 when disabled. */ + private long fetchTimeout = -1; + /** Accept partially fetched content as trimmed content */ private boolean partialContentAsTrimmed = false; @@ -210,6 +215,10 @@ public void configure(Config conf) { this.completionTimeout = ConfUtils.getInt(conf, "topology.message.timeout.secs", completionTimeout); + // clamped to the message timeout: the per-call deadline replaces the client-level + // callTimeout derived from it and must not loosen it + this.fetchTimeout = FetchTimeout.secs(conf); + this.partialContentAsTrimmed = ConfUtils.getBoolean(conf, "http.content.partial.as.trimmed", false); @@ -559,6 +568,11 @@ private void stripCredentialHeaders(Request.Builder builder) { } } + @Override + public boolean supportsFetchTimeout(String url, Metadata metadata) { + return fetchTimeout > 0; + } + @Override public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) throws Exception { @@ -721,6 +735,9 @@ public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) // every hop creates its own Call and DNS timing entry: track them all // so intermediate entries are cleaned up too, including on exceptions final List hopCalls = new ArrayList<>(); + // fetcher.thread.timeout, shared by all the hops of the chain + final long deadlineNanos = + fetchTimeout > 0 ? System.nanoTime() + TimeUnit.SECONDS.toNanos(fetchTimeout) : 0; try { for (int hops = 0; hops <= maxRedirectHops; hops++) { @@ -731,6 +748,16 @@ public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) } call = fetchClient.newCall(currentRequest); hopCalls.add(call); + if (deadlineNanos != 0) { + // hard deadline for the whole chain, enforced by okio's watchdog: on + // expiry the call is cancelled, the socket closed and execute() or the + // body read throw immediately. Every hop gets the time that is left + final long remaining = deadlineNanos - System.nanoTime(); + if (remaining <= 0) { + throw new FetchTimeoutException(url, fetchTimeout); + } + call.timeout().timeout(remaining, TimeUnit.NANOSECONDS); + } try { lastResponse = call.execute(); } catch (IOException | RuntimeException e) { @@ -889,6 +916,13 @@ public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) } return new ProtocolResponse(bytes, response.code(), responsemetadata); + } catch (InterruptedIOException e) { + if (deadlineNanos != 0 && call != null && call.isCanceled()) { + // cancelled by the watchdog at the deadline: tell it apart from the socket + // timeouts of http.timeout, which are SocketTimeoutExceptions + throw new FetchTimeoutException(url, fetchTimeout, e); + } + throw e; } finally { if (lastResponse != null) { lastResponse.close(); diff --git a/core/src/main/resources/crawler-default.yaml b/core/src/main/resources/crawler-default.yaml index 1f3a0be2a..3b6c6f3f6 100644 --- a/core/src/main/resources/crawler-default.yaml +++ b/core/src/main/resources/crawler-default.yaml @@ -29,9 +29,20 @@ config: fetcher.max.urls.in.queues: -1 fetcher.max.queue.size: -1 fetcher.timeout.queue: -1 - # hard timeout in seconds for a single protocol fetch at the bolt level; - # -1 disables (relies on protocol-level socket timeouts only) + # hard deadline in seconds for each protocol call of a URL: the robots.txt + # lookup (and each redirect it follows) and the page fetch (its redirect + # chain shares one deadline), independent of the protocol's socket + # timeouts; -1 disables it. Never exceeds topology.message.timeout.secs. + # With okhttp the HTTP call itself is cancelled when the deadline passes + # (socket closed, URL reported as FETCH_ERROR); other protocols run on a + # helper thread and are abandoned there on timeout. A robots.txt lookup + # which times out lets the page be fetched without rules on both paths fetcher.thread.timeout: -1 + # max helper threads per bolt for the abandoned fetches above; default is + # 2 x fetcher.threads.number (2 for SimpleFetcherBolt). When all are busy + # the URL is acked without a status, like one which waited too long in + # the queue, and the spout retries it later + # fetcher.thread.timeout.helpers: 100 # max. crawl-delay accepted in robots.txt (in seconds) fetcher.max.crawl.delay: 30 # behavior of fetcher when the crawl-delay in the robots.txt diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java index 35cd7b5f5..c3a3dc792 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java @@ -43,6 +43,7 @@ import org.apache.stormcrawler.TestUtil; import org.apache.stormcrawler.persistence.Status; import org.apache.stormcrawler.protocol.ProtocolFactory; +import org.apache.stormcrawler.protocol.StuckProtocol; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -53,8 +54,10 @@ abstract class AbstractFetcherBoltTest { BaseRichBolt bolt; @AfterEach - void cleanupParserBolt() { + void cleanupParserBolt() throws ReflectiveOperationException { bolt.cleanup(); + // the factory is a singleton configured once: never leak a protocol into the next test + resetProtocolFactory(); } @Test @@ -153,6 +156,133 @@ void testThreadTimeout(WireMockRuntimeInfo wmRuntimeInfo) { Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size()); } + /** + * A fetch that hits the bolt-level timeout must not hold up the fetches that follow it: with + * one fetcher thread, a stuck fetch followed by two fast ones must yield two pages and one + * FETCH_ERROR within a few seconds, not one FETCH_ERROR per URL. + */ + @Test + void stuckFetchDoesNotBlockTheFollowingFetches(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlMatching("/slow")) + .willReturn(aResponse().withStatus(200).withFixedDelay(10_000))); + stubFor(get(urlMatching("/fast.*")).willReturn(aResponse().withStatus(200).withBody("ok"))); + + resetProtocolFactory(); + TestOutputCollector output = new TestOutputCollector(); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.threads.number", 1); + config.put("fetcher.thread.timeout", 1L); + config.put("http.timeout", 30_000); + // same host: the second and third URL wait for the first one to release the queue + config.put("fetcher.server.delay", 0.0f); + bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + + String base = "http://localhost:" + wmRuntimeInfo.getHttpPort(); + for (String path : new String[] {"/slow", "/fast1", "/fast2"}) { + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("source"); + when(tuple.getStringByField("url")).thenReturn(base + path); + when(tuple.getValueByField("metadata")).thenReturn(null); + bolt.execute(tuple); + } + + await().atMost(6, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 3); + + List> statusTuples = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(1, statusTuples.size(), "only the slow URL should fail"); + Assertions.assertEquals(base + "/slow", statusTuples.get(0).get(0)); + Assertions.assertEquals(Status.FETCH_ERROR, statusTuples.get(0).get(2)); + Assertions.assertEquals(2, output.getEmitted(Utils.DEFAULT_STREAM_ID).size()); + } + + /** + * With a protocol that cannot be cancelled, timed-out fetches are abandoned on helper threads + * from a bounded pool shared by the bolt: every fetch actually starts until the pool is full, + * and the next one is rejected right away instead of queueing behind a stuck helper. + */ + @Test + void abandonedFetchesUseABoundedSharedPool() throws ReflectiveOperationException { + StuckProtocol.STARTED.set(0); + resetProtocolFactory(); + TestOutputCollector output = new TestOutputCollector(); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("http.protocol.implementation", StuckProtocol.class.getName()); + config.put("fetcher.threads.number", 1); + config.put("fetcher.thread.timeout", 1L); + config.put("fetcher.thread.timeout.helpers", 2); + config.put("fetcher.server.delay", 0.0f); + bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + + for (String path : new String[] {"/1", "/2", "/3"}) { + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("source"); + when(tuple.getStringByField("url")).thenReturn("http://stuck.example.com" + path); + when(tuple.getValueByField("metadata")).thenReturn(null); + bolt.execute(tuple); + } + + await().atMost(6, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 3); + + // pool of 2 (twice the fetcher threads): the first two fetches really started and + // timed out + Assertions.assertEquals(2, StuckProtocol.STARTED.get(), "fetches actually started"); + List> statusTuples = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(2, statusTuples.size()); + for (List t : statusTuples) { + Assertions.assertEquals(Status.FETCH_ERROR, t.get(2)); + Assertions.assertEquals( + "Socket timeout fetching", + ((Metadata) t.get(1)).getFirstValue("fetch.exception")); + } + // the third found no free helper: it never reached the network, so it is acked + // without a status, like a URL which waited too long in the queue, and the spout + // will retry it + Assertions.assertEquals(0, output.getFailedTuples().size()); + Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size()); + } + + /** + * The robots.txt lookup is covered by the timeout too, and a lookup which times out has the + * same outcome on every path: the page is fetched without rules, as HttpRobotRulesParser does + * with okhttp. Here the fetch hangs as well, so the URL ends up as FETCH_ERROR after two + * deadlines rather than one, and the fetch was really attempted. + */ + @Test + void hangingRobotsLookupDoesNotFailTheUrl() throws ReflectiveOperationException { + StuckProtocol.STARTED.set(0); + resetProtocolFactory(); + TestOutputCollector output = new TestOutputCollector(); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("http.protocol.implementation", StuckProtocol.class.getName()); + config.put(StuckProtocol.HANG_ROBOTS_KEY, true); + config.put("fetcher.threads.number", 1); + config.put("fetcher.thread.timeout", 1L); + bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("source"); + when(tuple.getStringByField("url")).thenReturn("http://stuck.example.com/robots"); + when(tuple.getValueByField("metadata")).thenReturn(null); + long start = System.currentTimeMillis(); + bolt.execute(tuple); + + await().atMost(6, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 1); + long elapsed = System.currentTimeMillis() - start; + Assertions.assertTrue(elapsed >= 2_000 && elapsed < 5_000, "took " + elapsed + " ms"); + Assertions.assertEquals(1, StuckProtocol.STARTED.get(), "the page fetch was attempted"); + List> statusTuples = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(1, statusTuples.size()); + Assertions.assertEquals(Status.FETCH_ERROR, statusTuples.get(0).get(2)); + Assertions.assertEquals( + "Socket timeout fetching", + ((Metadata) statusTuples.get(0).get(1)).getFirstValue("fetch.exception")); + } + @Test void invalidProxyMetadataEmitsFetchError(WireMockRuntimeInfo wmRuntimeInfo) throws ReflectiveOperationException { diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java new file mode 100644 index 000000000..6bb5504f4 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java @@ -0,0 +1,219 @@ +/* + * 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.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.protocol.DummyProtocol; +import org.apache.stormcrawler.protocol.FetchTimeoutException; +import org.apache.stormcrawler.protocol.Protocol; +import org.apache.stormcrawler.protocol.StuckProtocol; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(value = 20, unit = TimeUnit.SECONDS) +class FetchTimeoutHelpersTest { + + /** A protocol which claims to enforce the timeout itself. */ + private static final Protocol SELF_TIMING = + new DummyProtocol() { + @Override + public boolean supportsFetchTimeout(String url, Metadata metadata) { + return true; + } + }; + + private static final Protocol PLAIN = new DummyProtocol(); + + private FetchTimeoutHelpers helpers; + + private static Map conf(long timeoutSecs, Integer maxHelpers) { + Map conf = new HashMap<>(); + conf.put(Constants.FETCH_TIMEOUT_PARAM_KEY, timeoutSecs); + if (maxHelpers != null) { + conf.put(Constants.FETCH_TIMEOUT_HELPERS_PARAM_KEY, maxHelpers); + } + return conf; + } + + private FetchTimeoutHelpers helpers(long timeoutSecs, Integer maxHelpers) { + helpers = new FetchTimeoutHelpers(conf(timeoutSecs, maxHelpers), 4, "test-helper-"); + return helpers; + } + + @AfterEach + void shutdown() { + if (helpers != null) { + helpers.shutdown(); + } + } + + @Test + void disabledRunsOnTheCallingThread() throws Exception { + FetchTimeoutHelpers h = helpers(-1, null); + Assertions.assertFalse(h.enabled()); + Thread caller = Thread.currentThread(); + Thread ran = h.call(Thread::currentThread, PLAIN, "http://a.net/", null); + Assertions.assertSame(caller, ran); + Assertions.assertEquals(0, h.largestPoolSize()); + } + + @Test + void protocolEnforcingTheTimeoutRunsOnTheCallingThread() throws Exception { + FetchTimeoutHelpers h = helpers(1, null); + Assertions.assertTrue(h.enabled()); + Assertions.assertEquals(1, h.timeoutSecs()); + Thread ran = h.call(Thread::currentThread, SELF_TIMING, "http://a.net/", null); + Assertions.assertSame(Thread.currentThread(), ran); + Assertions.assertEquals(0, h.largestPoolSize()); + } + + @Test + void otherProtocolsRunOnAHelperAndReturnTheResult() throws Exception { + FetchTimeoutHelpers h = helpers(5, null); + Thread ran = h.call(Thread::currentThread, PLAIN, "http://a.net/", null); + Assertions.assertNotSame(Thread.currentThread(), ran); + Assertions.assertTrue(ran.getName().startsWith("test-helper-"), ran.getName()); + Assertions.assertTrue(ran.isDaemon()); + Assertions.assertEquals(1, h.largestPoolSize()); + } + + @Test + void protocolExceptionsPropagateUnwrapped() { + FetchTimeoutHelpers h = helpers(5, null); + IOException thrown = + Assertions.assertThrows( + IOException.class, + () -> + h.call( + () -> { + throw new IOException("boom"); + }, + PLAIN, + "http://a.net/", + null)); + Assertions.assertEquals("boom", thrown.getMessage()); + } + + @Test + void errorsAreWrappedInAnException() { + FetchTimeoutHelpers h = helpers(5, null); + Exception thrown = + Assertions.assertThrows( + Exception.class, + () -> + h.call( + () -> { + throw new AssertionError("not an exception"); + }, + PLAIN, + "http://a.net/", + null)); + Assertions.assertInstanceOf(AssertionError.class, thrown.getCause()); + } + + @Test + void deadlineThrowsTypedTimeoutAndInterruptsTheHelper() throws Exception { + FetchTimeoutHelpers h = helpers(1, null); + AtomicBoolean interrupted = new AtomicBoolean(); + CountDownLatch done = new CountDownLatch(1); + long start = System.currentTimeMillis(); + FetchTimeoutException thrown = + Assertions.assertThrows( + FetchTimeoutException.class, + () -> + h.call( + () -> { + try { + Thread.sleep(10_000); + } catch (InterruptedException e) { + interrupted.set(true); + } + done.countDown(); + return null; + }, + PLAIN, + "http://a.net/page", + null)); + Assertions.assertTrue(System.currentTimeMillis() - start < 3_000); + Assertions.assertTrue(thrown.getMessage().contains("http://a.net/page")); + // the helper is asked to stop, even though not every protocol honours it + Assertions.assertTrue(done.await(5, TimeUnit.SECONDS)); + Assertions.assertTrue(interrupted.get()); + } + + @Test + void saturationFailsFastWithTypedException() throws Exception { + FetchTimeoutHelpers h = helpers(1, 2); + Assertions.assertEquals(2, h.maxHelpers()); + StuckProtocol stuck = new StuckProtocol(); + // fill the two helpers with fetches that never return + for (int i = 0; i < 2; i++) { + Assertions.assertThrows( + FetchTimeoutException.class, + () -> + h.call( + () -> stuck.getProtocolOutput("http://a.net/", null), + stuck, + "u", + null)); + } + long start = System.currentTimeMillis(); + Assertions.assertThrows( + FetchTimeoutHelpers.SaturatedException.class, + () -> h.call(() -> "never", stuck, "http://a.net/3", null)); + Assertions.assertTrue(System.currentTimeMillis() - start < 500, "rejected immediately"); + Assertions.assertEquals(2, h.largestPoolSize()); + } + + /** The deadline is clamped to the message timeout on the helper path as well. */ + @Test + void timeoutIsClampedToTheMessageTimeout() { + Map conf = conf(60, null); + conf.put("topology.message.timeout.secs", 5); + helpers = new FetchTimeoutHelpers(conf, 4, "test-helper-"); + Assertions.assertEquals(5, helpers.timeoutSecs()); + } + + @Test + void poolBoundComesFromConfigOrDefault() { + Assertions.assertEquals(4, helpers(1, null).maxHelpers()); + helpers.shutdown(); + Assertions.assertEquals(7, helpers(1, 7).maxHelpers()); + helpers.shutdown(); + // never less than one helper + Assertions.assertEquals(1, helpers(1, 0).maxHelpers()); + } + + @Test + void afterShutdownCallsAreRejected() { + FetchTimeoutHelpers h = helpers(1, null); + h.shutdown(); + Assertions.assertThrows( + FetchTimeoutHelpers.SaturatedException.class, + () -> h.call(() -> "x", PLAIN, "http://a.net/", null)); + } +} diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java index e7f592e14..9d8ddd719 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java @@ -31,6 +31,7 @@ import org.apache.stormcrawler.Constants; import org.apache.stormcrawler.Metadata; import org.apache.stormcrawler.persistence.Status; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -138,4 +139,39 @@ void unforcedLongCrawlDelayStillEmitsCrawlDelayErrorWithoutMetadata( assertEquals("crawl_delay", md.getFirstValue(Constants.STATUS_ERROR_CAUSE)); assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); } + + @Test + void noHelperThreadsWithOkhttpAndFetchTimeout(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello"))); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.thread.timeout", 5L); + fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertEquals( + 0, + ((FetcherBolt) bolt).helperPoolSize(), + "okhttp cancels the call itself: no helper threads expected"); + } + + /** With okhttp the robots.txt fetch goes through the same call deadline as the page. */ + @Test + void slowRobotsTxtIsBoundedByTheFetchTimeoutWithOkhttp(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn(aResponse().withStatus(200).withFixedDelay(10_000))); + stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello"))); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("http.timeout", 30_000); + config.put("fetcher.thread.timeout", 1L); + long start = System.currentTimeMillis(); + // the robots.txt lookup fails at the deadline; the parser then allows the fetch + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + Assertions.assertNotNull(md); + Assertions.assertTrue( + System.currentTimeMillis() - start < 6_000, "robots.txt lookup was not bounded"); + assertEquals(0, ((FetcherBolt) bolt).helperPoolSize()); + } } diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/DelegationProtocolTest.java b/core/src/test/java/org/apache/stormcrawler/protocol/DelegationProtocolTest.java index 9b6ba8bce..079d82df4 100644 --- a/core/src/test/java/org/apache/stormcrawler/protocol/DelegationProtocolTest.java +++ b/core/src/test/java/org/apache/stormcrawler/protocol/DelegationProtocolTest.java @@ -18,6 +18,10 @@ package org.apache.stormcrawler.protocol; import java.io.FileNotFoundException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import org.apache.storm.Config; import org.apache.stormcrawler.Metadata; import org.apache.stormcrawler.protocol.DelegatorProtocol.FilteredProtocol; @@ -73,4 +77,74 @@ void getProtocolTest() throws FileNotFoundException { pf = superProto.getProtocolFor("https://www.example-two.com/large.doc", meta); Assertions.assertEquals("fourth", pf.id); } + + private static DelegatorProtocol delegator(String... classNames) { + Config conf = new Config(); + conf.put("http.agent.name", "this_is_only_a_test"); + conf.put("fetcher.thread.timeout", 1L); + List> entries = new ArrayList<>(); + for (int i = 0; i < classNames.length; i++) { + Map entry = new HashMap<>(); + entry.put("className", classNames[i]); + entry.put("id", "p" + i); + if (i < classNames.length - 1) { + entry.put("filters", Map.of("key" + i, "value")); + } + entries.add(entry); + } + conf.put("protocol.delegator.config", entries); + DelegatorProtocol delegator = new DelegatorProtocol(); + delegator.configure(conf); + return delegator; + } + + /** + * The capability is resolved per URL: a URL routed to okhttp is timed by okhttp even when + * another delegate can not cancel its fetches. The filters of the test delegator match on the + * metadata key "key0" for the first delegate, the last one is the default. + */ + @Test + void supportsFetchTimeoutIsResolvedPerUrl() { + String okhttp = org.apache.stormcrawler.protocol.okhttp.HttpProtocol.class.getName(); + String dummy = DummyProtocol.class.getName(); + String url = "http://example.com/page"; + Metadata toFirst = new Metadata(); + toFirst.setValue("key0", "value"); + Metadata toDefault = new Metadata(); + + Assertions.assertTrue(delegator(okhttp, okhttp).supportsFetchTimeout(url, toDefault)); + // routed to the default delegate, okhttp + Assertions.assertTrue(delegator(dummy, okhttp).supportsFetchTimeout(url, toDefault)); + // routed to the first delegate, which can not cancel + Assertions.assertFalse(delegator(dummy, okhttp).supportsFetchTimeout(url, toFirst)); + Assertions.assertFalse(delegator(okhttp, dummy).supportsFetchTimeout(url, toDefault)); + // the page goes to okhttp but the robots.txt lookup, routed without "key0", goes to + // the default delegate which can not cancel: the helper path is needed for the lookup + Assertions.assertFalse(delegator(okhttp, dummy).supportsFetchTimeout(url, toFirst)); + } + + /** + * The robots.txt lookup is routed with its own metadata: a delegate matching on it must support + * the timeout too, otherwise the lookup would run inline on a protocol which can not cancel it. + */ + @Test + void supportsFetchTimeoutRequiresTheRobotsDelegateToo() { + String okhttp = org.apache.stormcrawler.protocol.okhttp.HttpProtocol.class.getName(); + String dummy = DummyProtocol.class.getName(); + Config conf = new Config(); + conf.put("http.agent.name", "this_is_only_a_test"); + conf.put("fetcher.thread.timeout", 1L); + Map robotsDelegate = new HashMap<>(); + robotsDelegate.put("className", dummy); + robotsDelegate.put("id", "robots"); + robotsDelegate.put("filters", Map.of(DelegatorProtocol.ROBOTS, "true")); + Map pages = new HashMap<>(); + pages.put("className", okhttp); + pages.put("id", "pages"); + conf.put("protocol.delegator.config", List.of(robotsDelegate, pages)); + DelegatorProtocol delegator = new DelegatorProtocol(); + delegator.configure(conf); + Assertions.assertFalse( + delegator.supportsFetchTimeout("http://example.com/page", new Metadata())); + } } diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java b/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java new file mode 100644 index 000000000..59bf94278 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java @@ -0,0 +1,77 @@ +/* + * 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.protocol; + +import crawlercommons.robots.BaseRobotRules; +import crawlercommons.robots.SimpleRobotRules; +import crawlercommons.robots.SimpleRobotRules.RobotRulesMode; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.storm.Config; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.util.ConfUtils; + +/** + * A protocol whose fetches hang for a long time and ignore interruption, like a client blocked on a + * socket. Counts the fetches that actually started. + */ +public class StuckProtocol implements Protocol { + + public static final AtomicInteger STARTED = new AtomicInteger(); + + public static final long HANG_MILLIS = 10_000; + + /** Configuration key: when true, robots.txt lookups hang as well. */ + public static final String HANG_ROBOTS_KEY = "stuck.protocol.hang.robots"; + + private boolean hangRobots = false; + + @Override + public void configure(Config conf) { + hangRobots = ConfUtils.getBoolean(conf, HANG_ROBOTS_KEY, false); + } + + @Override + public ProtocolResponse getProtocolOutput(String url, Metadata metadata) throws Exception { + STARTED.incrementAndGet(); + hang(); + return new ProtocolResponse("late".getBytes(StandardCharsets.UTF_8), 200, new Metadata()); + } + + @Override + public BaseRobotRules getRobotRules(String url) { + if (hangRobots) { + hang(); + } + return new SimpleRobotRules(RobotRulesMode.ALLOW_ALL); + } + + private static void hang() { + long until = System.currentTimeMillis() + HANG_MILLIS; + while (System.currentTimeMillis() < until) { + try { + Thread.sleep(until - System.currentTimeMillis()); + } catch (InterruptedException e) { + // ignored on purpose: an interrupt does not free a blocked socket read either + } + } + } + + @Override + public void cleanup() {} +} diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocolFetchTimeoutTest.java b/core/src/test/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocolFetchTimeoutTest.java new file mode 100644 index 000000000..9fa6483f9 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocolFetchTimeoutTest.java @@ -0,0 +1,198 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.protocol.okhttp; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.get; +import static com.github.tomakehurst.wiremock.client.WireMock.stubFor; +import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; + +import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo; +import com.github.tomakehurst.wiremock.junit5.WireMockTest; +import java.io.InterruptedIOException; +import org.apache.storm.Config; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.protocol.FetchTimeoutException; +import org.apache.stormcrawler.protocol.ProtocolResponse; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +@WireMockTest +class HttpProtocolFetchTimeoutTest { + + private static HttpProtocol protocol(long fetchTimeoutSecs) { + return protocol(fetchTimeoutSecs, new Config()); + } + + private static HttpProtocol protocol(long fetchTimeoutSecs, Config conf) { + conf.put("http.agent.name", "this_is_only_a_test"); + // socket timeouts far above the fetch timeout so that only the latter can fire + conf.put("http.timeout", 30_000); + if (fetchTimeoutSecs > 0) { + conf.put(Constants.FETCH_TIMEOUT_PARAM_KEY, fetchTimeoutSecs); + } + HttpProtocol protocol = new HttpProtocol(); + protocol.configure(conf); + return protocol; + } + + @Test + void supportsFetchTimeoutOnlyWhenConfigured() { + Assertions.assertTrue(protocol(1).supportsFetchTimeout("http://a.net/", null)); + Assertions.assertFalse(protocol(-1).supportsFetchTimeout("http://a.net/", null)); + } + + @Test + void slowResponseIsCancelledAtTheFetchTimeout(WireMockRuntimeInfo wm) { + stubFor( + get(urlEqualTo("/slow")) + .willReturn(aResponse().withStatus(200).withFixedDelay(5_000))); + HttpProtocol protocol = protocol(1); + long start = System.currentTimeMillis(); + Exception thrown = + Assertions.assertThrows( + Exception.class, + () -> + protocol.getProtocolOutput( + wm.getHttpBaseUrl() + "/slow", new Metadata())); + long elapsed = System.currentTimeMillis() - start; + Assertions.assertTrue( + elapsed < 3_000, "fetch was not cancelled at the timeout, took " + elapsed + " ms"); + // the deadline, as opposed to a socket timeout + Assertions.assertInstanceOf(FetchTimeoutException.class, thrown); + Assertions.assertEquals(1, ((FetchTimeoutException) thrown).getTimeoutSecs()); + } + + /** A plain socket timeout (http.timeout) is not reported as the deadline. */ + @Test + void socketTimeoutIsNotTheDeadline(WireMockRuntimeInfo wm) { + stubFor( + get(urlEqualTo("/slow")) + .willReturn(aResponse().withStatus(200).withFixedDelay(5_000))); + Config conf = new Config(); + conf.put("http.agent.name", "this_is_only_a_test"); + // socket timeout below the deadline: it fires first + conf.put("http.timeout", 500); + conf.put(Constants.FETCH_TIMEOUT_PARAM_KEY, 10L); + HttpProtocol protocol = new HttpProtocol(); + protocol.configure(conf); + Exception thrown = + Assertions.assertThrows( + Exception.class, + () -> + protocol.getProtocolOutput( + wm.getHttpBaseUrl() + "/slow", new Metadata())); + Assertions.assertInstanceOf(InterruptedIOException.class, thrown); + Assertions.assertFalse(thrown instanceof FetchTimeoutException, thrown.toString()); + } + + /** The hops of a redirect chain share one deadline: it is per fetch, not per hop. */ + @Test + void redirectChainSharesTheDeadline(WireMockRuntimeInfo wm) { + // each hop takes 700 ms: with a 1 s deadline the chain can not complete, and a per-hop + // deadline would have let it run for over 2 s + stubFor( + get(urlEqualTo("/a")) + .willReturn( + aResponse() + .withStatus(302) + .withHeader("Location", "/b") + .withFixedDelay(700))); + stubFor( + get(urlEqualTo("/b")) + .willReturn( + aResponse() + .withStatus(302) + .withHeader("Location", "/c") + .withFixedDelay(700))); + stubFor( + get(urlEqualTo("/c")) + .willReturn( + aResponse().withStatus(200).withBody("ok").withFixedDelay(700))); + Config conf = new Config(); + conf.put("http.allow.redirects", true); + HttpProtocol protocol = protocol(1, conf); + long start = System.currentTimeMillis(); + Exception thrown = + Assertions.assertThrows( + Exception.class, + () -> + protocol.getProtocolOutput( + wm.getHttpBaseUrl() + "/a", new Metadata())); + long elapsed = System.currentTimeMillis() - start; + Assertions.assertInstanceOf(FetchTimeoutException.class, thrown); + Assertions.assertTrue(elapsed < 1_800, "deadline was not shared by the hops: " + elapsed); + } + + /** The fetch timeout must not loosen the deadline derived from the message timeout. */ + @Test + void fetchTimeoutIsClampedToTheMessageTimeout(WireMockRuntimeInfo wm) { + stubFor( + get(urlEqualTo("/slow")) + .willReturn(aResponse().withStatus(200).withFixedDelay(5_000))); + Config conf = new Config(); + conf.put("topology.message.timeout.secs", 1); + HttpProtocol protocol = protocol(60, conf); + long start = System.currentTimeMillis(); + Assertions.assertThrows( + Exception.class, + () -> protocol.getProtocolOutput(wm.getHttpBaseUrl() + "/slow", new Metadata())); + long elapsed = System.currentTimeMillis() - start; + Assertions.assertTrue(elapsed < 3_000, "message timeout was loosened, took " + elapsed); + } + + /** + * With http.content.partial.as.trimmed the content received before the deadline is kept and + * flagged as trimmed for "time"; without it the deadline is a plain failure. + */ + @Test + void deadlineDuringBodyHonoursPartialContentAsTrimmed(WireMockRuntimeInfo wm) throws Exception { + byte[] body = new byte[20_000]; + stubFor( + get(urlEqualTo("/dribble")) + .willReturn( + aResponse() + .withStatus(200) + // uncompressed, so that bytes reach the buffer as they + // arrive + .withHeader("Content-Encoding", "identity") + .withBody(body) + .withChunkedDribbleDelay(40, 8_000))); + Config keep = new Config(); + keep.put("http.content.partial.as.trimmed", true); + ProtocolResponse response = + protocol(1, keep) + .getProtocolOutput(wm.getHttpBaseUrl() + "/dribble", new Metadata()); + Assertions.assertEquals(200, response.getStatusCode()); + Assertions.assertTrue(response.getContent().length < body.length, "content was cut"); + Assertions.assertEquals( + "true", + response.getMetadata().getFirstValue(ProtocolResponse.TRIMMED_RESPONSE_KEY)); + Assertions.assertEquals( + "time", + response.getMetadata().getFirstValue(ProtocolResponse.TRIMMED_RESPONSE_REASON_KEY)); + + Assertions.assertThrows( + Exception.class, + () -> + protocol(1) + .getProtocolOutput( + wm.getHttpBaseUrl() + "/dribble", new Metadata())); + } +} diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index 093f60e5a..fbac9158d 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -192,6 +192,8 @@ is defined. | fetcher.threads.per.queue | 1 | Default number of threads per queue. Can be overridden. | fetcher.threads.start.delay | 10 | Delay (milliseconds) between starting next fetcher thread. Avoids overloading DNS or network resources during fetcher startup when all threads simultaneously start requesting pages. | fetcher.timeout.queue | -1 | Maximum wait time (seconds) for items in the queue. -1 disables timeout. +| fetcher.thread.timeout | -1 | Hard deadline (seconds) for each protocol call made for a URL, independent of the protocol's own socket timeouts: the robots.txt lookup, each redirect it follows (up to 5, each a call of its own), and the page fetch, whose redirect chain shares a single deadline. Worst case for one URL is therefore several times the value. -1 disables it. The deadline never exceeds `topology.message.timeout.secs`. With the default okhttp protocol the deadline is applied to the HTTP call itself: on expiry the call is cancelled, the socket closed and the URL reported as `FETCH_ERROR` with `fetch.exception` "Socket timeout fetching"; if `http.content.partial.as.trimmed` is on, content already received is kept instead and flagged as trimmed for "time". Protocols which do not enforce the deadline themselves (see `Protocol.supportsFetchTimeout(url, metadata)`, resolved per URL by `DelegatorProtocol`) run the call on a helper thread which is abandoned on timeout. On both paths a robots.txt lookup which times out lets the page be fetched without rules, like any failed lookup. The `fetch.timeout` counter counts every timeout, `fetch.deadline` only the ones where this deadline fired, `robots.timeout` the lookups abandoned on a helper thread. +| fetcher.thread.timeout.helpers | 2 x fetcher.threads.number (2 for SimpleFetcherBolt) | Maximum number of helper threads per bolt instance for the abandoned fetches above. Threads are created on demand and released after a minute idle. When every helper is busy the URL never reaches the network: it is acked without a status, like a URL which waited too long in the queue, so that the spout retries it later, and the `fetch.helper.rejected` counter is incremented. Helpers are shared by all hosts, so a host that never answers can make fetches of other hosts be deferred this way until its helpers time out; the `fetchhelpers` gauge shows how many are busy. | fetcherbolt.queue.debug.filepath | "" | Path to a debug log (e.g. /tmp/fetcher-dump-{port}). | http.agent.description | - | Description for the User-Agent header. | http.agent.email | - | Email address in User-Agent header. From 40380d66aa78fee407a801532739daaaa25acd44 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Wed, 16 Sep 2026 19:09:47 +0200 Subject: [PATCH 2/7] Fix open comments --- .../bolt/FetchTimeoutHelpers.java | 58 +++++++++++-- .../apache/stormcrawler/bolt/FetcherBolt.java | 81 ++++++++++++++++++- .../stormcrawler/bolt/SimpleFetcherBolt.java | 5 +- .../protocol/AbstractHttpProtocol.java | 7 ++ .../protocol/DelegatorProtocol.java | 12 +++ .../protocol/HttpRobotRulesParser.java | 21 +++++ .../stormcrawler/protocol/Protocol.java | 12 +++ .../protocol/RobotRulesParser.java | 10 +++ .../protocol/okhttp/HttpProtocol.java | 4 +- core/src/main/resources/crawler-default.yaml | 13 ++- .../bolt/AbstractFetcherBoltTest.java | 43 ++++++++++ .../bolt/FetchItemQueuesTest.java | 66 +++++++++++++++ .../bolt/FetchTimeoutHelpersTest.java | 47 +++++++++++ .../protocol/HttpRobotRulesParserTest.java | 24 ++++++ .../stormcrawler/protocol/StuckProtocol.java | 26 +++++- docs/src/main/asciidoc/configuration.adoc | 4 +- 16 files changed, 412 insertions(+), 21 deletions(-) diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java index 1a167f347..12c64bab7 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java @@ -22,11 +22,13 @@ import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.apache.storm.task.TopologyContext; import org.apache.stormcrawler.Constants; @@ -49,6 +51,12 @@ * once with {@link SaturatedException}. Helper threads are created on demand and released after a * minute of inactivity: with the default protocol none is ever created. * + *

The bound is a {@link Semaphore} of in-flight calls in front of a queueing executor, not the + * executor's own rejection: a worker hands its result back before it is back polling for work, so a + * pool with a {@code SynchronousQueue} would reject the page fetch submitted right after the + * robots.txt lookup whenever it is at its maximum. A permit is released when the call completes on + * the helper, whether the caller is still waiting for it or has given up. + * *

The deadline is the one of {@link FetchTimeout#secs(Map)}, clamped to the message timeout on * both paths. */ @@ -64,6 +72,9 @@ static final class SaturatedException extends Exception { private final long timeoutSecs; private final ThreadPoolExecutor helpers; + /** One permit per helper: a call runs only if it gets one, and holds it until it completes. */ + private final Semaphore permits; + /** * @param conf the bolt configuration * @param defaultMaxHelpers pool bound used unless {@code fetcher.thread.timeout.helpers} is set @@ -75,25 +86,34 @@ static final class SaturatedException extends Exception { ConfUtils.getInt( conf, Constants.FETCH_TIMEOUT_HELPERS_PARAM_KEY, defaultMaxHelpers); final AtomicInteger helperNum = new AtomicInteger(); + final int bound = Math.max(1, maxHelpers); + this.permits = new Semaphore(bound); + // core == max so a thread is started for every call while fewer than the bound exist; + // the queue only ever holds a call whose helper is between two tasks this.helpers = new ThreadPoolExecutor( - 0, - Math.max(1, maxHelpers), + bound, + bound, 60L, TimeUnit.SECONDS, - new SynchronousQueue<>(), + new LinkedBlockingQueue<>(), r -> { Thread t = new Thread(r, threadNamePrefix + helperNum.incrementAndGet()); t.setDaemon(true); return t; }); + this.helpers.allowCoreThreadTimeOut(true); } - /** Registers the {@code fetchhelpers} gauge: number of helper threads busy with a call. */ + /** Registers the {@code fetchhelpers} gauge: number of helpers busy with a call. */ void registerMetrics(TopologyContext context, Map conf, int bucketSecs) { - CrawlerMetrics.registerGauge( - context, conf, "fetchhelpers", helpers::getActiveCount, bucketSecs); + CrawlerMetrics.registerGauge(context, conf, "fetchhelpers", this::busy, bucketSecs); + } + + /** Number of calls in flight on the helpers, abandoned ones included. */ + int busy() { + return helpers.getMaximumPoolSize() - permits.availablePermits(); } /** Whether a timeout is configured at all. */ @@ -132,10 +152,27 @@ T call(Callable call, Protocol protocol, String url, Metadata metadata) if (timeoutSecs <= 0 || protocol.supportsFetchTimeout(url, metadata)) { return call.call(); } + if (!permits.tryAcquire()) { + throw new SaturatedException(url); + } + // the permit is given back by whoever knows the call is over: the helper once the call + // has run, or the caller if it cancelled the call before the helper ever started it + final AtomicBoolean started = new AtomicBoolean(false); final Future future; try { - future = helpers.submit(call); + future = + helpers.submit( + () -> { + started.set(true); + try { + return call.call(); + } finally { + permits.release(); + } + }); } catch (RejectedExecutionException e) { + // shut down + permits.release(); throw new SaturatedException(url); } try { @@ -143,6 +180,11 @@ T call(Callable call, Protocol protocol, String url, Metadata metadata) } catch (TimeoutException e) { // a courtesy for protocols which do honour interruption future.cancel(true); + // once cancelled the task can no longer start: if it had not, nobody else will + // release the permit + if (!started.get()) { + permits.release(); + } throw new FetchTimeoutException(url, timeoutSecs); } catch (CancellationException e) { throw new Exception("Fetch cancelled for " + url); 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 fb41b2081..8b140b81b 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -35,6 +35,7 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -240,6 +241,9 @@ static class FetchItemQueue { final AtomicLong crawlDelay; + /** Consecutive fetches of this queue rejected because every helper thread was busy. */ + private final AtomicInteger saturations = new AtomicInteger(); + public FetchItemQueue( String id, int maxThreads, long crawlDelay, long minCrawlDelay, int maxQueueSize) { this.id = id; @@ -336,9 +340,35 @@ boolean isReady(long now) { void finish(boolean asap) { inProgress.decrementAndGet(); + saturations.set(0); setNextFetchTime(System.currentTimeMillis(), asap); } + /** + * Like {@link #finish} for a fetch which never ran because every helper thread was busy. + * The helpers are shared by all queues and are not freed by trying again, so instead of + * being ready at once the queue is backed off exponentially: its own delay, doubled at each + * consecutive rejection, up to {@code maxBackoff}. Any other outcome resets the backoff. + * + *

The delay is spread with equal jitter, between half the computed value and the value + * itself: a stuck protocol rejects the fetches of many queues within the same instant, and + * without jitter they would all become ready together and hit the pool as a herd again. + * + * @return the delay applied, in milliseconds + */ + long finishSaturated(long maxBackoff) { + inProgress.decrementAndGet(); + // 2^30 x 1 s is already far beyond any sensible cap: bound the shift, not the counter + final int rejections = Math.min(saturations.incrementAndGet(), 30); + final long base = + Math.max(1000L, maxThreads > 1 ? minCrawlDelay.get() : crawlDelay.get()); + final long computed = Math.min(maxBackoff, base << (rejections - 1)); + final long half = computed / 2; + final long delay = half + ThreadLocalRandom.current().nextLong(computed - half + 1); + nextFetchTime.set(System.currentTimeMillis() + delay); + return delay; + } + private void setNextFetchTime(long endTime, boolean asap) { if (!asap) { nextFetchTime.set( @@ -386,6 +416,9 @@ static class FetchItemQueues { final long crawlDelay; final long minCrawlDelay; + /** Cap of the backoff applied to a queue whose fetches find every helper thread busy. */ + final long maxBackoff; + int maxQueueSize; final Config conf; @@ -419,6 +452,8 @@ public FetchItemQueues(Config conf) { if (this.maxQueueSize == -1) { this.maxQueueSize = Integer.MAX_VALUE; } + // a queue is never put off for longer than the longest politeness delay accepted + this.maxBackoff = ConfUtils.getInt(conf, "fetcher.max.crawl.delay", 30) * 1000L; // order is not guaranteed for (Entry e : conf.entrySet()) { @@ -465,6 +500,27 @@ public void finishFetchItem(FetchItem it, boolean asap) { return; } fiq.finish(asap); + rescheduleOrReap(fiq); + } + + /** + * Releases the slot of an item whose fetch found every helper thread busy and backs the + * queue off, see {@link FetchItemQueue#finishSaturated}. + * + * @return the delay applied to the queue in milliseconds, -1 if the queue is unknown + */ + public long backOffFetchItem(FetchItem it) { + FetchItemQueue fiq = queues.get(it.queueId); + if (fiq == null) { + LOG.warn("Attempting to back off item from unknown queue: {}", it.queueId); + return -1; + } + long delay = fiq.finishSaturated(maxBackoff); + rescheduleOrReap(fiq); + return delay; + } + + private void rescheduleOrReap(FetchItemQueue fiq) { if (fiq.queue.isEmpty()) { reapIfEmpty(fiq); } else { @@ -705,6 +761,8 @@ public void run() { String robotsCrawlDelaySecs = null; boolean asap = false; + // the fetch never ran because every helper thread was busy + boolean saturated = false; try { URL url = URLUtil.toURL(fit.url); @@ -725,12 +783,15 @@ public void run() { metadata); } catch (FetchTimeoutException e) { // same outcome as with okhttp, where HttpRobotRulesParser turns a - // failed lookup into empty rules: the page is fetched without rules + // failed lookup into empty rules: the page is fetched without rules. + // The protocol caches the failure so that the next URLs of the host + // do not each occupy a helper for a full deadline LOG.info( "[Fetcher #{}] robots.txt lookup timed out for {}", taskId, fit.url); eventCounter.scope("robots.timeout").incrBy(1); + protocol.robotRulesTimedOut(fit.url); rules = RobotRulesParser.EMPTY_RULES; } boolean fromCache = false; @@ -996,14 +1057,17 @@ public void run() { } catch (FetchTimeoutHelpers.SaturatedException e) { // the URL never reached the network: like a URL which waited too long in // the queue, it is acked without a status so that the spout retries it - // later, rather than taking a strike towards max.fetch.errors + // later, rather than taking a strike towards max.fetch.errors. The queue + // is backed off rather than made ready at once (see finally): the helpers + // are shared and retrying immediately would only drain the queue into + // more rejections eventCounter.scope("fetch.helper.rejected").incrBy(1); LOG.warn( "[Fetcher #{}] {}: all {} fetch helpers are busy", taskId, e.getMessage(), fetchHelpers.maxHelpers()); - asap = true; + saturated = true; } catch (Exception exece) { String message = exece.getMessage(); if (message == null) { @@ -1047,7 +1111,16 @@ public void run() { eventCounter.scope("exception").incrBy(1); } finally { - fetchQueues.finishFetchItem(fit, asap); + if (saturated) { + long delay = fetchQueues.backOffFetchItem(fit); + LOG.debug( + "[Fetcher #{}] queue {} backed off for {} ms", + taskId, + fit.queueId, + delay); + } else { + fetchQueues.finishFetchItem(fit, asap); + } activeThreads.decrementAndGet(); // count threads // ack it whatever happens collector.ack(fit.tuple); diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java index c4122f94d..a7eda57b8 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java @@ -296,9 +296,12 @@ public void execute(Tuple input) { metadata); } catch (FetchTimeoutException e) { // same outcome as with okhttp, where HttpRobotRulesParser turns a failed - // lookup into empty rules: the page is fetched without rules + // lookup into empty rules: the page is fetched without rules. The protocol + // caches the failure so that the next URLs of the host do not each occupy a + // helper for a full deadline LOG.info("[Fetcher #{}] robots.txt lookup timed out for {}", taskId, urlString); eventCounter.scope("robots.timeout").incrBy(1); + protocol.robotRulesTimedOut(urlString); rules = RobotRulesParser.EMPTY_RULES; } boolean fromCache = false; diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/AbstractHttpProtocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/AbstractHttpProtocol.java index a2f37b70b..6f9b091f7 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/AbstractHttpProtocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/AbstractHttpProtocol.java @@ -174,6 +174,13 @@ public BaseRobotRules getRobotRules(String url) { return robots.getRobotRulesSet(this, url); } + @Override + public void robotRulesTimedOut(String url) { + if (!this.skipRobots) { + robots.cacheLookupFailure(url); + } + } + @Override public void cleanup() {} diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java index db43d9a2b..9a602efa7 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java @@ -165,6 +165,10 @@ public BaseRobotRules getRobotRules(String url) { return protoInstance.getRobotRules(url); } + public void robotRulesTimedOut(String url) { + protoInstance.robotRulesTimedOut(url); + } + public void cleanup() { protoInstance.cleanup(); } @@ -338,6 +342,14 @@ public boolean supportsFetchTimeout(String url, Metadata metadata) { && forRobots.supportsFetchTimeout(url, metadata); } + @Override + public void robotRulesTimedOut(@NotNull String url) { + FilteredProtocol proto = getProtocolFor(url, robotsMetadata()); + if (proto != null) { + proto.robotRulesTimedOut(url); + } + } + private static Metadata robotsMetadata() { final Metadata m = new Metadata(); m.addValue(ROBOTS, "true"); diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java b/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java index c1419718e..7deb495e9 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java @@ -202,6 +202,27 @@ public BaseRobotRules getRobotRulesSetFromCache(URL url) { return EMPTY_RULES; } + /** + * Caches empty rules for the host of the URL in the error cache, as for any lookup which fails + * with an exception, unless rules for the host are already cached: a lookup abandoned at the + * deadline may have completed on its helper thread in the meantime. + */ + @Override + public void cacheLookupFailure(String url) { + URL u; + try { + u = URLUtil.toURL(url); + } catch (Exception e) { + return; + } + String cacheKey = getCacheKey(u); + if (ERRORCACHE.getIfPresent(cacheKey) != null || CACHE.getIfPresent(cacheKey) != null) { + return; + } + LOG.debug("Caching robots lookup failure for {} under key {}", url, cacheKey); + ERRORCACHE.put(cacheKey, new RobotRules(EMPTY_RULES)); + } + /** * Get the rules from robots.txt which applies for the given {@code url}. Robot rules are cached * for a unique combination of host, protocol, and port. If no rules are found in the cache, a diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java index 3bbdde803..642153c19 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java @@ -69,6 +69,18 @@ default boolean supportsFetchTimeout(String url, Metadata metadata) { BaseRobotRules getRobotRules(String url); + /** + * Called by the fetcher bolts when a {@link #getRobotRules(String)} call for the URL was + * abandoned on a helper thread because {@code fetcher.thread.timeout} passed. A protocol which + * caches robots.txt lookups should record the failure, so that later URLs of the same host do + * not each start a lookup of their own and wait a full deadline: the HTTP protocols cache the + * timeout like any other failed lookup, as empty rules in the robots error cache. Does nothing + * by default. + * + * @param url the URL whose robots.txt lookup timed out + */ + default void robotRulesTimedOut(String url) {} + void cleanup(); public static void main(Protocol protocol, String[] args) throws Exception { diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/RobotRulesParser.java b/core/src/main/java/org/apache/stormcrawler/protocol/RobotRulesParser.java index 121dae615..24220b474 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/RobotRulesParser.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/RobotRulesParser.java @@ -210,4 +210,14 @@ public BaseRobotRules getRobotRulesSet(Protocol protocol, String url) { } public abstract BaseRobotRules getRobotRulesSet(Protocol protocol, URL url); + + /** + * Records that a lookup of the robots.txt for the URL failed outside {@link + * #getRobotRulesSet(Protocol, URL)}, as when the fetcher abandoned it on a helper thread at the + * deadline, so that the rules used for the failure (allow all) are served from the error cache + * to later URLs of the host instead of being looked up again. Does nothing by default. + * + * @param url a URL of the host whose robots.txt could not be obtained + */ + public void cacheLookupFailure(String url) {} } diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java index bf4e144b6..8de292929 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java @@ -751,7 +751,9 @@ public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) if (deadlineNanos != 0) { // hard deadline for the whole chain, enforced by okio's watchdog: on // expiry the call is cancelled, the socket closed and execute() or the - // body read throw immediately. Every hop gets the time that is left + // body read throw immediately. Every hop gets the time that is left. + // DNS can still hold the fetcher thread past this deadline: cancellation + // cannot release the synchronous lookup until the resolver returns final long remaining = deadlineNanos - System.nanoTime(); if (remaining <= 0) { throw new FetchTimeoutException(url, fetchTimeout); diff --git a/core/src/main/resources/crawler-default.yaml b/core/src/main/resources/crawler-default.yaml index 3b6c6f3f6..c86d83109 100644 --- a/core/src/main/resources/crawler-default.yaml +++ b/core/src/main/resources/crawler-default.yaml @@ -34,14 +34,19 @@ config: # chain shares one deadline), independent of the protocol's socket # timeouts; -1 disables it. Never exceeds topology.message.timeout.secs. # With okhttp the HTTP call itself is cancelled when the deadline passes - # (socket closed, URL reported as FETCH_ERROR); other protocols run on a - # helper thread and are abandoned there on timeout. A robots.txt lookup - # which times out lets the page be fetched without rules on both paths + # (socket closed, URL reported as FETCH_ERROR), except that a DNS lookup + # cannot be interrupted and holds the thread until the resolver returns; + # other protocols run on a helper thread and are abandoned there on + # timeout. A robots.txt lookup which times out lets the page be fetched + # without rules on both paths and is cached in the robots error cache like + # any failed lookup fetcher.thread.timeout: -1 # max helper threads per bolt for the abandoned fetches above; default is # 2 x fetcher.threads.number (2 for SimpleFetcherBolt). When all are busy # the URL is acked without a status, like one which waited too long in - # the queue, and the spout retries it later + # the queue, and the spout retries it later; its queue is backed off + # exponentially (its crawl delay, doubled at each consecutive rejection, + # up to fetcher.max.crawl.delay, with equal jitter) instead of being retried at once # fetcher.thread.timeout.helpers: 100 # max. crawl-delay accepted in robots.txt (in seconds) fetcher.max.crawl.delay: 30 diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java index c3a3dc792..545251f48 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java @@ -32,6 +32,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.storm.task.OutputCollector; import org.apache.storm.topology.base.BaseRichBolt; @@ -254,6 +255,7 @@ void abandonedFetchesUseABoundedSharedPool() throws ReflectiveOperationException @Test void hangingRobotsLookupDoesNotFailTheUrl() throws ReflectiveOperationException { StuckProtocol.STARTED.set(0); + StuckProtocol.ROBOTS_TIMED_OUT.clear(); resetProtocolFactory(); TestOutputCollector output = new TestOutputCollector(); Map config = new HashMap<>(); @@ -283,6 +285,47 @@ void hangingRobotsLookupDoesNotFailTheUrl() throws ReflectiveOperationException ((Metadata) statusTuples.get(0).get(1)).getFirstValue("fetch.exception")); } + /** + * A robots.txt lookup abandoned at the deadline is reported to the protocol so that it can + * cache the failure: the next URL of the host must not start a lookup of its own and wait a + * full deadline on another helper. + */ + @Test + void timedOutRobotsLookupIsReportedToTheProtocolAndNotRepeated() + throws ReflectiveOperationException { + StuckProtocol.STARTED.set(0); + StuckProtocol.ROBOTS_HUNG.set(0); + StuckProtocol.ROBOTS_TIMED_OUT.clear(); + resetProtocolFactory(); + TestOutputCollector output = new TestOutputCollector(); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("http.protocol.implementation", StuckProtocol.class.getName()); + config.put(StuckProtocol.HANG_ROBOTS_KEY, true); + config.put("fetcher.threads.number", 1); + config.put("fetcher.thread.timeout", 1L); + // the first URL leaves two helpers stuck (robots, then page): room for the second page + config.put("fetcher.thread.timeout.helpers", 4); + config.put("fetcher.server.delay", 0.0f); + bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + + for (String path : new String[] {"/1", "/2"}) { + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("source"); + when(tuple.getStringByField("url")).thenReturn("http://stuck.example.com" + path); + when(tuple.getValueByField("metadata")).thenReturn(null); + bolt.execute(tuple); + } + + await().atMost(8, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 2); + Assertions.assertEquals( + Set.of("stuck.example.com"), + StuckProtocol.ROBOTS_TIMED_OUT, + "the protocol was told about the timed-out lookup"); + Assertions.assertEquals(1, StuckProtocol.ROBOTS_HUNG.get(), "robots.txt looked up once"); + Assertions.assertEquals(2, StuckProtocol.STARTED.get(), "both pages were fetched"); + } + @Test void invalidProxyMetadataEmitsFetchError(WireMockRuntimeInfo wmRuntimeInfo) throws ReflectiveOperationException { diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java index 6fe7670c2..c69073010 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java @@ -485,4 +485,70 @@ void staleTicketDoesNotHideAnotherReadyQueue() throws Exception { Assertions.assertNotNull(next, "returned nothing while b.net was ready"); Assertions.assertEquals("http://b.net/1", next.url); } + + /** Asserts that the delay is the equal-jittered value: within [computed / 2, computed]. */ + private static void assertJittered(long computed, long delay) { + Assertions.assertTrue( + delay >= computed / 2 && delay <= computed, + "delay " + delay + " not within [" + computed / 2 + ", " + computed + "]"); + } + + /** + * A fetch rejected because every helper thread is busy backs its queue off exponentially from + * the crawl delay up to fetcher.max.crawl.delay, with equal jitter, and a fetch which runs + * resets the backoff. + */ + @Test + void saturatedFetchBacksTheQueueOffExponentially() throws Exception { + FetchItemQueues q = queues("fetcher.server.delay", 2.0f, "fetcher.max.crawl.delay", 5); + // enough items for the queue to stay alive while the slot is retaken below + for (int i = 0; i < 8; i++) { + add(q, "http://a.net/" + i); + } + FetchItem it = q.getFetchItem(); + Assertions.assertNotNull(it); + FetchItemQueue fiq = q.queues.get(it.queueId); + + long[] expected = {2000, 4000, 5000, 5000}; + for (long computed : expected) { + long before = System.currentTimeMillis(); + long delay = q.backOffFetchItem(it); + assertJittered(computed, delay); + long next = fiq.getNextFetchTime(); + Assertions.assertTrue(next >= before + delay, "next fetch time reflects the backoff"); + Assertions.assertTrue(next <= System.currentTimeMillis() + delay); + Assertions.assertEquals(0, fiq.getInProgressSize(), "slot released"); + Assertions.assertNull(q.getFetchItem(), "queue not ready while backed off"); + // take the slot back for the next rejection + fiq.poll(); + Assertions.assertEquals(1, fiq.getInProgressSize()); + } + + // a fetch which runs resets the backoff: the next rejection starts again from the delay + q.finishFetchItem(it, true); + fiq.poll(); + assertJittered(2000, q.backOffFetchItem(it)); + } + + /** The backoff never starts below one second, even for a queue with no delay at all. */ + @Test + void saturatedFetchBackoffHasAOneSecondFloor() throws Exception { + FetchItemQueues q = queues("fetcher.server.delay", 0.0f); + add(q, "http://a.net/1"); + FetchItem it = q.getFetchItem(); + Assertions.assertNotNull(it); + assertJittered(1000, q.backOffFetchItem(it)); + } + + /** The jitter actually spreads the delays: repeated rejections do not all get the same one. */ + @Test + void saturatedFetchBackoffIsJittered() { + Set delays = ConcurrentHashMap.newKeySet(); + for (int i = 0; i < 50; i++) { + FetchItemQueue fiq = new FetchItemQueue("h" + i, 1, 10000, 0, Integer.MAX_VALUE); + fiq.poll(); + delays.add(fiq.finishSaturated(30000)); + } + Assertions.assertTrue(delays.size() > 1, "every delay identical: " + delays); + } } diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java index 6bb5504f4..ddd0e29bd 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java @@ -208,6 +208,53 @@ void poolBoundComesFromConfigOrDefault() { Assertions.assertEquals(1, helpers(1, 0).maxHelpers()); } + /** + * A helper hands its result back before it is back polling for work. A call submitted right + * then, as the page fetch is after the robots.txt lookup, must still find the helper free when + * the pool is at its bound: the bound counts calls in flight, not idle workers. + */ + @Test + void callSubmittedRightAfterAnotherCompletesIsNotRejected() throws Exception { + FetchTimeoutHelpers h = helpers(1, 2); + StuckProtocol stuck = new StuckProtocol(); + // one helper stuck for good: the pool is at its bound as soon as a second call runs + Assertions.assertThrows( + FetchTimeoutException.class, + () -> + h.call( + () -> stuck.getProtocolOutput("http://a.net/", null), + stuck, + "u", + null)); + Assertions.assertEquals(1, h.busy()); + for (int i = 0; i < 500; i++) { + // robots.txt lookup then, at once, the page fetch + Assertions.assertEquals("robots", h.call(() -> "robots", stuck, "http://a.net/", null)); + Assertions.assertEquals("page", h.call(() -> "page", stuck, "http://a.net/", null)); + } + Assertions.assertEquals(2, h.largestPoolSize()); + Assertions.assertEquals(1, h.busy(), "only the stuck call still holds a helper"); + } + + /** A call cancelled before its helper started it gives its permit back. */ + @Test + void timedOutCallWhichNeverStartedReleasesItsPermit() throws Exception { + FetchTimeoutHelpers h = helpers(1, 1); + StuckProtocol stuck = new StuckProtocol(); + Assertions.assertThrows( + FetchTimeoutException.class, + () -> + h.call( + () -> stuck.getProtocolOutput("http://a.net/", null), + stuck, + "u", + null)); + Assertions.assertEquals(1, h.busy()); + Assertions.assertThrows( + FetchTimeoutHelpers.SaturatedException.class, + () -> h.call(() -> "never", stuck, "http://a.net/2", null)); + } + @Test void afterShutdownCallsAreRejected() { FetchTimeoutHelpers h = helpers(1, null); diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java b/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java index 135c3df52..90f2be2d3 100644 --- a/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java +++ b/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java @@ -116,4 +116,28 @@ void testRobotRulesParsing(WireMockRuntimeInfo wmRuntimeInfo) { allowNone(403, modifiedConf, wmRuntimeInfo); allowAll(500, modifiedConf, wmRuntimeInfo); } + + /** + * A lookup failure reported from outside, as when the fetcher abandons the lookup at the + * deadline, is cached in the error cache like any other failure: the next lookup for the host + * is served from the cache, allows all, and sends no request. + */ + @Test + void reportedLookupFailureIsCachedAsEmptyRules(WireMockRuntimeInfo wmRuntimeInfo) { + stubFor( + get(urlPathEqualTo("/robots.txt")) + .willReturn(aResponse().withBody(body).withStatus(200))); + HttpRobotRulesParser parser = new HttpRobotRulesParser(); + parser.setConf(conf); + String base = wmRuntimeInfo.getHttpBaseUrl(); + parser.cacheLookupFailure(base + "/some/page"); + BaseRobotRules rules = parser.getRobotRulesSet(protocol, base + "/other"); + Assertions.assertTrue(rules.isAllowAll(), "the real robots.txt was not fetched"); + Assertions.assertInstanceOf(RobotRules.class, rules); + Assertions.assertEquals(0, ((RobotRules) rules).getContentLengthFetched().length); + Assertions.assertEquals( + 0, + wmRuntimeInfo.getWireMock().getServeEvents().size(), + "no request was sent to the server"); + } } diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java b/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java index 59bf94278..abbb64100 100644 --- a/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java +++ b/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java @@ -20,7 +20,10 @@ import crawlercommons.robots.BaseRobotRules; import crawlercommons.robots.SimpleRobotRules; import crawlercommons.robots.SimpleRobotRules.RobotRulesMode; +import java.net.URL; import java.nio.charset.StandardCharsets; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import org.apache.storm.Config; import org.apache.stormcrawler.Metadata; @@ -34,6 +37,12 @@ public class StuckProtocol implements Protocol { public static final AtomicInteger STARTED = new AtomicInteger(); + /** Number of robots.txt lookups which actually hung. */ + public static final AtomicInteger ROBOTS_HUNG = new AtomicInteger(); + + /** Hosts whose robots.txt lookup the fetcher reported as timed out: served at once after. */ + public static final Set ROBOTS_TIMED_OUT = ConcurrentHashMap.newKeySet(); + public static final long HANG_MILLIS = 10_000; /** Configuration key: when true, robots.txt lookups hang as well. */ @@ -55,12 +64,27 @@ public ProtocolResponse getProtocolOutput(String url, Metadata metadata) throws @Override public BaseRobotRules getRobotRules(String url) { - if (hangRobots) { + if (hangRobots && !ROBOTS_TIMED_OUT.contains(host(url))) { + ROBOTS_HUNG.incrementAndGet(); hang(); } return new SimpleRobotRules(RobotRulesMode.ALLOW_ALL); } + /** Like the HTTP protocols, remembers the failure so the host is not looked up again. */ + @Override + public void robotRulesTimedOut(String url) { + ROBOTS_TIMED_OUT.add(host(url)); + } + + private static String host(String url) { + try { + return new URL(url).getHost(); + } catch (Exception e) { + return url; + } + } + private static void hang() { long until = System.currentTimeMillis() + HANG_MILLIS; while (System.currentTimeMillis() < until) { diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index fbac9158d..1b58b2314 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -192,8 +192,8 @@ is defined. | fetcher.threads.per.queue | 1 | Default number of threads per queue. Can be overridden. | fetcher.threads.start.delay | 10 | Delay (milliseconds) between starting next fetcher thread. Avoids overloading DNS or network resources during fetcher startup when all threads simultaneously start requesting pages. | fetcher.timeout.queue | -1 | Maximum wait time (seconds) for items in the queue. -1 disables timeout. -| fetcher.thread.timeout | -1 | Hard deadline (seconds) for each protocol call made for a URL, independent of the protocol's own socket timeouts: the robots.txt lookup, each redirect it follows (up to 5, each a call of its own), and the page fetch, whose redirect chain shares a single deadline. Worst case for one URL is therefore several times the value. -1 disables it. The deadline never exceeds `topology.message.timeout.secs`. With the default okhttp protocol the deadline is applied to the HTTP call itself: on expiry the call is cancelled, the socket closed and the URL reported as `FETCH_ERROR` with `fetch.exception` "Socket timeout fetching"; if `http.content.partial.as.trimmed` is on, content already received is kept instead and flagged as trimmed for "time". Protocols which do not enforce the deadline themselves (see `Protocol.supportsFetchTimeout(url, metadata)`, resolved per URL by `DelegatorProtocol`) run the call on a helper thread which is abandoned on timeout. On both paths a robots.txt lookup which times out lets the page be fetched without rules, like any failed lookup. The `fetch.timeout` counter counts every timeout, `fetch.deadline` only the ones where this deadline fired, `robots.timeout` the lookups abandoned on a helper thread. -| fetcher.thread.timeout.helpers | 2 x fetcher.threads.number (2 for SimpleFetcherBolt) | Maximum number of helper threads per bolt instance for the abandoned fetches above. Threads are created on demand and released after a minute idle. When every helper is busy the URL never reaches the network: it is acked without a status, like a URL which waited too long in the queue, so that the spout retries it later, and the `fetch.helper.rejected` counter is incremented. Helpers are shared by all hosts, so a host that never answers can make fetches of other hosts be deferred this way until its helpers time out; the `fetchhelpers` gauge shows how many are busy. +| fetcher.thread.timeout | -1 | Hard deadline (seconds) for each protocol call made for a URL, independent of the protocol's own socket timeouts: the robots.txt lookup, each redirect it follows (up to 5, each a call of its own), and the page fetch, whose redirect chain shares a single deadline. Worst case for one URL is therefore several times the value. -1 disables it. The deadline never exceeds `topology.message.timeout.secs`. With the default okhttp protocol the deadline is applied to the HTTP call itself: on expiry the call is cancelled, the socket closed and the URL reported as `FETCH_ERROR` with `fetch.exception` "Socket timeout fetching"; if `http.content.partial.as.trimmed` is on, content already received is kept instead and flagged as trimmed for "time". One step escapes the cancellation: a DNS lookup is synchronous and cannot be interrupted, so a slow resolver holds the fetcher thread past the deadline until it returns, at which point the timeout is reported. Protocols which do not enforce the deadline themselves (see `Protocol.supportsFetchTimeout(url, metadata)`, resolved per URL by `DelegatorProtocol`) run the call on a helper thread which is abandoned on timeout. On both paths a robots.txt lookup which times out lets the page be fetched without rules and is cached in the robots error cache like any failed lookup (see `Protocol.robotRulesTimedOut(url)`), so later URLs of the host do not each wait for another deadline. The `fetch.timeout` counter counts every timeout, `fetch.deadline` only the ones where this deadline fired, `robots.timeout` the lookups abandoned on a helper thread. +| fetcher.thread.timeout.helpers | 2 x fetcher.threads.number (2 for SimpleFetcherBolt) | Maximum number of helper threads per bolt instance for the abandoned fetches above. Threads are created on demand and released after a minute idle. When every helper is busy the URL never reaches the network: it is acked without a status, like a URL which waited too long in the queue, so that the spout retries it later, and the `fetch.helper.rejected` counter is incremented. In `FetcherBolt` the queue of the URL is backed off exponentially rather than made ready again at once: its own crawl delay (at least 1 s), doubled at each consecutive rejection, capped by `fetcher.max.crawl.delay`, spread with equal jitter (between half the value and the value), and reset by any fetch of the queue that runs. Helpers are shared by all hosts, so a host that never answers can make fetches of other hosts be deferred this way until its helpers time out; the `fetchhelpers` gauge shows how many are busy. | fetcherbolt.queue.debug.filepath | "" | Path to a debug log (e.g. /tmp/fetcher-dump-{port}). | http.agent.description | - | Description for the User-Agent header. | http.agent.email | - | Email address in User-Agent header. From 9b23ff78499d2e249200f8507d1ff99a9b654009 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Thu, 17 Sep 2026 21:13:25 +0200 Subject: [PATCH 3/7] Update deprecation comment for FETCH_TIMEOUT_PARAM_KEY --- .../main/java/org/apache/stormcrawler/bolt/FetcherBolt.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 8b140b81b..bdf7b5053 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -85,9 +85,9 @@ public class FetcherBolt extends StatusEmitterBolt { public static final String QUEUED_TIMEOUT_PARAM_KEY = "fetcher.timeout.queue"; /** - * @deprecated since 3.8, use {@link Constants#FETCH_TIMEOUT_PARAM_KEY} + * @deprecated since 4.0, use {@link Constants#FETCH_TIMEOUT_PARAM_KEY} */ - @Deprecated(since = "3.8", forRemoval = true) + @Deprecated(since = "4.0", forRemoval = true) public static final String FETCH_TIMEOUT_PARAM_KEY = Constants.FETCH_TIMEOUT_PARAM_KEY; /** Key name of the custom crawl delay for a queue that may be present in the metadata. */ From b16ce2390324de62f9f88e5c26b5731e56103c90 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Thu, 17 Sep 2026 21:32:53 +0200 Subject: [PATCH 4/7] Configure HTTPRobot Cache as read first. Handle sequence: fetcher signals EMPTY handler ends after --- .../protocol/HttpRobotRulesParser.java | 9 +++++ .../protocol/HttpRobotRulesParserTest.java | 34 +++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java b/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java index 7deb495e9..4a5d42ba7 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java @@ -383,6 +383,12 @@ public BaseRobotRules getRobotRulesSet(Protocol http, URL url) { LOG.debug("Caching robots for {} under key {} in cache {}", url, cacheKey, cacheName); cacheToUse.put(cacheKey, cached); + if (cacheRule) { + // a lookup abandoned at the deadline and completed here on its helper thread may have + // been recorded as a failure in the meantime: the error cache is read first, so its + // entry would hide the rules for the whole error TTL + ERRORCACHE.invalidate(cacheKey); + } // cache robot rules for redirections // get here only if the target has not been found in the cache @@ -398,6 +404,9 @@ public BaseRobotRules getRobotRulesSet(Protocol http, URL url) { keyredir, cacheName); cacheToUse.put(keyredir, cached); + if (cacheRule) { + ERRORCACHE.invalidate(keyredir); + } } } diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java b/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java index 90f2be2d3..581d0ec67 100644 --- a/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java +++ b/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java @@ -25,6 +25,7 @@ import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo; import com.github.tomakehurst.wiremock.junit5.WireMockTest; import crawlercommons.robots.BaseRobotRules; +import java.util.concurrent.CompletableFuture; import org.apache.storm.Config; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; @@ -140,4 +141,37 @@ void reportedLookupFailureIsCachedAsEmptyRules(WireMockRuntimeInfo wmRuntimeInfo wmRuntimeInfo.getWireMock().getServeEvents().size(), "no request was sent to the server"); } + + /** + * A lookup abandoned at the deadline keeps running on its helper thread and completes after the + * failure was reported: the rules it obtained must win over the failure entry, as the + * robots.txt of the host was actually fetched. + */ + @Test + void rulesObtainedAfterReportedFailureReplaceIt(WireMockRuntimeInfo wmRuntimeInfo) { + stubFor( + get(urlPathEqualTo("/robots.txt")) + .willReturn( + aResponse().withBody(body).withStatus(200).withFixedDelay(1000))); + HttpRobotRulesParser parser = new HttpRobotRulesParser(); + parser.setConf(conf); + String base = wmRuntimeInfo.getHttpBaseUrl(); + // the lookup the fetcher abandons: it goes on while the failure is reported + CompletableFuture abandoned = + CompletableFuture.supplyAsync( + () -> parser.getRobotRulesSet(protocol, base + "/some/page")); + while (wmRuntimeInfo.getWireMock().getServeEvents().isEmpty() && !abandoned.isDone()) { + Thread.onSpinWait(); + } + parser.cacheLookupFailure(base + "/some/page"); + BaseRobotRules obtained = abandoned.join(); + Assertions.assertFalse(obtained.isAllowed(base + "/restricted/page")); + BaseRobotRules rules = parser.getRobotRulesSet(protocol, base + "/other"); + Assertions.assertFalse(rules.isAllowAll(), "the failure entry hides the fetched rules"); + Assertions.assertFalse(rules.isAllowed(base + "/restricted/page")); + Assertions.assertEquals( + 1, + wmRuntimeInfo.getWireMock().getServeEvents().size(), + "the rules were served from the cache"); + } } From 92a127d74e13cc2934e669c6b125c1cf8d5abf2e Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Thu, 17 Sep 2026 21:40:59 +0200 Subject: [PATCH 5/7] Handle nit on ROBOTS_HUNG, ROBOTS_TIMED_OUT (STARTED as well) on StuckProtocol --- .../bolt/AbstractFetcherBoltTest.java | 25 +++++++++------- .../stormcrawler/protocol/StuckProtocol.java | 30 ++++++++++++++----- 2 files changed, 36 insertions(+), 19 deletions(-) diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java index 545251f48..8614b5824 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java @@ -34,6 +34,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; +import org.apache.storm.Config; import org.apache.storm.task.OutputCollector; import org.apache.storm.topology.base.BaseRichBolt; import org.apache.storm.tuple.Tuple; @@ -206,7 +207,6 @@ void stuckFetchDoesNotBlockTheFollowingFetches(WireMockRuntimeInfo wmRuntimeInfo */ @Test void abandonedFetchesUseABoundedSharedPool() throws ReflectiveOperationException { - StuckProtocol.STARTED.set(0); resetProtocolFactory(); TestOutputCollector output = new TestOutputCollector(); Map config = new HashMap<>(); @@ -230,7 +230,7 @@ void abandonedFetchesUseABoundedSharedPool() throws ReflectiveOperationException // pool of 2 (twice the fetcher threads): the first two fetches really started and // timed out - Assertions.assertEquals(2, StuckProtocol.STARTED.get(), "fetches actually started"); + Assertions.assertEquals(2, stuckProtocol(config).started(), "fetches actually started"); List> statusTuples = output.getEmitted(Constants.StatusStreamName); Assertions.assertEquals(2, statusTuples.size()); for (List t : statusTuples) { @@ -254,8 +254,6 @@ void abandonedFetchesUseABoundedSharedPool() throws ReflectiveOperationException */ @Test void hangingRobotsLookupDoesNotFailTheUrl() throws ReflectiveOperationException { - StuckProtocol.STARTED.set(0); - StuckProtocol.ROBOTS_TIMED_OUT.clear(); resetProtocolFactory(); TestOutputCollector output = new TestOutputCollector(); Map config = new HashMap<>(); @@ -276,7 +274,7 @@ void hangingRobotsLookupDoesNotFailTheUrl() throws ReflectiveOperationException await().atMost(6, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 1); long elapsed = System.currentTimeMillis() - start; Assertions.assertTrue(elapsed >= 2_000 && elapsed < 5_000, "took " + elapsed + " ms"); - Assertions.assertEquals(1, StuckProtocol.STARTED.get(), "the page fetch was attempted"); + Assertions.assertEquals(1, stuckProtocol(config).started(), "the page fetch was attempted"); List> statusTuples = output.getEmitted(Constants.StatusStreamName); Assertions.assertEquals(1, statusTuples.size()); Assertions.assertEquals(Status.FETCH_ERROR, statusTuples.get(0).get(2)); @@ -293,9 +291,6 @@ void hangingRobotsLookupDoesNotFailTheUrl() throws ReflectiveOperationException @Test void timedOutRobotsLookupIsReportedToTheProtocolAndNotRepeated() throws ReflectiveOperationException { - StuckProtocol.STARTED.set(0); - StuckProtocol.ROBOTS_HUNG.set(0); - StuckProtocol.ROBOTS_TIMED_OUT.clear(); resetProtocolFactory(); TestOutputCollector output = new TestOutputCollector(); Map config = new HashMap<>(); @@ -318,12 +313,13 @@ void timedOutRobotsLookupIsReportedToTheProtocolAndNotRepeated() } await().atMost(8, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 2); + StuckProtocol stuck = stuckProtocol(config); Assertions.assertEquals( Set.of("stuck.example.com"), - StuckProtocol.ROBOTS_TIMED_OUT, + stuck.robotsTimedOut(), "the protocol was told about the timed-out lookup"); - Assertions.assertEquals(1, StuckProtocol.ROBOTS_HUNG.get(), "robots.txt looked up once"); - Assertions.assertEquals(2, StuckProtocol.STARTED.get(), "both pages were fetched"); + Assertions.assertEquals(1, stuck.robotsHung(), "robots.txt looked up once"); + Assertions.assertEquals(2, stuck.started(), "both pages were fetched"); } @Test @@ -417,6 +413,13 @@ List fetchAndGetStatusTuple( return statusTuples.get(0); } + /** The protocol instance the prepared bolt fetches with. */ + private static StuckProtocol stuckProtocol(Map config) { + Config conf = new Config(); + conf.putAll(config); + return (StuckProtocol) ProtocolFactory.getInstance(conf).getProtocol("http")[0]; + } + static void resetProtocolFactory() throws ReflectiveOperationException { Field instance = ProtocolFactory.class.getDeclaredField("single_instance"); instance.setAccessible(true); diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java b/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java index abbb64100..b83f7c029 100644 --- a/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java +++ b/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java @@ -31,17 +31,19 @@ /** * A protocol whose fetches hang for a long time and ignore interruption, like a client blocked on a - * socket. Counts the fetches that actually started. + * socket. Counts the fetches that actually started. All state is per instance: a test gets the + * instance a bolt uses from the {@link ProtocolFactory} once the bolt is prepared. */ public class StuckProtocol implements Protocol { - public static final AtomicInteger STARTED = new AtomicInteger(); + /** Number of fetches which actually started. */ + private final AtomicInteger started = new AtomicInteger(); /** Number of robots.txt lookups which actually hung. */ - public static final AtomicInteger ROBOTS_HUNG = new AtomicInteger(); + private final AtomicInteger robotsHung = new AtomicInteger(); /** Hosts whose robots.txt lookup the fetcher reported as timed out: served at once after. */ - public static final Set ROBOTS_TIMED_OUT = ConcurrentHashMap.newKeySet(); + private final Set robotsTimedOut = ConcurrentHashMap.newKeySet(); public static final long HANG_MILLIS = 10_000; @@ -57,15 +59,15 @@ public void configure(Config conf) { @Override public ProtocolResponse getProtocolOutput(String url, Metadata metadata) throws Exception { - STARTED.incrementAndGet(); + started.incrementAndGet(); hang(); return new ProtocolResponse("late".getBytes(StandardCharsets.UTF_8), 200, new Metadata()); } @Override public BaseRobotRules getRobotRules(String url) { - if (hangRobots && !ROBOTS_TIMED_OUT.contains(host(url))) { - ROBOTS_HUNG.incrementAndGet(); + if (hangRobots && !robotsTimedOut.contains(host(url))) { + robotsHung.incrementAndGet(); hang(); } return new SimpleRobotRules(RobotRulesMode.ALLOW_ALL); @@ -74,7 +76,19 @@ public BaseRobotRules getRobotRules(String url) { /** Like the HTTP protocols, remembers the failure so the host is not looked up again. */ @Override public void robotRulesTimedOut(String url) { - ROBOTS_TIMED_OUT.add(host(url)); + robotsTimedOut.add(host(url)); + } + + public int started() { + return started.get(); + } + + public int robotsHung() { + return robotsHung.get(); + } + + public Set robotsTimedOut() { + return Set.copyOf(robotsTimedOut); } private static String host(String url) { From 7a2dbfaf1d4cb187a6e6f52fe373e8d5175767ab Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Thu, 17 Sep 2026 21:45:32 +0200 Subject: [PATCH 6/7] Handle race: FutureTask can be canceled in the cs between run() and started.set(true) in FetchTimeoutHelpers --- .../bolt/FetchTimeoutHelpers.java | 45 ++++++++++++++----- .../bolt/FetchTimeoutHelpersTest.java | 45 +++++++++++++++++++ 2 files changed, 80 insertions(+), 10 deletions(-) diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java index 12c64bab7..27dc6df87 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java @@ -28,7 +28,6 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.apache.storm.task.TopologyContext; import org.apache.stormcrawler.Constants; @@ -54,8 +53,9 @@ *

The bound is a {@link Semaphore} of in-flight calls in front of a queueing executor, not the * executor's own rejection: a worker hands its result back before it is back polling for work, so a * pool with a {@code SynchronousQueue} would reject the page fetch submitted right after the - * robots.txt lookup whenever it is at its maximum. A permit is released when the call completes on - * the helper, whether the caller is still waiting for it or has given up. + * robots.txt lookup whenever it is at its maximum. A permit is released exactly once, by whoever + * wins the {@link Handoff}: the helper once the call has run, whether the caller is still waiting + * for it or has given up, or the caller if it gave up before the helper took the call. * *

The deadline is the one of {@link FetchTimeout#secs(Map)}, clamped to the message timeout on * both paths. @@ -69,6 +69,31 @@ static final class SaturatedException extends Exception { } } + /** + * Decides who releases the permit of a call: the helper which runs it, or the caller which + * abandons it before it ran. {@link Future#cancel} alone cannot tell: it succeeds until the + * call returns, also once {@link java.util.concurrent.FutureTask#run} is past its own check and + * about to invoke the call, so a flag set by the call would still be unset while the call is on + * its way. A single state moved by a compare-and-set settles it: exactly one side wins. + */ + static final class Handoff { + private static final int PENDING = 0; + private static final int RUNNING = 1; + private static final int ABANDONED = 2; + + private final AtomicInteger state = new AtomicInteger(PENDING); + + /** Called by the helper before the call: false if the caller abandoned it, so don't run. */ + boolean startedByHelper() { + return state.compareAndSet(PENDING, RUNNING); + } + + /** Called by the caller at the deadline: true if the helper never took the call. */ + boolean abandonedByCaller() { + return state.compareAndSet(PENDING, ABANDONED); + } + } + private final long timeoutSecs; private final ThreadPoolExecutor helpers; @@ -155,15 +180,17 @@ T call(Callable call, Protocol protocol, String url, Metadata metadata) if (!permits.tryAcquire()) { throw new SaturatedException(url); } - // the permit is given back by whoever knows the call is over: the helper once the call - // has run, or the caller if it cancelled the call before the helper ever started it - final AtomicBoolean started = new AtomicBoolean(false); + final Handoff handoff = new Handoff(); final Future future; try { future = helpers.submit( () -> { - started.set(true); + if (!handoff.startedByHelper()) { + // abandoned at the deadline before it ran: the caller has + // released the permit and is not waiting for a result + return null; + } try { return call.call(); } finally { @@ -180,9 +207,7 @@ T call(Callable call, Protocol protocol, String url, Metadata metadata) } catch (TimeoutException e) { // a courtesy for protocols which do honour interruption future.cancel(true); - // once cancelled the task can no longer start: if it had not, nobody else will - // release the permit - if (!started.get()) { + if (handoff.abandonedByCaller()) { permits.release(); } throw new FetchTimeoutException(url, timeoutSecs); diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java index ddd0e29bd..55a5317b8 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java @@ -255,6 +255,51 @@ void timedOutCallWhichNeverStartedReleasesItsPermit() throws Exception { () -> h.call(() -> "never", stuck, "http://a.net/2", null)); } + /** + * The permit of a call is released by exactly one side. Whatever the interleaving of the helper + * taking the call and the caller giving up on it, only the first to move wins the handoff, so a + * call abandoned while its task was already past the cancellation check cannot be released + * twice. + */ + @Test + void permitHandoffHasExactlyOneWinner() { + FetchTimeoutHelpers.Handoff helperFirst = new FetchTimeoutHelpers.Handoff(); + Assertions.assertTrue(helperFirst.startedByHelper(), "the helper took the call"); + Assertions.assertFalse(helperFirst.abandonedByCaller(), "the helper releases"); + Assertions.assertFalse(helperFirst.startedByHelper(), "moved once only"); + + FetchTimeoutHelpers.Handoff callerFirst = new FetchTimeoutHelpers.Handoff(); + Assertions.assertTrue(callerFirst.abandonedByCaller(), "the caller gave up first"); + Assertions.assertFalse(callerFirst.startedByHelper(), "the call must not run"); + Assertions.assertFalse(callerFirst.abandonedByCaller(), "moved once only"); + } + + /** + * Every call, timed out or not, leaves the pool as it found it: the number of busy helpers + * never goes negative and returns to the calls still stuck once the rest have completed. + */ + @Test + void permitsAreNeverReleasedTwice() throws Exception { + FetchTimeoutHelpers h = helpers(1, 2); + StuckProtocol stuck = new StuckProtocol(); + for (int i = 0; i < 20; i++) { + Assertions.assertEquals("x", h.call(() -> "x", PLAIN, "http://a.net/" + i, null)); + Assertions.assertTrue(h.busy() >= 0, "busy went negative"); + } + Assertions.assertEquals(0, h.busy()); + Assertions.assertThrows( + FetchTimeoutException.class, + () -> + h.call( + () -> stuck.getProtocolOutput("http://a.net/", null), + stuck, + "u", + null)); + Assertions.assertEquals(1, h.busy(), "the abandoned call still holds its helper"); + Assertions.assertEquals("y", h.call(() -> "y", PLAIN, "http://a.net/y", null)); + Assertions.assertEquals(1, h.busy(), "a completed call gave its permit back once"); + } + @Test void afterShutdownCallsAreRejected() { FetchTimeoutHelpers h = helpers(1, null); From a575b472989a2ebd3948132cfd56ac3c100e8b22 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Thu, 17 Sep 2026 22:10:26 +0200 Subject: [PATCH 7/7] Refactors the ERRORCACHE invalidation mechanism to properly handle the fetcher vs. helper sequence and expands test coverage --- .../protocol/HttpRobotRulesParser.java | 9 ++++++- .../bolt/FetchTimeoutHelpersTest.java | 26 ------------------- .../protocol/HttpRobotRulesParserTest.java | 8 +++--- 3 files changed, 13 insertions(+), 30 deletions(-) diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java b/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java index 4a5d42ba7..c20e42d35 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java @@ -205,7 +205,8 @@ public BaseRobotRules getRobotRulesSetFromCache(URL url) { /** * Caches empty rules for the host of the URL in the error cache, as for any lookup which fails * with an exception, unless rules for the host are already cached: a lookup abandoned at the - * deadline may have completed on its helper thread in the meantime. + * deadline may have completed on its helper thread in the meantime, before or after this call, + * and its rules must win either way. */ @Override public void cacheLookupFailure(String url) { @@ -221,6 +222,11 @@ public void cacheLookupFailure(String url) { } LOG.debug("Caching robots lookup failure for {} under key {}", url, cacheKey); ERRORCACHE.put(cacheKey, new RobotRules(EMPTY_RULES)); + // the lookup may have completed between the check above and the put: it invalidated an + // entry which was not there yet, so check again now that the entry is visible to it + if (CACHE.getIfPresent(cacheKey) != null) { + ERRORCACHE.invalidate(cacheKey); + } } /** @@ -314,6 +320,7 @@ public BaseRobotRules getRobotRulesSet(Protocol http, URL url) { keyredir, cacheKey); CACHE.put(cacheKey, cachedRediRobotRules); + ERRORCACHE.invalidate(cacheKey); return cachedRediRobotRules; } else { // Remember the target host/authority, we can cache the rules, too. diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java index 55a5317b8..5fd5150c4 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java @@ -274,32 +274,6 @@ void permitHandoffHasExactlyOneWinner() { Assertions.assertFalse(callerFirst.abandonedByCaller(), "moved once only"); } - /** - * Every call, timed out or not, leaves the pool as it found it: the number of busy helpers - * never goes negative and returns to the calls still stuck once the rest have completed. - */ - @Test - void permitsAreNeverReleasedTwice() throws Exception { - FetchTimeoutHelpers h = helpers(1, 2); - StuckProtocol stuck = new StuckProtocol(); - for (int i = 0; i < 20; i++) { - Assertions.assertEquals("x", h.call(() -> "x", PLAIN, "http://a.net/" + i, null)); - Assertions.assertTrue(h.busy() >= 0, "busy went negative"); - } - Assertions.assertEquals(0, h.busy()); - Assertions.assertThrows( - FetchTimeoutException.class, - () -> - h.call( - () -> stuck.getProtocolOutput("http://a.net/", null), - stuck, - "u", - null)); - Assertions.assertEquals(1, h.busy(), "the abandoned call still holds its helper"); - Assertions.assertEquals("y", h.call(() -> "y", PLAIN, "http://a.net/y", null)); - Assertions.assertEquals(1, h.busy(), "a completed call gave its permit back once"); - } - @Test void afterShutdownCallsAreRejected() { FetchTimeoutHelpers h = helpers(1, null); diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java b/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java index 581d0ec67..e900ebc70 100644 --- a/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java +++ b/core/src/test/java/org/apache/stormcrawler/protocol/HttpRobotRulesParserTest.java @@ -21,11 +21,13 @@ import static com.github.tomakehurst.wiremock.client.WireMock.get; import static com.github.tomakehurst.wiremock.client.WireMock.stubFor; import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo; +import static org.awaitility.Awaitility.await; import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo; import com.github.tomakehurst.wiremock.junit5.WireMockTest; import crawlercommons.robots.BaseRobotRules; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.storm.Config; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; @@ -160,9 +162,9 @@ void rulesObtainedAfterReportedFailureReplaceIt(WireMockRuntimeInfo wmRuntimeInf CompletableFuture abandoned = CompletableFuture.supplyAsync( () -> parser.getRobotRulesSet(protocol, base + "/some/page")); - while (wmRuntimeInfo.getWireMock().getServeEvents().isEmpty() && !abandoned.isDone()) { - Thread.onSpinWait(); - } + await().atMost(5, TimeUnit.SECONDS) + .until(() -> !wmRuntimeInfo.getWireMock().getServeEvents().isEmpty()); + Assertions.assertFalse(abandoned.isDone(), "the lookup is still in flight"); parser.cacheLookupFailure(base + "/some/page"); BaseRobotRules obtained = abandoned.join(); Assertions.assertFalse(obtained.isAllowed(base + "/restricted/page"));