fix: log Kafka errors according to severity

Farfetch · Oct 7, 2020 · 7300d3e · 7300d3e
1 parent 185e7b3
commit 7300d3e
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 5 deletions.
diff --git a/src/KafkaFlow/Consumers/KafkaConsumer.cs b/src/KafkaFlow/Consumers/KafkaConsumer.cs
@@ -41,7 +41,17 @@ public KafkaConsumer(
             this.consumerBuilder
                 .SetPartitionsAssignedHandler((consumer, partitions) => this.OnPartitionAssigned(consumer, partitions))
                 .SetPartitionsRevokedHandler((consumer, partitions) => this.OnPartitionRevoked(partitions))
-                .SetErrorHandler((p, error) => this.logHandler.Error("Kafka Consumer Error", null, new { Error = error }));
+                .SetErrorHandler((p, error) =>
+                {
+                    if (error.IsFatal)
+                    {
+                        this.logHandler.Error("Kafka Consumer Fatal Error", null, new { Error = error });
+                    }
+                    else
+                    {
+                        this.logHandler.Warning("Kafka Consumer Error", new { Error = error });
+                    }
+                });
         }
 
         private void OnPartitionRevoked(IReadOnlyCollection<TopicPartitionOffset> topicPartitions)

diff --git a/src/KafkaFlow/Producers/MessageProducer.cs b/src/KafkaFlow/Producers/MessageProducer.cs
@@ -165,10 +165,7 @@ private IProducer<byte[], byte[]> EnsureProducer()
                             {
                                 this.dependencyResolverScope.Resolver
                                     .Resolve<ILogHandler>()
-                                    .Error(
-                                        "Kafka Producer Error",
-                                        new KafkaException(error),
-                                        new { Error = error });
+                                    .Warning("Kafka Producer Error", new { Error = error });
                             }
                         })
                     .Build();