Fixes kafka consume excepiton for GroupLoadInProress errcode (#1085)

* Fixes and add options of kafka consume excepiton for GroupLoadInProress errcode. #1084 * clean code * Add kafka consume exception logs.
dotnetcore · Feb 15, 2022 · 2288025 · 2288025
1 parent 0b81a18
commit 2288025
Show file tree

Hide file tree

Showing 4 changed files with 32 additions and 3 deletions.
diff --git a/src/DotNetCore.CAP.Kafka/CAP.KafkaOptions.cs b/src/DotNetCore.CAP.Kafka/CAP.KafkaOptions.cs
@@ -24,6 +24,10 @@ public class KafkaOptions
         public KafkaOptions()
         {
             MainConfig = new Dictionary<string, string>();
+            RetriableErrorCodes = new List<ErrorCode>
+            {
+                ErrorCode.GroupLoadInProress
+            };
         }
 
         /// <summary>
@@ -43,5 +47,10 @@ public KafkaOptions()
         /// If you need to get offset and partition and so on.., you can use this function to write additional header into <see cref="CapHeader"/>
         /// </summary>
         public Func<ConsumeResult<string, byte[]>, List<KeyValuePair<string, string>>>? CustomHeaders { get; set; }
+
+        /// <summary>
+        /// New retriable error code (refer to https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafkacpp_8h.html#a4c6b7af48c215724c323c60ea4080dbf)
+        /// </summary>
+        public IList<ErrorCode> RetriableErrorCodes { get; set; }
     }
 }
diff --git a/src/DotNetCore.CAP.Kafka/KafkaConsumerClient.cs b/src/DotNetCore.CAP.Kafka/KafkaConsumerClient.cs
@@ -17,7 +17,7 @@ namespace DotNetCore.CAP.Kafka
 {
     public class KafkaConsumerClient : IConsumerClient
     {
-        private static readonly SemaphoreSlim ConnectionLock = new SemaphoreSlim(initialCount: 1, maxCount: 1);
+        private static readonly SemaphoreSlim ConnectionLock = new(initialCount: 1, maxCount: 1);
 
         private readonly string _groupId;
         private readonly KafkaOptions _kafkaOptions;
@@ -33,7 +33,7 @@ public KafkaConsumerClient(string groupId, IOptions<KafkaOptions> options)
 
         public event EventHandler<LogMessageEventArgs>? OnLog;
 
-        public BrokerAddress BrokerAddress => new ("Kafka", _kafkaOptions.Servers);
+        public BrokerAddress BrokerAddress => new("Kafka", _kafkaOptions.Servers);
 
         public ICollection<string> FetchTopics(IEnumerable<string> topicNames)
         {
@@ -89,7 +89,23 @@ public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
 
             while (true)
             {
-                var consumerResult = _consumerClient!.Consume(cancellationToken);
+                ConsumeResult<string, byte[]> consumerResult;
+
+                try
+                {
+                    consumerResult = _consumerClient!.Consume(cancellationToken);
+                }
+                catch (ConsumeException e) when (_kafkaOptions.RetriableErrorCodes.Contains(e.Error.Code))
+                {
+                    var logArgs = new LogMessageEventArgs
+                    {
+                        LogType = MqLogType.ConsumeRetries,
+                        Reason = e.Error.ToString()
+                    };
+                    OnLog?.Invoke(null, logArgs);
+
+                    continue;
+                }
 
                 if (consumerResult.IsPartitionEOF || consumerResult.Message.Value == null) continue;
 

diff --git a/src/DotNetCore.CAP/Internal/IConsumerRegister.Default.cs b/src/DotNetCore.CAP/Internal/IConsumerRegister.Default.cs
@@ -304,6 +304,9 @@ private void WriteLog(object sender, LogMessageEventArgs logmsg)
                 case MqLogType.ConsumeError:
                     _logger.LogError("Kafka client consume error. --> " + logmsg.Reason);
                     break;
+                case MqLogType.ConsumeRetries:
+                    _logger.LogWarning("Kafka client consume exception, retying... --> " + logmsg.Reason);
+                    break;
                 case MqLogType.ServerConnError:
                     _isHealthy = false;
                     _logger.LogCritical("Kafka server connection error. --> " + logmsg.Reason);

diff --git a/src/DotNetCore.CAP/Transport/MqLogType.cs b/src/DotNetCore.CAP/Transport/MqLogType.cs
@@ -15,6 +15,7 @@ public enum MqLogType
 
         //Kafka
         ConsumeError,
+        ConsumeRetries,
         ServerConnError,
 
         //AzureServiceBus