public static ByteBuffer writeSnapshot(File file, Map<Long, ProducerStateEntry> entries, boolean sync) throws IOException {
        Struct struct = new Struct(PID_SNAPSHOT_MAP_SCHEMA);
        struct.set(VERSION_FIELD, PRODUCER_SNAPSHOT_VERSION);
        struct.set(CRC_FIELD, 0L); // we'll fill this after writing the entries
        Struct[] structEntries = new Struct[entries.size()];
        int i = 0;
        for (Map.Entry<Long, ProducerStateEntry> producerIdEntry : entries.entrySet()) {
            Long producerId = producerIdEntry.getKey();
            ProducerStateEntry entry = producerIdEntry.getValue();
            // AutoMQ inject start
            // encode cached entries sequence range to the snapshot, so we can detect the duplicated message after partition reassignment
            // TODO: final solution, encoded full cached entries to the snapshot
            int offsetDelta = entry.lastSeq() - entry.firstSeq();
            // AutoMQ inject end
            Struct producerEntryStruct = struct.instance(PRODUCER_ENTRIES_FIELD);
            producerEntryStruct.set(PRODUCER_ID_FIELD, producerId)
                    .set(PRODUCER_EPOCH_FIELD, entry.producerEpoch())
                    .set(LAST_SEQUENCE_FIELD, entry.lastSeq())
                    .set(LAST_OFFSET_FIELD, entry.lastDataOffset())
                    .set(OFFSET_DELTA_FIELD, offsetDelta) // AutoMQ inject
                    .set(TIMESTAMP_FIELD, entry.lastTimestamp())
                    .set(COORDINATOR_EPOCH_FIELD, entry.coordinatorEpoch())
                    .set(CURRENT_TXN_FIRST_OFFSET_FIELD, entry.currentTxnFirstOffset().orElse(-1L));
            structEntries[i++] = producerEntryStruct;
        }
        struct.set(PRODUCER_ENTRIES_FIELD, structEntries);

        ByteBuffer buffer = ByteBuffer.allocate(struct.sizeOf());
        struct.writeTo(buffer);
        buffer.flip();

        // now fill in the CRC
        long crc = Crc32C.compute(buffer, PRODUCER_ENTRIES_OFFSET, buffer.limit() - PRODUCER_ENTRIES_OFFSET);
        ByteUtils.writeUnsignedInt(buffer, CRC_OFFSET, crc);

        // AutoMQ inject start
        if (file != null) {
            try (FileChannel fileChannel = FileChannel.open(file.toPath(), StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
                fileChannel.write(buffer.duplicate());
                if (sync) {
                    fileChannel.force(true);
                }
            }
        }
        return buffer;
        // AutoMQ inject end
    }
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