Loader to start healthy if no privilege to create tables but table already exists (close #29)

Author: istreeter

modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/processing/Processing.scala

@@ -37,6 +37,7 @@ object Processing {
   def stream[F[_]: Async](env: Environment[F]): Stream[F, Nothing] = {
     val eventProcessingConfig = EventProcessingConfig(EventProcessingConfig.NoWindowing)
     Stream.eval(env.tableManager.initializeEventsTable()) *>
+      Stream.eval(env.channel.opened.use_) *>
       env.source.stream(eventProcessingConfig, eventProcessor(env))
   }
 

modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/processing/SnowflakeRetrying.scala

@@ -55,7 +55,10 @@ object SnowflakeRetrying {
       // Authentication failure, i.e. user unrecognized or bad private key
       true.pure[F]
     case sql: java.sql.SQLException if sql.getErrorCode === 2003 =>
-      // Object does not exist or not authorized
+      // Object does not exist or not authorized to view it
+      true.pure[F]
+    case sql: java.sql.SQLException if sql.getErrorCode === 3001 =>
+      // Insufficient privileges
       true.pure[F]
     case _ =>
       false.pure[F]

modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/processing/TableManager.scala

@@ -59,6 +59,10 @@ object TableManager {
               Logger[ConnectionIO].info(s"Creating table $tableName if it does not already exist...") *>
                 sqlCreateTable(tableName).update.run.void
             }
+            .recoverWith {
+              case sql: java.sql.SQLException if sql.getErrorCode === 3001 =>
+                Logger[F].info(s"Access denied when trying to create table. Will ignore error and assume table already exists.")
+            }
         }
 
         def executeAddColumnsQuery(columns: List[String]): F[Unit] =

modules/core/src/test/scala/com.snowplowanalytics.snowplow.snowflake/processing/ProcessingSpec.scala

@@ -67,6 +67,7 @@ class ProcessingSpec extends Specification with CatsEffect {
       } yield state should beEqualTo(
         Vector(
           Action.InitEventsTable,
+          Action.OpenedChannel,
           Action.SentToBad(6),
           Action.AddedGoodCountMetric(0),
           Action.AddedBadCountMetric(6),
@@ -240,9 +241,9 @@ class ProcessingSpec extends Specification with CatsEffect {
       } yield state should beEqualTo(
         Vector(
           Action.InitEventsTable,
+          Action.OpenedChannel,
           Action.SetLatencyMetric(42123),
           Action.SetLatencyMetric(42123),
-          Action.OpenedChannel,
           Action.WroteRowsToSnowflake(4),
           Action.AddedGoodCountMetric(4),
           Action.AddedBadCountMetric(0),