Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java (Java)

Code before:
                        checkpointId,
                        directoryHandle.getDirectory());
                refCheckpointIds.add(checkpointId);
            }
        }

        void decreaseRefCountWhenCheckpointAbort(long checkpointId) {
            if (tracking) {
                LOG.debug(
                        "checkpoint:{} aborted, decrease ref-count to file-merging managed shared dir : {}",
                        checkpointId,
                        directoryHandle.getDirectory());
                refCheckpointIds.remove(checkpointId);
            }
        }

        void handoverOwnershipWhenCheckpointComplete(long checkpointId) {
            if (tracking) {
                LOG.debug(
                        "checkpoint:{} complete, handover ownership of file-merging managed shared dir to JobManager : {}",
                        checkpointId,
                        directoryHandle.getDirectory());
                tracking = false;
                refCheckpointIds.clear();
            }
        }

        void handoverOwnershipWhenCheckpointSubsumed(long checkpointId) {
            if (tracking) {
                LOG.debug(
                        "checkpoint:{} subsumed, handover ownership of file-merging managed shared dir to JobManager : {}",
                        checkpointId,
                        directoryHandle.getDirectory());
                tracking = false;
                refCheckpointIds.clear();
            }
        }

        void tryCleanupQuietly() {
            if (tracking && refCheckpointIds.isEmpty() && directoryHandle != null) {
                try {
                    directoryHandle.discardState();
                } catch (Exception e) {
                    // ignore
                }
            }
        }
    }
}


Review comment (suggestion):
I am curious - we come into a method with one checkpointId and we are clearing the set in 2 places, could you explain why we need to do this and not just remove the id we are called with; maybe add a comment with the explanation.

Code after:
                LOG.debug(
                        "checkpoint:{} start, add reference to file-merging managed shared dir : {}",
                        checkpointId,
                        directoryHandle.getDirectory());
                refCheckpointIds.add(checkpointId);
            }
        }

        void removeReferenceWhenCheckpointAbort(long checkpointId) {
            if (tracking) {
                LOG.debug(
                        "checkpoint:{} aborted, remove reference to file-merging managed shared dir : {}",
                        checkpointId,
                        directoryHandle.getDirectory());
                refCheckpointIds.remove(checkpointId);
            }
        }

        void handoverOwnershipWhenCheckpointComplete(long checkpointId) {
            if (tracking) {
                LOG.debug(
                        "checkpoint:{} complete, handover ownership of file-merging managed shared dir to JobManager : {}",
                        checkpointId,
                        directoryHandle.getDirectory());
                tracking = false;
                refCheckpointIds.clear();
            }
        }

        void handoverOwnershipWhenCheckpointSubsumed(long checkpointId) {
            if (tracking) {
                LOG.debug(
                        "checkpoint:{} subsumed, handover ownership of file-merging managed shared dir to JobManager : {}",
                        checkpointId,
                        directoryHandle.getDirectory());
                tracking = false;
                refCheckpointIds.clear();
            }
        }

        void tryCleanupQuietly() {
            if (tracking && refCheckpointIds.isEmpty() && directoryHandle != null) {
                try {
                    directoryHandle.discardState();
                } catch (Exception e) {
                    // ignore
                }
            }
        }
    }
}