apache/flink · error · IOException

Bulk copy failed

Error message

Bulk copy failed

What it means

Generic wrapper IOException thrown by NativeS3BulkCopyHelper when the bulk copy batch failed with a non-interruption, non-pool-exhaustion error. The real cause (S3Exception, NoSuchKey, credentials error, disk full, etc.) is attached as the cause and should be inspected first.

Solutions

1. Inspect getCause() of the IOException to identify the underlying failure; the message 'Bulk copy failed' alone is intentionally generic.
2. Fix the root cause: restore access to the object, grant permissions, free disk space, or re-create the client.
3. Re-run the copy; it is idempotent per file if destinations are overwritten.

try {
    s3Fs.copyFiles(requests, registry);
} catch (IOException e) {
    Throwable cause = e.getCause();
    if (cause instanceof NoSuchKeyException) {
        // source object vanished: log and skip / re-list
    } else if (cause instanceof S3Exception && ((S3Exception) cause).statusCode() >= 500) {
        // transient: retry with backoff after cleaning partial local files
    } else {
        throw e;
    }
}

When it happens

Trigger: Any ExecutionException from a copy future that is not connection-pool exhaustion: 404 NoSuchKey on a source object, 403 access denied, local disk I/O failure writing the destination, SDK client shutdown, or network errors.

Common situations: Source objects deleted between listing and download (race with compaction/cleanup), missing s3:GetObject permissions, local tmp directory full or read-only, or the S3 client closed while copies were pending.

Related errors

• Bulk copy interrupted
• Bulk copy not enabled. Set s3.bulk-copy.enabled=true
• Can't deserialize Debezium Avro message.
• Could not serialize row
• Only S3 to local copies are currently supported

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/8442c592ac822df3. Report an issue: GitHub.

Appendix: source

Thrown at flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:387

            Thread.currentThread().interrupt();
            throw new IOException("Bulk copy interrupted", e);
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            ExceptionUtils.rethrowIfFatalError(cause);
            if (isConnectionPoolExhausted(cause)) {
                throw new IOException(
                        String.format(
                                "S3 connection pool exhausted during bulk copy. "
                                        + "The configured connection pool size (%d) could not serve "
                                        + "the concurrent download requests (%d). "
                                        + "Consider reducing '%s' or increasing '%s'.",
                                maxConnections,
                                maxConcurrentCopies,
                                NativeS3FileSystemFactory.BULK_COPY_MAX_CONCURRENT.key(),
                                NativeS3FileSystemFactory.MAX_CONNECTIONS.key()),
                        cause);
            }
            throw new IOException("Bulk copy failed", cause);
        }
    }

    static boolean isSupportedS3Scheme(org.apache.flink.core.fs.Path path) {
        String scheme = path.toUri().getScheme();
        return "s3".equalsIgnoreCase(scheme) || "s3a".equalsIgnoreCase(scheme);
    }

    static boolean isSupportedLocalScheme(org.apache.flink.core.fs.Path path) {
        String scheme = path.toUri().getScheme();
        return scheme == null || "file".equalsIgnoreCase(scheme);
    }

    private static void abortAndClose(ResponseInputStream<GetObjectResponse> stream) {
        try {
            stream.abort();
        } catch (RuntimeException e) {
            LOG.debug("Error aborting S3 response stream during bulk-copy cancellation", e);

View on GitHub (pinned to 2f3c205e92)