public ValidationResult validateMessagesAndAssignOffsetsCompressed(LongRef offsetCounter,
                                                                       MetricsRecorder metricsRecorder,
                                                                       BufferSupplier bufferSupplier) {
        if (targetCompression.type() == CompressionType.ZSTD && interBrokerProtocolVersion.isLessThan(IBP_2_1_IV0))
            throw new UnsupportedCompressionTypeException("Produce requests to inter.broker.protocol.version < 2.1 broker " +
                "are not allowed to use ZStandard compression");

        // No in place assignment situation 1
        boolean inPlaceAssignment = sourceCompressionType == targetCompression.type();
        long now = time.milliseconds();

        long maxTimestamp = RecordBatch.NO_TIMESTAMP;
        LongRef expectedInnerOffset = PrimitiveRef.ofLong(0);
        List<Record> validatedRecords = new ArrayList<>();
        long offsetOfMaxTimestamp = -1;
        long initialOffset = offsetCounter.value;

        int uncompressedSizeInBytes = 0;

        // Assume there's only one batch with compressed memory records; otherwise, return InvalidRecordException
        // One exception though is that with format smaller than v2, if sourceCompression is noCompression, then each batch is actually
        // a single record so we'd need to special handle it by creating a single wrapper batch that includes all the records
        MutableRecordBatch firstBatch = getFirstBatchAndMaybeValidateNoMoreBatches(records, sourceCompressionType);

        // No in place assignment situation 2 and 3: we only need to check for the first batch because:
        //  1. For most cases (compressed records, v2, for example), there's only one batch anyways.
        //  2. For cases that there may be multiple batches, all batches' magic should be the same.
        if (firstBatch.magic() != toMagic || toMagic == RecordBatch.MAGIC_VALUE_V0)
            inPlaceAssignment = false;

        // Do not compress control records unless they are written compressed
        if (sourceCompressionType == CompressionType.NONE && firstBatch.isControlBatch())
            inPlaceAssignment = true;

        for (MutableRecordBatch batch : records.batches()) {
            validateBatch(topicPartition, firstBatch, batch, origin, toMagic, metricsRecorder);
            uncompressedSizeInBytes += AbstractRecords.recordBatchHeaderSizeInBytes(toMagic, batch.compressionType());

            // if we are on version 2 and beyond, and we know we are going for in place assignment,
            // then we can optimize the iterator to skip key / value / headers since they would not be used at all
            CloseableIterator<Record> recordsIterator;
            if (inPlaceAssignment && firstBatch.magic() >= RecordBatch.MAGIC_VALUE_V2)
                recordsIterator = batch.skipKeyValueIterator(bufferSupplier);
            else
                recordsIterator = batch.streamingIterator(bufferSupplier);

            try {
                List<ApiRecordError> recordErrors = new ArrayList<>(0);
                // this is a hot path and we want to avoid any unnecessary allocations.
                int recordIndex = 0;
                while (recordsIterator.hasNext()) {
                    Record record = recordsIterator.next();
                    long expectedOffset = expectedInnerOffset.value++;

                    Optional<ApiRecordError> recordError = validateRecordCompression(sourceCompressionType,
                        recordIndex, record);
                    if (!recordError.isPresent()) {
                        recordError = validateRecord(batch, topicPartition, record, recordIndex, now,
                            timestampType, timestampBeforeMaxMs, timestampAfterMaxMs, compactedTopic, metricsRecorder);
                    }

                    if (!recordError.isPresent()
                            && batch.magic() > RecordBatch.MAGIC_VALUE_V0
                            && toMagic > RecordBatch.MAGIC_VALUE_V0) {

                        if (record.timestamp() > maxTimestamp) {
                            maxTimestamp = record.timestamp();
                            // The offset is only increased when it is a valid record
                            offsetOfMaxTimestamp = initialOffset + validatedRecords.size();
                        }

                        // Some older clients do not implement the V1 internal offsets correctly.
                        // Historically the broker handled this by rewriting the batches rather
                        // than rejecting the request. We must continue this handling here to avoid
                        // breaking these clients.
                        if (record.offset() != expectedOffset)
                            inPlaceAssignment = false;
                    }

                    if (recordError.isPresent())
                        recordErrors.add(recordError.get());
                    else {
                        uncompressedSizeInBytes += record.sizeInBytes();
                        validatedRecords.add(record);
                    }

                    ++recordIndex;
                }

                processRecordErrors(recordErrors);

            } finally {
                recordsIterator.close();
            }
        }

        if (!inPlaceAssignment) {
            return buildRecordsAndAssignOffsets(offsetCounter, now, firstBatch, validatedRecords,
                uncompressedSizeInBytes);
        } else {
            // we can update the batch only and write the compressed payload as is;
            // again we assume only one record batch within the compressed set
            offsetCounter.value += validatedRecords.size();
            // there is only one batch in this path, so last offset can be viewed as shallowOffsetOfMaxTimestamp
            long lastOffset = offsetCounter.value - 1;
            firstBatch.setLastOffset(lastOffset);

            if (timestampType == TimestampType.LOG_APPEND_TIME) {
                maxTimestamp = now;
                offsetOfMaxTimestamp = initialOffset;
            }

            if (toMagic >= RecordBatch.MAGIC_VALUE_V1)
                firstBatch.setMaxTimestamp(timestampType, maxTimestamp);

            if (toMagic >= RecordBatch.MAGIC_VALUE_V2)
                firstBatch.setPartitionLeaderEpoch(partitionLeaderEpoch);

            RecordValidationStats recordValidationStats = new RecordValidationStats(uncompressedSizeInBytes, 0, 0);
            return new ValidationResult(
                now,
                records,
                maxTimestamp,
                offsetOfMaxTimestamp,
                false,
                recordValidationStats);
        }
    }
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