airbytehq · jbfbell · Jan 23, 2023 · Jan 5, 2023 · Jan 6, 2023 · Jan 12, 2023
diff --git a/...e-java-s3/src/main/java/io/airbyte/integrations/destination/s3/BlobStorageOperations.java b/...e-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.

diff --git a/...bases/base-java-s3/src/main/java/io/airbyte/integrations/destination/s3/S3BaseChecks.java b/...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);

diff --git a/...ase-java-s3/src/main/java/io/airbyte/integrations/destination/s3/S3StorageOperations.java b/...ase-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,17 @@ 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")));
+    }
   }
 
   /**

diff --git a/...ns/bases/base-java/src/main/java/io/airbyte/integrations/util/ConnectorExceptionUtil.java b/...ns/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());

@@ -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

@@ -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";

@@ -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

@@ -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