Address review feedback

snowplow · Aug 2, 2023 · e0580d2 · e0580d2
1 parent 8c650cd
commit e0580d2
Showing 1 changed file with 16 additions and 4 deletions.
diff --git a/.../main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisSink.scala b/.../main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KinesisSink.scala
@@ -221,8 +221,14 @@ class KinesisSink private (
           )
           // If Kinesis was already unhealthy, the background check is already running.
           // It can happen when the collector switches back and forth between Kinesis and SQS.
-          if (kinesisHealthy) checkKinesisHealth()
-          kinesisHealthy = false
+          if (kinesisHealthy) {
+            this.synchronized {
+              if (kinesisHealthy) {
+                kinesisHealthy = false
+                checkKinesisHealth()
+              }
+            }
+          }
           writeBatchToSqsWithRetries(failedRecords, sqs, minBackoff, maxRetries)
         case None =>
           log.error(s"No SQS buffer defined for stream $streamName. Retrying to send the events to Kinesis")
@@ -240,8 +246,14 @@ class KinesisSink private (
     } else {
       // If SQS was already unhealthy, the background check is already running.
       // It can happen when the collector switches back and forth between Kinesis and SQS.
-      if (sqsHealthy) checkSqsHealth()
-      sqsHealthy = false
+      if (sqsHealthy) {
+        this.synchronized {
+          if (sqsHealthy) {
+            sqsHealthy = false
+            checkSqsHealth()
+          }
+        }
+      }
       log.error(
         s"Maximum number of retries reached for SQS buffer ${sqs.sqsBufferName} for ${failedRecords.size} records. Retrying in Kinesis"
       )