94 lines
2.8 KiB
C#
94 lines
2.8 KiB
C#
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 the configured
|
|
/// maximum retry count 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]";
|
|
}
|