Add reentrant push consumer message receiving support for c# sdk

csharp/rocketmq-client-csharp/Consumer.cs

@@ -105,9 +105,10 @@ protected Proto.ReceiveMessageRequest WrapReceiveMessageRequest(int batchSize, M
  };
  }
 
- internal Proto.ReceiveMessageRequest WrapReceiveMessageRequest(int batchSize, MessageQueue mq,
- FilterExpression filterExpression, TimeSpan awaitDuration)
+ protected internal Proto.ReceiveMessageRequest WrapReceiveMessageRequest(int batchSize, MessageQueue mq,
+ FilterExpression filterExpression, TimeSpan awaitDuration, string attemptId)
  {
+ attemptId ??= Guid.NewGuid().ToString();
  var group = new Proto.Resource
  {
  ResourceNamespace = ClientConfig.Namespace,
@@ -120,7 +121,8 @@ internal Proto.ReceiveMessageRequest WrapReceiveMessageRequest(int batchSize, Me
  FilterExpression = WrapFilterExpression(filterExpression),
  LongPollingTimeout = Duration.FromTimeSpan(awaitDuration),
  BatchSize = batchSize,
- AutoRenew = true
+ AutoRenew = true,
+ AttemptId = attemptId
  };
  }
  }

csharp/rocketmq-client-csharp/ProcessQueue.cs

@@ -20,6 +20,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Apache.Rocketmq.V2;
+using Grpc.Core;
 using Microsoft.Extensions.Logging;
 using Org.Apache.Rocketmq.Error;
 
@@ -166,20 +167,20 @@ public async Task FetchMessageImmediately()
  /// <remarks>
  /// Make sure that no exception will be thrown.
  /// </remarks>
- public async Task OnReceiveMessageException(Exception t)
+ public async Task OnReceiveMessageException(Exception t, string attemptId)
  {
  var delay = t is TooManyRequestsException ? ReceivingFlowControlBackoffDelay : ReceivingFailureBackoffDelay;
- await ReceiveMessageLater(delay);
+ await ReceiveMessageLater(delay, attemptId);
  }
 
- private async Task ReceiveMessageLater(TimeSpan delay)
+ private async Task ReceiveMessageLater(TimeSpan delay, string attemptId)
  {
  var clientId = _consumer.GetClientId();
  try
  {
  Logger.LogInformation($"Try to receive message later, mq={_mq}, delay={delay}, clientId={clientId}");
  await Task.Delay(delay, _receiveMsgCts.Token);
- await ReceiveMessage();
+ await ReceiveMessage(attemptId);
  }
  catch (Exception ex)
  {
@@ -188,11 +189,21 @@ private async Task ReceiveMessageLater(TimeSpan delay)
  return;
  }
  Logger.LogError(ex, $"[Bug] Failed to schedule message receiving request, mq={_mq}, clientId={clientId}");
- await OnReceiveMessageException(ex);
+ await OnReceiveMessageException(ex, attemptId);
  }
  }
 
+ private string GenerateAttemptId()
+ {
+ return Guid.NewGuid().ToString();
+ }
+
  public async Task ReceiveMessage()
+ {
+ await ReceiveMessage(GenerateAttemptId());
+ }
+
+ public async Task ReceiveMessage(string attemptId)
  {
  var clientId = _consumer.GetClientId();
  if (_dropped)
@@ -203,36 +214,50 @@ public async Task ReceiveMessage()
  if (IsCacheFull())
  {
  Logger.LogWarning($"Process queue cache is full, would receive message later, mq={_mq}, clientId={clientId}");
- await ReceiveMessageLater(ReceivingBackoffDelayWhenCacheIsFull);
+ await ReceiveMessageLater(ReceivingBackoffDelayWhenCacheIsFull, attemptId);
  return;
  }
- await ReceiveMessageImmediately();
+ await ReceiveMessageImmediately(attemptId);
  }
 
  private async Task ReceiveMessageImmediately()
+ {
+ await ReceiveMessageImmediately(GenerateAttemptId());
+ }
+
+ private async Task ReceiveMessageImmediately(string attemptId)
  {
  var clientId = _consumer.GetClientId();
  if (_consumer.State != State.Running)
  {
  Logger.LogInformation($"Stop to receive message because consumer is not running, mq={_mq}, clientId={clientId}");
  return;
  }
+
+ var endpoints = _mq.Broker.Endpoints;
+ var batchSize = GetReceptionBatchSize();
+ var longPollingTimeout = _consumer.GetPushConsumerSettings().GetLongPollingTimeout();
+ var request = _consumer.WrapReceiveMessageRequest(batchSize, _mq, _filterExpression, longPollingTimeout, attemptId);
+
+ Interlocked.Exchange(ref _activityTime, DateTime.UtcNow.Ticks);
+
  try
  {
- var endpoints = _mq.Broker.Endpoints;
- var batchSize = GetReceptionBatchSize();
- var longPollingTimeout = _consumer.GetPushConsumerSettings().GetLongPollingTimeout();
- var request = _consumer.WrapReceiveMessageRequest(batchSize, _mq, _filterExpression, longPollingTimeout);
-
- Interlocked.Exchange(ref _activityTime, DateTime.UtcNow.Ticks);
-
  var result = await _consumer.ReceiveMessage(request, _mq, longPollingTimeout);
  await OnReceiveMessageResult(result);
  }
  catch (Exception ex)
  {
- Logger.LogError(ex, $"Exception raised during message reception, mq={_mq}, clientId={clientId}");
- await OnReceiveMessageException(ex);
+ string nextAttemptId = null;
+ if (ex is RpcException rpcException &&
+ StatusCode.DeadlineExceeded.Equals(rpcException.Status.StatusCode))
+ {
+ nextAttemptId = request.AttemptId;
+ }
+ Logger.LogError(ex, $"Exception raised during message reception, mq={_mq}," +
+ $" attemptId={request.AttemptId}, nextAttemptId={nextAttemptId}," +
+ $" clientId={clientId}");
+ await OnReceiveMessageException(ex, attemptId);
  }
  }