Pause a partition when its stream is ended

During a graceful shutdown the partition stream is ended. The runloop however, is still happily fetching records for that stream. These are then put in the stream's queue even though they will never be read. This is taking away network and CPU at a moment where the application should focus on doing a clean shutdown quickly. It could even cause an OOM for application that are tuned for the case where processing happens almost immediately.

With this change, when a stream is ended:
- the partition is paused,
- already fetched new records are not added to the stream's queue.

Also: less logging at start of a poll.

zio-kafka/src/main/scala/zio/kafka/consumer/internal/PartitionStreamControl.scala

@@ -52,16 +52,20 @@ final class PartitionStreamControl private (
 
   /** Offer new data for the stream to process. Should be called on every poll, also when `data.isEmpty` */
   private[internal] def offerRecords(data: Chunk[ByteArrayCommittableRecord]): ZIO[Any, Nothing, Unit] =
-    if (data.isEmpty) {
-      queueInfoRef.update(_.withEmptyPoll)
-    } else {
-      for {
-        now <- Clock.nanoTime
-        newPullDeadline = now + maxPollIntervalNanos
-        _ <- queueInfoRef.update(_.withOffer(newPullDeadline, data.size))
-        _ <- dataQueue.offer(Take.chunk(data))
-      } yield ()
-    }
+    ZIO
+      .whenZIO(isRunning) {
+        if (data.isEmpty) {
+          queueInfoRef.update(_.withEmptyPoll)
+        } else {
+          for {
+            now <- Clock.nanoTime
+            newPullDeadline = now + maxPollIntervalNanos
+            _ <- queueInfoRef.update(_.withOffer(newPullDeadline, data.size))
+            _ <- dataQueue.offer(Take.chunk(data))
+          } yield ()
+        }
+      }
+      .unit
 
   def queueSize: UIO[Int] = queueInfoRef.get.map(_.size)
 

zio-kafka/src/main/scala/zio/kafka/consumer/internal/Runloop.scala

@@ -460,11 +460,12 @@ private[consumer] final class Runloop private (
 
   private def handlePoll(state: State): Task[State] = {
     for {
-      partitionsToFetch <- settings.fetchStrategy.selectPartitionsToFetch(state.assignedStreams)
+      runningStreamsBeforePoll <- ZIO.filter(state.assignedStreams)(_.isRunning)
+      partitionsToFetch        <- settings.fetchStrategy.selectPartitionsToFetch(runningStreamsBeforePoll)
       _ <- ZIO.logDebug(
              s"Starting poll with ${state.pendingRequests.size} pending requests and" +
                s" ${state.pendingCommits.size} pending commits," +
-               s" resuming $partitionsToFetch partitions"
+               s" resuming ${partitionsToFetch.size} out of ${state.assignedStreams.size} partitions"
            )
       _ <- currentStateRef.set(state)
       pollResult <-