Repository: googleapis/java-storage
Pull request #1588 (2022-08-26 13:57:35)

test: add grpc integration tests for simplistic object read/write
Update BucketFixture to provide its own implementation of bulk object delete.
The method in RemoteStorageHelper relies upon batch methods which aren't
supported for grpc.




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/GrpcConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
index df0c4124..afba3bf9 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
@@ -825,7 +825,10 @@ final class GrpcConversions {
     if (from.hasCustomerEncryption()) {
       toBuilder.setCustomerEncryption(customerEncryptionCodec.decode(from.getCustomerEncryption()));
     }
-    ifNonNull(from.getStorageClass(), StorageClass::valueOf, toBuilder::setStorageClass);
+    String storageClass = from.getStorageClass();
+    if (!storageClass.isEmpty()) {
+      toBuilder.setStorageClass(StorageClass.valueOf(storageClass));
+    }
     if (from.hasUpdateStorageClassTime()) {
       toBuilder.setTimeStorageClassUpdatedOffsetDateTime(
           timestampCodec.decode(from.getUpdateStorageClassTime()));
@@ -836,7 +839,7 @@ final class GrpcConversions {
     if (from.hasEventBasedHold()) {
       toBuilder.setEventBasedHold(from.getEventBasedHold());
     }
-    ifNonNull(from.getTemporaryHold(), toBuilder::setTemporaryHold);
+    toBuilder.setTemporaryHold(from.getTemporaryHold());
     if (from.hasRetentionExpireTime()) {
       toBuilder.setRetentionExpirationTimeOffsetDateTime(
           timestampCodec.decode(from.getRetentionExpireTime()));
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..b949242f 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),
@@ -381,7 +382,9 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     GrpcCallContext grpcCallContext =
         opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
     GetObjectRequest.Builder builder =
-        GetObjectRequest.newBuilder().setBucket(blob.getBucket()).setObject(blob.getName());
+        GetObjectRequest.newBuilder()
+            .setBucket(bucketNameCodec.encode(blob.getBucket()))
+            .setObject(blob.getName());
     GetObjectRequest req = opts.getObjectsRequest().apply(builder).build();
     return Retrying.run(
         getOptions(),
@@ -409,9 +412,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 +431,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 +500,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 +526,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 +563,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 +579,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 +616,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 +629,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 +730,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 +902,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 +973,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 +1003,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 +1028,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 +1170,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/BucketFixture.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketFixture.java
index ef1a79ad..c81881dc 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketFixture.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketFixture.java
@@ -18,13 +18,21 @@ package com.google.cloud.storage;
 
 import static java.util.Objects.requireNonNull;
 
+import com.google.api.gax.paging.Page;
 import com.google.cloud.storage.conformance.retry.CleanupStrategy;
-import com.google.cloud.storage.testing.RemoteStorageHelper;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 import org.junit.rules.TestRule;
 import org.junit.runner.Description;
 import org.junit.runners.model.Statement;
@@ -94,7 +102,30 @@ public final class BucketFixture implements TestRule {
   private void doCleanup(String bucketName, Storage s) {
     LOGGER.info("Starting bucket cleanup...");
     try {
-      RemoteStorageHelper.forceDelete(s, bucketName, 5, TimeUnit.MINUTES);
+      // TODO: probe bucket existence, a bad test could have deleted the bucket
+      Page<Blob> page1 = s.list(bucketName);
+
+      int availableProcessors = Runtime.getRuntime().availableProcessors();
+      ListeningExecutorService exec =
+          MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2 * availableProcessors));
+
+      List<ListenableFuture<DeleteResult>> futures =
+          StreamSupport.stream(page1.iterateAll().spliterator(), false)
+              .map(b -> exec.submit(() -> new DeleteResult(b.getName(), b.delete())))
+              .collect(Collectors.toList());
+
+      List<DeleteResult> deleteResults = Futures.allAsList(futures).get(5, TimeUnit.MINUTES);
+      List<DeleteResult> failedDeletes =
+          deleteResults.stream().filter(r -> !r.success).collect(Collectors.toList());
+      failedDeletes.forEach(
+          r -> LOGGER.warning(String.format("Failed to delete object %s/%s", bucketName, r.name)));
+
+      if (failedDeletes.isEmpty()) {
+        s.delete(bucketName);
+      } else {
+        LOGGER.warning("Unable to delete bucket due to previous failed object deletes");
+      }
+      exec.shutdown();
       LOGGER.info("Bucket cleanup complete");
     } catch (Exception e) {
       LOGGER.log(Level.SEVERE, e, () -> "Error during bucket cleanup.");
@@ -127,4 +158,14 @@ public final class BucketFixture implements TestRule {
           requireNonNull(handle, "handle must be non null"), bucketNameFmtString, cleanupStrategy);
     }
   }
+
+  private static final class DeleteResult {
+    private final String name;
+    private final boolean success;
+
+    DeleteResult(String name, boolean success) {
+      this.name = name;
+      this.success = success;
+    }
+  }
 }
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);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
index cca6645d..53be1a63 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcTest.java
@@ -20,7 +20,9 @@ import static com.google.common.truth.Truth.assertThat;
 
 import com.google.api.gax.paging.Page;
 import com.google.cloud.NoCredentials;
+import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.BucketFixture;
 import com.google.cloud.storage.BucketInfo;
@@ -28,6 +30,10 @@ import com.google.cloud.storage.HmacKey;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
 import com.google.cloud.storage.HmacKey.HmacKeyState;
 import com.google.cloud.storage.ServiceAccount;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.Storage.CreateHmacKeyOption;
 import com.google.cloud.storage.Storage.ListHmacKeysOption;
 import com.google.cloud.storage.StorageFixture;
@@ -36,10 +42,17 @@ import com.google.cloud.storage.conformance.retry.CleanupStrategy;
 import com.google.cloud.storage.conformance.retry.TestBench;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import com.google.common.collect.ImmutableList;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
 import java.util.stream.IntStream;
 import java.util.stream.StreamSupport;
 import org.junit.ClassRule;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TestName;
 
 public final class ITGrpcTest {
   @ClassRule(order = 1)
@@ -64,6 +77,8 @@ public final class ITGrpcTest {
           .setHandle(storageFixture::getInstance)
           .build();
 
+  @Rule public final TestName testName = new TestName();
+
   @Test
   public void testCreateBucket() {
     final String bucketName = RemoteStorageHelper.generateBucketName();
@@ -74,13 +89,30 @@ public final class ITGrpcTest {
   @Test
   public void listBlobs() {
     BucketInfo bucketInfo = bucketFixture.getBucketInfo();
-    Page<Blob> list = storageFixture.getInstance().list(bucketInfo.getName());
-    ImmutableList<String> bucketNames =
+    byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+    String prefix = testName.getMethodName();
+    List<Blob> blobs =
+        IntStream.rangeClosed(1, 10)
+            .mapToObj(i -> String.format("%s/%02d", prefix, i))
+            .map(n -> BlobInfo.newBuilder(bucketInfo, n).build())
+            .map(
+                info ->
+                    storageFixture
+                        .getInstance()
+                        .create(info, content, BlobTargetOption.doesNotExist()))
+            .collect(ImmutableList.toImmutableList());
+
+    List<String> expected =
+        blobs.stream().map(Blob::getName).collect(ImmutableList.toImmutableList());
+
+    Page<Blob> list =
+        storageFixture.getInstance().list(bucketInfo.getName(), BlobListOption.prefix(prefix));
+    ImmutableList<String> actual =
         StreamSupport.stream(list.iterateAll().spliterator(), false)
             .map(Blob::getName)
             .collect(ImmutableList.toImmutableList());
 
-    assertThat(bucketNames).isEmpty();
+    assertThat(actual).isEqualTo(expected);
   }
 
   @Test
@@ -156,4 +188,53 @@ public final class ITGrpcTest {
     storageFixture.getInstance().updateHmacKeyState(hmacKey.getMetadata(), HmacKeyState.INACTIVE);
     storageFixture.getInstance().deleteHmacKey(hmacKey.getMetadata());
   }
+
+  @Test
+  public void object_writeGetRead() {
+    Storage s = storageFixture.getInstance();
+    BlobInfo info = BlobInfo.newBuilder(bucketFixture.getBucketInfo(), "writeGetRead").build();
+    byte[] content = "hello, world".getBytes(StandardCharsets.UTF_8);
+    s.create(info, content, BlobTargetOption.doesNotExist());
+
+    Blob blob = s.get(info.getBlobId());
+
+    byte[] actualContent = blob.getContent();
+    assertThat(actualContent).isEqualTo(content);
+  }
+
+  @Test
+  public void objectWrite_storage_create() {
+    BucketInfo bucketInfo = bucketFixture.getBucketInfo();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+    Blob blob = storageFixture.getInstance().create(info, content, BlobTargetOption.doesNotExist());
+    byte[] actual = blob.getContent();
+    assertThat(actual).isEqualTo(content);
+  }
+
+  @Test
+  public void objectWrite_storage_create_stream() {
+    BucketInfo bucketInfo = bucketFixture.getBucketInfo();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+    Blob blob =
+        storageFixture
+            .getInstance()
+            .create(info, new ByteArrayInputStream(content), BlobWriteOption.doesNotExist());
+    byte[] actual = blob.getContent();
+    assertThat(actual).isEqualTo(content);
+  }
+
+  @Test
+  public void objectWrite_storage_writer() throws IOException {
+    BucketInfo bucketInfo = bucketFixture.getBucketInfo();
+    BlobInfo info = BlobInfo.newBuilder(bucketInfo, testName.getMethodName()).build();
+    byte[] content = "Hello, World!".getBytes(StandardCharsets.UTF_8);
+    try (WriteChannel c =
+        storageFixture.getInstance().writer(info, BlobWriteOption.doesNotExist())) {
+      c.write(ByteBuffer.wrap(content));
+    }
+    byte[] actual = storageFixture.getInstance().readAllBytes(info.getBlobId());
+    assertThat(actual).isEqualTo(content);
+  }
 }