Use batch endpoint in URLFrontier module for DISCOVERED urls

[abhinav-phi]

Summary

URLFrontier 2.6 added a batched endpoint for ingesting discovered URLs, PutDiscovered(stream DiscoveredBatch) returns (stream BatchAck). The per-message cost of the streaming PutURLs endpoint is what caps the ingestion rate, and discovered URLs — the outlinks of the pages being parsed — are the bulk of what a crawl writes. This PR routes them through the batched endpoint, grouping up to urlfrontier.batch.size URLs (default 100) into one message, while known URLs (fetched, redirections, errors...) keep using the streaming mode, as suggested in #2062. The control-flow mechanism follows the best practices of the client shipped with URLFrontier (PutURLs.java).

Fixes #2062

What the bolt does now

StatusUpdaterBolt keeps its overall structure: a waitAck cache holds the tuples per URL until the frontier acks them, and a semaphore bounds the messages in flight. What changes is the path each URL takes:

Discovered URLs → PutDiscovered. A DISCOVERED URL is appended to a buffer instead of being sent on its own. The buffer leaves as one DiscoveredBatch message when:

• it reaches urlfrontier.batch.size entries;
• a known URL arrives — the outlinks buffered so far belong to the page whose status is now being updated, a natural end-of-page boundary for the batch (the release notes for 2.6 explicitly call the outlinks of a page a natural batch);
• a partially filled batch has been open for more than a second (checked by a scheduled flusher every 100 ms), so that acks are not delayed when the crawl tails off;
• the bolt is throttled because the semaphore is exhausted — pushing the batch out frees the permits sooner.

Each batch carries a unique ID (batch-N); the frontier echoes it in the BatchAck with one status per URL, in the order they were sent, so every URL can be resolved against waitAck and its tuples acked (or failed on FAIL) exactly as the streaming path did. Permits are released in bulk per URL, unchanged.

Known URLs → PutURLs, one message per URL, untouched behaviour.

Flow control without polling. store() used to busy-poll tryAcquire with a sleep. It now waits on a monitor (flow) and is woken by notifyAll() from the places that change its conditions: permit releases (acks, evictions), transport on-ready notifications, and stream errors. The wait timeout (urlfrontier.throttling.time.msec, unchanged, 10 ms) is a backstop that also drives the periodic waitAck.cleanUp() which prevents the deadlock the old code guarded against. This mirrors the isReady()/on-ready pattern of the reference client instead of polling.

Fallback for older frontiers. If the frontier does not implement PutDiscovered (i.e. predates 2.6), the stream fails with UNIMPLEMENTED. The bolt detects this, switches batching off permanently for its lifetime, and re-sends whatever was buffered or still in flight individually on the streaming endpoint — no URLs are lost, and a pre-2.6 frontier behaves exactly as before this PR. A dead batch stream (any other error) drops the pending batch IDs and opens a fresh stream on the next flush; the affected tuples are failed locally so Storm replays them.

Configuration

New key, documented in the module README:

# max number of discovered URLs per PutDiscovered message (default 100, 0 disables batching)
urlfrontier.batch.size: 100

The module now depends on urlfrontier-API 2.6 (bumped from 2.5). The schema stays wire-compatible with 2.5 apart from the semantics noted in the 2.6 release notes, and the fallback covers servers that do not implement the new RPC, so no frontier upgrade is strictly required — though the batching only pays off against a 2.6 server.

Tests

• StatusUpdaterBoltTest.acknowledgesDiscoveredURLsSentInBatches — 6 discovered URLs against a 2.6 frontier with batch size 2: all acked, ≥ 3 batches sent, nothing failed;
• StatusUpdaterBoltTest.sendsDiscoveredURLsIndividuallyWhenBatchingDisabled — urlfrontier.batch.size: 0 sends individually and never opens a batch;
• StatusUpdaterBoltTest.acksKnownURLsThroughStreamingEndpoint — known URLs still ack through the streaming endpoint;
• StatusUpdaterBoltFallbackTest — runs the bolt against a crawlercommons/url-frontier:2.5 container: the UNIMPLEMENTED error is caught, batching is switched off, and discovered URLs are acked through the streaming fallback;
• the existing StatusUpdaterBoltTest cases (ack with metadata, queue-stream emission, semaphore recovery after a frontier restart) pass unchanged on the new implementation.

All 64 module tests pass locally against real frontier containers (testcontainers), checkstyle, forbiddenapis and the google-java-format check are clean. The StatusUpdaterBolt JaCoCo ratios configured for this module still pass.

Notes for reviewers

• Dependency availability. The url-frontier 2.6 release published its artifacts to Sonatype Central via a workflow that completed successfully on 2026-08-19, but the deployment is not yet visible on Maven Central (urlfrontier-API still lists 2.5 as latest; the portal deployment likely awaits the manual publish step, since central-publishing-maven-plugin defaults to autoPublish=false). This PR builds against the 2.6 API regardless — the protobuf/gRPC API is stable in the 2.6 tag — but CI will need urlfrontier-API:2.6 to be resolvable; it may be worth pinging the crawler-commons maintainers to complete the release on the portal. If 2.6 cannot be published soon, an alternative is to keep the dependency at 2.5 and build the batch messages from the proto (not possible: DiscoveredBatch/BatchAck and the putDiscovered stub only exist in 2.6), so the version bump is inherent to the feature.
• The waitAck-cache eviction semantics, the semaphore accounting, the queue-stream emission and the channel management (multi-address assignment, reconnect on TRANSIENT_FAILURE) are carried over unchanged from the previous implementation.
• The PutDiscovered stream is deliberately not re-created on channel TRANSIENT_FAILURE (unlike the PutURLs stream): with waitForReady it survives connection blips, and abandoning it would orphan the batches already sent on it.