BatchAccumulator.java

class BatchAccumulator {
    // ...
    private Long append(int epoch, List<T> records, boolean isAtomic) {
        if (epoch != this.epoch) {
            return Long.MAX_VALUE;
        }

        ObjectSerializationCache serializationCache = new ObjectSerializationCache();
        int batchSize = 0;
        for (T record : records) {
            batchSize += serde.recordSize(record, serializationCache);
        }

        appendLock.lock();
        try {
            maybeCompleteDrain();

            BatchBuilder<T> batch = null;
            if (isAtomic) {
                batch = maybeAllocateBatch(batchSize);
            }

            for (T record : records) {
                if (!isAtomic) {
                    batch = maybeAllocateBatch(
                        serde.recordSize(record, serializationCache)
                    );
                }
                if (batch == null) {
                    return null;
                }

                batch.appendRecord(record, serializationCache);
                nextOffset += 1;
            }

            maybeResetLinger();

            return nextOffset - 1;
        } finally {
            appendLock.unlock();
        }
    }
}