Improve handling of pending commits (#1368)

Author: biochimia

modules/core/src/main/scala/fs2/kafka/internal/KafkaConsumerActor.scala

@@ -93,7 +93,7 @@ final private[kafka] class KafkaConsumerActor[F[_], K, V](
   private[this] def commit(request: Request.Commit[F]): F[Unit] =
     ref.flatModify { state =>
       val commitF = commitAsync(request.offsets, request.callback)
-      if (state.rebalancing) {
+      if (state.rebalancing || state.pendingCommits.nonEmpty) {
         val newState =
           state.withPendingCommit(commitF >> logging.log(CommittedPendingCommit(request)))
         (newState, logging.log(StoredPendingCommit(request, newState)))

modules/core/src/main/scala/fs2/kafka/internal/LogEntry.scala

@@ -145,7 +145,7 @@ private[kafka] object LogEntry {
 
     override def level: LogLevel = Debug
 
-    override def message: String = s"Committed pending commits [$pendingCommit]."
+    override def message: String = s"Committed pending commit [$pendingCommit]."
 
   }