You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stormcrawler.apache.org by dp...@apache.org on 2026/07/01 20:42:58 UTC
(stormcrawler) 01/01: feat: emit Retry-After host back-off and block the URLFrontier queue (#867, #784)
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/867-host-info-backoff
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 62b87124c84a3538c0a3a4d1040613bc3332960d
Author: Davide Polato <dp...@apache.org>
AuthorDate: Wed Jul 1 22:20:01 2026 +0200
feat: emit Retry-After host back-off and block the URLFrontier queue (#867, #784)
On an HTTP 429 or 503 response, FetcherBolt parses the Retry-After
header (delta-seconds or IMF-fixdate), caps it via the new
fetcher.max.retry.after setting (default 24h) and emits
(key, blockUntil) on the new "hostinfo" stream. The key is derived
with URLPartitioner from partition.url.mode so it matches the
URLFrontier queue key, which may differ from the fetcher's internal
queue mode.
The new HostBlockBolt (external/urlfrontier) consumes the stream and
calls blockQueueUntil, so the frontier stops handing out URLs for the
host until the requested time. The fetch pipeline itself never delays
or holds tuples.
---
.../java/org/apache/stormcrawler/Constants.java | 2 +
.../org/apache/stormcrawler/bolt/FetcherBolt.java | 131 +++++++++++++++++
core/src/main/resources/crawler-default.yaml | 9 ++
.../bolt/FetcherBoltRetryAfterParserTest.java | 141 ++++++++++++++++++
.../apache/stormcrawler/bolt/FetcherBoltTest.java | 126 ++++++++++++++++
.../stormcrawler/urlfrontier/HostBlockBolt.java | 117 +++++++++++++++
.../urlfrontier/HostBlockBoltTest.java | 162 +++++++++++++++++++++
7 files changed, 688 insertions(+)
diff --git a/core/src/main/java/org/apache/stormcrawler/Constants.java b/core/src/main/java/org/apache/stormcrawler/Constants.java
index 7e5ef863..411b7bb0 100644
--- a/core/src/main/java/org/apache/stormcrawler/Constants.java
+++ b/core/src/main/java/org/apache/stormcrawler/Constants.java
@@ -31,6 +31,8 @@ public class Constants {
public static final String StatusStreamName = "status";
+ public static final String HOST_INFO_STREAM_NAME = "hostinfo";
+
public static final String DELETION_STREAM_NAME = "deletion";
public static final String AllowRedirParamName = "redirections.allowed";
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 2439bd76..bde8eab4 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java
+++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java
@@ -24,7 +24,10 @@ import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.UnknownHostException;
+import java.time.DateTimeException;
import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -67,6 +70,7 @@ import org.apache.stormcrawler.protocol.ProtocolFactory;
import org.apache.stormcrawler.protocol.ProtocolResponse;
import org.apache.stormcrawler.protocol.RobotRules;
import org.apache.stormcrawler.util.ConfUtils;
+import org.apache.stormcrawler.util.URLPartitioner;
import org.apache.stormcrawler.util.URLUtil;
import org.slf4j.LoggerFactory;
@@ -94,6 +98,38 @@ public class FetcherBolt extends StatusEmitterBolt {
*/
public static final String FETCH_TIMEOUT_PARAM_KEY = "fetcher.thread.timeout";
+ /**
+ * Maximum delay in seconds honoured when a server requests a back-off via the <a
+ * href="https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After">Retry-After</a>
+ * HTTP response header. The parsed delay (capped to this value) is emitted on the {@link
+ * Constants#HOST_INFO_STREAM_NAME} stream as {@code blockUntil} for a host-aware consumer to
+ * act on; the fetcher itself does not delay the fetch. A value of {@code -1} disables the cap
+ * and emits the delay as-is; the default caps at 86400 (24h).
+ */
+ public static final String MAX_RETRY_AFTER_PARAM_KEY = "fetcher.max.retry.after";
+
+ /** Name of the Retry-After HTTP header, lower-cased as stored by the protocol layer. */
+ private static final String RETRY_AFTER_KEY = "retry-after";
+
+ /**
+ * Formatter for the HTTP-date form of the Retry-After header, e.g. {@code Wed, 21 Oct 2015
+ * 07:28:00 GMT}. {@link DateTimeFormatter} is immutable and thread-safe, unlike {@code
+ * SimpleDateFormat}, which matters here as it is shared across fetcher threads.
+ */
+ private static final DateTimeFormatter HTTP_DATE_FORMATTER =
+ DateTimeFormatter.ofPattern("EEE, dd MMM yyyy HH:mm:ss 'GMT'", Locale.ROOT)
+ .withZone(ZoneOffset.UTC);
+
+ /** Pre-compiled matcher for the numeric (delta-seconds) form of the Retry-After header. */
+ private static final Pattern RETRY_AFTER_SECONDS = Pattern.compile("[0-9]+");
+
+ /**
+ * Upper bound on the length of a Retry-After header value we will attempt to parse. A
+ * delta-seconds value is at most 19 digits and an HTTP date ~29 chars; anything longer is
+ * rejected to bound the work done on this attacker-controlled value.
+ */
+ private static final int MAX_RETRY_AFTER_VALUE_LENGTH = 64;
+
/** 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";
@@ -116,6 +152,14 @@ public class FetcherBolt extends StatusEmitterBolt {
private ProtocolFactory protocolFactory;
+ /**
+ * Derives the host-info back-off key the same way the frontier keys its queues (via {@code
+ * partition.url.mode}), so a {@code blockUntil} signal targets the right queue. Configured once
+ * and shared across fetcher threads; {@link URLPartitioner#getPartition} holds no mutable
+ * state.
+ */
+ private URLPartitioner partitioner;
+
private int taskId = -1;
boolean sitemapsAutoDiscovery = false;
@@ -483,6 +527,49 @@ public class FetcherBolt extends StatusEmitterBolt {
}
}
+ /**
+ * Parses the value of a <a
+ * href="https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After">Retry-After</a>
+ * HTTP response header into a delay expressed in milliseconds, relative to now. The value is
+ * either a number of seconds (e.g. {@code 120}) or an HTTP date (e.g. {@code Wed, 21 Oct 2015
+ * 07:28:00 GMT}).
+ *
+ * @return the delay in milliseconds (possibly {@code 0}), or {@code -1} if the header is
+ * absent, malformed, in the past or implausibly long.
+ */
+ static long parseRetryAfterDelay(String retryAfter) {
+ if (StringUtils.isBlank(retryAfter)) {
+ return -1;
+ }
+ retryAfter = retryAfter.trim();
+ // a valid Retry-After is either a delta-seconds value or an HTTP date;
+ // reject anything implausibly long to bound the work done on this
+ // attacker-controlled header value
+ if (retryAfter.length() > MAX_RETRY_AFTER_VALUE_LENGTH) {
+ return -1;
+ }
+ // delay expressed as a number of seconds
+ if (RETRY_AFTER_SECONDS.matcher(retryAfter).matches()) {
+ try {
+ return Math.multiplyExact(Long.parseLong(retryAfter), 1000L);
+ } catch (NumberFormatException | ArithmeticException e) {
+ // value too large to fit in a long, or its conversion to ms
+ // overflows - ignore
+ return -1;
+ }
+ }
+ // delay expressed as an HTTP date (IMF-fixdate form only, as produced by
+ // virtually all servers; the legacy RFC 850 / asctime forms are not accepted)
+ try {
+ Instant date = Instant.from(HTTP_DATE_FORMATTER.parse(retryAfter));
+ long delay = date.toEpochMilli() - System.currentTimeMillis();
+ return delay > 0 ? delay : -1;
+ } catch (DateTimeException e) {
+ LOG.debug("Invalid Retry-After header value: {}", retryAfter);
+ return -1;
+ }
+ }
+
/** This class picks items from queues and fetches the pages. */
private class FetcherThread extends Thread {
@@ -501,6 +588,9 @@ public class FetcherBolt extends StatusEmitterBolt {
/** Hard timeout in seconds for a single protocol fetch. -1 means disabled. */
private long fetchTimeout = -1;
+ /** Upper bound in ms for the recorded Retry-After delay; -1 means no cap. */
+ private long maxRetryAfter = -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.
@@ -521,6 +611,8 @@ public class FetcherBolt extends StatusEmitterBolt {
this.threadNum = num;
timeoutInQueues = ConfUtils.getLong(conf, QUEUED_TIMEOUT_PARAM_KEY, timeoutInQueues);
fetchTimeout = ConfUtils.getLong(conf, FETCH_TIMEOUT_PARAM_KEY, fetchTimeout);
+ long maxRetryAfterSecs = ConfUtils.getLong(conf, MAX_RETRY_AFTER_PARAM_KEY, -1L);
+ maxRetryAfter = maxRetryAfterSecs < 0 ? -1L : maxRetryAfterSecs * 1000L;
protocolMetadataPrefix =
ConfUtils.getString(
conf,
@@ -801,6 +893,41 @@ public class FetcherBolt extends StatusEmitterBolt {
mergedMetadata.setValue("fetch.timeInQueues", Long.toString(timeInQueues));
+ // report a server-requested back-off (Retry-After, e.g. on a
+ // 429 or 503) on the host-info stream so a host-aware consumer
+ // can block the host. The fetch itself is not delayed here.
+ // See #867 / #784.
+ // only on a rate-limit (429) or unavailable (503) response does
+ // Retry-After signal a host back-off worth acting on
+ int statusCode = response.getStatusCode();
+ if (statusCode == 429 || statusCode == 503) {
+ long retryAfter =
+ parseRetryAfterDelay(
+ response.getMetadata().getFirstValue(RETRY_AFTER_KEY));
+ if (retryAfter > 0 && maxRetryAfter >= 0 && retryAfter > maxRetryAfter) {
+ LOG.debug(
+ "Capping Retry-After for {} from {} to {} ms",
+ fit.url,
+ retryAfter,
+ maxRetryAfter);
+ retryAfter = maxRetryAfter;
+ }
+ if (retryAfter > 0) {
+ long blockUntilSecs = (System.currentTimeMillis() + retryAfter) / 1000L;
+ // key by the frontier's partition mode (partition.url.mode)
+ // so the consumer blocks the matching frontier queue.
+ // fit.queueId uses fetcher.queue.mode, which may differ.
+ String hostKey = partitioner.getPartition(fit.url, mergedMetadata);
+ if (hostKey == null) {
+ hostKey = fit.queueId;
+ }
+ collector.emit(
+ Constants.HOST_INFO_STREAM_NAME,
+ fit.tuple,
+ new Values(hostKey, blockUntilSecs));
+ }
+ }
+
// determine the status based on the status code
final Status status = Status.fromHTTPCode(response.getStatusCode());
@@ -962,6 +1089,9 @@ public class FetcherBolt extends StatusEmitterBolt {
this.fetchQueues = new FetchItemQueues(conf);
+ this.partitioner = new URLPartitioner();
+ this.partitioner.configure(conf);
+
this.taskId = context.getThisTaskId();
int threadCount = ConfUtils.getInt(conf, "fetcher.threads.number", 10);
@@ -1008,6 +1138,7 @@ public class FetcherBolt extends StatusEmitterBolt {
public void declareOutputFields(OutputFieldsDeclarer declarer) {
super.declareOutputFields(declarer);
declarer.declare(new Fields("url", "content", "metadata"));
+ declarer.declareStream(Constants.HOST_INFO_STREAM_NAME, new Fields("key", "blockUntil"));
}
@Override
diff --git a/core/src/main/resources/crawler-default.yaml b/core/src/main/resources/crawler-default.yaml
index d45cbc6b..44e567f8 100644
--- a/core/src/main/resources/crawler-default.yaml
+++ b/core/src/main/resources/crawler-default.yaml
@@ -52,6 +52,15 @@ config:
# use the delay specified in the robots.txt
fetcher.server.delay.force: false
+ # max. delay in seconds honoured when a server requests a back-off via the
+ # Retry-After HTTP response header (e.g. on a 429 or 503). The parsed delay is
+ # emitted on the "hostinfo" stream (key, blockUntil-in-epoch-seconds) for a
+ # host-aware consumer to act on - the fetch itself is not delayed here.
+ # (-1) honour the delay requested by the server as-is, with no upper bound
+ # (>=0) cap the delay to this many seconds. The default of 24h bounds a
+ # misbehaving or hostile server from pinning a host queue for too long.
+ fetcher.max.retry.after: 86400
+
# time bucket to use for the metrics sent by the Fetcher
fetcher.metrics.time.bucket.secs: 10
diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltRetryAfterParserTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltRetryAfterParserTest.java
new file mode 100644
index 00000000..ffc674e9
--- /dev/null
+++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltRetryAfterParserTest.java
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.stormcrawler.bolt;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Locale;
+import java.util.Random;
+import org.junit.jupiter.api.Test;
+
+/** Unit and randomised (fuzz) coverage for {@link FetcherBolt#parseRetryAfterDelay(String)}. */
+class FetcherBoltRetryAfterParserTest {
+
+ private static final DateTimeFormatter HTTP_DATE =
+ DateTimeFormatter.ofPattern("EEE, dd MMM yyyy HH:mm:ss 'GMT'", Locale.ROOT)
+ .withZone(ZoneOffset.UTC);
+
+ @Test
+ void parseRetryAfterSeconds() {
+ assertEquals(120_000L, FetcherBolt.parseRetryAfterDelay("120"));
+ // surrounding whitespace is tolerated
+ assertEquals(120_000L, FetcherBolt.parseRetryAfterDelay(" 120 "));
+ // zero is a valid (if pointless) delay and is not treated as "absent"
+ assertEquals(0L, FetcherBolt.parseRetryAfterDelay("0"));
+ }
+
+ @Test
+ void parseRetryAfterFutureDate() {
+ Instant future = Instant.now().plusSeconds(300);
+ long delay = FetcherBolt.parseRetryAfterDelay(HTTP_DATE.format(future));
+ // allow a small tolerance for the time elapsed during the call
+ assertTrue(delay > 250_000 && delay <= 300_000, "delay was " + delay);
+ }
+
+ @Test
+ void parseRetryAfterPastDateReturnsMinusOne() {
+ Instant past = Instant.now().minusSeconds(300);
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay(HTTP_DATE.format(past)));
+ }
+
+ @Test
+ void parseRetryAfterBlankOrMalformedReturnsMinusOne() {
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay(null));
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay(""));
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay(" "));
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay("not-a-date"));
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay("12.5"));
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay("-5"));
+ }
+
+ @Test
+ void parseRetryAfterOverlongValueReturnsMinusOne() {
+ // an implausibly long header value is rejected without further parsing
+ String tooLong = "1".repeat(65);
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay(tooLong));
+ }
+
+ @Test
+ void parseRetryAfterParseOverflowReturnsMinusOne() {
+ // too large to fit in a long at all
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay("999999999999999999999999999"));
+ }
+
+ @Test
+ void parseRetryAfterMillisecondOverflowReturnsMinusOne() {
+ // parses as a valid long, but * 1000 overflows a long (regression: was a
+ // silent overflow yielding a negative delay)
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay("3687950425865536959"));
+ assertEquals(-1L, FetcherBolt.parseRetryAfterDelay(Long.toString(Long.MAX_VALUE)));
+ }
+
+ @Test
+ void parseRetryAfterRandomSecondsRoundTrip() {
+ Random rnd = new Random(7);
+ for (int i = 0; i < 1_000; i++) {
+ long secs = rnd.nextInt(1_000_000);
+ assertEquals(secs * 1000L, FetcherBolt.parseRetryAfterDelay(Long.toString(secs)));
+ }
+ }
+
+ @Test
+ void parseRetryAfterNeverThrowsForRandomInput() {
+ Random rnd = new Random(42);
+ for (int i = 0; i < 10_000; i++) {
+ String input = randomRetryAfterValue(rnd);
+ long result = FetcherBolt.parseRetryAfterDelay(input);
+ // contract: never throws, never returns anything below -1
+ assertTrue(result >= -1L, "value " + result + " for input <" + input + ">");
+ }
+ }
+
+ /**
+ * Produces a wide spectrum of inputs: digit runs (including values that overflow a long or
+ * overflow when converted to ms), real HTTP dates (past and future), blanks, random printable
+ * garbage and nulls.
+ */
+ private static String randomRetryAfterValue(Random rnd) {
+ switch (rnd.nextInt(5)) {
+ case 0:
+ int len = 1 + rnd.nextInt(40);
+ StringBuilder digits = new StringBuilder();
+ for (int i = 0; i < len; i++) {
+ digits.append((char) ('0' + rnd.nextInt(10)));
+ }
+ return digits.toString();
+ case 1:
+ long offset = (rnd.nextBoolean() ? 1L : -1L) * rnd.nextInt(1_000_000);
+ return HTTP_DATE.format(Instant.now().plusSeconds(offset));
+ case 2:
+ return "";
+ case 3:
+ int glen = rnd.nextInt(30);
+ StringBuilder garbage = new StringBuilder();
+ for (int i = 0; i < glen; i++) {
+ garbage.append((char) (32 + rnd.nextInt(95)));
+ }
+ return garbage.toString();
+ default:
+ return 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 8019bcf3..69ea6b95 100644
--- a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java
@@ -17,7 +17,31 @@
package org.apache.stormcrawler.bolt;
+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.urlMatching;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.tuple.Tuple;
+import org.apache.stormcrawler.Constants;
+import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.TestOutputCollector;
+import org.apache.stormcrawler.TestUtil;
+import org.apache.stormcrawler.util.URLPartitioner;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
public class FetcherBoltTest extends AbstractFetcherBoltTest {
@@ -25,4 +49,106 @@ public class FetcherBoltTest extends AbstractFetcherBoltTest {
void setUpContext() throws Exception {
bolt = new FetcherBolt();
}
+
+ // --- a rate-limited server emits a host back-off signal on the host-info stream ---
+
+ @Test
+ void retryAfterEmitsHostInfo(WireMockRuntimeInfo wmRuntimeInfo) {
+ stubFor(
+ get(urlMatching(".+"))
+ .willReturn(aResponse().withStatus(429).withHeader("Retry-After", "120")));
+
+ TestOutputCollector output = runFetch(wmRuntimeInfo, new HashMap<>());
+
+ List<List<Object>> hostInfo = output.getEmitted(Constants.HOST_INFO_STREAM_NAME);
+ assertEquals(1, hostInfo.size());
+ // key is the lower-cased host of the fetched URL
+ assertEquals("localhost", hostInfo.get(0).get(0));
+ long blockUntil = (Long) hostInfo.get(0).get(1);
+ long nowSecs = System.currentTimeMillis() / 1000L;
+ // ~120s in the future, with a few seconds of slack
+ assertTrue(
+ blockUntil >= nowSecs + 110 && blockUntil <= nowSecs + 125,
+ "blockUntil was " + blockUntil + " now " + nowSecs);
+ }
+
+ @Test
+ void retryAfterKeyUsesFrontierPartitionMode(WireMockRuntimeInfo wmRuntimeInfo) {
+ stubFor(
+ get(urlMatching(".+"))
+ .willReturn(aResponse().withStatus(429).withHeader("Retry-After", "120")));
+ // the frontier partitions its queues by partition.url.mode; the emitted
+ // host-info key must match that queue key so blockQueueUntil hits the
+ // right queue - not the fetcher's own queue mode (fetcher.queue.mode,
+ // byHost by default). Set the two to different modes to prove it.
+ Map<String, Object> extra = new HashMap<>();
+ extra.put(Constants.PARTITION_MODEParamName, "byIP");
+
+ TestOutputCollector output = runFetch(wmRuntimeInfo, extra);
+
+ List<List<Object>> hostInfo = output.getEmitted(Constants.HOST_INFO_STREAM_NAME);
+ assertEquals(1, hostInfo.size());
+ String url = "http://localhost:" + wmRuntimeInfo.getHttpPort() + "/";
+ // expected key is what the frontier used to create the queue: the
+ // partition key under partition.url.mode (here the resolved IP)
+ String expectedKey = URLPartitioner.getPartition(url, new Metadata(), "byIP");
+ assertEquals(expectedKey, hostInfo.get(0).get(0));
+ }
+
+ @Test
+ void retryAfterCappedByConfig(WireMockRuntimeInfo wmRuntimeInfo) {
+ stubFor(
+ get(urlMatching(".+"))
+ .willReturn(aResponse().withStatus(429).withHeader("Retry-After", "3600")));
+ Map<String, Object> extra = new HashMap<>();
+ extra.put("fetcher.max.retry.after", 60L);
+
+ TestOutputCollector output = runFetch(wmRuntimeInfo, extra);
+
+ List<List<Object>> hostInfo = output.getEmitted(Constants.HOST_INFO_STREAM_NAME);
+ assertEquals(1, hostInfo.size());
+ long blockUntil = (Long) hostInfo.get(0).get(1);
+ long nowSecs = System.currentTimeMillis() / 1000L;
+ // capped to 60s, not 3600
+ assertTrue(blockUntil <= nowSecs + 65, "blockUntil was " + blockUntil);
+ }
+
+ @Test
+ void noHostInfoWhenHeaderAbsent(WireMockRuntimeInfo wmRuntimeInfo) {
+ stubFor(get(urlMatching(".+")).willReturn(aResponse().withStatus(429)));
+
+ TestOutputCollector output = runFetch(wmRuntimeInfo, new HashMap<>());
+
+ assertTrue(output.getEmitted(Constants.HOST_INFO_STREAM_NAME).isEmpty());
+ }
+
+ /** Fetches a single URL against the WireMock server and returns the collector. */
+ private TestOutputCollector runFetch(
+ WireMockRuntimeInfo wmRuntimeInfo, Map<String, Object> extraConfig) {
+ // allow robots.txt explicitly: the tests use a catch-all 429 stub, and a
+ // 429/error on robots.txt makes crawler-commons treat the host as
+ // "deny all", so the page would never be fetched. Higher priority than
+ // the catch-all so it always wins for /robots.txt.
+ stubFor(
+ get(urlPathEqualTo("/robots.txt"))
+ .atPriority(1)
+ .willReturn(aResponse().withStatus(404)));
+
+ TestOutputCollector output = new TestOutputCollector();
+ Map<String, Object> config = new HashMap<>();
+ config.put("http.agent.name", "this_is_only_a_test");
+ config.putAll(extraConfig);
+ bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+ Tuple tuple = mock(Tuple.class);
+ when(tuple.getSourceComponent()).thenReturn("source");
+ when(tuple.getStringByField("url"))
+ .thenReturn("http://localhost:" + wmRuntimeInfo.getHttpPort() + "/");
+ when(tuple.getValueByField("metadata")).thenReturn(null);
+ bolt.execute(tuple);
+
+ await().atMost(30, TimeUnit.SECONDS)
+ .until(() -> !output.getEmitted(Constants.StatusStreamName).isEmpty());
+ return output;
+ }
}
diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/HostBlockBolt.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/HostBlockBolt.java
new file mode 100644
index 00000000..d882d8a1
--- /dev/null
+++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/HostBlockBolt.java
@@ -0,0 +1,117 @@
+/*
+ * 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.urlfrontier;
+
+import crawlercommons.urlfrontier.CrawlID;
+import crawlercommons.urlfrontier.URLFrontierGrpc;
+import crawlercommons.urlfrontier.URLFrontierGrpc.URLFrontierStub;
+import crawlercommons.urlfrontier.Urlfrontier.BlockQueueParams;
+import crawlercommons.urlfrontier.Urlfrontier.Empty;
+import io.grpc.ManagedChannel;
+import io.grpc.stub.StreamObserver;
+import java.util.Map;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.tuple.Tuple;
+import org.apache.stormcrawler.util.ConfUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Consumes host back-off signals emitted by the {@code FetcherBolt} on the {@code hostinfo} stream
+ * and blocks the corresponding queue in URLFrontier until the requested time, via {@code
+ * blockQueueUntil}. See issues #867 and #784. Wire it with a fields grouping on {@code "key"} from
+ * the FetcherBolt's {@code hostinfo} stream. The {@code "key"} is the frontier queue key derived
+ * from {@code partition.url.mode} (the same setting the frontier uses to assign queues), so the
+ * block targets the matching queue. Connects via {@code urlfrontier.host} / {@code
+ * urlfrontier.port}; multi-node address resolution is not supported.
+ *
+ * <p>The block is fire-and-forget: a failed call is logged but the tuple is acked anyway. A missed
+ * block means the host is fetched once more and the next 429 re-emits the signal.
+ */
+public class HostBlockBolt extends BaseRichBolt {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HostBlockBolt.class);
+
+ private static final StreamObserver<Empty> NOOP_OBSERVER =
+ new StreamObserver<>() {
+ @Override
+ public void onNext(Empty value) {}
+
+ @Override
+ public void onError(Throwable t) {
+ LOG.warn("blockQueueUntil failed", t);
+ }
+
+ @Override
+ public void onCompleted() {}
+ };
+
+ private OutputCollector collector;
+ private ManagedChannel channel;
+ private URLFrontierStub frontier;
+ private String globalCrawlID;
+
+ @Override
+ public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector c) {
+ this.collector = c;
+ this.globalCrawlID =
+ ConfUtils.getString(conf, Constants.URLFRONTIER_CRAWL_ID_KEY, CrawlID.DEFAULT);
+ // build channel + stub as in Spout.prepare (host/port only)
+ this.channel =
+ ManagedChannelUtil.createChannel(
+ ConfUtils.getString(
+ conf,
+ Constants.URLFRONTIER_HOST_KEY,
+ Constants.URLFRONTIER_DEFAULT_HOST),
+ ConfUtils.getInt(
+ conf,
+ Constants.URLFRONTIER_PORT_KEY,
+ Constants.URLFRONTIER_DEFAULT_PORT));
+ this.frontier = URLFrontierGrpc.newStub(channel).withWaitForReady();
+ }
+
+ @Override
+ public void execute(Tuple t) {
+ String key = t.getStringByField("key");
+ long blockUntil = t.getLongByField("blockUntil");
+ BlockQueueParams params =
+ BlockQueueParams.newBuilder()
+ .setKey(key)
+ .setCrawlID(globalCrawlID)
+ .setTime(blockUntil)
+ .setLocal(false)
+ .build();
+ frontier.blockQueueUntil(params, NOOP_OBSERVER);
+ collector.ack(t);
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ // terminal bolt
+ }
+
+ @Override
+ public void cleanup() {
+ if (channel != null) {
+ channel.shutdown();
+ }
+ }
+}
diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/HostBlockBoltTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/HostBlockBoltTest.java
new file mode 100644
index 00000000..a3e87d16
--- /dev/null
+++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/HostBlockBoltTest.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.urlfrontier;
+
+import static org.awaitility.Awaitility.await;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import crawlercommons.urlfrontier.CrawlID;
+import crawlercommons.urlfrontier.URLFrontierGrpc;
+import crawlercommons.urlfrontier.URLFrontierGrpc.URLFrontierBlockingStub;
+import crawlercommons.urlfrontier.Urlfrontier.BlockQueueParams;
+import crawlercommons.urlfrontier.Urlfrontier.GetParams;
+import crawlercommons.urlfrontier.Urlfrontier.URLInfo;
+import io.grpc.ManagedChannel;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.tuple.Tuple;
+import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.TestOutputCollector;
+import org.apache.stormcrawler.TestUtil;
+import org.apache.stormcrawler.persistence.Status;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers(disabledWithoutDocker = true)
+class HostBlockBoltTest {
+
+ private static final String HOST = "example.com";
+
+ private URLFrontierContainer container;
+ private ManagedChannel channel;
+ private URLFrontierBlockingStub blocking;
+
+ @BeforeEach
+ void before() {
+ String image = "crawlercommons/url-frontier";
+ String version = System.getProperty("urlfrontier-version");
+ if (version != null) {
+ image += ":" + version;
+ }
+ container = new URLFrontierContainer(image);
+ container.start();
+ var connection = container.getFrontierConnection();
+ channel = ManagedChannelUtil.createChannel(connection.getHost(), connection.getPort());
+ blocking = URLFrontierGrpc.newBlockingStub(channel);
+ }
+
+ @AfterEach
+ void after() {
+ if (channel != null) {
+ channel.shutdownNow();
+ }
+ container.close();
+ }
+
+ private Map<String, Object> frontierConfig() {
+ var connection = container.getFrontierConnection();
+ Map<String, Object> config = new HashMap<>();
+ config.put(Constants.URLFRONTIER_HOST_KEY, connection.getHost());
+ config.put(Constants.URLFRONTIER_PORT_KEY, connection.getPort());
+ return config;
+ }
+
+ /** Seeds one DISCOVERED url for {@link #HOST} so the queue exists and is fetchable. */
+ private void seedOneUrl() {
+ StatusUpdaterBolt seeder = new StatusUpdaterBolt();
+ TestOutputCollector out = new TestOutputCollector();
+ Map<String, Object> config = frontierConfig();
+ config.put(
+ "urlbuffer.class", "org.apache.stormcrawler.persistence.urlbuffer.SimpleURLBuffer");
+ config.put("scheduler.class", "org.apache.stormcrawler.persistence.DefaultScheduler");
+ config.put("status.updater.cache.spec", "maximumSize=10000,expireAfterAccess=1h");
+ config.put("urlfrontier.updater.max.messages", 1);
+ config.put("urlfrontier.cache.expireafter.sec", 10);
+ seeder.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(out));
+
+ String url = "http://" + HOST + "/";
+ Tuple tuple = mock(Tuple.class);
+ when(tuple.getValueByField("status")).thenReturn(Status.DISCOVERED);
+ when(tuple.getStringByField("url")).thenReturn(url);
+ when(tuple.getValueByField("metadata")).thenReturn(new Metadata());
+ seeder.execute(tuple);
+
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(
+ () ->
+ out.getAckedTuples().stream()
+ .anyMatch(t -> url.equals(t.getStringByField("url"))));
+ seeder.cleanup();
+ }
+
+ /** Returns the set of queue keys currently handed out by getURLs. */
+ private Set<String> keysFromGetURLs() {
+ GetParams params =
+ GetParams.newBuilder()
+ .setCrawlID(CrawlID.DEFAULT)
+ .setMaxQueues(10)
+ .setMaxUrlsPerQueue(10)
+ .build();
+ Set<String> keys = new HashSet<>();
+ Iterator<URLInfo> it = blocking.getURLs(params);
+ while (it.hasNext()) {
+ keys.add(it.next().getKey());
+ }
+ return keys;
+ }
+
+ @Test
+ @Timeout(value = 2, unit = TimeUnit.MINUTES)
+ void blocksHostQueueUntilTime() {
+ seedOneUrl();
+
+ long nowSecs = System.currentTimeMillis() / 1000L;
+
+ // block the host for an hour via the bolt under test
+ HostBlockBolt bolt = new HostBlockBolt();
+ bolt.prepare(
+ frontierConfig(), TestUtil.getMockedTopologyContext(), mock(OutputCollector.class));
+ Tuple t = mock(Tuple.class);
+ when(t.getStringByField("key")).thenReturn(HOST);
+ when(t.getLongByField("blockUntil")).thenReturn(nowSecs + 3600);
+ bolt.execute(t);
+
+ // while blocked, getURLs must not return the host
+ await().atMost(15, TimeUnit.SECONDS).until(() -> !keysFromGetURLs().contains(HOST));
+
+ // unblock (a past time releases the queue) and the host becomes available again
+ blocking.blockQueueUntil(
+ BlockQueueParams.newBuilder()
+ .setKey(HOST)
+ .setCrawlID(CrawlID.DEFAULT)
+ .setTime(nowSecs - 1)
+ .build());
+ await().atMost(15, TimeUnit.SECONDS).until(() -> keysFromGetURLs().contains(HOST));
+
+ bolt.cleanup();
+ }
+}