Repository: apache/flink
File: flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/snapshot/ForStSnapshotStrategyBase.java (Java)

Code before:
        protected PreviousSnapshot(@Nullable Collection<HandleAndLocalPath> confirmedSstFiles) {
            this.confirmedSstFiles =
                    confirmedSstFiles != null
                            ? confirmedSstFiles.stream()
                                    .collect(
                                            Collectors.toMap(
                                                    HandleAndLocalPath::getLocalPath,
                                                    HandleAndLocalPath::getHandle))
                            : Collections.emptyMap();
        }

        /**
         * Remove the sst files which have been re-uploaded in the following checkpoint from the
         * confirmed sst files.
         *
         * @param followingUploadedSstFiles the sst files uploaded in the following checkpoint.
         */
        protected void removeReUploadedConfirmedSstFiles(
                @Nonnull Collection<HandleAndLocalPath> followingUploadedSstFiles) {
            if (!confirmedSstFiles.isEmpty()) {
                followingUploadedSstFiles.forEach(
                        e -> {
                            if (!(e.getHandle() instanceof PlaceholderStreamStateHandle)) {
                                // If it's not a placeholder handle, it means the sst file has been
                                // re-uploaded in the following checkpoint.
                                confirmedSstFiles.remove(e.getLocalPath());
                            }
                        });
            }
        }

        protected Optional<StreamStateHandle> getUploaded(String filename) {
            if (confirmedSstFiles.containsKey(filename)) {
                StreamStateHandle handle = confirmedSstFiles.get(filename);
                // We introduce a placeholder state handle to reduce network transfer overhead,
                // it will be replaced by the original handle from the shared state registry
                // (created from a previous checkpoint).
                return Optional.of(
                        new PlaceholderStreamStateHandle(
                                handle.getStreamStateHandleID(),
                                handle.getStateSize(),
                                FileMergingSnapshotManager.isFileMergingHandle(handle)));
            } else {
                // Don't use any uploaded but not confirmed handles because they might be deleted
                // (by TM) if the previous checkpoint failed. See FLINK-25395
                return Optional.empty();
            }
        }

        protected boolean isEmpty() {
            return confirmedSstFiles.isEmpty();

Review comment (suggestion):
Can't `confirmedSstFiles` be unmodifiable?

In `uploadSnapshotFiles()`, `uploadedSstFiles` is updated with
```
uploadedSstFiles.put(
        checkpointId, Collections.unmodifiableList(sstFiles));
```
and then we use `uploadedSstFiles` to populate `confirmedSstFiles`

Code after:
    /** Previous snapshot with uploaded sst files. */
    protected static class PreviousSnapshot {

        @Nonnull private final Map<String, StreamStateHandle> confirmedSstFiles;

        /**
         * Constructor of PreviousSnapshot. Giving a map of uploaded sst files in previous
         * checkpoints, prune the sst files which have been re-uploaded in the following
         * checkpoints. The prune logic is used to resolve the mismatch between TM and JM due to
         * notification delay. Following steps for example:
         *
         * <ul>
         *   <li>1) checkpoint 1 uses file 00001.SST uploaded as xxx.sst.
         *   <li>2) checkpoint 2 uses the same file 00001.SST but re-uploads it as yyy.sst because
         *       CP 1 wasn't yet confirmed.
         *   <li>3) TM get a confirmation of checkpoint 1.
         *   <li>4) JM completes checkpoint 2 and subsumes checkpoint 1 - removing xxx.sst.
         *   <li>5) checkpoint 3 tries to re-use file 00001.SST uploaded as xxx.sst in checkpoint 1,
         *       but it was deleted in (4) by JM.
         * </ul>
         *
         * @param currentUploadedSstFiles the sst files uploaded in previous checkpoints.
         * @param lastCompletedCheckpoint the last completed checkpoint id.
         */
        protected PreviousSnapshot(
                @Nullable SortedMap<Long, Collection<HandleAndLocalPath>> currentUploadedSstFiles,
                long lastCompletedCheckpoint) {
            this.confirmedSstFiles =
                    currentUploadedSstFiles != null
                            ? pruneFirstCheckpointSstFiles(
                                    currentUploadedSstFiles, lastCompletedCheckpoint)
                            : Collections.emptyMap();
        }

        /**
         * The last completed checkpoint's uploaded sst files are all included, then for each
         * following checkpoint, if a sst file has been re-uploaded, remove it from the first
         * checkpoint's sst files.
         *
         * @param currentUploadedSstFiles the sst files uploaded in the following checkpoint.
         * @param lastCompletedCheckpoint the last completed checkpoint id.
         */
        private Map<String, StreamStateHandle> pruneFirstCheckpointSstFiles(
                @Nonnull SortedMap<Long, Collection<HandleAndLocalPath>> currentUploadedSstFiles,
                long lastCompletedCheckpoint) {
            Map<String, StreamStateHandle> prunedSstFiles = null;
            int removedCount = 0;
            for (Map.Entry<Long, Collection<HandleAndLocalPath>> entry :
                    currentUploadedSstFiles.entrySet()) {
                // Iterate checkpoints in ascending order of checkpoint id.
                if (entry.getKey() == lastCompletedCheckpoint) {