feat: Add support to disable logging from bucket

google-cloud-storage/src/main/java/com/google/cloud/storage/BucketInfo.java

@@ -286,9 +286,14 @@ public String getLogObjectPrefix() {
     }
 
     Bucket.Logging toPb() {
-      Bucket.Logging logging = new Bucket.Logging();
-      logging.setLogBucket(logBucket);
-      logging.setLogObjectPrefix(logObjectPrefix);
+      Bucket.Logging logging;
+      if (logBucket != null || logObjectPrefix != null) {
+        logging = new Bucket.Logging();
+        logging.setLogBucket(logBucket);
+        logging.setLogObjectPrefix(logObjectPrefix);
+      } else {
+        logging = Data.nullOf(Bucket.Logging.class);
+      }
       return logging;
     }
 
@@ -1310,7 +1315,7 @@ public Builder setIamConfiguration(IamConfiguration iamConfiguration) {
 
     @Override
     public Builder setLogging(Logging logging) {
-      this.logging = logging;
+      this.logging = logging != null ? logging : BucketInfo.Logging.newBuilder().build();
       return this;
     }
 

google-cloud-storage/src/test/java/com/google/cloud/storage/BucketInfoTest.java

@@ -229,6 +229,7 @@ public void testToPbAndFromPb() {
         BucketInfo.newBuilder("b")
             .setDeleteRules(DELETE_RULES)
             .setLifecycleRules(LIFECYCLE_RULES)
+            .setLogging(LOGGING)
             .build();
     compareBuckets(bucketInfo, BucketInfo.fromPb(bucketInfo.toPb()));
   }

google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java

@@ -863,4 +863,27 @@ public void testDeleteLifecycleRules() {
     Bucket actualUpdatedBucket = updatedBucket.update();
     assertThat(actualUpdatedBucket.getLifecycleRules()).hasSize(0);
   }
+
+  @Test
+  public void testUpdateBucketLogging() {
+    initializeExpectedBucket(6);
+    BucketInfo.Logging logging =
+        BucketInfo.Logging.newBuilder()
+            .setLogBucket("logs-bucket")
+            .setLogObjectPrefix("test-logs")
+            .build();
+    BucketInfo bucketInfo = BucketInfo.newBuilder("b").setLogging(logging).build();
+    Bucket bucket = new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(bucketInfo));
+    assertThat(bucket.getLogging().getLogBucket()).isEqualTo("logs-bucket");
+    assertThat(bucket.getLogging().getLogObjectPrefix()).isEqualTo("test-logs");
+    Bucket expectedUpdatedBucket = bucket.toBuilder().setLogging(null).build();
+    expect(storage.getOptions()).andReturn(mockOptions).times(2);
+    expect(storage.update(expectedUpdatedBucket)).andReturn(expectedUpdatedBucket);
+    replay(storage);
+    initializeBucket();
+    Bucket updatedBucket = new Bucket(storage, new BucketInfo.BuilderImpl(expectedUpdatedBucket));
+    Bucket actualUpdatedBucket = updatedBucket.update();
+    assertThat(actualUpdatedBucket.getLogging().getLogBucket()).isNull();
+    assertThat(actualUpdatedBucket.getLogging().getLogObjectPrefix()).isNull();
+  }
 }

google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java

@@ -3227,13 +3227,7 @@ public void testBucketLogging() throws ExecutionException, InterruptedException
     try {
       assertNotNull(storage.create(BucketInfo.newBuilder(logsBucket).setLocation("us").build()));
       Policy policy = storage.getIamPolicy(logsBucket);
-      assertNotNull(
-          storage.setIamPolicy(
-              logsBucket,
-              policy
-                  .toBuilder()
-                  .addIdentity(StorageRoles.legacyBucketWriter(), Identity.allAuthenticatedUsers())
-                  .build()));
+      assertNotNull(policy);
       BucketInfo.Logging logging =
           BucketInfo.Logging.newBuilder()
               .setLogBucket(logsBucket)
@@ -3244,6 +3238,11 @@ public void testBucketLogging() throws ExecutionException, InterruptedException
               BucketInfo.newBuilder(loggingBucket).setLocation("us").setLogging(logging).build());
       assertEquals(logsBucket, bucket.getLogging().getLogBucket());
       assertEquals("test-logs", bucket.getLogging().getLogObjectPrefix());
+
+      // Disable bucket logging.
+      Bucket updatedBucket = bucket.toBuilder().setLogging(null).build().update();
+      assertNull(updatedBucket.getLogging());
+
     } finally {
       RemoteStorageHelper.forceDelete(storage, logsBucket, 5, TimeUnit.SECONDS);
       RemoteStorageHelper.forceDelete(storage, loggingBucket, 5, TimeUnit.SECONDS);