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

Code before:
         * files.
         *
         * @param currentUploadedSstFiles the sst files uploaded in the following checkpoint.
         */
        private Map<String, StreamStateHandle> pruneFirstCheckpointSstFiles(
                @Nonnull SortedMap<Long, Collection<HandleAndLocalPath>> currentUploadedSstFiles) {
            Map<String, StreamStateHandle> prunedSstFiles = null;
            for (Map.Entry<Long, Collection<HandleAndLocalPath>> entry :
                    currentUploadedSstFiles.entrySet()) {
                // Iterate checkpoints in ascending order of checkpoint id.
                if (prunedSstFiles == null) {
                    // The first checkpoint's uploaded sst files are all included.
                    prunedSstFiles =
                            entry.getValue().stream()
                                    .collect(
                                            Collectors.toMap(
                                                    HandleAndLocalPath::getLocalPath,
                                                    HandleAndLocalPath::getHandle));
                } else if (!prunedSstFiles.isEmpty()) {
                    // Prune sst files which have been re-uploaded in the following checkpoints.
                    for (HandleAndLocalPath handleAndLocalPath : entry.getValue()) {
                        if (!(handleAndLocalPath.getHandle()
                                instanceof PlaceholderStreamStateHandle)) {
                            // If it's not a placeholder handle, it means the sst file has been
                            // re-uploaded in the following checkpoint.
                            prunedSstFiles.remove(handleAndLocalPath.getLocalPath());
                        }
                    }
                }
            }
            return prunedSstFiles != null
                    ? Collections.unmodifiableMap(prunedSstFiles)
                    : Collections.emptyMap();
        }

        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();
            }

Review comment (bug):
NIT: log the number of removed files if it's > 0 (on DEBUG level)?

Code after:
                            ? 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) {
                    // The first checkpoint's uploaded sst files are all included.
                    prunedSstFiles =
                            entry.getValue().stream()
                                    .collect(
                                            Collectors.toMap(
                                                    HandleAndLocalPath::getLocalPath,
                                                    HandleAndLocalPath::getHandle));
                } else if (prunedSstFiles == null) {
                    // The last completed checkpoint's uploaded sst files are not existed.
                    // So we skip the pruning process.
                    break;
                } else if (!prunedSstFiles.isEmpty()) {
                    // Prune sst files which have been re-uploaded in the following checkpoints.
                    for (HandleAndLocalPath handleAndLocalPath : entry.getValue()) {
                        if (!(handleAndLocalPath.getHandle()
                                instanceof PlaceholderStreamStateHandle)) {
                            // If it's not a placeholder handle, it means the sst file has been
                            // re-uploaded in the following checkpoint.
                            if (prunedSstFiles.remove(handleAndLocalPath.getLocalPath()) != null) {
                                removedCount++;
                            }
                        }
                    }
                }
            }
            if (removedCount > 0 && LOG.isTraceEnabled()) {
                LOG.trace(
                        "Removed {} re-uploaded sst files from base file set for incremental "
                                + "checkpoint. Base checkpoint id: {}",