#2129 FetcherBolt: lock-free fetch queues ordered by next fetch time

[GGraziadei]

Fixes #2129.

FetchItemQueues used a single monitor for adding, taking and finishing items, and getFetchItem() rotated a LinkedHashMap linearly over every queue whose crawl delay had not elapsed. With many hosts the fetcher threads held that lock most of the time and the executor thread calling execute() stalled on every incoming tuple.

Change

• Queues live in a ConcurrentHashMap; the ones that may have an item ready are referenced from a DelayQueue of tickets ordered by their next fetch time. Taking an item is O(log n) and adding never waits for the fetcher threads.
• Per-queue state (size bound, in-progress count, next fetch time, crawl delays) uses atomics; a per-queue monitor is used only for the add/reap race. Empty queues are removed from the map as soon as they drain.
• FetchItem is built before any lock is taken, so a DNS lookup in byIP mode no longer blocks the fetcher threads.

Behaviour preserved

Politeness per queue, asap release, fetcher.threads.per.queue and fetcher.maxThreads.*, fetcher.max.queue.size, crawl delay overrides from metadata and robots.txt, queue modes, metrics and the debug dump.

Behaviour change: order in which hosts are served

The old implementation served hosts round-robin in LinkedHashMap insertion order, rotating past the ones whose crawl delay had not elapsed. The new one serves them in strict nextFetchTime order: the host that became ready the longest time ago goes first, and hosts whose next fetch time falls in the same millisecond are ordered arbitrarily (DelayQueue does not guarantee FIFO among equal delays).

In practice:

• A host is never starved: once its delay has elapsed it is ahead of every host that became ready later.
• Newly added hosts are ready immediately, so under a steady stream of new hosts they are served before hosts still waiting on their delay, exactly as before.
• Operators who relied on the insertion-order rotation to spread fetches evenly across a fixed set of hosts will see a different, but not less fair, interleaving. Per-host rate is unchanged: it is still bounded by the crawl delay.

Two smaller observable details, both stated in the javadoc: getQueueSize() can transiently over-report by one while an offer is being rejected, and the debug dump produced by the trigger file is no longer a point-in-time snapshot but a smear over the time it takes to iterate.

Tests

New FetchItemQueuesTest: politeness, asap, several hosts back to back, queue size bound, multiple threads per queue, crawl delay from metadata, reaping of empty queues, and a concurrent producers/consumers test checking that nothing is lost or duplicated. Existing FetcherBoltTest and SimpleFetcherBoltTest pass unchanged.

Benchmark: the case this targets

The cost the change removes is the linear rotation over queues that are not ready. It shows when a fetcher instance holds many hosts and most of them are waiting on their crawl delay, which is the steady state of a broad crawl: every getFetchItem() walked all of them under the global monitor, and the executor thread calling execute() queued behind the fetcher threads for that monitor.

Micro-benchmark on FetchItemQueues alone (main vs this branch): 50 fetcher threads, 1 producer thread emulating the executor, 20 URLs per host, fetcher.server.delay 1s so that almost every queue is waiting, instant fetch, 8s per run, p99 from sampled calls.

hosts	metric	before	after
20000	getFetchItem avg	1.58 ms	24 us
20000	getFetchItem p99	5.0 ms	24 us
20000	addFetchItem avg	14.1 ms	24 us
20000	addFetchItem p99	458 ms	0.11 ms
2000	addFetchItem p99	2.5 ms	0.13 ms
20000	fetched/s	20001	20000

Throughput is unchanged, as expected: it is bounded by politeness. What goes away is the executor stall and the CPU burnt scanning queues that are not ready. With fetcher.server.delay 0 (every queue always ready, no rotation) both versions do ~53k fetch/s, i.e. the new structure is not slower on the path where the old one was cheap.

With a few hundred hosts per fetcher instance the difference is not measurable; with tens of thousands it is the numbers above.

Verification

• FetcherBoltTest (8) and SimpleFetcherBoltTest (8) pass unchanged; FetchItemQueuesTest adds 9 tests including a concurrent producers/consumers run and a deterministic reproduction of the lost-wakeup race fixed in the second commit.
• Full mvn clean verify on the whole project: 675 tests, 0 failures, 0 errors (Docker-backed modules included).
• CI on this PR is green.

---

For all changes

• Is there a issue associated with this PR? Is it referenced in the commit message?
• Does your PR title start with #XXXX where XXXX is the issue number you are trying to resolve?
• Has your PR been rebased against the latest commit within the target branch (typically main)?
• Is your initial contribution a single, squashed commit?
• Is the code properly formatted with mvn git-code-format:format-code -Dgcf.globPattern="**/*" -Dskip.format.code=false?

For code changes

• Have you ensured that the full suite of tests is executed via mvn clean verify?
• Have you written or updated unit tests to verify your changes?
• If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0? (no new dependencies)
• If applicable, have you updated the LICENSE file, including the main LICENSE file? (not applicable)
• If applicable, have you updated the NOTICE file, including the main NOTICE file? (not applicable)

[rzo1]

Currently I am short on time, so first round AI-based review)

The design holds up: one ticket per queue guarded by the scheduled CAS, an immutable ordering key in QueueTicket, and FetchItem.create (which does DNS in byIP mode) moved out of the critical section.

One lost-wakeup window, inline on line 538.

Other comments:

• Fairness changes from LinkedHashMap round-robin to strict nextFetchTime order, with arbitrary tie-breaking among queues sharing a millisecond. This is a component operators tune by feel, so please put the change in the description.
• offer increments size before the bound check and decrements on overflow, so getQueueSize() can transiently over-report. Harmless, but worth a comment.
• logQueuesContent is no longer synchronized. Safe with weakly consistent iterators, but the dump is now a smear rather than a snapshot. Say so in its javadoc.
• This rewrites the hottest path in the crawler. Please include before/after numbers for the case it targets, and confirm the existing FetcherBolt integration tests pass.