public static List<SnapshotFile> listSnapshotFiles(File dir) throws IOException {
        if (dir.exists() && dir.isDirectory()) {
            try (Stream<Path> paths = Files.list(dir.toPath())) {
                return paths.filter(ProducerStateManager::isSnapshotFile)
                        .map(path -> new SnapshotFile(path.toFile())).collect(Collectors.toList());
            }
        } else {
            return Collections.emptyList();
        }
    }
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