From a70cf4b7ce9a1f6ce6157daed5e82e474ed60041 Mon Sep 17 00:00:00 2001
From: zenghongyao <873884283@qq.com>
Date: Wed, 30 Apr 2025 15:14:23 +0800
Subject: [PATCH 1/2] =?UTF-8?q?kafka=E4=BC=98=E5=8C=96=E5=BC=82=E5=B8=B8?=
 =?UTF-8?q?=E9=87=8D=E8=AF=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../Consumer/ConsumerService.cs               | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git a/modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs b/modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs
index 98e3166..bf752eb 100644
--- a/modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs
+++ b/modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs
@@ -183,6 +183,11 @@ namespace JiShe.CollectBus.Kafka.Consumer
                             _logger.LogError(ex, $"{string.Join("、", topics)}消息消费失败: {ex.Error.Reason}");
                             throw; // 抛出异常，以便重试
                         }
+                        catch (KafkaException ex) when (KafkaPollyPipeline.IsRecoverableError(ex))
+                        {
+                            _logger.LogError(ex, $"{string.Join("、", topics)} 消息消费失败: {ex.Error.Reason}");
+                            throw; // 抛出异常，以便重试
+                        }
                         catch (OperationCanceledException)
                         {
                             //ignore
@@ -403,13 +408,18 @@ namespace JiShe.CollectBus.Kafka.Consumer
                             _logger.LogError(ex, $"{string.Join("、", topics)} 消息消费失败: {ex.Error.Reason}");
                             throw; // 抛出异常，以便重试
                         }
+                        catch (KafkaException ex) when (KafkaPollyPipeline.IsRecoverableError(ex))
+                        {
+                            _logger.LogError(ex, $"{string.Join("、", topics)} 消息消费失败: {ex.Error.Reason}");
+                            throw; // 抛出异常，以便重试
+                        }
                         catch (OperationCanceledException)
                         {
                             //ignore
                         }
                         catch (Exception ex)
                         {
-                            _logger.LogError(ex, "处理批量消息时发生未知错误");
+                           _logger.LogError(ex, "处理批量消息时发生未知错误");
                         }
                     }
                 }, cts.Token);
@@ -540,13 +550,18 @@ namespace JiShe.CollectBus.Kafka.Consumer
                             _logger.LogError(ex, $"消息消费失败: {ex.Error.Reason}");
                             throw; // 抛出异常，以便重试
                         }
+                        catch (KafkaException ex) when (KafkaPollyPipeline.IsRecoverableError(ex))
+                        {
+                            _logger.LogError(ex, $"{string.Join("、", topics)} 消息消费失败: {ex.Error.Reason}");
+                            throw; // 抛出异常，以便重试
+                        }
                         catch (OperationCanceledException)
                         {
                             //ignore
                         }
                         catch (Exception ex)
                         {
-                            _logger.LogError(ex, "处理批量消息时发生未知错误");
+                           _logger.LogError(ex, "处理批量消息时发生未知错误");
                         }
                     }
                 }, cts.Token);

From 92b9cb37c169b5e8ce0916a5628b8976995e06c8 Mon Sep 17 00:00:00 2001
From: zenghongyao <873884283@qq.com>
Date: Wed, 30 Apr 2025 15:17:51 +0800
Subject: [PATCH 2/2] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E4=BB=A3=E7=A0=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs b/modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs
index 9750f86..6f1e4fd 100644
--- a/modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs
+++ b/modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs
@@ -33,7 +33,8 @@ namespace JiShe.CollectBus.Kafka.Internal
                 ErrorCode.RebalanceInProgress,
                 ErrorCode.NotCoordinatorForGroup,
                 ErrorCode.NetworkException,
-                ErrorCode.GroupCoordinatorNotAvailable
+                ErrorCode.GroupCoordinatorNotAvailable,
+                ErrorCode.InvalidGroupId
             };
             return ex switch
             {