private static Optional<ApiRecordError> validateRecord(RecordBatch batch,
                                                           TopicPartition topicPartition,
                                                           Record record,
                                                           int recordIndex,
                                                           long now,
                                                           TimestampType timestampType,
                                                           long timestampBeforeMaxMs,
                                                           long timestampAfterMaxMs,
                                                           boolean compactedTopic,
                                                           MetricsRecorder metricsRecorder) {
        if (!record.hasMagic(batch.magic())) {
            metricsRecorder.recordInvalidMagic();
            return Optional.of(new ApiRecordError(Errors.INVALID_RECORD,
                new RecordError(recordIndex, "Record " + record
                    + "'s magic does not match outer magic " + batch.magic() + " in topic partition "
                    + topicPartition)));
        }

        // verify the record-level CRC only if this is one of the deep entries of a compressed message
        // set for magic v0 and v1. For non-compressed messages, there is no inner record for magic v0 and v1,
        // so we depend on the batch-level CRC check in Log.analyzeAndValidateRecords(). For magic v2 and above,
        // there is no record-level CRC to check.
        if (batch.magic() <= RecordBatch.MAGIC_VALUE_V1 && batch.isCompressed()) {
            try {
                record.ensureValid();
            } catch (CorruptRecordException e) {
                metricsRecorder.recordInvalidChecksums();
                throw e;
            } catch (InvalidRecordException e) {
                metricsRecorder.recordInvalidChecksums();
                throw new CorruptRecordException(e.getMessage() + " in topic partition " + topicPartition);
            }
        }

        Optional<ApiRecordError> keyError = validateKey(record, recordIndex, topicPartition,
            compactedTopic, metricsRecorder);
        if (keyError.isPresent())
            return keyError;
        else
            return validateTimestamp(batch, record, recordIndex, now, timestampType, timestampBeforeMaxMs, timestampAfterMaxMs);
    }
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