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..27dc6df87
--- /dev/null
+++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java
@@ -0,0 +1,229 @@
+/*
+ * 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.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+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.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 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 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.
+ */
+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);
+ }
+ }
+
+ /**
+ * 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;
+
+ /** 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
+ * @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();
+ 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(
+ bound,
+ bound,
+ 60L,
+ TimeUnit.SECONDS,
+ 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 helpers busy with a call. */
+ void registerMetrics(TopologyContext context, Map conf, int 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. */
+ 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();
+ }
+ if (!permits.tryAcquire()) {
+ throw new SaturatedException(url);
+ }
+ final Handoff handoff = new Handoff();
+ final Future future;
+ try {
+ future =
+ helpers.submit(
+ () -> {
+ 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 {
+ permits.release();
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ // shut down
+ permits.release();
+ 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);
+ if (handoff.abandonedByCaller()) {
+ permits.release();
+ }
+ 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..bdf7b5053 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,12 @@
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.ThreadLocalRandom;
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 +58,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 +85,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 4.0, use {@link Constants#FETCH_TIMEOUT_PARAM_KEY}
*/
- public static final String FETCH_TIMEOUT_PARAM_KEY = "fetcher.thread.timeout";
+ @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. */
private static final String CRAWL_DELAY_KEY_NAME = "crawl.delay";
@@ -130,6 +125,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();
@@ -238,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;
@@ -334,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(
@@ -384,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;
@@ -417,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()) {
@@ -463,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 {
@@ -635,15 +693,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 +706,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
@@ -725,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);
@@ -735,7 +773,27 @@ 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.
+ // 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;
if (rules instanceof RobotRules
&& ((RobotRules) rules).getContentLengthFetched().length == 0) {
@@ -871,33 +929,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 +1054,20 @@ 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. 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());
+ saturated = true;
} catch (Exception exece) {
String message = exece.getMessage();
if (message == null) {
@@ -1024,10 +1075,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);
@@ -1055,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);
@@ -1137,6 +1202,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 +1253,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..a7eda57b8 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,24 @@ 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. 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;
if (rules instanceof RobotRules
&& ((RobotRules) rules).getContentLengthFetched().length == 0) {
@@ -446,28 +448,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 +567,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 +584,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/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 b19760b93..9a602efa7 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,10 +157,18 @@ 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);
}
+ public void robotRulesTimedOut(String url) {
+ protoInstance.robotRulesTimedOut(url);
+ }
+
public void cleanup() {
protoInstance.cleanup();
}
@@ -297,9 +305,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 +326,36 @@ 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);
+ }
+
+ @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");
+ 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/HttpRobotRulesParser.java b/core/src/main/java/org/apache/stormcrawler/protocol/HttpRobotRulesParser.java
index c1419718e..c20e42d35 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,33 @@ 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, before or after this call,
+ * and its rules must win either way.
+ */
+ @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));
+ // 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);
+ }
+ }
+
/**
* 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
@@ -293,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.
@@ -362,6 +390,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
@@ -377,6 +411,9 @@ public BaseRobotRules getRobotRulesSet(Protocol http, URL url) {
keyredir,
cacheName);
cacheToUse.put(keyredir, cached);
+ if (cacheRule) {
+ ERRORCACHE.invalidate(keyredir);
+ }
}
}
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..642153c19 100644
--- a/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java
+++ b/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java
@@ -53,8 +53,34 @@ 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);
+ /**
+ * 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 cda61b47d..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
@@ -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,18 @@ 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.
+ // 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);
+ }
+ call.timeout().timeout(remaining, TimeUnit.NANOSECONDS);
+ }
try {
lastResponse = call.execute();
} catch (IOException | RuntimeException e) {
@@ -889,6 +918,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..c86d83109 100644
--- a/core/src/main/resources/crawler-default.yaml
+++ b/core/src/main/resources/crawler-default.yaml
@@ -29,9 +29,25 @@ 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), 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; 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
# 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..8614b5824 100644
--- a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java
@@ -32,7 +32,9 @@
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.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
@@ -43,6 +45,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 +56,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 +158,170 @@ 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 {
+ 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(config).started(), "fetches actually started");
+ List> statusTuples = output.getEmitted(Constants.StatusStreamName);
+ Assertions.assertEquals(2, statusTuples.size());
+ for (List