[greyhound] parallel consumer - add disconnect on shutdown (#36660)

GitOrigin-RevId: 84bff3d0d5156c9cd3315423becf3811fe374776
wix · Sep 8, 2023 · 155a9cc · 155a9cc
1 parent 4de42ee
commit 155a9cc
Showing 1 changed file with 16 additions and 3 deletions.
diff --git a/core/src/main/scala/com/wixpress/dst/greyhound/core/consumer/EventLoop.scala b/core/src/main/scala/com/wixpress/dst/greyhound/core/consumer/EventLoop.scala
@@ -139,9 +139,10 @@ object EventLoop {
     for {
       _       <- report(StoppingEventLoop(clientId, group, consumerAttributes))
       _       <- running.set(ShuttingDown)
+      _       <- running.get.flatMap(currentState => report(EventLoopStateOnShutdown(clientId, group, currentState, consumerAttributes)))
       drained <-
       (joinFiberAndReport(group, clientId, consumerAttributes, fiber).interruptible *>
-        shutdownDispatcherAndReport(group, clientId, consumerAttributes, dispatcher))
+        shutdownDispatcherAndReport(group, clientId, consumerAttributes, dispatcher)).disconnect.interruptible
         .timeout(config.drainTimeout)
       _       <- ZIO.when(drained.isEmpty)(
                    report(DrainTimeoutExceeded(clientId, group, config.drainTimeout.toMillis, consumerAttributes)) *> fiber.interruptFork
@@ -202,8 +203,8 @@ object EventLoop {
           _       <- ZIO.when(records.isEmpty)(ZIO.sleep(50.millis))
         } yield true
 
-      case ShuttingDown => ZIO.succeed(false)
-      case Paused       => ZIO.sleep(100.millis).as(true)
+      case ShuttingDown => report(PollOnceFiberShuttingDown(clientId, group, consumer.config.consumerAttributes)) *> ZIO.succeed(false)
+      case Paused       => report(PollOnceFiberPaused(clientId, group, consumer.config.consumerAttributes)) *> ZIO.sleep(100.millis).as(true)
     }
 
   private def listener(
@@ -463,6 +464,18 @@ object EventLoopMetric {
 
   case class StoppingEventLoop(clientId: ClientId, group: Group, attributes: Map[String, String] = Map.empty) extends EventLoopMetric
 
+  case class EventLoopStateOnShutdown(
+    clientId: ClientId,
+    group: Group,
+    eventLoopState: EventLoopState,
+    attributes: Map[String, String] = Map.empty
+  ) extends EventLoopMetric
+
+  case class PollOnceFiberShuttingDown(clientId: ClientId, group: Group, attributes: Map[String, String] = Map.empty)
+      extends EventLoopMetric
+
+  case class PollOnceFiberPaused(clientId: ClientId, group: Group, attributes: Map[String, String] = Map.empty) extends EventLoopMetric
+
   case class StoppedEventLoop(clientId: ClientId, group: Group, attributes: Map[String, String] = Map.empty) extends EventLoopMetric
 
   case class CommittingOffsetsAndGaps(