Fixed NPE on GCSSource when bucket does not exist

Author: shweta-deosarkar

src/main/java/io/cdap/plugin/gcp/gcs/source/GCSSource.java

@@ -17,6 +17,7 @@
 package io.cdap.plugin.gcp.gcs.source;
 
 import com.google.auth.Credentials;
+import com.google.cloud.storage.Bucket;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageException;
 import com.google.common.base.Strings;
@@ -55,6 +56,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.regex.Pattern;
 import javax.annotation.Nullable;
 
@@ -119,16 +121,24 @@ public void prepareRun(BatchSourceContext context) throws Exception {
     }
 
     Storage storage = GCPUtils.getStorage(config.connection.getProject(), credentials);
+    Bucket bucket = storage.get(bucketName);
+    if (Objects.isNull(bucket)) {
+      String errorReason = String.format("Unable to access GCS bucket '%s'.",
+              bucketName);
+      collector.addFailure(String.format("%s Ensure you entered the correct bucket path.", errorReason),
+              null);
+      collector.getOrThrowException();
+    }
     String location = null;
     try {
       // Get location of the source for lineage
-      location = storage.get(bucketName).getLocation();
+      location = bucket.getLocation();
     } catch (StorageException e) {
       String errorReason = String.format("Error code: %s, Unable to access GCS bucket '%s'. ",
-        e.getCode(), bucketName);
+              e.getCode(), bucketName);
       collector.addFailure(String.format("%s %s", errorReason, e.getMessage()),
-          "Ensure you entered the correct bucket path and have permissions for it.")
-        .withStacktrace(e.getStackTrace());
+                      "Ensure you entered the correct bucket path and have permissions for it.")
+              .withStacktrace(e.getStackTrace());
       collector.getOrThrowException();
     }