private AcknowledgementBatch maybeCreateNewBatch(AcknowledgementBatch currentBatch, Long nextOffset, List<AcknowledgementBatch> batches) {
        if (nextOffset != currentBatch.lastOffset() + 1) {
            List<AcknowledgementBatch> optimalBatches = maybeOptimiseAcknowledgementTypes(currentBatch);

            optimalBatches.forEach(batch -> {
                if (canOptimiseForSingleAcknowledgeType(batch)) {
                    // If the batch had a single acknowledgement type, we optimise the array independent
                    // of the number of records.
                    batch.acknowledgeTypes().subList(1, batch.acknowledgeTypes().size()).clear();
                }
                batches.add(batch);
            });

            currentBatch = new AcknowledgementBatch();
            currentBatch.setFirstOffset(nextOffset);
        }

        return currentBatch;
    }
Creates a new current batch if the next offset is not one higher than the current batch's
last offset.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Acknowledgements.java