private CompletableFuture<Void> writeSnapshot(long offset, Map<Long, ProducerStateEntry> entries) {
        try {
            ByteBuffer buffer = ProducerStateManager.writeSnapshot(null, entries, false);
            byte[] rawSnapshot = new byte[buffer.remaining()];
            buffer.get(rawSnapshot);

            snapshotsMap.put(offset, ByteBuffer.wrap(rawSnapshot));
            ElasticPartitionProducerSnapshotsMeta meta = new ElasticPartitionProducerSnapshotsMeta(snapshotsMap);
            return persistSnapshots.persist(MetaKeyValue.of(MetaStream.PRODUCER_SNAPSHOTS_META_KEY, meta.encode()));
        } catch (IOException e) {
            return CompletableFuture.failedFuture(e);
        }
    }
Only keep the last snapshot which offset less than or equals to the recoveryPointCheckpoint
core/src/main/scala/kafka/log/streamaspect/ElasticProducerStateManager.java