fix: handle fatal producer errors

osskit · Jan 21, 2024 · f5db629 · f5db629
1 parent 4665883
commit f5db629
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 1 deletion.
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
     "name": "dafka-producer",
-    "version": "7.2.3",
+    "version": "7.2.4",
     "description": "Dockerized kafka producer",
     "private": true,
     "repository": {

diff --git a/src/main/scala/adapters/kafka/ProducerImpl.scala b/src/main/scala/adapters/kafka/ProducerImpl.scala
@@ -42,6 +42,14 @@ class ProducerImpl(
         }).map(seq => seq.asJava).orNull
       )
     )).map(_ => ())
+      .handleErrorWith(e => {
+        if (e.getMessage =="Cannot execute transactional method because we are in an error state") {
+          logger.error(e)("Unrecoverable Kafka error, aborting")
+          System.exit(-5)
+        }
+
+        return IO.raiseError(e)
+      })
 
   override def healthy(): IO[Boolean] = {
     readinessTopic.map { topic =>