Add KafkaErrorEvent.Handled to prevent built-in logging [#83]

Directory.Build.props

@@ -5,7 +5,7 @@
     <Description>Silverback is a simple but feature-rich framework to build reactive/event-driven applications or microservices.</Description>
     <License>MIT</License>
     <Copyright>Copyright (c) 2020 Sergio Aquilini</Copyright>
-    <VersionSuffix>-rc.12</VersionSuffix>
+    <VersionSuffix>-rc.13</VersionSuffix>
     <BaseVersion>2.1.1$(VersionSuffix)</BaseVersion>
     <ProjectUrl>https://beagle1984.github.io/silverback/</ProjectUrl>
     <RepositoryUrl>https://github.com/BEagle1984/silverback/</RepositoryUrl>

docs/_docs/4-Kafka/403-kafka-events.md

@@ -13,7 +13,7 @@ Event | Description
 `KafkaPartitionsAssignedEvent` | The event fired when a new consumer group partition assignment has been received by a consumer. Corresponding to each of this events there will be a `KafkaPartitionsRevokedEvent`. This event is important if you need to manually set the start offset (see the sample code below in the samples section).
 `KafkaPartitionsRevokedEvent` | The event fired prior to a group partition assignment being revoked. Corresponding to each of this events there will be a `KafkaPartitionsAssignedEvent`.
 `KafkaOffsetsCommittedEvent` | The event fired to report the result of the offset commits.
-`KafkaErrorEvent` | The event fired when an error is reported by the `Confluent.Kafka.Consumer` (e.g. connection failures or all brokers down). Note that the system (either the Kafka client itself or Silverback) will try to automatically recover from all errors automatically, so these errors have to be considered purely informational.
+`KafkaErrorEvent` | The event fired when an error is reported by the `Confluent.Kafka.Consumer` (e.g. connection failures or all brokers down). Note that the system (either the Kafka client itself or Silverback) will try to automatically recover from all errors automatically, so these errors have to be considered purely informational. The subscriber can set the `Handled` property to prevent Silverback to log the error (useful if you are logging it already).
 `KafkaStatisticsEvent` | The event fired when statistics are received. Statistics are provided as a JSON formatted string as defined in the [librdkafka documentation](https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md). You can enable statistics and set the statistics interval using the `StatisticsIntervalMs` configuration parameter (disabled by default).
 
 

src/Silverback.Integration.Kafka/Messaging/Broker/KafkaEventsHandler.cs

@@ -9,6 +9,7 @@
 
 namespace Silverback.Messaging.Broker
 {
+    // TODO: Test
     internal class KafkaEventsHandler
     {
         private readonly IServiceProvider _serviceProvider;
@@ -96,15 +97,29 @@ public void SetConsumerEventsHandlers(
                     // Ignore errors if not consuming anymore
                     // (lidrdkafka randomly throws some "brokers are down"
                     // while disconnecting)
-                    if (!ownerConsumer.IsConsuming) return;
+                    if (!ownerConsumer.IsConsuming)
+                        return;
 
+                    var kafkaErrorEvent = new KafkaErrorEvent(error);
+
+                    try
+                    {
+                        CreateScopeAndPublishEvent(kafkaErrorEvent);
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError(ex, "Error in KafkaErrorEvent subscriber.");
+                    }
+
+                    if (kafkaErrorEvent.Handled)
+                        return;
+                        
                     _logger.Log(
                         error.IsFatal ? LogLevel.Critical : LogLevel.Error,
                         "Error in Kafka consumer: {error} (topic(s): {topics})",
                         error,
                         ownerConsumer.Endpoint.Names);
 
-                    CreateScopeAndPublishEvent(new KafkaErrorEvent(error));
                 })
                 .SetStatisticsHandler((_, statistics) =>
                 {

src/Silverback.Integration.Kafka/Messaging/Messages/KafkaErrorEvent.cs

@@ -1,6 +1,8 @@
 // Copyright (c) 2020 Sergio Aquilini
 // This code is licensed under MIT license (see LICENSE file for details)
 
+using Confluent.Kafka;
+
 namespace Silverback.Messaging.Messages
 {
     /// <summary>
@@ -11,7 +13,7 @@ namespace Silverback.Messaging.Messages
     /// </summary>
     public class KafkaErrorEvent : IKafkaEvent
     {
-        public KafkaErrorEvent(Confluent.Kafka.Error error)
+        public KafkaErrorEvent(Error error)
         {
             Error = error;
         }
@@ -20,6 +22,12 @@ public KafkaErrorEvent(Confluent.Kafka.Error error)
         ///     Gets an <see cref="Confluent.Kafka.Error" /> instance representing the error that occured when
         ///     interacting with the Kafka broker or the librdkafka library.
         /// </summary>
-        public Confluent.Kafka.Error Error { get; }
+        public Error Error { get; }
+
+        /// <summary>
+        ///     Gets or sets a value indicating whether this error has been handled in the subscriber and doesn't need to be logged or
+        ///     handled in any other way by Silverback.
+        /// </summary>
+        public bool Handled { get; set; }
     }
 }