Repository: googleapis/java-storage
Pull request #1844 (2023-01-11 21:41:32)

chore(test): increase debug logging for failure cases in GapicUnbufferedWritableByteChannelTest
In very rare cases GapicUnbufferedWritableByteChannelTest#resumableUpload will fail with a permission denied error while running in CI. Thus far I've not be able to reproduce the failure locally even after running the tests thousands of times. This PR adds a log message with the expected vs resolved state in the case of failure in the hope we will be able to get more information the next time it does fail in CI.

Related to https://github.com/googleapis/java-storage/issues/1748


Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java
index a8b87186..9249e585 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryConversions.java
@@ -78,6 +78,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.function.Function;
 
 @InternalApi
@@ -373,7 +374,14 @@ final class ApiaryConversions {
         from.getDefaultKmsKeyName(),
         k -> new Encryption().setDefaultKmsKeyName(k),
         to::setEncryption);
-    ifNonNull(from.getLabels(), to::setLabels);
+    Map<String, String> pbLabels = from.getLabels();
+    if (pbLabels != null && !Data.isNull(pbLabels)) {
+      pbLabels = Maps.newHashMapWithExpectedSize(from.getLabels().size());
+      for (Map.Entry<String, String> entry : from.getLabels().entrySet()) {
+        pbLabels.put(entry.getKey(), firstNonNull(entry.getValue(), Data.nullOf(String.class)));
+      }
+    }
+    to.setLabels(pbLabels);
     maybeEncodeRetentionPolicy(from, to);
     ifNonNull(from.getIamConfiguration(), this::iamConfigEncode, to::setIamConfiguration);
     ifNonNull(from.getAutoclass(), this::autoclassEncode, to::setAutoclass);
@@ -412,7 +420,7 @@ final class ApiaryConversions {
         lift(Lifecycle::getRule).andThen(toImmutableListOf(lifecycleRule()::decode)),
         to::setLifecycleRules);
     ifNonNull(from.getDefaultEventBasedHold(), to::setDefaultEventBasedHold);
-    ifNonNull(from.getLabels(), to::setLabels);
+    ifNonNull(from.getLabels(), ApiaryConversions::replaceDataNullValuesWithNull, to::setLabels);
     ifNonNull(from.getBilling(), Billing::getRequesterPays, to::setRequesterPays);
     Encryption encryption = from.getEncryption();
     if (encryption != null
@@ -917,4 +925,21 @@ final class ApiaryConversions {
           to::setRetentionPeriodDuration);
     }
   }
+
+  private static Map<String, String> replaceDataNullValuesWithNull(Map<String, String> labels) {
+    boolean anyDataNull = labels.values().stream().anyMatch(Data::isNull);
+    if (anyDataNull) {
+      // If we received any Data null values, replace them with null before setting.
+      // Explicitly use a HashMap as it allows null values.
+      Map<String, String> tmp = Maps.newHashMapWithExpectedSize(labels.size());
+      for (Entry<String, String> e : labels.entrySet()) {
+        String k = e.getKey();
+        String v = e.getValue();
+        tmp.put(k, Data.isNull(v) ? null : v);
+      }
+      return Collections.unmodifiableMap(tmp);
+    } else {
+      return labels;
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java
new file mode 100644
index 00000000..c11c3845
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java
@@ -0,0 +1,149 @@
+/*
+ * Copyright 2023 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.ByteSizeConstants._2MiB;
+import static java.util.Objects.requireNonNull;
+
+import com.google.cloud.storage.BufferedReadableByteChannelSession.BufferedReadableByteChannel;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+abstract class BaseStorageReadChannel<T> implements StorageReadChannel {
+
+  private ByteRangeSpec byteRangeSpec;
+  private int chunkSize = _2MiB;
+  private BufferHandle bufferHandle;
+  private LazyReadChannel<T> lazyReadChannel;
+
+  @Nullable private T resolvedObject;
+
+  protected BaseStorageReadChannel() {
+    this.byteRangeSpec = ByteRangeSpec.nullRange();
+  }
+
+  @Override
+  public final synchronized void setChunkSize(int chunkSize) {
+    StorageException.wrapIOException(() -> maybeResetChannel(true));
+    this.chunkSize = chunkSize;
+  }
+
+  @Override
+  public final synchronized boolean isOpen() {
+    if (lazyReadChannel == null) {
+      return true;
+    } else {
+      LazyReadChannel<T> tmp = internalGetLazyChannel();
+      return tmp.isOpen();
+    }
+  }
+
+  @Override
+  public final synchronized void close() {
+    if (internalGetLazyChannel().isOpen()) {
+      StorageException.wrapIOException(internalGetLazyChannel().getChannel()::close);
+    }
+  }
+
+  @Override
+  public final synchronized StorageReadChannel setByteRangeSpec(ByteRangeSpec byteRangeSpec) {
+    requireNonNull(byteRangeSpec, "byteRangeSpec must be non null");
+    StorageException.wrapIOException(() -> maybeResetChannel(false));
+    this.byteRangeSpec = byteRangeSpec;
+    return this;
+  }
+
+  @Override
+  public final ByteRangeSpec getByteRangeSpec() {
+    return byteRangeSpec;
+  }
+
+  @Override
+  public final synchronized int read(ByteBuffer dst) throws IOException {
+    long diff = byteRangeSpec.length();
+    if (diff <= 0) {
+      close();
+      return -1;
+    }
+    try {
+      int read = internalGetLazyChannel().getChannel().read(dst);
+      if (read != -1) {
+        byteRangeSpec = byteRangeSpec.withShiftBeginOffset(read);
+      } else {
+        close();
+      }
+      return read;
+    } catch (StorageException e) {
+      if (e.getCode() == 416) {
+        // HttpStorageRpc turns 416 into a null etag with an empty byte array, leading
+        // BlobReadChannel to believe it read 0 bytes, returning -1 and leaving the channel open.
+        // Emulate that same behavior here to preserve behavior compatibility, though this should
+        // be removed in the next major version.
+        return -1;
+      } else {
+        throw new IOException(e);
+      }
+    } catch (IOException e) {
+      throw e;
+    } catch (Exception e) {
+      throw new IOException(StorageException.coalesce(e));
+    }
+  }
+
+  protected final BufferHandle getBufferHandle() {
+    if (bufferHandle == null) {
+      bufferHandle = BufferHandle.allocate(chunkSize);
+    }
+    return bufferHandle;
+  }
+
+  protected final int getChunkSize() {
+    return chunkSize;
+  }
+
+  @Nullable
+  protected T getResolvedObject() {
+    return resolvedObject;
+  }
+
+  protected void setResolvedObject(@Nullable T resolvedObject) {
+    this.resolvedObject = resolvedObject;
+  }
+
+  protected abstract LazyReadChannel<T> newLazyReadChannel();
+
+  private void maybeResetChannel(boolean freeBuffer) throws IOException {
+    if (lazyReadChannel != null && lazyReadChannel.isOpen()) {
+      try (BufferedReadableByteChannel ignore = lazyReadChannel.getChannel()) {
+        if (bufferHandle != null && !freeBuffer) {
+          bufferHandle.get().clear();
+        } else if (freeBuffer) {
+          bufferHandle = null;
+        }
+        lazyReadChannel = null;
+      }
+    }
+  }
+
+  private LazyReadChannel<T> internalGetLazyChannel() {
+    if (lazyReadChannel == null) {
+      lazyReadChannel = newLazyReadChannel();
+    }
+    return lazyReadChannel;
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java
index 95716f89..d33f3140 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobInfo.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.storage.BackwardCompatibilityUtils.millisOffsetDateTimeCodec;
+import static com.google.cloud.storage.Utils.diffMaps;
 import static com.google.common.base.MoreObjects.firstNonNull;
 import static com.google.common.base.Preconditions.checkNotNull;
 
@@ -24,6 +25,7 @@ import com.google.api.client.util.Data;
 import com.google.api.core.BetaApi;
 import com.google.cloud.storage.Storage.BlobField;
 import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.UnifiedOpts.NamedField;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -38,6 +40,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * Information about an object in Google Cloud Storage. A {@code BlobInfo} object includes the
@@ -100,7 +104,7 @@ public class BlobInfo implements Serializable {
   private final Boolean eventBasedHold;
   private final Boolean temporaryHold;
   private final OffsetDateTime retentionExpirationTime;
-  private final transient ImmutableSet<Storage.BlobField> modifiedFields;
+  private final transient ImmutableSet<NamedField> modifiedFields;
 
   /** This class is meant for internal use only. Users are discouraged from using this class. */
   public static final class ImmutableEmptyMap<K, V> extends AbstractMap<K, V> {
@@ -331,7 +335,7 @@ public class BlobInfo implements Serializable {
     }
 
     /** Sets the blob's user provided metadata. */
-    public abstract Builder setMetadata(Map<String, String> metadata);
+    public abstract Builder setMetadata(@Nullable Map<@NonNull String, @Nullable String> metadata);
 
     abstract Builder setMetageneration(Long metageneration);
 
@@ -502,7 +506,7 @@ public class BlobInfo implements Serializable {
     private Boolean eventBasedHold;
     private Boolean temporaryHold;
     private OffsetDateTime retentionExpirationTime;
-    private final ImmutableSet.Builder<Storage.BlobField> modifiedFields = ImmutableSet.builder();
+    private final ImmutableSet.Builder<NamedField> modifiedFields = ImmutableSet.builder();
 
     BuilderImpl(BlobId blobId) {
       this.blobId = blobId;
@@ -757,15 +761,18 @@ public class BlobInfo implements Serializable {
       return this;
     }
 
+    @SuppressWarnings({"UnnecessaryLocalVariable"})
     @Override
-    public Builder setMetadata(Map<String, String> metadata) {
-      if (!Objects.equals(this.metadata, metadata)) {
-        modifiedFields.add(BlobField.METADATA);
-      }
-      if (metadata != null) {
-        this.metadata = new HashMap<>(metadata);
-      } else {
-        this.metadata = (Map<String, String>) Data.nullOf(ImmutableEmptyMap.class);
+    public Builder setMetadata(@Nullable Map<@NonNull String, @Nullable String> metadata) {
+      Map<String, String> left = this.metadata;
+      Map<String, String> right = metadata;
+      if (!Objects.equals(left, right)) {
+        diffMaps(BlobField.METADATA, left, right, modifiedFields::add);
+        if (right != null) {
+          this.metadata = new HashMap<>(right);
+        } else {
+          this.metadata = (Map<String, String>) Data.nullOf(ImmutableEmptyMap.class);
+        }
       }
       return this;
     }
@@ -1317,7 +1324,8 @@ public class BlobInfo implements Serializable {
   }
 
   /** Returns blob's user provided metadata. */
-  public Map<String, String> getMetadata() {
+  @Nullable
+  public Map<@NonNull String, @Nullable String> getMetadata() {
     return metadata == null || Data.isNull(metadata) ? null : Collections.unmodifiableMap(metadata);
   }
 
@@ -1617,7 +1625,7 @@ public class BlobInfo implements Serializable {
         && Objects.equals(retentionExpirationTime, blobInfo.retentionExpirationTime);
   }
 
-  ImmutableSet<BlobField> getModifiedFields() {
+  ImmutableSet<NamedField> getModifiedFields() {
     return modifiedFields;
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java
index e768b04e..05b3513a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java
@@ -16,36 +16,23 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.ByteSizeConstants._2MiB;
-import static java.util.Objects.requireNonNull;
-
 import com.google.api.services.storage.Storage;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RestorableState;
 import com.google.cloud.storage.ApiaryUnbufferedReadableByteChannel.ApiaryReadRequest;
-import com.google.cloud.storage.BufferedReadableByteChannelSession.BufferedReadableByteChannel;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.base.MoreObjects;
-import java.io.IOException;
 import java.io.Serializable;
-import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.Objects;
 
-final class BlobReadChannelV2 implements StorageReadChannel {
+final class BlobReadChannelV2 extends BaseStorageReadChannel<StorageObject> {
 
   private final StorageObject storageObject;
   private final Map<StorageRpc.Option, ?> opts;
   private final BlobReadChannelContext blobReadChannelContext;
 
-  private LazyReadChannel<StorageObject> lazyReadChannel;
-  private StorageObject resolvedObject;
-  private ByteRangeSpec byteRangeSpec;
-
-  private int chunkSize = _2MiB;
-  private BufferHandle bufferHandle;
-
   BlobReadChannelV2(
       StorageObject storageObject,
       Map<StorageRpc.Option, ?> opts,
@@ -53,127 +40,30 @@ final class BlobReadChannelV2 implements StorageReadChannel {
     this.storageObject = storageObject;
     this.opts = opts;
     this.blobReadChannelContext = blobReadChannelContext;
-    this.byteRangeSpec = ByteRangeSpec.nullRange();
-  }
-
-  @Override
-  public synchronized void setChunkSize(int chunkSize) {
-    StorageException.wrapIOException(() -> maybeResetChannel(true));
-    this.chunkSize = chunkSize;
-  }
-
-  @Override
-  public synchronized boolean isOpen() {
-    if (lazyReadChannel == null) {
-      return true;
-    } else {
-      LazyReadChannel<StorageObject> tmp = internalGetLazyChannel();
-      return tmp.isOpen();
-    }
-  }
-
-  @Override
-  public synchronized void close() {
-    if (internalGetLazyChannel().isOpen()) {
-      StorageException.wrapIOException(internalGetLazyChannel().getChannel()::close);
-    }
-  }
-
-  @Override
-  public synchronized StorageReadChannel setByteRangeSpec(ByteRangeSpec byteRangeSpec) {
-    requireNonNull(byteRangeSpec, "byteRangeSpec must be non null");
-    StorageException.wrapIOException(() -> maybeResetChannel(false));
-    this.byteRangeSpec = byteRangeSpec;
-    return this;
-  }
-
-  @Override
-  public ByteRangeSpec getByteRangeSpec() {
-    return byteRangeSpec;
-  }
-
-  @Override
-  public synchronized int read(ByteBuffer dst) throws IOException {
-    long diff = byteRangeSpec.length();
-    if (diff <= 0) {
-      close();
-      return -1;
-    }
-    try {
-      int read = internalGetLazyChannel().getChannel().read(dst);
-      if (read != -1) {
-        byteRangeSpec = byteRangeSpec.withShiftBeginOffset(read);
-      } else {
-        close();
-      }
-      return read;
-    } catch (StorageException e) {
-      if (e.getCode() == 416) {
-        // HttpStorageRpc turns 416 into a null etag with an empty byte array, leading
-        // BlobReadChannel to believe it read 0 bytes, returning -1 and leaving the channel open.
-        // Emulate that same behavior here to preserve behavior compatibility, though this should
-        // be removed in the next major version.
-        return -1;
-      } else {
-        throw new IOException(e);
-      }
-    } catch (IOException e) {
-      throw e;
-    } catch (Exception e) {
-      throw new IOException(StorageException.coalesce(e));
-    }
   }
 
   @Override
   public RestorableState<ReadChannel> capture() {
     ApiaryReadRequest apiaryReadRequest = getApiaryReadRequest();
     return new BlobReadChannelV2State(
-        apiaryReadRequest, blobReadChannelContext.getStorageOptions(), chunkSize);
+        apiaryReadRequest, blobReadChannelContext.getStorageOptions(), getChunkSize());
   }
 
-  private void maybeResetChannel(boolean umallocBuffer) throws IOException {
-    if (lazyReadChannel != null && lazyReadChannel.isOpen()) {
-      try (BufferedReadableByteChannel ignore = lazyReadChannel.getChannel()) {
-        if (bufferHandle != null && !umallocBuffer) {
-          bufferHandle.get().clear();
-        } else if (umallocBuffer) {
-          bufferHandle = null;
-        }
-        lazyReadChannel = null;
-      }
-    }
-  }
-
-  private LazyReadChannel<StorageObject> internalGetLazyChannel() {
-    if (lazyReadChannel == null) {
-      lazyReadChannel = newLazyReadChannel();
-    }
-    return lazyReadChannel;
-  }
-
-  private LazyReadChannel<StorageObject> newLazyReadChannel() {
+  protected LazyReadChannel<StorageObject> newLazyReadChannel() {
     return new LazyReadChannel<>(
-        () -> {
-          if (bufferHandle == null) {
-            bufferHandle = BufferHandle.allocate(chunkSize);
-          }
-          return ResumableMedia.http()
-              .read()
-              .byteChannel(blobReadChannelContext)
-              .setCallback(this::setResolvedObject)
-              .buffered(bufferHandle)
-              .setApiaryReadRequest(getApiaryReadRequest())
-              .build();
-        });
-  }
-
-  private void setResolvedObject(StorageObject resolvedObject) {
-    this.resolvedObject = resolvedObject;
+        () ->
+            ResumableMedia.http()
+                .read()
+                .byteChannel(blobReadChannelContext)
+                .setCallback(this::setResolvedObject)
+                .buffered(getBufferHandle())
+                .setApiaryReadRequest(getApiaryReadRequest())
+                .build());
   }
 
   private ApiaryReadRequest getApiaryReadRequest() {
-    StorageObject object = resolvedObject != null ? resolvedObject : storageObject;
-    return new ApiaryReadRequest(object, opts, byteRangeSpec);
+    StorageObject object = getResolvedObject() != null ? getResolvedObject() : storageObject;
+    return new ApiaryReadRequest(object, opts, getByteRangeSpec());
   }
 
   static class BlobReadChannelV2State implements RestorableState<ReadChannel>, Serializable {
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 6147552c..af5c2cc2 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
@@ -45,6 +45,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * A Google cloud storage bucket.
@@ -546,7 +547,7 @@ public class Bucket extends BucketInfo {
     }
 
     @Override
-    public Builder setLabels(Map<String, String> labels) {
+    public Builder setLabels(@Nullable Map<@NonNull String, @Nullable String> labels) {
       infoBuilder.setLabels(labels);
       return this;
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
index 2a2c7256..d328fa9f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java
@@ -19,6 +19,7 @@ package com.google.cloud.storage;
 import static com.google.cloud.storage.BackwardCompatibilityUtils.millisOffsetDateTimeCodec;
 import static com.google.cloud.storage.BackwardCompatibilityUtils.millisUtcCodec;
 import static com.google.cloud.storage.BackwardCompatibilityUtils.nullableDurationSecondsCodec;
+import static com.google.cloud.storage.Utils.diffMaps;
 import static com.google.common.base.MoreObjects.firstNonNull;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.collect.Lists.newArrayList;
@@ -29,12 +30,13 @@ import com.google.api.client.util.DateTime;
 import com.google.api.core.BetaApi;
 import com.google.api.services.storage.model.Bucket.Lifecycle.Rule;
 import com.google.cloud.storage.Acl.Entity;
+import com.google.cloud.storage.BlobInfo.ImmutableEmptyMap;
 import com.google.cloud.storage.Storage.BucketField;
 import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.UnifiedOpts.NamedField;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Maps;
 import com.google.common.collect.Streams;
 import java.io.IOException;
 import java.io.ObjectInputStream;
@@ -43,6 +45,7 @@ import java.io.Serializable;
 import java.time.Duration;
 import java.time.OffsetDateTime;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -100,7 +103,7 @@ public class BucketInfo implements Serializable {
   private final String location;
   private final Rpo rpo;
   private final StorageClass storageClass;
-  private final Map<String, String> labels;
+  @Nullable private final Map<String, String> labels;
   private final String defaultKmsKeyName;
   private final Boolean defaultEventBasedHold;
   private final OffsetDateTime retentionEffectiveTime;
@@ -111,7 +114,7 @@ public class BucketInfo implements Serializable {
   private final String locationType;
   private final Logging logging;
   private final CustomPlacementConfig customPlacementConfig;
-  private final transient ImmutableSet<Storage.BucketField> modifiedFields;
+  private final transient ImmutableSet<NamedField> modifiedFields;
 
   /**
    * non-private for backward compatibility on message class. log messages are now emitted from
@@ -1477,7 +1480,7 @@ public class BucketInfo implements Serializable {
     public abstract Builder setDefaultAcl(Iterable<Acl> acl);
 
     /** Sets the label of this bucket. */
-    public abstract Builder setLabels(Map<String, String> labels);
+    public abstract Builder setLabels(@Nullable Map<@NonNull String, @Nullable String> labels);
 
     /** Sets the default Cloud KMS key name for this bucket. */
     public abstract Builder setDefaultKmsKeyName(String defaultKmsKeyName);
@@ -1630,7 +1633,7 @@ public class BucketInfo implements Serializable {
     private String locationType;
     private Logging logging;
     private CustomPlacementConfig customPlacementConfig;
-    private final ImmutableSet.Builder<Storage.BucketField> modifiedFields = ImmutableSet.builder();
+    private final ImmutableSet.Builder<NamedField> modifiedFields = ImmutableSet.builder();
 
     BuilderImpl(String name) {
       this.name = name;
@@ -1909,20 +1912,18 @@ public class BucketInfo implements Serializable {
       return this;
     }
 
+    @SuppressWarnings("UnnecessaryLocalVariable")
     @Override
-    public Builder setLabels(Map<String, String> labels) {
-      if (labels != null) {
-        Map<String, String> tmp =
-            Maps.transformValues(
-                labels,
-                input -> {
-                  // replace null values with empty strings
-                  return input == null ? Data.nullOf(String.class) : input;
-                });
-        if (!Objects.equals(this.labels, tmp)) {
-          modifiedFields.add(BucketField.LABELS);
+    public Builder setLabels(@Nullable Map<@NonNull String, @Nullable String> labels) {
+      Map<String, String> left = this.labels;
+      Map<String, String> right = labels;
+      if (!Objects.equals(left, right)) {
+        diffMaps(BucketField.LABELS, left, right, modifiedFields::add);
+        if (right != null) {
+          this.labels = new HashMap<>(right);
+        } else {
+          this.labels = (Map<String, String>) Data.nullOf(ImmutableEmptyMap.class);
         }
-        this.labels = tmp;
       }
       return this;
     }
@@ -2483,7 +2484,8 @@ public class BucketInfo implements Serializable {
   }
 
   /** Returns the labels for this bucket. */
-  public Map<String, String> getLabels() {
+  @Nullable
+  public Map<@NonNull String, @Nullable String> getLabels() {
     return labels;
   }
 
@@ -2692,7 +2694,7 @@ public class BucketInfo implements Serializable {
     return new Bucket(storage, new BucketInfo.BuilderImpl(this));
   }
 
-  ImmutableSet<BucketField> getModifiedFields() {
+  ImmutableSet<NamedField> getModifiedFields() {
     return modifiedFields;
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java
index 737e1a82..3b501eee 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java
@@ -19,6 +19,7 @@ package com.google.cloud.storage;
 import com.google.api.core.InternalApi;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.MoreObjects.ToStringHelper;
+import com.google.storage.v2.ReadObjectRequest;
 import java.io.Serializable;
 import java.util.Objects;
 import java.util.function.BiFunction;
@@ -80,6 +81,8 @@ abstract class ByteRangeSpec implements Serializable {
     return httpRangeHeader;
   }
 
+  abstract ReadObjectRequest.Builder seekReadObjectRequest(ReadObjectRequest.Builder b);
+
   @Nullable
   protected abstract String fmtAsHttpRangeHeader() throws ArithmeticException;
 
@@ -212,6 +215,11 @@ abstract class ByteRangeSpec implements Serializable {
       }
     }
 
+    @Override
+    public ReadObjectRequest.Builder seekReadObjectRequest(ReadObjectRequest.Builder b) {
+      return b.setReadOffset(beginOffset()).setReadLimit(length());
+    }
+
     @Override
     protected String fmtAsHttpRangeHeader() throws ArithmeticException {
       return String.format("bytes=%d-%d", beginOffset, endOffsetInclusive());
@@ -292,6 +300,11 @@ abstract class ByteRangeSpec implements Serializable {
       return new RelativeByteRangeSpec(beginOffset, relativeLength);
     }
 
+    @Override
+    public ReadObjectRequest.Builder seekReadObjectRequest(ReadObjectRequest.Builder b) {
+      return b.setReadOffset(beginOffset()).setReadLimit(length());
+    }
+
     @Override
     protected String fmtAsHttpRangeHeader() throws ArithmeticException {
       return String.format("bytes=%d-%d", beginOffset, endOffsetInclusive());
@@ -372,6 +385,11 @@ abstract class ByteRangeSpec implements Serializable {
       return new RelativeByteRangeSpec(beginOffset, relativeLength);
     }
 
+    @Override
+    public ReadObjectRequest.Builder seekReadObjectRequest(ReadObjectRequest.Builder b) {
+      return b.setReadOffset(beginOffset()).setReadLimit(length());
+    }
+
     @Override
     protected String fmtAsHttpRangeHeader() throws ArithmeticException {
       return String.format("bytes=%d-%d", beginOffset, endOffsetInclusive);
@@ -445,6 +463,11 @@ abstract class ByteRangeSpec implements Serializable {
       return new RelativeByteRangeSpec(beginOffset, relativeLength);
     }
 
+    @Override
+    public ReadObjectRequest.Builder seekReadObjectRequest(ReadObjectRequest.Builder b) {
+      return b.setReadOffset(beginOffset());
+    }
+
     @Override
     protected String fmtAsHttpRangeHeader() throws ArithmeticException {
       if (beginOffset > 0) {
@@ -530,6 +553,11 @@ abstract class ByteRangeSpec implements Serializable {
       }
     }
 
+    @Override
+    public ReadObjectRequest.Builder seekReadObjectRequest(ReadObjectRequest.Builder b) {
+      return b;
+    }
+
     @Override
     protected String fmtAsHttpRangeHeader() throws ArithmeticException {
       return null;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
index ba9b5b50..cd183118 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
@@ -16,78 +16,27 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.storage.ByteSizeConstants._16MiB;
-import static com.google.cloud.storage.Maths.sub;
-import static com.google.cloud.storage.StorageV2ProtoUtils.seekReadObjectRequest;
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkState;
-
-import com.google.api.core.ApiFuture;
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RestorableState;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ReadObjectRequest;
 import com.google.storage.v2.ReadObjectResponse;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-final class GrpcBlobReadChannel implements ReadChannel {
+import org.checkerframework.checker.nullness.qual.NonNull;
 
-  private final LazyReadChannel<Object> lazyReadChannel;
+final class GrpcBlobReadChannel extends BaseStorageReadChannel<Object> {
 
-  private Long position;
-  private Long limit;
-  private int chunkSize = _16MiB;
+  private final ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read;
+  private final ReadObjectRequest request;
+  private final boolean autoGzipDecompression;
 
   GrpcBlobReadChannel(
       ServerStreamingCallable<ReadObjectRequest, ReadObjectResponse> read,
       ReadObjectRequest request,
       boolean autoGzipDecompression) {
-    this.lazyReadChannel =
-        new LazyReadChannel<>(
-            () -> {
-              ReadObjectRequest req =
-                  seekReadObjectRequest(request, position, sub(limit, position));
-              return ResumableMedia.gapic()
-                  .read()
-                  .byteChannel(read)
-                  .setHasher(Hasher.noop())
-                  .setAutoGzipDecompression(autoGzipDecompression)
-                  .buffered(BufferHandle.allocate(chunkSize))
-                  .setReadObjectRequest(req)
-                  .build();
-            });
-  }
-
-  @Override
-  public void setChunkSize(int chunkSize) {
-    checkState(!isOpen(), "Unable to change chunkSize after read");
-    this.chunkSize = chunkSize;
-  }
-
-  @Override
-  public boolean isOpen() {
-    return lazyReadChannel.isOpen() && lazyReadChannel.getChannel().isOpen();
-  }
-
-  @Override
-  public void close() {
-    if (lazyReadChannel.isOpen()) {
-      try {
-        lazyReadChannel.getChannel().close();
-      } catch (IOException e) {
-        // why does ReadChannel remove IOException?!
-        throw StorageException.coalesce(e);
-      }
-    }
-  }
-
-  @Override
-  public void seek(long position) throws IOException {
-    checkArgument(position >= 0, "position must be >= 0");
-    checkState(!isOpen(), "Unable to change position after read");
-    this.position = position;
+    this.read = read;
+    this.request = request;
+    this.autoGzipDecompression = autoGzipDecompression;
   }
 
   @Override
@@ -96,35 +45,29 @@ final class GrpcBlobReadChannel implements ReadChannel {
   }
 
   @Override
-  public ReadChannel limit(long limit) {
-    checkArgument(limit >= 0, "limit must be >= 0");
-    checkState(!isOpen(), "Unable to change limit after read");
-    this.limit = limit;
-    return this;
-  }
-
-  @Override
-  public long limit() {
-    return limit != null ? limit : Long.MAX_VALUE;
+  protected LazyReadChannel<Object> newLazyReadChannel() {
+    return new LazyReadChannel<>(
+        () ->
+            ResumableMedia.gapic()
+                .read()
+                .byteChannel(read)
+                .setHasher(Hasher.noop())
+                .setAutoGzipDecompression(autoGzipDecompression)
+                .buffered(getBufferHandle())
+                .setReadObjectRequest(getReadObjectRequest())
+                .build());
   }
 
-  @Override
-  public int read(ByteBuffer dst) throws IOException {
-    Long diff = sub(limit, position);
-    if (diff != null && diff <= 0) {
-      close();
-      return -1;
-    }
-    try {
-      return lazyReadChannel.getChannel().read(dst);
-    } catch (IOException e) {
-      throw e;
-    } catch (Exception e) {
-      throw new IOException(StorageException.coalesce(e));
+  @NonNull
+  private ReadObjectRequest getReadObjectRequest() {
+    ByteRangeSpec rangeSpec = getByteRangeSpec();
+    ReadObjectRequest.Builder b = request.toBuilder();
+    if (request.getGeneration() == 0) {
+      Object resolvedObject = getResolvedObject();
+      if (resolvedObject != null) {
+        b.setGeneration(resolvedObject.getGeneration());
+      }
     }
-  }
-
-  ApiFuture<Object> getResults() {
-    return lazyReadChannel.getSession().get().getResult();
+    return rangeSpec.seekReadObjectRequest(b).build();
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
index 7e5e8c7e..57f2edf6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcConversions.java
@@ -333,7 +333,7 @@ final class GrpcConversions {
       to.setVersioning(versioningBuilder.build());
     }
     ifNonNull(from.getDefaultEventBasedHold(), to::setDefaultEventBasedHold);
-    ifNonNull(from.getLabels(), to::putAllLabels);
+    ifNonNull(from.getLabels(), this::removeNullValues, to::putAllLabels);
     // Do not use, #getLifecycleRules, it can not return null, which is important to our logic here
     List<? extends LifecycleRule> lifecycleRules = from.lifecycleRules;
     if (lifecycleRules != null) {
@@ -714,8 +714,12 @@ final class GrpcConversions {
               .collect(ImmutableList.toImmutableList());
       conditionBuilder.setMatchesStorageClass(collect);
     }
-    conditionBuilder.setMatchesPrefix(condition.getMatchesPrefixList());
-    conditionBuilder.setMatchesSuffix(condition.getMatchesSuffixList());
+    if (!condition.getMatchesPrefixList().isEmpty()) {
+      conditionBuilder.setMatchesPrefix(condition.getMatchesPrefixList());
+    }
+    if (!condition.getMatchesSuffixList().isEmpty()) {
+      conditionBuilder.setMatchesSuffix(condition.getMatchesSuffixList());
+    }
     return new BucketInfo.LifecycleRule(lifecycleAction, conditionBuilder.build());
   }
 
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 97389f92..b15f62bd 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
@@ -61,6 +61,7 @@ import com.google.cloud.storage.UnifiedOpts.HmacKeyListOpt;
 import com.google.cloud.storage.UnifiedOpts.HmacKeySourceOpt;
 import com.google.cloud.storage.UnifiedOpts.HmacKeyTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Mapper;
+import com.google.cloud.storage.UnifiedOpts.NamedField;
 import com.google.cloud.storage.UnifiedOpts.ObjectListOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
@@ -490,7 +491,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         .getUpdateMaskBuilder()
         .addAllPaths(
             bucketInfo.getModifiedFields().stream()
-                .map(BucketField::getGrpcName)
+                .map(NamedField::getGrpcName)
                 .collect(ImmutableList.toImmutableList()));
     UpdateBucketRequest req = builder.build();
     return Retrying.run(
@@ -512,7 +513,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         .getUpdateMaskBuilder()
         .addAllPaths(
             blobInfo.getModifiedFields().stream()
-                .map(BlobField::getGrpcName)
+                .map(NamedField::getGrpcName)
                 .collect(ImmutableList.toImmutableList()));
     UpdateObjectRequest req = builder.build();
     return Retrying.run(
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index efbbd998..cd6816e2 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -34,7 +34,6 @@ import com.google.cloud.BatchResult;
 import com.google.cloud.PageImpl;
 import com.google.cloud.PageImpl.NextPageFetcher;
 import com.google.cloud.Policy;
-import com.google.cloud.ReadChannel;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
 import com.google.cloud.storage.BlobReadChannelV2.BlobReadChannelContext;
@@ -581,12 +580,12 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   }
 
   @Override
-  public ReadChannel reader(String bucket, String blob, BlobSourceOption... options) {
+  public StorageReadChannel reader(String bucket, String blob, BlobSourceOption... options) {
     return reader(BlobId.of(bucket, blob), options);
   }
 
   @Override
-  public ReadChannel reader(BlobId blob, BlobSourceOption... options) {
+  public StorageReadChannel reader(BlobId blob, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options).resolveFrom(blob);
     StorageObject storageObject = Conversions.apiary().blobId().encode(blob);
     ImmutableMap<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageV2ProtoUtils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageV2ProtoUtils.java
index cbf49ef6..0bc8e8e1 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageV2ProtoUtils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageV2ProtoUtils.java
@@ -16,8 +16,6 @@
 
 package com.google.cloud.storage;
 
-import static com.google.common.base.Preconditions.checkArgument;
-
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.MessageOrBuilder;
 import com.google.protobuf.util.JsonFormat;
@@ -27,7 +25,6 @@ import com.google.storage.v2.ObjectAccessControl;
 import com.google.storage.v2.ReadObjectRequest;
 import java.util.function.Predicate;
 import org.checkerframework.checker.nullness.qual.NonNull;
-import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class StorageV2ProtoUtils {
 
@@ -39,44 +36,23 @@ final class StorageV2ProtoUtils {
 
   private StorageV2ProtoUtils() {}
 
+  // TODO: can we eliminate this method all together?
   @NonNull
   static ReadObjectRequest seekReadObjectRequest(
-      @NonNull ReadObjectRequest request, @Nullable Long offset, @Nullable Long limit) {
-    validate(offset, limit);
+      @NonNull ReadObjectRequest request, @NonNull ByteRangeSpec byteRangeSpec) {
+
+    long offset = byteRangeSpec.beginOffset();
+    long limit = byteRangeSpec.length();
     ReadObjectRequest req = request;
 
-    boolean setOffset = offset != null && (offset > 0 || offset != req.getReadOffset());
-    boolean setLimit = limit != null && (limit < Long.MAX_VALUE || limit != req.getReadLimit());
+    boolean setOffset = (offset > 0 && offset != req.getReadOffset());
+    boolean setLimit = (limit < ByteRangeSpec.EFFECTIVE_INFINITY && limit != req.getReadLimit());
     if (setOffset || setLimit) {
-      ReadObjectRequest.Builder b = req.toBuilder();
-      if (setOffset) {
-        b.setReadOffset(offset);
-      }
-      if (setLimit) {
-        b.setReadLimit(limit);
-      }
-      req = b.build();
+      req = byteRangeSpec.seekReadObjectRequest(req.toBuilder()).build();
     }
     return req;
   }
 
-  private static void validate(@Nullable Long offset, @Nullable Long limit) {
-    boolean offsetNull = offset == null;
-    boolean limitNull = limit == null;
-
-    if (offsetNull && limitNull) {
-      return;
-    }
-
-    if (!offsetNull) {
-      checkArgument(0 <= offset, VALIDATION_TEMPLATE, offset, limit);
-    }
-
-    if (!limitNull) {
-      checkArgument(0 <= limit, VALIDATION_TEMPLATE, offset, limit);
-    }
-  }
-
   @NonNull
   static String fmtProto(@NonNull final MessageOrBuilder msg) {
     try {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
index 6eecc74f..6e78001b 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnifiedOpts.java
@@ -2361,6 +2361,10 @@ final class UnifiedOpts {
     static NamedField literal(String name) {
       return new LiteralNamedField(name);
     }
+
+    static NamedField nested(NamedField parent, NamedField child) {
+      return new NestedNamedField(parent, child);
+    }
   }
 
   private static CommonObjectRequestParams.Builder customerSuppliedKey(
@@ -2376,7 +2380,7 @@ final class UnifiedOpts {
     private final String prefix;
     private final NamedField delegate;
 
-    public PrefixedNamedField(String prefix, NamedField delegate) {
+    private PrefixedNamedField(String prefix, NamedField delegate) {
       this.prefix = prefix;
       this.delegate = delegate;
     }
@@ -2417,11 +2421,11 @@ final class UnifiedOpts {
     }
   }
 
-  private static class LiteralNamedField implements NamedField {
+  private static final class LiteralNamedField implements NamedField {
 
     private final String name;
 
-    LiteralNamedField(String name) {
+    private LiteralNamedField(String name) {
       this.name = name;
     }
 
@@ -2457,4 +2461,46 @@ final class UnifiedOpts {
       return MoreObjects.toStringHelper(this).add("name", name).toString();
     }
   }
+
+  private static final class NestedNamedField implements NamedField {
+    private final NamedField parent;
+    private final NamedField child;
+
+    private NestedNamedField(NamedField parent, NamedField child) {
+      this.parent = parent;
+      this.child = child;
+    }
+
+    @Override
+    public String getApiaryName() {
+      return parent.getApiaryName() + "." + child.getApiaryName();
+    }
+
+    @Override
+    public String getGrpcName() {
+      return parent.getGrpcName() + "." + child.getGrpcName();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof NestedNamedField)) {
+        return false;
+      }
+      NestedNamedField that = (NestedNamedField) o;
+      return Objects.equals(parent, that.parent) && Objects.equals(child, that.child);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(parent, child);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("parent", parent).add("child", child).toString();
+    }
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
index 9a378556..b067e303 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Utils.java
@@ -21,8 +21,11 @@ import static java.util.Objects.requireNonNull;
 import com.google.api.client.util.DateTime;
 import com.google.api.core.InternalApi;
 import com.google.cloud.storage.Conversions.Codec;
+import com.google.cloud.storage.UnifiedOpts.NamedField;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.MapDifference;
+import com.google.common.collect.Maps;
 import com.google.common.io.BaseEncoding;
 import com.google.common.primitives.Ints;
 import com.google.storage.v2.BucketName;
@@ -34,10 +37,12 @@ import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Stream;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -234,6 +239,39 @@ final class Utils {
     throw new IllegalStateException("Unable to resolve non-null value");
   }
 
+  /**
+   * Diff two maps, and append each differing key to {@code sink} with the parent of {{@code parent}
+   */
+  @SuppressWarnings("ConstantValue")
+  static void diffMaps(
+      NamedField parent,
+      Map<String, String> left,
+      Map<String, String> right,
+      Consumer<NamedField> sink) {
+    final Stream<String> keys;
+    if (left != null && right == null) {
+      keys = left.keySet().stream();
+    } else if (left == null && right != null) {
+      keys = right.keySet().stream();
+    } else if (left != null && right != null) {
+      MapDifference<String, String> difference = Maps.difference(left, right);
+      keys =
+          Stream.of(
+                  // keys with modified values
+                  difference.entriesDiffering().keySet().stream(),
+                  // Only include keys to remove if ALL keys were removed
+                  right.isEmpty()
+                      ? difference.entriesOnlyOnLeft().keySet().stream()
+                      : Stream.<String>empty(),
+                  // new keys
+                  difference.entriesOnlyOnRight().keySet().stream())
+              .flatMap(x -> x);
+    } else {
+      keys = Stream.empty();
+    }
+    keys.map(NamedField::literal).map(k -> NamedField.nested(parent, k)).forEach(sink);
+  }
+
   private static int crc32cDecode(String from) {
     byte[] decodeCrc32c = BaseEncoding.base64().decode(from);
     return Ints.fromByteArray(decodeCrc32c);
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index cf93b127..d5d25b6a 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -86,7 +86,7 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-dependency-plugin</artifactId>
-        <version>3.4.0</version>
+        <version>3.5.0</version>
         <executions>
           <execution>
             <id>copy-dependencies</id>
@@ -124,7 +124,7 @@
           <plugin>
             <groupId>org.apache.maven.plugins</groupId>
             <artifactId>maven-surefire-plugin</artifactId>
-            <version>3.0.0-M7</version>
+            <version>3.0.0-M8</version>
             <configuration>
               <includes>
                 <include>**/*IT</include>


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java
index 4cb1ac54..a11d989b 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.storage.Acl.Project.ProjectRole.VIEWERS;
+import static com.google.cloud.storage.TestUtils.assertAll;
 import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -45,12 +46,10 @@ import com.google.cloud.storage.BucketInfo.LifecycleRule.SetStorageClassLifecycl
 import com.google.cloud.storage.BucketInfo.PublicAccessPrevention;
 import com.google.cloud.storage.Conversions.Codec;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
 import java.io.StringWriter;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.junit.Test;
@@ -100,16 +99,9 @@ public class BucketInfoTest {
   private static final String DEFAULT_KMS_KEY_NAME =
       "projects/p/locations/kr-loc/keyRings/kr/cryptoKeys/key";
   private static final Boolean VERSIONING_ENABLED = true;
-  private static final Map<String, String> BUCKET_LABELS;
+  private static final Map<String, String> BUCKET_LABELS =
+      TestUtils.hashMapOf("label1", "value1", "label2", null);
 
-  static {
-    BUCKET_LABELS = new HashMap<>();
-    BUCKET_LABELS.put("label1", "value1");
-    BUCKET_LABELS.put("label2", null);
-  }
-
-  private static final Map<String, String> BUCKET_LABELS_TARGET =
-      ImmutableMap.of("label1", "value1", "label2", "");
   private static final Boolean REQUESTER_PAYS = true;
   private static final Boolean DEFAULT_EVENT_BASED_HOLD = true;
   private static final Long RETENTION_EFFECTIVE_TIME = 10L;
@@ -186,7 +178,7 @@ public class BucketInfoTest {
   private static final Lifecycle EMPTY_LIFECYCLE = lifecycle(Collections.emptyList());
 
   @Test
-  public void testToBuilder() {
+  public void testToBuilder() throws Exception {
     compareBuckets(BUCKET_INFO, BUCKET_INFO.toBuilder().build());
     BucketInfo bucketInfo = BUCKET_INFO.toBuilder().setName("B").setGeneratedId("id").build();
     assertEquals("B", bucketInfo.getName());
@@ -197,7 +189,7 @@ public class BucketInfoTest {
   }
 
   @Test
-  public void testToBuilderIncomplete() {
+  public void testToBuilderIncomplete() throws Exception {
     BucketInfo incompleteBucketInfo = BucketInfo.newBuilder("b").build();
     compareBuckets(incompleteBucketInfo, incompleteBucketInfo.toBuilder().build());
   }
@@ -210,39 +202,40 @@ public class BucketInfoTest {
 
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
-  public void testBuilder() {
-    assertEquals("b", BUCKET_INFO.getName());
-    assertEquals(ACL, BUCKET_INFO.getAcl());
-    assertEquals(ETAG, BUCKET_INFO.getEtag());
-    assertEquals(GENERATED_ID, BUCKET_INFO.getGeneratedId());
-    assertEquals(META_GENERATION, BUCKET_INFO.getMetageneration());
-    assertEquals(OWNER, BUCKET_INFO.getOwner());
-    assertEquals(SELF_LINK, BUCKET_INFO.getSelfLink());
-    assertEquals(CREATE_TIME, BUCKET_INFO.getCreateTime());
-    assertEquals(UPDATE_TIME, BUCKET_INFO.getUpdateTime());
-    assertEquals(CORS, BUCKET_INFO.getCors());
-    assertEquals(DEFAULT_ACL, BUCKET_INFO.getDefaultAcl());
-    assertEquals(DELETE_RULES, BUCKET_INFO.getDeleteRules());
-    assertEquals(INDEX_PAGE, BUCKET_INFO.getIndexPage());
-    assertEquals(IAM_CONFIGURATION, BUCKET_INFO.getIamConfiguration());
-    assertEquals(NOT_FOUND_PAGE, BUCKET_INFO.getNotFoundPage());
-    assertEquals(LOCATION, BUCKET_INFO.getLocation());
-    assertEquals(STORAGE_CLASS, BUCKET_INFO.getStorageClass());
-    assertEquals(DEFAULT_KMS_KEY_NAME, BUCKET_INFO.getDefaultKmsKeyName());
-    assertEquals(VERSIONING_ENABLED, BUCKET_INFO.versioningEnabled());
-    assertEquals(BUCKET_LABELS_TARGET, BUCKET_INFO.getLabels());
-    assertEquals(REQUESTER_PAYS, BUCKET_INFO.requesterPays());
-    assertEquals(DEFAULT_EVENT_BASED_HOLD, BUCKET_INFO.getDefaultEventBasedHold());
-    assertEquals(RETENTION_EFFECTIVE_TIME, BUCKET_INFO.getRetentionEffectiveTime());
-    assertEquals(RETENTION_PERIOD, BUCKET_INFO.getRetentionPeriod());
-    assertEquals(RETENTION_POLICY_IS_LOCKED, BUCKET_INFO.retentionPolicyIsLocked());
-    assertTrue(LOCATION_TYPES.contains(BUCKET_INFO.getLocationType()));
-    assertEquals(LOGGING, BUCKET_INFO.getLogging());
+  public void testBuilder() throws Exception {
+    assertAll(
+        () -> assertEquals("b", BUCKET_INFO.getName()),
+        () -> assertEquals(ACL, BUCKET_INFO.getAcl()),
+        () -> assertEquals(ETAG, BUCKET_INFO.getEtag()),
+        () -> assertEquals(GENERATED_ID, BUCKET_INFO.getGeneratedId()),
+        () -> assertEquals(META_GENERATION, BUCKET_INFO.getMetageneration()),
+        () -> assertEquals(OWNER, BUCKET_INFO.getOwner()),
+        () -> assertEquals(SELF_LINK, BUCKET_INFO.getSelfLink()),
+        () -> assertEquals(CREATE_TIME, BUCKET_INFO.getCreateTime()),
+        () -> assertEquals(UPDATE_TIME, BUCKET_INFO.getUpdateTime()),
+        () -> assertEquals(CORS, BUCKET_INFO.getCors()),
+        () -> assertEquals(DEFAULT_ACL, BUCKET_INFO.getDefaultAcl()),
+        () -> assertEquals(DELETE_RULES, BUCKET_INFO.getDeleteRules()),
+        () -> assertEquals(INDEX_PAGE, BUCKET_INFO.getIndexPage()),
+        () -> assertEquals(IAM_CONFIGURATION, BUCKET_INFO.getIamConfiguration()),
+        () -> assertEquals(NOT_FOUND_PAGE, BUCKET_INFO.getNotFoundPage()),
+        () -> assertEquals(LOCATION, BUCKET_INFO.getLocation()),
+        () -> assertEquals(STORAGE_CLASS, BUCKET_INFO.getStorageClass()),
+        () -> assertEquals(DEFAULT_KMS_KEY_NAME, BUCKET_INFO.getDefaultKmsKeyName()),
+        () -> assertEquals(VERSIONING_ENABLED, BUCKET_INFO.versioningEnabled()),
+        () -> assertEquals(BUCKET_LABELS, BUCKET_INFO.getLabels()),
+        () -> assertEquals(REQUESTER_PAYS, BUCKET_INFO.requesterPays()),
+        () -> assertEquals(DEFAULT_EVENT_BASED_HOLD, BUCKET_INFO.getDefaultEventBasedHold()),
+        () -> assertEquals(RETENTION_EFFECTIVE_TIME, BUCKET_INFO.getRetentionEffectiveTime()),
+        () -> assertEquals(RETENTION_PERIOD, BUCKET_INFO.getRetentionPeriod()),
+        () -> assertEquals(RETENTION_POLICY_IS_LOCKED, BUCKET_INFO.retentionPolicyIsLocked()),
+        () -> assertTrue(LOCATION_TYPES.contains(BUCKET_INFO.getLocationType())),
+        () -> assertEquals(LOGGING, BUCKET_INFO.getLogging()));
   }
 
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
-  public void testToPbAndFromPb() {
+  public void testToPbAndFromPb() throws Exception {
     Codec<BucketInfo, Bucket> codec = Conversions.apiary().bucketInfo();
 
     Bucket encode1 = codec.encode(BUCKET_INFO);
@@ -260,37 +253,44 @@ public class BucketInfoTest {
     compareBuckets(bucketInfo, decode2);
   }
 
-  private void compareBuckets(BucketInfo expected, BucketInfo value) {
-    assertEquals(expected.getName(), value.getName());
-    assertEquals(expected.getAcl(), value.getAcl());
-    assertEquals(expected.getEtag(), value.getEtag());
-    assertEquals(expected.getGeneratedId(), value.getGeneratedId());
-    assertEquals(expected.getMetageneration(), value.getMetageneration());
-    assertEquals(expected.getOwner(), value.getOwner());
-    assertEquals(expected.getSelfLink(), value.getSelfLink());
-    assertEquals(expected.getCreateTimeOffsetDateTime(), value.getCreateTimeOffsetDateTime());
-    assertEquals(expected.getUpdateTimeOffsetDateTime(), value.getUpdateTimeOffsetDateTime());
-    assertEquals(expected.getCors(), value.getCors());
-    assertEquals(expected.getDefaultAcl(), value.getDefaultAcl());
-    assertEquals(expected.getDeleteRules(), value.getDeleteRules());
-    assertEquals(expected.getLifecycleRules(), value.getLifecycleRules());
-    assertEquals(expected.getIndexPage(), value.getIndexPage());
-    assertEquals(expected.getIamConfiguration(), value.getIamConfiguration());
-    assertEquals(expected.getNotFoundPage(), value.getNotFoundPage());
-    assertEquals(expected.getLocation(), value.getLocation());
-    assertEquals(expected.getStorageClass(), value.getStorageClass());
-    assertEquals(expected.getDefaultKmsKeyName(), value.getDefaultKmsKeyName());
-    assertEquals(expected.versioningEnabled(), value.versioningEnabled());
-    assertEquals(expected.getLabels(), value.getLabels());
-    assertEquals(expected.requesterPays(), value.requesterPays());
-    assertEquals(expected.getDefaultEventBasedHold(), value.getDefaultEventBasedHold());
-    assertEquals(
-        expected.getRetentionEffectiveTimeOffsetDateTime(),
-        value.getRetentionEffectiveTimeOffsetDateTime());
-    assertEquals(expected.getRetentionPeriodDuration(), value.getRetentionPeriodDuration());
-    assertEquals(expected.retentionPolicyIsLocked(), value.retentionPolicyIsLocked());
-    assertEquals(expected.getLogging(), value.getLogging());
-    assertEquals(expected, value);
+  private void compareBuckets(BucketInfo expected, BucketInfo value) throws Exception {
+    assertAll(
+        () -> assertEquals(expected.getName(), value.getName()),
+        () -> assertEquals(expected.getAcl(), value.getAcl()),
+        () -> assertEquals(expected.getEtag(), value.getEtag()),
+        () -> assertEquals(expected.getGeneratedId(), value.getGeneratedId()),
+        () -> assertEquals(expected.getMetageneration(), value.getMetageneration()),
+        () -> assertEquals(expected.getOwner(), value.getOwner()),
+        () -> assertEquals(expected.getSelfLink(), value.getSelfLink()),
+        () ->
+            assertEquals(
+                expected.getCreateTimeOffsetDateTime(), value.getCreateTimeOffsetDateTime()),
+        () ->
+            assertEquals(
+                expected.getUpdateTimeOffsetDateTime(), value.getUpdateTimeOffsetDateTime()),
+        () -> assertEquals(expected.getCors(), value.getCors()),
+        () -> assertEquals(expected.getDefaultAcl(), value.getDefaultAcl()),
+        () -> assertEquals(expected.getDeleteRules(), value.getDeleteRules()),
+        () -> assertEquals(expected.getLifecycleRules(), value.getLifecycleRules()),
+        () -> assertEquals(expected.getIndexPage(), value.getIndexPage()),
+        () -> assertEquals(expected.getIamConfiguration(), value.getIamConfiguration()),
+        () -> assertEquals(expected.getNotFoundPage(), value.getNotFoundPage()),
+        () -> assertEquals(expected.getLocation(), value.getLocation()),
+        () -> assertEquals(expected.getStorageClass(), value.getStorageClass()),
+        () -> assertEquals(expected.getDefaultKmsKeyName(), value.getDefaultKmsKeyName()),
+        () -> assertEquals(expected.versioningEnabled(), value.versioningEnabled()),
+        () -> assertEquals(expected.getLabels(), value.getLabels()),
+        () -> assertEquals(expected.requesterPays(), value.requesterPays()),
+        () -> assertEquals(expected.getDefaultEventBasedHold(), value.getDefaultEventBasedHold()),
+        () ->
+            assertEquals(
+                expected.getRetentionEffectiveTimeOffsetDateTime(),
+                value.getRetentionEffectiveTimeOffsetDateTime()),
+        () ->
+            assertEquals(expected.getRetentionPeriodDuration(), value.getRetentionPeriodDuration()),
+        () -> assertEquals(expected.retentionPolicyIsLocked(), value.retentionPolicyIsLocked()),
+        () -> assertEquals(expected.getLogging(), value.getLogging()),
+        () -> assertEquals(expected, value));
   }
 
   @Test
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ByteRangeSpecTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ByteRangeSpecTest.java
index 234a24b4..dce7190d 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ByteRangeSpecTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ByteRangeSpecTest.java
@@ -24,6 +24,7 @@ import static java.util.Objects.requireNonNull;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Streams;
+import com.google.storage.v2.ReadObjectRequest;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -31,6 +32,7 @@ import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.util.Arrays;
 import java.util.stream.Stream;
+import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.junit.Test;
 import org.junit.experimental.runners.Enclosed;
@@ -457,6 +459,13 @@ public final class ByteRangeSpecTest {
       assertThat(rs.getSpec().getHttpRangeHeader()).isEqualTo(expect.getHttpRange());
     }
 
+    @Test
+    public void seekReadObjectRequest() {
+      ByteRangeSpec spec = rs.getSpec();
+      ReadObjectRequest actual = spec.seekReadObjectRequest(ReadObjectRequest.newBuilder()).build();
+      assertThat(actual).isEqualTo(expect.getReadObjectRequest());
+    }
+
     @Test
     public void beginOffset() {
       assertThat(rs.getSpec().beginOffset()).isEqualTo(expect.getBeginOffset());
@@ -487,6 +496,7 @@ public final class ByteRangeSpecTest {
               .endOffsetInclusive(EFFECTIVE_INFINITY)
               .length(EFFECTIVE_INFINITY)
               .httpRange(null)
+              .readObjectRequest(reqId())
               .isApplicableTo(
                   ByteRangeSpec.relativeLength(null, null),
                   ByteRangeSpec.explicit(null, null),
@@ -509,6 +519,7 @@ public final class ByteRangeSpecTest {
               .endOffsetInclusive(EFFECTIVE_INFINITY)
               .length(EFFECTIVE_INFINITY)
               .httpRange(headerRangeOpen(3))
+              .readObjectRequest(reqOpen(3))
               .isApplicableTo(
                   ByteRangeSpec.relativeLength(3L, null),
                   ByteRangeSpec.explicit(3L, null),
@@ -571,6 +582,7 @@ public final class ByteRangeSpecTest {
                   .endOffsetInclusive(EFFECTIVE_INFINITY)
                   .length(EFFECTIVE_INFINITY)
                   .httpRange(headerRangeOpen(1L))
+                  .readObjectRequest(reqOpen(1L))
                   .isApplicableTo(
                       ByteRangeSpec.relativeLength(1L, null),
                       ByteRangeSpec.explicit(1L, null),
@@ -582,6 +594,7 @@ public final class ByteRangeSpecTest {
                   .endOffsetInclusive(EFFECTIVE_INFINITY)
                   .length(EFFECTIVE_INFINITY)
                   .httpRange(headerRangeOpen(EFFECTIVE_INFINITY))
+                  .readObjectRequest(reqOpen(EFFECTIVE_INFINITY))
                   .isApplicableTo(
                       ByteRangeSpec.relativeLength(EFFECTIVE_INFINITY, null),
                       ByteRangeSpec.explicit(EFFECTIVE_INFINITY, null),
@@ -624,6 +637,18 @@ public final class ByteRangeSpecTest {
     return String.format("bytes=%d-%d", min, max);
   }
 
+  private static ReadObjectRequest reqOpen(long offset) {
+    return ReadObjectRequest.newBuilder().setReadOffset(offset).build();
+  }
+
+  private static ReadObjectRequest reqBounded(long offset, long length) {
+    return ReadObjectRequest.newBuilder().setReadOffset(offset).setReadLimit(length).build();
+  }
+
+  private static ReadObjectRequest reqId() {
+    return ReadObjectRequest.getDefaultInstance();
+  }
+
   private static final class RangeScenario {
     private final ByteRangeSpec spec;
     private final Expectations expectations;
@@ -657,18 +682,21 @@ public final class ByteRangeSpecTest {
       private final long endOffsetInclusive;
       private final long length;
       @Nullable private final String httpRange;
+      @NonNull private final ReadObjectRequest readObjectRequest;
 
       private Expectations(
           long beginOffset,
           long endOffset,
           long endOffsetInclusive,
           long length,
-          @Nullable String httpRange) {
+          @Nullable String httpRange,
+          @NonNull ReadObjectRequest readObjectRequest) {
         this.beginOffset = beginOffset;
         this.endOffset = endOffset;
         this.endOffsetInclusive = endOffsetInclusive;
         this.length = length;
         this.httpRange = httpRange;
+        this.readObjectRequest = readObjectRequest;
       }
 
       public long getBeginOffset() {
@@ -691,6 +719,10 @@ public final class ByteRangeSpecTest {
         return httpRange;
       }
 
+      public @NonNull ReadObjectRequest getReadObjectRequest() {
+        return readObjectRequest;
+      }
+
       String testNameFormat() {
         return MoreObjects.toStringHelper("")
             .add("bo", fmt(beginOffset))
@@ -708,6 +740,7 @@ public final class ByteRangeSpecTest {
             .add("endOffsetInclusive", fmt(endOffsetInclusive))
             .add("length", fmt(length))
             .add("httpRange", httpRange)
+            .add("readObjectRequest", StorageV2ProtoUtils.fmtProto(readObjectRequest))
             .toString();
       }
 
@@ -736,6 +769,7 @@ public final class ByteRangeSpecTest {
       private Long endOffsetInclusive;
       private Long length;
       @Nullable private String httpRange;
+      @NonNull private ReadObjectRequest readObjectRequest;
 
       public ExpectationsBuilder beginOffset(Long beginOffset) {
         this.beginOffset = beginOffset;
@@ -762,8 +796,14 @@ public final class ByteRangeSpecTest {
         return this;
       }
 
+      public ExpectationsBuilder readObjectRequest(ReadObjectRequest readObjectRequest) {
+        this.readObjectRequest = readObjectRequest;
+        return this;
+      }
+
       public ExpectationsBuilder bounded() {
-        return this.httpRange(headerRangeClosed(beginOffset, endOffsetInclusive));
+        return this.httpRange(headerRangeClosed(beginOffset, endOffsetInclusive))
+            .readObjectRequest(reqBounded(beginOffset, length));
       }
 
       public Stream<RangeScenario> isApplicableTo(ByteRangeSpec... brss) {
@@ -777,7 +817,8 @@ public final class ByteRangeSpecTest {
             requireNonNull(endOffset, "endOffset must be non null"),
             requireNonNull(endOffsetInclusive, "endOffsetInclusive must be non null"),
             requireNonNull(length, "length must be non null"),
-            httpRange);
+            httpRange,
+            requireNonNull(readObjectRequest, "readObjectRequest must be non null"));
       }
     }
   }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
index 306b867d..30eb4ea0 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
@@ -22,10 +22,12 @@ import static com.google.common.truth.Truth.assertThat;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.rpc.DataLossException;
+import com.google.api.gax.rpc.PermissionDeniedException;
 import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.StartResumableWriteRequest;
@@ -40,14 +42,21 @@ import io.grpc.stub.CallStreamObserver;
 import io.grpc.stub.StreamObserver;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
+import java.util.logging.Logger;
+import java.util.stream.Collector;
+import java.util.stream.Collectors;
 import org.junit.Test;
 
 public final class GapicUnbufferedWritableByteChannelTest {
+  private static final Logger LOGGER =
+      Logger.getLogger(GapicUnbufferedWritableByteChannelTest.class.getName());
 
   private static final ChunkSegmenter segmenter =
       new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), 10, 5);
@@ -174,7 +183,7 @@ public final class GapicUnbufferedWritableByteChannelTest {
     try (FakeServer fake = FakeServer.of(service);
         StorageClient sc = StorageClient.create(fake.storageSettings())) {
       SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
-      try (GapicUnbufferedWritableByteChannel<?> c =
+      GapicUnbufferedWritableByteChannel<?> c =
           new GapicUnbufferedWritableByteChannel<>(
               result,
               segmenter,
@@ -182,10 +191,23 @@ public final class GapicUnbufferedWritableByteChannelTest {
               WriteFlushStrategy.fsyncEveryFlush(
                   sc.writeObjectCallable(),
                   RetryingDependencies.attemptOnce(),
-                  Retrying.neverRetry()))) {
+                  Retrying.neverRetry()));
+      ArrayList<String> debugMessages = new ArrayList<>();
+      try {
         ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(bytes, 10);
         for (ByteBuffer buf : buffers) {
-          c.write(buf);
+          debugMessages.add(String.format("Writing buffer. buf = %s", buf));
+          int written = c.write(buf);
+          debugMessages.add(String.format("Wrote bytes. written = %2d", written));
+        }
+        // explicitly only close on success so we can trap the original error that maybe have
+        // happened before we reach here.
+        // Realistically, calling close here isn't strictly necessary because once we leave the
+        // try block for FakeServer the server will shut down.
+        c.close();
+      } catch (PermissionDeniedException ignore) {
+        for (String debugMessage : debugMessages) {
+          LOGGER.warning(debugMessage);
         }
       }
       assertThat(result.get()).isEqualTo(resp5);
@@ -228,6 +250,14 @@ public final class GapicUnbufferedWritableByteChannelTest {
                 obs.onNext(resp5);
                 obs.onCompleted();
               } else {
+                DirectWriteService.logUnexpectedRequest(
+                    ImmutableSet.of(
+                        ImmutableList.of(req1),
+                        ImmutableList.of(req2),
+                        ImmutableList.of(req3),
+                        ImmutableList.of(req4),
+                        ImmutableList.of(req5)),
+                    requests);
                 obs.onError(
                     TestUtils.apiException(Code.PERMISSION_DENIED, "Unexpected request chain."));
               }
@@ -276,6 +306,7 @@ public final class GapicUnbufferedWritableByteChannelTest {
   }
 
   static class DirectWriteService extends StorageImplBase {
+    private static final Logger LOGGER = Logger.getLogger(DirectWriteService.class.getName());
     private final BiConsumer<StreamObserver<WriteObjectResponse>, List<WriteObjectRequest>> c;
 
     private ImmutableList.Builder<WriteObjectRequest> requests;
@@ -293,12 +324,27 @@ public final class GapicUnbufferedWritableByteChannelTest {
               obs.onNext(writes.get(build));
               obs.onCompleted();
             } else {
+              logUnexpectedRequest(writes.keySet(), build);
               obs.onError(
                   TestUtils.apiException(Code.PERMISSION_DENIED, "Unexpected request chain."));
             }
           });
     }
 
+    private static void logUnexpectedRequest(
+        Set<List<WriteObjectRequest>> writes, List<WriteObjectRequest> build) {
+      Collector<CharSequence, ?, String> joining = Collectors.joining(",\n\t", "[\n\t", "\n]");
+      Collector<CharSequence, ?, String> oneLine = Collectors.joining(",", "[", "]");
+      String msg =
+          String.format(
+              "Unexpected Request Chain.%nexpected one of: %s%n        but was: %s",
+              writes.stream()
+                  .map(l -> l.stream().map(StorageV2ProtoUtils::fmtProto).collect(oneLine))
+                  .collect(joining),
+              build.stream().map(StorageV2ProtoUtils::fmtProto).collect(oneLine));
+      LOGGER.warning(msg);
+    }
+
     @Override
     public StreamObserver<WriteObjectRequest> writeObject(StreamObserver<WriteObjectResponse> obs) {
       return new Adapter() {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageV2ProtoUtilsTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageV2ProtoUtilsTest.java
index a3b45a06..a6ccc8ff 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageV2ProtoUtilsTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageV2ProtoUtilsTest.java
@@ -16,10 +16,10 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.ByteRangeSpec.relativeLength;
 import static com.google.cloud.storage.StorageV2ProtoUtils.fmtProto;
 import static com.google.cloud.storage.StorageV2ProtoUtils.seekReadObjectRequest;
 import static com.google.common.truth.Truth.assertThat;
-import static org.junit.Assert.assertThrows;
 
 import com.google.cloud.storage.jqwik.StorageArbitraries;
 import com.google.storage.v2.ObjectAccessControl;
@@ -35,39 +35,6 @@ import net.jqwik.api.Provide;
 
 public final class StorageV2ProtoUtilsTest {
 
-  @Example
-  void validation_nullOffset_effectiveMin() {
-    seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), null, 0L);
-    seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), null, null);
-  }
-
-  @Example
-  void validation_nullLimit_effectiveMax() {
-    seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), 0L, null);
-    seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), null, null);
-  }
-
-  @Example
-  void validation_offset_lteq_limit() {
-    seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), 3L, 2L);
-    seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), 0L, 0L);
-    seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), 1L, 1L);
-  }
-
-  @Example
-  void validation_offset_gteq_0() {
-    assertThrows(
-        IllegalArgumentException.class,
-        () -> seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), -1L, null));
-  }
-
-  @Example
-  void validation_limit_gteq_0() {
-    assertThrows(
-        IllegalArgumentException.class,
-        () -> seekReadObjectRequest(ReadObjectRequest.getDefaultInstance(), null, -1L));
-  }
-
   @Example
   void objectAclEntityIdOrAltEq() {
     String entity = "project-viewer-123123";
@@ -86,21 +53,19 @@ public final class StorageV2ProtoUtilsTest {
   void seek(@ForAll("seekCases") SeekCase srr) {
     Long offset = srr.offset;
     Long limit = srr.limit;
+    ReadObjectRequest seek = seekReadObjectRequest(srr.req, relativeLength(offset, limit));
 
-    // I miss pattern matching...
+    // If both offset and limit are null, avoid allocating a new instance as we don't have any
+    // meaningful change to apply
     if (offset == null && limit == null) {
-      ReadObjectRequest seek = seekReadObjectRequest(srr.req, offset, limit);
       assertThat(seek).isSameInstanceAs(srr.req);
-    } else if (offset != null && limit == null) {
-      ReadObjectRequest seek = seekReadObjectRequest(srr.req, offset, limit);
-      assertThat(seek.getReadOffset()).isEqualTo(offset);
-    } else if (offset == null && limit != null) {
-      ReadObjectRequest seek = seekReadObjectRequest(srr.req, offset, limit);
-      assertThat(seek.getReadLimit()).isEqualTo(limit);
     } else {
-      ReadObjectRequest seek = seekReadObjectRequest(srr.req, offset, limit);
-      assertThat(seek.getReadOffset()).isEqualTo(offset);
-      assertThat(seek.getReadLimit()).isEqualTo(limit);
+      if (offset != null && offset != 0) {
+        assertThat(seek.getReadOffset()).isEqualTo(offset);
+      }
+      if (limit != null && limit != Long.MAX_VALUE) {
+        assertThat(seek.getReadLimit()).isEqualTo(limit);
+      }
     }
   }
 
@@ -128,14 +93,7 @@ public final class StorageV2ProtoUtilsTest {
 
     @Override
     public String toString() {
-      return "SeekReadRequest{"
-          + "req="
-          + fmtProto(req)
-          + ", offset="
-          + offset
-          + ", limit="
-          + limit
-          + '}';
+      return "SeekCase{" + "req=" + fmtProto(req) + ", offset=" + offset + ", limit=" + limit + '}';
     }
 
     private static SeekCase of(
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
index 03b287b3..0ce0b4c3 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
@@ -16,6 +16,8 @@
 
 package com.google.cloud.storage;
 
+import static java.util.Objects.requireNonNull;
+
 import com.google.api.core.ApiClock;
 import com.google.api.core.NanoClock;
 import com.google.api.gax.grpc.GrpcCallContext;
@@ -48,7 +50,10 @@ import java.io.OutputStream;
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.Callable;
 import java.util.function.Function;
@@ -239,4 +244,24 @@ public final class TestUtils {
             .collect(ImmutableList.toImmutableList());
     MultipleFailureException.assertEmpty(x);
   }
+
+  /** ImmutableMap does not allow null values, this method does */
+  public static Map<@NonNull String, @Nullable String> hashMapOf(
+      @NonNull String k1, @Nullable String v1) {
+    requireNonNull(k1, "k1 must be non null");
+    HashMap<String, String> map = new HashMap<>();
+    map.put(k1, v1);
+    return Collections.unmodifiableMap(map);
+  }
+
+  /** ImmutableMap does not allow null values, this method does */
+  public static Map<@NonNull String, @Nullable String> hashMapOf(
+      @NonNull String k1, @Nullable String v1, @NonNull String k2, @Nullable String v2) {
+    requireNonNull(k1, "k1 must be non null");
+    requireNonNull(k2, "k2 must be non null");
+    HashMap<String, String> map = new HashMap<>();
+    map.put(k1, v1);
+    map.put(k2, v2);
+    return Collections.unmodifiableMap(map);
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/UpdateMaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/UpdateMaskTest.java
index 89173046..6d343f2e 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/UpdateMaskTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/UpdateMaskTest.java
@@ -30,6 +30,7 @@ import com.google.cloud.storage.BucketInfo.LifecycleRule.LifecycleCondition;
 import com.google.cloud.storage.BucketInfo.Logging;
 import com.google.cloud.storage.Storage.BlobField;
 import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.UnifiedOpts.NamedField;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -64,7 +65,7 @@ public final class UpdateMaskTest {
       UpdateObjectRequest expected =
           UpdateObjectRequest.newBuilder()
               .setObject(expectedObject)
-              .setUpdateMask(FieldMask.newBuilder().addPaths("metadata").build())
+              .setUpdateMask(FieldMask.newBuilder().addPaths("metadata.x").build())
               .build();
 
       AtomicReference<UpdateObjectRequest> actualRequest = new AtomicReference<>();
@@ -95,7 +96,9 @@ public final class UpdateMaskTest {
 
     @Test
     public void blobInfo_field_metadata() {
-      testBlobField(b -> b.setMetadata(ImmutableMap.of("x", "X")), BlobField.METADATA);
+      testBlobField(
+          b -> b.setMetadata(ImmutableMap.of("x", "X")),
+          NamedField.nested(BlobField.METADATA, NamedField.literal("x")));
     }
 
     @Test
@@ -239,7 +242,7 @@ public final class UpdateMaskTest {
     }
 
     private static void testBlobField(
-        UnaryOperator<BlobInfo.Builder> f, BlobField... expectedModified) {
+        UnaryOperator<BlobInfo.Builder> f, NamedField... expectedModified) {
       BlobInfo actual1 = f.apply(base().toBuilder()).build();
       assertThat(actual1.getModifiedFields()).isEqualTo(ImmutableSet.copyOf(expectedModified));
       // verify that nothing is carried through from a previous state, and that setting the same
@@ -262,7 +265,7 @@ public final class UpdateMaskTest {
       UpdateBucketRequest expected =
           UpdateBucketRequest.newBuilder()
               .setBucket(expectedBucket)
-              .setUpdateMask(FieldMask.newBuilder().addPaths("labels").build())
+              .setUpdateMask(FieldMask.newBuilder().addPaths("labels.x").build())
               .build();
 
       AtomicReference<UpdateBucketRequest> actualRequest = new AtomicReference<>();
@@ -381,7 +384,9 @@ public final class UpdateMaskTest {
 
     @Test
     public void bucketInfo_field_setLabels() {
-      testBucketField(b -> b.setLabels(ImmutableMap.of("x", "X")), BucketField.LABELS);
+      testBucketField(
+          b -> b.setLabels(ImmutableMap.of("x", "X")),
+          NamedField.nested(BucketField.LABELS, NamedField.literal("x")));
     }
 
     @Test
@@ -445,7 +450,7 @@ public final class UpdateMaskTest {
     }
 
     private static void testBucketField(
-        UnaryOperator<BucketInfo.Builder> f, BucketField... expectedModified) {
+        UnaryOperator<BucketInfo.Builder> f, NamedField... expectedModified) {
       BucketInfo actual1 = f.apply(base().toBuilder()).build();
       assertThat(actual1.getModifiedFields()).isEqualTo(ImmutableSet.copyOf(expectedModified));
       // verify that nothing is carried through from a previous state, and that setting the same
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
index a4657db7..751b0975 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobReadChannelTest.java
@@ -30,7 +30,6 @@ import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.BucketInfo;
-import com.google.cloud.storage.DataGeneration;
 import com.google.cloud.storage.DataGenerator;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobSourceOption;
@@ -56,10 +55,10 @@ import java.nio.channels.Channels;
 import java.nio.channels.FileChannel;
 import java.nio.channels.WritableByteChannel;
 import java.nio.file.Files;
+import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.nio.file.StandardOpenOption;
 import java.util.Arrays;
-import java.util.Random;
 import java.util.zip.GZIPInputStream;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.junit.Rule;
@@ -80,8 +79,6 @@ public final class ITBlobReadChannelTest {
       BaseEncoding.base64()
           .decode("H4sIAAAAAAAAAPNIzcnJV3DPz0/PSVVwzskvTVEILskvSkxPVQQA/LySchsAAAA=");
 
-  @Rule public final DataGeneration dataGeneration = new DataGeneration(new Random(872364872));
-
   @Rule public final TemporaryFolder tmp = new TemporaryFolder();
 
   @Inject public Storage storage;
@@ -139,7 +136,7 @@ public final class ITBlobReadChannelTest {
   public void testLimit_downloadToFile() throws IOException {
     String blobName = String.format("%s/src", generator.randomObjectName());
     BlobId blobId = BlobId.of(bucket.getName(), blobName);
-    ByteBuffer content = dataGeneration.randByteBuffer(108);
+    ByteBuffer content = DataGenerator.base64Characters().genByteBuffer(108);
     try (WriteChannel writer = storage.writer(BlobInfo.newBuilder(blobId).build())) {
       writer.write(content);
     }
@@ -151,18 +148,21 @@ public final class ITBlobReadChannelTest {
     duplicate.position(14);
     duplicate.limit(37);
     duplicate.get(expectedBytes);
+    String xxdExpected = xxd(expectedBytes);
 
     try {
+      Path path = Paths.get(destFileName);
       try (ReadChannel from = storage.reader(blobId);
-          FileChannel to = FileChannel.open(Paths.get(destFileName), StandardOpenOption.WRITE)) {
+          FileChannel to = FileChannel.open(path, StandardOpenOption.WRITE)) {
         from.seek(14);
         from.limit(37);
 
         ByteStreams.copy(from, to);
       }
 
-      byte[] readBytes = Files.readAllBytes(Paths.get(destFileName));
-      assertThat(readBytes).isEqualTo(expectedBytes);
+      byte[] readBytes = Files.readAllBytes(path);
+      String xxdActual = xxd(readBytes);
+      assertThat(xxdActual).isEqualTo(xxdExpected);
     } finally {
       file.delete();
     }
@@ -344,7 +344,6 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @Exclude(transports = Transport.GRPC)
   public void seekAfterReadWorks() throws IOException {
     ObjectAndContent obj512KiB = objectsFixture.getObj512KiB();
     BlobInfo gen1 = obj512KiB.getInfo();
@@ -374,7 +373,6 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @Exclude(transports = Transport.GRPC)
   public void limitAfterReadWorks() throws IOException {
     ObjectAndContent obj512KiB = objectsFixture.getObj512KiB();
     BlobInfo gen1 = obj512KiB.getInfo();
@@ -404,7 +402,6 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @Exclude(transports = Transport.GRPC)
   public void readingLastByteReturnsOneByte_seekOnly() throws IOException {
     int length = 10;
     byte[] bytes = DataGenerator.base64Characters().genBytes(length);
@@ -426,7 +423,6 @@ public final class ITBlobReadChannelTest {
   }
 
   @Test
-  @Exclude(transports = Transport.GRPC)
   public void readingLastByteReturnsOneByte_seekAndLimit() throws IOException {
     int length = 10;
     byte[] bytes = DataGenerator.base64Characters().genBytes(length);
@@ -521,7 +517,7 @@ public final class ITBlobReadChannelTest {
       throws IOException {
     String blobName = String.format("%s/src", generator.randomObjectName());
     BlobInfo src = BlobInfo.newBuilder(bucket, blobName).build();
-    ByteBuffer content = dataGeneration.randByteBuffer(srcContentSize);
+    ByteBuffer content = DataGenerator.base64Characters().genByteBuffer(srcContentSize);
     ByteBuffer dup = content.duplicate();
     dup.position(rangeBegin);
     int newLimit = Math.min(dup.capacity(), rangeEnd);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketLifecycleRulesTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketLifecycleRulesTest.java
index 6e7eeadf..d7f14d82 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketLifecycleRulesTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBucketLifecycleRulesTest.java
@@ -72,6 +72,22 @@ public final class ITBucketLifecycleRulesTest {
     }
   }
 
+  @Test
+  public void condition_ageDays_0_shouldWork() throws Exception {
+    LifecycleRule d1 =
+        new LifecycleRule(
+            LifecycleAction.newAbortIncompleteMPUploadAction(),
+            LifecycleCondition.newBuilder().setAge(0).build());
+    BucketInfo info = baseInfo().setLifecycleRules(ImmutableList.of(d1)).build();
+
+    try (TemporaryBucket tmp =
+        TemporaryBucket.newBuilder().setBucketInfo(info).setStorage(storage).build()) {
+      BucketInfo bucket = tmp.getBucket();
+      Bucket update = storage.get(bucket.getName());
+      assertThat(update.getLifecycleRules()).isEqualTo(ImmutableList.of(d1));
+    }
+  }
+
   @Test
   public void deleteRule_modifyingLifecycleRulesMatchesLastOperation() throws Exception {
     BucketInfo info;
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITNestedUpdateMaskTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITNestedUpdateMaskTest.java
new file mode 100644
index 00000000..8e3b944d
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITNestedUpdateMaskTest.java
@@ -0,0 +1,176 @@
+/*
+ * 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.it;
+
+import static com.google.cloud.storage.TestUtils.hashMapOf;
+import static com.google.common.truth.Truth.assertThat;
+import static java.util.Objects.requireNonNull;
+
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.BucketTargetOption;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.ITNestedUpdateMaskTest.NestedUpdateMaskParametersProvider;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.annotations.Parameterized;
+import com.google.cloud.storage.it.runner.annotations.Parameterized.Parameter;
+import com.google.cloud.storage.it.runner.annotations.Parameterized.ParametersProvider;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+/**
+ * A set of tests to specifically test scenarios related to update handling of {@link
+ * BlobInfo#getMetadata()} and {@link BucketInfo#getLabels()} and the various permutations which can
+ * be used to add and remove keys.
+ */
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = Backend.PROD,
+    transports = {Transport.HTTP, Transport.GRPC})
+@Parameterized(NestedUpdateMaskParametersProvider.class)
+public final class ITNestedUpdateMaskTest {
+
+  @Inject public Generator generator;
+
+  @Inject public Storage storage;
+
+  @Inject public BucketInfo bucket;
+
+  @Parameter public Param param;
+
+  public static final class NestedUpdateMaskParametersProvider implements ParametersProvider {
+    private static final Map<String, String> empty = ImmutableMap.of();
+    private static final Map<String, String> k1a = ImmutableMap.of("k1", "a");
+    private static final Map<String, String> k2b = ImmutableMap.of("k2", "b");
+    private static final Map<String, String> k1z = ImmutableMap.of("k1", "z");
+    private static final Map<String, String> k1a_k2b = ImmutableMap.of("k1", "a", "k2", "b");
+    private static final Map<String, String> k1z_k2b = ImmutableMap.of("k1", "z", "k2", "b");
+    private static final Map<String, String> k1a_k2null = hashMapOf("k1", "a", "k2", null);
+    private static final Map<String, String> k1null = hashMapOf("k1", null);
+    private static final Map<String, String> k2null = hashMapOf("k2", null);
+
+    /**
+     *
+     *
+     * <pre>
+     * | base                | update               | expected            |
+     * |---------------------|----------------------|---------------------|
+     * | null                | {"k1":"a"}           | {"k1":"a"}          |
+     * | {}                  | {"k1":"a"}           | {"k1":"a"}          |
+     * | {"k1":"a"}          | {"k1":"a","k2":"b"}  | {"k1":"a","k2":"b"} |
+     * | {"k1":"a"}          | {"k2":"b"}           | {"k1":"a","k2":"b"} |
+     * | {"k1":"a","k2":"b"} | {"k1":"z","k2":"b"}  | {"k1":"z","k2":"b"} |
+     * | {"k1":"a","k2":"b"} | {"k1":"z"}           | {"k1":"z","k2":"b"} |
+     * | {"k1":"a","k2":"b"} | {"k1":"a","k2":null} | {"k1":"a"}          |
+     * | {"k1":"a","k2":"b"} | {"k2":null}          | {"k1":"a"}          |
+     * | {"k1":"a"}          | {}                   | null                |
+     * | {"k1":"a"}          | {"k1":null}          | null                |
+     * | {"k1":"a","k2":"b"} | null                 | null                |
+     * </pre>
+     */
+    @Override
+    public ImmutableList<Param> parameters() {
+      return ImmutableList.of(
+          new Param("null to 1", null, k1a, k1a),
+          new Param("empty to 1", empty, k1a, k1a),
+          new Param("1 to 2 set", k1a, k1a_k2b, k1a_k2b),
+          new Param("1 to 2 add", k1a, k2b, k1a_k2b),
+          new Param("2 keys, modify 1 value (full)", k1a_k2b, k1z_k2b, k1z_k2b),
+          new Param("2 keys, modify 1 value (fine)", k1a_k2b, k1z, k1z_k2b),
+          new Param("2 keys, modify 1 null (full)", k1a_k2b, k1a_k2null, k1a),
+          new Param("2 keys, modify 1 null (fine)", k1a_k2b, k2null, k1a),
+          new Param("1 key, set empty", k1a, empty, null),
+          new Param("1 key, null key", k1a, k1null, null),
+          new Param("2 keys, set null", k1a_k2b, null, null));
+    }
+  }
+
+  @Test
+  public void testBucketLabels() throws Exception {
+    BucketInfo bucket = newBucketInfo(param.initial);
+    try (TemporaryBucket tempB =
+        TemporaryBucket.newBuilder().setBucketInfo(bucket).setStorage(storage).build()) {
+      BucketInfo gen1 = tempB.getBucket();
+      BucketInfo modified = gen1.toBuilder().setLabels(param.update).build();
+      Bucket gen2 = storage.update(modified, BucketTargetOption.metagenerationMatch());
+      assertThat(gen2.getLabels()).isEqualTo(param.expected);
+    }
+  }
+
+  @Test
+  public void testBlobMetadata() {
+    BlobInfo blob = newBlobInfo(param.initial);
+    Blob gen1 = storage.create(blob, BlobTargetOption.doesNotExist());
+    BlobInfo modified = gen1.toBuilder().setMetadata(param.update).build();
+    Blob gen2 = storage.update(modified, BlobTargetOption.metagenerationMatch());
+    assertThat(gen2.getMetadata()).isEqualTo(param.expected);
+  }
+
+  private BlobInfo newBlobInfo(Map<String, String> metadata) {
+    String blobName = generator.randomObjectName();
+    BlobInfo.Builder builder = BlobInfo.newBuilder(bucket, blobName);
+    if (metadata != null) {
+      builder.setMetadata(metadata);
+    }
+    return builder.build();
+  }
+
+  private BucketInfo newBucketInfo(Map<String, String> metadata) {
+    BucketInfo.Builder builder = BucketInfo.newBuilder(generator.randomBucketName());
+    if (metadata != null) {
+      builder.setLabels(metadata);
+    }
+    return builder.build();
+  }
+
+  private static final class Param {
+    private final String description;
+    @Nullable private final Map<@NonNull String, @Nullable String> initial;
+    @Nullable private final Map<@NonNull String, @Nullable String> update;
+    @Nullable private final Map<@NonNull String, @Nullable String> expected;
+
+    private Param(
+        String description,
+        @Nullable Map<@NonNull String, @Nullable String> initial,
+        @Nullable Map<@NonNull String, @Nullable String> update,
+        @Nullable Map<@NonNull String, @Nullable String> expected) {
+      requireNonNull(description, "description must be non null");
+      this.description = description;
+      this.initial = initial;
+      this.update = update;
+      this.expected = expected;
+    }
+
+    @Override
+    public String toString() {
+      return description;
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
index 5a0fa76c..f2afbb2d 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITObjectTest.java
@@ -625,8 +625,6 @@ public class ITObjectTest {
   }
 
   @Test
-  // Metadata update bug b/230510191
-  @Exclude(transports = Transport.GRPC)
   public void testUpdateBlobUnsetMetadata() {
 
     String blobName = "test-update-blob-unset-metadata";