Improve Authentication Error handling

airbyte-integrations/bases/base-java-s3/src/main/java/io/airbyte/integrations/destination/s3/BlobStorageOperations.java

@@ -22,9 +22,9 @@ protected BlobStorageOperations() {
   public abstract String getBucketObjectPath(String namespace, String streamName, DateTime writeDatetime, String customFormat);
 
   /**
-   * Create a storage object where to store data in the destination for a @param objectPath
+   * Ensure that the bucket specified in the config exists
    */
-  public abstract void createBucketObjectIfNotExists(String objectPath) throws Exception;
+  public abstract void createBucketIfNotExists() throws Exception;
 
   /**
    * Upload the data files into the storage area.

airbyte-integrations/bases/base-java-s3/src/main/java/io/airbyte/integrations/destination/s3/S3BaseChecks.java

@@ -120,7 +120,7 @@ private static void attemptWriteAndDeleteS3Object(final S3StorageOperations stor
     final var bucketPath = s3Config.getBucketPath();
 
     if (!Strings.isNullOrEmpty(bucketPath)) {
-      storageOperations.createBucketObjectIfNotExists(bucketPath);
+      storageOperations.createBucketIfNotExists();
     }
     s3.putObject(s3Bucket, outputTableName, "check-content");
     testIAMUserHasListObjectPermission(s3, s3Bucket);

airbyte-integrations/bases/base-java-s3/src/main/java/io/airbyte/integrations/destination/s3/S3StorageOperations.java

@@ -8,13 +8,15 @@
 
 import alex.mojaki.s3upload.StreamTransferManager;
 import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.AmazonS3Exception;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion;
 import com.amazonaws.services.s3.model.ListObjectsRequest;
 import com.amazonaws.services.s3.model.ObjectListing;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.string.Strings;
 import io.airbyte.integrations.destination.NamingConventionTransformer;
 import io.airbyte.integrations.destination.record_buffer.SerializableBuffer;
@@ -95,23 +97,15 @@ public String getBucketObjectPath(final String namespace, final String streamNam
 
   /**
    * Create a directory object at the specified location. Creates the bucket if necessary.
-   *
-   * @param objectPath The directory to create. Must be a nonempty string.
    */
   @Override
-  public void createBucketObjectIfNotExists(final String objectPath) {
+  public void createBucketIfNotExists() {
     final String bucket = s3Config.getBucketName();
-    final String folderPath = objectPath.endsWith("/") ? objectPath : objectPath + "/";
     if (!doesBucketExist(bucket)) {
       LOGGER.info("Bucket {} does not exist; creating...", bucket);
       s3Client.createBucket(bucket);
       LOGGER.info("Bucket {} has been created.", bucket);
     }
-    if (!s3Client.doesObjectExist(bucket, folderPath)) {
-      LOGGER.info("Storage Object {}/{} does not exist in bucket; creating...", bucket, objectPath);
-      s3Client.putObject(bucket, folderPath, "");
-      LOGGER.info("Storage Object {}/{} has been created in bucket.", bucket, objectPath);
-    }
   }
 
   protected boolean doesBucketExist(final String bucket) {
@@ -138,7 +132,13 @@ public String uploadRecordsToBucket(final SerializableBuffer recordsData,
         exceptionsThrown.add(e);
       }
     }
-    throw new RuntimeException(String.format("Exceptions thrown while uploading records into storage: %s", Strings.join(exceptionsThrown, "\n")));
+    final boolean areAllExceptionsAuthExceptions = exceptionsThrown.stream().filter(e -> e instanceof AmazonS3Exception)
+        .map(s3e -> ((AmazonS3Exception) s3e).getStatusCode()).allMatch(code -> Integer.valueOf(403).equals(code));
+    if (areAllExceptionsAuthExceptions) {
+      throw new ConfigErrorException(exceptionsThrown.get(0).getMessage(), exceptionsThrown.get(0));
+    } else {
+      throw new RuntimeException(String.format("Exceptions thrown while uploading records into storage: %s", Strings.join(exceptionsThrown, "\n")));
+    }
   }
 
   /**

airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryGcsOperations.java

@@ -13,6 +13,8 @@
 import com.google.cloud.bigquery.LoadJobConfiguration;
 import com.google.cloud.bigquery.Schema;
 import com.google.cloud.bigquery.TableId;
+import com.google.common.collect.ImmutableList;
+import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.integrations.destination.StandardNameTransformer;
 import io.airbyte.integrations.destination.bigquery.uploader.AbstractBigQueryUploader;
 import io.airbyte.integrations.destination.gcs.GcsDestinationConfig;
@@ -31,6 +33,7 @@ public class BigQueryGcsOperations implements BigQueryStagingOperations {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryGcsOperations.class);
 
+  private static final List<Integer> AUTH_ERROR_CODES = ImmutableList.of(401, 403);
   private final BigQuery bigQuery;
   private final StandardNameTransformer gcsNameTransformer;
   private final GcsDestinationConfig gcsConfig;
@@ -84,7 +87,15 @@ public String getStagingFullPath(final String datasetId, final String stream) {
   public void createSchemaIfNotExists(final String datasetId, final String datasetLocation) {
     if (!existingSchemas.contains(datasetId)) {
       LOGGER.info("Creating dataset {}", datasetId);
-      BigQueryUtils.getOrCreateDataset(bigQuery, datasetId, datasetLocation);
+      try {
+        BigQueryUtils.getOrCreateDataset(bigQuery, datasetId, datasetLocation);
+      } catch (BigQueryException bqe) {
+        if (AUTH_ERROR_CODES.contains(bqe.getCode())) {
+          throw new ConfigErrorException(bqe.getMessage(), bqe);
+        } else {
+          throw bqe;
+        }
+      }
       existingSchemas.add(datasetId);
     }
   }
@@ -99,7 +110,7 @@ public void createTmpTableIfNotExists(final TableId tmpTableId, final Schema tab
   public void createStageIfNotExists(final String datasetId, final String stream) {
     final String objectPath = getStagingFullPath(datasetId, stream);
     LOGGER.info("Creating staging path for stream {} (dataset {}): {}", stream, datasetId, objectPath);
-    gcsStorageOperations.createBucketObjectIfNotExists(objectPath);
+    gcsStorageOperations.createBucketIfNotExists();
   }
 
   @Override

airbyte-integrations/connectors/destination-bigquery/src/test-integration/java/io/airbyte/integrations/destination/bigquery/BigQueryDestinationTest.java

@@ -81,7 +81,7 @@ class BigQueryDestinationTest {
   protected static final Path CREDENTIALS_WITH_GCS_STAGING_PATH =
       Path.of("secrets/credentials-gcs-staging.json");
 
-  protected static final Path[] ALL_PATHS = {CREDENTIALS_WITH_GCS_STAGING_PATH, CREDENTIALS_BAD_PROJECT_PATH, CREDENTIALS_NO_DATASET_CREATION_PATH,
+  protected static final Path[] ALL_PATHS = {CREDENTIALS_STANDARD_INSERT_PATH, CREDENTIALS_BAD_PROJECT_PATH, CREDENTIALS_NO_DATASET_CREATION_PATH,
       CREDENTIALS_NON_BILLABLE_PROJECT_PATH, CREDENTIALS_WITH_GCS_STAGING_PATH};
   private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryDestinationTest.class);
   private static final String DATASET_NAME_PREFIX = "bq_dest_integration_test";

airbyte-integrations/connectors/destination-redshift/src/main/java/io/airbyte/integrations/destination/redshift/operations/RedshiftS3StagingSqlOperations.java

@@ -77,7 +77,7 @@ public String getStagingPath(UUID connectionId, String namespace, String streamN
   public void createStageIfNotExists(JdbcDatabase database, String stageName) throws Exception {
     final String bucketPath = s3Config.getBucketPath();
     final String prefix = bucketPath.isEmpty() ? "" : bucketPath + (bucketPath.endsWith("/") ? "" : "/");
-    s3StorageOperations.createBucketObjectIfNotExists(prefix + stageName);
+    s3StorageOperations.createBucketIfNotExists();
   }
 
   @Override

airbyte-integrations/connectors/destination-snowflake/src/main/java/io/airbyte/integrations/destination/snowflake/SnowflakeS3StagingSqlOperations.java

@@ -82,7 +82,7 @@ public String uploadRecordsToStage(final JdbcDatabase database,
 
   @Override
   public void createStageIfNotExists(final JdbcDatabase database, final String stageName) {
-    s3StorageOperations.createBucketObjectIfNotExists(stageName);
+    s3StorageOperations.createBucketIfNotExists();
   }
 
   @Override

docs/integrations/destinations/redshift.md

@@ -178,5 +178,3 @@ Each stream will be output into its own raw table in Redshift. Each table will c
 | 0.3.13  | 2021-09-02 | [5745](https://github.com/airbytehq/airbyte/pull/5745)     | Disable STATUPDATE flag when using S3 staging to speed up performance                                                                                                                                            |
 | 0.3.12  | 2021-07-21 | [3555](https://github.com/airbytehq/airbyte/pull/3555)     | Enable partial checkpointing for halfway syncs                                                                                                                                                                   |
 | 0.3.11  | 2021-07-20 | [4874](https://github.com/airbytehq/airbyte/pull/4874)     | allow `additionalProperties` in connector spec                                                                                                                                                                   |
-
-