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

Revert "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 reverts commit d2f70e5c97.

Conflicts:
	contrib/src/main/java/org/archive/modules/postprocessor/KafkaCrawlLogFeed.java
This commit is contained in:
Noah Levitt
2014-11-10 13:55:13 -08:00
parent fa0a28f78b
commit a0f70d0865
1 changed files with 15 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/src/main/java/org/archive/modules/postprocessor/KafkaCrawlLogFeed.java
+15 -26
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,41 +164,30 @@ public class KafkaCrawlLogFeed extends Processor implements Lifecycle {
super.stop();
}
transient protected KafkaProducer kafkaProducer;
protected KafkaProducer kafkaProducer() {
transient protected Producer<byte[], byte[]> kafkaProducer;
protected Producer<byte[], byte[]> kafkaProducer() {
if (kafkaProducer == null) {
synchronized (this) {
if (kafkaProducer == null) {
Properties props = new Properties();
props.put("bootstrap.servers", getBrokerList());
props.put("acks", "1");
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("producer.type", "async");
kafkaProducer = new KafkaProducer(props);
ProducerConfig config = new ProducerConfig(props);
kafkaProducer = new Producer<byte[], byte[]>(config);
}
}
}
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);
ProducerRecord producerRecord = new ProducerRecord(getTopic(), message);
kafkaProducer().send(producerRecord, new KafkaResultCallback(curi));
KeyedMessage<byte[], byte[]> keyedMessage = new KeyedMessage<byte[], byte[]>(getTopic(), message);
// XXX no error handling
kafkaProducer().send(keyedMessage);
}
}