PDP-1323 Handle initialization errors for GCP

Author: benjben

modules/azure/src/main/scala/com.snowplowanalytics.snowplow.lakes/AzureApp.scala

@@ -14,6 +14,8 @@ import java.net.UnknownHostException
 
 import scala.reflect._
 
+import cats.implicits._
+
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException
 
 import com.snowplowanalytics.snowplow.sources.kafka.{KafkaSource, KafkaSourceConfig}
@@ -44,13 +46,13 @@ object AzureApp extends LoaderApp[KafkaSourceConfig, KafkaSinkConfig](BuildInfo)
     case AuthenticationError(e) =>
       e
     // Wrong container name
-    case e: AbfsRestOperationException if e.getStatusCode == 404 =>
+    case e: AbfsRestOperationException if e.getStatusCode === 404 =>
       s"The specified filesystem does not exist (e.g. wrong container name)"
     // Service principal missing permissions for container (role assignement missing or wrong role)
-    case e: AbfsRestOperationException if e.getStatusCode == 403 =>
+    case e: AbfsRestOperationException if e.getStatusCode === 403 =>
       s"Missing permissions for the destination (needs \"Storage Blob Data Contributor\" assigned to the service principal for the container)"
     // Soft delete not disabled
-    case e: AbfsRestOperationException if e.getStatusCode == 409 =>
+    case e: AbfsRestOperationException if e.getStatusCode === 409 =>
       "Blob soft delete must be disabled on the storage account"
     case _: UnknownHostException =>
       "Wrong storage name"

modules/gcp/src/main/scala/com.snowplowanalytics.snowplow.lakes/GcpApp.scala

@@ -10,6 +10,10 @@
 
 package com.snowplowanalytics.snowplow.lakes
 
+import cats.implicits._
+
+import com.google.api.client.googleapis.json.GoogleJsonResponseException
+
 import com.snowplowanalytics.snowplow.sources.pubsub.{PubsubSource, PubsubSourceConfig}
 import com.snowplowanalytics.snowplow.sinks.pubsub.{PubsubSink, PubsubSinkConfig}
 
@@ -19,5 +23,15 @@ object GcpApp extends LoaderApp[PubsubSourceConfig, PubsubSinkConfig](BuildInfo)
 
   override def badSink: SinkProvider = PubsubSink.resource(_)
 
-  override def isDestinationSetupError: DestinationSetupErrorCheck = TableFormatSetupError.check
+  override def isDestinationSetupError: DestinationSetupErrorCheck = {
+    // Destination bucket doesn't exist
+    case e: GoogleJsonResponseException if e.getDetails.getCode === 404 =>
+      "The specified bucket does not exist"
+    // Permissions missing for Cloud Storage
+    case e: GoogleJsonResponseException if e.getDetails.getCode === 403 =>
+      e.getDetails.getMessage
+    // Exceptions common to the table format - Delta/Iceberg/Hudi
+    case TableFormatSetupError.check(t) =>
+      t
+  }
 }

packaging/delta-iceberg/src/main/scala/com.snowplowanalytics.snowplow.lakes/TableFormatSetupError.scala

@@ -12,6 +12,8 @@ package com.snowplowanalytics.snowplow.lakes
 
 import org.apache.iceberg.exceptions.{ForbiddenException => IcebergForbiddenException, NotFoundException => IcebergNotFoundException}
 
+import org.apache.spark.sql.delta.DeltaAnalysisException
+
 object TableFormatSetupError {
 
   // Check if given exception is specific to iceberg format
@@ -22,5 +24,7 @@ object TableFormatSetupError {
       case e: IcebergForbiddenException =>
         // No permission to create a table in Glue catalog
         e.getMessage
+      case e: DeltaAnalysisException if e.errorClass == Some("DELTA_CREATE_TABLE_WITH_NON_EMPTY_LOCATION") =>
+        "Destination not empty and not a Delta table"
     }
 }

project/Dependencies.scala

@@ -132,9 +132,9 @@ object Dependencies {
 
   val icebergDeltaRuntimeDependencies = Seq(
     iceberg,
-    delta                     % Runtime,
-    Spark.coreForIcebergDelta % Runtime,
-    Spark.sqlForIcebergDelta  % Runtime
+    delta,
+    Spark.coreForIcebergDelta,
+    Spark.sqlForIcebergDelta
   )
 
   val coreDependencies = Seq(