gaborgsomogyi commented on code in PR #28136:
URL: https://github.com/apache/flink/pull/28136#discussion_r3550161153
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3ObjectOperations.java:
##########
@@ -380,15 +397,49 @@ public boolean deleteObject(String key) throws
IOException {
}
public long getObject(String key, File targetLocation) throws IOException {
+ java.nio.file.Path target = targetLocation.toPath().toAbsolutePath();
+ java.nio.file.Path parent = target.getParent();
+ if (parent != null) {
+ Files.createDirectories(parent);
+ }
+ java.nio.file.Path tempTarget =
+ NativeS3FileIoUtils.createTemporaryDownloadFile(parent,
target);
+ ResponseInputStream<GetObjectResponse> responseStream = null;
+ boolean success = false;
try {
GetObjectRequest request =
GetObjectRequest.builder().bucket(bucketName).key(key).build();
- ResponseTransformer<GetObjectResponse, GetObjectResponse>
responseTransformer =
- ResponseTransformer.toFile(targetLocation.toPath());
- s3Client.getObject(request, responseTransformer);
- return Files.size(targetLocation.toPath());
+ responseStream = s3Client.getObject(request);
+ NativeS3FileIoUtils.copyStream(responseStream, tempTarget,
DOWNLOAD_BUFFER_SIZE);
+ NativeS3FileIoUtils.moveFile(tempTarget, target);
+ success = true;
+ return Files.size(target);
} catch (S3Exception e) {
throw new IOException("Failed to get object for key: " + key, e);
+ } finally {
+ if (responseStream != null) {
+ if (!success) {
+ try {
+ responseStream.abort();
+ } catch (RuntimeException e) {
+ LOG.debug("Error aborting S3 response stream for key
{}", key, e);
+ }
+ try {
+ responseStream.close();
+ } catch (IOException e) {
+ LOG.debug("Error closing S3 response stream for key
{}", key, e);
+ }
Review Comment:
Maybe `abortAndClose`?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3ObjectOperations.java:
##########
@@ -380,15 +397,49 @@ public boolean deleteObject(String key) throws
IOException {
}
public long getObject(String key, File targetLocation) throws IOException {
+ java.nio.file.Path target = targetLocation.toPath().toAbsolutePath();
+ java.nio.file.Path parent = target.getParent();
+ if (parent != null) {
+ Files.createDirectories(parent);
+ }
+ java.nio.file.Path tempTarget =
+ NativeS3FileIoUtils.createTemporaryDownloadFile(parent,
target);
+ ResponseInputStream<GetObjectResponse> responseStream = null;
+ boolean success = false;
try {
GetObjectRequest request =
GetObjectRequest.builder().bucket(bucketName).key(key).build();
- ResponseTransformer<GetObjectResponse, GetObjectResponse>
responseTransformer =
- ResponseTransformer.toFile(targetLocation.toPath());
- s3Client.getObject(request, responseTransformer);
- return Files.size(targetLocation.toPath());
+ responseStream = s3Client.getObject(request);
+ NativeS3FileIoUtils.copyStream(responseStream, tempTarget,
DOWNLOAD_BUFFER_SIZE);
+ NativeS3FileIoUtils.moveFile(tempTarget, target);
+ success = true;
+ return Files.size(target);
} catch (S3Exception e) {
throw new IOException("Failed to get object for key: " + key, e);
+ } finally {
+ if (responseStream != null) {
+ if (!success) {
+ try {
+ responseStream.abort();
+ } catch (RuntimeException e) {
+ LOG.debug("Error aborting S3 response stream for key
{}", key, e);
+ }
+ try {
+ responseStream.close();
+ } catch (IOException e) {
+ LOG.debug("Error closing S3 response stream for key
{}", key, e);
+ }
+ } else {
+ try {
+ responseStream.close();
+ } catch (IOException e) {
+ LOG.debug("Error closing S3 response stream for key
{}", key, e);
+ }
Review Comment:
Maybe `closeQuietly`?
##########
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:
Same here
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java:
##########
@@ -486,21 +617,26 @@ public FileSystem create(URI fsUri) throws IOException {
.retryMaxBackoff(config.get(RETRY_MAX_BACKOFF))
.credentialsProviderClasses(credentialsProviderClasses)
.encryptionConfig(encryptionConfig)
+ .useCrt(crtEnabled)
+ .crtTargetThroughputGbps(crtTargetThroughputGbps)
+ .crtReadBufferSizeInBytes(
+ crtReadBufferSize == null ? null :
crtReadBufferSize.getBytes())
+ .crtMaxConcurrency(crtMaxConcurrency)
+ .crtMaxNativeMemoryLimitInBytes(
+ crtMaxNativeMemoryLimit == null
+ ? null
+ : crtMaxNativeMemoryLimit.getBytes())
+
.crtMinPartSizeInBytes(config.get(PART_UPLOAD_MIN_SIZE))
Review Comment:
If CRT is not enabled then there is no value to set those, right?
##########
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:
What's the point to have these ifs?
##########
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) {
Review Comment:
We're reimplementing partially `isSupportedS3Scheme` here? Or the other way
around.
--
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]
