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/09/30 14:49:20 UTC
(stormcrawler) branch fetcher-package created (now 0cff6394)
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a change to branch fetcher-package
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
at 0cff6394 #2057 FetcherBolt: move the fetcher internals to o.a.s.fetcher
This branch includes the following new commits:
new 0cff6394 #2057 FetcherBolt: move the fetcher internals to o.a.s.fetcher
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(stormcrawler) 01/01: #2057 FetcherBolt: move the fetcher internals to o.a.s.fetcher
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch fetcher-package
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 0cff6394dbdd5f8ca20630b37af53d3ecff2d552
Author: Davide Polato <dp...@apache.org>
AuthorDate: Wed Sep 30 16:41:03 2026 +0200
#2057 FetcherBolt: move the fetcher internals to o.a.s.fetcher
Signed-off-by: Davide Polato <dp...@apache.org>
---
.../org/apache/stormcrawler/bolt/FetcherBolt.java | 113 ++++++++++-----------
.../stormcrawler/bolt/SimpleFetcherBolt.java | 2 +
.../{bolt => fetcher}/CrawlDelayPolicy.java | 15 +--
.../stormcrawler/{bolt => fetcher}/FetchItem.java | 38 +++++--
.../{bolt => fetcher}/FetchItemQueues.java | 81 ++++++++++++---
.../{bolt => fetcher}/FetchOutcomes.java | 17 ++--
.../{bolt => fetcher}/FetchTimeoutHelpers.java | 21 ++--
.../{bolt => fetcher}/ProtocolMetrics.java | 11 +-
.../{bolt => fetcher}/RobotRulesLookup.java | 13 ++-
.../{bolt => fetcher}/CrawlDelayPolicyTest.java | 6 +-
.../{bolt => fetcher}/FetchItemQueuesTest.java | 28 ++---
.../{bolt => fetcher}/FetchOutcomesTest.java | 3 +-
.../{bolt => fetcher}/FetchTimeoutHelpersTest.java | 2 +-
.../{bolt => fetcher}/ProtocolMetricsTest.java | 2 +-
.../{bolt => fetcher}/RobotRulesLookupTest.java | 2 +-
15 files changed, 221 insertions(+), 133 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 61ca63b7..25aa5e0b 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,6 @@ import java.net.URL;
import java.time.Instant;
import java.util.Arrays;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.lang3.StringUtils;
import org.apache.storm.Config;
@@ -38,7 +37,14 @@ import org.apache.storm.utils.TupleUtils;
import org.apache.storm.utils.Utils;
import org.apache.stormcrawler.Constants;
import org.apache.stormcrawler.Metadata;
-import org.apache.stormcrawler.bolt.FetchItemQueues.FetchItemQueue;
+import org.apache.stormcrawler.fetcher.CrawlDelayPolicy;
+import org.apache.stormcrawler.fetcher.FetchItem;
+import org.apache.stormcrawler.fetcher.FetchItemQueues;
+import org.apache.stormcrawler.fetcher.FetchItemQueues.FetchItemQueue;
+import org.apache.stormcrawler.fetcher.FetchOutcomes;
+import org.apache.stormcrawler.fetcher.FetchTimeoutHelpers;
+import org.apache.stormcrawler.fetcher.ProtocolMetrics;
+import org.apache.stormcrawler.fetcher.RobotRulesLookup;
import org.apache.stormcrawler.metrics.CrawlerMetrics;
import org.apache.stormcrawler.metrics.ScopedCounter;
import org.apache.stormcrawler.metrics.ScopedReducedMetric;
@@ -152,7 +158,7 @@ public class FetcherBolt extends StatusEmitterBolt {
activeThreads.incrementAndGet(); // count threads
- beingFetched[threadNum] = fit.url;
+ beingFetched[threadNum] = fit.url();
LOG.debug(
"[Fetcher #{}] {} => activeThreads={}, spinWaiting={}, queueID={}",
@@ -160,14 +166,14 @@ public class FetcherBolt extends StatusEmitterBolt {
getName(),
activeThreads,
spinWaiting,
- fit.queueId);
+ fit.queueId());
- LOG.debug("[Fetcher #{}] {} : Fetching {}", taskId, getName(), fit.url);
+ LOG.debug("[Fetcher #{}] {} : Fetching {}", taskId, getName(), fit.url());
Metadata metadata = null;
- if (fit.tuple.contains("metadata")) {
- metadata = (Metadata) fit.tuple.getValueByField("metadata");
+ if (fit.tuple().contains("metadata")) {
+ metadata = (Metadata) fit.tuple().getValueByField("metadata");
}
if (metadata == null) {
metadata = new Metadata();
@@ -184,54 +190,54 @@ public class FetcherBolt extends StatusEmitterBolt {
boolean saturated = false;
try {
- URL url = URLUtil.toURL(fit.url);
+ URL url = URLUtil.toURL(fit.url());
Protocol protocol = protocolFactory.getProtocol(url);
if (protocol == null) {
throw new RuntimeException(
- "No protocol implementation found for " + fit.url);
+ "No protocol implementation found for " + fit.url());
}
RobotRulesLookup.Result robots =
- robotsLookup.lookup(protocol, fit.url, metadata);
+ robotsLookup.lookup(protocol, fit.url(), metadata);
BaseRobotRules rules = robots.rules();
- discoverSitemaps(fit.tuple, url, metadata, robots);
+ discoverSitemaps(fit.tuple(), url, metadata, robots);
- if (!rules.isAllowed(fit.url)) {
- LOG.info("Denied by robots.txt: {}", fit.url);
+ if (!rules.isAllowed(fit.url())) {
+ LOG.info("Denied by robots.txt: {}", fit.url());
// pass the info about denied by robots
metadata.setValue(Constants.STATUS_ERROR_CAUSE, "robots.txt");
collector.emit(
org.apache.stormcrawler.Constants.StatusStreamName,
- fit.tuple,
- new Values(fit.url, metadata, Status.ERROR));
+ fit.tuple(),
+ new Values(fit.url(), metadata, Status.ERROR));
// no need to wait next time as we won't request from
// that site
asap = true;
continue;
}
- FetchItemQueue fiq = fetchQueues.getFetchItemQueue(fit.queueId, metadata);
+ FetchItemQueue fiq = fetchQueues.getFetchItemQueue(fit.queueId(), metadata);
CrawlDelayPolicy.Decision decision =
crawlDelayPolicy.decide(
- fit.url,
- fit.queueId,
+ fit.url(),
+ fit.queueId(),
rules.getCrawlDelay(),
- fiq.crawlDelay.get());
+ fiq.getCrawlDelay());
if (decision.action() == CrawlDelayPolicy.Action.SKIP) {
// pass the info about crawl delay
metadata.setValue(Constants.STATUS_ERROR_CAUSE, "crawl_delay");
collector.emit(
org.apache.stormcrawler.Constants.StatusStreamName,
- fit.tuple,
- new Values(fit.url, metadata, Status.ERROR));
+ fit.tuple(),
+ new Values(fit.url(), metadata, Status.ERROR));
// no need to wait next time as we won't request
// from that site
asap = true;
continue;
}
if (decision.action() == CrawlDelayPolicy.Action.APPLY) {
- fiq.crawlDelay.set(decision.delay());
+ fiq.setCrawlDelay(decision.delay());
robotsCrawlDelaySecs = decision.robotsCrawlDelaySecs();
if (robotsCrawlDelaySecs != null) {
metadata.setValue(
@@ -240,13 +246,15 @@ public class FetcherBolt extends StatusEmitterBolt {
}
long start = System.currentTimeMillis();
- long timeInQueues = start - fit.creationTime;
+ long timeInQueues = start - fit.creationTime();
// waited longer than fetcher.timeout.queue: not fetched, acked without a
// status (see finally)
if (timeoutInQueues != -1 && timeInQueues > timeoutInQueues * 1000) {
LOG.info(
- "[Fetcher #{}] Waited in queue for too long - {}", taskId, fit.url);
+ "[Fetcher #{}] Waited in queue for too long - {}",
+ taskId,
+ fit.url());
eventCounter.scope("queue.timeout").incrBy(1);
// no need to wait next time as we won't request from
// that site
@@ -258,9 +266,9 @@ public class FetcherBolt extends StatusEmitterBolt {
ProtocolResponse response;
response =
fetchHelpers.call(
- () -> protocol.getProtocolOutput(fit.url, fetchMetadata),
+ () -> protocol.getProtocolOutput(fit.url(), fetchMetadata),
protocol,
- fit.url,
+ fit.url(),
fetchMetadata);
long timeFetching = System.currentTimeMillis() - start;
@@ -281,7 +289,7 @@ public class FetcherBolt extends StatusEmitterBolt {
LOG.info(
"[Fetcher #{}] Fetched {} with status {} in msec {}",
taskId,
- fit.url,
+ fit.url(),
response.getStatusCode(),
timeFetching);
@@ -296,8 +304,8 @@ public class FetcherBolt extends StatusEmitterBolt {
// send content for parsing
collector.emit(
Utils.DEFAULT_STREAM_ID,
- fit.tuple,
- new Values(fit.url, response.getContent(), outcome.metadata()));
+ fit.tuple(),
+ new Values(fit.url(), response.getContent(), outcome.metadata()));
} else {
// allowRedirs() can be overridden: it is asked for every redirect and
// nothing else (#954)
@@ -305,7 +313,7 @@ public class FetcherBolt extends StatusEmitterBolt {
&& allowRedirs()
&& outcome.redirectTarget() != null) {
emitOutlink(
- fit.tuple,
+ fit.tuple(),
url,
outcome.redirectTarget(),
outcome.metadata(),
@@ -313,8 +321,8 @@ public class FetcherBolt extends StatusEmitterBolt {
}
collector.emit(
Constants.StatusStreamName,
- fit.tuple,
- new Values(fit.url, outcome.metadata(), outcome.status()));
+ fit.tuple(),
+ new Values(fit.url(), outcome.metadata(), outcome.status()));
}
} catch (FetchTimeoutHelpers.SaturatedException e) {
@@ -333,25 +341,25 @@ public class FetcherBolt extends StatusEmitterBolt {
saturated = true;
} catch (Exception exece) {
FetchOutcomes.Outcome outcome =
- fetchOutcomes.ofFailure(exece, fit.url, metadata);
+ fetchOutcomes.ofFailure(exece, fit.url(), metadata);
collector.emit(
Constants.StatusStreamName,
- fit.tuple,
- new Values(fit.url, outcome.metadata(), outcome.status()));
+ fit.tuple(),
+ new Values(fit.url(), outcome.metadata(), outcome.status()));
} finally {
if (saturated) {
long delay = fetchQueues.backOffFetchItem(fit);
LOG.debug(
"[Fetcher #{}] queue {} backed off for {} ms",
taskId,
- fit.queueId,
+ fit.queueId(),
delay);
} else {
fetchQueues.finishFetchItem(fit, asap);
}
activeThreads.decrementAndGet(); // count threads
// ack it whatever happens
- collector.ack(fit.tuple);
+ collector.ack(fit.tuple());
beingFetched[threadNum] = "";
}
}
@@ -443,14 +451,14 @@ public class FetcherBolt extends StatusEmitterBolt {
context,
stormConf,
"in_queues",
- () -> fetchQueues.inQueues.get(),
+ () -> fetchQueues.numQueuedItems(),
metricsTimeBucketSecs);
CrawlerMetrics.registerGauge(
context,
stormConf,
"num_queues",
- () -> fetchQueues.queues.size(),
+ () -> fetchQueues.numQueues(),
metricsTimeBucketSecs);
this.averagedMetrics =
@@ -474,7 +482,7 @@ public class FetcherBolt extends StatusEmitterBolt {
new CrawlDelayPolicy(
ConfUtils.getInt(conf, "fetcher.max.crawl.delay", 30) * 1000L,
ConfUtils.getBoolean(conf, "fetcher.max.crawl.delay.force", false),
- fetchQueues.crawlDelay,
+ fetchQueues.defaultCrawlDelay(),
ConfUtils.getBoolean(conf, "fetcher.server.delay.force", false));
this.taskId = context.getThisTaskId();
@@ -555,7 +563,7 @@ public class FetcherBolt extends StatusEmitterBolt {
}
if (this.maxNumberUrlsInQueues != -1) {
- while (this.activeThreads.get() + this.fetchQueues.inQueues.get()
+ while (this.activeThreads.get() + this.fetchQueues.numQueuedItems()
>= maxNumberUrlsInQueues) {
try {
Thread.sleep(500);
@@ -567,8 +575,8 @@ public class FetcherBolt extends StatusEmitterBolt {
"[Fetcher #{}] Threads : {}\tqueues : {}\tin_queues : {}",
taskId,
this.activeThreads.get(),
- this.fetchQueues.queues.size(),
- this.fetchQueues.inQueues.get());
+ this.fetchQueues.numQueues(),
+ this.fetchQueues.numQueuedItems());
}
}
@@ -610,24 +618,11 @@ public class FetcherBolt extends StatusEmitterBolt {
}
/**
- * Logs the content of the queues and the URLs being fetched. Not synchronized: the queues and
- * their items are iterated with weakly consistent iterators while the fetcher threads keep
- * working, so the dump is a smear over the time it takes to produce it rather than a
- * point-in-time snapshot. Sizes and item lists may not add up exactly.
+ * Logs the content of the queues and the URLs being fetched. The URLs being fetched are read
+ * without locking, like the queues, so the two lists may not match exactly.
*/
private void logQueuesContent() {
- StringBuilder sb = new StringBuilder();
- sb.append("\nNum queues : ").append(fetchQueues.queues.size());
- for (Entry<String, FetchItemQueue> entry : fetchQueues.queues.entrySet()) {
- sb.append("\nQueue ID : ").append(entry.getKey());
- FetchItemQueue fiq = entry.getValue();
- sb.append("\t size : ").append(fiq.getQueueSize());
- sb.append("\t in progress : ").append(fiq.getInProgressSize());
- for (FetchItem fetchItem : fiq.queue) {
- sb.append("\n\t").append(fetchItem.url);
- }
- }
- LOG.info("Dumping queue content {}", sb.toString());
+ LOG.info("Dumping queue content {}", fetchQueues.dump());
StringBuilder sb2 = new StringBuilder("\n");
// dump the list of URLs being fetched
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 968faae0..bdd03eb2 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java
+++ b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java
@@ -42,6 +42,8 @@ import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
import org.apache.stormcrawler.Constants;
import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.fetcher.FetchTimeoutHelpers;
+import org.apache.stormcrawler.fetcher.ProtocolMetrics;
import org.apache.stormcrawler.metrics.CrawlerMetrics;
import org.apache.stormcrawler.metrics.ScopedCounter;
import org.apache.stormcrawler.metrics.ScopedReducedMetric;
diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/CrawlDelayPolicy.java b/core/src/main/java/org/apache/stormcrawler/fetcher/CrawlDelayPolicy.java
similarity index 91%
rename from core/src/main/java/org/apache/stormcrawler/bolt/CrawlDelayPolicy.java
rename to core/src/main/java/org/apache/stormcrawler/fetcher/CrawlDelayPolicy.java
index 448b311f..c41973e5 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/CrawlDelayPolicy.java
+++ b/core/src/main/java/org/apache/stormcrawler/fetcher/CrawlDelayPolicy.java
@@ -15,8 +15,9 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
+import org.apache.stormcrawler.bolt.FetcherBolt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -24,13 +25,15 @@ import org.slf4j.LoggerFactory;
* Decides what the crawl delay of robots.txt does to the fetch queue of a URL, following
* fetcher.max.crawl.delay, fetcher.max.crawl.delay.force, fetcher.server.delay and
* fetcher.server.delay.force. Delays are in milliseconds.
+ *
+ * <p>For internal use only. Not part of StormCrawler's public API.
*/
-final class CrawlDelayPolicy {
+public final class CrawlDelayPolicy {
// the bolt's category: log configurations for FetcherBolt keep covering these lines
private static final Logger LOG = LoggerFactory.getLogger(FetcherBolt.class);
- enum Action {
+ public enum Action {
/**
* Leave the fetch queue as it is: robots.txt has no delay, or the one the queue already
* has, even above the cap. The caller must not write back the delay it read, which could
@@ -47,7 +50,7 @@ final class CrawlDelayPolicy {
* @param robotsCrawlDelaySecs the robots.txt delay in seconds, rounded up, when it was longer
* than fetcher.max.crawl.delay and capped; null otherwise
*/
- record Decision(Action action, long delay, String robotsCrawlDelaySecs) {
+ public record Decision(Action action, long delay, String robotsCrawlDelaySecs) {
static final Decision UNCHANGED = new Decision(Action.UNCHANGED, 0, null);
static final Decision SKIP = new Decision(Action.SKIP, 0, null);
}
@@ -62,7 +65,7 @@ final class CrawlDelayPolicy {
// specifies a shorter crawl-delay
private final boolean serverDelayForce;
- CrawlDelayPolicy(
+ public CrawlDelayPolicy(
long maxCrawlDelay,
boolean maxCrawlDelayForce,
long serverDelay,
@@ -78,7 +81,7 @@ final class CrawlDelayPolicy {
* @param queueId the ID of the fetch queue, for the log
* @param robotsDelay the crawl delay of robots.txt, not positive when there is none
*/
- Decision decide(String url, String queueId, long robotsDelay, long queueDelay) {
+ public Decision decide(String url, String queueId, long robotsDelay, long queueDelay) {
if (robotsDelay <= 0 || robotsDelay == queueDelay) {
return Decision.UNCHANGED;
}
diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchItem.java b/core/src/main/java/org/apache/stormcrawler/fetcher/FetchItem.java
similarity index 81%
rename from core/src/main/java/org/apache/stormcrawler/bolt/FetchItem.java
rename to core/src/main/java/org/apache/stormcrawler/fetcher/FetchItem.java
index 7ce1955d..16b6ce36 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/FetchItem.java
+++ b/core/src/main/java/org/apache/stormcrawler/fetcher/FetchItem.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import crawlercommons.domains.PaidLevelDomain;
import java.net.InetAddress;
@@ -24,18 +24,23 @@ import java.net.UnknownHostException;
import java.util.Locale;
import org.apache.commons.lang3.StringUtils;
import org.apache.storm.tuple.Tuple;
+import org.apache.stormcrawler.bolt.FetcherBolt;
import org.apache.stormcrawler.util.URLUtil;
import org.slf4j.LoggerFactory;
-/** This class described the item to be fetched. */
-class FetchItem {
+/**
+ * This class described the item to be fetched.
+ *
+ * <p>For internal use only. Not part of StormCrawler's public API.
+ */
+public final class FetchItem {
// the bolt's category: log configurations for FetcherBolt keep covering these lines
private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(FetcherBolt.class);
- String queueId;
- String url;
- Tuple tuple;
- long creationTime;
+ private final String queueId;
+ private final String url;
+ private final Tuple tuple;
+ private final long creationTime;
private FetchItem(String url, Tuple t, String queueId) {
this.url = url;
@@ -48,7 +53,7 @@ class FetchItem {
* Create an item. Queue id will be created based on <code>queueMode</code> argument, either as
* a protocol + hostname pair, protocol + IP address pair or protocol+domain pair.
*/
- public static FetchItem create(URL u, String url, Tuple t, String queueMode) {
+ static FetchItem create(URL u, String url, Tuple t, String queueMode) {
String queueId;
@@ -93,4 +98,21 @@ class FetchItem {
queueId = key.toLowerCase(Locale.ROOT);
return new FetchItem(url, t, queueId);
}
+
+ public String queueId() {
+ return queueId;
+ }
+
+ public String url() {
+ return url;
+ }
+
+ public Tuple tuple() {
+ return tuple;
+ }
+
+ /** When the item was created, in epoch milliseconds. */
+ public long creationTime() {
+ return creationTime;
+ }
}
diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchItemQueues.java b/core/src/main/java/org/apache/stormcrawler/fetcher/FetchItemQueues.java
similarity index 89%
rename from core/src/main/java/org/apache/stormcrawler/bolt/FetchItemQueues.java
rename to core/src/main/java/org/apache/stormcrawler/fetcher/FetchItemQueues.java
index dc28c8e4..b64b4f6b 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/FetchItemQueues.java
+++ b/core/src/main/java/org/apache/stormcrawler/fetcher/FetchItemQueues.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import java.net.URL;
import java.util.HashMap;
@@ -35,6 +35,7 @@ import java.util.regex.Pattern;
import org.apache.storm.Config;
import org.apache.storm.tuple.Tuple;
import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.bolt.FetcherBolt;
import org.apache.stormcrawler.util.ConfUtils;
import org.slf4j.LoggerFactory;
@@ -46,8 +47,10 @@ import org.slf4j.LoggerFactory;
* referenced from a {@link DelayQueue} ordered by their next fetch time: taking an item is O(log n)
* and does not require a global lock, so the executor thread adding URLs is never blocked by the
* fetcher threads.
+ *
+ * <p>For internal use only. Not part of StormCrawler's public API.
*/
-class FetchItemQueues {
+public final class FetchItemQueues {
// the bolt's category: log configurations for FetcherBolt keep covering these lines
private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(FetcherBolt.class);
@@ -80,9 +83,9 @@ class FetchItemQueues {
final Config conf;
- public static final String QUEUE_MODE_HOST = "byHost";
- public static final String QUEUE_MODE_DOMAIN = "byDomain";
- public static final String QUEUE_MODE_IP = "byIP";
+ static final String QUEUE_MODE_HOST = "byHost";
+ static final String QUEUE_MODE_DOMAIN = "byDomain";
+ static final String QUEUE_MODE_IP = "byIP";
String queueMode;
@@ -134,7 +137,7 @@ class FetchItemQueues {
final FetchItem it = FetchItem.create(u, url, input, queueMode);
final Metadata metadata = (Metadata) input.getValueByField("metadata");
while (true) {
- FetchItemQueue fiq = getFetchItemQueue(it.queueId, metadata);
+ FetchItemQueue fiq = getFetchItemQueue(it.queueId(), metadata);
synchronized (fiq) {
if (fiq.removed) {
// reaped concurrently: get a fresh one
@@ -146,15 +149,15 @@ class FetchItemQueues {
}
inQueues.incrementAndGet();
schedule(fiq, fiq.getNextFetchTime());
- LOG.debug("{} added to queue {}", url, it.queueId);
+ LOG.debug("{} added to queue {}", url, it.queueId());
return true;
}
}
public void finishFetchItem(FetchItem it, boolean asap) {
- FetchItemQueue fiq = queues.get(it.queueId);
+ FetchItemQueue fiq = queues.get(it.queueId());
if (fiq == null) {
- LOG.warn("Attempting to finish item from unknown queue: {}", it.queueId);
+ LOG.warn("Attempting to finish item from unknown queue: {}", it.queueId());
return;
}
fiq.finish(asap);
@@ -168,9 +171,9 @@ class FetchItemQueues {
* @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);
+ FetchItemQueue fiq = queues.get(it.queueId());
if (fiq == null) {
- LOG.warn("Attempting to back off item from unknown queue: {}", it.queueId);
+ LOG.warn("Attempting to back off item from unknown queue: {}", it.queueId());
return -1;
}
long delay = fiq.finishSaturated(maxBackoff);
@@ -335,12 +338,44 @@ class FetchItemQueues {
return null;
}
+ public int numQueues() {
+ return queues.size();
+ }
+
+ public int numQueuedItems() {
+ return inQueues.get();
+ }
+
+ /** Default delay between two fetches from a queue, fetcher.server.delay in milliseconds. */
+ public long defaultCrawlDelay() {
+ return crawlDelay;
+ }
+
+ /**
+ * Describes every fetch queue: its ID, size, fetches in progress and the URLs waiting. The
+ * fetcher threads keep working meanwhile, so the figures may not add up exactly.
+ */
+ public String dump() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("\nNum queues : ").append(queues.size());
+ for (Entry<String, FetchItemQueue> entry : queues.entrySet()) {
+ sb.append("\nQueue ID : ").append(entry.getKey());
+ FetchItemQueue fiq = entry.getValue();
+ sb.append("\t size : ").append(fiq.getQueueSize());
+ sb.append("\t in progress : ").append(fiq.getInProgressSize());
+ for (FetchItem fetchItem : fiq.queue) {
+ sb.append("\n\t").append(fetchItem.url());
+ }
+ }
+ return sb.toString();
+ }
+
/**
* This class handles FetchItems which come from the same host ID (be it a proto/hostname or
* proto/IP pair). It also keeps track of requests in progress and elapsed time between
* requests.
*/
- static class FetchItemQueue {
+ public static class FetchItemQueue {
final Queue<FetchItem> queue = new ConcurrentLinkedQueue<>();
final String id;
@@ -372,7 +407,7 @@ class FetchItemQueues {
/** Consecutive fetches of this queue rejected because every helper thread was busy. */
private final AtomicInteger saturations = new AtomicInteger();
- public FetchItemQueue(
+ FetchItemQueue(
String id, int maxThreads, long crawlDelay, long minCrawlDelay, int maxQueueSize) {
this.id = id;
this.maxThreads = maxThreads;
@@ -383,15 +418,31 @@ class FetchItemQueues {
setNextFetchTime(System.currentTimeMillis(), true);
}
- public int getQueueSize() {
+ int getQueueSize() {
// never negative for the metrics, even while a poll of an empty queue is in flight
return Math.max(0, size.get());
}
- public int getInProgressSize() {
+ int getInProgressSize() {
return inProgress.get();
}
+ /**
+ * The crawl delay of this queue, in milliseconds. A queue allowing more than one thread
+ * spaces its fetches by the min crawl delay instead.
+ */
+ public long getCrawlDelay() {
+ return crawlDelay.get();
+ }
+
+ /**
+ * Sets the crawl delay of this queue, in milliseconds. It overwrites the value: unlike the
+ * maximum taken when URLs are added, it can lower the delay.
+ */
+ public void setCrawlDelay(long delay) {
+ crawlDelay.set(delay);
+ }
+
long getNextFetchTime() {
return nextFetchTime.get();
}
diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchOutcomes.java b/core/src/main/java/org/apache/stormcrawler/fetcher/FetchOutcomes.java
similarity index 94%
rename from core/src/main/java/org/apache/stormcrawler/bolt/FetchOutcomes.java
rename to core/src/main/java/org/apache/stormcrawler/fetcher/FetchOutcomes.java
index 3c0b6192..46b52d3f 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/FetchOutcomes.java
+++ b/core/src/main/java/org/apache/stormcrawler/fetcher/FetchOutcomes.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import java.io.InterruptedIOException;
import java.net.UnknownHostException;
@@ -23,6 +23,9 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHeaders;
import org.apache.stormcrawler.Constants;
import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.bolt.FetcherBolt;
+import org.apache.stormcrawler.bolt.SiteMapParserBolt;
+import org.apache.stormcrawler.bolt.StatusEmitterBolt;
import org.apache.stormcrawler.metrics.ScopedCounter;
import org.apache.stormcrawler.persistence.Status;
import org.apache.stormcrawler.protocol.AbstractHttpProtocol;
@@ -36,8 +39,10 @@ import org.slf4j.LoggerFactory;
* {@link FetcherBolt} reports for the URL. Counts each response in status_{code}, each failure in
* exception, timeouts also in fetch.timeout and the fetcher.thread.timeout deadline also in
* fetch.deadline. The bolt does the emitting.
+ *
+ * <p>For internal use only. Not part of StormCrawler's public API.
*/
-final class FetchOutcomes {
+public final class FetchOutcomes {
// the bolt's category: log configurations for FetcherBolt keep covering these lines
private static final Logger LOG = LoggerFactory.getLogger(FetcherBolt.class);
@@ -55,7 +60,7 @@ final class FetchOutcomes {
* @param redirectKeyValues custom key/values for the outlink to the redirect target, as {@link
* StatusEmitterBolt#emitOutlink} takes them
*/
- record Outcome(
+ public record Outcome(
Status status,
Metadata metadata,
boolean parse,
@@ -65,7 +70,7 @@ final class FetchOutcomes {
private final String protocolMetadataPrefix;
private final ScopedCounter eventCounter;
- FetchOutcomes(String protocolMetadataPrefix, ScopedCounter eventCounter) {
+ public FetchOutcomes(String protocolMetadataPrefix, ScopedCounter eventCounter) {
this.protocolMetadataPrefix = protocolMetadataPrefix;
this.eventCounter = eventCounter;
}
@@ -76,7 +81,7 @@ final class FetchOutcomes {
* @param timeFetching milliseconds spent fetching
* @param timeInQueues milliseconds the URL waited in the fetch queues
*/
- Outcome ofResponse(
+ public Outcome ofResponse(
ProtocolResponse response,
Metadata metadata,
String robotsCrawlDelaySecs,
@@ -154,7 +159,7 @@ final class FetchOutcomes {
* @param metadata the metadata the URL came with; the failure is written to it, or to a new
* instance when it is empty, since it may then be the read-only {@link Metadata#empty}
*/
- Outcome ofFailure(Exception e, String url, Metadata metadata) {
+ public Outcome ofFailure(Exception e, String url, Metadata metadata) {
String message = e.getMessage();
if (message == null) {
message = "";
diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java b/core/src/main/java/org/apache/stormcrawler/fetcher/FetchTimeoutHelpers.java
similarity index 93%
rename from core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java
rename to core/src/main/java/org/apache/stormcrawler/fetcher/FetchTimeoutHelpers.java
index 27dc6df8..2589e988 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java
+++ b/core/src/main/java/org/apache/stormcrawler/fetcher/FetchTimeoutHelpers.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import java.util.Map;
import java.util.concurrent.Callable;
@@ -59,11 +59,13 @@ import org.apache.stormcrawler.util.ConfUtils;
*
* <p>The deadline is the one of {@link FetchTimeout#secs(Map)}, clamped to the message timeout on
* both paths.
+ *
+ * <p>For internal use only. Not part of StormCrawler's public API.
*/
-final class FetchTimeoutHelpers {
+public final class FetchTimeoutHelpers {
/** Thrown when no helper thread is available to run a call with a timeout. */
- static final class SaturatedException extends Exception {
+ public static final class SaturatedException extends Exception {
SaturatedException(String url) {
super("No fetch helper available for " + url);
}
@@ -105,7 +107,8 @@ final class FetchTimeoutHelpers {
* @param defaultMaxHelpers pool bound used unless {@code fetcher.thread.timeout.helpers} is set
* @param threadNamePrefix prefix of the helper thread names
*/
- FetchTimeoutHelpers(Map<String, Object> conf, int defaultMaxHelpers, String threadNamePrefix) {
+ public FetchTimeoutHelpers(
+ Map<String, Object> conf, int defaultMaxHelpers, String threadNamePrefix) {
this.timeoutSecs = FetchTimeout.secs(conf);
final int maxHelpers =
ConfUtils.getInt(
@@ -132,7 +135,7 @@ final class FetchTimeoutHelpers {
}
/** Registers the {@code fetchhelpers} gauge: number of helpers busy with a call. */
- void registerMetrics(TopologyContext context, Map<String, Object> conf, int bucketSecs) {
+ public void registerMetrics(TopologyContext context, Map<String, Object> conf, int bucketSecs) {
CrawlerMetrics.registerGauge(context, conf, "fetchhelpers", this::busy, bucketSecs);
}
@@ -152,12 +155,12 @@ final class FetchTimeoutHelpers {
}
/** Pool bound. */
- int maxHelpers() {
+ public int maxHelpers() {
return helpers.getMaximumPoolSize();
}
/** Largest number of helper threads ever alive. */
- int largestPoolSize() {
+ public int largestPoolSize() {
return helpers.getLargestPoolSize();
}
@@ -172,7 +175,7 @@ final class FetchTimeoutHelpers {
* @throws FetchTimeoutException when the deadline passed
* @throws Exception the protocol's own exception
*/
- <T> T call(Callable<T> call, Protocol protocol, String url, Metadata metadata)
+ public <T> T call(Callable<T> call, Protocol protocol, String url, Metadata metadata)
throws Exception {
if (timeoutSecs <= 0 || protocol.supportsFetchTimeout(url, metadata)) {
return call.call();
@@ -223,7 +226,7 @@ final class FetchTimeoutHelpers {
}
}
- void shutdown() {
+ public void shutdown() {
helpers.shutdownNow();
}
}
diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/ProtocolMetrics.java b/core/src/main/java/org/apache/stormcrawler/fetcher/ProtocolMetrics.java
similarity index 86%
rename from core/src/main/java/org/apache/stormcrawler/bolt/ProtocolMetrics.java
rename to core/src/main/java/org/apache/stormcrawler/fetcher/ProtocolMetrics.java
index b9588d96..a32480b0 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/ProtocolMetrics.java
+++ b/core/src/main/java/org/apache/stormcrawler/fetcher/ProtocolMetrics.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.stormcrawler.Metadata;
@@ -25,9 +25,12 @@ import org.slf4j.LoggerFactory;
/**
* Copies the timings a protocol reports under {@value #METRICS_PREFIX} into the metric registry of
- * a fetcher bolt. Shared by {@link FetcherBolt} and {@link SimpleFetcherBolt}.
+ * a fetcher bolt. Shared by {@link org.apache.stormcrawler.bolt.FetcherBolt} and {@link
+ * org.apache.stormcrawler.bolt.SimpleFetcherBolt}.
+ *
+ * <p>For internal use only. Not part of StormCrawler's public API.
*/
-final class ProtocolMetrics {
+public final class ProtocolMetrics {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolMetrics.class);
@@ -43,7 +46,7 @@ final class ProtocolMetrics {
* prefix can come from the fetched server rather than from the protocol itself. A value which
* is not a number is skipped: it must not fail a fetch which otherwise completed.
*/
- static void update(ScopedReducedMetric averagedMetrics, Metadata metadata) {
+ public static void update(ScopedReducedMetric averagedMetrics, Metadata metadata) {
for (String key : metadata.keySet(METRICS_PREFIX)) {
final String value = metadata.getFirstValue(key);
final long parsed;
diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/RobotRulesLookup.java b/core/src/main/java/org/apache/stormcrawler/fetcher/RobotRulesLookup.java
similarity index 88%
rename from core/src/main/java/org/apache/stormcrawler/bolt/RobotRulesLookup.java
rename to core/src/main/java/org/apache/stormcrawler/fetcher/RobotRulesLookup.java
index 7054107c..348434c1 100644
--- a/core/src/main/java/org/apache/stormcrawler/bolt/RobotRulesLookup.java
+++ b/core/src/main/java/org/apache/stormcrawler/fetcher/RobotRulesLookup.java
@@ -15,10 +15,11 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import crawlercommons.robots.BaseRobotRules;
import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.bolt.FetcherBolt;
import org.apache.stormcrawler.metrics.ScopedCounter;
import org.apache.stormcrawler.protocol.FetchTimeoutException;
import org.apache.stormcrawler.protocol.Protocol;
@@ -31,8 +32,10 @@ import org.slf4j.LoggerFactory;
* Looks up the robots.txt rules of a URL under fetcher.thread.timeout, for the fetcher threads of
* {@link FetcherBolt}. A lookup which times out yields empty rules. Each lookup which returns rules
* counts in robots.fetched or robots.fromCache, each timeout also in robots.timeout.
+ *
+ * <p>For internal use only. Not part of StormCrawler's public API.
*/
-final class RobotRulesLookup {
+public final class RobotRulesLookup {
// the bolt's category: log configurations for FetcherBolt keep covering these lines
private static final Logger LOG = LoggerFactory.getLogger(FetcherBolt.class);
@@ -42,13 +45,13 @@ final class RobotRulesLookup {
* convention, cached rules carry no fetched content lengths. The empty rules of a timed-out
* lookup are not from the cache.
*/
- record Result(BaseRobotRules rules, boolean fromCache) {}
+ public record Result(BaseRobotRules rules, boolean fromCache) {}
private final FetchTimeoutHelpers helpers;
private final ScopedCounter eventCounter;
private final int taskId;
- RobotRulesLookup(FetchTimeoutHelpers helpers, ScopedCounter eventCounter, int taskId) {
+ public RobotRulesLookup(FetchTimeoutHelpers helpers, ScopedCounter eventCounter, int taskId) {
this.helpers = helpers;
this.eventCounter = eventCounter;
this.taskId = taskId;
@@ -62,7 +65,7 @@ final class RobotRulesLookup {
* @throws FetchTimeoutHelpers.SaturatedException when every helper thread is busy
* @throws Exception the protocol's own exception
*/
- Result lookup(Protocol protocol, String url, Metadata metadata) throws Exception {
+ public Result lookup(Protocol protocol, String url, Metadata metadata) throws Exception {
BaseRobotRules rules;
try {
rules = helpers.call(() -> protocol.getRobotRules(url), protocol, url, metadata);
diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/CrawlDelayPolicyTest.java b/core/src/test/java/org/apache/stormcrawler/fetcher/CrawlDelayPolicyTest.java
similarity index 95%
rename from core/src/test/java/org/apache/stormcrawler/bolt/CrawlDelayPolicyTest.java
rename to core/src/test/java/org/apache/stormcrawler/fetcher/CrawlDelayPolicyTest.java
index b7d94d76..4f8598ce 100644
--- a/core/src/test/java/org/apache/stormcrawler/bolt/CrawlDelayPolicyTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/fetcher/CrawlDelayPolicyTest.java
@@ -15,12 +15,12 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import org.apache.stormcrawler.bolt.CrawlDelayPolicy.Action;
-import org.apache.stormcrawler.bolt.CrawlDelayPolicy.Decision;
+import org.apache.stormcrawler.fetcher.CrawlDelayPolicy.Action;
+import org.apache.stormcrawler.fetcher.CrawlDelayPolicy.Decision;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java b/core/src/test/java/org/apache/stormcrawler/fetcher/FetchItemQueuesTest.java
similarity index 96%
rename from core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java
rename to core/src/test/java/org/apache/stormcrawler/fetcher/FetchItemQueuesTest.java
index 7bbeed33..1e486b45 100644
--- a/core/src/test/java/org/apache/stormcrawler/bolt/FetchItemQueuesTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/fetcher/FetchItemQueuesTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -33,7 +33,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.storm.Config;
import org.apache.storm.tuple.Tuple;
import org.apache.stormcrawler.Metadata;
-import org.apache.stormcrawler.bolt.FetchItemQueues.FetchItemQueue;
+import org.apache.stormcrawler.fetcher.FetchItemQueues.FetchItemQueue;
import org.apache.stormcrawler.util.URLUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -88,7 +88,7 @@ class FetchItemQueuesTest {
FetchItem first = q.getFetchItem();
Assertions.assertNotNull(first);
- Assertions.assertEquals("http://a.net/1", first.url);
+ Assertions.assertEquals("http://a.net/1", first.url());
Assertions.assertEquals(1, q.inQueues.get());
// one thread per host: nothing else from a.net while the fetch is in progress
Assertions.assertNull(q.getFetchItem());
@@ -98,7 +98,7 @@ class FetchItemQueuesTest {
Assertions.assertNull(q.getFetchItem());
FetchItem second = awaitItem(q, 2000);
Assertions.assertNotNull(second);
- Assertions.assertEquals("http://a.net/2", second.url);
+ Assertions.assertEquals("http://a.net/2", second.url());
Assertions.assertEquals(0, q.inQueues.get());
}
@@ -111,7 +111,7 @@ class FetchItemQueuesTest {
q.finishFetchItem(first, true);
FetchItem second = q.getFetchItem();
Assertions.assertNotNull(second);
- Assertions.assertEquals("http://a.net/2", second.url);
+ Assertions.assertEquals("http://a.net/2", second.url());
}
@Test
@@ -124,7 +124,7 @@ class FetchItemQueuesTest {
for (int i = 0; i < 3; i++) {
FetchItem it = q.getFetchItem();
Assertions.assertNotNull(it);
- got.add(it.queueId);
+ got.add(it.queueId());
}
Assertions.assertEquals(Set.of("a.net", "b.net", "c.net"), got);
Assertions.assertNull(q.getFetchItem());
@@ -219,7 +219,7 @@ class FetchItemQueuesTest {
Thread.yield();
continue;
}
- if (!seen.add(it.url)) {
+ if (!seen.add(it.url())) {
failed.set(true);
}
fetched.incrementAndGet();
@@ -286,7 +286,7 @@ class FetchItemQueuesTest {
// whoever polls next must get the second URL
FetchItem second = polled[0] != null ? polled[0] : awaitItem(q, 2000);
Assertions.assertNotNull(second, "second URL lost: no ticket left for the queue");
- Assertions.assertEquals("http://a.net/2", second.url);
+ Assertions.assertEquals("http://a.net/2", second.url());
}
/**
@@ -337,7 +337,7 @@ class FetchItemQueuesTest {
poller.join();
Assertions.assertNotNull(polled[0]);
- Assertions.assertEquals("http://a.net/2", polled[0].url);
+ Assertions.assertEquals("http://a.net/2", polled[0].url());
Assertions.assertEquals(1, hooked.getInProgressSize());
// the dispatched item finishes on its own queue, which is then reaped
q.finishFetchItem(polled[0], true);
@@ -472,7 +472,7 @@ class FetchItemQueuesTest {
add(q, "http://b.net/1");
FetchItem a1 = q.getFetchItem();
Assertions.assertNotNull(a1);
- Assertions.assertEquals("http://a.net/1", a1.url);
+ Assertions.assertEquals("http://a.net/1", a1.url());
// a second URL for a.net issues a ticket at the queue's current (past) next fetch time
add(q, "http://a.net/2");
// finishing with the 10 s delay pushes a.net's next fetch time into the future: the
@@ -481,7 +481,7 @@ class FetchItemQueuesTest {
FetchItem next = q.getFetchItem();
Assertions.assertNotNull(next, "returned nothing while b.net was ready");
- Assertions.assertEquals("http://b.net/1", next.url);
+ Assertions.assertEquals("http://b.net/1", next.url());
}
/** Asserts that the delay is the equal-jittered value: within [computed / 2, computed]. */
@@ -505,7 +505,7 @@ class FetchItemQueuesTest {
}
FetchItem it = q.getFetchItem();
Assertions.assertNotNull(it);
- FetchItemQueue fiq = q.queues.get(it.queueId);
+ FetchItemQueue fiq = q.queues.get(it.queueId());
long[] expected = {2000, 4000, 5000, 5000};
for (long computed : expected) {
@@ -540,7 +540,7 @@ class FetchItemQueuesTest {
}
FetchItem it = q.getFetchItem();
Assertions.assertNotNull(it);
- FetchItemQueue fiq = q.queues.get(it.queueId);
+ FetchItemQueue fiq = q.queues.get(it.queueId());
long[] expected = {8000, 16000, 30000, 30000};
for (long computed : expected) {
@@ -558,7 +558,7 @@ class FetchItemQueuesTest {
}
FetchItem it = q.getFetchItem();
Assertions.assertNotNull(it);
- FetchItemQueue fiq = q.queues.get(it.queueId);
+ FetchItemQueue fiq = q.queues.get(it.queueId());
for (int i = 0; i < 3; i++) {
assertJittered(10000, q.backOffFetchItem(it));
diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchOutcomesTest.java b/core/src/test/java/org/apache/stormcrawler/fetcher/FetchOutcomesTest.java
similarity index 99%
rename from core/src/test/java/org/apache/stormcrawler/bolt/FetchOutcomesTest.java
rename to core/src/test/java/org/apache/stormcrawler/fetcher/FetchOutcomesTest.java
index c019dcba..cf39a5a9 100644
--- a/core/src/test/java/org/apache/stormcrawler/bolt/FetchOutcomesTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/fetcher/FetchOutcomesTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -32,6 +32,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.stormcrawler.Constants;
import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.bolt.SiteMapParserBolt;
import org.apache.stormcrawler.metrics.ScopedCounter;
import org.apache.stormcrawler.persistence.Status;
import org.apache.stormcrawler.protocol.AbstractHttpProtocol;
diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java b/core/src/test/java/org/apache/stormcrawler/fetcher/FetchTimeoutHelpersTest.java
similarity index 99%
rename from core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java
rename to core/src/test/java/org/apache/stormcrawler/fetcher/FetchTimeoutHelpersTest.java
index 55a5317b..7760f045 100644
--- a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/fetcher/FetchTimeoutHelpersTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import java.io.IOException;
import java.util.HashMap;
diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/ProtocolMetricsTest.java b/core/src/test/java/org/apache/stormcrawler/fetcher/ProtocolMetricsTest.java
similarity index 97%
rename from core/src/test/java/org/apache/stormcrawler/bolt/ProtocolMetricsTest.java
rename to core/src/test/java/org/apache/stormcrawler/fetcher/ProtocolMetricsTest.java
index 12e7d8c6..35807898 100644
--- a/core/src/test/java/org/apache/stormcrawler/bolt/ProtocolMetricsTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/fetcher/ProtocolMetricsTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import static org.junit.jupiter.api.Assertions.assertEquals;
diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/RobotRulesLookupTest.java b/core/src/test/java/org/apache/stormcrawler/fetcher/RobotRulesLookupTest.java
similarity index 99%
rename from core/src/test/java/org/apache/stormcrawler/bolt/RobotRulesLookupTest.java
rename to core/src/test/java/org/apache/stormcrawler/fetcher/RobotRulesLookupTest.java
index 512140a5..09bb3890 100644
--- a/core/src/test/java/org/apache/stormcrawler/bolt/RobotRulesLookupTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/fetcher/RobotRulesLookupTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.stormcrawler.bolt;
+package org.apache.stormcrawler.fetcher;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;