feat: add librdkafka log handler

Author: Alexander Shuvalov

src/KafkaFlow/Clusters/ClusterManager.cs

@@ -7,6 +7,7 @@
 using Confluent.Kafka.Admin;
 using KafkaFlow.Authentication;
 using KafkaFlow.Configuration;
+using KafkaFlow.Extensions;
 
 namespace KafkaFlow.Clusters;
 
@@ -33,7 +34,8 @@ public ClusterManager(ILogHandler logHandler, ClusterConfiguration configuration
 
                 config.ReadSecurityInformationFrom(configuration);
 
-                var adminClientBuilder = new AdminClientBuilder(config);
+                var adminClientBuilder = new AdminClientBuilder(config)
+                    .SetLogHandler((_, log) => _logHandler.Log(log));
 
                 var security = configuration.GetSecurityInformation();
 

src/KafkaFlow/Consumers/Consumer.cs

@@ -253,7 +253,8 @@ private void EnsureConsumer()
             .SetPartitionsAssignedHandler(FirePartitionsAssignedHandlers)
             .SetPartitionsRevokedHandler(FirePartitionRevokedHandlers)
             .SetErrorHandler((consumer, error) => _errorsHandlers.ForEach(x => x(consumer, error)))
-            .SetStatisticsHandler((consumer, statistics) => _statisticsHandlers.ForEach(x => x(consumer, statistics)));
+            .SetStatisticsHandler((consumer, statistics) => _statisticsHandlers.ForEach(x => x(consumer, statistics)))
+            .SetLogHandler((_, log) => _logHandler.Log(log));
 
         var security = this.Configuration.ClusterConfiguration.GetSecurityInformation();
 

src/KafkaFlow/Extensions/LogHandlerExtensions.cs

@@ -0,0 +1,41 @@
+using System;
+using Confluent.Kafka;
+
+namespace KafkaFlow.Extensions;
+
+internal static class LogHandlerExtensions
+{
+    public static void Log(this ILogHandler logHandler, LogMessage logMessage)
+    {
+        var message = "librdkafka log";
+        var data = new
+        {
+            Level = logMessage.Level.ToString(),
+            logMessage.Name,
+            logMessage.Facility,
+            logMessage.Message,
+        };
+
+        switch (logMessage.Level)
+        {
+            case SyslogLevel.Emergency:
+            case SyslogLevel.Alert:
+            case SyslogLevel.Critical:
+            case SyslogLevel.Error:
+                logHandler.Error(message, null, data);
+                break;
+            case SyslogLevel.Warning:
+                logHandler.Warning(message, null, data);
+                break;
+            case SyslogLevel.Notice:
+            case SyslogLevel.Info:
+                logHandler.Info(message, data);
+                break;
+            case SyslogLevel.Debug:
+                logHandler.Verbose(message, data);
+                break;
+            default:
+                throw new ArgumentOutOfRangeException(nameof(logMessage.Level), logMessage.Level, null);
+        }
+    }
+}

src/KafkaFlow/Producers/MessageProducer.cs

@@ -4,6 +4,7 @@
 using Confluent.Kafka;
 using KafkaFlow.Authentication;
 using KafkaFlow.Configuration;
+using KafkaFlow.Extensions;
 
 namespace KafkaFlow.Producers;
 
@@ -280,7 +281,8 @@ private IProducer<byte[], byte[]> EnsureProducer()
                                 });
                             }
                         }
-                    });
+                    })
+                .SetLogHandler((_, log) => _logHandler.Log(log));
 
             var security = _configuration.Cluster.GetSecurityInformation();