Support to set max retries for kafka pubsub processing

src/CloudEventDotNet.Kafka/KafkaAtLeastOnceConsumer.cs

@@ -39,7 +39,7 @@ public KafkaAtLeastOnceConsumer(
             .SetLogHandler((_, log) => _telemetry.OnConsumerLog(log))
             .Build();
 
-        _workItemContext = new KafkaWorkItemContext(registry, new(options, _telemetry));
+        _workItemContext = new KafkaWorkItemContext(registry, new KafkaCloudEventRepublisher(options, _telemetry));
         _topics = registry.GetSubscribedTopics(pubSubName).ToArray();
         _telemetry.Logger.LogDebug("KafkaAtLeastOnceConsumer created");
     }

src/CloudEventDotNet.Kafka/KafkaAtMostOnceConsumer.cs

@@ -34,7 +34,7 @@ public KafkaAtMostOnceConsumer(
             Acks = Acks.Leader,
             LingerMs = 10
         };
-        _workItemContext = new KafkaWorkItemContext(registry, new(options, _telemetry));
+        _workItemContext = new KafkaWorkItemContext(registry, new KafkaCloudEventRepublisher(options, _telemetry));
         _topics = registry.GetSubscribedTopics(pubSubName).ToArray();
 
         var channelContext = new KafkaMessageChannelContext(

src/CloudEventDotNet.Kafka/KafkaCloudEventPublisher.cs

@@ -9,9 +9,11 @@ internal sealed class KafkaCloudEventPublisher : ICloudEventPublisher
 {
     private readonly IProducer<byte[], byte[]> _producer;
     private readonly KafkaProducerTelemetry _telemetry;
+    private readonly ILogger _logger;
 
     public KafkaCloudEventPublisher(
         string pubSubName,
+        ILogger<KafkaCloudEventPublisher> logger,
         KafkaPublishOptions options,
         ILoggerFactory loggerFactory)
     {
@@ -20,6 +22,7 @@ public KafkaCloudEventPublisher(
             .SetErrorHandler((_, e) => _telemetry.OnProducerError(e))
             .SetLogHandler((_, log) => _telemetry.OnProducerLog(log))
             .Build();
+        _logger = logger;
     }
 
     public async Task PublishAsync<TData>(string topic, CloudEvent<TData> cloudEvent)
@@ -32,4 +35,5 @@ public async Task PublishAsync<TData>(string topic, CloudEvent<TData> cloudEvent
         DeliveryResult<byte[], byte[]> result = await _producer.ProduceAsync(topic, message).ConfigureAwait(false);
         _telemetry.OnMessageProduced(result, _producer.Name);
     }
+
 }

src/CloudEventDotNet.Kafka/KafkaCloudEventRepublisher.cs

@@ -0,0 +1,65 @@
+
+using Confluent.Kafka;
+using Microsoft.Extensions.Logging;
+
+namespace CloudEventDotNet.Kafka;
+
+internal class KafkaCloudEventRepublisher : ICloudEventRepublisher
+{
+    private readonly IProducer<byte[], byte[]> _producer;
+    private readonly KafkaSubscribeOptions _options;
+    private readonly KafkaConsumerTelemetry _telemetry;
+
+    public KafkaCloudEventRepublisher(
+        KafkaSubscribeOptions options,
+        KafkaConsumerTelemetry telemetry
+    )
+    {
+        var producerConfig = new ProducerConfig()
+        {
+            BootstrapServers = options.ConsumerConfig.BootstrapServers,
+            Acks = Acks.Leader,
+            LingerMs = 10,
+
+            // producer retries sending messages by default
+        };
+
+        _producer = new ProducerBuilder<byte[], byte[]>(producerConfig)
+            .SetErrorHandler((_, e) => telemetry.OnProducerError(e))
+            .SetLogHandler((_, log) => telemetry.OnProducerLog(log))
+            .Build();
+        _options = options;
+        _telemetry = telemetry;
+    }
+
+    public async Task RepublishAsync(string topic, CloudEvent cloudEvent)
+    {
+        int retry = cloudEvent.Retry;
+
+        if (retry >= _options.MaxRetries)
+        {
+            CloudEventProcessingTelemetry.LogOnCloudEventDropped(_telemetry.Logger, cloudEvent.Id);
+            return;
+        }
+
+        cloudEvent.Retry = retry + 1;
+
+        var message = new Message<byte[], byte[]>
+        {
+            Value = JSON.SerializeToUtf8Bytes(cloudEvent)
+        };
+        DeliveryResult<byte[], byte[]> result;
+        try
+        {
+            result = await _producer.ProduceAsync(topic, message).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            _telemetry.Logger.LogError(ex, "Failed to republish cloud event {id}", cloudEvent.Id);
+            CloudEventProcessingTelemetry.LogOnCloudEventDropped(_telemetry.Logger, cloudEvent.Id);
+            return;
+        }
+        _telemetry.Logger.LogInformation("Republished cloud event {id} at {offset}", cloudEvent.Id, result.TopicPartitionOffset);
+
+    }
+}

src/CloudEventDotNet.Kafka/KafkaMessageWorkItem.cs

@@ -1,4 +1,4 @@
-using System.Text.Json;
+using System.Diagnostics;
 using Confluent.Kafka;
 using Microsoft.Extensions.Logging;
 
@@ -49,6 +49,7 @@ public ValueTask WaitToCompleteAsync()
 
     internal async Task ExecuteAsync()
     {
+        Activity? activity = null;
         try
         {
             var cloudEvent = JSON.Deserialize<CloudEvent>(_message.Message.Value)!;
@@ -58,12 +59,14 @@ internal async Task ExecuteAsync()
                 CloudEventProcessingTelemetry.OnHandlerNotFound(_telemetry.Logger, metadata);
                 return;
             }
-            bool succeed = await handler.ProcessAsync(cloudEvent, _cancellationTokenSource.Token).ConfigureAwait(false);
-            KafkaConsumerTelemetry.OnConsumed(_channelContext.ConsumerName, _channelContext.ConsumerGroup);
 
+            // processing
+            activity = handler.StartProcessing(cloudEvent);
+            KafkaConsumerTelemetry.OnConsuming(activity, _channelContext.ConsumerName, _channelContext.ConsumerGroup);
+            bool succeed = await handler.ProcessAsync(cloudEvent, _cancellationTokenSource.Token).ConfigureAwait(false);
             if (!succeed)
             {
-                await _context.Producer.ReproduceAsync(_message).ConfigureAwait(false);
+                await _context.Republisher.RepublishAsync(_message.Topic, cloudEvent).ConfigureAwait(false);
             }
         }
         catch (Exception ex)
@@ -73,6 +76,7 @@ internal async Task ExecuteAsync()
         finally
         {
             _waiter.SetResult();
+            activity?.Dispose();
         }
     }
 }

src/CloudEventDotNet.Kafka/KafkaPubSubOptions.cs

@@ -32,6 +32,11 @@ public class KafkaSubscribeOptions
     /// The limit of unprocessed CloudEvents in local process queue.
     /// </summary>
     public int RunningWorkItemLimit { get; set; } = 1024;
+
+    /// <summary>
+    /// Max retries for cloud events failed to processing, set 0 to disable retries, default to int.MaxValue.
+    /// </summary>
+    public int MaxRetries { get; set; } = int.MaxValue;
 }
 
 public enum DeliveryGuarantee

src/CloudEventDotNet.Kafka/KafkaTelemetry.Consumer.cs

@@ -138,16 +138,17 @@ public void OnProducerLog(LogMessage log)
     )]
     public partial void OnCommitLoopError(Exception exception);
 
-    public static void OnConsumed(
+    public static void OnConsuming(
+        Activity? activity,
         string consumerName,
         string consumerGroup)
     {
-        var activity = Activity.Current;
         if (activity is not null)
         {
             // activity.SetTag("messaging.kafka.message_key", message.Key);
             activity.SetTag("messaging.kafka.client_id", consumerName);
             activity.SetTag("messaging.kafka.consumer_group", consumerGroup);
         }
     }
+
 }

src/CloudEventDotNet.Kafka/KafkaTelemetry.Producer.cs

@@ -1,4 +1,5 @@
 using System.Diagnostics;
+using System.Net.Http.Headers;
 using Confluent.Kafka;
 using Microsoft.Extensions.Logging;
 
@@ -19,18 +20,20 @@ public KafkaProducerTelemetry(string pubSubName, ILoggerFactory loggerFactory)
         Message = "Produced message {topic}:{partition}:{offset}"
     )]
     partial void LogOnMessageProduced(string topic, int partition, long offset);
+    public void LogOnMessageProduced(DeliveryResult<byte[], byte[]> result)
+        => LogOnMessageProduced(result.Topic, result.Partition.Value, result.Offset.Value);
     public void OnMessageProduced(DeliveryResult<byte[], byte[]> result, string clientId)
     {
-        var activity = Activity.Current;
-        if (activity is not null)
+        if (Activity.Current is { } activity)
         {
             activity.SetTag("messaging.kafka.client_id", clientId);
             activity.SetTag("messaging.kafka.partition", result.Partition.Value);
         }
-        LogOnMessageProduced(result.Topic, result.Partition.Value, result.Offset.Value);
+        LogOnMessageProduced(result);
     }
 
 
+
     [LoggerMessage(
         Level = LogLevel.Error,
         Message = "Producer error: {error}"

src/CloudEventDotNet.Kafka/KafkaWorkItemContext.cs

@@ -4,4 +4,4 @@ namespace CloudEventDotNet.Kafka;
 
 internal record KafkaWorkItemContext(
     Registry Registry,
-    KafkaRedeliverProducer Producer);
+    ICloudEventRepublisher Republisher);

src/CloudEventDotNet.Redis/RedisMessageWorkItem.cs

@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using System.Text.Json;
 using CloudEventDotNet.Redis.Instruments;
 using StackExchange.Redis;
@@ -45,6 +46,8 @@ public ValueTask WaitToCompleteAsync()
 
     internal async Task ExecuteAsync()
     {
+        Activity? activity = null;
+
         try
         {
             var cloudEvent = JSON.Deserialize<CloudEvent>((byte[])Message["data"]!)!;
@@ -53,6 +56,11 @@ internal async Task ExecuteAsync()
             {
                 return;
             }
+
+
+            // processing
+            activity = handler.StartProcessing(cloudEvent);
+            RedisTelemetry.OnMessageProcessing(activity, ChannelContext.ConsumerGroup, ChannelContext.ConsumerName);
             bool succeed = await handler.ProcessAsync(cloudEvent, _cancellationTokenSource.Token).ConfigureAwait(false);
             if (succeed)
             {
@@ -62,7 +70,6 @@ await _context.Redis.StreamAcknowledgeAsync(
                     Message.Id).ConfigureAwait(false);
                 _context.RedisTelemetry.OnMessageAcknowledged(Message.Id.ToString());
             }
-            RedisTelemetry.OnMessageProcessed(ChannelContext.ConsumerGroup, ChannelContext.ConsumerName);
         }
         catch (Exception ex)
         {

src/CloudEventDotNet.Redis/RedisTelemetry.cs

@@ -25,12 +25,12 @@ public static void OnMessageProduced(ILogger logger, IConnectionMultiplexer mult
         activity?.SetTag("messaging.redis.client_name", multiplexer.ClientName);
     }
 
-    public static void OnMessageProcessed(
+    public static void OnMessageProcessing(
+        Activity? activity,
         string consumerGroup,
         string consumerName
     )
     {
-        Activity? activity = Activity.Current;
         if (activity is not null)
         {
             activity.SetTag("messaging.redis.client_id", consumerName);

src/CloudEventDotNet/CloudEvent.cs

@@ -28,6 +28,19 @@ internal record CloudEvent(
 
     [JsonExtensionData]
     public Dictionary<string, JsonElement> Extensions { get; set; } = new();
+
+    [JsonIgnore]
+    public int Retry
+    {
+        get
+        {
+            return Extensions["retry"].GetInt32();
+        }
+        set
+        {
+            Extensions["retry"] = JsonSerializer.SerializeToElement(value);
+        }
+    }
 }
 
 /// <summary>