fix: Kafka consumer poison pill causes infinite retry

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);
                 }
             }

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);
                 }

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);
                 }

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);
+                    if (guard.ShouldSkip(result, ex))
-                    // Do not commit — message will be reprocessed after consumer restart.
+                    {
+                        consumer.Commit(result);
+                        continue;
+                    }
+
+                    _logger.LogError(ex, "NotificationPublisher failed to process message from {Topic} — will retry", result.Topic);
                 }
             }
         }

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);
                 }

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);
                 }

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);
                 }

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);
                 }

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;
 }

VigilCareClinicalAPI/Infrastructure/PoisonPillGuard.cs

@@ -0,0 +1,94 @@
+using System.Text.Json;
+using Confluent.Kafka;
+using Prometheus;
+
+/// <summary>
+/// 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
+/// <paramref name="maxRetries"/> times before the offset is committed and
+/// the message is abandoned.
+/// </summary>
+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<string, string> 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<string, string> 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]";
+}