Store the gcs path into GcsWritableByteChannel.

Author: shunping

sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java

@@ -103,7 +103,7 @@ public GcsUtilV2 create(PipelineOptions options) {
     "nullness" // For Creating AccessDeniedException FileNotFoundException, and
     // FileAlreadyExistsException with null.
   })
-  private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+  private static IOException translateStorageException(GcsPath gcsPath, StorageException e) {
     switch (e.getCode()) {
       case 403:
         return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
@@ -568,18 +568,19 @@ public SeekableByteChannel open(GcsPath path, BlobSourceOption... sourceOptions)
   /** A bridge that allows a GCS WriteChannel to behave as a WritableByteChannel. */
   private static class GcsWritableByteChannel implements WritableByteChannel {
     private final WriteChannel writer;
+    private final GcsPath gcsPath;
 
-    GcsWritableByteChannel(WriteChannel writer) {
+    GcsWritableByteChannel(WriteChannel writer, GcsPath gcsPath) {
       this.writer = writer;
+      this.gcsPath = gcsPath;
     }
 
     @Override
     public int write(ByteBuffer src) throws IOException {
       try {
         return writer.write(src);
       } catch (StorageException e) {
-        // In a real implementation, you'd use your translateStorageException here
-        throw new IOException(e);
+        throw translateStorageException(gcsPath, e);
       }
     }
 
@@ -623,7 +624,7 @@ public WritableByteChannel create(
       }
 
       // Return the bridge wrapper
-      return new GcsWritableByteChannel(writer);
+      return new GcsWritableByteChannel(writer, path);
 
     } catch (StorageException e) {
       throw translateStorageException(path, e);

sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilParameterizedIT.java

@@ -659,11 +659,13 @@ public void testWriteAndRead() throws IOException {
     try {
       createTestBucketHelper(bucketName, false);
 
+      // Write content to a GCS file
       CreateOptions options = CreateOptions.builder().setExpectFileToNotExist(true).build();
       try (WritableByteChannel writer = gcsUtil.create(targetPath, options)) {
         writer.write(ByteBuffer.wrap(content.getBytes(StandardCharsets.UTF_8)));
       }
 
+      // Read content into a buffer
       StringBuilder readContent = new StringBuilder();
       try (ReadableByteChannel reader = gcsUtil.open(targetPath)) {
         ByteBuffer buffer = ByteBuffer.allocate(1024);
@@ -673,6 +675,8 @@ public void testWriteAndRead() throws IOException {
           buffer.clear();
         }
       }
+
+      // Verify content
       assertEquals(content, readContent.toString());
     } finally {
       tearDownTestBucketHelper(bucketName);