Reduces noise from LibrdKafka as per #3064

BrighterCommand · May 7, 2024 · 8c8d249 · 8c8d249
1 parent 476e3ad
commit 8c8d249
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 4 deletions.
diff --git a/src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessageConsumer.cs b/src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessageConsumer.cs
@@ -196,8 +196,12 @@ public KafkaMessageConsumer(
                 })
                 .SetErrorHandler((consumer, error) =>
                 {
-                    s_logger.LogError("Code: {ErrorCode}, Reason: {ErrorMessage}, Fatal: {FatalError}", error.Code,
-                        error.Reason, error.IsFatal);
+                    var hasFatalError = error.IsFatal;
+                    
+                    if (hasFatalError ) 
+                        s_logger.LogError("Code: {ErrorCode}, Reason: {ErrorMessage}, Fatal: {FatalError}", error.Code, error.Reason, true);
+                    else
+                        s_logger.LogWarning("Code: {ErrorCode}, Reason: {ErrorMessage}, Fatal: {FatalError}", error.Code, error.Reason, false);
                 })
                 .Build();
 

diff --git a/src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessageProducer.cs b/src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessageProducer.cs
@@ -132,9 +132,13 @@ public void Init()
             _producer = new ProducerBuilder<string, byte[]>(_producerConfig)
                 .SetErrorHandler((_, error) =>
                 {
-                    s_logger.LogError("Code: {ErrorCode}, Reason: {ErrorMessage}, Fatal: {FatalError}", error.Code, error.Reason,
-                        error.IsFatal);
                     _hasFatalProducerError = error.IsFatal;
+                    
+                    if (_hasFatalProducerError) 
+                        s_logger.LogError("Code: {ErrorCode}, Reason: {ErrorMessage}, Fatal: {FatalError}", error.Code, error.Reason, true);
+                    else
+                        s_logger.LogWarning("Code: {ErrorCode}, Reason: {ErrorMessage}, Fatal: {FatalError}", error.Code, error.Reason, false);
+                    
                 })
                 .Build();
             _publisher = new KafkaMessagePublisher(_producer, _headerBuilder);