private List<CompletedBatch<T>> drainCompleted() {
        List<CompletedBatch<T>> res = new ArrayList<>(completed.size());
        while (true) {
            CompletedBatch<T> batch = completed.poll();
            if (batch == null) {
                return res;
            } else {
                res.add(batch);
            }
        }
    }
Drain completed batches. The caller is expected to first check whether
{@link #needsDrain(long)} returns true in order to avoid unnecessary draining.

Note on thread-safety: this method is safe in the presence of concurrent
appends, but it assumes a single thread is responsible for draining.

This call will not block, but the drain may require multiple attempts before
it can be completed if the thread responsible for appending is holding the
append lock. In the worst case, the append will be completed on the next
call to {@link #append(int, List, OptionalLong, boolean)} following the
initial call to this method.

The caller should respect the time to the next flush as indicated by
{@link #timeUntilDrain(long)}.

@return the list of completed batches
raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java