class StandardProcessSession implements ProcessSession, ProvenanceEventEnricher {

    @Override
    public FlowFile append(FlowFile source, final OutputStreamCallback writer) {
        verifyTaskActive();

        source = validateRecordState(source);
        final StandardRepositoryRecord record = records.get(source);
        long newSize = 0L;

        // Get the current Content Claim from the record and see if we already have
        // an OutputStream that we can append to.
        final ContentClaim oldClaim = record.getCurrentClaim();
        ByteCountingOutputStream outStream = oldClaim == null ? null : appendableStreams.get(oldClaim);
        long originalByteWrittenCount = 0;

        ContentClaim newClaim = null;
        try {
            if (outStream == null) {
                claimCache.flush(oldClaim);

                try (final InputStream oldClaimIn = read(source)) {
                    newClaim = context.getContentRepository().create(context.getConnectable().isLossTolerant());
                    claimLog.debug("Creating ContentClaim {} for 'append' for {}", newClaim, source);

                    final OutputStream rawOutStream = context.getContentRepository().write(newClaim);
                    final OutputStream bufferedOutStream = new BufferedOutputStream(rawOutStream);
                    outStream = new ByteCountingOutputStream(bufferedOutStream);
                    originalByteWrittenCount = 0;

                    appendableStreams.put(newClaim, outStream);

                    // We need to copy all of the data from the old claim to the new claim
                    StreamUtils.copy(oldClaimIn, outStream);

                    // wrap our OutputStreams so that the processor cannot close it
                    try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream)) {
                        writeRecursionSet.add(source);
                        writer.process(new FlowFileAccessOutputStream(disableOnClose, source));
                    } finally {
                        writeRecursionSet.remove(source);
                    }
                }
            } else {
                newClaim = oldClaim;
                originalByteWrittenCount = outStream.getBytesWritten();

                // wrap our OutputStreams so that the processor cannot close it
                try (final OutputStream disableOnClose = new DisableOnCloseOutputStream(outStream);
                    final OutputStream flowFileAccessOutStream = new FlowFileAccessOutputStream(disableOnClose, source)) {
                    writeRecursionSet.add(source);
                    writer.process(flowFileAccessOutStream);
                } finally {
                    writeRecursionSet.remove(source);
                }
            }

            // update the newSize to reflect the number of bytes written
            newSize = outStream.getBytesWritten();
        } catch (final ContentNotFoundException nfe) {
            resetWriteClaims(); // need to reset write claim before we can remove the claim

            // If the content claim changed, then we should destroy the new one. We do this
            // because the new content claim will never get set as the 'working claim' for the FlowFile
            // record since we will throw an Exception. As a result, we need to ensure that we have
            // appropriately decremented the claimant count and can destroy the content if it is no
            // longer in use. However, it is critical that we do this ONLY if the content claim has
            // changed. Otherwise, the FlowFile already has a reference to this Content Claim and
            // whenever the FlowFile is removed, the claim count will be decremented; if we decremented
            // it here also, we would be decrementing the claimant count twice!
            if (newClaim != oldClaim) {
                destroyContent(newClaim);
            }

            handleContentNotFound(nfe, record);
        } catch (final IOException ioe) {
            resetWriteClaims(); // need to reset write claim before we can remove the claim

            // See above explanation for why this is done only if newClaim != oldClaim
            if (newClaim != oldClaim) {
                destroyContent(newClaim);
            }

            throw new ProcessException("IOException thrown from " + connectableDescription + ": " + ioe.toString(), ioe);
        } catch (final Throwable t) {
            resetWriteClaims(); // need to reset write claim before we can remove the claim

            // See above explanation for why this is done only if newClaim != oldClaim
            if (newClaim != oldClaim) {
                destroyContent(newClaim);
            }

            throw t;
        } finally {
            if (outStream != null) {
                final long bytesWrittenThisIteration = outStream.getBytesWritten() - originalByteWrittenCount;
                bytesWritten += bytesWrittenThisIteration;
            }
        }

        // If the record already has a working claim, and this is the first time that we are appending to the FlowFile,
        // destroy the current working claim because it is a temporary claim that
        // is no longer going to be used, as we are about to set a new working claim. This would happen, for instance, if
        // the FlowFile was written to, via #write() and then append() was called.
        if (newClaim != oldClaim) {
            removeTemporaryClaim(record);
        }

        final FlowFileRecord newFile = new StandardFlowFileRecord.Builder().fromFlowFile(record.getCurrent()).contentClaim(newClaim).contentClaimOffset(0).size(newSize).build();
        record.setWorking(newFile);
        return newFile;
    }

    public  StandardProcessSession(final RepositoryContext context, final TaskTermination taskTermination);

}