Improve Authentication Error handling

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -40,7 +40,7 @@
 - name: BigQuery
   destinationDefinitionId: 22f6c74f-5699-40ff-833c-4a879ea40133
   dockerRepository: airbyte/destination-bigquery
-  dockerImageTag: 1.2.11
+  dockerImageTag: 1.2.12
   documentationUrl: https://docs.airbyte.com/integrations/destinations/bigquery
   icon: bigquery.svg
   normalizationConfig:
@@ -139,7 +139,7 @@
 - name: Google Cloud Storage (GCS)
   destinationDefinitionId: ca8f6566-e555-4b40-943a-545bf123117a
   dockerRepository: airbyte/destination-gcs
-  dockerImageTag: 0.2.12
+  dockerImageTag: 0.2.13
   documentationUrl: https://docs.airbyte.com/integrations/destinations/gcs
   icon: googlecloudstorage.svg
   resourceRequirements:
@@ -290,7 +290,7 @@
 - name: Redshift
   destinationDefinitionId: f7a7d195-377f-cf5b-70a5-be6b819019dc
   dockerRepository: airbyte/destination-redshift
-  dockerImageTag: 0.3.53
+  dockerImageTag: 0.3.54
   documentationUrl: https://docs.airbyte.com/integrations/destinations/redshift
   icon: redshift.svg
   normalizationConfig:
@@ -321,7 +321,7 @@
 - name: S3
   destinationDefinitionId: 4816b78f-1489-44c1-9060-4b19d5fa9362
   dockerRepository: airbyte/destination-s3
-  dockerImageTag: 0.3.18
+  dockerImageTag: 0.3.19
   documentationUrl: https://docs.airbyte.com/integrations/destinations/s3
   icon: s3.svg
   resourceRequirements:
@@ -348,7 +348,7 @@
 - name: Snowflake
   destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba
   dockerRepository: airbyte/destination-snowflake
-  dockerImageTag: 0.4.43
+  dockerImageTag: 0.4.44
   documentationUrl: https://docs.airbyte.com/integrations/destinations/snowflake
   icon: snowflake.svg
   normalizationConfig:

airbyte-config/init/src/main/resources/seed/destination_specs.yaml

@@ -621,7 +621,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-bigquery:1.2.11"
+- dockerImage: "airbyte/destination-bigquery:1.2.12"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/destinations/bigquery"
     connectionSpecification:
@@ -2325,7 +2325,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-gcs:0.2.12"
+- dockerImage: "airbyte/destination-gcs:0.2.13"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/destinations/gcs"
     connectionSpecification:
@@ -5123,7 +5123,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-redshift:0.3.53"
+- dockerImage: "airbyte/destination-redshift:0.3.54"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/destinations/redshift"
     connectionSpecification:
@@ -5492,7 +5492,7 @@
     supported_destination_sync_modes:
     - "append"
     - "overwrite"
-- dockerImage: "airbyte/destination-s3:0.3.18"
+- dockerImage: "airbyte/destination-s3:0.3.19"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/destinations/s3"
     connectionSpecification:
@@ -6109,7 +6109,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-snowflake:0.4.43"
+- dockerImage: "airbyte/destination-snowflake:0.4.44"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/destinations/snowflake"
     connectionSpecification:

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,19 +8,22 @@
 
 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;
 import io.airbyte.integrations.destination.s3.template.S3FilenameTemplateManager;
 import io.airbyte.integrations.destination.s3.template.S3FilenameTemplateParameterObject;
 import io.airbyte.integrations.destination.s3.util.StreamTransferManagerFactory;
+import io.airbyte.integrations.util.ConnectorExceptionUtil;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -95,23 +98,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 +133,18 @@ 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")));
+    // Verifying that ALL exceptions are authentication related before assuming this is a configuration
+    // issue
+    // reduces risk of misidentifying errors or reporting a transient error.
+    final boolean areAllExceptionsAuthExceptions = exceptionsThrown.stream().filter(e -> e instanceof AmazonS3Exception)
+        .map(s3e -> ((AmazonS3Exception) s3e).getStatusCode())
+        .filter(ConnectorExceptionUtil.HTTP_AUTHENTICATION_ERROR_CODES::contains)
+        .count() == exceptionsThrown.size();
+    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/bases/base-java/src/main/java/io/airbyte/integrations/util/ConnectorExceptionUtil.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.util;
 
+import com.google.common.collect.ImmutableList;
 import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.exceptions.ConnectionErrorException;
 import io.airbyte.integrations.base.errors.messages.ErrorMessage;
@@ -22,6 +23,8 @@ public class ConnectorExceptionUtil {
   static final String RECOVERY_CONNECTION_ERROR_MESSAGE =
       "We're having issues syncing from a Postgres replica that is configured as a hot standby server. " +
           "Please see https://docs.airbyte.com/integrations/sources/postgres/#sync-data-from-postgres-hot-standby-server for options and workarounds";
+
+  public static final List<Integer> HTTP_AUTHENTICATION_ERROR_CODES = ImmutableList.of(401, 403);
   private static final List<Predicate<Throwable>> configErrorPredicates =
       List.of(getConfigErrorPredicate(), getConnectionErrorPredicate(),
           isRecoveryConnectionExceptionPredicate(), isUnknownColumnInFieldListException());

airbyte-integrations/connectors/destination-bigquery-denormalized/Dockerfile

@@ -17,5 +17,5 @@ ENV ENABLE_SENTRY true
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=1.2.11
+LABEL io.airbyte.version=1.2.12
 LABEL io.airbyte.name=airbyte/destination-bigquery-denormalized

airbyte-integrations/connectors/destination-bigquery/Dockerfile

@@ -17,5 +17,5 @@ ENV ENABLE_SENTRY true
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=1.2.11
+LABEL io.airbyte.version=1.2.12
 LABEL io.airbyte.name=airbyte/destination-bigquery

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

@@ -13,11 +13,14 @@
 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;
 import io.airbyte.integrations.destination.gcs.GcsStorageOperations;
 import io.airbyte.integrations.destination.record_buffer.SerializableBuffer;
+import io.airbyte.integrations.util.ConnectorExceptionUtil;
 import io.airbyte.protocol.models.v0.DestinationSyncMode;
 import java.util.HashSet;
 import java.util.List;
@@ -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 e) {
+        if (ConnectorExceptionUtil.HTTP_AUTHENTICATION_ERROR_CODES.contains(e.getCode())) {
+          throw new ConfigErrorException(e.getMessage(), e);
+        } else {
+          throw e;
+        }
+      }
       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

@@ -84,8 +84,10 @@ 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_NO_EDIT_PUBLIC_SCHEMA_ROLE_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-gcs/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION destination-gcs
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.2.12
+LABEL io.airbyte.version=0.2.13
 LABEL io.airbyte.name=airbyte/destination-gcs

airbyte-integrations/connectors/destination-redshift/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION destination-redshift
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.53
+LABEL io.airbyte.version=0.3.54
 LABEL io.airbyte.name=airbyte/destination-redshift

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-s3/Dockerfile

@@ -40,5 +40,5 @@ RUN /bin/bash -c 'set -e && \
        echo "unknown arch" ;\
     fi'
 
-LABEL io.airbyte.version=0.3.18
+LABEL io.airbyte.version=0.3.19
 LABEL io.airbyte.name=airbyte/destination-s3

airbyte-integrations/connectors/destination-snowflake/Dockerfile

@@ -20,5 +20,5 @@ RUN tar xf ${APPLICATION}.tar --strip-components=1
 
 ENV ENABLE_SENTRY true
 
-LABEL io.airbyte.version=0.4.43
+LABEL io.airbyte.version=0.4.44
 LABEL io.airbyte.name=airbyte/destination-snowflake

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/bigquery.md

@@ -136,6 +136,7 @@ Now that you have set up the BigQuery destination connector, check out the follo
 
 | Version | Date       | Pull Request                                              | Subject                                                                                                                  |
 |:--------|:-----------|:----------------------------------------------------------|:-------------------------------------------------------------------------------------------------------------------------|
+| 1.2.12  | 2023-01-18 | [#21087](https://github.com/airbytehq/airbyte/pull/21087) | Wrap Authentication Errors as Config Exceptions                                                                          |
 | 1.2.11  | 2023-01-18 | [#21144](https://github.com/airbytehq/airbyte/pull/21144) | Added explicit error message if sync fails due to a config issue                                                         |
 | 1.2.9   | 2022-12-14 | [#20501](https://github.com/airbytehq/airbyte/pull/20501) | Report GCS staging failures that occur during connection check                                                           |
 | 1.2.8   | 2022-11-22 | [#19489](https://github.com/airbytehq/airbyte/pull/19489) | Added non-billable projects handle to check connection stage                                                             |