Update kafka-clients to 3.6.0 in series/3.x (#1263)

* Update kafka-clients to 3.6.0 in series/3.x * `InvalidProducerEpochException` fixed to `ProducerFencedException` --------- Co-authored-by: a.artigao <[email protected]>
fd4s · Oct 21, 2023 · 4f51f27 · 4f51f27
1 parent 9572ca0
commit 4f51f27
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 6 deletions.
diff --git a/build.sbt b/build.sbt
@@ -6,7 +6,7 @@ val confluentVersion = "7.5.1"
 
 val fs2Version = "3.9.2"
 
-val kafkaVersion = "3.5.1"
+val kafkaVersion = "3.6.0"
 
 val testcontainersScalaVersion = "0.41.0"
 

diff --git a/modules/core/src/test/scala/fs2/kafka/TransactionalKafkaProducerSpec.scala b/modules/core/src/test/scala/fs2/kafka/TransactionalKafkaProducerSpec.scala
@@ -17,7 +17,7 @@ import fs2.kafka.internal.converters.collection._
 import fs2.kafka.producer.MkProducer
 import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerGroupMetadata, OffsetAndMetadata}
 import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.errors.InvalidProducerEpochException
+import org.apache.kafka.common.errors.ProducerFencedException
 import org.apache.kafka.common.serialization.ByteArraySerializer
 import org.scalatest.EitherValues
 import scala.concurrent.duration._
@@ -418,9 +418,6 @@ class TransactionalKafkaProducerSpec extends BaseKafkaSpec with EitherValues {
   }
 }
 
-// TODO: after switching from ForEachTestContainer to ForAllTestContainer, this fails
-// if run with a shared container with the following error:
-// org.apache.kafka.common.errors.ProducerFencedException: There is a newer producer with the same transactionalId which fences the current one. was not an instance of org.apache.kafka.common.errors.InvalidProducerEpochException, but an instance of org.apache.kafka.common.errors.ProducerFencedException
 class TransactionalKafkaProducerTimeoutSpec extends BaseKafkaSpec with EitherValues {
   it("should use user-specified transaction timeouts") {
     withTopic { topic =>
@@ -466,7 +463,7 @@ class TransactionalKafkaProducerTimeoutSpec extends BaseKafkaSpec with EitherVal
           result <- Stream.eval(producer.produce(records).attempt)
         } yield result).compile.lastOrError.unsafeRunSync()
 
-      produced.left.value shouldBe an[InvalidProducerEpochException]
+      produced.left.value shouldBe an[ProducerFencedException]
 
       val consumedOrError = {
         Either.catchNonFatal(