From f1a3e6a86ddeefc5ae30f31643aadedae7edcd30 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Wed, 30 Sep 2026 16:41:03 +0200 Subject: [PATCH] #2057 FetcherBolt: move the fetcher internals to o.a.s.fetcher Signed-off-by: Davide Polato --- .../apache/stormcrawler/bolt/FetcherBolt.java | 113 +++++++++--------- .../stormcrawler/bolt/SimpleFetcherBolt.java | 2 + .../{bolt => fetcher}/CrawlDelayPolicy.java | 15 ++- .../{bolt => fetcher}/FetchItem.java | 38 ++++-- .../{bolt => fetcher}/FetchItemQueues.java | 81 ++++++++++--- .../{bolt => fetcher}/FetchOutcomes.java | 17 ++- .../FetchTimeoutHelpers.java | 21 ++-- .../{bolt => fetcher}/ProtocolMetrics.java | 11 +- .../{bolt => fetcher}/RobotRulesLookup.java | 13 +- .../CrawlDelayPolicyTest.java | 6 +- .../FetchItemQueuesTest.java | 28 ++--- .../{bolt => fetcher}/FetchOutcomesTest.java | 3 +- .../FetchTimeoutHelpersTest.java | 2 +- .../ProtocolMetricsTest.java | 2 +- .../RobotRulesLookupTest.java | 2 +- 15 files changed, 221 insertions(+), 133 deletions(-) rename core/src/main/java/org/apache/stormcrawler/{bolt => fetcher}/CrawlDelayPolicy.java (91%) rename core/src/main/java/org/apache/stormcrawler/{bolt => fetcher}/FetchItem.java (81%) rename core/src/main/java/org/apache/stormcrawler/{bolt => fetcher}/FetchItemQueues.java (89%) rename core/src/main/java/org/apache/stormcrawler/{bolt => fetcher}/FetchOutcomes.java (94%) rename core/src/main/java/org/apache/stormcrawler/{bolt => fetcher}/FetchTimeoutHelpers.java (93%) rename core/src/main/java/org/apache/stormcrawler/{bolt => fetcher}/ProtocolMetrics.java (86%) rename core/src/main/java/org/apache/stormcrawler/{bolt => fetcher}/RobotRulesLookup.java (88%) rename core/src/test/java/org/apache/stormcrawler/{bolt => fetcher}/CrawlDelayPolicyTest.java (95%) rename core/src/test/java/org/apache/stormcrawler/{bolt => fetcher}/FetchItemQueuesTest.java (96%) rename core/src/test/java/org/apache/stormcrawler/{bolt => fetcher}/FetchOutcomesTest.java (99%) rename core/src/test/java/org/apache/stormcrawler/{bolt => fetcher}/FetchTimeoutHelpersTest.java (99%) rename core/src/test/java/org/apache/stormcrawler/{bolt => fetcher}/ProtocolMetricsTest.java (97%) rename core/src/test/java/org/apache/stormcrawler/{bolt => fetcher}/RobotRulesLookupTest.java (99%) 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 61ca63b7c..25aa5e0b2 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.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.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 void run() { activeThreads.incrementAndGet(); // count threads - beingFetched[threadNum] = fit.url; + beingFetched[threadNum] = fit.url(); LOG.debug( "[Fetcher #{}] {} => activeThreads={}, spinWaiting={}, queueID={}", @@ -160,14 +166,14 @@ public void run() { 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 void run() { 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 void run() { } 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 void run() { 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 void run() { LOG.info( "[Fetcher #{}] Fetched {} with status {} in msec {}", taskId, - fit.url, + fit.url(), response.getStatusCode(), timeFetching); @@ -296,8 +304,8 @@ public void run() { // 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 void run() { && allowRedirs() && outcome.redirectTarget() != null) { emitOutlink( - fit.tuple, + fit.tuple(), url, outcome.redirectTarget(), outcome.metadata(), @@ -313,8 +321,8 @@ && allowRedirs() } 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 @@ && allowRedirs() 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 void prepare( 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 void prepare( 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 void execute(Tuple input) { } 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 void execute(Tuple input) { "[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 void execute(Tuple input) { } /** - * 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 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 968faae02..bdd03eb2f 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.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 448b311f9..c41973e50 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 @@ * 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. + * + *

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 @@ enum Action { * @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 @@ record Decision(Action action, long delay, String robotsCrawlDelaySecs) { // specifies a shorter crawl-delay private final boolean serverDelayForce; - CrawlDelayPolicy( + public CrawlDelayPolicy( long maxCrawlDelay, boolean maxCrawlDelayForce, long serverDelay, @@ -78,7 +81,7 @@ record Decision(Action action, long delay, String robotsCrawlDelaySecs) { * @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 7ce1955d6..16b6ce36d 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.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. + * + *

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 @@ private FetchItem(String url, Tuple t, String queueId) { * Create an item. Queue id will be created based on queueMode 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 @@ public static FetchItem create(URL u, String url, Tuple t, String queueMode) { 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 dc28c8e44..b64b4f6bf 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 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 @@ * 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. + * + *

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 @@ public boolean addFetchItem(URL u, String url, Tuple input) { 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 @@ public boolean addFetchItem(URL u, String url, Tuple input) { } 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 @@ public void finishFetchItem(FetchItem it, boolean asap) { * @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 @@ public FetchItem getFetchItem() { 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 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 queue = new ConcurrentLinkedQueue<>(); final String id; @@ -372,7 +407,7 @@ static class FetchItemQueue { /** 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 @@ public FetchItemQueue( 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 3c0b61924..46b52d3fe 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.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 @@ * {@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. + * + *

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 @@ record Outcome( 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 @@ record Outcome( * @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 @@ Outcome ofResponse( * @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 27dc6df87..2589e9881 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 @@ * *

The deadline is the one of {@link FetchTimeout#secs(Map)}, clamped to the message timeout on * both paths. + * + *

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 @@ boolean abandonedByCaller() { * @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) { + public FetchTimeoutHelpers( + Map conf, int defaultMaxHelpers, String threadNamePrefix) { this.timeoutSecs = FetchTimeout.secs(conf); final int maxHelpers = ConfUtils.getInt( @@ -132,7 +135,7 @@ boolean abandonedByCaller() { } /** Registers the {@code fetchhelpers} gauge: number of helpers busy with a call. */ - void registerMetrics(TopologyContext context, Map conf, int bucketSecs) { + public void registerMetrics(TopologyContext context, Map conf, int bucketSecs) { CrawlerMetrics.registerGauge(context, conf, "fetchhelpers", this::busy, bucketSecs); } @@ -152,12 +155,12 @@ long timeoutSecs() { } /** 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 @@ int largestPoolSize() { * @throws FetchTimeoutException when the deadline passed * @throws Exception the protocol's own exception */ - T call(Callable call, Protocol protocol, String url, Metadata metadata) + public T call(Callable call, Protocol protocol, String url, Metadata metadata) throws Exception { if (timeoutSecs <= 0 || protocol.supportsFetchTimeout(url, metadata)) { return call.call(); @@ -223,7 +226,7 @@ T call(Callable call, Protocol protocol, String url, Metadata metadata) } } - 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 b9588d968..a32480b07 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 @@ /** * 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}. + * + *

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 @@ private 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 7054107c2..348434c18 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 @@ * 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. + * + *

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 @@ record Result(BaseRobotRules rules, boolean fromCache) {} * @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 b7d94d761..4f8598cec 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 7bbeed337..1e486b45e 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 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 @@ void itemIsReturnedOnceAndHostWaitsForCrawlDelay() throws Exception { 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 @@ void itemIsReturnedOnceAndHostWaitsForCrawlDelay() throws Exception { 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 @@ void finishingAsapMakesHostImmediatelyAvailable() throws Exception { 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 @@ void differentHostsAreServedBackToBack() throws Exception { 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 @@ void concurrentProducersAndConsumersLoseNothing() throws Exception { Thread.yield(); continue; } - if (!seen.add(it.url)) { + if (!seen.add(it.url())) { failed.set(true); } fetched.incrementAndGet(); @@ -286,7 +286,7 @@ boolean hasFreeSlot() { // 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 @@ void afterDequeue() { 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 @@ void staleTicketDoesNotHideAnotherReadyQueue() throws Exception { 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 @@ void staleTicketDoesNotHideAnotherReadyQueue() throws Exception { 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 @@ void saturatedFetchBacksTheQueueOffExponentially() throws Exception { } 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 @@ void saturatedFetchBackoffWithUnboundedMaxCrawlDelay() throws Exception { } 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 @@ void saturatedFetchBackoffNeverBelowTheQueueDelay() throws Exception { } 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 c019dcba6..cf39a5a9e 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.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 55a5317b8..7760f0459 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 12e7d8c69..35807898c 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 512140a5d..09bb3890d 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;