shunping commented on code in PR #40677:
URL: https://github.com/apache/beam/pull/40677#discussion_r4332055076
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -994,35 +1018,73 @@ public SeekableByteChannel open(GcsPath path,
BlobSourceOption... sourceOptions)
}
}
- /** A bridge that allows a GCS WriteChannel to behave as a
WritableByteChannel. */
+ /**
+ * A bridge that allows a GCS WriteChannel to behave as a
WritableByteChannel.
+ *
+ * <p>The WriteChannel is opened in the background, because java-storage
starts the resumable
+ * upload session with a blocking request when it opens one. Waiting for it
on the first write, or
+ * on close, rather than in {@link #create} lets that request overlap with
the caller's own work,
+ * as gcsio does for {@code GcsUtilV1}. A failure to start the session
surfaces there instead.
+ */
private static class GcsWritableByteChannel implements WritableByteChannel {
- private final WriteChannel writer;
+ private final Future<WriteChannel> pendingWriter;
private final GcsPath gcsPath;
+ private boolean closed = false;
- GcsWritableByteChannel(WriteChannel writer, GcsPath gcsPath) {
- this.writer = writer;
+ GcsWritableByteChannel(Future<WriteChannel> pendingWriter, GcsPath
gcsPath) {
+ this.pendingWriter = pendingWriter;
this.gcsPath = gcsPath;
}
+ /** Waits for the upload session to start and returns its channel. */
+ private WriteChannel writer() throws IOException {
+ try {
+ return pendingWriter.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ pendingWriter.cancel(true);
+ throw new InterruptedIOException("Interrupted while starting the
upload to " + gcsPath);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof StorageException) {
+ throw translateStorageException(gcsPath, (StorageException) cause);
+ }
+ if (cause instanceof IOException) {
+ throw (IOException) cause;
+ }
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new IOException("Failed to start the upload to " + gcsPath,
cause);
+ }
+ }
+
@Override
public int write(ByteBuffer src) throws IOException {
try {
- return writer.write(src);
+ return writer().write(src);
} catch (StorageException e) {
throw translateStorageException(gcsPath, e);
}
}
@Override
public boolean isOpen() {
- return writer.isOpen();
+ return !closed;
}
@Override
public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
// The upload is finalized here, so this is where a failed precondition
surfaces.
try {
- writer.close();
+ writer().close();
} catch (StorageException e) {
throw translateStorageException(gcsPath, e);
}
Review Comment:
> If write() is interrupted while waiting in writer(), writer() calls
pendingWriter.cancel(true) and throws InterruptedIOException. When
try-with-resources then calls channel.close(), close() calls writer().close()
-> pendingWriter.get(), which throws an unchecked CancellationException.
> If pendingWriter fails with a StorageException (e.g. 403 Forbidden) during
write(), write() throws AccessDeniedException, and then try-with-resources
calls close(), which calls writer().close() and throws a second
AccessDeniedException.
Added `sessionFailed` in write() and it was checked in closed(), so not
calling writer().close() again if the session is failed (interrupted, canceled,
failed).
--
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]
