admin/JiShe.CollectBus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Merge branch 'dev' of https://310.jisheyun.com/daizan/JiShe.CollectBus into dev

Browse Source
This commit is contained in:
ChenYi 2025-04-30 15:58:08 +08:00
commit fd9e3e5af3
2 changed files with 19 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs
View File

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

3
modules/JiShe.CollectBus.Kafka/Internal/KafkaPollyPipeline.cs
View File

@ -33,7 +33,8 @@ namespace JiShe.CollectBus.Kafka.Internal
ErrorCode.RebalanceInProgress, ErrorCode.RebalanceInProgress,
ErrorCode.NotCoordinatorForGroup, ErrorCode.NotCoordinatorForGroup,
ErrorCode.NetworkException, ErrorCode.NetworkException,
ErrorCode.GroupCoordinatorNotAvailable ErrorCode.GroupCoordinatorNotAvailable,
ErrorCode.InvalidGroupId
}; };
return ex switch return ex switch
{ {