Add kafka sink healthcheck

snowplow · Oct 11, 2023 · 14c8e13 · 14c8e13
1 parent c578a77
commit 14c8e13
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/...rc/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KafkaSink.scala b/...rc/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/KafkaSink.scala
@@ -34,7 +34,9 @@ class KafkaSink[F[_]: Sync](
 
   private lazy val log = LoggerFactory.getLogger(getClass())
 
-  override def isHealthy: F[Boolean] = Sync[F].pure(true)
+
+  @volatile private var kafkaHealthy: Boolean = false
+  override def isHealthy: F[Boolean] = Sync[F].pure(kafkaHealthy)
 
   /**
     * Store raw events to the topic
@@ -49,7 +51,12 @@ class KafkaSink[F[_]: Sync](
         new ProducerRecord(topicName, key, event),
         new Callback {
           override def onCompletion(metadata: RecordMetadata, e: Exception): Unit =
-            if (e != null) log.error(s"Sending event failed: ${e.getMessage}")
+            if (e != null) {
+              kafkaHealthy = false
+              log.error(s"Sending event failed: ${e.getMessage}")
+            } else {
+              kafkaHealthy = true
+            }
         }
       )
     }