lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-09-20 12:46:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

switch to org.apache.kafka.clients.producer.KafkaProducer to see if we can avoid the InterruptedException at the end of the crawl (and log warnings on failure to send message)

Browse Source
This commit is contained in:
Noah Levitt
2014-11-06 19:26:32 -08:00
parent 23fa29d673
commit d2f70e5c97
1 changed files with 25 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/src/main/java/org/archive/modules/postprocessor/KafkaCrawlLogFeed.java
+25 -14
View File
@@ -23,11 +23,11 @@ import java.util.Map;
import java.util.Properties;
import java.util.logging.Logger;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.commons.collections.Closure;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.archive.crawler.framework.Frontier;
import org.archive.crawler.frontier.AbstractFrontier;
import org.archive.crawler.frontier.BdbFrontier;
@@ -164,30 +164,41 @@ public class KafkaCrawlLogFeed extends Processor implements Lifecycle {
super.stop();
}
transient protected Producer<byte[], byte[]> kafkaProducer;
protected Producer<byte[], byte[]> kafkaProducer() {
transient protected KafkaProducer kafkaProducer;
protected KafkaProducer kafkaProducer() {
if (kafkaProducer == null) {
synchronized (this) {
if (kafkaProducer == null) {
Properties props = new Properties();
props.put("metadata.broker.list", getBrokerList());
// XXX in asynchronous mode, we have no way to handle delivery
// problems, so it doesn't matter if they are acked or not
props.put("request.required.acks", "0");
props.put("request.required.acks", "1");
props.put("producer.type", "async");
ProducerConfig config = new ProducerConfig(props);
kafkaProducer = new Producer<byte[], byte[]>(config);
kafkaProducer = new KafkaProducer(props);
}
}
}
return kafkaProducer;
}
protected static class KafkaResultCallback implements Callback {
private CrawlURI curi;
public KafkaResultCallback(CrawlURI curi) {
this.curi = curi;
}
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
logger.warning("kafka delivery failed for " + curi + " - " + exception);
}
}
}
@Override
protected void innerProcess(CrawlURI curi) throws InterruptedException {
byte[] message = buildMessage(curi);
KeyedMessage<byte[], byte[]> keyedMessage = new KeyedMessage<byte[], byte[]>(getTopic(), message);
// XXX no error handling
kafkaProducer().send(keyedMessage);
ProducerRecord producerRecord = new ProducerRecord(getTopic(), message);
kafkaProducer().send(producerRecord, new KafkaResultCallback(curi));
}
}