lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-07-23 07:51:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #178 from internetarchive/kafka-mem-hog

Browse Source 
KafkaCrawlLogFeed had been using lots of heap because each callback i…
This commit is contained in:
Barbara Miller
2017-03-07 11:11:23 -08:00
committed by GitHub
parent 0ca88c6071 bd6607739d
commit 548ea5eb84
1 changed files with 15 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/src/main/java/org/archive/modules/postprocessor/KafkaCrawlLogFeed.java
+15 -8
View File
@@ -164,6 +164,9 @@ public class KafkaCrawlLogFeed extends Processor implements Lifecycle {
}
}
String rateStr = String.format("%0.01f", 0.01 * stats.errors / stats.total);
logger.info("final error count: " + stats.errors + "/" + stats.total + " (" + rateStr + "%)");
if (kafkaProducer != null) {
kafkaProducer.close();
kafkaProducer = null;
@@ -224,25 +227,29 @@ public class KafkaCrawlLogFeed extends Processor implements Lifecycle {
return kafkaProducer;
}
protected static class KafkaResultCallback implements Callback {
private CrawlURI curi;
public KafkaResultCallback(CrawlURI curi) {
this.curi = curi;
}
protected final class StatsCallback implements Callback {
public long errors = 0l;
public long total = 0l;
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
total++;
if (exception != null) {
logger.warning("kafka delivery failed for " + curi + " - " + exception);
errors++;
}
if (total % 10000 == 0) {
String rateStr = String.format("%0.01f", 0.01 * errors / total);
logger.info("error count so far: " + errors + "/" + total + " (" + rateStr + "%)");
}
}
}
protected StatsCallback stats = new StatsCallback();
@Override
protected void innerProcess(CrawlURI curi) throws InterruptedException {
byte[] message = buildMessage(curi);
ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<String,byte[]>(getTopic(), message);
kafkaProducer().send(producerRecord, new KafkaResultCallback(curi));
kafkaProducer().send(producerRecord, stats);
}
}