{"record":{"id":"d15f4017d73872fb","repo":"apache/flink","slug":"bulk-copy-interrupted","errorCode":null,"errorMessage":"Bulk copy interrupted","messagePattern":"Bulk copy interrupted","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"warning","filePath":"flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java","lineNumber":370,"sourceCode":"            // Fail fast: complete as soon as either all downloads finish successfully or the first\n            // one fails, rather than waiting for every in-flight download to run to completion. The\n            // outer copyFiles handler aborts the remaining streams and shuts the pool down on the\n            // resulting exception.\n            CompletableFuture<Void> allDone =\n                    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));\n            CompletableFuture<Void> firstFailure = new CompletableFuture<>();\n            for (CompletableFuture<Void> future : futures) {\n                future.whenComplete(\n                        (ignored, error) -> {\n                            if (error != null) {\n                                firstFailure.completeExceptionally(error);\n                            }\n                        });\n            }\n            CompletableFuture.anyOf(allDone, firstFailure).get();\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n            throw new IOException(\"Bulk copy interrupted\", e);\n        } catch (ExecutionException e) {\n            Throwable cause = e.getCause();\n            ExceptionUtils.rethrowIfFatalError(cause);\n            if (isConnectionPoolExhausted(cause)) {\n                throw new IOException(\n                        String.format(\n                                \"S3 connection pool exhausted during bulk copy. \"\n                                        + \"The configured connection pool size (%d) could not serve \"\n                                        + \"the concurrent download requests (%d). \"\n                                        + \"Consider reducing '%s' or increasing '%s'.\",\n                                maxConnections,\n                                maxConcurrentCopies,\n                                NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT.key(),\n                                NativeS3FileSystemFactory.MAX_CONNECTIONS.key()),\n                        cause);\n            }\n            throw new IOException(\"Bulk copy failed\", cause);\n        }","sourceCodeStart":352,"sourceCodeEnd":388,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java#L352-L388","documentation":"NativeS3BulkCopyHelper wraps InterruptedException from waiting on the batch of copy futures as an IOException('Bulk copy interrupted') after restoring the thread's interrupt flag. It means the copying thread was interrupted (e.g. task cancellation) rather than an S3 failure.","triggerScenarios":"The thread executing copyFiles is interrupted while blocked in CompletableFuture.anyOf(...).get() — typically job/task cancellation, operator close during a bulk download, or an explicit thread interrupt from shutdown logic.","commonSituations":"Cancelling a Flink job or stopping a task while a bulk copy is in flight; test harnesses that interrupt worker threads; shutdown hooks racing with active downloads.","solutions":["Treat this as a cancellation signal: stop submitting further batches and let the closeableRegistry/cancellation mechanism abort in-flight downloads.","If it appears without an intentional cancel, find who interrupts the thread (task cancellation, executor shutdownNow) and sequence shutdown after copies complete.","Do not retry blindly; re-run the bulk copy only after the interruption source is resolved and clean up partially written local files first."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    s3Fs.copyFiles(requests, registry);\n} catch (IOException e) {\n    if (e.getCause() instanceof InterruptedException) {\n        Thread.currentThread().interrupt();\n        // cancellation path: stop cleanly, do not retry\n        return;\n    }\n    throw e;\n}","preventionTips":["Treat 'Bulk copy interrupted' as cancellation, not failure — check whether the job was cancelled before debugging S3.","Sequence executor shutdown after bulk copies complete to avoid spurious interrupts.","Clean partially written local destination files before any re-run."],"tags":["s3","bulk-copy","interruption","cancellation"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}