Repository: googleapis/java-storage
Pull request #2472 (2024-03-27 15:23:07)

feat: port DefaultBlobWriteSessionConfig to work with HttpStorageOptions


Patch:
diff --git a/.cloudbuild/samples_build.yaml b/.cloudbuild/samples_build.yaml
new file mode 100644
index 00000000..c22f1335
--- /dev/null
+++ b/.cloudbuild/samples_build.yaml
@@ -0,0 +1,35 @@
+steps:
+- name: gcr.io/cloud-devrel-public-resources/java8
+  entrypoint: ls
+  args: [
+    '-alt',
+  ]
+- name: gcr.io/cloud-devrel-public-resources/java8
+  entrypoint: curl
+  args: [
+    '--header',
+    'Metadata-Flavor: Google',
+    'http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/email'
+  ]
+- name: gcr.io/cloud-devrel-public-resources/java8
+  entrypoint: pwd
+- name: gcr.io/cloud-devrel-public-resources/java8
+  entrypoint: bash
+  args: [
+    '.kokoro/build.sh'
+  ]
+  env:
+  - 'JOB_TYPE=samples'
+  - 'GOOGLE_CLOUD_PROJECT=cloud-java-ci-sample'
+  - 'GOOGLE_CLOUD_PROJECT_NUMBER=615621127317'
+  - 'IT_SERVICE_ACCOUNT_EMAIL=samples@cloud-java-ci-sample.iam.gserviceaccount.com'
+- name: gcr.io/cloud-devrel-public-resources/java8
+  entrypoint: echo
+  args: [
+    'Sample job succeeded',
+  ]
+timeout: 3600s
+options:
+  defaultLogsBucketBehavior: REGIONAL_USER_OWNED_BUCKET
+
+
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
index b52e9bd4..441f78ac 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
@@ -53,6 +53,15 @@ public abstract class BlobWriteSessionConfig implements Serializable {
         StorageInternal s, BlobInfo info, Opts<ObjectTargetOpt> opts);
   }
 
+  /**
+   * Internal marker interface to signify an implementation of {@link BlobWriteSessionConfig} is
+   * compatible with {@link com.google.cloud.storage.TransportCompatibility.Transport#HTTP}
+   *
+   * <p>We could evaluate the annotations, but the code for that is more complicated and probably
+   * not worth the effort.
+   */
+  interface HttpCompatible {}
+
   /**
    * Internal marker interface to signify an implementation of {@link BlobWriteSessionConfig} is
    * compatible with {@link com.google.cloud.storage.TransportCompatibility.Transport#GRPC}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
index 43b68d8a..781411c6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
@@ -55,7 +55,7 @@ import java.util.Collection;
  *       full or close. Buffer size is configurable via
  *       {@link DefaultBlobWriteSessionConfig#withChunkSize(int)}
  *     </td>
- *     <td>gRPC</td>
+ *     <td>gRPC, HTTP</td>
  *     <td>The network will only be used for the following operations:
  *     <ol>
  *       <li>Creating the Resumable Upload Session</li>
@@ -241,7 +241,7 @@ public final class BlobWriteSessionConfigs {
    * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
-  @TransportCompatibility({Transport.GRPC})
+  @TransportCompatibility({Transport.GRPC, Transport.HTTP})
   public static DefaultBlobWriteSessionConfig getDefault() {
     return new DefaultBlobWriteSessionConfig(ByteSizeConstants._16MiB);
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
index 3a4dd2be..1d45c9e3 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
@@ -21,17 +21,21 @@ import com.google.api.core.ApiFutures;
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
 import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import java.nio.channels.WritableByteChannel;
 import java.time.Clock;
+import java.util.Map;
+import java.util.function.Supplier;
 import javax.annotation.concurrent.Immutable;
 
 /**
@@ -55,9 +59,9 @@ import javax.annotation.concurrent.Immutable;
  */
 @Immutable
 @BetaApi
-@TransportCompatibility({Transport.GRPC})
+@TransportCompatibility({Transport.GRPC, Transport.HTTP})
 public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
-    implements BlobWriteSessionConfig.GrpcCompatible {
+    implements BlobWriteSessionConfig.HttpCompatible, BlobWriteSessionConfig.GrpcCompatible {
   private static final long serialVersionUID = -6873740918589930633L;
 
   private final int chunkSize;
@@ -146,6 +150,39 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
                           .build();
                     })),
             WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER);
+      } else if (s instanceof StorageImpl) {
+        StorageImpl json = (StorageImpl) s;
+
+        return new DecoratedWritableByteChannelSession<>(
+            new LazySession<>(
+                new LazyWriteChannel<>(
+                    () -> {
+                      final Map<StorageRpc.Option, ?> optionsMap = opts.getRpcOptions();
+                      BlobInfo.Builder builder = info.toBuilder().setMd5(null).setCrc32c(null);
+                      BlobInfo updated = opts.blobInfoMapper().apply(builder).build();
+
+                      StorageObject encode = Conversions.json().blobInfo().encode(updated);
+                      Supplier<String> uploadIdSupplier =
+                          ResumableMedia.startUploadForBlobInfo(
+                              json.getOptions(),
+                              updated,
+                              optionsMap,
+                              json.retryAlgorithmManager.getForResumableUploadSessionCreate(
+                                  optionsMap));
+                      ApiFuture<JsonResumableWrite> startAsync =
+                          ApiFutures.immediateFuture(
+                              JsonResumableWrite.of(
+                                  encode, optionsMap, uploadIdSupplier.get(), 0L));
+
+                      return ResumableMedia.http()
+                          .write()
+                          .byteChannel(HttpClientContext.from(json.storageRpc))
+                          .resumable()
+                          .buffered(BufferHandle.allocate(chunkSize))
+                          .setStartAsync(startAsync)
+                          .build();
+                    })),
+            Conversions.json().blobInfo());
       } else {
         throw new IllegalStateException(
             "Unknown Storage implementation: " + s.getClass().getName());
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
index fd2a5731..684f3f15 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
@@ -16,6 +16,7 @@
 
 package com.google.cloud.storage;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiClock;
@@ -30,6 +31,7 @@ import com.google.cloud.TransportOptions;
 import com.google.cloud.http.HttpTransportOptions;
 import com.google.cloud.spi.ServiceRpcFactory;
 import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.StorageRpcFactory;
 import com.google.cloud.storage.spi.v1.HttpStorageRpc;
@@ -39,7 +41,9 @@ import com.google.common.collect.ImmutableSet;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.Serializable;
+import java.time.Clock;
 import java.util.Set;
+import org.checkerframework.checker.nullness.qual.NonNull;
 
 /** @since 2.14.0 This new api is in preview and is subject to breaking changes. */
 @BetaApi
@@ -55,6 +59,7 @@ public class HttpStorageOptions extends StorageOptions {
 
   private final HttpRetryAlgorithmManager retryAlgorithmManager;
   private transient RetryDependenciesAdapter retryDepsAdapter;
+  private final BlobWriteSessionConfig blobWriteSessionConfig;
 
   private HttpStorageOptions(Builder builder, StorageDefaults serviceDefaults) {
     super(builder, serviceDefaults);
@@ -63,6 +68,7 @@ public class HttpStorageOptions extends StorageOptions {
             MoreObjects.firstNonNull(
                 builder.storageRetryStrategy, defaults().getStorageRetryStrategy()));
     retryDepsAdapter = new RetryDependenciesAdapter();
+    blobWriteSessionConfig = builder.blobWriteSessionConfig;
   }
 
   @Override
@@ -120,6 +126,8 @@ public class HttpStorageOptions extends StorageOptions {
   public static class Builder extends StorageOptions.Builder {
 
     private StorageRetryStrategy storageRetryStrategy;
+    private BlobWriteSessionConfig blobWriteSessionConfig =
+        HttpStorageDefaults.INSTANCE.getDefaultStorageWriterConfig();
 
     Builder() {}
 
@@ -218,6 +226,24 @@ public class HttpStorageOptions extends StorageOptions {
       return this;
     }
 
+    /**
+     * @see BlobWriteSessionConfig
+     * @see BlobWriteSessionConfigs
+     * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
+     * @see HttpStorageDefaults#getDefaultStorageWriterConfig()
+     * @since 2.29.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    public HttpStorageOptions.Builder setBlobWriteSessionConfig(
+        @NonNull BlobWriteSessionConfig blobWriteSessionConfig) {
+      requireNonNull(blobWriteSessionConfig, "blobWriteSessionConfig must be non null");
+      checkArgument(
+          blobWriteSessionConfig instanceof BlobWriteSessionConfig.HttpCompatible,
+          "The provided instance of BlobWriteSessionConfig is not compatible with this HTTP transport.");
+      this.blobWriteSessionConfig = blobWriteSessionConfig;
+      return this;
+    }
+
     @Override
     public HttpStorageOptions build() {
       return new HttpStorageOptions(this, defaults());
@@ -249,6 +275,12 @@ public class HttpStorageOptions extends StorageOptions {
     public StorageRetryStrategy getStorageRetryStrategy() {
       return StorageRetryStrategy.getDefaultStorageRetryStrategy();
     }
+
+    /** @since 2.29.0 This new api is in preview and is subject to breaking changes. */
+    @BetaApi
+    public BlobWriteSessionConfig getDefaultStorageWriterConfig() {
+      return BlobWriteSessionConfigs.getDefault();
+    }
   }
 
   /**
@@ -287,7 +319,14 @@ public class HttpStorageOptions extends StorageOptions {
     public Storage create(StorageOptions options) {
       if (options instanceof HttpStorageOptions) {
         HttpStorageOptions httpStorageOptions = (HttpStorageOptions) options;
-        return new StorageImpl(httpStorageOptions);
+        Clock clock = Clock.systemUTC();
+        try {
+          return new StorageImpl(
+              httpStorageOptions, httpStorageOptions.blobWriteSessionConfig.createFactory(clock));
+        } catch (IOException e) {
+          throw new IllegalStateException(
+              "Unable to instantiate HTTP com.google.cloud.storage.Storage client.", e);
+        }
       } else {
         throw new IllegalArgumentException("Only HttpStorageOptions supported");
       }
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 a5df68b8..1a9bd8d0 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
@@ -38,6 +38,7 @@ import com.google.cloud.Policy;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
 import com.google.cloud.storage.BlobReadChannelV2.BlobReadChannelContext;
+import com.google.cloud.storage.BlobWriteSessionConfig.WriterFactory;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
 import com.google.cloud.storage.PostPolicyV4.ConditionV4Type;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
@@ -92,7 +93,7 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
-final class StorageImpl extends BaseService<StorageOptions> implements Storage {
+final class StorageImpl extends BaseService<StorageOptions> implements Storage, StorageInternal {
 
   private static final byte[] EMPTY_BYTE_ARRAY = {};
   private static final String EMPTY_BYTE_ARRAY_MD5 = "1B2M2Y8AsgTpgAmY7PhCfg==";
@@ -115,11 +116,13 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   final HttpRetryAlgorithmManager retryAlgorithmManager;
   final StorageRpc storageRpc;
+  final WriterFactory writerFactory;
 
-  StorageImpl(HttpStorageOptions options) {
+  StorageImpl(HttpStorageOptions options, WriterFactory writerFactory) {
     super(options);
     this.retryAlgorithmManager = options.getRetryAlgorithmManager();
     this.storageRpc = options.getStorageRpcV1();
+    this.writerFactory = writerFactory;
   }
 
   @Override
@@ -1635,4 +1638,13 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
         () -> storageRpc.get(bucketPb, optionsMap),
         (b) -> Conversions.json().bucketInfo().decode(b).asBucket(this));
   }
+
+  @Override
+  public BlobWriteSession blobWriteSession(BlobInfo blobInfo, BlobWriteOption... options) {
+    Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo);
+
+    WritableByteChannelSession<?, BlobInfo> writableByteChannelSession =
+        writerFactory.writeSession(this, blobInfo, opts);
+    return BlobWriteSessions.of(writableByteChannelSession);
+  }
 }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
index 763346b1..b8f89726 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
@@ -26,6 +26,7 @@ import com.google.cloud.storage.BlobWriteSessionConfigs;
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.DataGenerator;
 import com.google.cloud.storage.GrpcStorageOptions;
+import com.google.cloud.storage.HttpStorageOptions;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.StorageException;
@@ -49,7 +50,7 @@ import org.junit.runner.RunWith;
 
 @RunWith(StorageITRunner.class)
 @CrossRun(
-    transports = {Transport.GRPC},
+    transports = {Transport.HTTP, Transport.GRPC},
     backends = {Backend.PROD})
 public final class ITBlobWriteSessionTest {
 
@@ -68,6 +69,7 @@ public final class ITBlobWriteSessionTest {
   }
 
   @Test
+  @CrossRun.Exclude(transports = Transport.HTTP)
   public void bufferToTempDirThenUpload() throws Exception {
     StorageOptions options = null;
     if (transport == Transport.GRPC) {
@@ -94,6 +96,13 @@ public final class ITBlobWriteSessionTest {
               .setBlobWriteSessionConfig(
                   BlobWriteSessionConfigs.getDefault().withChunkSize(256 * 1024))
               .build();
+    } else if (transport == Transport.HTTP) {
+      options =
+          ((HttpStorageOptions) storage.getOptions())
+              .toBuilder()
+              .setBlobWriteSessionConfig(
+                  BlobWriteSessionConfigs.getDefault().withChunkSize(256 * 1024))
+              .build();
     }
     assertWithMessage("unable to resolve options").that(options).isNotNull();
     //noinspection DataFlowIssue
@@ -103,6 +112,7 @@ public final class ITBlobWriteSessionTest {
   }
 
   @Test
+  @CrossRun.Exclude(transports = Transport.HTTP)
   public void bidiTest() throws Exception {
     StorageOptions options = null;
     if (transport == Transport.GRPC) {
diff --git a/samples/snippets/src/test/java/com/example/storage/ITBucketSnippets.java b/samples/snippets/src/test/java/com/example/storage/ITBucketSnippets.java
index 2f61fdbb..dc68167a 100644
--- a/samples/snippets/src/test/java/com/example/storage/ITBucketSnippets.java
+++ b/samples/snippets/src/test/java/com/example/storage/ITBucketSnippets.java
@@ -112,8 +112,8 @@ public class ITBucketSnippets {
   private static final String BUCKET = RemoteStorageHelper.generateBucketName();
   private static final String PROJECT_ID = System.getenv("GOOGLE_CLOUD_PROJECT");
   private static final String KMS_KEY_NAME =
-      "projects/java-docs-samples-testing/locations/us/keyRings/"
-          + "jds_test_kms_key_ring/cryptoKeys/gcs_kms_key_one";
+      "projects/cloud-java-ci-sample/locations/us/keyRings/"
+          + "gcs_test_kms_key_ring/cryptoKeys/gcs_kms_key_one";
   private static final RetrySettings RETRY_SETTINGS =
       RetrySettings.newBuilder()
           .setInitialRetryDelay(Duration.ofSeconds(2))
@@ -465,8 +465,8 @@ public class ITBucketSnippets {
         .get(BUCKET)
         .toBuilder()
         .setDefaultKmsKeyName(
-            "projects/java-docs-samples-testing/locations/us/keyRings/"
-                + "jds_test_kms_key_ring/cryptoKeys/gcs_kms_key_one")
+            "projects/cloud-java-ci-sample/locations/us/keyRings/"
+                + "gcs_test_kms_key_ring/cryptoKeys/gcs_kms_key_one")
         .build()
         .update();
     TestUtils.retryAssert(
diff --git a/samples/snippets/src/test/java/com/example/storage/ITObjectSnippets.java b/samples/snippets/src/test/java/com/example/storage/ITObjectSnippets.java
index feca2ca8..0ebaa55c 100644
--- a/samples/snippets/src/test/java/com/example/storage/ITObjectSnippets.java
+++ b/samples/snippets/src/test/java/com/example/storage/ITObjectSnippets.java
@@ -97,8 +97,8 @@ public class ITObjectSnippets {
   private static final byte[] CONTENT = STRING_CONTENT.getBytes(UTF_8);
   private static final String PROJECT_ID = System.getenv("GOOGLE_CLOUD_PROJECT");
   private static final String KMS_KEY_NAME =
-      "projects/java-docs-samples-testing/locations/us/keyRings/"
-          + "jds_test_kms_key_ring/cryptoKeys/gcs_kms_key_one";
+      "projects/cloud-java-ci-sample/locations/us/keyRings/"
+          + "gcs_test_kms_key_ring/cryptoKeys/gcs_kms_key_one";
 
   private static Storage storage;
 
diff --git a/samples/snippets/src/test/java/com/example/storage/bucket/CreateBucketPubSubNotificationTest.java b/samples/snippets/src/test/java/com/example/storage/bucket/CreateBucketPubSubNotificationTest.java
index a3ff52a3..7bfd9524 100644
--- a/samples/snippets/src/test/java/com/example/storage/bucket/CreateBucketPubSubNotificationTest.java
+++ b/samples/snippets/src/test/java/com/example/storage/bucket/CreateBucketPubSubNotificationTest.java
@@ -40,6 +40,8 @@ public class CreateBucketPubSubNotificationTest extends TestBase {
       Notification.PayloadFormat.JSON_API_V1.JSON_API_V1;
   private static final Map<String, String> CUSTOM_ATTRIBUTES = ImmutableMap.of("label1", "value1");
   private static final String PROJECT = System.getenv("GOOGLE_CLOUD_PROJECT");
+  private static final String PROJECT_NUMBER = System.getenv("GOOGLE_CLOUD_PROJECT_NUMBER");
+
   private static final String ID = UUID.randomUUID().toString().substring(0, 8);
   private static final String TOPIC =
       String.format("projects/%s/topics/new-topic-create-%s", PROJECT, ID);
@@ -57,8 +59,15 @@ public class CreateBucketPubSubNotificationTest extends TestBase {
       GetIamPolicyRequest getIamPolicyRequest =
           GetIamPolicyRequest.newBuilder().setResource(TOPIC).build();
       com.google.iam.v1.Policy policy = topicAdminClient.getIamPolicy(getIamPolicyRequest);
-      Binding binding =
-          Binding.newBuilder().setRole("roles/owner").addMembers("allAuthenticatedUsers").build();
+      // For available bindings identities, see
+      // https://cloud.google.com/iam/docs/overview#concepts_related_identity
+      String member =
+          PROJECT_NUMBER != null
+              ? "serviceAccount:service-"
+                  + PROJECT_NUMBER
+                  + "@gs-project-accounts.iam.gserviceaccount.com"
+              : "allAuthenticatedUsers";
+      Binding binding = Binding.newBuilder().setRole("roles/owner").addMembers(member).build();
       SetIamPolicyRequest setIamPolicyRequest =
           SetIamPolicyRequest.newBuilder()
               .setResource(TOPIC)
diff --git a/samples/snippets/src/test/java/com/example/storage/bucket/DeleteBucketPubSubNotificationTest.java b/samples/snippets/src/test/java/com/example/storage/bucket/DeleteBucketPubSubNotificationTest.java
index d83960c4..aab1073b 100644
--- a/samples/snippets/src/test/java/com/example/storage/bucket/DeleteBucketPubSubNotificationTest.java
+++ b/samples/snippets/src/test/java/com/example/storage/bucket/DeleteBucketPubSubNotificationTest.java
@@ -44,6 +44,8 @@ public class DeleteBucketPubSubNotificationTest extends TestBase {
       Notification.PayloadFormat.JSON_API_V1.JSON_API_V1;
   private static final Map<String, String> CUSTOM_ATTRIBUTES = ImmutableMap.of("label1", "value1");
   private static final String PROJECT = System.getenv("GOOGLE_CLOUD_PROJECT");
+  private static final String PROJECT_NUMBER = System.getenv("GOOGLE_CLOUD_PROJECT_NUMBER");
+
   private static final String ID = UUID.randomUUID().toString().substring(0, 8);
   private static final String TOPIC =
       String.format("projects/%s/topics/new-topic-delete-%s", PROJECT, ID);
@@ -62,8 +64,15 @@ public class DeleteBucketPubSubNotificationTest extends TestBase {
       GetIamPolicyRequest getIamPolicyRequest =
           GetIamPolicyRequest.newBuilder().setResource(TOPIC).build();
       com.google.iam.v1.Policy policy = topicAdminClient.getIamPolicy(getIamPolicyRequest);
-      Binding binding =
-          Binding.newBuilder().setRole("roles/owner").addMembers("allAuthenticatedUsers").build();
+      // For available bindings identities, see
+      // https://cloud.google.com/iam/docs/overview#concepts_related_identity
+      String member =
+          PROJECT_NUMBER != null
+              ? "serviceAccount:service-"
+                  + PROJECT_NUMBER
+                  + "@gs-project-accounts.iam.gserviceaccount.com"
+              : "allAuthenticatedUsers";
+      Binding binding = Binding.newBuilder().setRole("roles/owner").addMembers(member).build();
       SetIamPolicyRequest setIamPolicyRequest =
           SetIamPolicyRequest.newBuilder()
               .setResource(TOPIC)
diff --git a/samples/snippets/src/test/java/com/example/storage/bucket/ListPubSubNotificationsTest.java b/samples/snippets/src/test/java/com/example/storage/bucket/ListPubSubNotificationsTest.java
index 643c51bf..88a4613e 100644
--- a/samples/snippets/src/test/java/com/example/storage/bucket/ListPubSubNotificationsTest.java
+++ b/samples/snippets/src/test/java/com/example/storage/bucket/ListPubSubNotificationsTest.java
@@ -44,6 +44,8 @@ public class ListPubSubNotificationsTest extends TestBase {
       Notification.PayloadFormat.JSON_API_V1.JSON_API_V1;
   private static final Map<String, String> CUSTOM_ATTRIBUTES = ImmutableMap.of("label1", "value1");
   private static final String PROJECT = System.getenv("GOOGLE_CLOUD_PROJECT");
+  private static final String PROJECT_NUMBER = System.getenv("GOOGLE_CLOUD_PROJECT_NUMBER");
+
   private static final String ID = UUID.randomUUID().toString().substring(0, 8);
   private static final String TOPIC =
       String.format("projects/%s/topics/new-topic-list-%s", PROJECT, ID);
@@ -57,8 +59,15 @@ public class ListPubSubNotificationsTest extends TestBase {
       GetIamPolicyRequest getIamPolicyRequest =
           GetIamPolicyRequest.newBuilder().setResource(TOPIC).build();
       com.google.iam.v1.Policy policy = topicAdminClient.getIamPolicy(getIamPolicyRequest);
-      Binding binding =
-          Binding.newBuilder().setRole("roles/owner").addMembers("allAuthenticatedUsers").build();
+      // For available bindings identities, see
+      // https://cloud.google.com/iam/docs/overview#concepts_related_identity
+      String member =
+          PROJECT_NUMBER != null
+              ? "serviceAccount:service-"
+                  + PROJECT_NUMBER
+                  + "@gs-project-accounts.iam.gserviceaccount.com"
+              : "allAuthenticatedUsers";
+      Binding binding = Binding.newBuilder().setRole("roles/owner").addMembers(member).build();
       SetIamPolicyRequest setIamPolicyRequest =
           SetIamPolicyRequest.newBuilder()
               .setResource(TOPIC)
diff --git a/samples/snippets/src/test/java/com/example/storage/bucket/PrintPubSubNotificationTest.java b/samples/snippets/src/test/java/com/example/storage/bucket/PrintPubSubNotificationTest.java
index ce3dd804..d342795a 100644
--- a/samples/snippets/src/test/java/com/example/storage/bucket/PrintPubSubNotificationTest.java
+++ b/samples/snippets/src/test/java/com/example/storage/bucket/PrintPubSubNotificationTest.java
@@ -40,6 +40,8 @@ public class PrintPubSubNotificationTest extends TestBase {
       Notification.PayloadFormat.JSON_API_V1.JSON_API_V1;
   private static final Map<String, String> CUSTOM_ATTRIBUTES = ImmutableMap.of("label1", "value1");
   private static final String PROJECT = System.getenv("GOOGLE_CLOUD_PROJECT");
+  private static final String PROJECT_NUMBER = System.getenv("GOOGLE_CLOUD_PROJECT_NUMBER");
+
   private static final String ID = UUID.randomUUID().toString().substring(0, 8);
   private static final String TOPIC =
       String.format("projects/%s/topics/new-topic-print-%s", PROJECT, ID);
@@ -53,8 +55,15 @@ public class PrintPubSubNotificationTest extends TestBase {
       GetIamPolicyRequest getIamPolicyRequest =
           GetIamPolicyRequest.newBuilder().setResource(TOPIC).build();
       com.google.iam.v1.Policy policy = topicAdminClient.getIamPolicy(getIamPolicyRequest);
-      Binding binding =
-          Binding.newBuilder().setRole("roles/owner").addMembers("allAuthenticatedUsers").build();
+      // For available bindings identities, see
+      // https://cloud.google.com/iam/docs/overview#concepts_related_identity
+      String member =
+          PROJECT_NUMBER != null
+              ? "serviceAccount:service-"
+                  + PROJECT_NUMBER
+                  + "@gs-project-accounts.iam.gserviceaccount.com"
+              : "allAuthenticatedUsers";
+      Binding binding = Binding.newBuilder().setRole("roles/owner").addMembers(member).build();
       SetIamPolicyRequest setIamPolicyRequest =
           SetIamPolicyRequest.newBuilder()
               .setResource(TOPIC)