public static List<ProducerStateEntry> readSnapshot0(File file, byte[] buffer) throws IOException {
        try {
            // AutoMQ inject start
            if (buffer == null) {
                buffer = Files.readAllBytes(file.toPath());
            }
            // AutoMQ inject end
            Struct struct = PID_SNAPSHOT_MAP_SCHEMA.read(ByteBuffer.wrap(buffer));

            Short version = struct.getShort(VERSION_FIELD);
            if (version != PRODUCER_SNAPSHOT_VERSION)
                throw new CorruptSnapshotException("Snapshot contained an unknown file version " + version);

            long crc = struct.getUnsignedInt(CRC_FIELD);
            long computedCrc = Crc32C.compute(buffer, PRODUCER_ENTRIES_OFFSET, buffer.length - PRODUCER_ENTRIES_OFFSET);
            if (crc != computedCrc)
                throw new CorruptSnapshotException("Snapshot is corrupt (CRC is no longer valid). Stored crc: " + crc
                        + ". Computed crc: " + computedCrc);

            Object[] producerEntryFields = struct.getArray(PRODUCER_ENTRIES_FIELD);
            List<ProducerStateEntry> entries = new ArrayList<>(producerEntryFields.length);
            for (Object producerEntryObj : producerEntryFields) {
                Struct producerEntryStruct = (Struct) producerEntryObj;
                long producerId = producerEntryStruct.getLong(PRODUCER_ID_FIELD);
                short producerEpoch = producerEntryStruct.getShort(PRODUCER_EPOCH_FIELD);
                int seq = producerEntryStruct.getInt(LAST_SEQUENCE_FIELD);
                long offset = producerEntryStruct.getLong(LAST_OFFSET_FIELD);
                long timestamp = producerEntryStruct.getLong(TIMESTAMP_FIELD);
                int offsetDelta = producerEntryStruct.getInt(OFFSET_DELTA_FIELD);
                int coordinatorEpoch = producerEntryStruct.getInt(COORDINATOR_EPOCH_FIELD);
                long currentTxnFirstOffset = producerEntryStruct.getLong(CURRENT_TXN_FIRST_OFFSET_FIELD);

                OptionalLong currentTxnFirstOffsetVal = currentTxnFirstOffset >= 0 ? OptionalLong.of(currentTxnFirstOffset) : OptionalLong.empty();
                Optional<BatchMetadata> batchMetadata =
                        (offset >= 0) ? Optional.of(new BatchMetadata(seq, offset, offsetDelta, timestamp)) : Optional.empty();
                // AutoMQ inject start
                batchMetadata.ifPresent(m -> m.recovered = true);
                // AutoMQ inject end
                entries.add(new ProducerStateEntry(producerId, producerEpoch, coordinatorEpoch, timestamp, currentTxnFirstOffsetVal, batchMetadata));
            }

            return entries;
        } catch (SchemaException e) {
            throw new CorruptSnapshotException("Snapshot failed schema validation: " + e.getMessage());
        }
    }
Removes the producer state snapshot file metadata corresponding to the provided offset if it exists from this
ProducerStateManager, and renames the backing snapshot file to have the Log.DeletionSuffix.
<p>
Note: This method is safe to use with async deletes. If a race occurs and the snapshot file
is deleted without this ProducerStateManager instance knowing, the resulting exception on
SnapshotFile rename will be ignored and {@link Optional#empty()} will be returned.
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java