From d6eb68a26501366e0e60fc3a67c2d327085874b5 Mon Sep 17 00:00:00 2001 From: Adam Miller Date: Mon, 5 Apr 2021 16:01:27 -0700 Subject: [PATCH] Revert "Fix error log reporting of batch size for trough crawl logs" --- .../postprocessor/TroughCrawlLogFeed.java | 114 ++++++++---------- 1 file changed, 53 insertions(+), 61 deletions(-) diff --git a/contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java b/contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java index 732ef5d3..6d664b0b 100644 --- a/contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java +++ b/contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java @@ -89,7 +89,7 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle { protected static final Logger logger = Logger.getLogger(TroughCrawlLogFeed.class.getName()); - protected static final int BATCH_MAX_TIME_MS = 60 * 1000; + protected static final int BATCH_MAX_TIME_MS = 20 * 1000; protected static final int BATCH_MAX_SIZE = 400; protected KeyedProperties kp = new KeyedProperties(); @@ -252,75 +252,67 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle { } protected void postCrawledBatch() { - Object[] flattenedValues = null; - StringBuffer sqlTmpl = new StringBuffer(); - synchronized (crawledBatch) { - if (uncrawledBatch.size() >= BATCH_MAX_SIZE || System.currentTimeMillis() - uncrawledBatchLastTime > BATCH_MAX_TIME_MS) { - crawledBatchLastTime = System.currentTimeMillis(); - if (!crawledBatch.isEmpty()) { - logger.info("posting batch of " + crawledBatch.size() + " crawled urls trough segment " + getSegmentId()); - sqlTmpl.append("insert into crawled_url (" - + "timestamp, status_code, size, payload_size, url, hop_path, is_seed_redirect, " - + "via, mimetype, content_digest, seed, is_duplicate, warc_filename, " - + "warc_offset, warc_content_bytes, host) values " - + "(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"); - for (int i = 1; i < crawledBatch.size(); i++) { - sqlTmpl.append(", (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"); - } + logger.info("posting batch of " + crawledBatch.size() + " crawled urls trough segment " + getSegmentId()); + Object[] flattenedValues = null; + StringBuffer sqlTmpl = new StringBuffer(); + synchronized (crawledBatch) { + if (uncrawledBatch.size() >= BATCH_MAX_SIZE || System.currentTimeMillis() - uncrawledBatchLastTime > BATCH_MAX_TIME_MS) { + crawledBatchLastTime = System.currentTimeMillis(); + if (!crawledBatch.isEmpty()) { + sqlTmpl.append("insert into crawled_url (" + + "timestamp, status_code, size, payload_size, url, hop_path, is_seed_redirect, " + + "via, mimetype, content_digest, seed, is_duplicate, warc_filename, " + + "warc_offset, warc_content_bytes, host) values " + + "(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"); + for (int i = 1; i < crawledBatch.size(); i++) { + sqlTmpl.append(", (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"); + } - flattenedValues = new Object[16 * crawledBatch.size()]; - for (int i = 0; i < crawledBatch.size(); i++) { - System.arraycopy(crawledBatch.get(i), 0, flattenedValues, 16 * i, 16); - } - crawledBatch.clear(); - } - } - } - if(flattenedValues !=null && flattenedValues.length > 0) { - try { - synchronized (getSegmentId()) { //avoids 500 due to locked db from posting from uncrawled batch - troughClient().write(getSegmentId(), sqlTmpl.toString(), flattenedValues); - } - } catch (Exception e) { - logger.log(Level.WARNING, "problem posting batch of " + (flattenedValues.length/16) + " crawled urls to trough segment " + getSegmentId(), e); - } - crawledBatchLastTime = System.currentTimeMillis(); - } + flattenedValues = new Object[16 * crawledBatch.size()]; + for (int i = 0; i < crawledBatch.size(); i++) { + System.arraycopy(crawledBatch.get(i), 0, flattenedValues, 16 * i, 16); + } + crawledBatch.clear(); + } + } + } + if(flattenedValues !=null && flattenedValues.length > 0) { + try { + troughClient().write(getSegmentId(), sqlTmpl.toString(), flattenedValues); + } catch (Exception e) { + logger.log(Level.WARNING, "problem posting batch of " + flattenedValues.length + " crawled urls to trough segment " + getSegmentId(), e); + } + + crawledBatchLastTime = System.currentTimeMillis(); + } } protected void postUncrawledBatch() { - Object[] flattenedValues = null; - StringBuffer sqlTmpl = new StringBuffer(); + logger.info("posting batch of " + uncrawledBatch.size() + " uncrawled urls trough segment " + getSegmentId()); synchronized (uncrawledBatch) { - if (uncrawledBatch.size() >= BATCH_MAX_SIZE || System.currentTimeMillis() - uncrawledBatchLastTime > BATCH_MAX_TIME_MS) { - uncrawledBatchLastTime = System.currentTimeMillis(); - if (!uncrawledBatch.isEmpty()) { - logger.info("posting batch of " + uncrawledBatch.size() + " uncrawled urls trough segment " + getSegmentId()); - sqlTmpl.append( - "insert into uncrawled_url (timestamp, url, hop_path, status_code, via, seed, host)" - + " values (%s, %s, %s, %s, %s, %s, %s)"); + if (!uncrawledBatch.isEmpty()) { + StringBuffer sqlTmpl = new StringBuffer(); + sqlTmpl.append( + "insert into uncrawled_url (timestamp, url, hop_path, status_code, via, seed, host)" + + " values (%s, %s, %s, %s, %s, %s, %s)"); - for (int i = 1; i < uncrawledBatch.size(); i++) { - sqlTmpl.append(", (%s, %s, %s, %s, %s, %s, %s)"); - } - - flattenedValues = new Object[7 * uncrawledBatch.size()]; - for (int i = 0; i < uncrawledBatch.size(); i++) { - System.arraycopy(uncrawledBatch.get(i), 0, flattenedValues, 7 * i, 7); - } - uncrawledBatch.clear(); + for (int i = 1; i < uncrawledBatch.size(); i++) { + sqlTmpl.append(", (%s, %s, %s, %s, %s, %s, %s)"); } - } - } - if(flattenedValues !=null && flattenedValues.length > 0) { - try { - synchronized (getSegmentId()) { + + Object[] flattenedValues = new Object[7 * uncrawledBatch.size()]; + for (int i = 0; i < uncrawledBatch.size(); i++) { + System.arraycopy(uncrawledBatch.get(i), 0, flattenedValues, 7 * i, 7); + } + + try { troughClient().write(getSegmentId(), sqlTmpl.toString(), flattenedValues); + } catch (Exception e) { + logger.log(Level.WARNING, "problem posting batch of " + uncrawledBatch.size() + " uncrawled urls to trough segment " + getSegmentId(), e); } - } catch (Exception e) { - logger.log(Level.WARNING, "problem posting batch of " + uncrawledBatch.size() + " uncrawled urls to trough segment " + getSegmentId(), e); - } - uncrawledBatchLastTime = System.currentTimeMillis(); + uncrawledBatchLastTime = System.currentTimeMillis(); + uncrawledBatch.clear(); + } } } }