diff --git a/VigilCareClinicalAPI/BackgroundServices/ElasticsSearch/EsIndexerService.cs b/VigilCareClinicalAPI/BackgroundServices/ElasticsSearch/EsIndexerService.cs index 1de5753..62b559a 100644 --- a/VigilCareClinicalAPI/BackgroundServices/ElasticsSearch/EsIndexerService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/ElasticsSearch/EsIndexerService.cs @@ -54,6 +54,8 @@ public class EsIndexerService : BackgroundService _logger.LogInformation("EsIndexerService started. Subscribed to 5 topics."); + var guard = new PoisonPillGuard("es-indexer", _kafkaOptions.MaxPoisonRetries, _logger); + try { while (!stoppingToken.IsCancellationRequested) @@ -64,6 +66,7 @@ public class EsIndexerService : BackgroundService result = consumer.Consume(stoppingToken); await DispatchAsync(result.Topic, result.Message.Value, stoppingToken); consumer.Commit(result); + guard.OnSuccess(); } catch (OperationCanceledException) { @@ -71,10 +74,15 @@ public class EsIndexerService : BackgroundService } catch (Exception ex) { + if (result is not null && guard.ShouldSkip(result, ex)) + { + consumer.Commit(result); + continue; + } + _logger.LogError(ex, - "EsIndexer failed processing topic={Topic} offset={Offset} — not committing", + "EsIndexer failed processing topic={Topic} offset={Offset} — will retry", result?.Topic, result?.Offset.Value); - // Do not commit: message will be redelivered on restart await Task.Delay(1000, stoppingToken); } } diff --git a/VigilCareClinicalAPI/BackgroundServices/GcsScoringService.cs b/VigilCareClinicalAPI/BackgroundServices/GcsScoringService.cs index dc7598c..3479e0b 100644 --- a/VigilCareClinicalAPI/BackgroundServices/GcsScoringService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/GcsScoringService.cs @@ -33,6 +33,8 @@ public class GcsScoringService : BackgroundService _logger.LogInformation("GcsScoringService started — consumer group: gcs-scoring"); + var guard = new PoisonPillGuard("gcs-scoring", _kafkaOptions.MaxPoisonRetries, _logger); + try { while (!stoppingToken.IsCancellationRequested) @@ -62,6 +64,7 @@ public class GcsScoringService : BackgroundService evt.EncounterId, outcome.TotalScore, outcome.Classification); consumer.Commit(result); + guard.OnSuccess(); } catch (OperationCanceledException) { @@ -69,8 +72,14 @@ public class GcsScoringService : BackgroundService } catch (Exception ex) { + if (result is not null && guard.ShouldSkip(result, ex)) + { + consumer.Commit(result); + continue; + } + _logger.LogError(ex, - "GcsScoringService failed on topic={Topic} offset={Offset} — not committing", + "GcsScoringService failed on topic={Topic} offset={Offset} — will retry", result?.Topic, result?.Offset.Value); await Task.Delay(2000, stoppingToken); } diff --git a/VigilCareClinicalAPI/BackgroundServices/News2ScoringService.cs b/VigilCareClinicalAPI/BackgroundServices/News2ScoringService.cs index 42d0830..e683bb0 100644 --- a/VigilCareClinicalAPI/BackgroundServices/News2ScoringService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/News2ScoringService.cs @@ -33,6 +33,8 @@ public class News2ScoringService : BackgroundService _logger.LogInformation("News2ScoringService started — consumer group: news2-scoring"); + var guard = new PoisonPillGuard("news2-scoring", _kafkaOptions.MaxPoisonRetries, _logger); + try { while (!stoppingToken.IsCancellationRequested) @@ -62,6 +64,7 @@ public class News2ScoringService : BackgroundService evt.EncounterId, outcome.TotalScore, outcome.RiskLevel); consumer.Commit(result); + guard.OnSuccess(); } catch (OperationCanceledException) { @@ -69,8 +72,14 @@ public class News2ScoringService : BackgroundService } catch (Exception ex) { + if (result is not null && guard.ShouldSkip(result, ex)) + { + consumer.Commit(result); + continue; + } + _logger.LogError(ex, - "News2ScoringService failed on topic={Topic} offset={Offset} — not committing", + "News2ScoringService failed on topic={Topic} offset={Offset} — will retry", result?.Topic, result?.Offset.Value); await Task.Delay(2000, stoppingToken); } diff --git a/VigilCareClinicalAPI/BackgroundServices/Notifications/NotificationPublisherService.cs b/VigilCareClinicalAPI/BackgroundServices/Notifications/NotificationPublisherService.cs index b6eb4ef..68625d4 100644 --- a/VigilCareClinicalAPI/BackgroundServices/Notifications/NotificationPublisherService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/Notifications/NotificationPublisherService.cs @@ -50,6 +50,8 @@ public sealed class NotificationPublisherService : BackgroundService _logger.LogInformation("NotificationPublisherService started — consuming {Topics}", string.Join(", ", _kafkaOptions.Topics.AlertGenerated, _kafkaOptions.Topics.EncounterStatusChanged)); + var guard = new PoisonPillGuard("notification-publisher", _kafkaOptions.MaxPoisonRetries, _logger); + try { while (!stoppingToken.IsCancellationRequested) @@ -75,11 +77,17 @@ public sealed class NotificationPublisherService : BackgroundService await HandleEncounterStatusChangedAsync(chan, props, result.Message.Value, stoppingToken); consumer.Commit(result); + guard.OnSuccess(); } catch (Exception ex) { - _logger.LogError(ex, "NotificationPublisher failed to process message from {Topic}", result.Topic); - // Do not commit — message will be reprocessed after consumer restart. + if (guard.ShouldSkip(result, ex)) + { + consumer.Commit(result); + continue; + } + + _logger.LogError(ex, "NotificationPublisher failed to process message from {Topic} — will retry", result.Topic); } } } diff --git a/VigilCareClinicalAPI/BackgroundServices/SepsisEngineService.cs b/VigilCareClinicalAPI/BackgroundServices/SepsisEngineService.cs index 16cf0b9..052ec49 100644 --- a/VigilCareClinicalAPI/BackgroundServices/SepsisEngineService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/SepsisEngineService.cs @@ -39,6 +39,8 @@ public class SepsisEngineService : BackgroundService _logger.LogInformation( "SepsisEngineService started — consumer group: sepsis-engine (qSOFA screening only)"); + var guard = new PoisonPillGuard("sepsis-engine", _kafkaOptions.MaxPoisonRetries, _logger); + try { while (!stoppingToken.IsCancellationRequested) @@ -68,6 +70,7 @@ public class SepsisEngineService : BackgroundService evt.EncounterId, evt.ObservationCode, evt.Value); consumer.Commit(result); + guard.OnSuccess(); } catch (OperationCanceledException) { @@ -75,8 +78,14 @@ public class SepsisEngineService : BackgroundService } catch (Exception ex) { + if (result is not null && guard.ShouldSkip(result, ex)) + { + consumer.Commit(result); + continue; + } + _logger.LogError(ex, - "SepsisEngine failed on topic={Topic} offset={Offset} — not committing", + "SepsisEngine failed on topic={Topic} offset={Offset} — will retry", result?.Topic, result?.Offset.Value); await Task.Delay(2000, stoppingToken); } diff --git a/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs b/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs index adfa9a6..2ede97b 100644 --- a/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs @@ -37,6 +37,8 @@ public class SofaScoringService : BackgroundService _logger.LogInformation("SofaScoringService started — consumer group: sofa-scoring"); + var guard = new PoisonPillGuard("sofa-scoring", _kafkaOptions.MaxPoisonRetries, _logger); + try { while (!stoppingToken.IsCancellationRequested) @@ -92,6 +94,7 @@ public class SofaScoringService : BackgroundService } consumer.Commit(result); + guard.OnSuccess(); } catch (OperationCanceledException) { @@ -99,8 +102,14 @@ public class SofaScoringService : BackgroundService } catch (Exception ex) { + if (result is not null && guard.ShouldSkip(result, ex)) + { + consumer.Commit(result); + continue; + } + _logger.LogError(ex, - "SofaScoringService failed on topic={Topic} offset={Offset} — not committing", + "SofaScoringService failed on topic={Topic} offset={Offset} — will retry", result?.Topic, result?.Offset.Value); await Task.Delay(2000, stoppingToken); } diff --git a/VigilCareClinicalAPI/BackgroundServices/TrendAnalyzerService.cs b/VigilCareClinicalAPI/BackgroundServices/TrendAnalyzerService.cs index cf96662..4045ff6 100644 --- a/VigilCareClinicalAPI/BackgroundServices/TrendAnalyzerService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/TrendAnalyzerService.cs @@ -33,6 +33,8 @@ public class TrendAnalyzerService : BackgroundService _logger.LogInformation("TrendAnalyzerService started — consumer group: trend-analyzer"); + var guard = new PoisonPillGuard("trend-analyzer", _kafkaOptions.MaxPoisonRetries, _logger); + try { while (!stoppingToken.IsCancellationRequested) @@ -63,6 +65,7 @@ public class TrendAnalyzerService : BackgroundService evt.EncounterId, outcome.ObservationCode, outcome.RatePerMinute); consumer.Commit(result); + guard.OnSuccess(); } catch (OperationCanceledException) { @@ -70,8 +73,14 @@ public class TrendAnalyzerService : BackgroundService } catch (Exception ex) { + if (result is not null && guard.ShouldSkip(result, ex)) + { + consumer.Commit(result); + continue; + } + _logger.LogError(ex, - "TrendAnalyzerService failed on topic={Topic} offset={Offset} — not committing", + "TrendAnalyzerService failed on topic={Topic} offset={Offset} — will retry", result?.Topic, result?.Offset.Value); await Task.Delay(2000, stoppingToken); } diff --git a/VigilCareClinicalAPI/BackgroundServices/WarningAlertService.cs b/VigilCareClinicalAPI/BackgroundServices/WarningAlertService.cs index 65cfa73..a3c2ab9 100644 --- a/VigilCareClinicalAPI/BackgroundServices/WarningAlertService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/WarningAlertService.cs @@ -33,6 +33,8 @@ public class WarningAlertService : BackgroundService _logger.LogInformation("WarningAlertService started — consumer group: warning-evaluator"); + var guard = new PoisonPillGuard("warning-evaluator", _kafkaOptions.MaxPoisonRetries, _logger); + try { while (!stoppingToken.IsCancellationRequested) @@ -58,6 +60,7 @@ public class WarningAlertService : BackgroundService stoppingToken); consumer.Commit(result); + guard.OnSuccess(); } catch (OperationCanceledException) { @@ -65,8 +68,14 @@ public class WarningAlertService : BackgroundService } catch (Exception ex) { + if (result is not null && guard.ShouldSkip(result, ex)) + { + consumer.Commit(result); + continue; + } + _logger.LogError(ex, - "WarningAlertService failed on topic={Topic} offset={Offset} — not committing", + "WarningAlertService failed on topic={Topic} offset={Offset} — will retry", result?.Topic, result?.Offset.Value); await Task.Delay(2000, stoppingToken); } diff --git a/VigilCareClinicalAPI/Configuration/KafkaOptions.cs b/VigilCareClinicalAPI/Configuration/KafkaOptions.cs index 7865fdd..0132abf 100644 --- a/VigilCareClinicalAPI/Configuration/KafkaOptions.cs +++ b/VigilCareClinicalAPI/Configuration/KafkaOptions.cs @@ -7,4 +7,5 @@ public class KafkaOptions public short ReplicationFactor { get; set; } = 3; public int OutboxBatchSize { get; set; } = 100; public int OutboxPollIntervalMs { get; set; } = 500; + public int MaxPoisonRetries { get; set; } = 5; } diff --git a/VigilCareClinicalAPI/Infrastructure/PoisonPillGuard.cs b/VigilCareClinicalAPI/Infrastructure/PoisonPillGuard.cs new file mode 100644 index 0000000..06543f1 --- /dev/null +++ b/VigilCareClinicalAPI/Infrastructure/PoisonPillGuard.cs @@ -0,0 +1,94 @@ +using System.Text.Json; +using Confluent.Kafka; +using Prometheus; + +/// +/// Prevents a single un-processable Kafka message from blocking a consumer +/// partition forever. Permanent errors (malformed JSON, bad format) are +/// skipped immediately; transient errors are retried up to +/// times before the offset is committed and +/// the message is abandoned. +/// +public sealed class PoisonPillGuard +{ + private static readonly Counter PoisonPillsSkipped = Metrics.CreateCounter( + "kafka_poison_pills_skipped_total", + "Messages skipped as poison pills, labeled by consumer group and topic.", + labelNames: new[] { "consumer_group", "topic" }); + + private readonly string _consumerGroup; + private readonly int _maxRetries; + private readonly ILogger _logger; + + private (string Topic, int Partition, long Offset)? _lastFailedKey; + private int _retryCount; + + public PoisonPillGuard(string consumerGroup, int maxRetries, ILogger logger) + { + _consumerGroup = consumerGroup; + _maxRetries = maxRetries; + _logger = logger; + } + + public bool ShouldSkip(ConsumeResult result, Exception ex) + { + if (IsPermanent(ex)) + { + LogSkip(result, ex, "permanent"); + return true; + } + + var key = (result.Topic, result.Partition.Value, result.Offset.Value); + + if (_lastFailedKey == key) + { + _retryCount++; + } + else + { + _lastFailedKey = key; + _retryCount = 1; + } + + if (_retryCount >= _maxRetries) + { + LogSkip(result, ex, $"transient after {_retryCount} retries"); + return true; + } + + return false; + } + + public void OnSuccess() + { + _lastFailedKey = null; + _retryCount = 0; + } + + private void LogSkip(ConsumeResult result, Exception ex, string reason) + { + PoisonPillsSkipped.WithLabels(_consumerGroup, result.Topic).Inc(); + + _logger.LogCritical(ex, + "Poison pill skipped ({Reason}): consumer_group={Group} topic={Topic} " + + "partition={Partition} offset={Offset} payload={Payload}", + reason, _consumerGroup, result.Topic, + result.Partition.Value, result.Offset.Value, + Truncate(result.Message.Value, 2000)); + } + + private static bool IsPermanent(Exception ex) => + GetRoot(ex) is JsonException or FormatException or ArgumentNullException; + + private static Exception GetRoot(Exception ex) + { + while (ex.InnerException is not null) + ex = ex.InnerException; + return ex; + } + + private static string? Truncate(string? value, int max) => + value is null ? null + : value.Length <= max ? value + : value[..max] + "…[truncated]"; +}