Handle responses where all records were invalid (#392)

Co-authored-by: Eric Meisel <[email protected]>
etspaceman · Oct 11, 2024 · f632c2e · f632c2e
1 parent 0957fec
commit f632c2e
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 2 deletions.
diff --git a/shared/src/main/scala/kinesis4cats/producer/Producer.scala b/shared/src/main/scala/kinesis4cats/producer/Producer.scala
@@ -182,8 +182,7 @@ abstract class Producer[F[_], PutReq, PutRes] private[kinesis4cats] (
       )
       finalRes <- retryingOnFailuresAndAllErrors(
         config.retryPolicy,
-        (x: Producer.Result[PutRes]) =>
-          F.pure(x.isSuccessful || (x.isPartiallySuccessful && !x.hasFailed)),
+        (x: Producer.Result[PutRes]) => F.pure(!x.hasFailed),
         (x: Producer.Result[PutRes], details: RetryDetails) =>
           for {
             failed <- F.fromOption(

diff --git a/shared/src/test/scala/kinesis4cats/producer/ProducerSpec.scala b/shared/src/test/scala/kinesis4cats/producer/ProducerSpec.scala
@@ -142,6 +142,28 @@ class ProducerSpec extends munit.CatsEffectSuite {
       assert(res === expected, s"res: $res\nexp: $expected")
     }
   }
+
+  fixture(false).test("It should not retry when all Results are invalid") {
+    producer =>
+      val tooSmallPartitionKey = ""
+      val record = Record(Array.fill(50)(1), tooSmallPartitionKey)
+
+      val data = NonEmptyList.of(record, record, record)
+
+      val expected: Producer.Result[MockPutResponse] = Producer.Result(
+        successful = Nil,
+        invalid = List(
+          Producer.InvalidRecord.InvalidPartitionKey(tooSmallPartitionKey),
+          Producer.InvalidRecord.InvalidPartitionKey(tooSmallPartitionKey),
+          Producer.InvalidRecord.InvalidPartitionKey(tooSmallPartitionKey)
+        ),
+        failed = Nil
+      )
+
+      producer.put(data).map { res =>
+        assert(res === expected, s"res: $res\nexp: $expected")
+      }
+  }
 }
 
 class MockProducer(