private static void processRecordErrors(List<ApiRecordError> recordErrors) {
        if (!recordErrors.isEmpty()) {
            List<RecordError> errors = recordErrors.stream().map(e -> e.recordError).collect(Collectors.toList());
            if (recordErrors.stream().anyMatch(e -> e.apiError == Errors.INVALID_TIMESTAMP)) {
                throw new RecordValidationException(new InvalidTimestampException(
                    "One or more records have been rejected due to invalid timestamp"), errors);
            } else {
                throw new RecordValidationException(new InvalidRecordException(
                    "One or more records have been rejected due to " + errors.size() + " record errors "
                    + "in total, and only showing the first three errors at most: " + errors.subList(0, Math.min(errors.size(), 3))), errors);
            }
        }
    }
We cannot do in place assignment in one of the following situations:
1. Source and target compression codec are different
2. When the target magic is not equal to batches' magic, meaning format conversion is needed.
3. When the target magic is equal to V0, meaning absolute offsets need to be re-assigned.
storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java