enrich-kinesis: Recover from losing lease to a new worker (close #649)

snowplow · Jul 1, 2022 · 88c8181 · 88c8181
1 parent d41b6cf
commit 88c8181
Showing 1 changed file with 12 additions and 1 deletion.
diff --git a/...les/kinesis/src/main/scala/com/snowplowanalytics/snowplow/enrich/kinesis/KinesisRun.scala b/...les/kinesis/src/main/scala/com/snowplowanalytics/snowplow/enrich/kinesis/KinesisRun.scala
@@ -22,6 +22,8 @@ import scala.concurrent.ExecutionContext
 
 import fs2.aws.kinesis.CommittableRecord
 
+import software.amazon.kinesis.exceptions.ShutdownException
+
 import com.snowplowanalytics.snowplow.enrich.common.fs2.Run
 import com.snowplowanalytics.snowplow.enrich.common.fs2.Telemetry
 
@@ -73,5 +75,14 @@ object KinesisRun {
               biggest
           } :: acc
       }
-      .parTraverse_(record => Sync[F].delay(record.checkpointer.checkpoint))
+      .parTraverse_ { record =>
+        Sync[F]
+          .delay(record.checkpointer.checkpoint)
+          .recover {
+            // The ShardRecordProcessor instance has been shutdown. This just means another KCL
+            // worker has stolen our lease. It is expected during autoscaling of instances, and is
+            // safe to ignore.
+            case _: ShutdownException => ()
+          }
+      }
 }