Add Producer.sendOffsetsToTransaction

build.sbt

@@ -55,6 +55,9 @@ ThisBuild / mimaBinaryIssueFilters ++= Seq(
 //  ProblemFilters.exclude[ReversedMissingMethodProblem](
 //    "com.evolutiongaming.skafka.consumer.Consumer.clientInstanceId"
 //  ),
+  ProblemFilters.exclude[ReversedMissingMethodProblem](
+    "com.evolutiongaming.skafka.producer.Producer.sendOffsetsToTransaction"
+  ),
 )
 ThisBuild / versionPolicyIgnored ++= Seq(
   // add libraries, that are known to be binary compatible, here, like:

skafka/src/main/scala/com/evolutiongaming/skafka/producer/Producer.scala

@@ -1,13 +1,16 @@
 package com.evolutiongaming.skafka
 package producer
 
+import cats.data.NonEmptyMap as Nem
 import cats.effect.{Async, Deferred, Resource, Sync}
 import cats.effect.implicits.*
 import cats.implicits.*
 import cats.{Applicative, Functor, Monad, MonadError, MonadThrow, ~>}
 import com.evolutiongaming.catshelper.CatsHelper.*
 import com.evolutiongaming.catshelper.{Log, MeasureDuration, ToTry}
 import com.evolutiongaming.skafka.Converters.*
+import com.evolutiongaming.skafka.consumer.ConsumerGroupMetadata
+import com.evolutiongaming.skafka.consumer.ConsumerConverters.*
 import com.evolutiongaming.skafka.producer.ProducerConverters.*
 import org.apache.kafka.clients.producer.{
   Callback,
@@ -33,6 +36,15 @@ trait Producer[F[_]] {
 
   def abortTransaction: F[Unit]
 
+  /** Sends consumed offsets to the consumer group coordinator as part of the current transaction, fencing zombie
+    * writers by the consumer group generation (KIP-447). The `consumerGroupMetadata` must come from the input
+    * consumer's `groupMetadata()`; offsets become committed only if the transaction commits.
+    */
+  def sendOffsetsToTransaction(
+    offsets: Nem[TopicPartition, OffsetAndMetadata],
+    consumerGroupMetadata: ConsumerGroupMetadata
+  ): F[Unit]
+
   /** @return
     *   Outer F[_] is about sending event (including batching if required), inner F[_] is about waiting and getting the
     *   result of the send operation.
@@ -70,6 +82,11 @@ object Producer {
 
       def abortTransaction: F[Unit] = empty
 
+      def sendOffsetsToTransaction(
+        offsets: Nem[TopicPartition, OffsetAndMetadata],
+        consumerGroupMetadata: ConsumerGroupMetadata
+      ): F[Unit] = empty
+
       def send[K, V](
         record: ProducerRecord[K, V]
       )(implicit toBytesK: ToBytes[F, K], toBytesV: ToBytes[F, V]): F[F[RecordMetadata]] = {
@@ -121,6 +138,13 @@ object Producer {
           blocking { producer.abortTransaction() }
         }
 
+        def sendOffsetsToTransaction(
+          offsets: Nem[TopicPartition, OffsetAndMetadata],
+          consumerGroupMetadata: ConsumerGroupMetadata
+        ): F[Unit] = {
+          blocking { producer.sendOffsetsToTransaction(asOffsetsAndMetadataJ(offsets), consumerGroupMetadata.asJava) }
+        }
+
         def send[K, V](
           record: ProducerRecord[K, V]
         )(implicit toBytesK: ToBytes[F, K], toBytesV: ToBytes[F, V]): F[F[RecordMetadata]] = {
@@ -250,6 +274,19 @@ object Producer {
         } yield r
       }
 
+      def sendOffsetsToTransaction(
+        offsets: Nem[TopicPartition, OffsetAndMetadata],
+        consumerGroupMetadata: ConsumerGroupMetadata
+      ): F[Unit] = {
+        for {
+          d <- MeasureDuration[F].start
+          r <- producer.sendOffsetsToTransaction(offsets, consumerGroupMetadata).attempt
+          d <- d
+          _ <- metrics.sendOffsetsToTransaction(d)
+          r <- r.liftTo[F]
+        } yield r
+      }
+
       def send[K, V](
         record: ProducerRecord[K, V]
       )(implicit toBytesK: ToBytes[F, K], toBytesV: ToBytes[F, V]): F[F[RecordMetadata]] = {
@@ -345,6 +382,11 @@ object Producer {
 
       def abortTransaction: G[Unit] = fg(self.abortTransaction)
 
+      def sendOffsetsToTransaction(
+        offsets: Nem[TopicPartition, OffsetAndMetadata],
+        consumerGroupMetadata: ConsumerGroupMetadata
+      ): G[Unit] = fg(self.sendOffsetsToTransaction(offsets, consumerGroupMetadata))
+
       def send[K, V](
         record: ProducerRecord[K, V]
       )(implicit toBytesK: ToBytes[G, K], toBytesV: ToBytes[G, V]): G[G[RecordMetadata]] = {

skafka/src/main/scala/com/evolutiongaming/skafka/producer/ProducerLogging.scala

@@ -1,9 +1,11 @@
 package com.evolutiongaming.skafka.producer
 
+import cats.data.NonEmptyMap as Nem
 import cats.implicits.*
 import cats.MonadThrow
 import com.evolutiongaming.catshelper.{Log, MeasureDuration}
-import com.evolutiongaming.skafka.{ClientMetric, PartitionInfo, ToBytes, Topic}
+import com.evolutiongaming.skafka.{ClientMetric, OffsetAndMetadata, PartitionInfo, ToBytes, Topic, TopicPartition}
+import com.evolutiongaming.skafka.consumer.ConsumerGroupMetadata
 import org.apache.kafka.common.Uuid
 import org.apache.kafka.common.errors.RecordTooLargeException
 
@@ -30,6 +32,11 @@ object ProducerLogging {
 
       def abortTransaction: F[Unit] = producer.abortTransaction
 
+      def sendOffsetsToTransaction(
+        offsets: Nem[TopicPartition, OffsetAndMetadata],
+        consumerGroupMetadata: ConsumerGroupMetadata
+      ): F[Unit] = producer.sendOffsetsToTransaction(offsets, consumerGroupMetadata)
+
       def send[K, V](
         record: ProducerRecord[K, V]
       )(implicit toBytesK: ToBytes[F, K], toBytesV: ToBytes[F, V]): F[F[RecordMetadata]] = {

skafka/src/test/scala/com/evolutiongaming/skafka/producer/ProducerSpec.scala

@@ -3,13 +3,16 @@ package com.evolutiongaming.skafka.producer
 import java.util
 import java.util.concurrent.{CompletableFuture, Future as FutureJ}
 import cats.arrow.FunctionK
+import cats.data.NonEmptyMap as Nem
 import cats.effect.IO
 import cats.implicits.*
 import com.evolutiongaming.catshelper.MeasureDuration
 import com.evolutiongaming.catshelper.CatsHelper.*
+import com.evolutiongaming.skafka.Converters.*
 import com.evolutiongaming.skafka.IOMatchers.*
+import com.evolutiongaming.skafka.consumer.ConsumerGroupMetadata
 import com.evolutiongaming.skafka.producer.ProducerConverters.*
-import com.evolutiongaming.skafka.{Bytes, Partition, PartitionInfo, TopicPartition}
+import com.evolutiongaming.skafka.{Bytes, OffsetAndMetadata, Partition, PartitionInfo, TopicPartition}
 import com.evolutiongaming.skafka.IOSuite.*
 import org.apache.kafka.clients.consumer.{
   ConsumerGroupMetadata as ConsumerGroupMetadataJ,
@@ -61,6 +64,15 @@ class ProducerSpec extends AnyWordSpec with Matchers {
       }
     }
 
+    "proxy sendOffsetsToTransaction" in new Scope {
+      val offsets       = Nem.of(topicPartition -> OffsetAndMetadata())
+      val groupMetadata = ConsumerGroupMetadata("group", 1, "member", None)
+      verify(producer.sendOffsetsToTransaction(offsets, groupMetadata)) { _ =>
+        sentOffsets.map(_.asScala.toMap) shouldEqual Some(Map(topicPartition.asJava -> OffsetAndMetadata().asJava))
+        sentGroupMetadata.map(_.groupId) shouldEqual Some("group")
+      }
+    }
+
     "proxy send" in new Scope {
       val record = ProducerRecord(topic = topic, value = "val", key = "key")
       producer.send(record).flatten should produce(metadata)
@@ -116,6 +128,11 @@ class ProducerSpec extends AnyWordSpec with Matchers {
       verify(empty.abortTransaction) { _ => }
     }
 
+    "sendOffsetsToTransaction" in new Scope {
+      val offsets = Nem.of(topicPartition -> OffsetAndMetadata())
+      verify(empty.sendOffsetsToTransaction(offsets, ConsumerGroupMetadata("group", 1, "member", None))) { _ => }
+    }
+
     "send" in {
       val record = ProducerRecord(topic = topic, value = "val", key = "key")
       empty.send(record).flatten should produce(metadata)
@@ -131,15 +148,15 @@ class ProducerSpec extends AnyWordSpec with Matchers {
   }
 
   private trait Scope {
-    var flushCalled                                               = false
-    var commitTransaction                                         = false
-    var beginTransaction                                          = false
-    var initTransactions                                          = false
-    var abortTransaction                                          = false
-    var partitionsFor                                             = ""
-    var sendOffsetsToTransaction                                  = ""
-    var sendOffsetsToTransaction1: Option[ConsumerGroupMetadataJ] = none[ConsumerGroupMetadataJ]
-    val completableFuture: CompletableFuture[RecordMetadataJ]     = CompletableFuture.completedFuture(metadata.asJava)
+    var flushCalled                                                        = false
+    var commitTransaction                                                  = false
+    var beginTransaction                                                   = false
+    var initTransactions                                                   = false
+    var abortTransaction                                                   = false
+    var partitionsFor                                                      = ""
+    var sentOffsets: Option[util.Map[TopicPartitionJ, OffsetAndMetadataJ]] = none
+    var sentGroupMetadata: Option[ConsumerGroupMetadataJ]                  = none
+    val completableFuture: CompletableFuture[RecordMetadataJ] = CompletableFuture.completedFuture(metadata.asJava)
 
     val jProducer: ProducerJ[Bytes, Bytes] = new ProducerJ[Bytes, Bytes] {
 
@@ -151,7 +168,8 @@ class ProducerSpec extends AnyWordSpec with Matchers {
         offsets: util.Map[TopicPartitionJ, OffsetAndMetadataJ],
         groupMetadata: ConsumerGroupMetadataJ
       ): Unit = {
-        Scope.this.sendOffsetsToTransaction1 = groupMetadata.some
+        Scope.this.sentOffsets       = offsets.some
+        Scope.this.sentGroupMetadata = groupMetadata.some
       }
 
       def flush(): Unit = flushCalled = true