Repository: googleapis/java-storage
Pull request #1592 (2022-08-26 17:10:09)

chore: update WriteFlushStrategy to inject bucket id
Update WriteFlushStrategy to inject bucket id into x-goog-request-params upon opening the stream.

Fixes [b/238792234](http://b/238792234)


Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
index 240113b8..442335b6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
@@ -57,7 +57,10 @@ final class GapicUnbufferedWritableByteChannel<
 
     this.writeCtx = new WriteCtx<>(requestFactory);
     this.flusher =
-        flusherFactory.newFlusher(writeCtx.getConfirmedBytes()::addAndGet, resultFuture::set);
+        flusherFactory.newFlusher(
+            requestFactory.bucketName(),
+            writeCtx.getConfirmedBytes()::addAndGet,
+            resultFuture::set);
   }
 
   @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index dcdf2952..63649166 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -26,6 +26,7 @@ import static com.google.common.base.MoreObjects.firstNonNull;
 import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiFuture;
+import com.google.api.core.BetaApi;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.grpc.GrpcStatusCode;
 import com.google.api.gax.paging.AbstractPage;
@@ -36,6 +37,7 @@ import com.google.api.gax.rpc.ApiExceptionFactory;
 import com.google.api.gax.rpc.ApiExceptions;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.UnaryCallable;
+import com.google.api.gax.rpc.UnimplementedException;
 import com.google.cloud.BaseService;
 import com.google.cloud.Policy;
 import com.google.cloud.WriteChannel;
@@ -126,6 +128,7 @@ import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+@BetaApi
 final class GrpcStorageImpl extends BaseService<StorageOptions> implements Storage {
 
   private static final byte[] ZERO_BYTES = new byte[0];
@@ -356,8 +359,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GetBucketRequest.Builder builder =
         GetBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket));
     GetBucketRequest req = opts.getBucketsRequest().apply(builder).build();
-    // TODO(frankyn): Do we care about projection because Apiary uses FULL for projection? Missing
-    // projection=full
     return Retrying.run(
         getOptions(),
         retryAlgorithmManager.getFor(req),
@@ -409,9 +410,13 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .apply(ListBucketsRequest.newBuilder())
             .build();
     ListBucketsPagedResponse call = listBucketsCallable.call(request, grpcCallContext);
-    ListBucketsPage page = call.getPage();
-    return new TransformingPageDecorator<>(
-        page, syntaxDecoders.bucket, getOptions(), retryAlgorithmManager.getFor(request));
+    try {
+      ListBucketsPage page = call.getPage();
+      return new TransformingPageDecorator<>(
+          page, syntaxDecoders.bucket, getOptions(), retryAlgorithmManager.getFor(request));
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
   }
 
   @Override
@@ -424,10 +429,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     ListObjectsRequest.Builder builder =
         ListObjectsRequest.newBuilder().setParent(bucketNameCodec.encode(bucket));
     ListObjectsRequest req = opts.listObjectsRequest().apply(builder).build();
-    ListObjectsPagedResponse call = listObjectsCallable.call(req, grpcCallContext);
-    ListObjectsPage page = call.getPage();
-    return new TransformingPageDecorator<>(
-        page, syntaxDecoders.blob, getOptions(), retryAlgorithmManager.getFor(req));
+    try {
+      ListObjectsPagedResponse call = listObjectsCallable.call(req, grpcCallContext);
+      ListObjectsPage page = call.getPage();
+      return new TransformingPageDecorator<>(
+          page, syntaxDecoders.blob, getOptions(), retryAlgorithmManager.getFor(req));
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
   }
 
   @Override
@@ -489,8 +498,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           () -> storageClient.deleteBucketCallable().call(req, grpcCallContext),
           Decoder.identity());
       return true;
-    } catch (ApiException e) {
-      // TODO: We should throw a StorageException instead of ApiException when making the
+    } catch (StorageException e) {
       return false;
     }
   }
@@ -516,8 +524,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           () -> storageClient.deleteObjectCallable().call(req, grpcCallContext),
           Decoder.identity());
       return true;
-    } catch (ApiException e) {
-      // TODO: We should throw a StorageException instead of ApiException when making the
+    } catch (StorageException e) {
       return false;
     }
   }
@@ -554,7 +561,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public CopyWriter copy(CopyRequest copyRequest) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("copy", CopyRequest.class));
   }
 
   @Override
@@ -570,7 +577,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Channels.newChannel(baos)) {
       ByteStreams.copy(r, w);
-    } catch (IOException e) {
+    } catch (ApiException | IOException e) {
       throw StorageException.coalesce(e);
     }
     return baos.toByteArray();
@@ -607,7 +614,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Files.newByteChannel(path, WRITE_OPS)) {
       ByteStreams.copy(r, w);
-    } catch (IOException e) {
+    } catch (ApiException | IOException e) {
       throw StorageException.coalesce(e);
     }
   }
@@ -620,7 +627,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     try (UnbufferedReadableByteChannel r = session.open();
         WritableByteChannel w = Channels.newChannel(outputStream)) {
       ByteStreams.copy(r, w);
-    } catch (IOException e) {
+    } catch (ApiException | IOException e) {
       throw StorageException.coalesce(e);
     }
   }
@@ -721,132 +728,137 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public List<Blob> get(BlobId... blobIds) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("get", BlobId[].class));
   }
 
   @Override
   public List<Blob> get(Iterable<BlobId> blobIds) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("get", Iterable.class));
   }
 
   @Override
   public List<Blob> update(BlobInfo... blobInfos) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("update", BlobInfo[].class));
   }
 
   @Override
   public List<Blob> update(Iterable<BlobInfo> blobInfos) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("update", Iterable.class));
   }
 
   @Override
   public List<Boolean> delete(BlobId... blobIds) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("delete", BlobId[].class));
   }
 
   @Override
   public List<Boolean> delete(Iterable<BlobId> blobIds) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("delete", Iterable.class));
   }
 
   @Override
   public Acl getAcl(String bucket, Entity entity, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("getAcl", String.class, Entity.class, BucketSourceOption[].class));
   }
 
   @Override
   public Acl getAcl(String bucket, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("getAcl", String.class, Entity.class));
   }
 
   @Override
   public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("deleteAcl", String.class, Entity.class, BucketSourceOption[].class));
   }
 
   @Override
   public boolean deleteAcl(String bucket, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("deleteAcl", String.class, Entity.class));
   }
 
   @Override
   public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("createAcl", String.class, Acl.class, BucketSourceOption[].class));
   }
 
   @Override
   public Acl createAcl(String bucket, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("createAcl", String.class, Acl.class));
   }
 
   @Override
   public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("updateAcl", String.class, Acl.class, BucketSourceOption[].class));
   }
 
   @Override
   public Acl updateAcl(String bucket, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("updateAcl", String.class, Acl.class));
   }
 
   @Override
   public List<Acl> listAcls(String bucket, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("listAcls", String.class, BucketSourceOption[].class));
   }
 
   @Override
   public List<Acl> listAcls(String bucket) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("listAcls", String.class));
   }
 
   @Override
   public Acl getDefaultAcl(String bucket, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("getDefaultAcl", String.class, Entity.class));
   }
 
   @Override
   public boolean deleteDefaultAcl(String bucket, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("deleteDefaultAcl", String.class, Entity.class));
   }
 
   @Override
   public Acl createDefaultAcl(String bucket, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("createDefaultAcl", String.class, Acl.class));
   }
 
   @Override
   public Acl updateDefaultAcl(String bucket, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("updateDefaultAcl", String.class, Acl.class));
   }
 
   @Override
   public List<Acl> listDefaultAcls(String bucket) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("listDefaultAcls", String.class));
   }
 
   @Override
   public Acl getAcl(BlobId blob, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("getAcl", BlobId.class, Entity.class));
   }
 
   @Override
   public boolean deleteAcl(BlobId blob, Entity entity) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("deleteAcl", BlobId.class, Entity.class));
   }
 
   @Override
   public Acl createAcl(BlobId blob, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("createAcl", BlobId.class, Acl.class));
   }
 
   @Override
   public Acl updateAcl(BlobId blob, Acl acl) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("updateAcl", BlobId.class, Acl.class));
   }
 
   @Override
   public List<Acl> listAcls(BlobId blob) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("listAcls", BlobId.class));
   }
 
   @Override
@@ -888,10 +900,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .andThen(opts.listHmacKeysRequest())
             .apply(ListHmacKeysRequest.newBuilder())
             .build();
-    ListHmacKeysPagedResponse call = listHmacKeysCallable.call(request, grpcCallContext);
-    ListHmacKeysPage page = call.getPage();
-    return new TransformingPageDecorator<>(
-        page, codecs.hmacKeyMetadata(), getOptions(), retryAlgorithmManager.getFor(request));
+    try {
+      ListHmacKeysPagedResponse call = listHmacKeysCallable.call(request, grpcCallContext);
+      ListHmacKeysPage page = call.getPage();
+      return new TransformingPageDecorator<>(
+          page, codecs.hmacKeyMetadata(), getOptions(), retryAlgorithmManager.getFor(request));
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
   }
 
   @Override
@@ -955,18 +971,21 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Policy getIamPolicy(String bucket, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("getIamPolicy", String.class, BucketSourceOption[].class));
   }
 
   @Override
   public Policy setIamPolicy(String bucket, Policy policy, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("setIamPolicy", String.class, Policy.class, BucketSourceOption[].class));
   }
 
   @Override
   public List<Boolean> testIamPermissions(
       String bucket, List<String> permissions, BucketSourceOption... options) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("testIamPermissions", String.class, List.class, BucketSourceOption.class));
   }
 
   @Override
@@ -982,22 +1001,23 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Notification createNotification(String bucket, NotificationInfo notificationInfo) {
-    return todo();
+    return throwNotYetImplemented(
+        fmtMethodName("createNotification", String.class, NotificationInfo.class));
   }
 
   @Override
   public Notification getNotification(String bucket, String notificationId) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("getNotification", String.class, String.class));
   }
 
   @Override
   public List<Notification> listNotifications(String bucket) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("listNotifications", String.class));
   }
 
   @Override
   public boolean deleteNotification(String bucket, String notificationId) {
-    return todo();
+    return throwNotYetImplemented(fmtMethodName("deleteNotification", String.class, String.class));
   }
 
   @Override
@@ -1006,9 +1026,12 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   }
 
   private Blob getBlob(ApiFuture<WriteObjectResponse> result) {
-    // TODO: investigate if we need to unnest any exception
-    WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(result);
-    return syntaxDecoders.blob.decode(response.getResource());
+    try {
+      WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(result);
+      return syntaxDecoders.blob.decode(response.getResource());
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
   }
 
   /** Bind some decoders for our "Syntax" classes to this instance of GrpcStorageImpl */
@@ -1145,6 +1168,15 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     throw new UnsupportedOperationException(message);
   }
 
+  private <T> T throwNotYetImplemented(String methodName) {
+    String message =
+        String.format(
+            "%s#%s is not yet implemented for GRPC transport. Please use StorageOptions.http() to construct a compatible instance in the interim.",
+            Storage.class.getName(), methodName);
+    throw new UnimplementedException(
+        message, null, statusCodeFor(StatusCode.Code.UNIMPLEMENTED), false);
+  }
+
   private static String fmtMethodName(String name, Class<?>... args) {
     return name
         + "("
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
index b804ba4c..47996e7b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
@@ -24,6 +24,7 @@ import com.google.storage.v2.StartResumableWriteResponse;
 import com.google.storage.v2.WriteObjectRequest;
 import java.util.Objects;
 import java.util.function.Function;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class ResumableWrite implements WriteObjectRequestBuilderFactory {
 
@@ -55,6 +56,14 @@ final class ResumableWrite implements WriteObjectRequestBuilderFactory {
     return writeRequest.toBuilder();
   }
 
+  @Override
+  public @Nullable String bucketName() {
+    if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
+      return req.getWriteObjectSpec().getResource().getBucket();
+    }
+    return null;
+  }
+
   @Override
   public String toString() {
     return "ResumableWrite{" + "req=" + fmtProto(req) + ", res=" + fmtProto(res) + '}';
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
index 9e55d8cc..2488cfcb 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
@@ -23,6 +23,7 @@ import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.storage.v2.WriteObjectRequest;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class WriteCtx<RequestFactoryT extends WriteObjectRequestBuilderFactory> {
 
@@ -74,10 +75,12 @@ final class WriteCtx<RequestFactoryT extends WriteObjectRequestBuilderFactory> {
         + '}';
   }
 
-  @FunctionalInterface
   interface WriteObjectRequestBuilderFactory {
     WriteObjectRequest.Builder newBuilder();
 
+    @Nullable
+    String bucketName();
+
     static SimpleWriteObjectRequestBuilderFactory simple(WriteObjectRequest req) {
       return new SimpleWriteObjectRequestBuilderFactory(req);
     }
@@ -96,6 +99,14 @@ final class WriteCtx<RequestFactoryT extends WriteObjectRequestBuilderFactory> {
       return req.toBuilder();
     }
 
+    @Override
+    public @Nullable String bucketName() {
+      if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
+        return req.getWriteObjectSpec().getResource().getBucket();
+      }
+      return null;
+    }
+
     @Override
     public String toString() {
       return "SimpleWriteObjectRequestBuilderFactory{" + "req=" + fmtProto(req) + '}';
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
index 8dfe99f2..6da9c5e3 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
@@ -17,9 +17,11 @@
 package com.google.cloud.storage;
 
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import java.util.List;
@@ -51,22 +53,34 @@ final class WriteFlushStrategy {
    */
   static FlusherFactory fsyncEveryFlush(
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
-    return (LongConsumer committedTotalBytesCallback,
+    return (String bucketName,
+        LongConsumer committedTotalBytesCallback,
         Consumer<WriteObjectResponse> onSuccessCallback) ->
-        new FsyncEveryFlusher(write, committedTotalBytesCallback, onSuccessCallback);
+        new FsyncEveryFlusher(write, bucketName, committedTotalBytesCallback, onSuccessCallback);
   }
 
   /**
    * Create a {@link Flusher} which will "fsync" only on {@link Flusher#close(WriteObjectRequest)}.
    * Calls to {@link Flusher#flush(List)} will be sent but not synced.
    *
-   * @see FlusherFactory#newFlusher(LongConsumer, Consumer)
+   * @see FlusherFactory#newFlusher(String, LongConsumer, Consumer)
    */
   static FlusherFactory fsyncOnClose(
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
-    return (LongConsumer committedTotalBytesCallback,
+    return (String bucketName,
+        LongConsumer committedTotalBytesCallback,
         Consumer<WriteObjectResponse> onSuccessCallback) ->
-        new FsyncOnClose(write, committedTotalBytesCallback, onSuccessCallback);
+        new FsyncOnClose(write, bucketName, committedTotalBytesCallback, onSuccessCallback);
+  }
+
+  private static GrpcCallContext contextWithBucketName(String bucketName) {
+    GrpcCallContext ret = GrpcCallContext.createDefault();
+    if (bucketName != null && !bucketName.isEmpty()) {
+      return ret.withExtraHeaders(
+          ImmutableMap.of(
+              "x-goog-request-params", ImmutableList.of(String.format("bucket=%s", bucketName))));
+    }
+    return ret;
   }
 
   @FunctionalInterface
@@ -77,7 +91,9 @@ final class WriteFlushStrategy {
      * @param onSuccessCallback Callback to signal success, and provide the final response.
      */
     Flusher newFlusher(
-        LongConsumer committedTotalBytesCallback, Consumer<WriteObjectResponse> onSuccessCallback);
+        String bucketName,
+        LongConsumer committedTotalBytesCallback,
+        Consumer<WriteObjectResponse> onSuccessCallback);
   }
 
   interface Flusher {
@@ -89,14 +105,17 @@ final class WriteFlushStrategy {
   private static final class FsyncEveryFlusher implements Flusher {
 
     private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+    private final String bucketName;
     private final LongConsumer sizeCallback;
     private final Consumer<WriteObjectResponse> completeCallback;
 
     private FsyncEveryFlusher(
         ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        String bucketName,
         LongConsumer sizeCallback,
         Consumer<WriteObjectResponse> completeCallback) {
       this.write = write;
+      this.bucketName = bucketName;
       this.sizeCallback = sizeCallback;
       this.completeCallback = completeCallback;
     }
@@ -104,7 +123,9 @@ final class WriteFlushStrategy {
     public void flush(@NonNull List<WriteObjectRequest> segments) {
 
       Observer observer = new Observer(sizeCallback, completeCallback);
-      ApiStreamObserver<WriteObjectRequest> write = this.write.clientStreamingCall(observer);
+      GrpcCallContext internalContext = contextWithBucketName(bucketName);
+      ApiStreamObserver<WriteObjectRequest> write =
+          this.write.withDefaultCallContext(internalContext).clientStreamingCall(observer);
 
       boolean first = true;
       for (WriteObjectRequest message : segments) {
@@ -129,6 +150,7 @@ final class WriteFlushStrategy {
   private static final class FsyncOnClose implements Flusher {
 
     private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+    private final String bucketName;
     private final Observer responseObserver;
 
     private volatile ApiStreamObserver<WriteObjectRequest> stream;
@@ -136,9 +158,11 @@ final class WriteFlushStrategy {
 
     private FsyncOnClose(
         ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        String bucketName,
         LongConsumer sizeCallback,
         Consumer<WriteObjectResponse> completeCallback) {
       this.write = write;
+      this.bucketName = bucketName;
       this.responseObserver = new Observer(sizeCallback, completeCallback);
     }
 
@@ -172,7 +196,11 @@ final class WriteFlushStrategy {
       if (stream == null) {
         synchronized (this) {
           if (stream == null) {
-            stream = write.clientStreamingCall(responseObserver);
+            GrpcCallContext internalContext = contextWithBucketName(bucketName);
+            stream =
+                this.write
+                    .withDefaultCallContext(internalContext)
+                    .clientStreamingCall(responseObserver);
           }
         }
       }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
index bc32ccea..70ad6a34 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
@@ -77,7 +77,7 @@ public final class GapicUnbufferedWritableByteChannelTest {
         WriteObjectResponse.newBuilder().setResource(obj.toBuilder().setSize(40)).build();
 
     WriteObjectRequest base = WriteObjectRequest.newBuilder().setWriteObjectSpec(spec).build();
-    WriteObjectRequestBuilderFactory reqFactory = base::toBuilder;
+    WriteObjectRequestBuilderFactory reqFactory = WriteObjectRequestBuilderFactory.simple(base);
 
     StorageImplBase service =
         new DirectWriteService(
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
new file mode 100644
index 00000000..9ad61533
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.gax.rpc.ApiCallContext;
+import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.cloud.storage.WriteFlushStrategy.Flusher;
+import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import org.junit.Test;
+
+public final class WriteFlushStrategyTest {
+  private static final Map<String, List<String>> expectedHeaderNonNullNonEmpty =
+      ImmutableMap.of("x-goog-request-params", ImmutableList.of("bucket=bucket-name"));
+  private static final Map<String, List<String>> expectedHeaderNonNullEmpty = ImmutableMap.of();
+  private static final Map<String, List<String>> expectedHeaderNull = ImmutableMap.of();
+
+  @Test
+  public void bucketNameAddedToXGoogRequestParams_nonNull_nonEmpty_fsyncEveryFlush() {
+    doTest(WriteFlushStrategy::fsyncEveryFlush, "bucket-name", expectedHeaderNonNullNonEmpty);
+  }
+
+  @Test
+  public void bucketNameAddedToXGoogRequestParams_nonNull_nonEmpty_fsyncOnClose() {
+    doTest(WriteFlushStrategy::fsyncOnClose, "bucket-name", expectedHeaderNonNullNonEmpty);
+  }
+
+  @Test
+  public void bucketNameNotAddedToXGoogRequestParams_nonNull_empty_fsyncEveryFlush() {
+    doTest(WriteFlushStrategy::fsyncEveryFlush, "", expectedHeaderNonNullEmpty);
+  }
+
+  @Test
+  public void bucketNameNotAddedToXGoogRequestParams_nonNull_empty_fsyncOnClose() {
+    doTest(WriteFlushStrategy::fsyncOnClose, "", expectedHeaderNonNullEmpty);
+  }
+
+  @Test
+  public void bucketNameNotAddedToXGoogRequestParams_null_fsyncEveryFlush() {
+    doTest(WriteFlushStrategy::fsyncEveryFlush, null, expectedHeaderNull);
+  }
+
+  @Test
+  public void bucketNameNotAddedToXGoogRequestParams_null_fsyncOnClose() {
+    doTest(WriteFlushStrategy::fsyncOnClose, null, expectedHeaderNull);
+  }
+
+  private static void doTest(
+      Function<ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse>, FlusherFactory> ff,
+      String bucketName,
+      Map<String, List<String>> expectedHeader) {
+    AtomicLong c = new AtomicLong(0);
+    AtomicReference<WriteObjectResponse> ref = new AtomicReference<>();
+    AtomicReference<Map<String, List<String>>> actualHeader = new AtomicReference<>();
+    ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write =
+        new ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse>() {
+          @Override
+          public ApiStreamObserver<WriteObjectRequest> clientStreamingCall(
+              ApiStreamObserver<WriteObjectResponse> responseObserver, ApiCallContext context) {
+            Map<String, List<String>> extraHeaders = context.getExtraHeaders();
+            actualHeader.compareAndSet(null, extraHeaders);
+            return new ApiStreamObserver<WriteObjectRequest>() {
+              @Override
+              public void onNext(WriteObjectRequest value) {}
+
+              @Override
+              public void onError(Throwable t) {}
+
+              @Override
+              public void onCompleted() {
+                responseObserver.onCompleted();
+              }
+            };
+          }
+        };
+    FlusherFactory factory = ff.apply(write);
+    Flusher flusher = factory.newFlusher(bucketName, c::addAndGet, ref::set);
+    flusher.flush(Collections.emptyList());
+    flusher.close(null);
+    assertThat(actualHeader.get()).isEqualTo(expectedHeader);
+  }
+}