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

Code before:
     * possible.
     *
     * @param sourceHandle The source state handle
     * @param checkpointStreamFactory The checkpoint stream factory
     * @param stateScope The state scope
     * @return The target state handle if path-copying is successful, otherwise null
     */
    private @Nullable StreamStateHandle tryPathCopyingToCheckpoint(
            @Nonnull StreamStateHandle sourceHandle,
            CheckpointStreamFactory checkpointStreamFactory,
            CheckpointedStateScope stateScope,
            CloseableRegistry tmpResourcesRegistry) {

        try {
            if (!checkpointStreamFactory.canFastDuplicate(sourceHandle, stateScope)) {
                return null;
            }

            List<StreamStateHandle> result =
                    checkpointStreamFactory.duplicate(
                            Collections.singletonList(sourceHandle), stateScope);
            StreamStateHandle resultStateHandle = result.get(0);
            tmpResourcesRegistry.registerCloseable(
                    () -> StateUtil.discardStateObjectQuietly(resultStateHandle));
            return resultStateHandle;
        } catch (Exception e) {
            LOG.warn("Failed to duplicate file to checkpoint: {} {}", sourceHandle, stateScope, e);
        }
        return null;
    }

    /** Write file to checkpoint storage through {@link CheckpointStateOutputStream}. */
    private @Nullable StreamStateHandle bytesCopyingToCheckpoint(
            Path filePath,
            long maxTransferBytes,
            CheckpointStreamFactory checkpointStreamFactory,
            CheckpointedStateScope stateScope,
            CloseableRegistry closeableRegistry,
            CloseableRegistry tmpResourcesRegistry)
            throws IOException {
        InputStream inputStream = null;
        CheckpointStateOutputStream outputStream = null;

        try {
            final byte[] buffer = new byte[READ_BUFFER_SIZE];

            inputStream = dbFileSystem.open(filePath, READ_BUFFER_SIZE);
            closeableRegistry.registerCloseable(inputStream);

            outputStream = checkpointStreamFactory.createCheckpointStateOutputStream(stateScope);
            closeableRegistry.registerCloseable(outputStream);

Review comment (bug):
I think it would be useful to add comments / javadoc  around this logic. It is not obvious to me why we are discarding state quietly in a copy method - some comments to detail the thinking here would be good. 

I think we are copying information for a failure case. It it worth adding at least a debug around this cleanup. 

The Jira says : clean up half-uploaded checkpoints. How do we get half up loaded checkpoints. Is there a case to fix this at source to prevent these half uploaded files from occurring in the first place. 

Code after:
     * possible.
     *
     * @param sourceHandle The source state handle
     * @param checkpointStreamFactory The checkpoint stream factory
     * @param stateScope The state scope
     * @param tmpResourcesRegistry The temporary resources registry
     * @return The target state handle if path-copying is successful, otherwise null
     */
    private @Nullable StreamStateHandle tryPathCopyingToCheckpoint(
            @Nonnull StreamStateHandle sourceHandle,
            CheckpointStreamFactory checkpointStreamFactory,
            CheckpointedStateScope stateScope,
            CloseableRegistry tmpResourcesRegistry) {

        try {
            // copy the file by duplicating
            if (!checkpointStreamFactory.canFastDuplicate(sourceHandle, stateScope)) {
                return null;
            }
            List<StreamStateHandle> result =
                    checkpointStreamFactory.duplicate(
                            Collections.singletonList(sourceHandle), stateScope);
            StreamStateHandle resultStateHandle = result.get(0);

            // register the clean-up logic of the uploaded file
            tmpResourcesRegistry.registerCloseable(
                    () -> StateUtil.discardStateObjectQuietly(resultStateHandle));
            return resultStateHandle;
        } catch (Exception e) {
            LOG.warn("Failed to duplicate file to checkpoint: {} {}", sourceHandle, stateScope, e);
        }
        return null;
    }

    /** Write file to checkpoint storage through {@link CheckpointStateOutputStream}. */
    private @Nullable StreamStateHandle bytesCopyingToCheckpoint(
            Path filePath,
            long maxTransferBytes,
            CheckpointStreamFactory checkpointStreamFactory,
            CheckpointedStateScope stateScope,
            CloseableRegistry closeableRegistry,
            CloseableRegistry tmpResourcesRegistry)
            throws IOException {
        // copy the file by bytes
        InputStream inputStream = null;
        CheckpointStateOutputStream outputStream = null;

        try {
            final byte[] buffer = new byte[READ_BUFFER_SIZE];

            inputStream = dbFileSystem.open(filePath, READ_BUFFER_SIZE);