public Optional<BatchMetadata> findDuplicateBatch(RecordBatch batch) {
        // AutoMQ inject start
        if (batch.producerEpoch() != producerEpoch) {
            return Optional.empty();
        }
        Optional<BatchMetadata> metadata = batchWithSequenceRange(batch.baseSequence(), batch.lastSequence());
        if (metadata.isPresent()) {
            return metadata;
        }
        BatchMetadata front = batchMetadata.peek();
        if (front != null && front.recovered) {
            // the batch metadata (`front`) is recovered from snapshot
            boolean batchFallInFront = contains(front.firstSeq(), front.lastSeq, batch.baseSequence())
                && contains(front.firstSeq(), front.lastSeq, batch.lastSequence());
            if (batchFallInFront) {
                throw new DuplicateSequenceException(
                    String.format("The batch is duplicated (recover from snapshot), broker cached metadata is %s, the record batch is [%s, %s]",
                        this, batch.baseSequence(), batch.lastSequence())
                );
            }
        }
        if (front != null) {
            // regard the batch as duplicated if it is before the first cached batch
            boolean batchBeforeFront = contains(
                decrementSequence(front.firstSeq(), 65536),
                decrementSequence(front.firstSeq(), 1),
                batch.lastSequence()
            );
            if (batchBeforeFront) {
                throw new DuplicateSequenceException(
                    String.format("The batch is duplicated, broker cached metadata is %s, the record batch is [%s, %s]",
                        this, batch.baseSequence(), batch.lastSequence())
                );
            }
        }
        return metadata;
        // AutoMQ inject end
    }
Returns a new instance with the provided parameters (when present) and the values from the current instance
otherwise.
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateEntry.java