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

Code before:
                            tmpResourcesRegistry);
            LOG.trace("Bytes-copy file to checkpoint: {}, {}", dbFilePath, targetStateHandle);
        }

        return HandleAndLocalPath.of(targetStateHandle, dbFilePath.getName());
    }

    /**
     * Duplicate file to checkpoint storage by calling {@link CheckpointStreamFactory#duplicate} if
     * 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,
            long maxTransferBytes) {

        try {
            if (sourceHandle.getStateSize() > maxTransferBytes) {
                return null;
            }

            // 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,

Review comment (suggestion):
Is there any chance the `sourceHandle.getStateSize() <= maxTransferBytes` here, but dont match when copying...

Code after:
                            tmpResourcesRegistry);
            LOG.trace("Bytes-copy file to checkpoint: {}, {}", dbFilePath, targetStateHandle);
        }

        return HandleAndLocalPath.of(targetStateHandle, dbFilePath.getName());
    }

    /**
     * Duplicate file to checkpoint storage by calling {@link CheckpointStreamFactory#duplicate} if
     * possible.
     *
     * @param sourceHandle The source state handle
     * @param checkpointStreamFactory The checkpoint stream factory
     * @param stateScope The state scope
     * @param tmpResourcesRegistry The temporary resources registry
     * @param maxTransferBytes The max transfer bytes
     * @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,
            long maxTransferBytes) {

        try {
            // skip if there is a limit of transfer bytes
            if (maxTransferBytes > 0 && maxTransferBytes != Long.MAX_VALUE) {
                return null;
            }

            // 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}. */