class KafkaCrawlLogFeed extends Processor implements Lifecycle {

    protected static final Logger logger = Logger.getLogger(KafkaCrawlLogFeed.class.getName());

    protected Frontier frontier;
    public Frontier getFrontier() {
        return this.frontier;
    }
    /** Autowired frontier, needed to determine when a url is finished. */
    @Autowired
    public void setFrontier(Frontier frontier) {
        this.frontier = frontier;
    }

    protected ServerCache serverCache;
    public ServerCache getServerCache() {
        return this.serverCache;
    }
    @Autowired
    public void setServerCache(ServerCache serverCache) {
        this.serverCache = serverCache;
    }

    protected Map<String,String> extraFields;
    public Map<String, String> getExtraFields() {
        return extraFields;
    }
    public void setExtraFields(Map<String, String> extraFields) {
        this.extraFields = extraFields;
    }

    protected boolean dumpPendingAtClose = false;
    public boolean getDumpPendingAtClose() {
        return dumpPendingAtClose;
    }
    /**
     * If true, publish all pending urls (i.e. queued urls still in the
     * frontier) when crawl job is stopping. They are recognizable by the status
     * field which has the value 0.
     *
     * @see BdbFrontier#setDumpPendingAtClose(boolean)
     */
    public void setDumpPendingAtClose(boolean dumpPendingAtClose) {
        this.dumpPendingAtClose = dumpPendingAtClose;
    }

    protected String brokerList = "localhost:9092";
    /** Kafka broker list (kafka property "metadata.broker.list"). */
    public void setBrokerList(String brokerList) {
        this.brokerList = brokerList;
    }
    public String getBrokerList() {
        return brokerList;
    }

    protected String topic = "heritrix-crawl-log";
    public void setTopic(String topic) {
        this.topic = topic;
    }
    public String getTopic() {
        return topic;
    }

    protected byte[] buildMessage(CrawlURI curi) {
        JSONObject jo = CrawlLogJsonBuilder.buildJson(curi, getExtraFields(), getServerCache());
        try {
            return jo.toString().getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    protected boolean shouldProcess(CrawlURI curi) {
        if (frontier instanceof AbstractFrontier) {
            return !((AbstractFrontier) frontier).needsReenqueuing(curi);
        } else {
            return false;
        }
    }

    private transient long pendingDumpedCount = 0l;
    @Override
    public synchronized void stop() {
        if (!isRunning) {
            return;
        }

        if (dumpPendingAtClose) {
            if (frontier instanceof BdbFrontier) {

                Closure closure = new Closure() {
                    public void execute(Object curi) {
                        try {
                            innerProcess((CrawlURI) curi);
                            pendingDumpedCount++;
                        } catch (InterruptedException e) {
                        }
                    }
                };

                logger.info("dumping " + frontier.queuedUriCount() + " queued urls to kafka feed");
                ((BdbFrontier) frontier).forAllPendingDo(closure);
                logger.info("dumped " + pendingDumpedCount + " queued urls to kafka feed");
            } else {
                logger.warning("frontier is not a BdbFrontier, cannot dumpPendingAtClose");
            }
        }

        String rateStr = String.format("%1.1f", 0.01 * stats.errors / stats.total);
        logger.info("final error count: " + stats.errors + "/" + stats.total + " (" + rateStr + "%)");

        if (kafkaProducer != null) {
            kafkaProducer.close();
            kafkaProducer = null;
        }
        if (kafkaProducerThreads != null) {
            kafkaProducerThreads.destroy();
            kafkaProducerThreads = null;
        }

        super.stop();
    }

    private transient ThreadGroup kafkaProducerThreads;

    transient protected KafkaProducer<String, byte[]> kafkaProducer;
    protected KafkaProducer<String, byte[]> kafkaProducer() {<FILL_FUNCTION_BODY>}

    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) {
                errors++;
            }

            if (total % 10000 == 0) {
                String rateStr = String.format("%1.1f", 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, stats);
    }
}

        if (kafkaProducer == null) {
            synchronized (this) {
                if (kafkaProducer == null) {
                    final Properties props = new Properties();
                    props.put("bootstrap.servers", getBrokerList());
                    props.put("acks", "1");
                    props.put("producer.type", "async");
                    props.put("key.serializer", StringSerializer.class.getName());
                    props.put("value.serializer", ByteArraySerializer.class.getName());

                    /*
                     * XXX This mess here exists so that the kafka producer
                     * thread is in a thread group that is not the ToePool,
                     * so that it doesn't get interrupted at the end of the
                     * crawl in ToePool.cleanup(). 
                     */
                    kafkaProducerThreads = new ThreadGroup(Thread.currentThread().getThreadGroup().getParent(), "KafkaProducerThreads");
                    ThreadFactory threadFactory = new ThreadFactory() {
                        public Thread newThread(Runnable r) {
                            return new Thread(kafkaProducerThreads, r);
                        }
                    };
                    Callable<KafkaProducer<String,byte[]>> task = new Callable<KafkaProducer<String,byte[]>>() {
                        public KafkaProducer<String, byte[]> call() throws InterruptedException {
                            return new KafkaProducer<String,byte[]>(props);
                        }
                    };
                    ExecutorService executorService = Executors.newFixedThreadPool(1, threadFactory);
                    Future<KafkaProducer<String, byte[]>> future = executorService.submit(task);
                    try {
                        kafkaProducer = future.get();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    } catch (ExecutionException e) {
                        throw new RuntimeException(e);
                    } finally {
                        executorService.shutdown();
                    }
                }
            }
        }
        return kafkaProducer;
    
<methods>public non-sealed void <init>() ,public void doCheckpoint(org.archive.checkpointing.Checkpoint) throws java.io.IOException,public void finishCheckpoint(org.archive.checkpointing.Checkpoint) ,public static java.lang.String flattenVia(org.archive.modules.CrawlURI) ,public java.lang.String getBeanName() ,public boolean getEnabled() ,public org.archive.spring.KeyedProperties getKeyedProperties() ,public static long getRecordedSize(org.archive.modules.CrawlURI) ,public org.archive.modules.deciderules.DecideRule getShouldProcessRule() ,public long getURICount() ,public static boolean hasHttpAuthenticationCredential(org.archive.modules.CrawlURI) ,public boolean isRunning() ,public static boolean isSuccess(org.archive.modules.CrawlURI) ,public org.archive.modules.ProcessResult process(org.archive.modules.CrawlURI) throws java.lang.InterruptedException,public java.lang.String report() ,public void setBeanName(java.lang.String) ,public void setEnabled(boolean) ,public void setRecoveryCheckpoint(org.archive.checkpointing.Checkpoint) ,public void setShouldProcessRule(org.archive.modules.deciderules.DecideRule) ,public void start() ,public void startCheckpoint(org.archive.checkpointing.Checkpoint) ,public void stop() <variables>protected java.lang.String beanName,protected boolean isRunning,protected org.archive.spring.KeyedProperties kp,protected org.archive.checkpointing.Checkpoint recoveryCheckpoint,protected java.util.concurrent.atomic.AtomicLong uriCount