some producer ops using the new send operation with batches

core/src/main/scala/com/banno/kafka/producer/ProducerOps.scala

@@ -17,6 +17,7 @@
 package com.banno.kafka.producer
 
 import cats.*
+import cats.data.NonEmptyList
 import cats.syntax.all.*
 import fs2.*
 import org.apache.kafka.common.*
@@ -40,12 +41,99 @@ case class ProducerOps[F[_], K, V](producer: ProducerApi[F, K, V]) {
   )(implicit F: Applicative[F]): F[G[RecordMetadata]] =
     records.traverse(producer.sendAsync)
 
+  /** Sends all of the records to the producer (synchronously), so the producer
+    * may batch them. After all records are sent, asynchronously waits for all
+    * acks. Returns the write metadatas, in order. This is the only batch write
+    * operation that allows the producer to perform its own batching, while
+    * semantically blocking until all writes have succeeded. It maximizes
+    * concurrency and producer batching, and also simplicity of usage. Fails if
+    * any individual send or ack fails.
+    */
+  def sendBatch[G[_]: Traverse](
+      records: G[ProducerRecord[K, V]]
+  )(implicit F: Monad[F]): F[G[RecordMetadata]] = {
+    val sends: G[F[F[RecordMetadata]]] = records.map(producer.send)
+    for {
+      acks <- sends.sequence
+      rms <- acks.sequence
+    } yield rms
+  }
+
+  def sendBatchNonEmpty[G[_]: NonEmptyTraverse](
+      records: G[ProducerRecord[K, V]]
+  )(implicit F: FlatMap[F]): F[G[RecordMetadata]] = {
+    val sends: G[F[F[RecordMetadata]]] = records.map(producer.send)
+    for {
+      acks <- sends.nonEmptySequence
+      rms <- acks.nonEmptySequence
+    } yield rms
+  }
+
+  def sendBatchWithCallbacks[G[_]: Traverse](
+      records: G[ProducerRecord[K, V]],
+      onSend: ProducerRecord[K, V] => F[Unit],
+  )(implicit F: Monad[F]): F[G[RecordMetadata]] = {
+    val sends: G[F[F[RecordMetadata]]] =
+      records.map(r => producer.send(r) <* onSend(r))
+    for {
+      acks <- sends.sequence
+      rms <- acks.sequence
+    } yield rms
+  }
+
+  def sendBatchWithCallbacks[G[_]: NonEmptyTraverse](
+      records: G[ProducerRecord[K, V]],
+      onSend: ProducerRecord[K, V] => F[Unit],
+  )(implicit F: FlatMap[F]): F[G[RecordMetadata]] = {
+    val sends: G[F[F[RecordMetadata]]] =
+      records.map(r => producer.send(r) <* onSend(r))
+    for {
+      acks <- sends.nonEmptySequence
+      rms <- acks.nonEmptySequence
+    } yield rms
+  }
+
   def pipeSync: Pipe[F, ProducerRecord[K, V], RecordMetadata] =
     _.evalMap(producer.sendSync)
 
   def pipeAsync: Pipe[F, ProducerRecord[K, V], RecordMetadata] =
     _.evalMap(producer.sendAsync)
 
+  def pipeSendBatch[G[_]: Traverse](implicit
+      F: Monad[F]
+  ): Pipe[F, G[ProducerRecord[K, V]], G[RecordMetadata]] =
+    _.evalMap(sendBatch[G])
+
+  def pipeSendBatchNonEmpty[G[_]: NonEmptyTraverse](implicit
+      F: FlatMap[F]
+  ): Pipe[F, G[ProducerRecord[K, V]], G[RecordMetadata]] =
+    _.evalMap(sendBatchNonEmpty[G])
+
+  /** Uses the stream's chunks as batches of records to send to the producer. */
+  def pipeSendBatchChunks(implicit
+      F: Monad[F]
+  ): Pipe[F, ProducerRecord[K, V], RecordMetadata] =
+    s =>
+      pipeSendBatch[Chunk](Traverse[Chunk], F)(s.chunks).flatMap(Stream.chunk)
+
+  def pipeSendBatchChunksNonEmpty(implicit
+      F: FlatMap[F]
+  ): Pipe[F, ProducerRecord[K, V], RecordMetadata] =
+    s =>
+      pipeSendBatchNonEmpty[NonEmptyList](NonEmptyTraverse[NonEmptyList], F)(
+        s.chunks.map(_.toNel).unNone
+      ).flatMap(nel => Stream.emits(nel.toList))
+
+  /** Calls chunkN on the input stream, to create chunks of size `n`, and sends
+    * those chunks as batches to the producer.
+    */
+  def pipeSendBatchChunkN(n: Int, allowFewer: Boolean = true)(implicit
+      F: Monad[F]
+  ): Pipe[F, ProducerRecord[K, V], RecordMetadata] =
+    s =>
+      pipeSendBatch[Chunk](Traverse[Chunk], F)(s.chunkN(n, allowFewer))
+        .flatMap(Stream.chunk)
+
   def sink: Pipe[F, ProducerRecord[K, V], Unit] =
     _.evalMap(producer.sendAndForget)
 
@@ -55,6 +143,26 @@ case class ProducerOps[F[_], K, V](producer: ProducerApi[F, K, V]) {
   def sinkAsync: Pipe[F, ProducerRecord[K, V], Unit] =
     pipeAsync.apply(_).void
 
+  def sinkSendBatch[G[_]: Traverse](implicit
+      F: Monad[F]
+  ): Pipe[F, G[ProducerRecord[K, V]], Unit] =
+    pipeSendBatch.apply(_).void
+
+  def sinkSendBatch[G[_]: NonEmptyTraverse](implicit
+      F: FlatMap[F]
+  ): Pipe[F, G[ProducerRecord[K, V]], Unit] =
+    pipeSendBatchNonEmpty.apply(_).void
+
+  def sinkSendBatchChunks(implicit
+      F: Monad[F]
+  ): Pipe[F, ProducerRecord[K, V], Unit] =
+    pipeSendBatchChunks.apply(_).void
+
+  def sinkSendBatchChunkN(n: Int, allowFewer: Boolean = true)(implicit
+      F: Monad[F]
+  ): Pipe[F, ProducerRecord[K, V], Unit] =
+    pipeSendBatchChunkN(n, allowFewer).apply(_).void
+
   def transaction[G[_]: Foldable](
       records: G[ProducerRecord[K, V]]
   )(implicit F: MonadError[F, Throwable]): F[Unit] =