public CompletableFuture<Void> takeSnapshotAndRemoveExpired(long recoveryPointCheckpoint) {
        Long lastSnapshotOffset = snapshotsMap.floorKey(recoveryPointCheckpoint);
        if (lastSnapshotOffset != null) {
            List<Long> expiredSnapshotOffsets = new ArrayList<>(snapshotsMap.headMap(lastSnapshotOffset, false).keySet());
            expiredSnapshotOffsets.forEach(offset -> {
                snapshotsMap.remove(offset);
                snapshots.remove(offset);
            });
        }
        return takeSnapshot0();
    }
Only keep the last snapshot which offset less than or equals to the recoveryPointCheckpoint
core/src/main/scala/kafka/log/streamaspect/ElasticProducerStateManager.java