Fix slow rebalanceSafeCommits behavior (#1358)

After consumer 1 is shutdown (using stopConsumption), rebalances happen
and partitions from consumer 2 are assigned. These streams are never
started, so the finalizer completing completedPromise is never called.
Waiting for these to complete takes 3 minutes (default
maxRebalanceDuration).

In case that streams were assigned and no record was ever put in their
queues, there's no use in waiting for the stream to complete by
committing some offset.

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

@@ -65,6 +65,8 @@ final class PartitionStreamControl private (
 
   def queueSize: UIO[Int] = queueInfoRef.get.map(_.size)
 
+  def lastPulledOffset: UIO[Option[Offset]] = queueInfoRef.get.map(_.lastPulledOffset)
+
   /**
    * @return
    *   the number of polls there are records idling in the queue. It is increased on every poll (when the queue is

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

@@ -130,9 +130,10 @@ private[consumer] final class Runloop private (
           streamResults <-
             ZIO.foreach(streamsToEnd) { stream =>
               for {
-                isDone    <- stream.completedPromise.isDone
-                endOffset <- if (isDone) stream.completedPromise.await else ZIO.none
-              } yield (isDone, endOffset)
+                isDone           <- stream.completedPromise.isDone
+                lastPulledOffset <- stream.lastPulledOffset
+                endOffset        <- if (isDone) stream.completedPromise.await else ZIO.none
+              } yield (isDone || lastPulledOffset.isEmpty, endOffset)
             }
           committedOffsets <- committedOffsetsRef.get
         } yield {