Merge pull request #584 from fd4s/single-thread-for-consumer

Use single thread for KafkaConsumer
fd4s · Apr 8, 2021 · 4a841de · 4a841de
2 parents 38a94a2 + c0757cd
commit 4a841de
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 14 deletions.
diff --git a/modules/core/src/main/scala/fs2/kafka/internal/Blocking.scala b/modules/core/src/main/scala/fs2/kafka/internal/Blocking.scala
@@ -6,7 +6,11 @@
 
 package fs2.kafka.internal
 
-import cats.effect.Sync
+import cats.effect.{Async, Resource, Sync}
+
+import java.util.concurrent.Executors
+import scala.concurrent.ExecutionContext
+import cats.effect.syntax.async._
 
 private[kafka] trait Blocking[F[_]] {
   def apply[A](a: => A): F[A]
@@ -16,4 +20,30 @@ private[kafka] object Blocking {
   def apply[F[_]: Sync]: Blocking[F] = new Blocking[F] {
     override def apply[A](a: => A): F[A] = Sync[F].blocking(a)
   }
+
+  def singleThreaded[F[_]](name: String)(implicit F: Async[F]): Resource[F, Blocking[F]] =
+    Resource {
+      F.delay {
+        val executor =
+          Executors.newSingleThreadExecutor(
+            (runnable: Runnable) => {
+              val thread = new Thread(runnable)
+              thread.setName(s"$name-${thread.getId}")
+              thread.setDaemon(true)
+              thread
+            }
+          )
+
+        val ec = ExecutionContext.fromExecutor(executor)
+
+        val blocking: Blocking[F] = new Blocking[F] {
+          def apply[A](a: => A): F[A] = F.delay(a).evalOn(ec)
+        }
+
+        val shutdown =
+          F.delay(executor.shutdown())
+
+        (blocking, shutdown)
+      }
+    }
 }
diff --git a/modules/core/src/main/scala/fs2/kafka/internal/WithConsumer.scala b/modules/core/src/main/scala/fs2/kafka/internal/WithConsumer.scala
@@ -22,21 +22,23 @@ private[kafka] sealed abstract class WithConsumer[F[_]] {
 }
 
 private[kafka] object WithConsumer {
-  def apply[F[_], K, V](
+  def apply[F[_]: Async, K, V](
     mk: MkConsumer[F],
     settings: ConsumerSettings[F, K, V]
-  )(
-    implicit F: Async[F]
   ): Resource[F, WithConsumer[F]] =
-    Resource.make {
-      (mk(settings), Semaphore[F](1L))
-        .mapN { (consumer, semaphore) =>
-          new WithConsumer[F] {
-            override def apply[A](f: (KafkaByteConsumer, Blocking[F]) => F[A]): F[A] =
-              semaphore.permit.use { _ =>
-                f(consumer, Blocking[F])
-              }
+    Blocking.singleThreaded[F]("fs2-kafka-consumer").flatMap { b =>
+      Resource.make {
+        (mk(settings), Semaphore[F](1L))
+          .mapN { (consumer, semaphore) =>
+            new WithConsumer[F] {
+              override def apply[A](f: (KafkaByteConsumer, Blocking[F]) => F[A]): F[A] =
+                semaphore.permit.use { _ =>
+                  f(consumer, b)
+                }
+            }
           }
-        }
-    }(_.blocking { _.close(settings.closeTimeout.asJava) })
+      }(_.blocking {
+        _.close(settings.closeTimeout.asJava)
+      })
+    }
 }