Merge pull request data-integrations#1359 from data-integrations/fix/plugin-1738

[PLUGIN-1738] Fix bucket creation issue for GCS Copy/Move Plugins

Author: dj-smart

src/main/java/io/cdap/plugin/gcp/gcs/StorageClient.java

@@ -115,16 +115,20 @@ public void mapMetaDataForAllBlobs(String path, Consumer<Map<String, String>> fu
    * @param cmekKeyName  the name of the cmek key
    */
   public void createBucketIfNotExists(GCSPath path, @Nullable String location, @Nullable CryptoKeyName cmekKeyName) {
-    Bucket bucket = null;
     try {
-      bucket = storage.get(path.getBucket());
-    } catch (StorageException e) {
-      throw new RuntimeException(
-        String.format("Unable to access bucket %s. ", path.getBucket())
-          + "Ensure you entered the correct bucket path and have permissions for it.", e);
-    }
-    if (bucket == null) {
       GCPUtils.createBucket(storage, path.getBucket(), location, cmekKeyName);
+      LOG.info("Bucket {} has been created successfully", path.getBucket());
+    } catch (StorageException e) {
+      // Don't throw error if bucket already exists
+      // https://cloud.google.com/storage/docs/json_api/v1/status-codes#409_Conflict
+      if (e.getCode() == 409) {
+        LOG.warn("Getting 409 Conflict: {} Bucket at destination path {} may already exist.",
+                 e.getMessage(), path.getUri());
+      } else {
+        throw new RuntimeException(
+          String.format("Unable to create bucket %s. Ensure you entered the correct bucket path and " +
+                          "have permissions for it.", path.getBucket()), e);
+      }
     }
   }
 

src/test/java/io/cdap/plugin/gcp/gcs/StorageClientTest.java

@@ -17,14 +17,52 @@
 package io.cdap.plugin.gcp.gcs;
 
 import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import org.junit.After;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.slf4j.Logger;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
 
 /**
  * Tests for storage client
  */
 public class StorageClientTest {
 
+  @Mock
+  private Storage storage;
+
+  private StorageClient storageClient;
+
+  private final ByteArrayOutputStream outContent = new ByteArrayOutputStream();
+
+  private final PrintStream originalOut = System.out;
+
+  @Before
+  public void setUp() {
+    MockitoAnnotations.initMocks(this);
+    storageClient = new StorageClient(storage);
+    System.setOut(new PrintStream(outContent));
+  }
+
+  @After
+  public void restoreStreams() {
+    System.setOut(originalOut);
+  }
+
   @Test
   public void testAppend() {
     Assert.assertEquals("a/b/c", StorageClient.append("a/", "/b/c"));
@@ -67,4 +105,43 @@ public void testNonExistingDestinationEndingSlash() {
     Assert.assertEquals(BlobId.of("b0", "subdir/dir2/a/b/c"),
                         StorageClient.resolve("dir1/dir2", "dir1/dir2/a/b/c", GCSPath.from("b0/subdir/"), false));
   }
+
+  @Test
+  public void testCreateBucketIfNotExists() {
+    // Test successful bucket creation
+    GCSPath path = GCSPath.from("my-bucket");
+    storageClient.createBucketIfNotExists(path, "us", null);
+    // No exception is thrown and method storage.create() is invoked once
+    verify(storage, times(1)).create(any(BucketInfo.class));
+
+    // Test bucket already exists
+    GCSPath existingPath = GCSPath.from("existing-bucket");
+
+    when(storage.create(any(BucketInfo.class))).thenThrow(new StorageException(409, "Conflict"));
+
+    storageClient.createBucketIfNotExists(existingPath, "existing-location", null);
+    // The exception thrown should be caught and warning log should be printed
+    Assert.assertTrue(outContent.toString().contains("Getting 409 Conflict"));
+    // The method storage.create() is invoked 2 times in total
+    verify(storage, times(2)).create(any(BucketInfo.class));
+
+    // Test bucket creation failure
+    GCSPath failurePath = GCSPath.from("failed-bucket");
+
+    when(storage.create(any(BucketInfo.class))).thenThrow(new StorageException(500, "Internal Server Error"));
+
+    try {
+      storageClient.createBucketIfNotExists(failurePath, "failed-location", null);
+    } catch (Exception e) {
+      // Verify that RuntimeException is caught
+      if (!(e instanceof RuntimeException)) {
+        Assert.fail(String.format("Test for detecting bucket creation failure did not succeed. " +
+                                    "Unexpected Exception caught: %s", e));
+      }
+      // The method storage.create() is invoked 3 times in total
+      verify(storage, times(3)).create(any(BucketInfo.class));
+      return;
+    }
+    Assert.fail("Test for detecting bucket creation failure did not succeed. No exception caught");
+  }
 }