abhinav-phi commented on code in PR #2117:
URL: https://github.com/apache/stormcrawler/pull/2117#discussion_r3951961605
##########
external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java:
##########
@@ -253,10 +403,168 @@ public void onNext(final
crawlercommons.urlfrontier.Urlfrontier.AckMessage confi
return;
}
+ completeTuples(url, values, confirmation.getStatus());
+ }
+
+ /**
+ * Acknowledges a whole batch: one status per URL, in the order the batch
was sent.
+ *
+ * @param confirmation the BatchAck received from the PutDiscovered
endpoint
+ */
+ private void onNext(final BatchAck confirmation) {
+ if (closed) {
+ return;
+ }
+
+ final List<URLItem> items;
+ synchronized (batchLock) {
+ items = pendingBatches.remove(confirmation.getID());
+ }
+
+ if (items == null) {
+ LOG.debug("Could not find batch with ID `{}`.",
confirmation.getID());
+ return;
+ }
+
+ final List<AckMessage.Status> statuses =
confirmation.getStatusesList();
+ if (statuses.size() != items.size()) {
+ LOG.warn(
+ "BatchAck {} carries {} status(es) for {} URL(s).",
+ confirmation.getID(),
+ statuses.size(),
+ items.size());
+ }
+
+ // URLs without a status, e.g. on a protocol breach, are left to the
waitAck eviction
+ int numStatuses = Math.min(statuses.size(), items.size());
+ for (int i = 0; i < numStatuses; i++) {
+ final String url = items.get(i).getID();
+ final List<Tuple> values = detachWaitAck(url);
+ if (values == null) {
+ LOG.debug("Could not find unacked tuple for id `{}`.", url);
+ continue;
+ }
+ completeTuples(url, values, statuses.get(i));
+ }
+ }
+
+ private void onBatchError(final Throwable t) {
+ if (closed) {
+ return;
+ }
+
+ // a frontier older than 2.6 does not know the PutDiscovered endpoint;
instead of
+ // dropping the discovered URLs, send them individually on the
streaming endpoint
+ // and keep doing so for the rest of the bolt's life
+ if (t instanceof StatusRuntimeException
+ && ((StatusRuntimeException) t).getStatus().getCode()
+ == io.grpc.Status.Code.UNIMPLEMENTED) {
+ LOG.warn(
+ "The frontier does not implement PutDiscovered
(URLFrontier < 2.6) - sending discovered URLs individually on the streaming
endpoint.");
+ disableBatchingAndResend();
+ return;
+ }
+
+ LOG.error("Error received on the batch stream: {}", t.getMessage());
+ LOG.debug("Error received on the batch stream", t);
+
+ // the stream is dead: forget the batches it carried, their tuples are
failed by the
+ // waitAck cache eviction and replayed by Storm. A new stream is
opened on the next flush.
+ synchronized (batchLock) {
+ pendingBatches.clear();
+ batchRequestObserver = null;
+ batchTransport = null;
+ }
+ synchronized (flow) {
+ flow.notifyAll();
+ }
+ }
+
+ /** Stops batching and pushes everything buffered or in flight through the
streaming endpoint. */
+ private void disableBatchingAndResend() {
+ final List<URLItem> toResend = new ArrayList<>();
+ synchronized (batchLock) {
+ batching = false;
+ for (List<URLItem> items : pendingBatches.values()) {
+ toResend.addAll(items);
+ }
+ pendingBatches.clear();
Review Comment:
Fixed in 41ea0ba5 as suggested: the batches still tracked in
`pendingBatches` are drained under `batchLock` and their tuples failed
immediately, consistent with the RuntimeException path in `flushBatch`. Nothing
depends on the `waitAck` cache expiry anymore.
##########
external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java:
##########
@@ -294,43 +615,55 @@ public void store(
// First get processing permit. Otherwise, starvation possible.
var hasPermit = false;
var timeSpent = 0L;
+ boolean throttled = false;
while (!hasPermit) {
- try {
- hasPermit = inFlightSemaphore.tryAcquire(throttleTimeMS,
TimeUnit.MILLISECONDS);
- if (!hasPermit) {
- LOG.trace(
- "{} messages in flight, time spent throttling {}",
- inFlightSemaphore.getQueueLength(),
- timeSpent);
-
eventCounter.scope("timeSpentThrottling").incrBy(throttleTimeMS);
- timeSpent += throttleTimeMS;
- if (timeSpent >= 30000L) {
+ hasPermit = inFlightSemaphore.tryAcquire();
+ if (!hasPermit) {
+ throttled = true;
+ LOG.trace(
+ "{} messages in flight, time spent throttling {}",
+ inFlightSemaphore.getQueueLength(),
+ timeSpent);
+ // wait for room on the monitor: woken as soon as an ack
releases permits or the
+ // transport becomes ready again. The timeout is a backstop,
not a poll interval.
+ synchronized (flow) {
+ try {
+ flow.wait(throttleTimeMS);
+ } catch (InterruptedException e) {
LOG.warn(
- "Waiting more than {} ms for processing. There
are {} permits available for {} waiting threads.",
- timeSpent,
- inFlightSemaphore.availablePermits(),
+ "InterruptedException - (approx.) {} messages
in flight.",
inFlightSemaphore.getQueueLength());
- }
- // To prevent a deadlock, it is necessary to periodically
clean up the waitAck
- // cache. Otherwise, in case of a frontier-side or
connection-wise error, all
- // incoming URLs will after some time be all caught up in
this loop without
- // touching the cache, possibly leading to no eviction and
thus leading to no
- // release of inFlightSemaphore permits.
- waitAckLock.lock();
- try {
- waitAck.cleanUp();
- } finally {
- waitAckLock.unlock();
+ Thread.currentThread().interrupt();
}
}
- } catch (InterruptedException e) {
- LOG.warn(
- "InterruptedException - (approx.) {} messages in
flight.",
- inFlightSemaphore.getQueueLength());
- Thread.currentThread().interrupt();
+
eventCounter.scope("timeSpentThrottling").incrBy(throttleTimeMS);
Review Comment:
Fixed in 41ea0ba5: the wait is bracketed with `System.nanoTime()` and the
metric now records the time actually spent in `flow.wait` - an ack waking the
wait early no longer bills the full throttle interval. The 30-second warn
threshold uses the measured sum as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
