public void deleteSnapshotsBefore(long offset) throws IOException {
        for (SnapshotFile snapshot : snapshots.subMap(0L, offset).values()) {
            removeAndDeleteSnapshot(snapshot.offset);
        }
    }
Deletes the producer snapshot files until the given offset (exclusive) in a thread safe manner.

@param offset offset number
@throws IOException if any IOException occurs while deleting the files.
storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java