Samrat002 commented on code in PR #28136:
URL: https://github.com/apache/flink/pull/28136#discussion_r3551495716
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3BulkCopyHelper.java:
##########
@@ -166,69 +202,174 @@ public void copyFiles(
}
LOG.info("Completed bulk copy of {} files", totalFiles);
- } catch (Exception e) {
- if (!copyFutures.isEmpty()) {
- LOG.warn(
- "Error during bulk copy, waiting for {} in-flight
operations to complete",
- copyFutures.size());
- try {
- waitForCopies(copyFutures);
- } catch (IOException waitError) {
- LOG.warn(
- "Error waiting for in-flight copy operations: {}",
- waitError.getMessage());
- e.addSuppressed(waitError);
- }
- }
+ } catch (Throwable e) {
+ cancellation.close();
if (e instanceof IOException) {
throw (IOException) e;
+ } else if (e instanceof Error) {
+ throw (Error) e;
} else {
throw new IOException(e);
}
Review Comment:
This is exactly the same as
`org.apache.flink.util.ExceptionUtils.rethrowIOException` which already exists.
i have reused it.
```
/**
* Re-throws the given {@code Throwable} in scenarios where the
signatures allows only
* IOExceptions (and RuntimeException and Error).
*
* <p>Throws this exception directly, if it is an IOException, a
RuntimeException, or an Error.
* Otherwise it wraps it in an IOException and throws it.
*
* @param t The Throwable to be thrown.
*/
public static void rethrowIOException(Throwable t) throws IOException {
if (t instanceof IOException) {
throw (IOException) t;
} else if (t instanceof RuntimeException) {
throw (RuntimeException) t;
} else if (t instanceof Error) {
throw (Error) t;
} else {
throw new IOException(t.getMessage(), t);
}
}
```
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystem.java:
##########
@@ -512,7 +512,17 @@ public String generateEntropy() {
@Override
public boolean canCopyPaths(Path source, Path destination) {
- return bulkCopyHelper != null;
+ return bulkCopyHelper != null && isS3Path(source) &&
isLocalPath(destination);
+ }
+
+ private static boolean isS3Path(Path path) {
+ final String scheme = path.toUri().getScheme();
+ return "s3".equalsIgnoreCase(scheme) || "s3a".equalsIgnoreCase(scheme);
+ }
+
+ private static boolean isLocalPath(Path path) {
Review Comment:
removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
