Merge branch 'zhy_feat_cev_v7' into dev

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,6 +408,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
@@ -540,6 +550,11 @@ 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