private void maybeEnsureValid(FetchConfig fetchConfig, RecordBatch batch) {
        if (fetchConfig.checkCrcs && batch.magic() >= RecordBatch.MAGIC_VALUE_V2) {
            try {
                batch.ensureValid();
            } catch (CorruptRecordException e) {
                throw new KafkaException("Record batch for partition " + partition + " at offset " +
                        batch.baseOffset() + " is invalid, cause: " + e.getMessage());
            }
        }
    }
Draining a {@link CompletedFetch} will signal that the data has been consumed and the underlying resources
are closed. This is somewhat analogous to {@link Closeable#close() closing}, though no error will result if a
caller invokes {@link #fetchRecords(FetchConfig, Deserializers, int)}; an empty {@link List list} will be
returned instead.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java