Repository: googleapis/java-storage
Pull request #1458 (2022-06-16 22:58:22)

test: add property test for DefaultBufferedWritableByteChannel
New property test against arbitrarily generated data along with buffer size, and app write size.

Ensure the expected number of writes and their sizes are performed for a specific data size.

Ensure the expected number of flushes are performed, and at the correct intervals.

Ensure channel is closed after full read, and throws a ClosedChannelException when attempted to read after close




Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
index 5a92301b..5f3f850f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
@@ -29,6 +29,7 @@ import com.google.cloud.storage.Storage.BlobTargetOption;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.Storage.CopyRequest;
 import com.google.cloud.storage.Storage.SignUrlOption;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.io.BaseEncoding;
 import java.io.IOException;
@@ -225,6 +226,7 @@ public class Blob extends BlobInfo {
    * @throws StorageException upon failure
    * @see Storage#downloadTo(BlobId, Path, Storage.BlobSourceOption...)
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public void downloadTo(Path path, BlobSourceOption... options) {
     storage.downloadTo(this.getBlobId(), path, toSourceOptions(this, options));
   }
@@ -237,6 +239,7 @@ public class Blob extends BlobInfo {
    * @throws StorageException upon failure
    * @see Storage#downloadTo(BlobId, OutputStream, Storage.BlobSourceOption...)
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public void downloadTo(OutputStream outputStream, BlobSourceOption... options) {
     storage.downloadTo(this.getBlobId(), outputStream, toSourceOptions(this, options));
   }
@@ -250,6 +253,7 @@ public class Blob extends BlobInfo {
    * @param path destination
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public void downloadTo(Path path) {
     downloadTo(path, new BlobSourceOption[0]);
   }
@@ -536,6 +540,7 @@ public class Blob extends BlobInfo {
    * @return true if this blob exists, false otherwise
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public boolean exists(BlobSourceOption... options) {
     int length = options.length;
     Storage.BlobGetOption[] getOptions = Arrays.copyOf(toGetOptions(this, options), length + 1);
@@ -556,6 +561,7 @@ public class Blob extends BlobInfo {
    * @param options blob read options
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public byte[] getContent(BlobSourceOption... options) {
     return storage.readAllBytes(getBlobId(), toSourceOptions(this, options));
   }
@@ -592,6 +598,7 @@ public class Blob extends BlobInfo {
    * @return a {@code Blob} object with latest information or {@code null} if no longer exists.
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Blob reload(BlobSourceOption... options) {
     // BlobId with generation unset is needed to retrieve the latest version of the Blob
     BlobId idWithoutGeneration = BlobId.of(getBucket(), getName());
@@ -621,6 +628,7 @@ public class Blob extends BlobInfo {
    * @see <a
    *     href="https://cloud.google.com/storage/docs/json_api/v1/objects/update">https://cloud.google.com/storage/docs/json_api/v1/objects/update</a>
    */
+  @TransportCompatibility({Transport.HTTP})
   public Blob update(BlobTargetOption... options) {
     return storage.update(this, options);
   }
@@ -644,6 +652,7 @@ public class Blob extends BlobInfo {
    * @return {@code true} if blob was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public boolean delete(BlobSourceOption... options) {
     return storage.delete(getBlobId(), toSourceOptions(this, options));
   }
@@ -667,6 +676,7 @@ public class Blob extends BlobInfo {
    *     blob or to complete the copy if more than one RPC request is needed
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public CopyWriter copyTo(BlobId targetBlob, BlobSourceOption... options) {
     CopyRequest copyRequest =
         CopyRequest.newBuilder()
@@ -695,6 +705,7 @@ public class Blob extends BlobInfo {
    *     blob or to complete the copy if more than one RPC request is needed
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public CopyWriter copyTo(String targetBucket, BlobSourceOption... options) {
     return copyTo(targetBucket, getName(), options);
   }
@@ -729,6 +740,7 @@ public class Blob extends BlobInfo {
    *     blob or to complete the copy if more than one RPC request is needed
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public CopyWriter copyTo(String targetBucket, String targetBlob, BlobSourceOption... options) {
     return copyTo(BlobId.of(targetBucket, targetBlob), options);
   }
@@ -765,6 +777,7 @@ public class Blob extends BlobInfo {
    * @param options blob read options
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public ReadChannel reader(BlobSourceOption... options) {
     return storage.reader(getBlobId(), toSourceOptions(this, options));
   }
@@ -789,6 +802,7 @@ public class Blob extends BlobInfo {
    * @param options target blob options
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public WriteChannel writer(BlobWriteOption... options) {
     return storage.writer(this, options);
   }
@@ -854,6 +868,7 @@ public class Blob extends BlobInfo {
    * @throws SigningException if the attempt to sign the URL failed
    * @see <a href="https://cloud.google.com/storage/docs/access-control#Signed-URLs">Signed-URLs</a>
    */
+  @TransportCompatibility(Transport.HTTP)
   public URL signUrl(long duration, TimeUnit unit, SignUrlOption... options) {
     return storage.signUrl(this, duration, unit, options);
   }
@@ -869,6 +884,7 @@ public class Blob extends BlobInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl getAcl(Entity entity) {
     return storage.getAcl(getBlobId(), entity);
   }
@@ -890,6 +906,7 @@ public class Blob extends BlobInfo {
    * @return {@code true} if the ACL was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public boolean deleteAcl(Entity entity) {
     return storage.deleteAcl(getBlobId(), entity);
   }
@@ -905,6 +922,7 @@ public class Blob extends BlobInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl createAcl(Acl acl) {
     return storage.createAcl(getBlobId(), acl);
   }
@@ -920,6 +938,7 @@ public class Blob extends BlobInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl updateAcl(Acl acl) {
     return storage.updateAcl(getBlobId(), acl);
   }
@@ -938,6 +957,7 @@ public class Blob extends BlobInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public List<Acl> listAcls() {
     return storage.listAcls(getBlobId());
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
index a7206811..0bfb5154 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Bucket.java
@@ -27,6 +27,7 @@ import com.google.cloud.storage.Acl.Entity;
 import com.google.cloud.storage.Storage.BlobGetOption;
 import com.google.cloud.storage.Storage.BlobListOption;
 import com.google.cloud.storage.Storage.BucketTargetOption;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
@@ -766,6 +767,7 @@ public class Bucket extends BucketInfo {
    * @return true if this bucket exists, false otherwise
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public boolean exists(BucketSourceOption... options) {
     int length = options.length;
     Storage.BucketGetOption[] getOptions = Arrays.copyOf(toGetOptions(this, options), length + 1);
@@ -790,6 +792,7 @@ public class Bucket extends BucketInfo {
    * @return a {@code Bucket} object with latest information or {@code null} if not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Bucket reload(BucketSourceOption... options) {
     return storage.get(getName(), toGetOptions(this, options));
   }
@@ -811,6 +814,7 @@ public class Bucket extends BucketInfo {
    * @return a {@code Bucket} object with updated information
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Bucket update(BucketTargetOption... options) {
     return storage.update(this, options);
   }
@@ -834,6 +838,7 @@ public class Bucket extends BucketInfo {
    * @return {@code true} if bucket was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public boolean delete(BucketSourceOption... options) {
     return storage.delete(getName(), toSourceOptions(this, options));
   }
@@ -855,6 +860,7 @@ public class Bucket extends BucketInfo {
    * @param options options for listing blobs
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Page<Blob> list(BlobListOption... options) {
     return storage.list(getName(), options);
   }
@@ -875,6 +881,7 @@ public class Bucket extends BucketInfo {
    * @param options blob search options
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Blob get(String blob, BlobGetOption... options) {
     return storage.get(BlobId.of(getName(), blob), options);
   }
@@ -901,6 +908,7 @@ public class Bucket extends BucketInfo {
    * @return an immutable list of {@code Blob} objects
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public List<Blob> get(String blobName1, String blobName2, String... blobNames) {
     List<BlobId> blobIds = Lists.newArrayListWithCapacity(blobNames.length + 2);
     blobIds.add(BlobId.of(getName(), blobName1));
@@ -934,6 +942,7 @@ public class Bucket extends BucketInfo {
    * @return an immutable list of {@code Blob} objects
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public List<Blob> get(Iterable<String> blobNames) {
     ImmutableList.Builder<BlobId> builder = ImmutableList.builder();
     for (String blobName : blobNames) {
@@ -962,6 +971,7 @@ public class Bucket extends BucketInfo {
    * @return a complete blob information
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Blob create(String blob, byte[] content, String contentType, BlobTargetOption... options) {
     BlobInfo blobInfo =
         BlobInfo.newBuilder(BlobId.of(getName(), blob)).setContentType(contentType).build();
@@ -990,6 +1000,7 @@ public class Bucket extends BucketInfo {
    * @return a complete blob information
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Blob create(
       String blob, InputStream content, String contentType, BlobWriteOption... options) {
     BlobInfo blobInfo =
@@ -1018,6 +1029,7 @@ public class Bucket extends BucketInfo {
    * @return a complete blob information
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Blob create(String blob, byte[] content, BlobTargetOption... options) {
     BlobInfo blobInfo = BlobInfo.newBuilder(BlobId.of(getName(), blob)).build();
     Tuple<BlobInfo, Storage.BlobTargetOption[]> target =
@@ -1044,6 +1056,7 @@ public class Bucket extends BucketInfo {
    * @return a complete blob information
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   public Blob create(String blob, InputStream content, BlobWriteOption... options) {
     BlobInfo blobInfo = BlobInfo.newBuilder(BlobId.of(getName(), blob)).build();
     Tuple<BlobInfo, Storage.BlobWriteOption[]> write =
@@ -1062,6 +1075,7 @@ public class Bucket extends BucketInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl getAcl(Entity entity) {
     return storage.getAcl(getName(), entity);
   }
@@ -1083,6 +1097,7 @@ public class Bucket extends BucketInfo {
    * @return {@code true} if the ACL was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public boolean deleteAcl(Entity entity) {
     return storage.deleteAcl(getName(), entity);
   }
@@ -1098,6 +1113,7 @@ public class Bucket extends BucketInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl createAcl(Acl acl) {
     return storage.createAcl(getName(), acl);
   }
@@ -1113,6 +1129,7 @@ public class Bucket extends BucketInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl updateAcl(Acl acl) {
     return storage.updateAcl(getName(), acl);
   }
@@ -1131,6 +1148,7 @@ public class Bucket extends BucketInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public List<Acl> listAcls() {
     return storage.listAcls(getName());
   }
@@ -1150,6 +1168,7 @@ public class Bucket extends BucketInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl getDefaultAcl(Entity entity) {
     return storage.getDefaultAcl(getName(), entity);
   }
@@ -1174,6 +1193,7 @@ public class Bucket extends BucketInfo {
    * @return {@code true} if the ACL was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public boolean deleteDefaultAcl(Entity entity) {
     return storage.deleteDefaultAcl(getName(), entity);
   }
@@ -1192,6 +1212,7 @@ public class Bucket extends BucketInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl createDefaultAcl(Acl acl) {
     return storage.createDefaultAcl(getName(), acl);
   }
@@ -1210,6 +1231,7 @@ public class Bucket extends BucketInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Acl updateDefaultAcl(Acl acl) {
     return storage.updateDefaultAcl(getName(), acl);
   }
@@ -1231,6 +1253,7 @@ public class Bucket extends BucketInfo {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public List<Acl> listDefaultAcls() {
     return storage.listDefaultAcls(getName());
   }
@@ -1256,6 +1279,7 @@ public class Bucket extends BucketInfo {
    * @return a {@code Bucket} object of the locked bucket
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   public Bucket lockRetentionPolicy(BucketTargetOption... options) {
     return storage.lockRetentionPolicy(this, options);
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedWritableByteChannel.java
index 8381adf6..27e7a8e6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedWritableByteChannel.java
@@ -22,6 +22,37 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 
+/**
+ * Buffering {@link java.nio.channels.WritableByteChannel} which attempts to maximize the amount of
+ * bytes written to the underlying {@link UnbufferedWritableByteChannel} while minimizing
+ * unnecessary copying of said bytes.
+ *
+ * <p>Our flushing strategy is "eager", meaning as soon as we have enough total bytes equal to the
+ * capacity of our buffer we will write to the underlying channel.
+ *
+ * <p>A few strategies are employed to meet the state goals.
+ *
+ * <ol>
+ *   <li>If we do not have any bytes in our buffer and {@code src} is the same size as our buffer,
+ *       simply {@link UnbufferedWritableByteChannel#write(ByteBuffer) write(src)} to the the
+ *       underlying channel
+ *   <li>If we do not have any bytes in our buffer and {@code src} is larger than the size of our
+ *       buffer, take a slice of {@code src} the same size as our buffer and {@link
+ *       UnbufferedWritableByteChannel#write(ByteBuffer[]) write([buffer, slice])} before enqueuing
+ *       any outstanding bytes which are smaller than our buffer.
+ *   <li>If we do not have any bytes in our buffer and {@code src} is smaller than the size of our
+ *       buffer, enqueue it in full
+ *   <li>If we do have enqueued bytes and {@code src} is the size of our remaining buffer space
+ *       {@link UnbufferedWritableByteChannel#write(ByteBuffer[]) write([buffer, src])} to the
+ *       underlying channel
+ *   <li>If we do have enqueued bytes and {@code src} is larger than the size of our remaining
+ *       buffer space, take a slice of {@code src} the same size as our buffer and {@link
+ *       UnbufferedWritableByteChannel#write(ByteBuffer[]) write([buffer, slice])} to the underlying
+ *       channel before enqueuing any outstanding bytes which are smaller than our buffer.
+ *   <li>If we do have enqueued bytes and {@code src} is smaller than our remaining buffer space,
+ *       enqueue it in full
+ * </ol>
+ */
 final class DefaultBufferedWritableByteChannel implements BufferedWritableByteChannel {
 
   private final ByteBuffer buffer;
@@ -38,6 +69,7 @@ final class DefaultBufferedWritableByteChannel implements BufferedWritableByteCh
     return channel.isComplete();
   }
 
+  @SuppressWarnings("UnnecessaryLocalVariable")
   @Override
   public int write(ByteBuffer src) throws IOException {
     if (!channel.isOpen()) {
@@ -46,41 +78,85 @@ final class DefaultBufferedWritableByteChannel implements BufferedWritableByteCh
     int bytesConsumed = 0;
 
     while (src.hasRemaining()) {
-      int bufferRemaining = buffer.remaining();
       int srcRemaining = src.remaining();
+      int srcPosition = src.position();
 
-      final int tmpBytesConsumed;
-      if (!enqueuedBytes() && bufferRemaining == srcRemaining) {
-        channel.write(src);
-        tmpBytesConsumed = bufferRemaining;
-      } else if (!enqueuedBytes() && bufferRemaining <= srcRemaining) {
-        // no enqueued data and the src provided is larger than our buffer
-        // rather than copying into the buffer, simply flush a slice from the source
-        ByteBuffer slice = src.slice();
-        Buffers.limit(slice, bufferRemaining);
-        int write = channel.write(slice);
-        Buffers.position(src, src.position() + write);
-        tmpBytesConsumed = bufferRemaining;
+      int capacity = buffer.capacity();
+      int bufferRemaining = buffer.remaining();
+      int bufferPending = capacity - bufferRemaining;
+
+      boolean enqueuedBytes = buffer.hasRemaining();
+      if (enqueuedBytes && srcRemaining < bufferRemaining) {
+        // srcRemaining is smaller than the remaining space in our buffer, enqueue it in full
+        buffer.put(src);
+        bytesConsumed += srcRemaining;
+        break;
+      } else if (enqueuedBytes) {
+        // between bufferPending and srcRemaining we have a full buffers worth of data
+        // Figure out what we need to do before flushing it
+
+        ByteBuffer buf;
+        int sliceLimit = bufferRemaining; // alias for easier readability below
+        boolean usingSlice = false;
+        if (srcRemaining == bufferRemaining) {
+          // our buffer and all of src are equal to a full buffer, no need to slice src
+          buf = src;
+        } else {
+          // our buffer and all of src are larger than a full buffer, take a slice of src such that
+          // the total number of bytes are equal to capacity
+          ByteBuffer slice = src.slice();
+          Buffers.limit(slice, sliceLimit);
+          usingSlice = true;
+          buf = slice;
+        }
+
+        Buffers.flip(buffer);
+        ByteBuffer[] srcs = {buffer, buf};
+        long write = channel.write(srcs);
+        if (write == capacity) {
+          // we successfully wrote all the bytes we wanted to
+          Buffers.clear(buffer);
+          if (usingSlice) {
+            Buffers.position(src, srcPosition + sliceLimit);
+          }
+          bytesConsumed += sliceLimit;
+        } else {
+          if (buffer.hasRemaining()) {
+            // we didn't write enough bytes to consume the whole buffer. Do not advance the
+            // position of src
+            buffer.compact();
+          } else {
+            // we wrote enough to consume the buffer
+            Buffers.clear(buffer);
+            // we didn't write enough to consume the whole slice, determine how much of the slice
+            // was written and advance the position of src
+            int sliceWritten = Math.toIntExact(write - bufferPending);
+            Buffers.position(src, srcPosition + sliceWritten);
+            bytesConsumed += sliceWritten;
+          }
+        }
       } else {
-        if (bufferRemaining <= srcRemaining) {
-          // some enqueued bytes in the buffer, fill the remaining capacity of buffer
-          // before yielding
+        // no enqueued data, see if we can simply write the provided src or a slice of it
+        // since our buffer is empty
+        if (bufferRemaining == srcRemaining) {
+          // the capacity of buffer and the bytes remaining in src are the same, directly
+          // write src
+          bytesConsumed += channel.write(src);
+        } else if (bufferRemaining <= srcRemaining) {
+          // the src provided is larger than our buffer. rather than copying into the buffer, simply
+          // write a slice
           ByteBuffer slice = src.slice();
           Buffers.limit(slice, bufferRemaining);
-          buffer.put(slice);
-          Buffers.position(src, src.position() + bufferRemaining);
-          tmpBytesConsumed = bufferRemaining;
+          int write = channel.write(slice);
+          Buffers.position(src, srcPosition + write);
+          bytesConsumed += write;
         } else {
-          // the available in src is less than buffers remaining capacity, enqueue it in full
+          // srcRemaining is smaller than the remaining space in our buffer, enqueue it in full
           buffer.put(src);
-          tmpBytesConsumed = srcRemaining;
+          bytesConsumed += srcRemaining;
         }
-        maybeFlushBuffer();
       }
-
-      bytesConsumed += tmpBytesConsumed;
     }
-
     return bytesConsumed;
   }
 
@@ -98,22 +174,14 @@ final class DefaultBufferedWritableByteChannel implements BufferedWritableByteCh
 
   @Override
   public void flush() throws IOException {
-    if (enqueuedBytes()) {
-      ByteBuffer b = buffer.duplicate();
-      Buffers.position(b, 0);
-      Buffers.limit(b, buffer.position());
-      channel.write(b);
-      Buffers.position(buffer, 0);
-    }
-  }
-
-  private void maybeFlushBuffer() throws IOException {
-    if (enqueuedBytes() && !buffer.hasRemaining()) {
-      flush();
+    if (buffer.hasRemaining()) {
+      Buffers.flip(buffer);
+      channel.write(buffer);
+      if (buffer.hasRemaining()) {
+        buffer.compact();
+      } else {
+        Buffers.clear(buffer);
+      }
     }
   }
-
-  private boolean enqueuedBytes() {
-    return buffer.position() > 0;
-  }
 }
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 d7a041d1..07bf75a3 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
@@ -25,6 +25,7 @@ import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.gax.paging.AbstractPage;
 import com.google.api.gax.paging.Page;
+import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiExceptions;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.BaseService;
@@ -41,6 +42,9 @@ import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.io.ByteStreams;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.storage.v2.BucketName;
+import com.google.storage.v2.CreateBucketRequest;
+import com.google.storage.v2.DeleteBucketRequest;
 import com.google.storage.v2.DeleteHmacKeyRequest;
 import com.google.storage.v2.GetBucketRequest;
 import com.google.storage.v2.GetObjectRequest;
@@ -74,11 +78,13 @@ import java.nio.file.Files;
 import java.nio.file.OpenOption;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 final class GrpcStorageImpl extends BaseService<StorageOptions> implements Storage {
 
@@ -112,7 +118,28 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) {
-    return todo();
+    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    com.google.storage.v2.Bucket bucket =
+        com.google.storage.v2.Bucket.newBuilder()
+            .setName(BucketName.format(this.getOptions().getProjectId(), bucketInfo.getName()))
+            .setProject(this.getOptions().getProjectId())
+            .build();
+    CreateBucketRequest.Builder bucketRequestBuilder =
+        CreateBucketRequest.newBuilder().setBucket(bucket);
+    ifNonNull(
+        (String) optionsMap.get(StorageRpc.Option.PREDEFINED_ACL),
+        bucketRequestBuilder::setPredefinedAcl);
+    ifNonNull(
+        (String) optionsMap.get(StorageRpc.Option.PREDEFINED_DEFAULT_OBJECT_ACL),
+        bucketRequestBuilder::setPredefinedDefaultObjectAcl);
+    CreateBucketRequest req = bucketRequestBuilder.build();
+    // TODO(frankyn): Do we care about projection because Apiary uses FULL for projection? Missing
+    // projection=full
+    return Retrying.run(
+        getOptions(),
+        retryAlgorithmManager.getFor(req),
+        () -> grpcStorageStub.createBucketCallable().call(req),
+        syntaxDecoders.bucket);
   }
 
   @Override
@@ -197,8 +224,6 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public Bucket get(String bucket, BucketGetOption... options) {
-    UnaryCallable<GetBucketRequest, com.google.storage.v2.Bucket> bucketCallable =
-        grpcStorageStub.getBucketCallable();
     final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
     GetBucketRequest.Builder bucketRequestBuilder = GetBucketRequest.newBuilder().setName(bucket);
     ifNonNull(
@@ -312,7 +337,29 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public boolean delete(String bucket, BucketSourceOption... options) {
-    return todo();
+    final Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    DeleteBucketRequest.Builder bucketRequestBuilder =
+        DeleteBucketRequest.newBuilder()
+            .setName(BucketName.format(this.getOptions().getProjectId(), bucket));
+    ifNonNull(
+        (Long) optionsMap.get(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH),
+        bucketRequestBuilder::setIfMetagenerationNotMatch);
+    ifNonNull(
+        (Long) optionsMap.get(StorageRpc.Option.IF_METAGENERATION_MATCH),
+        bucketRequestBuilder::setIfMetagenerationNotMatch);
+    DeleteBucketRequest req = bucketRequestBuilder.build();
+    try {
+      Retrying.run(
+          getOptions(),
+          retryAlgorithmManager.getFor(req),
+          () -> grpcStorageStub.deleteBucketCallable().call(req),
+          Decoder.identity());
+      return true;
+    } catch (ApiException e) {
+      // TODO: We should throw a StorageException instead of ApiException when making the
+      // deleteBucketCallable().call(req)
+      return false;
+    }
   }
 
   @Override
@@ -359,7 +406,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public StorageBatch batch() {
-    return todo();
+    return throwHttpJsonOnly("batch()");
   }
 
   @Override
@@ -427,12 +474,13 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public WriteChannel writer(URL signedURL) {
-    return todo();
+    return throwHttpJsonOnly(fmtMethodName("writer", URL.class));
   }
 
   @Override
   public URL signUrl(BlobInfo blobInfo, long duration, TimeUnit unit, SignUrlOption... options) {
-    return todo();
+    return throwHttpJsonOnly(
+        fmtMethodName("signUrl", BlobInfo.class, long.class, TimeUnit.class, SignUrlOption.class));
   }
 
   @Override
@@ -443,7 +491,15 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       PostFieldsV4 fields,
       PostConditionsV4 conditions,
       PostPolicyV4Option... options) {
-    return todo();
+    return throwHttpJsonOnly(
+        fmtMethodName(
+            "generateSignedPostPolicyV4",
+            BlobInfo.class,
+            long.class,
+            TimeUnit.class,
+            PostFieldsV4.class,
+            PostConditionsV4.class,
+            PostPolicyV4Option.class));
   }
 
   @Override
@@ -453,7 +509,14 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       TimeUnit unit,
       PostFieldsV4 fields,
       PostPolicyV4Option... options) {
-    return todo();
+    return throwHttpJsonOnly(
+        fmtMethodName(
+            "generateSignedPostPolicyV4",
+            BlobInfo.class,
+            long.class,
+            TimeUnit.class,
+            PostFieldsV4.class,
+            PostPolicyV4Option.class));
   }
 
   @Override
@@ -463,13 +526,26 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       TimeUnit unit,
       PostConditionsV4 conditions,
       PostPolicyV4Option... options) {
-    return todo();
+    return throwHttpJsonOnly(
+        fmtMethodName(
+            "generateSignedPostPolicyV4",
+            BlobInfo.class,
+            long.class,
+            TimeUnit.class,
+            PostConditionsV4.class,
+            PostPolicyV4Option.class));
   }
 
   @Override
   public PostPolicyV4 generateSignedPostPolicyV4(
       BlobInfo blobInfo, long duration, TimeUnit unit, PostPolicyV4Option... options) {
-    return todo();
+    return throwHttpJsonOnly(
+        fmtMethodName(
+            "generateSignedPostPolicyV4",
+            BlobInfo.class,
+            long.class,
+            TimeUnit.class,
+            PostPolicyV4Option.class));
   }
 
   @Override
@@ -816,4 +892,19 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       }
     }
   }
+
+  private <T> T throwHttpJsonOnly(String methodName) {
+    String message =
+        String.format(
+            "%s#%s is only supported for HTTP_JSON transport. Please use StorageOptions.http() to construct a compatible instance.",
+            Storage.class.getName(), methodName);
+    throw new UnsupportedOperationException(message);
+  }
+
+  private static String fmtMethodName(String name, Class<?>... args) {
+    return name
+        + "("
+        + Arrays.stream(args).map(Class::getName).collect(Collectors.joining(", "))
+        + ")";
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 05cd78f4..8ceef950 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -19,7 +19,9 @@ package com.google.cloud.storage;
 import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiClock;
+import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
+import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.ClientContext;
 import com.google.api.gax.rpc.HeaderProvider;
@@ -30,6 +32,7 @@ import com.google.cloud.ServiceRpc;
 import com.google.cloud.TransportOptions;
 import com.google.cloud.grpc.GrpcTransportOptions;
 import com.google.cloud.spi.ServiceRpcFactory;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.StorageRpcFactory;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableSet;
@@ -39,6 +42,8 @@ import java.io.IOException;
 import java.net.URI;
 import java.util.Set;
 
+@BetaApi
+@TransportCompatibility(Transport.GRPC)
 public final class GrpcStorageOptions extends StorageOptions {
 
   private static final long serialVersionUID = 4165732727259088956L;
@@ -48,6 +53,7 @@ public final class GrpcStorageOptions extends StorageOptions {
 
   private final GrpcRetryAlgorithmManager retryAlgorithmManager;
 
+  @BetaApi
   public GrpcStorageOptions(Builder builder, StorageDefaults serviceDefaults) {
     super(builder, serviceDefaults);
     this.retryAlgorithmManager =
@@ -65,6 +71,7 @@ public final class GrpcStorageOptions extends StorageOptions {
     return retryAlgorithmManager;
   }
 
+  @BetaApi
   @Override
   public GrpcStorageOptions.Builder toBuilder() {
     return new GrpcStorageOptions.Builder(this);
@@ -80,18 +87,22 @@ public final class GrpcStorageOptions extends StorageOptions {
     return obj instanceof StorageOptions && baseEquals((StorageOptions) obj);
   }
 
+  @BetaApi
   public static GrpcStorageOptions.Builder newBuilder() {
     return new GrpcStorageOptions.Builder().setHost(DEFAULT_HOST);
   }
 
+  @BetaApi
   public static GrpcStorageOptions getDefaultInstance() {
     return newBuilder().build();
   }
 
+  @BetaApi
   public static GrpcStorageOptions.GrpcStorageDefaults defaults() {
     return GrpcStorageOptions.GrpcStorageDefaults.INSTANCE;
   }
 
+  @BetaApi
   public static class Builder extends StorageOptions.Builder {
 
     private StorageRetryStrategy storageRetryStrategy;
@@ -102,6 +113,7 @@ public final class GrpcStorageOptions extends StorageOptions {
       super(options);
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setTransportOptions(TransportOptions transportOptions) {
       if (!(transportOptions instanceof GrpcTransportOptions)) {
@@ -118,6 +130,7 @@ public final class GrpcStorageOptions extends StorageOptions {
      * @return the builder
      * @see StorageRetryStrategy#getDefaultStorageRetryStrategy()
      */
+    @BetaApi
     public GrpcStorageOptions.Builder setStorageRetryStrategy(
         StorageRetryStrategy storageRetryStrategy) {
       this.storageRetryStrategy =
@@ -130,6 +143,7 @@ public final class GrpcStorageOptions extends StorageOptions {
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setServiceFactory(
         ServiceFactory<Storage, StorageOptions> serviceFactory) {
@@ -137,36 +151,42 @@ public final class GrpcStorageOptions extends StorageOptions {
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setClock(ApiClock clock) {
       super.setClock(clock);
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setProjectId(String projectId) {
       super.setProjectId(projectId);
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setHost(String host) {
       super.setHost(host);
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setCredentials(Credentials credentials) {
       super.setCredentials(credentials);
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setRetrySettings(RetrySettings retrySettings) {
       super.setRetrySettings(retrySettings);
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setServiceRpcFactory(
         ServiceRpcFactory<StorageOptions> serviceRpcFactory) {
@@ -174,30 +194,35 @@ public final class GrpcStorageOptions extends StorageOptions {
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setHeaderProvider(HeaderProvider headerProvider) {
       super.setHeaderProvider(headerProvider);
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setClientLibToken(String clientLibToken) {
       super.setClientLibToken(clientLibToken);
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions.Builder setQuotaProjectId(String quotaProjectId) {
       super.setQuotaProjectId(quotaProjectId);
       return this;
     }
 
+    @BetaApi
     @Override
     public GrpcStorageOptions build() {
       return new GrpcStorageOptions(this, defaults());
     }
   }
 
+  @BetaApi
   public static final class GrpcStorageDefaults extends StorageDefaults {
     static final GrpcStorageDefaults INSTANCE = new GrpcStorageOptions.GrpcStorageDefaults();
     static final StorageFactory STORAGE_FACTORY = new GrpcStorageFactory();
@@ -205,21 +230,25 @@ public final class GrpcStorageOptions extends StorageOptions {
 
     private GrpcStorageDefaults() {}
 
+    @BetaApi
     @Override
     public StorageFactory getDefaultServiceFactory() {
       return STORAGE_FACTORY;
     }
 
+    @BetaApi
     @Override
     public StorageRpcFactory getDefaultRpcFactory() {
       return STORAGE_RPC_FACTORY;
     }
 
+    @BetaApi
     @Override
     public GrpcTransportOptions getDefaultTransportOptions() {
       return GrpcTransportOptions.newBuilder().build();
     }
 
+    @BetaApi
     public StorageRetryStrategy getStorageRetryStrategy() {
       return StorageRetryStrategy.getDefaultStorageRetryStrategy();
     }
@@ -266,6 +295,7 @@ public final class GrpcStorageOptions extends StorageOptions {
           int p = uri.getPort() > 0 ? uri.getPort() : uri.getScheme().equals("http") ? 80 : 443;
           String hp = String.format("%s:%d", h, p);
           StorageSettings.Builder builder = StorageSettings.newBuilder().setEndpoint(hp);
+          builder.setCredentialsProvider(FixedCredentialsProvider.create(options.getCredentials()));
           if (!"storage.googleapis.com:443".equals(hp)) { // TODO: make this more formal
             options =
                 options
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
index 0c6960f8..23d1a8f5 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -34,6 +34,7 @@ import com.google.cloud.storage.Acl.Entity;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
@@ -1878,6 +1879,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return a complete bucket
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Bucket create(BucketInfo bucketInfo, BucketTargetOption... options);
 
   /**
@@ -1896,6 +1898,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return a {@code Blob} with complete information
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob create(BlobInfo blobInfo, BlobTargetOption... options);
 
   /**
@@ -1919,6 +1922,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws StorageException upon failure
    * @see <a href="https://cloud.google.com/storage/docs/hashes-etags">Hashes and ETags</a>
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob create(BlobInfo blobInfo, byte[] content, BlobTargetOption... options);
 
   /**
@@ -1942,6 +1946,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws StorageException upon failure
    * @see <a href="https://cloud.google.com/storage/docs/hashes-etags">Hashes and ETags</a>
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob create(
       BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options);
 
@@ -1985,6 +1990,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws StorageException upon failure
    */
   @Deprecated
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options);
 
   /**
@@ -2011,6 +2017,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws StorageException on server side error
    * @see #createFrom(BlobInfo, Path, int, BlobWriteOption...)
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob createFrom(BlobInfo blobInfo, Path path, BlobWriteOption... options) throws IOException;
 
   /**
@@ -2044,6 +2051,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws IOException on I/O error
    * @throws StorageException on server side error
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob createFrom(BlobInfo blobInfo, Path path, int bufferSize, BlobWriteOption... options)
       throws IOException;
 
@@ -2071,6 +2079,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws StorageException on server side error
    * @see #createFrom(BlobInfo, InputStream, int, BlobWriteOption...)
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob createFrom(BlobInfo blobInfo, InputStream content, BlobWriteOption... options)
       throws IOException;
 
@@ -2095,6 +2104,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws IOException on I/O error
    * @throws StorageException on server side error
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob createFrom(
       BlobInfo blobInfo, InputStream content, int bufferSize, BlobWriteOption... options)
       throws IOException;
@@ -2117,6 +2127,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Bucket get(String bucket, BucketGetOption... options);
 
   /**
@@ -2140,6 +2151,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return a {@code Bucket} object of the locked bucket
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Bucket lockRetentionPolicy(BucketInfo bucket, BucketTargetOption... options);
 
   /**
@@ -2161,6 +2173,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob get(String bucket, String blob, BlobGetOption... options);
 
   /**
@@ -2197,6 +2210,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob get(BlobId blob, BlobGetOption... options);
 
   /**
@@ -2213,6 +2227,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Blob get(BlobId blob);
 
   /**
@@ -2233,6 +2248,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Page<Bucket> list(BucketListOption... options);
 
   /**
@@ -2255,6 +2271,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Page<Blob> list(String bucket, BlobListOption... options);
 
   /**
@@ -2274,6 +2291,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return the updated bucket
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Bucket update(BucketInfo bucketInfo, BucketTargetOption... options);
 
   /**
@@ -2313,6 +2331,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @see <a
    *     href="https://cloud.google.com/storage/docs/json_api/v1/objects/update">https://cloud.google.com/storage/docs/json_api/v1/objects/update</a>
    */
+  @TransportCompatibility({Transport.HTTP})
   Blob update(BlobInfo blobInfo, BlobTargetOption... options);
 
   /**
@@ -2345,6 +2364,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @see <a
    *     href="https://cloud.google.com/storage/docs/json_api/v1/objects/update">https://cloud.google.com/storage/docs/json_api/v1/objects/update</a>
    */
+  @TransportCompatibility({Transport.HTTP})
   Blob update(BlobInfo blobInfo);
 
   /**
@@ -2371,6 +2391,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return {@code true} if bucket was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   boolean delete(String bucket, BucketSourceOption... options);
 
   /**
@@ -2395,6 +2416,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return {@code true} if blob was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   boolean delete(String bucket, String blob, BlobSourceOption... options);
 
   /**
@@ -2422,6 +2444,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return {@code true} if blob was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   boolean delete(BlobId blob, BlobSourceOption... options);
 
   /**
@@ -2444,6 +2467,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return {@code true} if blob was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   boolean delete(BlobId blob);
 
   /**
@@ -2472,6 +2496,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return the composed blob
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Blob compose(ComposeRequest composeRequest);
 
   /**
@@ -2538,6 +2563,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws StorageException upon failure
    * @see <a href="https://cloud.google.com/storage/docs/json_api/v1/objects/rewrite">Rewrite</a>
    */
+  @TransportCompatibility({Transport.HTTP})
   CopyWriter copy(CopyRequest copyRequest);
 
   /**
@@ -2557,6 +2583,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return the blob's content
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   byte[] readAllBytes(String bucket, String blob, BlobSourceOption... options);
 
   /**
@@ -2586,6 +2613,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return the blob's content
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   byte[] readAllBytes(BlobId blob, BlobSourceOption... options);
 
   /**
@@ -2615,6 +2643,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * Blob blob = result.get(); // returns get result or throws StorageException
    * }</pre>
    */
+  @TransportCompatibility(Transport.HTTP)
   StorageBatch batch();
 
   /**
@@ -2639,6 +2668,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   ReadChannel reader(String bucket, String blob, BlobSourceOption... options);
 
   /**
@@ -2671,6 +2701,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   ReadChannel reader(BlobId blob, BlobSourceOption... options);
 
   /**
@@ -2690,6 +2721,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   void downloadTo(BlobId blob, Path path, BlobSourceOption... options);
 
   /**
@@ -2710,6 +2742,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param outputStream
    * @param options
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options);
 
   /**
@@ -2734,6 +2767,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   WriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options);
 
   /**
@@ -2758,6 +2792,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility(Transport.HTTP)
   WriteChannel writer(URL signedURL);
 
   /**
@@ -2870,6 +2905,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @throws SigningException if the attempt to sign the URL failed
    * @see <a href="https://cloud.google.com/storage/docs/access-control#Signed-URLs">Signed-URLs</a>
    */
+  @TransportCompatibility(Transport.HTTP)
   URL signUrl(BlobInfo blobInfo, long duration, TimeUnit unit, SignUrlOption... options);
 
   /**
@@ -2921,6 +2957,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *     href="https://cloud.google.com/storage/docs/xml-api/post-object#usage_and_examples">POST
    *     Object</a>
    */
+  @TransportCompatibility(Transport.HTTP)
   PostPolicyV4 generateSignedPostPolicyV4(
       BlobInfo blobInfo,
       long duration,
@@ -2934,6 +2971,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * conditions. See full documentation for {@link #generateSignedPostPolicyV4(BlobInfo, long,
    * TimeUnit, PostPolicyV4.PostFieldsV4, PostPolicyV4.PostConditionsV4, PostPolicyV4Option...)}.
    */
+  @TransportCompatibility(Transport.HTTP)
   PostPolicyV4 generateSignedPostPolicyV4(
       BlobInfo blobInfo,
       long duration,
@@ -2946,6 +2984,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * See full documentation for {@link #generateSignedPostPolicyV4(BlobInfo, long, TimeUnit,
    * PostPolicyV4.PostFieldsV4, PostPolicyV4.PostConditionsV4, PostPolicyV4Option...)}.
    */
+  @TransportCompatibility(Transport.HTTP)
   PostPolicyV4 generateSignedPostPolicyV4(
       BlobInfo blobInfo,
       long duration,
@@ -2959,6 +2998,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * #generateSignedPostPolicyV4(BlobInfo, long, TimeUnit, PostPolicyV4.PostFieldsV4,
    * PostPolicyV4.PostConditionsV4, PostPolicyV4Option...)}.
    */
+  @TransportCompatibility(Transport.HTTP)
   PostPolicyV4 generateSignedPostPolicyV4(
       BlobInfo blobInfo, long duration, TimeUnit unit, PostPolicyV4Option... options);
 
@@ -2981,6 +3021,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *     been denied the corresponding item in the list is {@code null}.
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Blob> get(BlobId... blobIds);
 
   /**
@@ -3003,6 +3044,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *     been denied the corresponding item in the list is {@code null}.
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Blob> get(Iterable<BlobId> blobIds);
 
   /**
@@ -3029,6 +3071,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *     been denied the corresponding item in the list is {@code null}.
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Blob> update(BlobInfo... blobInfos);
 
   /**
@@ -3056,6 +3099,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *     been denied the corresponding item in the list is {@code null}.
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Blob> update(Iterable<BlobInfo> blobInfos);
 
   /**
@@ -3078,6 +3122,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *     was denied the corresponding item is {@code false}.
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Boolean> delete(BlobId... blobIds);
 
   /**
@@ -3101,6 +3146,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *     was denied the corresponding item is {@code false}.
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Boolean> delete(Iterable<BlobId> blobIds);
 
   /**
@@ -3129,9 +3175,11 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options extra parameters to apply to this operation
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl getAcl(String bucket, Entity entity, BucketSourceOption... options);
 
   /** @see #getAcl(String, Entity, BucketSourceOption...) */
+  @TransportCompatibility({Transport.HTTP})
   Acl getAcl(String bucket, Entity entity);
 
   /**
@@ -3164,9 +3212,11 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return {@code true} if the ACL was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... options);
 
   /** @see #deleteAcl(String, Entity, BucketSourceOption...) */
+  @TransportCompatibility({Transport.HTTP})
   boolean deleteAcl(String bucket, Entity entity);
 
   /**
@@ -3192,9 +3242,11 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options extra parameters to apply to this operation
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl createAcl(String bucket, Acl acl, BucketSourceOption... options);
 
   /** @see #createAcl(String, Acl, BucketSourceOption...) */
+  @TransportCompatibility({Transport.HTTP})
   Acl createAcl(String bucket, Acl acl);
 
   /**
@@ -3220,9 +3272,11 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options extra parameters to apply to this operation
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options);
 
   /** @see #updateAcl(String, Acl, BucketSourceOption...) */
+  @TransportCompatibility({Transport.HTTP})
   Acl updateAcl(String bucket, Acl acl);
 
   /**
@@ -3253,9 +3307,11 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options any number of BucketSourceOptions to apply to this operation
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Acl> listAcls(String bucket, BucketSourceOption... options);
 
   /** @see #listAcls(String, BucketSourceOption...) */
+  @TransportCompatibility({Transport.HTTP})
   List<Acl> listAcls(String bucket);
 
   /**
@@ -3274,6 +3330,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl getDefaultAcl(String bucket, Entity entity);
 
   /**
@@ -3297,6 +3354,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return {@code true} if the ACL was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   boolean deleteDefaultAcl(String bucket, Entity entity);
 
   /**
@@ -3315,6 +3373,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl createDefaultAcl(String bucket, Acl acl);
 
   /**
@@ -3333,6 +3392,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl updateDefaultAcl(String bucket, Acl acl);
 
   /**
@@ -3353,6 +3413,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Acl> listDefaultAcls(String bucket);
 
   /**
@@ -3381,6 +3442,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl getAcl(BlobId blob, Entity entity);
 
   /**
@@ -3404,6 +3466,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return {@code true} if the ACL was deleted, {@code false} if it was not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   boolean deleteAcl(BlobId blob, Entity entity);
 
   /**
@@ -3431,6 +3494,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl createAcl(BlobId blob, Acl acl);
 
   /**
@@ -3448,6 +3512,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Acl updateAcl(BlobId blob, Acl acl);
 
   /**
@@ -3468,6 +3533,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Acl> listAcls(BlobId blob);
 
   /**
@@ -3487,6 +3553,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   HmacKey createHmacKey(ServiceAccount serviceAccount, CreateHmacKeyOption... options);
 
   /**
@@ -3521,6 +3588,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options the options to apply to this operation
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   Page<HmacKeyMetadata> listHmacKeys(ListHmacKeysOption... options);
 
   /**
@@ -3537,6 +3605,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   HmacKeyMetadata getHmacKey(String accessId, GetHmacKeyOption... options);
 
   /**
@@ -3556,6 +3625,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   void deleteHmacKey(HmacKeyMetadata hmacKeyMetadata, DeleteHmacKeyOption... options);
 
   /**
@@ -3572,6 +3642,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    *
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   HmacKeyMetadata updateHmacKeyState(
       final HmacKeyMetadata hmacKeyMetadata,
       final HmacKey.HmacKeyState state,
@@ -3596,6 +3667,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options extra parameters to apply to this operation
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Policy getIamPolicy(String bucket, BucketSourceOption... options);
 
   /**
@@ -3623,6 +3695,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options extra parameters to apply to this operation
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Policy setIamPolicy(String bucket, Policy policy, BucketSourceOption... options);
 
   /**
@@ -3647,6 +3720,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @param options extra parameters to apply to this operation
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Boolean> testIamPermissions(
       String bucket, List<String> permissions, BucketSourceOption... options);
 
@@ -3664,6 +3738,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return the service account associated with this project
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP, Transport.GRPC})
   ServiceAccount getServiceAccount(String projectId);
 
   /**
@@ -3687,6 +3762,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return the created notification
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Notification createNotification(String bucket, NotificationInfo notificationInfo);
 
   /**
@@ -3705,6 +3781,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return the {@code Notification} object with the given id or {@code null} if not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   Notification getNotification(String bucket, String notificationId);
 
   /**
@@ -3721,6 +3798,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return a list of {@link Notification} objects added to the bucket.
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   List<Notification> listNotifications(String bucket);
 
   /**
@@ -3744,6 +3822,7 @@ public interface Storage extends Service<StorageOptions>, AutoCloseable {
    * @return {@code true} if the notification has been deleted, {@code false} if not found
    * @throws StorageException upon failure
    */
+  @TransportCompatibility({Transport.HTTP})
   boolean deleteNotification(String bucket, String notificationId);
 
   @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
index 862fc1c4..7073cb32 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage;
 
+import com.google.api.core.BetaApi;
 import com.google.cloud.NoCredentials;
 import com.google.cloud.ServiceDefaults;
 import com.google.cloud.ServiceOptions;
@@ -23,6 +24,7 @@ import com.google.cloud.http.HttpTransportOptions;
 import com.google.cloud.storage.HttpStorageOptions.HttpStorageDefaults;
 import com.google.cloud.storage.HttpStorageOptions.HttpStorageFactory;
 import com.google.cloud.storage.HttpStorageOptions.HttpStorageRpcFactory;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.StorageRpcFactory;
 
 public abstract class StorageOptions extends ServiceOptions<Storage, StorageOptions> {
@@ -95,23 +97,29 @@ public abstract class StorageOptions extends ServiceOptions<Storage, StorageOpti
   public abstract boolean equals(Object obj);
 
   /** Returns a default {@code StorageOptions} instance. */
+  @TransportCompatibility(Transport.HTTP)
   public static StorageOptions getDefaultInstance() {
     return HttpStorageOptions.newBuilder().build();
   }
 
   /** Returns a unauthenticated {@code StorageOptions} instance. */
+  @TransportCompatibility(Transport.HTTP)
   public static StorageOptions getUnauthenticatedInstance() {
     return HttpStorageOptions.newBuilder().setCredentials(NoCredentials.getInstance()).build();
   }
 
+  @TransportCompatibility(Transport.HTTP)
   public static StorageOptions.Builder newBuilder() {
     return http();
   }
 
+  @TransportCompatibility(Transport.GRPC)
   public static HttpStorageOptions.Builder http() {
     return HttpStorageOptions.newBuilder();
   }
 
+  @BetaApi
+  @TransportCompatibility(Transport.GRPC)
   public static GrpcStorageOptions.Builder grpc() {
     return GrpcStorageOptions.newBuilder();
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/TransportCompatibility.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/TransportCompatibility.java
new file mode 100644
index 00000000..c5c8dea9
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/TransportCompatibility.java
@@ -0,0 +1,62 @@
+/*
+ * 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 java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Annotation which is used to convey which Cloud Storage API a class or method has compatibility
+ * with.
+ *
+ * <p>Not all operations are compatible with all transports.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE, ElementType.METHOD})
+@Documented
+@Inherited
+@interface TransportCompatibility {
+
+  Transport[] value();
+
+  /**
+   * Enum representing the transports {@code com.google.cloud.storage} classes have implementations
+   * for.
+   */
+  enum Transport {
+    /**
+     * Value indicating use of the <a target="_blank" rel="noopener noreferrer"
+     * href="https://cloud.google.com/storage/docs/json_api">Cloud Storage JSON API</a>
+     *
+     * @see StorageOptions#http()
+     */
+    HTTP,
+
+    /**
+     * Value indicating usa of the <a target="_blank" rel="noopener noreferrer"
+     * href="https://github.com/googleapis/googleapis/blob/master/google/storage/v2/storage.proto">Cloud
+     * Storage v2 gRPC API</a> TODO: link to public docs when published.
+     *
+     * @see StorageOptions#grpc()
+     */
+    GRPC
+  }
+}


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BaseConvertablePropertyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BaseConvertablePropertyTest.java
index 5070a7c6..4e020a73 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BaseConvertablePropertyTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BaseConvertablePropertyTest.java
@@ -37,7 +37,9 @@ abstract class BaseConvertablePropertyTest<ModelT, ProtoT extends Message> {
   final void edgeCases() {
     TypeUsage baseTypeUsage = findBaseTypeUsage(this.getClass());
     TypeUsage protoTType = baseTypeUsage.getTypeArgument(1);
-    report(protoTType);
+    if (!CIUtils.isRunningOnGitHubActions() || CIUtils.isJobTypeIntegration()) {
+      report(protoTType);
+    }
   }
 
   /**
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/CIUtils.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/CIUtils.java
new file mode 100644
index 00000000..1352337f
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/CIUtils.java
@@ -0,0 +1,42 @@
+/*
+ * 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;
+
+public final class CIUtils {
+
+  private CIUtils() {}
+
+  public static boolean isRunningInCI() {
+    return isJobTypeUnit() || isJobTypeIntegration();
+  }
+
+  public static boolean isJobTypeUnit() {
+    return isJobTypeEq("test");
+  }
+
+  public static boolean isJobTypeIntegration() {
+    return isJobTypeEq("integration");
+  }
+
+  public static boolean isRunningOnGitHubActions() {
+    return System.getenv("GITHUB_JOB") != null;
+  }
+
+  private static boolean isJobTypeEq(String integration) {
+    return integration.equals(System.getenv("JOB_TYPE"));
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/DefaultBufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/DefaultBufferedWritableByteChannelTest.java
new file mode 100644
index 00000000..3dac1b5d
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/DefaultBufferedWritableByteChannelTest.java
@@ -0,0 +1,570 @@
+/*
+ * 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.cloud.storage.ChunkSegmenterTest.TestData.fmt;
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth.assertWithMessage;
+import static org.junit.Assert.assertThrows;
+
+import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.common.collect.ImmutableList;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.WritableByteChannel;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.List;
+import java.util.Objects;
+import net.jqwik.api.Arbitraries;
+import net.jqwik.api.Arbitrary;
+import net.jqwik.api.Combinators;
+import net.jqwik.api.Example;
+import net.jqwik.api.ForAll;
+import net.jqwik.api.Property;
+import net.jqwik.api.Provide;
+import net.jqwik.api.providers.TypeUsage;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+public final class DefaultBufferedWritableByteChannelTest {
+
+  @Example
+  void edgeCases() {
+    JqwikTest.report(TypeUsage.of(WriteOps.class), arbitraryWriteOps());
+  }
+
+  @Property
+  void bufferingEagerlyFlushesWhenFull(@ForAll("WriteOps") WriteOps writeOps) throws IOException {
+    ByteBuffer buffer = ByteBuffer.allocate(writeOps.bufferSize);
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        CountingWritableByteChannelAdapter adapter =
+            new CountingWritableByteChannelAdapter(Channels.newChannel(baos));
+        BufferedWritableByteChannel c = new DefaultBufferedWritableByteChannel(buffer, adapter)) {
+
+      List<Integer> actualWriteSizes = new ArrayList<>();
+
+      for (ByteBuffer buf : writeOps.writes) {
+        int write = c.write(buf);
+        actualWriteSizes.add(write);
+      }
+
+      c.close();
+      assertThrows(ClosedChannelException.class, () -> c.write(null));
+
+      assertWithMessage("Unexpected write size")
+          .that(actualWriteSizes)
+          .isEqualTo(writeOps.writeSizes);
+      assertWithMessage("Unexpected total flushed length")
+          .that(adapter.writeEndPoints)
+          .isEqualTo(writeOps.expectedFlushes);
+      assertThat(baos.toByteArray()).isEqualTo(writeOps.bytes);
+    }
+  }
+
+  /**
+   * Scenario A:
+   *
+   * <p>Data size, and write size are smaller than buffer size
+   */
+  @Example
+  void scenario_a() throws IOException {
+    bufferingEagerlyFlushesWhenFull(WriteOps.of(1, 2, 1));
+  }
+
+  /** Scenario B: Data size and buffer size are equal, while write size may be larger than both */
+  @Example
+  void scenario_b() throws IOException {
+    bufferingEagerlyFlushesWhenFull(WriteOps.of(1, 1, 2));
+  }
+
+  /**
+   * Scenario C:
+   *
+   * <ul>
+   *   <li>data size is evenly divisible by buffer size and write size
+   *   <li>buffer size is larger than write size
+   *   <li>buffer size is not evenly divisible by write size
+   * </ul>
+   */
+  @Example
+  void scenario_c() throws IOException {
+    bufferingEagerlyFlushesWhenFull(WriteOps.of(105, 15, 7));
+  }
+
+  /**
+   * Scenario D:
+   *
+   * <ul>
+   *   <li>write and buffer size are smaller than data
+   *   <li>data size is not evenly divisible by either write size nor buffer size
+   *   <li>buffer size is smaller than write size
+   *   <li>write size is not evenly divisible by buffer size
+   * </ul>
+   */
+  @Example
+  void scenario_d() throws IOException {
+    bufferingEagerlyFlushesWhenFull(WriteOps.of(61, 3, 16));
+  }
+
+  /**
+   * Scenario E:
+   *
+   * <p>Some flushes are only partially consumed. Ensure we proceed with consuming the buffer
+   * provided to {@code write}
+   *
+   * <pre>
+   *           0                        27
+   * data:    |--------------------------|
+   *               5       14 17        27
+   * writes:  |----|--------|--|---------|
+   *                   10
+   * flush 1: |---------|
+   *            2        12
+   * flush 2:   |---------|
+   *                     12        22
+   * flush 3:             |---------|
+   *                              19    27
+   * flush 4:                     |------|
+   * </pre>
+   */
+  @Example
+  void partialFlushOfEnqueuedBytesFlushesMultipleTimes() throws IOException {
+    ByteBuffer buffer = ByteBuffer.allocate(10);
+
+    ByteBuffer data1 = DataGenerator.base64Characters().genByteBuffer(5);
+    ByteBuffer data2 = DataGenerator.base64Characters().genByteBuffer(9);
+    ByteBuffer data3 = DataGenerator.base64Characters().genByteBuffer(3);
+    ByteBuffer data4 = DataGenerator.base64Characters().genByteBuffer(10);
+
+    ImmutableList<ByteBuffer> buffers = ImmutableList.of(data1, data2, data3, data4);
+
+    int allDataSize = buffers.stream().mapToInt(ByteBuffer::remaining).sum();
+    byte[] allData =
+        buffers.stream().reduce(ByteBuffer.allocate(allDataSize), ByteBuffer::put).array();
+    buffers.forEach(b -> b.position(0));
+
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        CountingWritableByteChannelAdapter adapter =
+            new CountingWritableByteChannelAdapter(Channels.newChannel(baos));
+        BufferedWritableByteChannel c = new DefaultBufferedWritableByteChannel(buffer, adapter)) {
+
+      c.write(data1); // write 5 bytes, which should enqueue in full
+      // before the next write, limit the number of bytes the underlying channel will consume to 2.
+      adapter.nextWriteMaxConsumptionLimit = 2L;
+      // write 9 bytes, which should trigger a flush - limited to 2 bytes, leaving 3 bytes in the
+      // buffer and not consuming any of the 9 bytes. Since 3 + 9 is still larger than our buffer
+      // attempt another flush of 10 bytes which will all be consumed. Enqueue the remaining 2
+      // bytes from data2.
+      c.write(data2);
+
+      // write 3 bytes, which should enqueue in full, leaving the buffer with 5 bytes enqueued
+      c.write(data3);
+      // before the next write, limit the number of bytes the underlying channel will consume to 7.
+      adapter.nextWriteMaxConsumptionLimit = 7L;
+      // write 10 bytes, which should trigger a flush - limited to 7 bytes, consuming all of the
+      // buffer, but only consuming 2 bytes written data. The remaining 8 bytes should be
+      // enqueued in full.
+      c.write(data4);
+
+      // close the channel, causing a flush of the 8 outstanding bytes in buffer.
+      c.close();
+      assertThrows(ClosedChannelException.class, () -> c.write(null));
+
+      assertWithMessage("Unexpected total flushed length")
+          .that(adapter.writeEndPoints)
+          .isEqualTo(ImmutableList.of(2L, 12L, 19L, 27L));
+      assertThat(baos.toByteArray()).isEqualTo(allData);
+    }
+  }
+
+  /**
+   * Ensure manually calling flush works.
+   *
+   * <pre>
+   *           0         12
+   * data:    |-----------|
+   *             3  6  9
+   * writes:  |--|--|--|--|
+   *             3
+   * flush 1: |--|
+   *             3  6
+   * flush 2:    |--|
+   *               5   10
+   * flush 3:      |----|
+   *                   10 12
+   * flush 4:           |-|
+   * </pre>
+   */
+  @Example
+  void manualFlushingIsAccurate() throws IOException {
+    ByteBuffer buffer = ByteBuffer.allocate(5);
+
+    ByteBuffer data1 = DataGenerator.base64Characters().genByteBuffer(3);
+    ByteBuffer data2 = DataGenerator.base64Characters().genByteBuffer(3);
+    ByteBuffer data3 = DataGenerator.base64Characters().genByteBuffer(3);
+    ByteBuffer data4 = DataGenerator.base64Characters().genByteBuffer(3);
+
+    ImmutableList<ByteBuffer> buffers = ImmutableList.of(data1, data2, data3, data4);
+
+    int allDataSize = buffers.stream().mapToInt(ByteBuffer::remaining).sum();
+    byte[] allData =
+        buffers.stream().reduce(ByteBuffer.allocate(allDataSize), ByteBuffer::put).array();
+    buffers.forEach(b -> b.position(0));
+
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        CountingWritableByteChannelAdapter adapter =
+            new CountingWritableByteChannelAdapter(Channels.newChannel(baos));
+        BufferedWritableByteChannel c = new DefaultBufferedWritableByteChannel(buffer, adapter)) {
+
+      c.write(data1); // write 3 bytes, which should enqueue in full
+      c.flush(); // flush all enqueued bytes
+
+      c.write(data2); // write 3 bytes, which should enqueue in full
+      // before we call flush, limit how many bytes the underlying channel will consume to 2.
+      // This should leave 1 byte in the buffer
+      adapter.nextWriteMaxConsumptionLimit = 2L;
+      c.flush(); // attempt to flush all enqueued bytes, however only 2 of the 3 will be consumed
+      c.write(data3); // write 3 bytes, which should enqueue in full
+      // after this write, our buffer should contain 4 bytes of its 5 byte capacity
+      // on the next write, 5 bytes should be flushed. 4 from the buffer, 1 from the written data
+      c.write(data4); // 1 of the 3 bytes will be flushed, leaving 2 bytes in the buffer
+
+      c.close(); // close the channel, which should flush the 2 outstanding buffered bytes
+      assertThrows(ClosedChannelException.class, () -> c.write(null));
+
+      assertWithMessage("Unexpected total flushed length")
+          .that(adapter.writeEndPoints)
+          .isEqualTo(ImmutableList.of(3L, 5L, 10L, 12L));
+      assertThat(baos.toByteArray()).isEqualTo(allData);
+    }
+  }
+
+  @Provide("WriteOps")
+  static Arbitrary<WriteOps> arbitraryWriteOps() {
+    return Combinators.combine(
+            Arbitraries.integers().between(1, 256 * 1024),
+            Arbitraries.integers().between(1, 16 * 1024),
+            Arbitraries.integers().between(1, 64 * 1024))
+        .as(WriteOps::of);
+  }
+
+  /**
+   *
+   *
+   * <pre>
+   *           0                                                                                                     105
+   * data:    |--------------------------------------------------------------------------------------------------------|
+   *                 7     14     21     28     35     42     49     56     63     70     77     84     91     98    105
+   * writes:  |------|------|------|------|------|------|------|------|------|------|------|------|------|------|------|
+   *                        15             30             45             60             75             90            105
+   * flushes: |--------------|--------------|--------------|--------------|--------------|--------------|--------------|
+   * </pre>
+   */
+  @Example
+  void writeOpsOfGeneratesAccurately_1() {
+    int dataSize = 105;
+    int bufferSize = 15;
+    int writeSize = 7;
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(dataSize);
+    ImmutableList<ByteBuffer> writes =
+        ImmutableList.of(
+            ByteBuffer.wrap(bytes, 0, writeSize),
+            ByteBuffer.wrap(bytes, 7, writeSize),
+            ByteBuffer.wrap(bytes, 14, writeSize),
+            ByteBuffer.wrap(bytes, 21, writeSize),
+            ByteBuffer.wrap(bytes, 28, writeSize),
+            ByteBuffer.wrap(bytes, 35, writeSize),
+            ByteBuffer.wrap(bytes, 42, writeSize),
+            ByteBuffer.wrap(bytes, 49, writeSize),
+            ByteBuffer.wrap(bytes, 56, writeSize),
+            ByteBuffer.wrap(bytes, 63, writeSize),
+            ByteBuffer.wrap(bytes, 70, writeSize),
+            ByteBuffer.wrap(bytes, 77, writeSize),
+            ByteBuffer.wrap(bytes, 84, writeSize),
+            ByteBuffer.wrap(bytes, 91, writeSize),
+            ByteBuffer.wrap(bytes, 98, writeSize));
+    ImmutableList<Long> flushes = ImmutableList.of(15L, 30L, 45L, 60L, 75L, 90L, 105L);
+    String z = "[0x00000007 * 0x0000000f]";
+    WriteOps expected = new WriteOps(bytes, bufferSize, writeSize, writes, flushes, z);
+    assertThat(WriteOps.of(dataSize, bufferSize, writeSize)).isEqualTo(expected);
+  }
+
+  /**
+   *
+   *
+   * <pre>
+   *           0                                                          61
+   * data:    |------------------------------------------------------------|
+   *                         16         (16) 32         (16) 48      (13) 61
+   * writes:  |---------------|---------------|---------------|------------|
+   *             3  6  9 12 15 18 21 24 27 30 33 36 39 42 45 48 51 54 57 60
+   * flushes: |--|--|--|--|--|--|--|--|--|--|--|--|--|--|--|--|--|--|--|--||
+   * </pre>
+   */
+  @Example
+  void writeOpsOfGeneratesAccurately_2() {
+    int dataSize = 61;
+    int bufferSize = 3;
+    int writeSize = 16;
+    byte[] bytes = DataGenerator.base64Characters().genBytes(dataSize);
+    ImmutableList<ByteBuffer> writes =
+        ImmutableList.of(
+            ByteBuffer.wrap(bytes, 0, writeSize),
+            ByteBuffer.wrap(bytes, 16, writeSize),
+            ByteBuffer.wrap(bytes, 32, writeSize),
+            ByteBuffer.wrap(bytes, 48, 13));
+    ImmutableList<Long> flushes =
+        ImmutableList.of(
+            3L, 6L, 9L, 12L, 15L, 18L, 21L, 24L, 27L, 30L, 33L, 36L, 39L, 42L, 45L, 48L, 51L, 54L,
+            57L, 60L, 61L);
+    String z = "[0x00000010 * 0x00000003, 0x0000000d]";
+    WriteOps expected = new WriteOps(bytes, bufferSize, writeSize, writes, flushes, z);
+    WriteOps actual = WriteOps.of(dataSize, bufferSize, writeSize);
+    assertThat(actual).isEqualTo(expected);
+  }
+
+  private static final class WriteOps {
+    private final byte[] bytes;
+    private final int bufferSize;
+    private final int writeSize;
+    private final ImmutableList<Integer> writeSizes;
+    private final ImmutableList<ByteBuffer> writes;
+    private final ImmutableList<Long> expectedFlushes;
+    private final String dbgExpectedWriteSizes;
+
+    public WriteOps(
+        byte[] bytes,
+        int bufferSize,
+        int writeSize,
+        ImmutableList<ByteBuffer> writes,
+        ImmutableList<Long> expectedFlushes,
+        String dbgExpectedWriteSizes) {
+      this.bytes = bytes;
+      this.bufferSize = bufferSize;
+      this.writeSize = writeSize;
+      this.writeSizes =
+          writes.stream().map(ByteBuffer::remaining).collect(ImmutableList.toImmutableList());
+      this.writes = writes;
+      this.expectedFlushes = expectedFlushes;
+      this.dbgExpectedWriteSizes = dbgExpectedWriteSizes;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof WriteOps)) {
+        return false;
+      }
+      WriteOps writeOps = (WriteOps) o;
+      return bufferSize == writeOps.bufferSize
+          && writeSize == writeOps.writeSize
+          && Arrays.equals(bytes, writeOps.bytes)
+          && Objects.equals(writes, writeOps.writes)
+          && Objects.equals(expectedFlushes, writeOps.expectedFlushes)
+          && Objects.equals(dbgExpectedWriteSizes, writeOps.dbgExpectedWriteSizes);
+    }
+
+    @Override
+    public int hashCode() {
+      int result =
+          Objects.hash(bufferSize, writeSize, writes, expectedFlushes, dbgExpectedWriteSizes);
+      result = 31 * result + Arrays.hashCode(bytes);
+      return result;
+    }
+
+    @Override
+    public String toString() {
+      return "WriteOps{"
+          + "bytes.length="
+          + fmt(bytes.length)
+          + ", bufferSize="
+          + fmt(bufferSize)
+          + ", writeSize="
+          + fmt(writeSize)
+          + ", writes.size()="
+          + fmt(writes.size())
+          + ", expectedFlushes.size()="
+          + fmt(expectedFlushes.size())
+          + ", expectedWriteSizes="
+          + dbgExpectedWriteSizes
+          + '}';
+    }
+
+    @NonNull
+    static WriteOps of(int byteSize, int bufferSize, int writeSize) {
+      byte[] bytes = DataGenerator.base64Characters().genBytes(byteSize);
+
+      List<ByteBuffer> writes = new ArrayList<>();
+      Deque<Long> expectedFlushes = new ArrayDeque<>();
+
+      int length = bytes.length;
+
+      int fullWriteCount = 0;
+      int remainingWrite = 0;
+      int prevWriteEndOffset = 0;
+      for (int i = 1; i <= length; i++) {
+        boolean flushBoundary = (i % bufferSize == 0) || bufferSize == 1;
+        boolean writeBoundary = (i % writeSize == 0) || writeSize == 1;
+        boolean eof = i == length;
+
+        if (flushBoundary) {
+          expectedFlushes.addLast((long) i);
+        }
+
+        if (writeBoundary) {
+          writes.add(ByteBuffer.wrap(bytes, prevWriteEndOffset, writeSize));
+          fullWriteCount++;
+          prevWriteEndOffset += writeSize;
+        }
+
+        if (eof) {
+          // We expect a flush during close in the following scenarios:
+          // the buffer size is larger than our data size (peekLast == null)
+          // data size is not evenly divisible by bufferSize
+          if (expectedFlushes.peekLast() == null || expectedFlushes.peekLast() != length) {
+            expectedFlushes.addLast((long) length);
+          }
+
+          // If the data size is not evenly divisible by writeSize we will have an extra
+          // smaller write
+          if (prevWriteEndOffset != length) {
+            int writeLen = Math.min(length - prevWriteEndOffset, writeSize);
+            writes.add(ByteBuffer.wrap(bytes, prevWriteEndOffset, writeLen));
+            remainingWrite = writeLen;
+            prevWriteEndOffset += writeLen;
+          }
+        }
+      }
+
+      String dbgExpectedWriteSizes;
+      if (fullWriteCount > 0 && remainingWrite > 0) {
+        dbgExpectedWriteSizes =
+            String.format(
+                "[%s * %s, %s]", fmt(writeSize), fmt(fullWriteCount), fmt(remainingWrite));
+      } else if (remainingWrite > 0) {
+        dbgExpectedWriteSizes = String.format("[%s]", fmt(remainingWrite));
+      } else {
+        dbgExpectedWriteSizes = String.format("[%s * %s]", fmt(writeSize), fmt(fullWriteCount));
+      }
+      return new WriteOps(
+          bytes,
+          bufferSize,
+          writeSize,
+          ImmutableList.copyOf(writes),
+          ImmutableList.copyOf(expectedFlushes),
+          dbgExpectedWriteSizes);
+    }
+  }
+
+  /**
+   * Adapter to make any {@link WritableByteChannel} into an {@link UnbufferedWritableByteChannel}
+   */
+  private static final class CountingWritableByteChannelAdapter
+      implements UnbufferedWritableByteChannel {
+
+    private final WritableByteChannel c;
+
+    private final List<Long> writeEndPoints;
+    private long totalBytesWritten;
+
+    private long nextWriteMaxConsumptionLimit = Long.MAX_VALUE;
+
+    private CountingWritableByteChannelAdapter(WritableByteChannel c) {
+      this.c = c;
+      writeEndPoints = new ArrayList<>();
+    }
+
+    @Override
+    public boolean isComplete() {
+      return c.isOpen();
+    }
+
+    @Override
+    public int write(ByteBuffer src) throws IOException {
+      return Math.toIntExact(write(new ByteBuffer[] {src}, 0, 1));
+    }
+
+    @Override
+    public long write(ByteBuffer[] srcs) throws IOException {
+      return write(srcs, 0, srcs.length);
+    }
+
+    @Override
+    public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+      if (!c.isOpen()) {
+        return -1;
+      }
+
+      long budgetRemaining = nextWriteMaxConsumptionLimit;
+      nextWriteMaxConsumptionLimit = Long.MAX_VALUE;
+
+      long bytesWriten = 0;
+      for (int i = offset; i < length && budgetRemaining > 0; i++) {
+        ByteBuffer src = srcs[i];
+        if (src.hasRemaining()) {
+          ByteBuffer slice = src.slice();
+          int remaining = src.remaining();
+          int newLimit = Math.toIntExact(Math.min(budgetRemaining, remaining));
+          slice.limit(newLimit);
+          int write = c.write(slice);
+          if (write == -1) {
+            if (bytesWriten == 0) {
+              c.close();
+              return -1;
+            } else {
+              break;
+            }
+          } else if (write == 0) {
+            break;
+          } else {
+            src.position(src.position() + write);
+          }
+          budgetRemaining -= write;
+          bytesWriten += write;
+        }
+      }
+      incr(bytesWriten);
+      return bytesWriten;
+    }
+
+    @Override
+    public boolean isOpen() {
+      return c.isOpen();
+    }
+
+    @Override
+    public void close() throws IOException {
+      c.close();
+    }
+
+    private void incr(long bytesWritten) {
+      if (bytesWritten > 0) {
+        totalBytesWritten += bytesWritten;
+        writeEndPoints.add(totalBytesWritten);
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java
new file mode 100644
index 00000000..e2042b01
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TransportCompatibilityTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.cloud.NoCredentials;
+import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
+import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.common.collect.ImmutableList;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+import org.junit.Test;
+
+public final class TransportCompatibilityTest {
+
+  @Test
+  public void verifyUnsupportedMethodsGenerateMeaningfulException() {
+    Storage s =
+        StorageOptions.grpc()
+            .setProjectId("blank")
+            .setCredentials(NoCredentials.getInstance())
+            .build()
+            .getService();
+    ImmutableList<String> messages =
+        Stream.<Supplier<?>>of(
+                s::batch,
+                () -> s.writer(null),
+                () -> s.signUrl(null, 0, null),
+                () -> s.generateSignedPostPolicyV4(null, 0, null, null, null, null),
+                () -> s.generateSignedPostPolicyV4(null, 0, null, (PostFieldsV4) null),
+                () -> s.generateSignedPostPolicyV4(null, 0, null, (PostConditionsV4) null),
+                () -> s.generateSignedPostPolicyV4(null, 0, null))
+            .map(
+                sup -> {
+                  try {
+                    sup.get();
+                    return null;
+                  } catch (UnsupportedOperationException e) {
+                    return e.getMessage();
+                  }
+                })
+            .collect(ImmutableList.toImmutableList());
+
+    for (String message : messages) {
+      assertThat(message).contains("only supported for " + Transport.HTTP);
+      assertThat(message)
+          .contains("Please use StorageOptions.http() to construct a compatible instance");
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/GracefulConformanceEnforcement.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/GracefulConformanceEnforcement.java
index 7d792c42..0b11f83d 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/GracefulConformanceEnforcement.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/GracefulConformanceEnforcement.java
@@ -18,6 +18,7 @@ package com.google.cloud.storage.conformance.retry;
 
 import static org.junit.Assert.assertNotNull;
 
+import com.google.cloud.storage.CIUtils;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.io.CharStreams;
 import java.io.IOException;
@@ -64,7 +65,7 @@ final class GracefulConformanceEnforcement implements TestRule {
           if (!testNamesWhichCanFail.contains(testName)) {
             throw t;
           } else {
-            if (isRunningInCI()) {
+            if (CIUtils.isRunningInCI()) {
               throw new AssumptionViolatedException(
                   String.format(
                       "Test %s is allowed to fail, downgrading failure to ignored.", testName),
@@ -78,11 +79,6 @@ final class GracefulConformanceEnforcement implements TestRule {
     };
   }
 
-  private static boolean isRunningInCI() {
-    return "test".equals(System.getenv("JOB_TYPE"))
-        || "integration".equals(System.getenv("JOB_TYPE"));
-  }
-
   private static Set<String> loadTestNamesWhichCanFail() {
     ClassLoader cl = Thread.currentThread().getContextClassLoader();
     InputStream inputStream =