Adding exception logging for consumer's batch processing loop

CHANGELOG.md

@@ -13,6 +13,9 @@ The `Unreleased` section name is replaced by the expected version of next releas
 ## [Unreleased]
 
 ### Added
+
+- Added a logging for exception caught in the batch processing loop. Any uncaught exception from the handler will be logged here as well.  
+
 ### Changed
 ### Removed
 ### Fixed

src/FsKafka/FsKafka.fs

@@ -462,18 +462,22 @@ module private ConsumerImpl =
             let buffer = Array.zeroCreate buf.maxBatchSize
             let nextBatch () =
                 let count = collection.FillBuffer(buffer, buf.maxBatchDelay)
-                if count <> 0 then log.Debug("Consuming {count}", count)
                 let batch = Array.init count (fun i -> buffer.[i])
                 Array.Clear(buffer, 0, count)
                 batch
 
             let rec loop () = async {
                 if not collection.IsCompleted then
+                    let batchWatch = System.Diagnostics.Stopwatch()
+                    let mutable batchLen, batchSize = 0, 0L
+                    use __ = Serilog.Context.LogContext.PushProperty("partition", Binding.partitionValue key.Partition)
                     try match nextBatch() with
                         | [||] -> ()
                         | batch ->
-                            use __ = Serilog.Context.LogContext.PushProperty("partition", Binding.partitionValue key.Partition)
-                            log.Debug("Dispatching {count} message(s) to handler", batch.Length)
+                            batchLen <- batch.Length
+                            batchSize <- batch |> Array.sumBy approximateMessageBytes
+                            batchWatch.Start()
+                            log.ForContext("batchSize", batchSize).Debug("Dispatching {batchLen} message(s) to handler", batchLen)
                             // run the handler function
                             do! handler batch
 
@@ -482,9 +486,10 @@ module private ConsumerImpl =
                             consumer.StoreOffset(lastItem)
 
                             // decrement in-flight message counter
-                            let batchSize = batch |> Array.sumBy approximateMessageBytes
                             counter.Delta(-batchSize)
                     with e ->
+                        log.ForContext("batchSize", batchSize).ForContext("batchLen", batchLen).ForContext("handlerDuration", batchWatch.Elapsed)
+                            .Information(e, "Exiting batch processing loop due to handler exception") 
                         tcs.TrySetException e |> ignore
                         cts.Cancel()
                     return! loop() }

src/FsKafka0/FsKafka.fs

@@ -480,21 +480,27 @@ module private ConsumerImpl =
 
             let rec loop () = async {
                 if not collection.IsCompleted then
+                    let batchWatch = System.Diagnostics.Stopwatch()
+                    let mutable batchLen, batchSize = 0, 0L
+                    use __ = Serilog.Context.LogContext.PushProperty("partition", Binding.partitionValue key.Partition)
                     try match nextBatch() with
                         | [||] -> ()
                         | batch ->
-                            use __ = Serilog.Context.LogContext.PushProperty("partition", Binding.partitionValue key.Partition)
-                            log.Debug("Dispatching {count} message(s) to handler", batch.Length)
+                            batchLen <- batch.Length
+                            batchSize <- batch |> Array.sumBy (fun m -> getMessageSize m)
+                            batchWatch.Start()
+                            log.ForContext("batchSize", batchSize).Debug("Dispatching {batchLen} message(s) to handler", batchLen)
                             // run the handler function
                             do! handler batch
 
                             // store completed offsets
                             storeOffset consumer (getBatchOffset batch)
 
                             // decrement in-flight message counter
-                            let batchSize = batch |> Array.sumBy (fun m -> getMessageSize m)
                             counter.Delta -batchSize
                     with e ->
+                        log.ForContext("batchSize", batchSize).ForContext("batchLen", batchLen).ForContext("handlerDuration", batchWatch.Elapsed)
+                            .Information(e, "Exiting batch processing loop due to handler exception")
                         tcs.TrySetException e |> ignore
                         cts.Cancel()
                     return! loop() }