abhinav-phi commented on code in PR #2117:
URL: https://github.com/apache/stormcrawler/pull/2117#discussion_r3951960717
##########
external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java:
##########
@@ -418,7 +751,142 @@ public void store(
KnownURLItem.newBuilder().setInfo(info).setRefetchableFromDate(date).build());
}
- requestObserver.onNext(itemBuilder.setID(url).build());
+ final URLItem item = itemBuilder.setID(url).build();
+
+ // discovered URLs travel in batches on the PutDiscovered endpoint,
known URLs keep
+ // using the streaming endpoint
+ if (status.equals(Status.DISCOVERED)) {
+ boolean shouldBatch;
+ boolean flushNow = false;
+ synchronized (batchLock) {
+ // re-read inside the lock: batching can be disabled
concurrently by the
+ // fallback for frontiers without the PutDiscovered endpoint
+ shouldBatch = batching;
+ if (shouldBatch) {
+ if (batchBuffer.isEmpty()) {
+ oldestBufferedAt = System.currentTimeMillis();
+ }
+ batchBuffer.add(item);
+ flushNow = batchBuffer.size() >= batchSize;
+ }
+ }
+ if (!shouldBatch) {
+ sendOnStreamingEndpoint(item);
+ return;
+ }
+ if (flushNow) {
+ flushBatch();
+ }
+ return;
+ }
+
+ if (batching) {
+ // the outlinks buffered so far belong to the page whose status is
now updated:
+ // a natural boundary for the batch
+ flushBatch();
+ }
+
+ sendOnStreamingEndpoint(item);
+ }
+
+ /** Sends the buffered discovered URLs as one batch, if any. */
+ private void flushBatch() {
+ flushBatch(false);
+ }
+
+ /**
+ * Sends the buffered discovered URLs as one batch, if any.
+ *
+ * @param awaitTransport whether to wait briefly for the transport to
become ready before
+ * sending; only the flusher thread may do so, the Storm executor
thread never stalls
+ */
+ private void flushBatch(boolean awaitTransport) {
+ final List<URLItem> items;
+ final StreamObserver<DiscoveredBatch> stream;
+ synchronized (batchLock) {
+ if (!batching || batchBuffer.isEmpty()) {
+ return;
+ }
+ if (batchRequestObserver == null) {
+ // the previous stream died: open a new one
+ batchRequestObserver = newPutDiscoveredStream();
+ }
+ stream = batchRequestObserver;
+ items = new ArrayList<>(batchBuffer);
+ batchBuffer.clear();
+ oldestBufferedAt = 0;
+ }
+
+ final String batchID = "batch-" + batchSequences.incrementAndGet();
+ final DiscoveredBatch.Builder batchBuilder =
DiscoveredBatch.newBuilder().setID(batchID);
+ for (URLItem buffered : items) {
+ batchBuilder.addItems(buffered.getDiscovered().getInfo());
+ }
+ final DiscoveredBatch batch = batchBuilder.build();
+
+ // registered before the send so that a fast ack can never miss it
+ synchronized (batchLock) {
+ pendingBatches.put(batchID, items);
+ }
+
+ try {
+ if (awaitTransport) {
+ // follow the transport's lead: wait briefly for it to take
the batch without
+ // buffering it, woken by the on-ready handler. The timeout is
a backstop, not a
+ // poll interval.
+ final ClientCallStreamObserver<DiscoveredBatch> transport =
batchTransport;
+ if (transport != null && !transport.isReady()) {
+ synchronized (flow) {
+ flow.wait(BATCH_FLUSH_DELAY_MS);
+ }
+ }
+ }
+ synchronized (sendLock) {
+ stream.onNext(batch);
+ }
+ eventCounter.scope("batched").incrBy(items.size());
+ eventCounter.scope("batches").incrBy(1);
+ LOG.debug("Sent batch {} with {} discovered URL(s).", batchID,
items.size());
+ } catch (InterruptedException e) {
Review Comment:
Fixed in 41ea0ba5: the interrupt handler now removes the batch from
`pendingBatches` and fails its tuples locally so Storm replays them - exactly
as suggested. The batch had never been handed to the transport at that point,
so nothing that reached the frontier is dropped.
--
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]
