private static Optional<ApiRecordError> validateTimestamp(RecordBatch batch,
                                                              Record record,
                                                              int recordIndex,
                                                              long now,
                                                              TimestampType timestampType,
                                                              long timestampBeforeMaxMs,
                                                              long timestampAfterMaxMs) {
        if (timestampType == TimestampType.CREATE_TIME && record.timestamp() != RecordBatch.NO_TIMESTAMP) {
            if (recordHasInvalidTimestamp(record, now, timestampBeforeMaxMs, timestampAfterMaxMs)) {
                return Optional.of(new ApiRecordError(Errors.INVALID_TIMESTAMP, new RecordError(recordIndex,
                    "Timestamp " + record.timestamp() + " of message with offset " + record.offset()
                        + " is out of range. The timestamp should be within [" + (now - timestampBeforeMaxMs)
                        + ", " + (now + timestampAfterMaxMs) + "]")));
            }
        } else if (batch.timestampType() == TimestampType.LOG_APPEND_TIME)
            return Optional.of(new ApiRecordError(Errors.INVALID_TIMESTAMP, new RecordError(recordIndex,
                "Invalid timestamp type in message " + record + ". Producer should not set timestamp "
                + "type to LogAppendTime.")));
        return Optional.empty();
    }
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