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;
 
@@ -95,40 +97,49 @@ public void prepareRun(BatchSourceContext context) throws Exception {
       bucketName = GCSPath.from(path).getBucket();
     } catch (IllegalArgumentException e) {
       collector.addFailure(e.getMessage(), null)
-        .withStacktrace(e.getStackTrace());
+          .withStacktrace(e.getStackTrace());
       collector.getOrThrowException();
     }
 
     Boolean isServiceAccountFilePath = config.connection.isServiceAccountFilePath();
     if (isServiceAccountFilePath == null) {
       collector.addFailure("Service account type is undefined.",
-        "Must be `filePath` or `JSON`");
+          "Must be `filePath` or `JSON`");
       collector.getOrThrowException();
     }
 
     Credentials credentials = null;
     try {
       credentials = config.connection.getServiceAccount() == null ?
-        null : GCPUtils.loadServiceAccountCredentials(config.connection.getServiceAccount(),
-        isServiceAccountFilePath);
+          null : GCPUtils.loadServiceAccountCredentials(config.connection.getServiceAccount(),
+          isServiceAccountFilePath);
     } catch (Exception e) {
       String errorReason = "Unable to load service account credentials.";
       collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), null)
-        .withStacktrace(e.getStackTrace());
+          .withStacktrace(e.getStackTrace());
       collector.getOrThrowException();
     }
 
     Storage storage = GCPUtils.getStorage(config.connection.getProject(), credentials);
     String location = null;
     try {
       // Get location of the source for lineage
-      location = storage.get(bucketName).getLocation();
+      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();
+      }
+      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();
     }
 
@@ -142,7 +153,7 @@ public void prepareRun(BatchSourceContext context) throws Exception {
 
     // set error details provider
     context.setErrorDetailsProvider(
-      new ErrorDetailsProviderSpec(GCSErrorDetailsProvider.class.getName()));
+        new ErrorDetailsProviderSpec(GCSErrorDetailsProvider.class.getName()));
 
     // super is called down here to avoid instantiating the lineage recorder with a null asset
     super.prepareRun(context);
@@ -151,12 +162,12 @@ public void prepareRun(BatchSourceContext context) throws Exception {
   @Override
   protected Map<String, String> getFileSystemProperties(BatchSourceContext context) {
     Map<String, String> properties = GCPUtils.getFileSystemProperties(config.connection, config.getPath(),
-                                                                      new HashMap<>(config.getFileSystemProperties()));
+        new HashMap<>(config.getFileSystemProperties()));
     if (config.isCopyHeader()) {
       properties.put(PathTrackingInputFormat.COPY_HEADER, Boolean.TRUE.toString());
     }
     if (config.getFileEncoding() != null
-      && !config.getFileEncoding().equalsIgnoreCase(AbstractFileSourceConfig.DEFAULT_FILE_ENCODING)) {
+        && !config.getFileEncoding().equalsIgnoreCase(AbstractFileSourceConfig.DEFAULT_FILE_ENCODING)) {
       properties.put(PathTrackingInputFormat.SOURCE_FILE_ENCODING, config.getFileEncoding());
     }
     if (config.getMinSplitSize() != null) {
@@ -179,17 +190,17 @@ protected LineageRecorder getLineageRecorder(BatchSourceContext context) {
   @Override
   protected void recordLineage(LineageRecorder lineageRecorder, List<String> outputFields) {
     lineageRecorder.recordRead("Read", String.format("Read %s from Google Cloud Storage.",
-                                                     config.isEncrypted() ? " and decrypt " : " "), outputFields);
+        config.isEncrypted() ? " and decrypt " : " "), outputFields);
   }
 
   @Override
   protected boolean shouldGetSchema() {
     return !config.containsMacro(GCPConnectorConfig.NAME_PROJECT) &&
-             !config.containsMacro(GCSSourceConfig.NAME_PATH) && !config.containsMacro(GCSSourceConfig.NAME_FORMAT) &&
-             !config.containsMacro(GCSSourceConfig.NAME_DELIMITER) &&
-             !config.containsMacro(GCSSourceConfig.NAME_FILE_SYSTEM_PROPERTIES) &&
-             !config.containsMacro(GCPConnectorConfig.NAME_SERVICE_ACCOUNT_FILE_PATH) &&
-             !config.containsMacro(GCPConnectorConfig.NAME_SERVICE_ACCOUNT_JSON);
+        !config.containsMacro(GCSSourceConfig.NAME_PATH) && !config.containsMacro(GCSSourceConfig.NAME_FORMAT) &&
+        !config.containsMacro(GCSSourceConfig.NAME_DELIMITER) &&
+        !config.containsMacro(GCSSourceConfig.NAME_FILE_SYSTEM_PROPERTIES) &&
+        !config.containsMacro(GCPConnectorConfig.NAME_SERVICE_ACCOUNT_FILE_PATH) &&
+        !config.containsMacro(GCPConnectorConfig.NAME_SERVICE_ACCOUNT_JSON);
   }
 
   /**
@@ -225,19 +236,19 @@ public static class GCSSourceConfig extends AbstractFileSourceConfig implements
     @Macro
     @Nullable
     @Description("Whether the data file is encrypted. If it is set to 'true', a associated metadata file needs to be "
-      + "provided for each data file. Please refer to the Documentation for the details of the metadata file content.")
+        + "provided for each data file. Please refer to the Documentation for the details of the metadata file content.")
     private Boolean encrypted;
 
     @Macro
     @Nullable
     @Description("The file name suffix for the metadata file of the encrypted data file. "
-      + "The default is '" + DEFAULT_ENCRYPTED_METADATA_SUFFIX + "'.")
+        + "The default is '" + DEFAULT_ENCRYPTED_METADATA_SUFFIX + "'.")
     private String encryptedMetadataSuffix;
 
     @Macro
     @Nullable
     @Description("A list of columns with the corresponding data types for whom the automatic data type detection gets" +
-      " skipped.")
+        " skipped.")
     private String override;
 
     @Macro
@@ -270,23 +281,23 @@ public void validate(FailureCollector collector) {
           GCSPath.from(path);
         } catch (IllegalArgumentException e) {
           collector.addFailure(e.getMessage(), null).withConfigProperty(NAME_PATH)
-            .withStacktrace(e.getStackTrace());
+              .withStacktrace(e.getStackTrace());
         }
       }
       if (!containsMacro(NAME_FILE_SYSTEM_PROPERTIES)) {
         try {
           getFileSystemProperties();
         } catch (Exception e) {
           collector.addFailure("File system properties must be a valid json.", null)
-            .withConfigProperty(NAME_FILE_SYSTEM_PROPERTIES).withStacktrace(e.getStackTrace());
+              .withConfigProperty(NAME_FILE_SYSTEM_PROPERTIES).withStacktrace(e.getStackTrace());
         }
       }
       if (!containsMacro(NAME_FILE_REGEX)) {
         try {
           getFilePattern();
         } catch (IllegalArgumentException e) {
           collector.addFailure(e.getMessage(), null).withConfigProperty(NAME_FILE_REGEX)
-            .withStacktrace(e.getStackTrace());
+              .withStacktrace(e.getStackTrace());
         }
       }
     }
@@ -320,7 +331,7 @@ public boolean isEncrypted() {
 
     public String getEncryptedMetadataSuffix() {
       return Strings.isNullOrEmpty(encryptedMetadataSuffix) ?
-        DEFAULT_ENCRYPTED_METADATA_SUFFIX : encryptedMetadataSuffix;
+          DEFAULT_ENCRYPTED_METADATA_SUFFIX : encryptedMetadataSuffix;
     }
 
     Map<String, String> getFileSystemProperties() {
@@ -330,4 +341,4 @@ Map<String, String> getFileSystemProperties() {
       return GSON.fromJson(fileSystemProperties, MAP_STRING_STRING_TYPE);
     }
   }
-}
+}