public static void validateRecords(short version, BaseRecords baseRecords) {
        if (version >= 3) {
            if (baseRecords instanceof Records) {
                Records records = (Records) baseRecords;
                Iterator<? extends RecordBatch> iterator = records.batches().iterator();
                if (!iterator.hasNext())
                    throw new InvalidRecordException("Produce requests with version " + version + " must have at least " +
                            "one record batch per partition");

                RecordBatch entry = iterator.next();
                if (entry.magic() != RecordBatch.MAGIC_VALUE_V2)
                    throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " +
                            "contain record batches with magic version 2");
                if (version < 7 && entry.compressionType() == CompressionType.ZSTD) {
                    throw new UnsupportedCompressionTypeException("Produce requests with version " + version + " are not allowed to " +
                            "use ZStandard compression");
                }

                if (iterator.hasNext())
                    throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " +
                            "contain exactly one record batch per partition");
            }
        }

        // Note that we do not do similar validation for older versions to ensure compatibility with
        // clients which send the wrong magic version in the wrong version of the produce request. The broker
        // did not do this validation before, so we maintain that behavior here.
    }
@return data or IllegalStateException if the data is removed (to prevent unnecessary memory retention).
clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java