private static void validateBatch(TopicPartition topicPartition,
                                      RecordBatch firstBatch,
                                      RecordBatch batch,
                                      AppendOrigin origin,
                                      byte toMagic,
                                      MetricsRecorder metricsRecorder) {
        // batch magic byte should have the same magic as the first batch
        if (firstBatch.magic() != batch.magic()) {
            metricsRecorder.recordInvalidMagic();
            throw new InvalidRecordException("Batch magic " + batch.magic() + " is not the same as the first batch's magic byte "
                + firstBatch.magic() + " in topic partition " + topicPartition);
        }

        if (origin == AppendOrigin.CLIENT) {
            if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2) {
                long countFromOffsets = batch.lastOffset() - batch.baseOffset() + 1;
                if (countFromOffsets <= 0) {
                    metricsRecorder.recordInvalidOffset();
                    throw new InvalidRecordException("Batch has an invalid offset range: [" + batch.baseOffset() + ", "
                        + batch.lastOffset() + "] in topic partition " + topicPartition);
                }

                // v2 and above messages always have a non-null count
                long count = batch.countOrNull();
                if (count <= 0) {
                    metricsRecorder.recordInvalidOffset();
                    throw new InvalidRecordException("Invalid reported count for record batch: " + count
                        + " in topic partition " + topicPartition);
                }

                if (countFromOffsets != count) {
                    metricsRecorder.recordInvalidOffset();
                    throw new InvalidRecordException("Inconsistent batch offset range [" + batch.baseOffset() + ", "
                        + batch.lastOffset() + "] and count of records " + count + " in topic partition " + topicPartition);
                }
            }

            if (batch.isControlBatch()) {
                metricsRecorder.recordInvalidOffset();
                throw new InvalidRecordException("Clients are not allowed to write control records in topic partition " + topicPartition);
            }

            if (batch.hasProducerId() && batch.baseSequence() < 0) {
                metricsRecorder.recordInvalidSequence();
                throw new InvalidRecordException("Invalid sequence number " + batch.baseSequence() + " in record batch with producerId "
                    + batch.producerId() + " in topic partition " + topicPartition);
            }
        }

        if (batch.isTransactional() && toMagic < RecordBatch.MAGIC_VALUE_V2)
            throw new UnsupportedForMessageFormatException("Transactional records cannot be used with magic version " + toMagic);

        if (batch.hasProducerId() && toMagic < RecordBatch.MAGIC_VALUE_V2)
            throw new UnsupportedForMessageFormatException("Idempotent records cannot be used with magic version " + toMagic);
    }
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