class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements HasKeyedProperties {
    private static final Logger logger = 
            Logger.getLogger(WARCWriterChainProcessor.class.getName());
    
    {
        setChain(Arrays.asList(
                new DnsResponseRecordBuilder(),
                new HttpResponseRecordBuilder(),
                new WhoisResponseRecordBuilder(),
                new FtpControlConversationRecordBuilder(),
                new FtpResponseRecordBuilder(),
                new RevisitRecordBuilder(),
                new HttpRequestRecordBuilder(),
                new MetadataRecordBuilder()));
    }
    @SuppressWarnings("unchecked")
    public List<? extends WARCRecordBuilder> getChain() {
        return (List<WARCRecordBuilder>) kp.get("chain");
    }
    public void setChain(List<? extends WARCRecordBuilder> chain) {
        kp.put("chain", chain);
    }

    @Override
    protected boolean shouldWrite(CrawlURI curi) {
        if (getSkipIdenticalDigests()
                && IdenticalDigestDecideRule.hasIdenticalDigest(curi)) {
            curi.getAnnotations().add(ANNOTATION_UNWRITTEN 
                    + ":identicalDigest");
            return false;
        }

        // WARCWriterProcessor has seemingly unnecessarily complicated logic
        if (curi.getFetchStatus() <= 0) {
            curi.getAnnotations().add(ANNOTATION_UNWRITTEN + ":status");
            return false;
        }
        
        return true;
    }
    
    @Override
    protected ProcessResult innerProcessResult(CrawlURI curi) {
        try {
            if (shouldWrite(curi)) {
                return write(curi);
            } else {
                copyForwardWriteTagIfDupe(curi);
            }
        } catch (IOException e) {
            curi.getNonFatalFailures().add(e);
            logger.log(Level.SEVERE, "Failed write of Records: " +
                curi.toString(), e);
        }
        return ProcessResult.PROCEED;
    }
    
    protected ProcessResult write(final CrawlURI curi)
    throws IOException {
        WARCWriter writer = (WARCWriter) getPool().borrowFile();

        // Reset writer temp stats so they reflect only this set of records.
        writer.resetTmpStats();
        writer.resetTmpRecordLog();

        long position = writer.getPosition();
        try {
            // Roll over to new warc file if we've exceeded maxBytes.
            writer.checkSize();
            if (writer.getPosition() != position) {
                // We rolled over to a new warc and wrote a warcinfo record.
                // Tally stats and reset temp stats, to avoid including warcinfo
                // record in stats for current url.
                addTotalBytesWritten(writer.getPosition() - position);
                addStats(writer.getTmpStats());
                writer.resetTmpStats();
                writer.resetTmpRecordLog();

                position = writer.getPosition();
            }

            writeRecords(curi, writer);
        } catch (IOException e) {
            // Invalidate this file (It gets a '.invalid' suffix).
            getPool().invalidateFile(writer);
            // Set the writer to null otherwise the pool accounting
            // of how many active writers gets skewed if we subsequently
            // do a returnWriter call on this object in the finally block.
            writer = null;
            throw e;
        } finally {
            if (writer != null) {
                updateMetadataAfterWrite(curi, writer, position);
                getPool().returnFile(writer);
            }
        }
        // XXX this looks wrong, check should happen *before* writing the
        // record, the way checkBytesWritten() currently works
        return checkBytesWritten();
    }

    protected void writeRecords(CrawlURI curi, WARCWriter writer) throws IOException {<FILL_FUNCTION_BODY>}
    @Override
    protected JSONObject toCheckpointJson() throws JSONException {
        JSONObject json = super.toCheckpointJson();
        json.put("urlsWritten", urlsWritten);
        json.put("stats", stats);
        return json;
    }
    
    @Override
    protected void fromCheckpointJson(JSONObject json) throws JSONException {
        super.fromCheckpointJson(json);

        // conditionals below are for backward compatibility with old checkpoints
        
        if (json.has("urlsWritten")) {
            urlsWritten.set(json.getLong("urlsWritten"));
        }
        
        if (json.has("stats")) {
            HashMap<String, Map<String, Long>> cpStats = new HashMap<String, Map<String, Long>>();
            JSONObject jsonStats = json.getJSONObject("stats");
            if (JSONObject.getNames(jsonStats) != null) {
                for (String key1: JSONObject.getNames(jsonStats)) {
                    JSONObject jsonSubstats = jsonStats.getJSONObject(key1);
                    if (!cpStats.containsKey(key1)) {
                        cpStats.put(key1, new HashMap<String, Long>());
                    }
                    Map<String, Long> substats = cpStats.get(key1);

                    for (String key2: JSONObject.getNames(jsonSubstats)) {
                        long value = jsonSubstats.getLong(key2);
                        substats.put(key2, value);
                    }
                }
                addStats(cpStats);
            }
        }
    }
}

        URI concurrentTo = null;
        for (WARCRecordBuilder recordBuilder: getChain()) {
            if (recordBuilder.shouldBuildRecord(curi)) {
                WARCRecordInfo record = recordBuilder.buildRecord(curi, concurrentTo);
                if (record != null) {
                    writer.writeRecord(record);
                    InputStream is = null;
                    try {
                        is = record.getContentStream();
                        is.close();
                    }
                    catch (Exception e){
                        logger.log(Level.WARNING, "problem closing Warc Record Content Stream " + e);
                    }
                    finally {
                        IOUtils.closeQuietly(record.getContentStream()); //Closing one way or the other seems to leave some file handles open. Calling close() and using closeQuietly() handles both FileStreams and FileChannels 
                    }
                    if (concurrentTo == null) {
                        concurrentTo = record.getRecordId();
                    }
                }
            }
        }
    
<methods>public non-sealed void <init>() ,public long getDefaultMaxFileSize() ,public List<org.archive.spring.ConfigPath> getDefaultStorePaths() ,public List<java.lang.String> getMetadata() ,public RecordIDGenerator getRecordIDGenerator() ,public ConcurrentMap<java.lang.String,ConcurrentMap<java.lang.String,java.util.concurrent.atomic.AtomicLong>> getStats() ,public java.lang.String report() ,public void setRecordIDGenerator(RecordIDGenerator) <variables>private transient List<java.lang.String> cachedMetadata,protected RecordIDGenerator generator,private static final java.util.logging.Logger logger,protected ConcurrentMap<java.lang.String,ConcurrentMap<java.lang.String,java.util.concurrent.atomic.AtomicLong>> stats,protected java.util.concurrent.atomic.AtomicLong urlsWritten