You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2026/09/16 06:04:44 UTC
(rocketmq-clients) branch master updated: [C#] Align half-open gRPC recovery with Java (#1371)
This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new fc4b90bd [C#] Align half-open gRPC recovery with Java (#1371)
fc4b90bd is described below
commit fc4b90bdb144c62556f35d59ad7a6fbce30c1092
Author: guyinyou <36...@users.noreply.github.com>
AuthorDate: Wed Sep 16 14:04:37 2026 +0800
[C#] Align half-open gRPC recovery with Java (#1371)
* fix(csharp): recover half-open gRPC connections
Align heartbeat-driven transport recovery and telemetry renewal with Java
to avoid repeated timeouts on half-open connections. Guard lifecycle
transitions and normalize transport throttling failures.
Reference implementation: https://github.com/apache/rocketmq-clients/pull/1310
Java commit: 9fe1449d19449b41442aa3a97ab168ed6b5bd6b1
Validation: net8.0/net10.0 builds and formatting checks passed;
226 tests passed in each of three full-suite runs.
BREAKING CHANGE: Custom IClientManager implementations must add
Reconnect(Endpoints); custom IRpcClient implementations must add State
and ResetTransport().
* fix(csharp): use codespell-compatible recovery test names
Rename local telemetry variables to pass CodeSpell 2.1.0 without changing
test behavior or weakening spelling checks.
* fix(csharp): make recovery tests deterministic across CI runners
Use fresh metadata for each recovery heartbeat to avoid accumulating
caller headers beyond server limits. Hold the first transport reset until
all contenders finish instead of relying on thread-pool timing.
---------
Co-authored-by: guyinyou <gu...@alibaba-inc.com>
---
.github/workflows/csharp_build.yml | 1 +
csharp/rocketmq-client-csharp/Client.cs | 303 ++++++--
csharp/rocketmq-client-csharp/ClientManager.cs | 369 ++++++++-
.../Error/ClientException.cs | 5 +
.../Error/TooManyRequestsException.cs | 7 +
csharp/rocketmq-client-csharp/IClientManager.cs | 6 +
csharp/rocketmq-client-csharp/IRpcClient.cs | 10 +
csharp/rocketmq-client-csharp/PushConsumer.cs | 17 +-
csharp/rocketmq-client-csharp/RpcClient.cs | 80 +-
csharp/rocketmq-client-csharp/Session.cs | 397 ++++++++--
csharp/tests/ClientManagerRpcExceptionTest.cs | 105 +++
csharp/tests/ClientManagerTest.cs | 845 ++++++++++++++++++++-
csharp/tests/ClientTest.cs | 385 +++++++---
csharp/tests/MockServer.cs | 27 +
.../ProducerHalfOpenTcpRecoveryIntegrationTest.cs | 340 +++++++++
csharp/tests/PushConsumerRecoveryTest.cs | 272 +++++++
csharp/tests/RecoveryTestSupport.cs | 317 ++++++++
csharp/tests/SessionTest.cs | 422 +++++++++-
csharp/tests/TransportRecoveryConcurrencyTest.cs | 190 +++++
csharp/tests/TransportRecoveryLeakTest.cs | 177 +++++
csharp/tests/UnobservedTaskExceptionTest.cs | 311 ++++++++
21 files changed, 4292 insertions(+), 294 deletions(-)
diff --git a/.github/workflows/csharp_build.yml b/.github/workflows/csharp_build.yml
index c90105dd..20afb473 100644
--- a/.github/workflows/csharp_build.yml
+++ b/.github/workflows/csharp_build.yml
@@ -20,6 +20,7 @@ jobs:
dotnet-version: |
6.0.x
8.0.x
+ 10.0.x
- name: Build artifacts
working-directory: ./csharp
run: |
diff --git a/csharp/rocketmq-client-csharp/Client.cs b/csharp/rocketmq-client-csharp/Client.cs
index 5c2960fa..efa40b87 100644
--- a/csharp/rocketmq-client-csharp/Client.cs
+++ b/csharp/rocketmq-client-csharp/Client.cs
@@ -50,6 +50,9 @@ namespace Org.Apache.Rocketmq
private static readonly TimeSpan StatsSchedulePeriod = TimeSpan.FromSeconds(60);
private readonly CancellationTokenSource _statsCts;
+ // Larger than the close timeout of a single session, which force closes itself: this one only backs it up.
+ internal static readonly TimeSpan SessionCloseTimeout = TimeSpan.FromSeconds(10);
+
protected readonly ClientConfig ClientConfig;
protected readonly Endpoints Endpoints;
protected IClientManager ClientManager;
@@ -60,7 +63,9 @@ namespace Org.Apache.Rocketmq
private readonly ConcurrentDictionary<string, TopicRouteData> _topicRouteCache;
private readonly Dictionary<Endpoints, Session> _sessionsTable;
+ private readonly Dictionary<Endpoints, int> _initializingEndpoints = new Dictionary<Endpoints, int>();
private readonly ReaderWriterLockSlim _sessionLock;
+ private bool _sessionsClosed;
internal volatile State State;
@@ -111,6 +116,37 @@ namespace Org.Apache.Rocketmq
_settingsSyncCts.Cancel();
_statsCts.Cancel();
NotifyClientTermination();
+
+ List<Session> sessions;
+ _sessionLock.EnterWriteLock();
+ try
+ {
+ _sessionsClosed = true;
+ sessions = _sessionsTable.Values.ToList();
+ foreach (var session in sessions)
+ {
+ session.MarkClosed();
+ }
+
+ _sessionsTable.Clear();
+ }
+ finally
+ {
+ _sessionLock.ExitWriteLock();
+ }
+
+ // Disposing streams can resume their readers; never do it while holding the session table lock.
+ try
+ {
+ await Task.WhenAll(sessions.Select(session => session.CloseAsync()))
+ .WaitAsync(SessionCloseTimeout);
+ }
+ catch (Exception e)
+ {
+ Logger.LogWarning(e, $"Failed to close the client sessions in {SessionCloseTimeout}, " +
+ $"clientId={ClientId}");
+ }
+
await ClientManager.Shutdown();
ClientMeterManager.Shutdown();
Logger.LogDebug($"Shutdown the rocketmq client successfully, clientId={ClientId}");
@@ -121,6 +157,7 @@ namespace Org.Apache.Rocketmq
_sessionLock.EnterReadLock();
try
{
+ ThrowIfSessionsClosed();
// Session exists, return in advance.
if (_sessionsTable.TryGetValue(endpoints, out var session))
{
@@ -135,6 +172,7 @@ namespace Org.Apache.Rocketmq
_sessionLock.EnterWriteLock();
try
{
+ ThrowIfSessionsClosed();
// Session exists, return in advance.
if (_sessionsTable.TryGetValue(endpoints, out var session))
{
@@ -152,38 +190,157 @@ namespace Org.Apache.Rocketmq
}
}
+ private bool SessionsClosed => _sessionsClosed || State == State.Stopping
+ || State == State.Terminated || State == State.Failed;
+
+ private void ThrowIfSessionsClosed()
+ {
+ if (SessionsClosed)
+ {
+ throw new ObjectDisposedException(nameof(Client));
+ }
+ }
+
+ internal Session GetSessionIfPresent(Endpoints endpoints)
+ {
+ _sessionLock.EnterReadLock();
+ try
+ {
+ return !SessionsClosed && _sessionsTable.TryGetValue(endpoints, out var session) ? session : null;
+ }
+ finally
+ {
+ _sessionLock.ExitReadLock();
+ }
+ }
+
protected abstract IEnumerable<string> GetTopics();
+ internal virtual bool IsEndpointsDeprecated(Endpoints endpoints)
+ {
+ _sessionLock.EnterReadLock();
+ try
+ {
+ return SessionsClosed || (!_initializingEndpoints.ContainsKey(endpoints)
+ && !GetTotalRouteEndpoints().Contains(endpoints));
+ }
+ finally
+ {
+ _sessionLock.ExitReadLock();
+ }
+ }
+
+ internal virtual async Task<bool> RemoveSession(Endpoints endpoints, Session session)
+ {
+ _sessionLock.EnterWriteLock();
+ try
+ {
+ if (!_sessionsTable.TryGetValue(endpoints, out var existing) || !ReferenceEquals(existing, session)
+ || (!SessionsClosed && (_initializingEndpoints.ContainsKey(endpoints)
+ || GetTotalRouteEndpoints().Contains(endpoints))))
+ {
+ return false;
+ }
+
+ session.MarkClosed();
+ _sessionsTable.Remove(endpoints);
+ }
+ finally
+ {
+ _sessionLock.ExitWriteLock();
+ }
+
+ await session.CloseAsync();
+ return true;
+ }
+
internal abstract Proto::HeartbeatRequest WrapHeartbeatRequest();
protected abstract void OnTopicRouteDataUpdated0(string topic, TopicRouteData topicRouteData);
internal async Task OnTopicRouteDataFetched(string topic, TopicRouteData topicRouteData)
{
- var routeEndpoints = new HashSet<Endpoints>();
- foreach (var mq in topicRouteData.MessageQueues)
+ var routeEndpoints = new HashSet<Endpoints>(topicRouteData.MessageQueues.Select(mq => mq.Broker.Endpoints));
+ _sessionLock.EnterWriteLock();
+ try
{
- routeEndpoints.Add(mq.Broker.Endpoints);
+ ThrowIfSessionsClosed();
+ foreach (var endpoints in routeEndpoints)
+ {
+ _initializingEndpoints.TryGetValue(endpoints, out var count);
+ _initializingEndpoints[endpoints] = count + 1;
+ }
+ }
+ finally
+ {
+ _sessionLock.ExitWriteLock();
}
- var existedRouteEndpoints = GetTotalRouteEndpoints();
- var newEndpoints = routeEndpoints.Except(existedRouteEndpoints);
-
- foreach (var endpoints in newEndpoints)
+ try
{
- var (created, session) = GetSession(endpoints);
- if (!created)
+ foreach (var endpoints in routeEndpoints)
{
- continue;
+ var (_, session) = GetSession(endpoints);
+ await session.InitializeAsync();
}
- Logger.LogInformation($"Begin to establish session for endpoints={endpoints}, clientId={ClientId}");
- await session.SyncSettings(true);
- Logger.LogInformation($"Establish session for endpoints={endpoints} successfully, clientId={ClientId}");
+ _sessionLock.EnterWriteLock();
+ try
+ {
+ ThrowIfSessionsClosed();
+ _topicRouteCache[topic] = topicRouteData;
+ OnTopicRouteDataUpdated0(topic, topicRouteData);
+ }
+ finally
+ {
+ _sessionLock.ExitWriteLock();
+ }
}
+ finally
+ {
+ List<Session> retired;
+ _sessionLock.EnterWriteLock();
+ try
+ {
+ foreach (var endpoints in routeEndpoints)
+ {
+ var count = _initializingEndpoints[endpoints] - 1;
+ if (count == 0)
+ {
+ _initializingEndpoints.Remove(endpoints);
+ }
+ else
+ {
+ _initializingEndpoints[endpoints] = count;
+ }
+ }
+
+ var active = GetTotalRouteEndpoints();
+ retired = new List<Session>();
+ foreach (var entry in _sessionsTable.ToArray())
+ {
+ if (active.Contains(entry.Key) || _initializingEndpoints.ContainsKey(entry.Key))
+ {
+ continue;
+ }
- _topicRouteCache[topic] = topicRouteData;
- OnTopicRouteDataUpdated0(topic, topicRouteData);
+ entry.Value.MarkClosed();
+ _sessionsTable.Remove(entry.Key);
+ retired.Add(entry.Value);
+ }
+ }
+ finally
+ {
+ _sessionLock.ExitWriteLock();
+ }
+
+ if (ClientManager is ClientManager manager)
+ {
+ manager.PruneTransportRecoveryStates();
+ }
+
+ await Task.WhenAll(retired.Select(session => session.CloseAsync()));
+ }
}
/**
@@ -410,10 +567,10 @@ namespace Org.Apache.Rocketmq
var request = WrapNotifyClientTerminationRequest();
foreach (var item in endpoints)
{
- var invocation =
- await ClientManager.NotifyClientTermination(item, request, ClientConfig.RequestTimeout);
try
{
+ var invocation =
+ await ClientManager.NotifyClientTermination(item, request, ClientConfig.RequestTimeout);
StatusChecker.Check(invocation.Response.Status, request, invocation.RequestId);
}
catch (Exception e)
@@ -441,6 +598,28 @@ namespace Org.Apache.Rocketmq
return ClientManager;
}
+ /// <summary>
+ /// Rebuilds the telemetry stream of the specified endpoints, invoked right after the gRPC transport has been
+ /// replaced.
+ /// </summary>
+ internal virtual void ReconnectTelemetry(Endpoints endpoints, Session expectedSession = null)
+ {
+ var session = GetSessionIfPresent(endpoints);
+ if (null != expectedSession && !ReferenceEquals(session, expectedSession))
+ {
+ return;
+ }
+
+ if (null == session)
+ {
+ Logger.LogWarning($"Failed to rebuild telemetry because client session does not exist, " +
+ $"endpoints={endpoints}, clientId={ClientId}");
+ return;
+ }
+
+ session.Reconnect();
+ }
+
// Only for testing
internal void SetClientManager(IClientManager clientManager)
{
@@ -454,52 +633,66 @@ namespace Org.Apache.Rocketmq
$"clientId={ClientId}, endpoints={endpoints}");
}
- internal virtual async void OnVerifyMessageCommand(Endpoints endpoints, Proto.VerifyMessageCommand command)
+ internal virtual async Task OnVerifyMessageCommand(Endpoints endpoints, Proto.VerifyMessageCommand command)
{
- // Only push consumer support message consumption verification.
- Logger.LogWarning($"Ignore verify message command from remote, which is not expected, clientId={ClientId}, " +
- $"endpoints={endpoints}, command={command}");
- var status = new Proto.Status
- {
- Code = Proto.Code.Unsupported,
- Message = "Message consumption verification is not supported"
- };
- var verifyMessageResult = new Proto.VerifyMessageResult
- {
- Nonce = command.Nonce
- };
-
- var telemetryCommand = new Proto.TelemetryCommand
- {
- VerifyMessageResult = verifyMessageResult,
- Status = status
- };
- var (_, session) = GetSession(endpoints);
- await session.WriteAsync(telemetryCommand);
+ try
+ {
+ var session = GetSessionIfPresent(endpoints);
+ if (null == session)
+ {
+ return;
+ }
+
+ var telemetryCommand = new Proto.TelemetryCommand
+ {
+ VerifyMessageResult = new Proto.VerifyMessageResult { Nonce = command.Nonce },
+ Status = new Proto.Status
+ {
+ Code = Proto.Code.Unsupported,
+ Message = "Message consumption verification is not supported"
+ }
+ };
+ await session.WriteAsync(telemetryCommand);
+ }
+ catch (Exception e)
+ {
+ Logger.LogWarning(e, $"Failed to reply to message verification, endpoints={endpoints}, clientId={ClientId}");
+ }
}
- internal async void OnPrintThreadStackTraceCommand(Endpoints endpoints,
+ internal async Task OnPrintThreadStackTraceCommand(Endpoints endpoints,
Proto.PrintThreadStackTraceCommand command)
{
- Logger.LogWarning("Ignore thread stack trace printing command from remote because it is still not supported, " +
- $"clientId={ClientId}, endpoints={endpoints}");
- var status = new Proto.Status
- {
- Code = Proto.Code.Unsupported,
- Message = "C# don't support thread stack trace printing"
- };
- var threadStackTrace = new Proto.ThreadStackTrace
+ try
{
- Nonce = command.Nonce
- };
+ var session = GetSessionIfPresent(endpoints);
+ if (null == session)
+ {
+ return;
+ }
- var telemetryCommand = new Proto.TelemetryCommand
+ var telemetryCommand = new Proto.TelemetryCommand
+ {
+ ThreadStackTrace = new Proto.ThreadStackTrace { Nonce = command.Nonce },
+ Status = new Proto.Status
+ {
+ Code = Proto.Code.Unsupported,
+ Message = "C# don't support thread stack trace printing"
+ }
+ };
+ await session.WriteAsync(telemetryCommand);
+ }
+ catch (Exception e)
{
- ThreadStackTrace = threadStackTrace,
- Status = status,
- };
- var (_, session) = GetSession(endpoints);
- await session.WriteAsync(telemetryCommand);
+ Logger.LogWarning(e, $"Failed to reply to thread stack trace request, endpoints={endpoints}, clientId={ClientId}");
+ }
+ }
+
+ internal virtual void OnReconnectEndpointsCommand(Endpoints endpoints, Proto.ReconnectEndpointsCommand command)
+ {
+ Logger.LogInformation($"Receive reconnect endpoints command from remote, endpoints={endpoints}, " +
+ $"command={command}, clientId={ClientId}");
+ ClientManager.Reconnect(endpoints);
}
internal virtual void OnSettingsCommand(Endpoints endpoints, Proto.Settings settings)
diff --git a/csharp/rocketmq-client-csharp/ClientManager.cs b/csharp/rocketmq-client-csharp/ClientManager.cs
index eca4881c..ba9ff296 100644
--- a/csharp/rocketmq-client-csharp/ClientManager.cs
+++ b/csharp/rocketmq-client-csharp/ClientManager.cs
@@ -17,19 +17,33 @@
using Proto = Apache.Rocketmq.V2;
using System;
+using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using grpcLib = Grpc.Core;
using System.Collections.Generic;
using System.Linq;
+using Microsoft.Extensions.Logging;
+using Org.Apache.Rocketmq.Error;
namespace Org.Apache.Rocketmq
{
public class ClientManager : IClientManager
{
+ internal const int HeartbeatFailureThreshold = 2;
+ internal static readonly TimeSpan TransportRecoveryCooldown = TimeSpan.FromSeconds(30);
+
+ private static readonly ILogger Logger = MqLogManager.CreateLogger<ClientManager>();
+
private readonly Client _client;
private readonly Dictionary<Endpoints, IRpcClient> _rpcClients;
private readonly ReaderWriterLockSlim _clientLock;
+ private readonly object _transportRecoveryLock = new object();
+ private readonly Dictionary<Endpoints, TransportRecoveryState> _transportRecoveryStates =
+ new Dictionary<Endpoints, TransportRecoveryState>();
+ // A retired state must not release a reset which is still using the endpoint's cached RPC client.
+ private readonly HashSet<Endpoints> _recoveringEndpoints = new HashSet<Endpoints>();
+ private bool _stopped;
public ClientManager(Client client)
{
@@ -74,19 +88,312 @@ namespace Org.Apache.Rocketmq
}
}
- public async Task Shutdown()
+ private IRpcClient GetRpcClientIfPresent(Endpoints endpoints)
{
_clientLock.EnterReadLock();
try
{
- var tasks = _rpcClients.Select(item => item.Value.Shutdown()).ToList();
+ return _rpcClients.TryGetValue(endpoints, out var cachedClient) ? cachedClient : null;
+ }
+ finally
+ {
+ _clientLock.ExitReadLock();
+ }
+ }
+
+ public void Reconnect(Endpoints endpoints)
+ {
+ var rpcClient = GetRpcClientIfPresent(endpoints);
+ if (null == rpcClient)
+ {
+ Logger.LogWarning($"Failed to reconnect because rpc client does not exist, endpoints={endpoints}, " +
+ $"clientId={_client.GetClientId()}");
+ return;
+ }
+
+ Reconnect(endpoints, rpcClient);
+ }
+
+ internal void Reconnect(Endpoints endpoints, IRpcClient rpcClient)
+ {
+ var recoveryState = GetTransportRecoveryState(endpoints);
+ lock (_transportRecoveryLock)
+ {
+ if (!TryAdmitTransportRecovery(endpoints, recoveryState, false))
+ {
+ return;
+ }
+ }
+
+ RecoverTransport(endpoints, rpcClient, recoveryState, "server reconnect command");
+ }
+
+ private TransportRecoveryState GetTransportRecoveryState(Endpoints endpoints)
+ {
+ lock (_transportRecoveryLock)
+ {
+ if (_stopped || _client.IsEndpointsDeprecated(endpoints))
+ {
+ RetireTransportRecoveryState(endpoints);
+ return null;
+ }
+
+ var session = _client.GetSessionIfPresent(endpoints);
+ if (!_transportRecoveryStates.TryGetValue(endpoints, out var recoveryState)
+ || !ReferenceEquals(recoveryState.Session, session))
+ {
+ RetireTransportRecoveryState(endpoints);
+ recoveryState = new TransportRecoveryState(session);
+ _transportRecoveryStates.Add(endpoints, recoveryState);
+ }
+
+ return recoveryState;
+ }
+ }
+
+ // Called outside Client's session lock: the lock order is recovery -> session, never the reverse.
+ internal void PruneTransportRecoveryStates()
+ {
+ lock (_transportRecoveryLock)
+ {
+ foreach (var endpoints in _transportRecoveryStates.Keys.ToArray())
+ {
+ if (_client.IsEndpointsDeprecated(endpoints)
+ || !ReferenceEquals(_transportRecoveryStates[endpoints].Session, _client.GetSessionIfPresent(endpoints)))
+ {
+ RetireTransportRecoveryState(endpoints);
+ }
+ }
+ }
+ }
+
+ // The following state helpers are only used while holding _transportRecoveryLock.
+ private void RetireTransportRecoveryState(Endpoints endpoints)
+ {
+ if (_transportRecoveryStates.Remove(endpoints, out var recoveryState))
+ {
+ recoveryState.Heartbeats.Clear();
+ }
+ }
+
+ private bool IsTransportRecoveryStateActive(Endpoints endpoints, TransportRecoveryState recoveryState)
+ {
+ if (_stopped || null == recoveryState
+ || !_transportRecoveryStates.TryGetValue(endpoints, out var activeState)
+ || !ReferenceEquals(activeState, recoveryState))
+ {
+ return false;
+ }
+
+ if (_client.IsEndpointsDeprecated(endpoints)
+ || !ReferenceEquals(recoveryState.Session, _client.GetSessionIfPresent(endpoints)))
+ {
+ RetireTransportRecoveryState(endpoints);
+ return false;
+ }
+
+ return true;
+ }
+
+ /// <summary>
+ /// Watches the outcome of a heartbeat and drives transport recovery from connectivity failures only. Ordinary
+ /// RPC failures never reset the transport, the native reconnection of gRPC stays in charge of those.
+ /// </summary>
+ internal Task MonitorHeartbeat(Endpoints endpoints, IRpcClient rpcClient,
+ Task<Proto.HeartbeatResponse> future)
+ {
+ return MonitorHeartbeat(endpoints, rpcClient, GetTransportRecoveryState(endpoints), future);
+ }
+
+ private Task MonitorHeartbeat(Endpoints endpoints, IRpcClient rpcClient, TransportRecoveryState recoveryState,
+ Task<Proto.HeartbeatResponse> future)
+ {
+ lock (_transportRecoveryLock)
+ {
+ if (IsTransportRecoveryStateActive(endpoints, recoveryState))
+ {
+ // Register before attaching the callback, including for an already completed heartbeat.
+ recoveryState.Heartbeats.Enqueue(future);
+ }
+ }
+
+ return future.ContinueWith(task =>
+ {
+ // Even a retired heartbeat must have its exception observed.
+ _ = task.Exception;
+ string recoveryReason = null;
+ try
+ {
+ lock (_transportRecoveryLock)
+ {
+ if (!IsTransportRecoveryStateActive(endpoints, recoveryState))
+ {
+ return Task.CompletedTask;
+ }
+
+ // Completion callbacks may run out of order. Only completed queue heads affect the counter.
+ while (recoveryState.Heartbeats.Count > 0 && recoveryState.Heartbeats.Peek().IsCompleted)
+ {
+ var completed = recoveryState.Heartbeats.Dequeue();
+ var statusCode =
+ (completed.Exception?.Flatten().InnerException as grpcLib.RpcException)?.StatusCode;
+ var recover = false;
+ switch (statusCode)
+ {
+ case grpcLib.StatusCode.Unavailable:
+ recoveryState.HeartbeatFailureAttempts = 0;
+ // Other states mean gRPC is already reconnecting on its own.
+ recover = grpcLib.ConnectivityState.Ready == rpcClient.State;
+ break;
+ case grpcLib.StatusCode.DeadlineExceeded:
+ recoveryState.HeartbeatFailureAttempts++;
+ recover = recoveryState.HeartbeatFailureAttempts >= HeartbeatFailureThreshold;
+ break;
+ default:
+ recoveryState.HeartbeatFailureAttempts = 0;
+ break;
+ }
+
+ if (recover && TryAdmitTransportRecovery(endpoints, recoveryState, true))
+ {
+ recoveryReason = $"heartbeat failure, statusCode={statusCode}";
+ }
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ Logger.LogError(e, $"[Bug] unexpected exception raised while monitoring heartbeat, " +
+ $"endpoints={endpoints}, clientId={_client.GetClientId()}");
+ }
+
+ // Only admitted recovery goes to the pool. The heartbeat caller never waits for transport or streams.
+ return null == recoveryReason ? Task.CompletedTask : Task.Run(() =>
+ RecoverTransport(endpoints, rpcClient, recoveryState, recoveryReason));
+ }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default).Unwrap();
+ }
+
+ private bool TryAdmitTransportRecovery(Endpoints endpoints, TransportRecoveryState recoveryState,
+ bool respectHeartbeatCooldown)
+ {
+ if (!IsTransportRecoveryStateActive(endpoints, recoveryState) || State.Running != _client.State
+ || _recoveringEndpoints.Contains(endpoints))
+ {
+ return false;
+ }
+
+ var now = Stopwatch.GetTimestamp();
+ var previous = recoveryState.LastRecoveryTimestamp;
+ if (respectHeartbeatCooldown && TransportRecoveryState.NoRecoveryTimestamp != previous
+ && Stopwatch.GetElapsedTime(previous, now) < TransportRecoveryCooldown)
+ {
+ // Keep failures accumulated during cooldown so the next failure after expiry can recover.
+ return false;
+ }
+
+ _recoveringEndpoints.Add(endpoints);
+ recoveryState.LastRecoveryTimestamp = now;
+ recoveryState.HeartbeatFailureAttempts = 0;
+ return true;
+ }
+
+ private void RecoverTransport(Endpoints endpoints, IRpcClient rpcClient, TransportRecoveryState recoveryState,
+ string reason)
+ {
+ try
+ {
+ lock (_transportRecoveryLock)
+ {
+ if (!IsTransportRecoveryStateActive(endpoints, recoveryState) || State.Running != _client.State)
+ {
+ return;
+ }
+ }
+
+ Logger.LogWarning($"Try to recover transport, endpoints={endpoints}, reason={reason}, " +
+ $"clientId={_client.GetClientId()}");
+ try
+ {
+ rpcClient.ResetTransport();
+ }
+ catch (Exception e)
+ {
+ Logger.LogWarning(e, $"Failed to reset transport while recovering, endpoints={endpoints}, " +
+ $"reason={reason}, clientId={_client.GetClientId()}");
+ }
+
+ lock (_transportRecoveryLock)
+ {
+ if (!IsTransportRecoveryStateActive(endpoints, recoveryState) || State.Running != _client.State)
+ {
+ return;
+ }
+ }
+
+ try
+ {
+ // Never call Client's session-writing operations while holding the recovery lock.
+ _client.ReconnectTelemetry(endpoints, recoveryState.Session);
+ }
+ catch (Exception e)
+ {
+ Logger.LogWarning(e, $"Failed to rebuild telemetry while recovering transport, " +
+ $"endpoints={endpoints}, reason={reason}, clientId={_client.GetClientId()}");
+ }
+ }
+ finally
+ {
+ lock (_transportRecoveryLock)
+ {
+ _recoveringEndpoints.Remove(endpoints);
+ }
+ }
+ }
+
+ /// <summary>
+ /// Maps a transport level RESOURCE_EXHAUSTED failure onto the throttling exception of the SDK so that the
+ /// retry and the backoff logic of the callers can recognize it. Every other outcome is passed through.
+ /// </summary>
+ internal static async Task<TResponse> NormalizeTransportException<TResponse>(grpcLib.Metadata metadata,
+ Task<TResponse> task)
+ {
+ try
+ {
+ return await task.ConfigureAwait(false);
+ }
+ catch (grpcLib.RpcException e) when (grpcLib.StatusCode.ResourceExhausted == e.StatusCode)
+ {
+ var requestId = metadata?.GetValue(MetadataConstants.RequestIdKey);
+ var description = string.IsNullOrEmpty(e.Status.Detail) ? e.Message : e.Status.Detail;
+ throw new TooManyRequestsException((int)Proto.Code.TooManyRequests, requestId, description, e);
+ }
+ }
+
+ public async Task Shutdown()
+ {
+ lock (_transportRecoveryLock)
+ {
+ _stopped = true;
+ foreach (var recoveryState in _transportRecoveryStates.Values)
+ {
+ recoveryState.Heartbeats.Clear();
+ }
- await Task.WhenAll(tasks);
+ _transportRecoveryStates.Clear();
+ }
+
+ List<IRpcClient> rpcClients;
+ _clientLock.EnterReadLock();
+ try
+ {
+ rpcClients = _rpcClients.Values.ToList();
}
finally
{
_clientLock.ExitReadLock();
}
+
+ await Task.WhenAll(rpcClients.Select(rpcClient => rpcClient.Shutdown())).ConfigureAwait(false);
}
public grpcLib::AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> Telemetry(
@@ -99,7 +406,8 @@ namespace Org.Apache.Rocketmq
Endpoints endpoints, Proto.QueryRouteRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).QueryRoute(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).QueryRoute(metadata, request, timeout));
return new RpcInvocation<Proto.QueryRouteRequest, Proto.QueryRouteResponse>(request, response, metadata);
}
@@ -107,7 +415,11 @@ namespace Org.Apache.Rocketmq
Proto.HeartbeatRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).Heartbeat(metadata, request, timeout);
+ var rpcClient = GetRpcClient(endpoints);
+ var recoveryState = GetTransportRecoveryState(endpoints);
+ var future = rpcClient.Heartbeat(metadata, request, timeout);
+ _ = MonitorHeartbeat(endpoints, rpcClient, recoveryState, future);
+ var response = await NormalizeTransportException(metadata, future);
return new RpcInvocation<Proto.HeartbeatRequest, Proto.HeartbeatResponse>(request, response, metadata);
}
@@ -115,7 +427,8 @@ namespace Org.Apache.Rocketmq
NotifyClientTermination(Endpoints endpoints, Proto.NotifyClientTerminationRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).NotifyClientTermination(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).NotifyClientTermination(metadata, request, timeout));
return new RpcInvocation<Proto.NotifyClientTerminationRequest, Proto.NotifyClientTerminationResponse>(
request, response, metadata);
}
@@ -124,7 +437,8 @@ namespace Org.Apache.Rocketmq
RecallMessage(Endpoints endpoints, Proto.RecallMessageRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).RecallMessage(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).RecallMessage(metadata, request, timeout));
return new RpcInvocation<Proto.RecallMessageRequest, Proto.RecallMessageResponse>(
request, response, metadata);
}
@@ -133,7 +447,8 @@ namespace Org.Apache.Rocketmq
Endpoints endpoints, Proto::SendMessageRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).SendMessage(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).SendMessage(metadata, request, timeout));
return new RpcInvocation<Proto.SendMessageRequest, Proto.SendMessageResponse>(
request, response, metadata);
}
@@ -142,7 +457,8 @@ namespace Org.Apache.Rocketmq
Endpoints endpoints, Proto.QueryAssignmentRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).QueryAssignment(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).QueryAssignment(metadata, request, timeout));
return new RpcInvocation<Proto.QueryAssignmentRequest, Proto.QueryAssignmentResponse>(
request, response, metadata);
}
@@ -151,7 +467,8 @@ namespace Org.Apache.Rocketmq
ReceiveMessage(Endpoints endpoints, Proto.ReceiveMessageRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).ReceiveMessage(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).ReceiveMessage(metadata, request, timeout));
return new RpcInvocation<Proto.ReceiveMessageRequest, List<Proto.ReceiveMessageResponse>>(
request, response, metadata);
@@ -161,7 +478,8 @@ namespace Org.Apache.Rocketmq
Endpoints endpoints, Proto.AckMessageRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).AckMessage(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).AckMessage(metadata, request, timeout));
return new RpcInvocation<Proto.AckMessageRequest, Proto.AckMessageResponse>(
request, response, metadata);
}
@@ -171,7 +489,8 @@ namespace Org.Apache.Rocketmq
Proto.ChangeInvisibleDurationRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).ChangeInvisibleDuration(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).ChangeInvisibleDuration(metadata, request, timeout));
return new RpcInvocation<Proto.ChangeInvisibleDurationRequest, Proto.ChangeInvisibleDurationResponse>(
request, response, metadata);
}
@@ -181,7 +500,8 @@ namespace Org.Apache.Rocketmq
Proto.ForwardMessageToDeadLetterQueueRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).ForwardMessageToDeadLetterQueue(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).ForwardMessageToDeadLetterQueue(metadata, request, timeout));
return new RpcInvocation<Proto.ForwardMessageToDeadLetterQueueRequest, Proto.ForwardMessageToDeadLetterQueueResponse>(
request, response, metadata);
}
@@ -190,7 +510,8 @@ namespace Org.Apache.Rocketmq
Endpoints endpoints, Proto.EndTransactionRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).EndTransaction(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).EndTransaction(metadata, request, timeout));
return new RpcInvocation<Proto.EndTransactionRequest, Proto.EndTransactionResponse>(
request, response, metadata);
}
@@ -199,9 +520,25 @@ namespace Org.Apache.Rocketmq
Endpoints endpoints, Proto.SyncLiteSubscriptionRequest request, TimeSpan timeout)
{
var metadata = _client.Sign();
- var response = await GetRpcClient(endpoints).SyncLiteSubscription(metadata, request, timeout);
+ var response = await NormalizeTransportException(metadata,
+ GetRpcClient(endpoints).SyncLiteSubscription(metadata, request, timeout));
return new RpcInvocation<Proto.SyncLiteSubscriptionRequest, Proto.SyncLiteSubscriptionResponse>(
request, response, metadata);
}
+
+ private sealed class TransportRecoveryState
+ {
+ internal const long NoRecoveryTimestamp = long.MinValue;
+
+ internal readonly Session Session;
+ internal readonly Queue<Task<Proto.HeartbeatResponse>> Heartbeats = new Queue<Task<Proto.HeartbeatResponse>>();
+ internal int HeartbeatFailureAttempts;
+ internal long LastRecoveryTimestamp = NoRecoveryTimestamp;
+
+ internal TransportRecoveryState(Session session)
+ {
+ Session = session;
+ }
+ }
}
-}
\ No newline at end of file
+}
diff --git a/csharp/rocketmq-client-csharp/Error/ClientException.cs b/csharp/rocketmq-client-csharp/Error/ClientException.cs
index 78063531..ec6f92b0 100644
--- a/csharp/rocketmq-client-csharp/Error/ClientException.cs
+++ b/csharp/rocketmq-client-csharp/Error/ClientException.cs
@@ -37,6 +37,11 @@ namespace Org.Apache.Rocketmq.Error
{
}
+ protected ClientException(int responseCode, string requestId, string message, Exception innerException) : base(
+ $"[{RequestIdKey}={requestId}, {ResponseCodeKey}={responseCode}] {message}", innerException)
+ {
+ }
+
protected ClientException(int responseCode, string message) : base(
$"[{ResponseCodeKey}={responseCode}] {message}")
{
diff --git a/csharp/rocketmq-client-csharp/Error/TooManyRequestsException.cs b/csharp/rocketmq-client-csharp/Error/TooManyRequestsException.cs
index c8f0655b..c632ec46 100644
--- a/csharp/rocketmq-client-csharp/Error/TooManyRequestsException.cs
+++ b/csharp/rocketmq-client-csharp/Error/TooManyRequestsException.cs
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+using System;
+
namespace Org.Apache.Rocketmq.Error
{
/// <summary>
@@ -27,6 +29,11 @@ namespace Org.Apache.Rocketmq.Error
{
}
+ public TooManyRequestsException(int responseCode, string requestId, string message, Exception innerException)
+ : base(responseCode, requestId, message, innerException)
+ {
+ }
+
public TooManyRequestsException(int responseCode, string message) : base(responseCode, message)
{
}
diff --git a/csharp/rocketmq-client-csharp/IClientManager.cs b/csharp/rocketmq-client-csharp/IClientManager.cs
index 60177eb6..9f178d06 100644
--- a/csharp/rocketmq-client-csharp/IClientManager.cs
+++ b/csharp/rocketmq-client-csharp/IClientManager.cs
@@ -25,6 +25,12 @@ namespace Org.Apache.Rocketmq
{
public interface IClientManager
{
+ /// <summary>
+ /// Reconnect to the specified endpoints, replacing the gRPC transport and rebuilding the telemetry stream.
+ /// </summary>
+ /// <param name="endpoints">The target endpoints.</param>
+ void Reconnect(Endpoints endpoints);
+
/// <summary>
/// Establish a telemetry channel between client and remote endpoints.
/// </summary>
diff --git a/csharp/rocketmq-client-csharp/IRpcClient.cs b/csharp/rocketmq-client-csharp/IRpcClient.cs
index e4206395..b9233526 100644
--- a/csharp/rocketmq-client-csharp/IRpcClient.cs
+++ b/csharp/rocketmq-client-csharp/IRpcClient.cs
@@ -25,6 +25,16 @@ namespace Org.Apache.Rocketmq
{
public interface IRpcClient
{
+ /// <summary>
+ /// Connectivity state of the underlying gRPC channel.
+ /// </summary>
+ ConnectivityState State { get; }
+
+ /// <summary>
+ /// Replaces the gRPC channel of this client, dropping the connections it holds.
+ /// </summary>
+ void ResetTransport();
+
AsyncDuplexStreamingCall<TelemetryCommand, TelemetryCommand> Telemetry(Metadata metadata);
Task<QueryRouteResponse> QueryRoute(Metadata metadata, QueryRouteRequest request, TimeSpan timeout);
diff --git a/csharp/rocketmq-client-csharp/PushConsumer.cs b/csharp/rocketmq-client-csharp/PushConsumer.cs
index 94eeeaef..a7cf44da 100644
--- a/csharp/rocketmq-client-csharp/PushConsumer.cs
+++ b/csharp/rocketmq-client-csharp/PushConsumer.cs
@@ -589,15 +589,19 @@ namespace Org.Apache.Rocketmq
{
}
- internal override async void OnVerifyMessageCommand(Endpoints endpoints, VerifyMessageCommand command)
+ internal override async Task OnVerifyMessageCommand(Endpoints endpoints, VerifyMessageCommand command)
{
- var nonce = command.Nonce;
- var messageView = MessageView.FromProtobuf(command.Message);
- var messageId = messageView.MessageId;
Proto.TelemetryCommand telemetryCommand = null;
-
try
{
+ var session = GetSessionIfPresent(endpoints);
+ if (null == session)
+ {
+ return;
+ }
+
+ var nonce = command.Nonce;
+ var messageView = MessageView.FromProtobuf(command.Message);
var consumeResult = await _consumeService.Consume(messageView);
var code = consumeResult == ConsumeResult.SUCCESS ? Code.Ok : Code.FailedToConsumeMessage;
var status = new Status
@@ -613,13 +617,12 @@ namespace Org.Apache.Rocketmq
VerifyMessageResult = verifyMessageResult,
Status = status
};
- var (_, session) = GetSession(endpoints);
await session.WriteAsync(telemetryCommand);
}
catch (Exception e)
{
Logger.LogError(e,
- $"Failed to send message verification result command, endpoints={Endpoints}, command={telemetryCommand}, messageId={messageId}, clientId={ClientId}");
+ $"Failed to send message verification result command, endpoints={endpoints}, command={telemetryCommand}, clientId={ClientId}");
}
}
diff --git a/csharp/rocketmq-client-csharp/RpcClient.cs b/csharp/rocketmq-client-csharp/RpcClient.cs
index 0001ad7f..06ae6965 100644
--- a/csharp/rocketmq-client-csharp/RpcClient.cs
+++ b/csharp/rocketmq-client-csharp/RpcClient.cs
@@ -32,21 +32,82 @@ namespace Org.Apache.Rocketmq
public class RpcClient : IRpcClient
{
private static readonly ILogger Logger = MqLogManager.CreateLogger<RpcClient>();
- private readonly Proto::MessagingService.MessagingServiceClient _stub;
- private readonly GrpcChannel _channel;
private readonly string _target;
+ private readonly object _transportLock = new object();
+ private volatile GrpcChannel _channel;
+ private volatile Proto::MessagingService.MessagingServiceClient _stub;
public RpcClient(Endpoints endpoints, bool sslEnabled)
{
_target = endpoints.GrpcTarget(sslEnabled);
- _channel = GrpcChannel.ForAddress(_target, new GrpcChannelOptions
+ _channel = CreateChannel(_target);
+ _stub = CreateStub(_channel);
+ }
+
+ /// <summary>
+ /// Connectivity state of the underlying gRPC channel. <see cref="ConnectivityState.Idle"/> is returned when the
+ /// state is not resolvable, which keeps the native reconnection of gRPC in control instead of forcing a reset.
+ /// </summary>
+ public ConnectivityState State
+ {
+ get
+ {
+ try
+ {
+ return _channel.State;
+ }
+ catch (Exception e)
+ {
+ Logger.LogWarning(e, $"Failed to resolve the connectivity state, target={_target}");
+ return ConnectivityState.Idle;
+ }
+ }
+ }
+
+ /// <summary>
+ /// Replaces the gRPC channel of this client, dropping the connections it holds. Every in-flight call of the
+ /// replaced channel is cancelled, including the long-lived telemetry stream.
+ /// </summary>
+ public void ResetTransport()
+ {
+ GrpcChannel previous;
+ lock (_transportLock)
+ {
+ var channel = CreateChannel(_target);
+ var stub = CreateStub(channel);
+ previous = _channel;
+ _channel = channel;
+ _stub = stub;
+ }
+
+ // Dispose outside of the lock, the disposal cancels every in-flight call of the replaced channel.
+ try
+ {
+ previous?.Dispose();
+ }
+ catch (Exception e)
{
- HttpHandler = CreateHttpHandler(),
+ Logger.LogWarning(e, $"Failed to dispose the replaced gRPC channel, target={_target}");
+ }
+ }
+
+ private static GrpcChannel CreateChannel(string target)
+ {
+ return GrpcChannel.ForAddress(target, new GrpcChannelOptions
+ {
+ HttpHandler = CreateGrpcHttpHandler(),
+ // Dispose the handler together with the channel, otherwise replacing the transport would leave the
+ // previous handler and the connections it pools alive.
+ DisposeHttpClient = true,
// Disable auto-retry.
MaxRetryAttempts = 0
});
- var invoker = _channel.Intercept(new ClientLoggerInterceptor());
- _stub = new Proto::MessagingService.MessagingServiceClient(invoker);
+ }
+
+ private static Proto::MessagingService.MessagingServiceClient CreateStub(GrpcChannel channel)
+ {
+ var invoker = channel.Intercept(new ClientLoggerInterceptor());
+ return new Proto::MessagingService.MessagingServiceClient(invoker);
}
public async Task Shutdown()
@@ -73,6 +134,13 @@ namespace Org.Apache.Rocketmq
return handler;
}
+ internal static SocketsHttpHandler CreateGrpcHttpHandler()
+ {
+ var handler = new SocketsHttpHandler();
+ handler.SslOptions.RemoteCertificateValidationCallback = CertValidator;
+ return handler;
+ }
+
public AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> Telemetry(Metadata metadata)
{
var deadline = DateTime.UtcNow.Add(TimeSpan.FromDays(3650));
diff --git a/csharp/rocketmq-client-csharp/Session.cs b/csharp/rocketmq-client-csharp/Session.cs
index 2d898ad4..0ac6c6b7 100644
--- a/csharp/rocketmq-client-csharp/Session.cs
+++ b/csharp/rocketmq-client-csharp/Session.cs
@@ -30,14 +30,26 @@ namespace Org.Apache.Rocketmq
private static readonly ILogger Logger = MqLogManager.CreateLogger<Session>();
private static readonly TimeSpan SettingsInitializationTimeout = TimeSpan.FromSeconds(3);
- private readonly ManualResetEventSlim _event = new ManualResetEventSlim(false);
+ private static readonly TimeSpan StreamingCallRenewBackoffDelay = TimeSpan.FromSeconds(1);
+
+ // A healthy writer holds the stream lock for the duration of a single small command only, so exceeding this
+ // means the write is stuck on an unusable connection and will never complete on its own.
+ internal static readonly TimeSpan CloseTimeout = TimeSpan.FromSeconds(3);
- private readonly AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand>
- _streamingCall;
+ private readonly ManualResetEventSlim _event = new ManualResetEventSlim(false);
private readonly Client _client;
private readonly Endpoints _endpoints;
private readonly SemaphoreSlim _semaphore;
+ private readonly Lazy<Task> _initialization;
+
+ // Serializes the writers against the replacement of the streaming call. It is never held while acquiring
+ // _semaphore, and _semaphore is held across a blocking wait, so the two must never be nested the other way.
+ private readonly SemaphoreSlim _streamLock = new SemaphoreSlim(1);
+
+ private AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> _streamingCall;
+ private int _closed;
+ private int _reconnecting;
public Session(Endpoints endpoints,
AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> streamingCall,
@@ -47,29 +59,63 @@ namespace Org.Apache.Rocketmq
_semaphore = new SemaphoreSlim(1);
_streamingCall = streamingCall;
_client = client;
- Loop();
+ _initialization = new Lazy<Task>(() => SyncSettings(true));
+ StartReadLoop(streamingCall);
+ }
+
+ internal Task InitializeAsync()
+ {
+ return _initialization.Value;
+ }
+
+ public Task WriteAsync(Proto.TelemetryCommand telemetryCommand)
+ {
+ return WriteAsync(telemetryCommand, null);
+ }
+
+ private async Task WriteAsync(Proto.TelemetryCommand telemetryCommand,
+ AsyncDuplexStreamingCall<Proto.TelemetryCommand, Proto.TelemetryCommand> expectedCall)
+ {
+ await _streamLock.WaitAsync();
+ try
+ {
+ var streamingCall = Volatile.Read(ref _streamingCall);
+ if (1 == Volatile.Read(ref _closed) || null == streamingCall
+ || (null != expectedCall && !ReferenceEquals(streamingCall, expectedCall)))
+ {
+ return;
+ }
+
+ await streamingCall.RequestStream.WriteAsync(telemetryCommand);
+ }
+ finally
+ {
+ _streamLock.Release();
+ }
}
- public async Task WriteAsync(Proto.TelemetryCommand telemetryCommand)
+ public Task SyncSettings(bool awaitResp)
{
- var writer = _streamingCall.RequestStream;
- await writer.WriteAsync(telemetryCommand);
+ return SyncSettings(awaitResp, null);
}
- // TODO: Test concurrency.
- public async Task SyncSettings(bool awaitResp)
+ private async Task SyncSettings(bool awaitResp,
+ AsyncDuplexStreamingCall<Proto.TelemetryCommand, Proto.TelemetryCommand> expectedCall)
{
- // Add more buffer time.
await _semaphore.WaitAsync();
try
{
- var settings = _client.GetSettings();
+ if (1 == Volatile.Read(ref _closed)
+ || (null != expectedCall && !ReferenceEquals(Volatile.Read(ref _streamingCall), expectedCall)))
+ {
+ return;
+ }
+
var telemetryCommand = new Proto.TelemetryCommand
{
- Settings = settings.ToProtobuf()
+ Settings = _client.GetSettings().ToProtobuf()
};
- await WriteAsync(telemetryCommand);
- // await writer.CompleteAsync();
+ await WriteAsync(telemetryCommand, expectedCall);
if (awaitResp)
{
_event.Wait(_client.GetClientConfig().RequestTimeout.Add(SettingsInitializationTimeout));
@@ -81,60 +127,293 @@ namespace Org.Apache.Rocketmq
}
}
- private void Loop()
+ internal void Reconnect()
{
- Task.Run(async () =>
+ TryReconnect(null);
+ }
+
+ private bool TryReconnect(
+ AsyncDuplexStreamingCall<Proto.TelemetryCommand, Proto.TelemetryCommand> expectedCall)
+ {
+ if (0 != Interlocked.CompareExchange(ref _reconnecting, 1, 0))
{
- await foreach (var response in _streamingCall.ResponseStream.ReadAllAsync())
+ return false;
+ }
+
+ _ = ReconnectAsync(expectedCall);
+ return true;
+ }
+
+ private async Task ReconnectAsync(
+ AsyncDuplexStreamingCall<Proto.TelemetryCommand, Proto.TelemetryCommand> expectedCall)
+ {
+ AsyncDuplexStreamingCall<Proto.TelemetryCommand, Proto.TelemetryCommand> renewed = null;
+ try
+ {
+ if (null != expectedCall && !ReferenceEquals(Volatile.Read(ref _streamingCall), expectedCall))
{
- switch (response.CommandCase)
+ return;
+ }
+
+ while (null == renewed)
+ {
+ if (1 == Volatile.Read(ref _closed)
+ || (_client.State != State.Running && _client.State != State.Starting))
{
- case Proto.TelemetryCommand.CommandOneofCase.Settings:
- {
- Logger.LogInformation(
- $"Receive setting from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
- _client.OnSettingsCommand(_endpoints, response.Settings);
- _event.Set();
- break;
- }
- case Proto.TelemetryCommand.CommandOneofCase.RecoverOrphanedTransactionCommand:
- {
- Logger.LogInformation(
- $"Receive orphaned transaction recovery command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
- _client.OnRecoverOrphanedTransactionCommand(_endpoints,
- response.RecoverOrphanedTransactionCommand);
- break;
- }
- case Proto.TelemetryCommand.CommandOneofCase.VerifyMessageCommand:
- {
- Logger.LogInformation(
- $"Receive message verification command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
- _client.OnVerifyMessageCommand(_endpoints, response.VerifyMessageCommand);
- break;
- }
- case Proto.TelemetryCommand.CommandOneofCase.PrintThreadStackTraceCommand:
- {
- Logger.LogInformation(
- $"Receive thread stack print command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
- _client.OnPrintThreadStackTraceCommand(_endpoints, response.PrintThreadStackTraceCommand);
- break;
- }
- case Proto.TelemetryCommand.CommandOneofCase.NotifyUnsubscribeLiteCommand:
+ return;
+ }
+
+ if (_client.IsEndpointsDeprecated(_endpoints)
+ && await _client.RemoveSession(_endpoints, this))
+ {
+ return;
+ }
+
+ var previous = Volatile.Read(ref _streamingCall);
+ // A terminated reader can leave a write blocked; disposing the call releases its write lock.
+ CancelStreamingCall(previous);
+ try
+ {
+ var streamingCall = _client.GetClientManager().Telemetry(_endpoints);
+ await _streamLock.WaitAsync();
+ try
+ {
+ if (1 == Volatile.Read(ref _closed))
{
- Logger.LogInformation(
- $"Receive notify unsubscribe lite command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
- _client.OnNotifyUnsubscribeLiteCommand(_endpoints, response.NotifyUnsubscribeLiteCommand);
- break;
+ CancelStreamingCall(streamingCall);
+ return;
}
- default:
+
+ Interlocked.Exchange(ref _streamingCall, streamingCall);
+ if (1 == Volatile.Read(ref _closed))
{
- Logger.LogWarning(
- $"Receive unrecognized command from remote, endpoints={_endpoints}, command={response}, clientId={_client.GetClientId()}");
- break;
+ Interlocked.CompareExchange(ref _streamingCall, null, streamingCall);
+ CancelStreamingCall(streamingCall);
+ return;
}
+
+ renewed = streamingCall;
+ }
+ finally
+ {
+ _streamLock.Release();
+ }
+ }
+ catch (Exception e)
+ {
+ Logger.LogWarning(e, $"Failed to renew telemetry, endpoints={_endpoints}, " +
+ $"delay={StreamingCallRenewBackoffDelay}, clientId={_client.GetClientId()}");
+ await Task.Delay(StreamingCallRenewBackoffDelay);
}
}
+ }
+ catch (Exception e)
+ {
+ Logger.LogError(e, $"Failed to reconnect telemetry, endpoints={_endpoints}, " +
+ $"clientId={_client.GetClientId()}");
+ }
+ finally
+ {
+ Interlocked.Exchange(ref _reconnecting, 0);
+ }
+
+ if (null == renewed)
+ {
+ return;
+ }
+
+ // Release the renewal guard before the new reader can signal its own termination.
+ StartReadLoop(renewed);
+ try
+ {
+ await SyncSettings(false, renewed);
+ }
+ catch (Exception e)
+ {
+ Logger.LogWarning(e, $"Failed to sync settings on renewed telemetry, endpoints={_endpoints}, " +
+ $"clientId={_client.GetClientId()}");
+ }
+ }
+
+ internal void MarkClosed()
+ {
+ Interlocked.Exchange(ref _closed, 1);
+ _event.Set();
+ }
+
+ internal async Task CloseAsync()
+ {
+ MarkClosed();
+
+ if (await _streamLock.WaitAsync(CloseTimeout))
+ {
+ AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> streamingCall;
+ try
+ {
+ streamingCall = _streamingCall;
+ _streamingCall = null;
+ }
+ finally
+ {
+ _streamLock.Release();
+ }
+
+ CancelStreamingCall(streamingCall);
+ return;
+ }
+
+ Logger.LogWarning($"A writer holds the streaming call, force closing it, endpoints={_endpoints}, " +
+ $"timeout={CloseTimeout}, clientId={_client.GetClientId()}");
+ // Disposing the call is what releases the stuck writer, its await faults and drops the lock afterwards.
+ CancelStreamingCall(Interlocked.Exchange(ref _streamingCall, null));
+ }
+
+ private void CancelStreamingCall(
+ AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> streamingCall)
+ {
+ if (null == streamingCall)
+ {
+ return;
+ }
+
+ try
+ {
+ streamingCall.Dispose();
+ }
+ catch (Exception e)
+ {
+ Logger.LogWarning(e, $"Failed to cancel the replaced streaming call, endpoints={_endpoints}, " +
+ $"clientId={_client.GetClientId()}");
+ }
+ }
+
+ private void StartReadLoop(
+ AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> streamingCall)
+ {
+ Task.Run(async () =>
+ {
+ try
+ {
+ await foreach (var response in streamingCall.ResponseStream.ReadAllAsync())
+ {
+ if (1 == Volatile.Read(ref _closed)
+ || !ReferenceEquals(Volatile.Read(ref _streamingCall), streamingCall))
+ {
+ return;
+ }
+
+ switch (response.CommandCase)
+ {
+ case Proto.TelemetryCommand.CommandOneofCase.Settings:
+ {
+ Logger.LogInformation(
+ $"Receive setting from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
+ _client.OnSettingsCommand(_endpoints, response.Settings);
+ _event.Set();
+ break;
+ }
+ case Proto.TelemetryCommand.CommandOneofCase.RecoverOrphanedTransactionCommand:
+ {
+ Logger.LogInformation(
+ $"Receive orphaned transaction recovery command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
+ _client.OnRecoverOrphanedTransactionCommand(_endpoints,
+ response.RecoverOrphanedTransactionCommand);
+ break;
+ }
+ case Proto.TelemetryCommand.CommandOneofCase.VerifyMessageCommand:
+ {
+ Logger.LogInformation(
+ $"Receive message verification command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
+ _ = _client.OnVerifyMessageCommand(_endpoints, response.VerifyMessageCommand);
+ break;
+ }
+ case Proto.TelemetryCommand.CommandOneofCase.PrintThreadStackTraceCommand:
+ {
+ Logger.LogInformation(
+ $"Receive thread stack print command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
+ _ = _client.OnPrintThreadStackTraceCommand(_endpoints, response.PrintThreadStackTraceCommand);
+ break;
+ }
+ case Proto.TelemetryCommand.CommandOneofCase.ReconnectEndpointsCommand:
+ {
+ Logger.LogInformation(
+ $"Receive reconnect endpoints command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
+ _client.OnReconnectEndpointsCommand(_endpoints, response.ReconnectEndpointsCommand);
+ break;
+ }
+ case Proto.TelemetryCommand.CommandOneofCase.NotifyUnsubscribeLiteCommand:
+ {
+ Logger.LogInformation(
+ $"Receive notify unsubscribe lite command from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
+ _client.OnNotifyUnsubscribeLiteCommand(_endpoints, response.NotifyUnsubscribeLiteCommand);
+ break;
+ }
+ default:
+ {
+ Logger.LogWarning(
+ $"Receive unrecognized command from remote, endpoints={_endpoints}, command={response}, clientId={_client.GetClientId()}");
+ break;
+ }
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ // The loop of a replaced or closed session is cancelled on purpose.
+ if (ReferenceEquals(Volatile.Read(ref _streamingCall), streamingCall) && 0 == Volatile.Read(ref _closed))
+ {
+ Logger.LogError(e, $"Exception raised from the telemetry stream, endpoints={_endpoints}, " +
+ $"clientId={_client.GetClientId()}");
+ }
+ else
+ {
+ Logger.LogInformation($"Telemetry stream is cancelled for reconnecting, endpoints={_endpoints}, " +
+ $"clientId={_client.GetClientId()}");
+ }
+ }
+
+ try
+ {
+ await RenewAfterTerminationAsync(streamingCall);
+ }
+ catch (Exception e)
+ {
+ Logger.LogError(e, $"[Bug] unexpected exception raised while handling the termination of the " +
+ $"telemetry stream, endpoints={_endpoints}, clientId={_client.GetClientId()}");
+ }
});
}
- };
-}
\ No newline at end of file
+
+ private bool IsRenewalObsolete(
+ AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> streamingCall)
+ {
+ // The loop of a replaced or closed session ends on purpose, and a client which is not running must not be
+ // kept alive by a renewal of its telemetry stream.
+ return !ReferenceEquals(Volatile.Read(ref _streamingCall), streamingCall)
+ || 1 == Volatile.Read(ref _closed)
+ || (_client.State != State.Running && _client.State != State.Starting);
+ }
+
+ private async Task RenewAfterTerminationAsync(
+ AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand> streamingCall)
+ {
+ if (IsRenewalObsolete(streamingCall))
+ {
+ return;
+ }
+
+ Logger.LogWarning($"Telemetry stream terminated, attempt to renew it later, endpoints={_endpoints}, " +
+ $"delay={StreamingCallRenewBackoffDelay}, clientId={_client.GetClientId()}");
+ // A stale callback may briefly own the guard without renewing the current stream.
+ do
+ {
+ await Task.Delay(StreamingCallRenewBackoffDelay);
+ if (IsRenewalObsolete(streamingCall) || TryReconnect(streamingCall))
+ {
+ return;
+ }
+ }
+ while (!IsRenewalObsolete(streamingCall));
+ }
+ }
+}
diff --git a/csharp/tests/ClientManagerRpcExceptionTest.cs b/csharp/tests/ClientManagerRpcExceptionTest.cs
new file mode 100644
index 00000000..b443f271
--- /dev/null
+++ b/csharp/tests/ClientManagerRpcExceptionTest.cs
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System.Threading.Tasks;
+using Grpc.Core;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.Rocketmq;
+using Org.Apache.Rocketmq.Error;
+using Proto = Apache.Rocketmq.V2;
+
+namespace tests
+{
+ /// <summary>
+ /// Transport level RESOURCE_EXHAUSTED must reach the callers as the throttling exception of the SDK, so that the
+ /// producer takes its backoff path instead of retrying immediately.
+ /// </summary>
+ [TestClass]
+ public class ClientManagerRpcExceptionTest
+ {
+ private const string RequestId = "fake-request-id";
+
+ private static Metadata CreateMetadata()
+ {
+ return new Metadata { { MetadataConstants.RequestIdKey, RequestId } };
+ }
+
+ [TestMethod]
+ public async Task TestResourceExhaustedIsNormalized()
+ {
+ var rpcException = new RpcException(new Status(StatusCode.ResourceExhausted, "flow controlled"));
+ var task = Task.FromException<Proto.HeartbeatResponse>(rpcException);
+
+ try
+ {
+ await ClientManager.NormalizeTransportException(CreateMetadata(), task);
+ Assert.Fail("Expected the transport failure to be normalized");
+ }
+ catch (TooManyRequestsException e)
+ {
+ StringAssert.Contains(e.Message, "response-code=42900");
+ StringAssert.Contains(e.Message, $"request-id={RequestId}");
+ StringAssert.Contains(e.Message, "flow controlled");
+ Assert.AreSame(rpcException, e.InnerException);
+ }
+ }
+
+ [TestMethod]
+ public async Task TestResourceExhaustedWithoutDetailFallsBackToTheExceptionMessage()
+ {
+ var rpcException = new RpcException(new Status(StatusCode.ResourceExhausted, string.Empty));
+ var task = Task.FromException<Proto.HeartbeatResponse>(rpcException);
+
+ try
+ {
+ await ClientManager.NormalizeTransportException(CreateMetadata(), task);
+ Assert.Fail("Expected the transport failure to be normalized");
+ }
+ catch (TooManyRequestsException e)
+ {
+ StringAssert.Contains(e.Message, rpcException.Message);
+ Assert.AreSame(rpcException, e.InnerException);
+ }
+ }
+
+ [TestMethod]
+ public async Task TestOtherTransportExceptionIsUnchanged()
+ {
+ var rpcException = new RpcException(new Status(StatusCode.Unavailable, "connection refused"));
+ var task = Task.FromException<Proto.HeartbeatResponse>(rpcException);
+
+ try
+ {
+ await ClientManager.NormalizeTransportException(CreateMetadata(), task);
+ Assert.Fail("Expected the transport failure to be passed through");
+ }
+ catch (RpcException e)
+ {
+ Assert.AreSame(rpcException, e);
+ }
+ }
+
+ [TestMethod]
+ public async Task TestSuccessfulResponseIsUnchanged()
+ {
+ var response = new Proto.HeartbeatResponse();
+ var task = Task.FromResult(response);
+
+ Assert.AreSame(response, await ClientManager.NormalizeTransportException(CreateMetadata(), task));
+ }
+ }
+}
diff --git a/csharp/tests/ClientManagerTest.cs b/csharp/tests/ClientManagerTest.cs
index be3697d6..2c7f8cdf 100644
--- a/csharp/tests/ClientManagerTest.cs
+++ b/csharp/tests/ClientManagerTest.cs
@@ -16,19 +16,29 @@
*/
using System;
+using System.Collections;
using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
using Apache.Rocketmq.V2;
using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Moq;
using Org.Apache.Rocketmq;
using Endpoints = Org.Apache.Rocketmq.Endpoints;
+using grpcLib = Grpc.Core;
namespace tests
{
[TestClass]
public class ClientManagerTest
{
- private static readonly Endpoints FakeEndpoints = new Endpoints("127.0.0.1:8080");
- private static IClientManager _clientManager;
+ private static readonly Endpoints FakeEndpoints = RecoveryTestSupport.FakeEndpoints;
+ private ClientManager _clientManager;
+ private Mock<IRpcClient> _rpcClient;
private readonly ClientConfig _clientConfig = new ClientConfig.Builder()
.SetEndpoints("127.0.0.1:8080")
@@ -38,101 +48,852 @@ namespace tests
public void Initialize()
{
_clientManager = new ClientManager(CreateTestClient());
+ _rpcClient = new Mock<IRpcClient>();
+ _rpcClient.Setup(c => c.Shutdown()).Returns(Task.CompletedTask);
+ CacheRpcClient(_clientManager, _rpcClient.Object);
+ }
+
+ [TestCleanup]
+ public async Task Cleanup()
+ {
+ await _clientManager.Shutdown();
}
[TestMethod]
- public void TestHeartbeat()
+ public async Task TestHeartbeat()
{
var request = new HeartbeatRequest();
- _clientManager.Heartbeat(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.Heartbeat(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new HeartbeatResponse();
+ _rpcClient.Setup(c => c.Heartbeat(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.Heartbeat(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.Heartbeat(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestSendMessage()
+ public async Task TestSendMessage()
{
var request = new SendMessageRequest();
- _clientManager.SendMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.SendMessage(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new SendMessageResponse();
+ _rpcClient.Setup(c => c.SendMessage(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.SendMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.SendMessage(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestQueryAssignment()
+ public async Task TestQueryAssignment()
{
var request = new QueryAssignmentRequest();
- _clientManager.QueryAssignment(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.QueryAssignment(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new QueryAssignmentResponse();
+ _rpcClient.Setup(c => c.QueryAssignment(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.QueryAssignment(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.QueryAssignment(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestReceiveMessage()
+ public async Task TestReceiveMessage()
{
var request = new ReceiveMessageRequest();
- _clientManager.ReceiveMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.ReceiveMessage(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new List<ReceiveMessageResponse>();
+ _rpcClient.Setup(c => c.ReceiveMessage(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.ReceiveMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.ReceiveMessage(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestAckMessage()
+ public async Task TestAckMessage()
{
var request = new AckMessageRequest();
- _clientManager.AckMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.AckMessage(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new AckMessageResponse();
+ _rpcClient.Setup(c => c.AckMessage(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.AckMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.AckMessage(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestChangeInvisibleDuration()
+ public async Task TestChangeInvisibleDuration()
{
var request = new ChangeInvisibleDurationRequest();
- _clientManager.ChangeInvisibleDuration(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.ChangeInvisibleDuration(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new ChangeInvisibleDurationResponse();
+ _rpcClient.Setup(c => c.ChangeInvisibleDuration(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.ChangeInvisibleDuration(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.ChangeInvisibleDuration(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestForwardMessageToDeadLetterQueue()
+ public async Task TestForwardMessageToDeadLetterQueue()
{
var request = new ForwardMessageToDeadLetterQueueRequest();
- _clientManager.ForwardMessageToDeadLetterQueue(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.ForwardMessageToDeadLetterQueue(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new ForwardMessageToDeadLetterQueueResponse();
+ _rpcClient.Setup(c => c.ForwardMessageToDeadLetterQueue(It.IsAny<grpcLib.Metadata>(), request,
+ It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.ForwardMessageToDeadLetterQueue(FakeEndpoints, request,
+ TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.ForwardMessageToDeadLetterQueue(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestEndTransaction()
+ public async Task TestEndTransaction()
{
var request = new EndTransactionRequest();
- _clientManager.EndTransaction(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.EndTransaction(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new EndTransactionResponse();
+ _rpcClient.Setup(c => c.EndTransaction(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.EndTransaction(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.EndTransaction(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestNotifyClientTermination()
+ public async Task TestNotifyClientTermination()
{
var request = new NotifyClientTerminationRequest();
- _clientManager.NotifyClientTermination(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.NotifyClientTermination(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new NotifyClientTerminationResponse();
+ _rpcClient.Setup(c => c.NotifyClientTermination(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.NotifyClientTermination(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.NotifyClientTermination(null, request, TimeSpan.FromSeconds(1)));
}
[TestMethod]
- public void TestRecallMessage()
+ public async Task TestRecallMessage()
{
var request = new RecallMessageRequest();
- _clientManager.RecallMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
- _clientManager.RecallMessage(null, request, TimeSpan.FromSeconds(1));
- // Expect no exception thrown.
+ var response = new RecallMessageResponse();
+ _rpcClient.Setup(c => c.RecallMessage(It.IsAny<grpcLib.Metadata>(), request, It.IsAny<TimeSpan>()))
+ .ReturnsAsync(response);
+
+ var invocation = await _clientManager.RecallMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+ Assert.AreSame(response, invocation.Response);
+ await Assert.ThrowsExactlyAsync<ArgumentNullException>(() =>
+ _clientManager.RecallMessage(null, request, TimeSpan.FromSeconds(1)));
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatDeadlineExceededTriggersRecoveryAfterThreshold()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ rpcClient.Verify(c => c.ResetTransport(), Times.Once);
+ Assert.AreEqual(1, client.ReconnectTelemetryCalls);
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatSuccessResetsFailureAttempts()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ Task.FromResult(new HeartbeatResponse()));
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatUnavailableTriggersRecoveryWhenChannelIsReady()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.State).Returns(grpcLib.ConnectivityState.Ready);
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.Unavailable));
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Once);
+ Assert.AreEqual(1, client.ReconnectTelemetryCalls);
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatUnavailableDoesNotRecoverChannelWhichIsNotReady()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.State).Returns(grpcLib.ConnectivityState.TransientFailure);
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.Unavailable));
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatResourceExhaustedDoesNotTriggerRecovery()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.State).Returns(grpcLib.ConnectivityState.Ready);
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.ResourceExhausted));
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.ResourceExhausted));
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatRecoveryHasCooldown()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+
+ for (var i = 0; i < 2 * ClientManager.HeartbeatFailureThreshold; i++)
+ {
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ }
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Once);
+ Assert.AreEqual(1, client.ReconnectTelemetryCalls);
+ }
+
+ [TestMethod]
+ public async Task TestServerReconnectIgnoresHeartbeatCooldown()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+
+ for (var i = 0; i < ClientManager.HeartbeatFailureThreshold; i++)
+ {
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ }
+
+ clientManager.Reconnect(FakeEndpoints, rpcClient.Object);
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Exactly(2));
+ Assert.AreEqual(2, client.ReconnectTelemetryCalls);
+ }
+
+ [TestMethod]
+ [Timeout(30000)]
+ public async Task TestConcurrentReconnectOnlyRecoversOnce()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ const int threadCount = 8;
+
+ var recoveryStarted =
+ new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var allowRecoveryToComplete =
+ new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ rpcClient.Setup(c => c.ResetTransport()).Callback(() =>
+ {
+ recoveryStarted.TrySetResult(true);
+ allowRecoveryToComplete.Task.GetAwaiter().GetResult();
+ });
+
+ // Hold the winner off the thread pool so slow worker injection cannot release it ahead of the contenders.
+ var firstRecovery = Task.Factory.StartNew(() => clientManager.Reconnect(FakeEndpoints, rpcClient.Object),
+ CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
+ var reconnects = Array.Empty<Task>();
+ try
+ {
+ await recoveryStarted.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ reconnects = Enumerable.Range(0, threadCount - 1)
+ .Select(_ => Task.Run(() => clientManager.Reconnect(FakeEndpoints, rpcClient.Object)))
+ .ToArray();
+ await Task.WhenAll(reconnects).WaitAsync(TimeSpan.FromSeconds(5));
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Once);
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ allowRecoveryToComplete.TrySetResult(true);
+ await firstRecovery.WaitAsync(TimeSpan.FromSeconds(5));
+ await Task.WhenAll(reconnects).WaitAsync(TimeSpan.FromSeconds(5));
+ await clientManager.Shutdown();
+ }
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Once);
+ Assert.AreEqual(1, client.ReconnectTelemetryCalls);
+ }
+
+ [TestMethod]
+ public async Task TestTransportRecoveryIsSkippedWhenClientIsNotRunning()
+ {
+ var client = CreateRecoveryTestClient();
+ client.State = Org.Apache.Rocketmq.State.Stopping;
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.State).Returns(grpcLib.ConnectivityState.Ready);
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.Unavailable));
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+ }
+
+ [TestMethod]
+ public void TestReconnectWithoutCachedRpcClientIsIgnored()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+
+ clientManager.Reconnect(FakeEndpoints);
+
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+ }
+
+ [TestMethod]
+ public void TestRecoveryReplacesTheTransportBeforeItRebuildsTheTelemetry()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+
+ var order = new List<string>();
+ rpcClient.Setup(c => c.ResetTransport()).Callback(() =>
+ {
+ lock (order)
+ {
+ order.Add("transport");
+ }
+ });
+ client.ReconnectTelemetryHook = _ =>
+ {
+ lock (order)
+ {
+ order.Add("telemetry");
+ }
+ };
+
+ clientManager.Reconnect(FakeEndpoints, rpcClient.Object);
+
+ // Disposing the replaced channel is what faults a writer stuck on a half-open connection, releasing the
+ // stream lock which the renewal then needs. Renewing first would let it stall behind that writer.
+ Assert.AreEqual("transport,telemetry", string.Join(",", order));
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatCallbacksDrainInRegistrationOrder()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ var first = new TaskCompletionSource<HeartbeatResponse>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var success = new TaskCompletionSource<HeartbeatResponse>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var last = new TaskCompletionSource<HeartbeatResponse>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var firstMonitor = clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object, first.Task);
+ var successMonitor = clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object, success.Task);
+ var lastMonitor = clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object, last.Task);
+ try
+ {
+ first.SetException(new grpcLib.RpcException(new grpcLib.Status(grpcLib.StatusCode.DeadlineExceeded, "first")));
+ await firstMonitor.WaitAsync(TimeSpan.FromSeconds(5));
+ last.SetException(new grpcLib.RpcException(new grpcLib.Status(grpcLib.StatusCode.DeadlineExceeded, "last")));
+ // Force the last callback to run before the success callback. Its result must stay behind success.
+ await lastMonitor.WaitAsync(TimeSpan.FromSeconds(5));
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+
+ success.SetResult(new HeartbeatResponse());
+ await successMonitor.WaitAsync(TimeSpan.FromSeconds(5));
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ first.TrySetResult(new HeartbeatResponse());
+ success.TrySetResult(new HeartbeatResponse());
+ last.TrySetResult(new HeartbeatResponse());
+ await Task.WhenAll(firstMonitor, successMonitor, lastMonitor).WaitAsync(TimeSpan.FromSeconds(5));
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestAlternatingCompletedHeartbeatsOnOnePoolCallerNeverRecover()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ try
+ {
+ await Task.Run(async () =>
+ {
+ var monitors = new List<Task>();
+ for (var i = 0; i < 1000; i++)
+ {
+ monitors.Add(clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded)));
+ monitors.Add(clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ Task.FromResult(new HeartbeatResponse())));
+ }
+
+ await Task.WhenAll(monitors);
+ }).WaitAsync(TimeSpan.FromSeconds(5));
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatRecoveryResumesAfterCooldownExpires()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ try
+ {
+ for (var i = 0; i < 2 * ClientManager.HeartbeatFailureThreshold; i++)
+ {
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ }
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Once);
+ ExpireRecoveryCooldown(clientManager);
+ // Failures accumulated during cooldown still count; one additional deadline is enough.
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+
+ rpcClient.Verify(c => c.ResetTransport(), Times.Exactly(2));
+ Assert.AreEqual(2, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ [DataRow(false)]
+ [DataRow(true)]
+ public async Task TestLateHeartbeatsCannotRecreateRetiredState(bool shutdown)
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.State).Returns(grpcLib.ConnectivityState.Ready);
+ var heartbeats = Enumerable.Range(0, 2).Select(_ =>
+ new TaskCompletionSource<HeartbeatResponse>(TaskCreationOptions.RunContinuationsAsynchronously)).ToArray();
+ var monitors = heartbeats.Select(source =>
+ clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object, source.Task)).ToArray();
+ try
+ {
+ Assert.AreEqual(1, RecoveryStateCount(clientManager));
+ if (shutdown)
+ {
+ await clientManager.Shutdown();
+ }
+ else
+ {
+ client.EndpointsDeprecated = true;
+ clientManager.PruneTransportRecoveryStates();
+ }
+
+ Assert.AreEqual(0, RecoveryStateCount(clientManager));
+ foreach (var heartbeat in heartbeats)
+ {
+ heartbeat.SetException(new grpcLib.RpcException(
+ new grpcLib.Status(grpcLib.StatusCode.DeadlineExceeded, "late heartbeat")));
+ }
+
+ await Task.WhenAll(monitors).WaitAsync(TimeSpan.FromSeconds(5));
+ // Neither a new monitor nor a server command may reopen recovery after shutdown/deprecation.
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.Unavailable));
+ clientManager.Reconnect(FakeEndpoints, rpcClient.Object);
+
+ Assert.AreEqual(0, RecoveryStateCount(clientManager));
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ foreach (var heartbeat in heartbeats)
+ {
+ heartbeat.TrySetResult(new HeartbeatResponse());
+ }
+
+ await Task.WhenAll(monitors).WaitAsync(TimeSpan.FromSeconds(5));
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ [DataRow(false)]
+ [DataRow(true)]
+ public async Task TestReaddedEndpointHasFreshCountersAndCooldownAndIgnoresOldCallbacks(bool recoverFirst)
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ var heartbeats = Enumerable.Range(0, 3).Select(_ =>
+ new TaskCompletionSource<HeartbeatResponse>(TaskCreationOptions.RunContinuationsAsynchronously)).ToArray();
+ var monitors = new List<Task>();
+ try
+ {
+ var initialFailures = recoverFirst ? ClientManager.HeartbeatFailureThreshold : 1;
+ for (var i = 0; i < initialFailures; i++)
+ {
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ }
+
+ var initialResets = recoverFirst ? 1 : 0;
+ var retiredState = RecoveryState(clientManager);
+ monitors.AddRange(heartbeats.Select(source =>
+ clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object, source.Task)));
+ client.EndpointsDeprecated = true;
+ clientManager.PruneTransportRecoveryStates();
+ Assert.AreEqual(0, RecoveryStateCount(clientManager));
+ client.EndpointsDeprecated = false;
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ var activeState = RecoveryState(clientManager);
+ Assert.AreNotSame(retiredState, activeState);
+ rpcClient.Verify(c => c.ResetTransport(), Times.Exactly(initialResets));
+
+ heartbeats[0].SetException(new grpcLib.RpcException(
+ new grpcLib.Status(grpcLib.StatusCode.DeadlineExceeded, "retired first")));
+ heartbeats[1].SetException(new grpcLib.RpcException(
+ new grpcLib.Status(grpcLib.StatusCode.DeadlineExceeded, "retired second")));
+ heartbeats[2].SetResult(new HeartbeatResponse());
+ await Task.WhenAll(monitors).WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.AreSame(activeState, RecoveryState(clientManager));
+ rpcClient.Verify(c => c.ResetTransport(), Times.Exactly(initialResets));
+
+ // The old success must not erase the new failure; the old cooldown must not suppress recovery.
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ rpcClient.Verify(c => c.ResetTransport(), Times.Exactly(initialResets + 1));
+ Assert.AreEqual(initialResets + 1, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ foreach (var heartbeat in heartbeats)
+ {
+ heartbeat.TrySetResult(new HeartbeatResponse());
+ }
+
+ await Task.WhenAll(monitors).WaitAsync(TimeSpan.FromSeconds(5));
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatCapturesRecoveryStateBeforeSending()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.Shutdown()).Returns(Task.CompletedTask);
+ CacheRpcClient(clientManager, rpcClient.Object);
+ rpcClient.Setup(c => c.Heartbeat(It.IsAny<grpcLib.Metadata>(), It.IsAny<HeartbeatRequest>(),
+ It.IsAny<TimeSpan>())).Returns(() =>
+ {
+ client.EndpointsDeprecated = true;
+ clientManager.PruneTransportRecoveryStates();
+ client.EndpointsDeprecated = false;
+ return RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded);
+ });
+ try
+ {
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ await Assert.ThrowsExactlyAsync<grpcLib.RpcException>(() =>
+ clientManager.Heartbeat(FakeEndpoints, new HeartbeatRequest(), TimeSpan.FromSeconds(1)));
+
+ Assert.AreEqual(0, RecoveryStateCount(clientManager),
+ "A heartbeat sent before retirement must not register against the re-added endpoint");
+ rpcClient.Verify(c => c.ResetTransport(), Times.Never);
+ }
+ finally
+ {
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestHeartbeatCallerDoesNotWaitForTransportRecovery()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.Shutdown()).Returns(Task.CompletedTask);
+ CacheRpcClient(clientManager, rpcClient.Object);
+ var started = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ rpcClient.Setup(c => c.ResetTransport()).Callback(() =>
+ {
+ started.TrySetResult(true);
+ release.Task.GetAwaiter().GetResult();
+ });
+ rpcClient.Setup(c => c.Heartbeat(It.IsAny<grpcLib.Metadata>(), It.IsAny<HeartbeatRequest>(),
+ It.IsAny<TimeSpan>())).Returns(() =>
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ try
+ {
+ for (var i = 0; i < ClientManager.HeartbeatFailureThreshold; i++)
+ {
+ await Assert.ThrowsExactlyAsync<grpcLib.RpcException>(() =>
+ clientManager.Heartbeat(FakeEndpoints, new HeartbeatRequest(), TimeSpan.FromSeconds(1))
+ .WaitAsync(TimeSpan.FromSeconds(5)));
+ }
+
+ await started.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+ release.SetResult(true);
+ await TestAwaiter.Until(() => !RecoveryInProgress(clientManager), "the background heartbeat recovery");
+ Assert.AreEqual(1, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ release.TrySetResult(true);
+ await TestAwaiter.Until(() => !RecoveryInProgress(clientManager), "the background recovery to exit");
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestRetiredBusyResetCannotOverlapReaddedEndpointOrRenewOldTelemetry()
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.State).Returns(grpcLib.ConnectivityState.Ready);
+ rpcClient.Setup(c => c.Shutdown()).Returns(Task.CompletedTask);
+ CacheRpcClient(clientManager, rpcClient.Object);
+ var started = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var resetCalls = 0;
+ rpcClient.Setup(c => c.ResetTransport()).Callback(() =>
+ {
+ if (Interlocked.Increment(ref resetCalls) == 1)
+ {
+ started.TrySetResult(true);
+ release.Task.GetAwaiter().GetResult();
+ }
+ });
+ var oldRecovery = clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.Unavailable));
+ try
+ {
+ await started.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ client.EndpointsDeprecated = true;
+ clientManager.PruneTransportRecoveryStates();
+ Assert.AreEqual(0, RecoveryStateCount(clientManager));
+ client.EndpointsDeprecated = false;
+
+ for (var i = 0; i < ClientManager.HeartbeatFailureThreshold; i++)
+ {
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ }
+
+ clientManager.Reconnect(FakeEndpoints);
+ Assert.AreEqual(1, Volatile.Read(ref resetCalls), "Retirement must not release the old reset's guard");
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+
+ release.SetResult(true);
+ await oldRecovery.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls, "The retired reset must not renew the new session");
+
+ await clientManager.MonitorHeartbeat(FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ Assert.AreEqual(2, Volatile.Read(ref resetCalls));
+ Assert.AreEqual(1, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ release.TrySetResult(true);
+ await oldRecovery.WaitAsync(TimeSpan.FromSeconds(5));
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ [DataRow(false)]
+ [DataRow(true)]
+ public async Task TestTransportRecoveryDoesNotRenewTelemetryAfterStop(bool shutdown)
+ {
+ var client = CreateRecoveryTestClient();
+ var clientManager = new ClientManager(client);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.ResetTransport()).Callback(() =>
+ {
+ if (shutdown)
+ {
+ clientManager.Shutdown().GetAwaiter().GetResult();
+ }
+ else
+ {
+ client.State = Org.Apache.Rocketmq.State.Stopping;
+ }
+ });
+ try
+ {
+ clientManager.Reconnect(FakeEndpoints, rpcClient.Object);
+ rpcClient.Verify(c => c.ResetTransport(), Times.Once);
+ Assert.AreEqual(0, client.ReconnectTelemetryCalls);
+ }
+ finally
+ {
+ await clientManager.Shutdown();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestShutdownInvokesAndAwaitsRpcShutdownOutsideClientLock()
+ {
+ var clientLock = PrivateField<ReaderWriterLockSlim>(_clientManager, "_clientLock");
+ var completion = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var heldDuringShutdown = false;
+ _rpcClient.Setup(c => c.Shutdown()).Returns(() =>
+ {
+ heldDuringShutdown = clientLock.IsReadLockHeld;
+ return completion.Task;
+ });
+ var shutdown = _clientManager.Shutdown();
+ try
+ {
+ Assert.IsFalse(shutdown.IsCompleted);
+ Assert.IsFalse(heldDuringShutdown, "RPC shutdown must be invoked outside the cache lock");
+ Assert.AreEqual(0, clientLock.CurrentReadCount, "The cache lock must not be held across await");
+ }
+ finally
+ {
+ completion.TrySetResult(true);
+ await shutdown.WaitAsync(TimeSpan.FromSeconds(5));
+ }
+ }
+
+ private static T PrivateField<T>(object instance, string name)
+ {
+ var field = instance.GetType().GetField(name, BindingFlags.Instance | BindingFlags.NonPublic);
+ Assert.IsNotNull(field, $"Missing field {name}");
+ return (T)field.GetValue(instance);
+ }
+
+ private static void CacheRpcClient(ClientManager clientManager, IRpcClient rpcClient)
+ {
+ var clientLock = PrivateField<ReaderWriterLockSlim>(clientManager, "_clientLock");
+ clientLock.EnterWriteLock();
+ try
+ {
+ PrivateField<Dictionary<Endpoints, IRpcClient>>(clientManager, "_rpcClients")[FakeEndpoints] = rpcClient;
+ }
+ finally
+ {
+ clientLock.ExitWriteLock();
+ }
+ }
+
+ private static int RecoveryStateCount(ClientManager clientManager)
+ {
+ lock (PrivateField<object>(clientManager, "_transportRecoveryLock"))
+ {
+ return PrivateField<IDictionary>(clientManager, "_transportRecoveryStates").Count;
+ }
+ }
+
+ private static object RecoveryState(ClientManager clientManager)
+ {
+ lock (PrivateField<object>(clientManager, "_transportRecoveryLock"))
+ {
+ return PrivateField<IDictionary>(clientManager, "_transportRecoveryStates")[FakeEndpoints];
+ }
+ }
+
+ private static bool RecoveryInProgress(ClientManager clientManager)
+ {
+ lock (PrivateField<object>(clientManager, "_transportRecoveryLock"))
+ {
+ return PrivateField<HashSet<Endpoints>>(clientManager, "_recoveringEndpoints").Contains(FakeEndpoints);
+ }
+ }
+
+ private static void ExpireRecoveryCooldown(ClientManager clientManager)
+ {
+ lock (PrivateField<object>(clientManager, "_transportRecoveryLock"))
+ {
+ var state = RecoveryState(clientManager);
+ Assert.IsNotNull(state);
+ var timestamp = state.GetType().GetField("LastRecoveryTimestamp",
+ BindingFlags.Instance | BindingFlags.NonPublic);
+ Assert.IsNotNull(timestamp);
+ var elapsed = ClientManager.TransportRecoveryCooldown + TimeSpan.FromSeconds(1);
+ timestamp.SetValue(state, Stopwatch.GetTimestamp() - (long)(elapsed.TotalSeconds * Stopwatch.Frequency));
+ }
}
private Client CreateTestClient()
{
return new Producer(_clientConfig, new ConcurrentDictionary<string, bool>(), 1, null);
}
+
+ private RecoveryTestClient CreateRecoveryTestClient()
+ {
+ return new RecoveryTestClient(_clientConfig);
+ }
}
}
diff --git a/csharp/tests/ClientTest.cs b/csharp/tests/ClientTest.cs
index e428137e..aadb2f17 100644
--- a/csharp/tests/ClientTest.cs
+++ b/csharp/tests/ClientTest.cs
@@ -17,9 +17,8 @@
using System;
using System.Collections.Concurrent;
-using System.Threading;
+using System.Linq;
using System.Threading.Tasks;
-using Apache.Rocketmq.V2;
using Grpc.Core;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Moq;
@@ -32,136 +31,344 @@ namespace tests
[TestClass]
public class ClientTest
{
+ private const string Topic = "testTopic";
+
[TestMethod]
- public async Task TestOnVerifyMessageCommand()
+ [DataRow(false)]
+ [DataRow(true)]
+ public async Task TestUnsupportedCommandsReplyOnExistingSession(bool verify)
{
- var testClient = CreateTestClient();
- var endpoints = new Endpoints("testEndpoints");
- var command = new VerifyMessageCommand { Nonce = "testNonce" };
-
- var mockCall = new AsyncDuplexStreamingCall<TelemetryCommand, TelemetryCommand>(
- new MockClientStreamWriter<TelemetryCommand>(),
- new MockAsyncStreamReader<TelemetryCommand>(),
- null,
- null,
- null,
- null);
- var mockClientManager = new Mock<IClientManager>();
- mockClientManager.Setup(cm => cm.Telemetry(endpoints)).Returns(mockCall);
+ var (client, manager, stream, endpoints) = await ArrangeClient();
+ try
+ {
+ await Reply(client, endpoints, verify);
+ var response = stream.WrittenCommand(1);
+ Assert.AreEqual(Proto.Code.Unsupported, response.Status.Code);
+ Assert.AreEqual("nonce", verify ? response.VerifyMessageResult.Nonce : response.ThreadStackTrace.Nonce);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ }
+ finally
+ {
+ await client.DisposeAsync();
+ }
+ }
- testClient.SetClientManager(mockClientManager.Object);
+ [TestMethod]
+ [DataRow(false)]
+ [DataRow(true)]
+ [Timeout(20000)]
+ public async Task TestCancelledCommandReplyDoesNotEscape(bool verify)
+ {
+ var (client, manager, stream, endpoints) = await ArrangeClient();
+ Task reply = null;
+ try
+ {
+ stream.Writer.WriteGate = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ reply = Reply(client, endpoints, verify);
+ await TestAwaiter.Until(() => stream.WrittenCommands == 2, "the command reply to block in the writer");
+ Assert.IsFalse(reply.IsCompleted);
- testClient.OnVerifyMessageCommand(endpoints, command);
+ await client.DisposeAsync();
+ await reply.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.AreEqual(1, stream.DisposeCalls);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ }
+ finally
+ {
+ await client.DisposeAsync();
+ if (reply != null)
+ {
+ await reply;
+ }
+ }
+ }
- mockClientManager.Verify(cm => cm.Telemetry(endpoints), Times.Once);
+ [TestMethod]
+ public async Task TestLateCommandsDoNotRecreateSessions()
+ {
+ var (client, manager, stream, endpoints) = await ArrangeClient();
+ await client.DisposeAsync();
+ await Reply(client, endpoints, false);
+ await Reply(client, endpoints, true);
+ client.ReconnectTelemetry(endpoints);
+ Assert.AreEqual(1, stream.DisposeCalls);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
}
[TestMethod]
public async Task TestOnTopicRouteDataFetchedFailure()
{
- var testClient = CreateTestClient();
- var endpoints = new Endpoints("testEndpoints");
- var mq = new Proto.MessageQueue
+ var client = CreateTestClient();
+ var route = CreateRoute();
+ var endpoints = route.MessageQueues.First().Broker.Endpoints;
+ var manager = new Mock<IClientManager>();
+ var failed = new FakeTelemetryCall();
+ var gate = new TaskCompletionSource<bool>();
+ gate.SetException(new RpcException(new Status(StatusCode.Unavailable, "settings write failed")));
+ failed.Writer.WriteGate = gate;
+ var renewed = new FakeTelemetryCall(true, SettingsCommand(client));
+ manager.SetupSequence(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(failed.Call).Returns(renewed.Call);
+ client.SetClientManager(manager.Object);
+ try
{
- Topic = new Proto::Resource
- {
- ResourceNamespace = "testNamespace",
- Name = "testTopic"
- },
- Id = 0,
- Permission = Proto.Permission.ReadWrite,
- Broker = new Proto::Broker
- {
- Name = "testBroker",
- Id = 0,
- Endpoints = new Proto.Endpoints
- {
- Scheme = Proto.AddressScheme.Ipv4,
- Addresses = { new Proto.Address { Host = "127.0.0.1", Port = 8080 } }
- }
- }
- };
- var topicRouteData = new TopicRouteData(new[] { mq });
-
- var mockCall = new AsyncDuplexStreamingCall<TelemetryCommand, TelemetryCommand>(
- new MockClientStreamWriter<TelemetryCommand>(),
- new MockAsyncStreamReader<TelemetryCommand>(),
- null,
- null,
- null,
- null);
- var mockClientManager = new Mock<IClientManager>();
- mockClientManager.Setup(cm => cm.Telemetry(endpoints)).Returns(mockCall);
+ await Assert.ThrowsExactlyAsync<RpcException>(() => client.OnTopicRouteDataFetched(Topic, route));
+ Assert.IsTrue(client.IsEndpointsDeprecated(endpoints));
+ Assert.AreEqual(1, failed.DisposeCalls);
- testClient.SetClientManager(mockClientManager.Object);
+ await client.OnTopicRouteDataFetched(Topic, route);
+ Assert.IsFalse(client.IsEndpointsDeprecated(endpoints));
+ Assert.IsNotNull(renewed.WrittenCommand(0).Settings);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Exactly(2));
+ }
+ finally
+ {
+ await client.DisposeAsync();
+ failed.Call.Dispose();
+ renewed.Call.Dispose();
+ }
+ }
+ [TestMethod]
+ [DataRow(State.Starting)]
+ [DataRow(State.Running)]
+ [Timeout(20000)]
+ public async Task TestInitializingEndpointCanRenewBeforeRoutePublication(State state)
+ {
+ var client = CreateTestClient();
+ client.State = state;
+ var route = CreateRoute();
+ var endpoints = route.MessageQueues.First().Broker.Endpoints;
+ var first = new FakeTelemetryCall(false);
+ var renewed = new FakeTelemetryCall(true, SettingsCommand(client));
+ var manager = new Mock<IClientManager>();
+ manager.SetupSequence(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(first.Call).Returns(renewed.Call);
+ client.SetClientManager(manager.Object);
try
{
- await testClient.OnTopicRouteDataFetched("testTopic", topicRouteData);
- Assert.Fail();
+ await Task.Run(() => client.OnTopicRouteDataFetched(Topic, route)).WaitAsync(TimeSpan.FromSeconds(10));
+ await TestAwaiter.Until(() => renewed.WrittenCommands == 1, "settings on the replacement stream");
+ Assert.IsFalse(client.IsEndpointsDeprecated(endpoints));
+ Assert.AreEqual(1, first.DisposeCalls);
+ await client.OnTopicRouteDataFetched(Topic, route);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Exactly(2));
}
- catch (Exception)
+ finally
{
- mockClientManager.Verify(cm => cm.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ await client.DisposeAsync();
+ first.Call.Dispose();
+ renewed.Call.Dispose();
}
}
[TestMethod]
- public async Task TestOnPrintThreadStackTraceCommand()
+ [Timeout(15000)]
+ public async Task TestConcurrentRoutesShareInitializationAndKeepEndpointLive()
{
- var testClient = CreateTestClient();
- var endpoints = new Endpoints("testEndpoints");
- var command = new PrintThreadStackTraceCommand { Nonce = "testNonce" };
- var mockCall = new AsyncDuplexStreamingCall<TelemetryCommand, TelemetryCommand>(
- new MockClientStreamWriter<TelemetryCommand>(),
- new MockAsyncStreamReader<TelemetryCommand>(),
- null,
- null,
- null,
- null);
-
- var mockClientManager = new Mock<IClientManager>();
- mockClientManager.Setup(cm => cm.Telemetry(endpoints)).Returns(mockCall);
+ var client = CreateTestClient();
+ var route = CreateRoute();
+ var endpoints = route.MessageQueues.First().Broker.Endpoints;
+ var gate = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var stream = new FakeTelemetryCall(true, SettingsCommand(client));
+ stream.Writer.WriteGate = gate;
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(stream.Call);
+ client.SetClientManager(manager.Object);
+ var first = client.OnTopicRouteDataFetched(Topic, route);
+ var second = client.OnTopicRouteDataFetched("anotherTopic", route);
+ try
+ {
+ await TestAwaiter.Until(() => stream.WrittenCommands == 1, "the shared initialization write");
+ Assert.IsFalse(first.IsCompleted);
+ Assert.IsFalse(second.IsCompleted);
+ await client.OnTopicRouteDataFetched(Topic, new TopicRouteData(Array.Empty<Proto.MessageQueue>()));
+ Assert.IsFalse(client.IsEndpointsDeprecated(endpoints));
+ Assert.AreEqual(0, stream.DisposeCalls);
- testClient.SetClientManager(mockClientManager.Object);
+ gate.SetResult(true);
+ await Task.WhenAll(first, second);
+ Assert.AreEqual(1, stream.WrittenCommands);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ }
+ finally
+ {
+ gate.TrySetResult(true);
+ await Task.WhenAll(first, second);
+ await client.DisposeAsync();
+ }
+ }
- // Act
- testClient.OnPrintThreadStackTraceCommand(endpoints, command);
+ [TestMethod]
+ public async Task TestLateRouteResponseCannotCreateSessionAfterShutdown()
+ {
+ var client = CreateTestClient();
+ var manager = new Mock<IClientManager>();
+ var responseSource = new TaskCompletionSource<RpcInvocation<Proto.QueryRouteRequest, Proto.QueryRouteResponse>>(
+ TaskCreationOptions.RunContinuationsAsynchronously);
+ manager.Setup(m => m.QueryRoute(It.IsAny<Endpoints>(), It.IsAny<Proto.QueryRouteRequest>(), It.IsAny<TimeSpan>()))
+ .Returns(responseSource.Task);
+ client.SetClientManager(manager.Object);
+ var pending = client.FetchRoute(Topic);
+ await client.DisposeAsync();
+ var response = new Proto.QueryRouteResponse { Status = new Proto.Status { Code = Proto.Code.Ok } };
+ response.MessageQueues.Add(CreateMessageQueue());
+ responseSource.SetResult(new RpcInvocation<Proto.QueryRouteRequest, Proto.QueryRouteResponse>(
+ new Proto.QueryRouteRequest(), response, new Metadata()));
- // Assert
- mockClientManager.Verify(cm => cm.Telemetry(endpoints), Times.Once);
+ await Assert.ThrowsExactlyAsync<ObjectDisposedException>(() => pending);
+ Assert.AreEqual(State.Terminated, client.State);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Never);
}
- private Client CreateTestClient()
+ [TestMethod]
+ public async Task TestShutdownCancelsInitializationAndPreventsRoutePublication()
{
- return new Producer(new ClientConfig.Builder().SetEndpoints("127.0.0.1:9876").Build(),
- new ConcurrentDictionary<string, bool>(), 1, null);
+ var client = CreateTestClient();
+ var route = CreateRoute();
+ var endpoints = route.MessageQueues.First().Broker.Endpoints;
+ var stream = new FakeTelemetryCall();
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(stream.Call);
+ client.SetClientManager(manager.Object);
+ var pending = Task.Run(() => client.OnTopicRouteDataFetched(Topic, route));
+ await TestAwaiter.Until(() => stream.WrittenCommands == 1, "initial settings to be waiting for a response");
+ await client.DisposeAsync();
+ await Assert.ThrowsExactlyAsync<ObjectDisposedException>(() => pending);
+ Assert.IsTrue(client.IsEndpointsDeprecated(endpoints));
+ Assert.AreEqual(1, stream.DisposeCalls);
}
- private class MockClientStreamWriter<T> : IClientStreamWriter<T>
+ [TestMethod]
+ public async Task TestOnReconnectEndpointsCommand()
{
- public Task WriteAsync(T message)
+ var client = CreateTestClient();
+ var manager = new Mock<IClientManager>();
+ client.SetClientManager(manager.Object);
+ var endpoints = RecoveryTestSupport.FakeEndpoints;
+ try
{
- // Simulate async operation
- return Task.CompletedTask;
+ client.OnReconnectEndpointsCommand(endpoints, new Proto.ReconnectEndpointsCommand { Nonce = "nonce" });
+ manager.Verify(m => m.Reconnect(endpoints), Times.Once);
}
+ finally
+ {
+ await client.DisposeAsync();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestReaddedSessionIgnoresOldRecoveryEvenWhenPruningIsDelayed()
+ {
+ var (client, manager, first, endpoints) = await ArrangeClient();
+ using var rejoined = new FakeTelemetryCall(true, SettingsCommand(client));
+ using var renewed = new FakeTelemetryCall(true, SettingsCommand(client));
+ manager.SetupSequence(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(rejoined.Call).Returns(renewed.Call);
+ var recovery = new ClientManager(client);
+ var rpc = new Mock<IRpcClient>();
+ var oldHeartbeat = new TaskCompletionSource<Proto.HeartbeatResponse>(TaskCreationOptions.RunContinuationsAsynchronously);
+ Task oldMonitor = null;
+ try
+ {
+ var oldSession = client.GetSessionIfPresent(endpoints);
+ await recovery.MonitorHeartbeat(endpoints, rpc.Object,
+ RecoveryTestSupport.FailedHeartbeat(StatusCode.DeadlineExceeded));
+ oldMonitor = recovery.MonitorHeartbeat(endpoints, rpc.Object, oldHeartbeat.Task);
+ await client.OnTopicRouteDataFetched(Topic, new TopicRouteData(Array.Empty<Proto.MessageQueue>()));
+ await client.OnTopicRouteDataFetched(Topic, CreateRoute());
+ Assert.AreNotSame(oldSession, client.GetSessionIfPresent(endpoints));
+ Assert.AreEqual(1, first.DisposeCalls);
+
+ client.ReconnectTelemetry(endpoints, oldSession);
+ Assert.AreEqual(0, rejoined.DisposeCalls);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Exactly(2));
- public WriteOptions WriteOptions { get; set; }
+ oldHeartbeat.SetException(new RpcException(new Status(StatusCode.DeadlineExceeded, "retired heartbeat")));
+ await oldMonitor;
+ await recovery.MonitorHeartbeat(endpoints, rpc.Object,
+ RecoveryTestSupport.FailedHeartbeat(StatusCode.DeadlineExceeded));
+ rpc.Verify(r => r.ResetTransport(), Times.Never);
- public Task CompleteAsync()
+ await recovery.MonitorHeartbeat(endpoints, rpc.Object,
+ RecoveryTestSupport.FailedHeartbeat(StatusCode.DeadlineExceeded));
+ await TestAwaiter.Until(() => renewed.WrittenCommands == 1, "recovery of the new session only");
+ rpc.Verify(r => r.ResetTransport(), Times.Once);
+ }
+ finally
{
- throw new NotImplementedException();
+ oldHeartbeat.TrySetResult(new Proto.HeartbeatResponse());
+ if (oldMonitor != null)
+ {
+ await oldMonitor;
+ }
+
+ await recovery.Shutdown();
+ await client.DisposeAsync();
}
}
- private class MockAsyncStreamReader<T> : IAsyncStreamReader<T>
+ private static Task Reply(Client client, Endpoints endpoints, bool verify)
+ {
+ return verify
+ ? client.OnVerifyMessageCommand(endpoints, new Proto.VerifyMessageCommand { Nonce = "nonce" })
+ : client.OnPrintThreadStackTraceCommand(endpoints, new Proto.PrintThreadStackTraceCommand { Nonce = "nonce" });
+ }
+
+ private static async Task<(RouteTestClient, Mock<IClientManager>, FakeTelemetryCall, Endpoints)> ArrangeClient()
+ {
+ var client = CreateTestClient();
+ var stream = new FakeTelemetryCall(true, SettingsCommand(client));
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(stream.Call);
+ client.SetClientManager(manager.Object);
+ var route = CreateRoute();
+ await client.OnTopicRouteDataFetched(Topic, route);
+ return (client, manager, stream, route.MessageQueues.First().Broker.Endpoints);
+ }
+
+ private static RouteTestClient CreateTestClient()
{
- public Task<bool> MoveNext(CancellationToken cancellationToken)
+ return new RouteTestClient();
+ }
+
+ private static Proto.TelemetryCommand SettingsCommand(Client client)
+ {
+ return new Proto.TelemetryCommand { Settings = client.GetSettings().ToProtobuf() };
+ }
+
+ private static TopicRouteData CreateRoute()
+ {
+ return new TopicRouteData(new[] { CreateMessageQueue() });
+ }
+
+ private static Proto.MessageQueue CreateMessageQueue()
+ {
+ return new Proto.MessageQueue
{
- throw new System.NotImplementedException();
+ Topic = new Proto.Resource { Name = Topic },
+ Id = 0,
+ Permission = Proto.Permission.ReadWrite,
+ Broker = new Proto.Broker
+ {
+ Name = "broker", Id = 0,
+ Endpoints = new Proto.Endpoints
+ {
+ Scheme = Proto.AddressScheme.Ipv4,
+ Addresses = { new Proto.Address { Host = "127.0.0.1", Port = 8080 } }
+ }
+ }
+ };
+ }
+
+ private sealed class RouteTestClient : Producer
+ {
+ internal RouteTestClient() : base(RecoveryTestSupport.CreateClientConfig(), new ConcurrentDictionary<string, bool>(), 1, null)
+ {
+ State = State.Running;
}
- public T Current => throw new NotImplementedException();
+ internal Task<TopicRouteData> FetchRoute(string topic)
+ {
+ return GetRouteData(topic);
+ }
}
}
}
diff --git a/csharp/tests/MockServer.cs b/csharp/tests/MockServer.cs
index 6f658420..02a89083 100644
--- a/csharp/tests/MockServer.cs
+++ b/csharp/tests/MockServer.cs
@@ -21,6 +21,7 @@ using System.Threading;
using System.Threading.Tasks;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
+using MessageIdGenerator = Org.Apache.Rocketmq.MessageIdGenerator;
using Proto = Apache.Rocketmq.V2;
namespace tests
@@ -87,6 +88,32 @@ namespace tests
return Task.FromResult(response);
}
+ public override Task<Proto.SendMessageResponse> SendMessage(Proto.SendMessageRequest request,
+ ServerCallContext context)
+ {
+ var response = new Proto.SendMessageResponse
+ {
+ Status = _mockStatus,
+ Entries =
+ {
+ new Proto.SendResultEntry
+ {
+ Status = _mockStatus,
+ MessageId = MessageIdGenerator.GetInstance().Next(),
+ Offset = 1
+ }
+ }
+ };
+ return Task.FromResult(response);
+ }
+
+ public override Task<Proto.NotifyClientTerminationResponse> NotifyClientTermination(
+ Proto.NotifyClientTerminationRequest request, ServerCallContext context)
+ {
+ var response = new Proto.NotifyClientTerminationResponse { Status = _mockStatus };
+ return Task.FromResult(response);
+ }
+
public override Task<Proto.QueryAssignmentResponse> QueryAssignment(Proto.QueryAssignmentRequest request,
ServerCallContext context)
{
diff --git a/csharp/tests/ProducerHalfOpenTcpRecoveryIntegrationTest.cs b/csharp/tests/ProducerHalfOpenTcpRecoveryIntegrationTest.cs
new file mode 100644
index 00000000..91f131ed
--- /dev/null
+++ b/csharp/tests/ProducerHalfOpenTcpRecoveryIntegrationTest.cs
@@ -0,0 +1,340 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.Rocketmq;
+
+namespace tests
+{
+ /// <summary>
+ /// A peer which disappears without sending either FIN or RST leaves the connection half-open: the transport still
+ /// considers it usable, so every request written onto it blocks until its own deadline expires. Recovery has to be
+ /// driven by the heartbeat of the client, which replaces the transport once it timed out twice in a row.
+ /// </summary>
+ [TestClass]
+ public class ProducerHalfOpenTcpRecoveryIntegrationTest : GrpcServerIntegrationTest
+ {
+ private const string Loopback = "127.0.0.1";
+ private const string Topic = "topic";
+ private const string Broker = "broker";
+
+ private static readonly TimeSpan MaxRecoveryTime = TimeSpan.FromSeconds(40);
+
+ private Server _server;
+ private MockServer _mockServer;
+
+ [TestInitialize]
+ public void SetUp()
+ {
+ _mockServer = new MockServer(Topic, Broker, new List<string>());
+ _server = SetUpServer(_mockServer);
+ _mockServer.Port = Port;
+ }
+
+ [TestCleanup]
+ public async Task TearDown()
+ {
+ await _server.ShutdownAsync().WaitAsync(TimeSpan.FromSeconds(10));
+ }
+
+ [TestMethod]
+ [Timeout(60000)]
+ public async Task TestProducerRecoversFromHalfOpenTcpAfterHeartbeatTimeouts()
+ {
+ var proxy = new BlackholeTcpProxy(Port);
+ Producer producer = null;
+ // Everything below can fail while the proxy is already listening, so the try has to cover it or the
+ // listener and its accept loop would outlive this test and disturb the ones running after it.
+ try
+ {
+ // The route data must advertise the proxy, otherwise the producer would talk to the server directly.
+ _mockServer.Port = proxy.Port;
+
+ var credentialsProvider = new StaticSessionCredentialsProvider("accessKey", "secretKey");
+ var clientConfig = new ClientConfig.Builder()
+ .SetEndpoints(Loopback + ":" + proxy.Port)
+ .SetCredentialsProvider(credentialsProvider)
+ .EnableSsl(false)
+ .SetRequestTimeout(TimeSpan.FromSeconds(3))
+ .Build();
+
+ producer = await new Producer.Builder()
+ .SetClientConfig(clientConfig)
+ .SetTopics(Topic)
+ .SetMaxAttempts(1)
+ .Build();
+
+ var message = new Message.Builder()
+ .SetTopic(Topic)
+ .SetBody(Encoding.UTF8.GetBytes("tcp-blackhole"))
+ .Build();
+
+ await producer.Send(message);
+ proxy.AssertHealthy();
+ var initialConnectionCount = proxy.AcceptedConnectionCount;
+ Assert.IsTrue(initialConnectionCount > 0);
+
+ proxy.BlackholeExistingConnections();
+ var blackholeStart = Stopwatch.GetTimestamp();
+
+ await AssertSendFailure(producer, message);
+ // No connection was torn down, the blackholed one is half-open rather than closed.
+ Assert.AreEqual(0, proxy.ClosedConnectionCount);
+
+ while (Stopwatch.GetElapsedTime(blackholeStart) < MaxRecoveryTime)
+ {
+ proxy.AssertHealthy();
+ try
+ {
+ await producer.Send(message);
+ }
+ catch (Exception)
+ {
+ // Keep sending until the heartbeat recovery replaces the blackholed connection.
+ continue;
+ }
+
+ Assert.IsTrue(proxy.AcceptedConnectionCount > initialConnectionCount,
+ "The producer recovered without establishing a new connection");
+ return;
+ }
+
+ proxy.AssertHealthy();
+ Assert.Fail($"Producer did not recover from the half-open TCP connection within {MaxRecoveryTime}");
+ }
+ finally
+ {
+ if (null != producer)
+ {
+ await producer.DisposeAsync();
+ }
+
+ proxy.Dispose();
+ }
+ }
+
+ private static async Task AssertSendFailure(Producer producer, Message message)
+ {
+ Exception failure = null;
+ try
+ {
+ await producer.Send(message);
+ }
+ catch (Exception e)
+ {
+ failure = e;
+ }
+
+ Assert.IsNotNull(failure, "Message should time out on the blackholed TCP connection");
+ }
+
+ /// <summary>
+ /// Forwards traffic to the backend until the connections established so far are blackholed. A blackholed
+ /// connection keeps being drained but nothing is forwarded anymore, so neither peer ever observes a FIN or a
+ /// RST and both sides keep the connection in their pool.
+ /// </summary>
+ private sealed class BlackholeTcpProxy : IDisposable
+ {
+ private readonly int _backendPort;
+ private readonly TcpListener _listener;
+ private readonly CancellationTokenSource _cts = new CancellationTokenSource();
+ private readonly ConcurrentBag<SocketPair> _connections = new ConcurrentBag<SocketPair>();
+
+ private int _acceptedConnectionCount;
+ private int _closedConnectionCount;
+ private int _blackholeThroughConnectionId;
+ private Exception _acceptFailure;
+
+ internal BlackholeTcpProxy(int backendPort)
+ {
+ _backendPort = backendPort;
+ _listener = new TcpListener(IPAddress.Loopback, 0);
+ _listener.Start();
+ Port = ((IPEndPoint)_listener.LocalEndpoint).Port;
+ _ = AcceptConnections();
+ }
+
+ internal int Port { get; }
+
+ internal int AcceptedConnectionCount => Volatile.Read(ref _acceptedConnectionCount);
+
+ internal int ClosedConnectionCount => Volatile.Read(ref _closedConnectionCount);
+
+ internal void BlackholeExistingConnections()
+ {
+ Interlocked.Exchange(ref _blackholeThroughConnectionId,
+ Volatile.Read(ref _acceptedConnectionCount));
+ }
+
+ /// <summary>
+ /// Surfaces a failure of the accept loop, which would otherwise make the producer hang instead of failing.
+ /// </summary>
+ internal void AssertHealthy()
+ {
+ var failure = Volatile.Read(ref _acceptFailure);
+ if (null == failure)
+ {
+ return;
+ }
+
+ throw new AssertFailedException("TCP proxy stopped accepting connections", failure);
+ }
+
+ private async Task AcceptConnections()
+ {
+ while (!_cts.IsCancellationRequested)
+ {
+ TcpClient downstream = null;
+ try
+ {
+ downstream = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false);
+ downstream.NoDelay = true;
+
+ var upstream = new TcpClient();
+ upstream.NoDelay = true;
+ await upstream.ConnectAsync(IPAddress.Loopback, _backendPort, _cts.Token).ConfigureAwait(false);
+
+ var connection = new SocketPair(this,
+ Interlocked.Increment(ref _acceptedConnectionCount), downstream, upstream);
+ _connections.Add(connection);
+ _ = Forward(connection, downstream, upstream);
+ _ = Forward(connection, upstream, downstream);
+ }
+ catch (Exception e)
+ {
+ if (!_cts.IsCancellationRequested)
+ {
+ Interlocked.CompareExchange(ref _acceptFailure, e, null);
+ }
+
+ downstream?.Dispose();
+ return;
+ }
+ }
+ }
+
+ private async Task Forward(SocketPair connection, TcpClient source, TcpClient destination)
+ {
+ var buffer = new byte[8192];
+ try
+ {
+ var input = source.GetStream();
+ var output = destination.GetStream();
+ while (!_cts.IsCancellationRequested)
+ {
+ var length = await input.ReadAsync(buffer.AsMemory(), _cts.Token).ConfigureAwait(false);
+ if (0 == length)
+ {
+ break;
+ }
+
+ if (connection.Id <= Volatile.Read(ref _blackholeThroughConnectionId))
+ {
+ continue;
+ }
+
+ await output.WriteAsync(buffer.AsMemory(0, length), _cts.Token).ConfigureAwait(false);
+ await output.FlushAsync(_cts.Token).ConfigureAwait(false);
+ }
+ }
+ catch (Exception)
+ {
+ // Socket closure is expected during transport recovery and test cleanup.
+ }
+ finally
+ {
+ connection.Close();
+ }
+ }
+
+ public void Dispose()
+ {
+ if (_cts.IsCancellationRequested)
+ {
+ return;
+ }
+
+ _cts.Cancel();
+ try
+ {
+ _listener.Stop();
+ }
+ catch (Exception)
+ {
+ // Ignore exception on purpose.
+ }
+
+ foreach (var connection in _connections)
+ {
+ connection.Close();
+ }
+ }
+
+ private sealed class SocketPair
+ {
+ private readonly BlackholeTcpProxy _proxy;
+ private readonly TcpClient _downstream;
+ private readonly TcpClient _upstream;
+ private int _closed;
+
+ internal SocketPair(BlackholeTcpProxy proxy, int id, TcpClient downstream, TcpClient upstream)
+ {
+ _proxy = proxy;
+ Id = id;
+ _downstream = downstream;
+ _upstream = upstream;
+ }
+
+ internal int Id { get; }
+
+ internal void Close()
+ {
+ if (0 != Interlocked.CompareExchange(ref _closed, 1, 0))
+ {
+ return;
+ }
+
+ Dispose(_downstream);
+ Dispose(_upstream);
+ Interlocked.Increment(ref _proxy._closedConnectionCount);
+ }
+
+ private static void Dispose(TcpClient socket)
+ {
+ try
+ {
+ socket.Dispose();
+ }
+ catch (Exception)
+ {
+ // Ignore exception on purpose.
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/csharp/tests/PushConsumerRecoveryTest.cs b/csharp/tests/PushConsumerRecoveryTest.cs
new file mode 100644
index 00000000..09aa8f7b
--- /dev/null
+++ b/csharp/tests/PushConsumerRecoveryTest.cs
@@ -0,0 +1,272 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+using System.Collections.Concurrent;
+using System.Threading.Tasks;
+using Google.Protobuf.WellKnownTypes;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Moq;
+using Org.Apache.Rocketmq;
+using Endpoints = Org.Apache.Rocketmq.Endpoints;
+using Proto = Apache.Rocketmq.V2;
+
+namespace tests
+{
+ /// <summary>
+ /// Consumer recovery must follow the subscription route: renew active sessions, retire obsolete ones, and create
+ /// a fresh session when the same endpoints return.
+ /// </summary>
+ [TestClass]
+ public class PushConsumerRecoveryTest
+ {
+ private const string Topic = "topic";
+ private const string Broker = "broker";
+ private const string TestConsumerGroup = "testGroup";
+
+ [TestMethod]
+ [Timeout(30000)]
+ public async Task TestPushConsumerRenewsTelemetryForEndpointsStillInItsRouteTable()
+ {
+ using var first = new FakeTelemetryCall(true, SettingsCommand());
+ using var renewed = new FakeTelemetryCall(true, SettingsCommand());
+ var consumer = CreateConsumer(out var clientManager, first, renewed);
+ try
+ {
+ var route = CreateTopicRouteData();
+ var endpoints = route.MessageQueues[0].Broker.Endpoints;
+ await consumer.OnTopicRouteDataFetched(Topic, route);
+ AssertSettingsWritten(first);
+ Assert.IsFalse(consumer.IsEndpointsDeprecated(endpoints));
+
+ consumer.ReconnectTelemetry(endpoints);
+
+ await TestAwaiter.Until(() => renewed.WrittenCommands > 0 && 1 == first.DisposeCalls,
+ "consumer renewal to write settings and dispose the old stream");
+ AssertSettingsWritten(renewed);
+ Assert.AreEqual(0, renewed.DisposeCalls);
+ Assert.IsFalse(consumer.IsEndpointsDeprecated(endpoints));
+ clientManager.Verify(m => m.Telemetry(endpoints), Times.Exactly(2));
+ }
+ finally
+ {
+ await consumer.DisposeAsync();
+ }
+ }
+
+ [TestMethod]
+ [Timeout(30000)]
+ public async Task TestPushConsumerDropsObsoleteSessionAndReaddsSameEndpointsWithANewStream()
+ {
+ using var first = new FakeTelemetryCall(true, SettingsCommand());
+ using var rejoined = new FakeTelemetryCall(true, SettingsCommand());
+ var consumer = CreateConsumer(out var clientManager, first, rejoined);
+ try
+ {
+ var route = CreateTopicRouteData();
+ var endpoints = route.MessageQueues[0].Broker.Endpoints;
+ await consumer.OnTopicRouteDataFetched(Topic, route);
+ AssertSettingsWritten(first);
+
+ await consumer.OnTopicRouteDataFetched(Topic, new TopicRouteData(Array.Empty<Proto.MessageQueue>()));
+
+ Assert.IsTrue(consumer.IsEndpointsDeprecated(endpoints));
+ Assert.AreEqual(1, first.DisposeCalls,
+ "publishing an empty route must close the obsolete stream without a reconnect callback");
+ Assert.AreEqual(0, rejoined.WrittenCommands);
+ clientManager.Verify(m => m.Telemetry(endpoints), Times.Once);
+
+ await consumer.OnTopicRouteDataFetched(Topic, route);
+
+ Assert.IsFalse(consumer.IsEndpointsDeprecated(endpoints));
+ AssertSettingsWritten(rejoined);
+ Assert.AreEqual(0, rejoined.DisposeCalls);
+ Assert.AreEqual(1, first.DisposeCalls);
+ clientManager.Verify(m => m.Telemetry(endpoints), Times.Exactly(2));
+ }
+ finally
+ {
+ await consumer.DisposeAsync();
+ }
+ }
+
+ [TestMethod]
+ [Timeout(30000)]
+ public async Task TestPushConsumerShutdownClosesTheSessionsOfItsRouteTable()
+ {
+ using var stream = new FakeTelemetryCall(true, SettingsCommand());
+ var consumer = CreateConsumer(out var clientManager, stream);
+ try
+ {
+ var route = CreateTopicRouteData();
+ var endpoints = route.MessageQueues[0].Broker.Endpoints;
+ await consumer.OnTopicRouteDataFetched(Topic, route);
+ AssertSettingsWritten(stream);
+
+ await consumer.DisposeAsync();
+
+ Assert.AreEqual(State.Terminated, consumer.State);
+ Assert.AreEqual(1, stream.DisposeCalls,
+ "public consumer disposal must close its telemetry stream");
+ consumer.ReconnectTelemetry(endpoints);
+ clientManager.Verify(m => m.Telemetry(endpoints), Times.Once);
+ }
+ finally
+ {
+ await consumer.DisposeAsync();
+ }
+ }
+
+ [TestMethod]
+ [Timeout(30000)]
+ public async Task TestPushConsumerAutomaticallyRenewsTelemetryAfterEof()
+ {
+ using var first = new FakeTelemetryCall(true, SettingsCommand());
+ using var renewed = new FakeTelemetryCall(true, SettingsCommand());
+ var consumer = CreateConsumer(out var clientManager, first, renewed);
+ try
+ {
+ var route = CreateTopicRouteData();
+ var endpoints = route.MessageQueues[0].Broker.Endpoints;
+ await consumer.OnTopicRouteDataFetched(Topic, route);
+ AssertSettingsWritten(first);
+
+ first.Reader.Close();
+
+ await TestAwaiter.Until(() => renewed.WrittenCommands > 0 && 1 == first.DisposeCalls,
+ "EOF to renew an active consumer's stream without an explicit reconnect");
+ AssertSettingsWritten(renewed);
+ Assert.AreEqual(0, renewed.DisposeCalls);
+ Assert.IsFalse(consumer.IsEndpointsDeprecated(endpoints));
+ clientManager.Verify(m => m.Telemetry(endpoints), Times.Exactly(2));
+ }
+ finally
+ {
+ await consumer.DisposeAsync();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestMalformedVerificationCommandDoesNotEscape()
+ {
+ using var stream = new FakeTelemetryCall(true, SettingsCommand());
+ var consumer = CreateConsumer(out var manager, stream);
+ try
+ {
+ var route = CreateTopicRouteData();
+ var endpoints = route.MessageQueues[0].Broker.Endpoints;
+ await consumer.OnTopicRouteDataFetched(Topic, route);
+ await consumer.OnVerifyMessageCommand(endpoints, new Proto.VerifyMessageCommand { Nonce = "nonce" });
+ Assert.AreEqual(1, stream.WrittenCommands);
+ Assert.AreEqual(0, stream.DisposeCalls);
+ manager.Verify(m => m.Telemetry(endpoints), Times.Once);
+ }
+ finally
+ {
+ await consumer.DisposeAsync();
+ }
+ }
+
+ private static PushConsumer CreateConsumer(out Mock<IClientManager> clientManager,
+ params FakeTelemetryCall[] streams)
+ {
+ var clientConfig = new ClientConfig.Builder()
+ .SetEndpoints("127.0.0.1:8080")
+ .EnableSsl(false)
+ .Build();
+ var consumer = new PushConsumer(clientConfig, TestConsumerGroup,
+ new ConcurrentDictionary<string, FilterExpression>(), new NoopMessageListener(), 10, 10, 1)
+ {
+ State = State.Running
+ };
+ clientManager = new Mock<IClientManager>();
+ var telemetry = clientManager.SetupSequence(m => m.Telemetry(It.IsAny<Endpoints>()));
+ foreach (var stream in streams)
+ {
+ telemetry.Returns(stream.Call);
+ }
+
+ telemetry.Throws(new InvalidOperationException("Unexpected telemetry stream creation"));
+ consumer.SetClientManager(clientManager.Object);
+ return consumer;
+ }
+
+ private static void AssertSettingsWritten(FakeTelemetryCall stream)
+ {
+ Assert.IsTrue(stream.WrittenCommands > 0, "the consumer must write settings on the stream");
+ Assert.IsNotNull(stream.WrittenCommand(0).Settings);
+ }
+
+ private static TopicRouteData CreateTopicRouteData()
+ {
+ var messageQueue = new Proto.MessageQueue
+ {
+ Topic = new Proto.Resource
+ {
+ ResourceNamespace = "namespace",
+ Name = Topic
+ },
+ Id = 0,
+ Permission = Proto.Permission.ReadWrite,
+ Broker = new Proto.Broker
+ {
+ Name = Broker,
+ Id = 0,
+ Endpoints = new Proto.Endpoints
+ {
+ Scheme = Proto.AddressScheme.Ipv4,
+ Addresses = { new Proto.Address { Host = "127.0.0.1", Port = 8080 } }
+ }
+ }
+ };
+ return new TopicRouteData(new[] { messageQueue });
+ }
+
+ /// <summary>
+ /// Settings a push consumer accepts: its subscription settings reject a retry policy which carries no strategy,
+ /// and an exponential backoff without its durations would fault the read loop which delivers them.
+ /// </summary>
+ private static Proto.TelemetryCommand SettingsCommand()
+ {
+ return new Proto.TelemetryCommand
+ {
+ Settings = new Proto.Settings
+ {
+ Subscription = new Proto.Subscription(),
+ BackoffPolicy = new Proto.RetryPolicy
+ {
+ MaxAttempts = 3,
+ ExponentialBackoff = new Proto.ExponentialBackoff
+ {
+ Initial = new Duration { Seconds = 1 },
+ Max = new Duration { Seconds = 10 },
+ Multiplier = 2
+ }
+ }
+ }
+ };
+ }
+
+ private sealed class NoopMessageListener : IMessageListener
+ {
+ public ConsumeResult Consume(MessageView messageView)
+ {
+ return ConsumeResult.SUCCESS;
+ }
+ }
+ }
+}
diff --git a/csharp/tests/RecoveryTestSupport.cs b/csharp/tests/RecoveryTestSupport.cs
new file mode 100644
index 00000000..746d30aa
--- /dev/null
+++ b/csharp/tests/RecoveryTestSupport.cs
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.Rocketmq;
+using Endpoints = Org.Apache.Rocketmq.Endpoints;
+using Proto = Apache.Rocketmq.V2;
+
+namespace tests
+{
+ /// <summary>
+ /// Client pinned to the running state whose session lifecycle hooks are observable, so that transport recovery can
+ /// be driven without a live broker.
+ /// </summary>
+ internal class RecoveryTestClient : Producer
+ {
+ private readonly TaskCompletionSource<Proto.ReconnectEndpointsCommand> _firstReconnectCommand =
+ new TaskCompletionSource<Proto.ReconnectEndpointsCommand>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ private int _reconnectTelemetryCalls;
+ private int _removeSessionCalls;
+
+ internal RecoveryTestClient(ClientConfig clientConfig)
+ : base(clientConfig, new ConcurrentDictionary<string, bool>(), 1, null)
+ {
+ State = Org.Apache.Rocketmq.State.Running;
+ }
+
+ /// <summary>
+ /// Whether the endpoints must be reported as dropped from the topic route table.
+ /// </summary>
+ internal volatile bool EndpointsDeprecated;
+
+ /// <summary>
+ /// Whether the telemetry renewal must fail, so that the recovery paths can be driven into their failure
+ /// branches.
+ /// </summary>
+ internal volatile bool FailReconnectTelemetry;
+
+ internal int ReconnectTelemetryCalls => Volatile.Read(ref _reconnectTelemetryCalls);
+
+ internal int RemoveSessionCalls => Volatile.Read(ref _removeSessionCalls);
+
+ internal Task<Proto.ReconnectEndpointsCommand> FirstReconnectCommand => _firstReconnectCommand.Task;
+
+ /// <summary>
+ /// Runs after the renewal counter is incremented, so that a test can pin the order in which recovery replaces
+ /// the transport and rebuilds the telemetry.
+ /// </summary>
+ internal Action<Endpoints> ReconnectTelemetryHook;
+
+ internal override void ReconnectTelemetry(Endpoints endpoints, Session expectedSession = null)
+ {
+ if (FailReconnectTelemetry)
+ {
+ throw new RpcException(new Status(StatusCode.Unavailable, "no telemetry"));
+ }
+
+ Interlocked.Increment(ref _reconnectTelemetryCalls);
+ ReconnectTelemetryHook?.Invoke(endpoints);
+ }
+
+ internal override bool IsEndpointsDeprecated(Endpoints endpoints)
+ {
+ return EndpointsDeprecated;
+ }
+
+ internal override async Task<bool> RemoveSession(Endpoints endpoints, Session session)
+ {
+ Interlocked.Increment(ref _removeSessionCalls);
+ await session.CloseAsync();
+ return true;
+ }
+
+ internal override void OnReconnectEndpointsCommand(Endpoints endpoints, Proto.ReconnectEndpointsCommand command)
+ {
+ _firstReconnectCommand.TrySetResult(command);
+ }
+ }
+
+ /// <summary>
+ /// In-memory telemetry stream whose writes and disposal are observable, standing in for a gRPC call.
+ /// </summary>
+ internal sealed class FakeTelemetryCall : IDisposable
+ {
+ internal readonly RecordingStreamWriter Writer;
+ internal readonly FakeStreamReader Reader;
+ internal readonly AsyncDuplexStreamingCall<Proto.TelemetryCommand, Proto.TelemetryCommand> Call;
+
+ private int _disposeCalls;
+
+ internal FakeTelemetryCall(bool keepStreamOpen = true, params Proto.TelemetryCommand[] responses)
+ {
+ Writer = new RecordingStreamWriter();
+ Reader = new FakeStreamReader(keepStreamOpen, responses);
+ Call = new AsyncDuplexStreamingCall<Proto.TelemetryCommand, Proto.TelemetryCommand>(
+ Writer, Reader, null, null, null, () =>
+ {
+ if (0 != Interlocked.Exchange(ref _disposeCalls, 1))
+ {
+ return;
+ }
+
+ // A real call cancels pending reads and writes, not the caller's test gate.
+ Reader.Close();
+ Writer.Close();
+ });
+ }
+
+ public void Dispose()
+ {
+ Call.Dispose();
+ }
+
+ internal int DisposeCalls => Volatile.Read(ref _disposeCalls);
+
+ internal int WrittenCommands
+ {
+ get
+ {
+ lock (Writer.Commands)
+ {
+ return Writer.Commands.Count;
+ }
+ }
+ }
+
+ internal Proto.TelemetryCommand WrittenCommand(int index)
+ {
+ lock (Writer.Commands)
+ {
+ return Writer.Commands[index];
+ }
+ }
+ }
+
+ internal sealed class RecordingStreamWriter : IClientStreamWriter<Proto.TelemetryCommand>
+ {
+ internal readonly List<Proto.TelemetryCommand> Commands = new List<Proto.TelemetryCommand>();
+
+ private readonly CancellationTokenSource _closed = new CancellationTokenSource();
+ private bool _disposed;
+
+ /// <summary>
+ /// When set, writes are recorded and then wait for this gate or call disposal. Disposal leaves the gate itself
+ /// untouched, so an unused gate never acquires an unobserved exception.
+ /// </summary>
+ internal TaskCompletionSource<bool> WriteGate { get; set; }
+
+ public WriteOptions WriteOptions { get; set; }
+
+ public async Task WriteAsync(Proto.TelemetryCommand message)
+ {
+ Task gate;
+ CancellationToken cancellationToken;
+ lock (Commands)
+ {
+ if (_disposed)
+ {
+ throw new RpcException(new Status(StatusCode.Cancelled, "telemetry call disposed"));
+ }
+
+ cancellationToken = _closed.Token;
+ gate = WriteGate?.Task;
+ Commands.Add(message);
+ }
+
+ try
+ {
+ if (null != gate)
+ {
+ await gate.WaitAsync(cancellationToken);
+ }
+
+ cancellationToken.ThrowIfCancellationRequested();
+ }
+ catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+ {
+ throw new RpcException(new Status(StatusCode.Cancelled, "telemetry call disposed"));
+ }
+ }
+
+ internal void Close()
+ {
+ lock (Commands)
+ {
+ if (_disposed)
+ {
+ return;
+ }
+
+ _disposed = true;
+ _closed.Cancel();
+ _closed.Dispose();
+ }
+ }
+
+ public Task CompleteAsync()
+ {
+ return Task.CompletedTask;
+ }
+ }
+
+ internal sealed class FakeStreamReader : IAsyncStreamReader<Proto.TelemetryCommand>
+ {
+ private readonly Queue<Proto.TelemetryCommand> _responses;
+ private readonly bool _keepOpen;
+ private readonly TaskCompletionSource<bool> _closed =
+ new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ internal FakeStreamReader(bool keepOpen, params Proto.TelemetryCommand[] responses)
+ {
+ _keepOpen = keepOpen;
+ _responses = new Queue<Proto.TelemetryCommand>(responses);
+ }
+
+ public Proto.TelemetryCommand Current { get; private set; }
+
+ internal void Close()
+ {
+ lock (_responses)
+ {
+ _responses.Clear();
+ _closed.TrySetResult(true);
+ }
+ }
+
+ public async Task<bool> MoveNext(CancellationToken cancellationToken)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ lock (_responses)
+ {
+ if (_closed.Task.IsCompleted)
+ {
+ return false;
+ }
+
+ if (_responses.Count > 0)
+ {
+ Current = _responses.Dequeue();
+ return true;
+ }
+ }
+
+ if (!_keepOpen)
+ {
+ return false;
+ }
+
+ await _closed.Task.WaitAsync(cancellationToken);
+ return false;
+ }
+ }
+
+ internal static class TestAwaiter
+ {
+ internal static async Task Until(Func<bool> condition, string description, TimeSpan? timeout = null)
+ {
+ var deadline = DateTime.UtcNow.Add(timeout ?? TimeSpan.FromSeconds(5));
+ while (!condition())
+ {
+ Assert.IsTrue(DateTime.UtcNow < deadline, $"Timed out waiting for {description}");
+ await Task.Delay(20);
+ }
+ }
+ }
+
+ internal static class RecoveryTestSupport
+ {
+ internal static readonly Endpoints FakeEndpoints = new Endpoints("127.0.0.1:8080");
+
+ internal static ClientConfig CreateClientConfig()
+ {
+ return new ClientConfig.Builder().SetEndpoints("127.0.0.1:9876").Build();
+ }
+
+ internal static Task<Proto.HeartbeatResponse> FailedHeartbeat(StatusCode statusCode)
+ {
+ return Task.FromException<Proto.HeartbeatResponse>(new RpcException(new Status(statusCode, null)));
+ }
+
+ /// <summary>
+ /// An unobserved task exception and a leaked descriptor both surface only once the offending object is
+ /// finalized, which a full collection forces.
+ /// </summary>
+ internal static void ForceFinalization()
+ {
+ for (var i = 0; i < 3; i++)
+ {
+ GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, true, true);
+ GC.WaitForPendingFinalizers();
+ }
+
+ GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, true, true);
+ }
+ }
+}
diff --git a/csharp/tests/SessionTest.cs b/csharp/tests/SessionTest.cs
index ec53ef10..cadd112f 100644
--- a/csharp/tests/SessionTest.cs
+++ b/csharp/tests/SessionTest.cs
@@ -15,7 +15,11 @@
* limitations under the License.
*/
+using System;
using System.Collections.Concurrent;
+using System.Linq;
+using System.Reflection;
+using System.Threading;
using System.Threading.Tasks;
using Grpc.Core;
using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -29,35 +33,413 @@ namespace tests
[TestClass]
public class SessionTests
{
- private static Client CreateTestClient()
+ [TestMethod]
+ public async Task TestSyncSettings()
{
- var clientConfig = new ClientConfig.Builder().SetEndpoints("127.0.0.1:9876").Build();
- return new Producer(clientConfig, new ConcurrentDictionary<string, bool>(), 1, null);
+ await using var client = new Producer(RecoveryTestSupport.CreateClientConfig(), new ConcurrentDictionary<string, bool>(), 1, null);
+ using var stream = new FakeTelemetryCall(true, new Proto.TelemetryCommand { Settings = client.GetSettings().ToProtobuf() });
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, stream.Call, client);
+ try
+ {
+ await session.SyncSettings(true);
+ Assert.AreEqual(1, stream.WrittenCommands);
+ Assert.IsNotNull(stream.WrittenCommand(0).Settings);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
}
[TestMethod]
- public async Task TestSyncSettings()
+ public async Task TestReconnectRenewsTelemetryAndSyncsSettings()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(renewed.Call);
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, first.Call, client);
+ try
+ {
+ session.Reconnect();
+ await TestAwaiter.Until(() => renewed.WrittenCommands == 1, "settings on renewed telemetry");
+ Assert.IsNotNull(renewed.WrittenCommand(0).Settings);
+ Assert.AreEqual(1, first.DisposeCalls);
+ Assert.AreEqual(0, renewed.DisposeCalls);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+
+ Assert.AreEqual(1, renewed.DisposeCalls);
+ }
+
+ [TestMethod]
+ [Timeout(20000)]
+ public async Task TestConcurrentReconnectOnlyRenewsTelemetryOnce()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ const int threadCount = 8;
+ var started = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(() =>
+ {
+ started.TrySetResult(true);
+ release.Task.GetAwaiter().GetResult();
+ return renewed.Call;
+ });
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, first.Call, client);
+ using var barrier = new Barrier(threadCount + 1);
+ var reconnects = Enumerable.Range(0, threadCount).Select(_ => Task.Run(() =>
+ {
+ Assert.IsTrue(barrier.SignalAndWait(TimeSpan.FromSeconds(10)));
+ session.Reconnect();
+ })).ToArray();
+ try
+ {
+ Assert.IsTrue(barrier.SignalAndWait(TimeSpan.FromSeconds(10)));
+ await started.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ await TestAwaiter.Until(() => reconnects.Count(task => task.IsCompleted) == threadCount - 1,
+ "all losing reconnect attempts to exit while the winner is blocked");
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ release.SetResult(true);
+ await Task.WhenAll(reconnects);
+ await TestAwaiter.Until(() => renewed.WrittenCommands == 1, "the winning renewal to synchronize settings");
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ }
+ finally
+ {
+ release.TrySetResult(true);
+ await Task.WhenAll(reconnects);
+ await session.CloseAsync();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestReconnectIsForgivenWhileClientIsNotRunningAndRetriedAfterwards()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(renewed.Call);
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, first.Call, client);
+ try
+ {
+ client.State = State.Stopping;
+ session.Reconnect();
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Never);
+ client.State = State.Running;
+ session.Reconnect();
+ await TestAwaiter.Until(() => renewed.WrittenCommands == 1, "renewal after the client resumes");
+ Assert.AreEqual(1, first.DisposeCalls);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestReconnectRemovesSessionWhenEndpointsAreDeprecated()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ client.EndpointsDeprecated = true;
+ using var stream = new FakeTelemetryCall();
+ var manager = new Mock<IClientManager>();
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, stream.Call, client);
+ try
+ {
+ session.Reconnect();
+ await TestAwaiter.Until(() => client.RemoveSessionCalls == 1 && stream.DisposeCalls == 1,
+ "the deprecated session to be removed and closed");
+ client.EndpointsDeprecated = false;
+ session.Reconnect();
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Never);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestReconnectEndpointsCommandIsDispatched()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var stream = new FakeTelemetryCall(true, new Proto.TelemetryCommand
+ {
+ ReconnectEndpointsCommand = new Proto.ReconnectEndpointsCommand { Nonce = "nonce" }
+ });
+ var manager = new Mock<IClientManager>();
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, stream.Call, client);
+ try
+ {
+ var received = await client.FirstReconnectCommand.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.AreEqual("nonce", received.Nonce);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Never);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestWriteAsyncIsDroppedOnceTheSessionIsClosed()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var stream = new FakeTelemetryCall();
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, stream.Call, client);
+ await session.CloseAsync();
+ await session.WriteAsync(new Proto.TelemetryCommand());
+ Assert.AreEqual(0, stream.WrittenCommands);
+ Assert.AreEqual(1, stream.DisposeCalls);
+ }
+
+ [TestMethod]
+ [Timeout(15000)]
+ public async Task TestCloseIsNotHeldUpByAStuckWrite()
{
- var testClient = CreateTestClient();
- var endpoints = new Endpoints(testClient.GetClientConfig().Endpoints);
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var stream = new FakeTelemetryCall();
+ stream.Writer.WriteGate = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, stream.Call, client);
+ var write = session.WriteAsync(new Proto.TelemetryCommand());
+ await TestAwaiter.Until(() => stream.WrittenCommands == 1, "the blocked write");
+ await session.CloseAsync().WaitAsync(Session.CloseTimeout.Add(TimeSpan.FromSeconds(5)));
+ var error = await Assert.ThrowsExactlyAsync<RpcException>(() => write);
+ Assert.AreEqual(StatusCode.Cancelled, error.StatusCode);
+ Assert.AreEqual(1, stream.DisposeCalls);
+ }
+
+ [TestMethod]
+ public async Task TestRenewalDiscardsTheStreamOfASessionClosedWhileItWasCreated()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ Session session = null;
+ Task close = null;
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(() =>
+ {
+ close = session.CloseAsync();
+ return renewed.Call;
+ });
+ client.SetClientManager(manager.Object);
+ session = new Session(RecoveryTestSupport.FakeEndpoints, first.Call, client);
+ try
+ {
+ session.Reconnect();
+ await (close ?? Task.CompletedTask);
+ await TestAwaiter.Until(() => renewed.DisposeCalls == 1, "the unused replacement to be disposed");
+ Assert.AreEqual(0, renewed.WrittenCommands);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+ }
+
+ [TestMethod]
+ public async Task TestReadLoopTerminationRenewsActiveSession()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(renewed.Call);
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, first.Call, client);
+ try
+ {
+ first.Reader.Close();
+ await TestAwaiter.Until(() => renewed.WrittenCommands == 1, "EOF to renew telemetry and synchronize settings");
+ Assert.AreEqual(1, first.DisposeCalls);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+ }
+
+ [TestMethod]
+ [Timeout(20000)]
+ public async Task TestReplacementEofIsNotLostWhileSettingsSynchronizationIsPending()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var first = new FakeTelemetryCall();
+ using var ended = new FakeTelemetryCall(false);
+ ended.Writer.WriteGate = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ using var healthy = new FakeTelemetryCall(true, new Proto.TelemetryCommand { Settings = client.GetSettings().ToProtobuf() });
+ var manager = new Mock<IClientManager>();
+ manager.SetupSequence(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(ended.Call).Returns(healthy.Call);
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, first.Call, client);
+ var initialSync = Task.Run(() => session.SyncSettings(true));
+ try
+ {
+ await TestAwaiter.Until(() => first.WrittenCommands == 1, "initial settings to await their response");
+ session.Reconnect();
+ await TestAwaiter.Until(() => healthy.WrittenCommands == 1,
+ "an immediately terminated replacement to renew again", TimeSpan.FromSeconds(10));
+ await initialSync;
+ Assert.AreEqual(1, ended.DisposeCalls);
+ Assert.IsNotNull(healthy.WrittenCommand(0).Settings);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Exactly(2));
+ }
+ finally
+ {
+ await session.CloseAsync();
+ await initialSync;
+ }
+ }
+
+ [TestMethod]
+ [Timeout(15000)]
+ public async Task TestCommandFailureCancelsOldCallBeforeWaitingForWriter()
+ {
+ await using var client = new ThrowingSettingsClient();
+ using var first = new FakeTelemetryCall(true, new Proto.TelemetryCommand { Settings = client.GetSettings().ToProtobuf() });
+ first.Writer.WriteGate = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ using var renewed = new FakeTelemetryCall();
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(renewed.Call);
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, first.Call, client);
+ var write = session.WriteAsync(new Proto.TelemetryCommand());
+ try
+ {
+ await TestAwaiter.Until(() => first.WrittenCommands == 1, "the old writer to block");
+ client.FailSettings.TrySetResult(true);
+ await TestAwaiter.Until(() => renewed.WrittenCommands == 1, "command failure to renew despite the blocked writer");
+ var error = await Assert.ThrowsExactlyAsync<RpcException>(() => write);
+ Assert.AreEqual(StatusCode.Cancelled, error.StatusCode);
+ Assert.AreEqual(1, first.DisposeCalls);
+ }
+ finally
+ {
+ client.FailSettings.TrySetResult(true);
+ await session.CloseAsync();
+ await Assert.ThrowsExactlyAsync<RpcException>(() => write);
+ }
+ }
+
+ [TestMethod]
+ [Timeout(15000)]
+ public async Task TestCurrentStreamEofRetriesAfterLosingRenewalGuard()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ var manager = new Mock<IClientManager>();
+ manager.Setup(m => m.Telemetry(It.IsAny<Endpoints>())).Returns(renewed.Call);
+ client.SetClientManager(manager.Object);
+ var session = new Session(RecoveryTestSupport.FakeEndpoints, first.Call, client);
+ var guard = typeof(Session).GetField("_reconnecting", BindingFlags.Instance | BindingFlags.NonPublic);
+ var onTermination = typeof(Session).GetMethod("RenewAfterTerminationAsync",
+ BindingFlags.Instance | BindingFlags.NonPublic);
+ Assert.IsNotNull(guard);
+ Assert.IsNotNull(onTermination);
+ using var context = new PausedRenewalContext();
+ Task renewal = null;
+ try
+ {
+ guard.SetValue(session, 1);
+ var previousContext = SynchronizationContext.Current;
+ try
+ {
+ SynchronizationContext.SetSynchronizationContext(context);
+ renewal = (Task)onTermination.Invoke(session, new object[] { first.Call });
+ }
+ finally
+ {
+ SynchronizationContext.SetSynchronizationContext(previousContext);
+ }
+
+ // Execute the EOF continuation while a stale callback still holds the renewal guard.
+ await context.RunNextAsync();
+ Assert.IsFalse(renewal.IsCompleted, "a current stream's EOF must not be lost when admission is busy");
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Never);
- var mockStreamWriter = new Mock<IClientStreamWriter<Proto.TelemetryCommand>>();
- var mockAsyncStreamReader = new Mock<IAsyncStreamReader<Proto.TelemetryCommand>>();
- var mockClientManager = new Mock<IClientManager>();
- var mockGrpcCall = new AsyncDuplexStreamingCall<Proto.TelemetryCommand, Proto.TelemetryCommand>(
- mockStreamWriter.Object, mockAsyncStreamReader.Object, null, null, null, null);
+ guard.SetValue(session, 0);
+ await context.RunNextAsync();
+ await renewal.WaitAsync(TimeSpan.FromSeconds(5));
+ await TestAwaiter.Until(() => renewed.WrittenCommands == 1, "the retained EOF to renew and sync settings");
+ Assert.AreEqual(1, first.DisposeCalls);
+ manager.Verify(m => m.Telemetry(It.IsAny<Endpoints>()), Times.Once);
+ }
+ finally
+ {
+ guard.SetValue(session, 0);
+ await session.CloseAsync();
+ if (renewal != null && !renewal.IsCompleted)
+ {
+ await context.RunNextAsync();
+ await renewal.WaitAsync(TimeSpan.FromSeconds(5));
+ }
+ }
+ }
+
+ private sealed class PausedRenewalContext : SynchronizationContext, IDisposable
+ {
+ private readonly ConcurrentQueue<Action> _continuations = new ConcurrentQueue<Action>();
+ private readonly SemaphoreSlim _posted = new SemaphoreSlim(0);
+
+ public override void Post(SendOrPostCallback callback, object state)
+ {
+ _continuations.Enqueue(() => callback(state));
+ _posted.Release();
+ }
- mockClientManager.Setup(cm => cm.Telemetry(endpoints)).Returns(mockGrpcCall);
- var session = new Session(endpoints, mockGrpcCall, testClient);
+ internal async Task RunNextAsync()
+ {
+ Assert.IsTrue(await _posted.WaitAsync(TimeSpan.FromSeconds(5)), "the renewal backoff must complete");
+ Assert.IsTrue(_continuations.TryDequeue(out var continuation));
+ var previousContext = Current;
+ try
+ {
+ SetSynchronizationContext(this);
+ continuation();
+ }
+ finally
+ {
+ SetSynchronizationContext(previousContext);
+ }
+ }
- var settings = new Proto.Settings();
- mockStreamWriter.Setup(m => m.WriteAsync(It.Is<Proto.TelemetryCommand>(tc => tc.Settings == settings)))
- .Returns(Task.CompletedTask);
- testClient.SetClientManager(mockClientManager.Object);
+ public void Dispose()
+ {
+ _posted.Dispose();
+ }
+ }
+
+ private sealed class ThrowingSettingsClient : RecoveryTestClient
+ {
+ internal readonly TaskCompletionSource<bool> FailSettings = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
- await session.SyncSettings(true);
+ internal ThrowingSettingsClient() : base(RecoveryTestSupport.CreateClientConfig())
+ {
+ }
- mockStreamWriter.Verify(m => m.WriteAsync(It.IsAny<Proto.TelemetryCommand>()), Times.Once);
+ internal override void OnSettingsCommand(Endpoints endpoints, Proto.Settings settings)
+ {
+ FailSettings.Task.GetAwaiter().GetResult();
+ throw new InvalidOperationException("invalid settings");
+ }
}
}
-}
\ No newline at end of file
+}
diff --git a/csharp/tests/TransportRecoveryConcurrencyTest.cs b/csharp/tests/TransportRecoveryConcurrencyTest.cs
new file mode 100644
index 00000000..2b55ca0b
--- /dev/null
+++ b/csharp/tests/TransportRecoveryConcurrencyTest.cs
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Moq;
+using Org.Apache.Rocketmq;
+using Endpoints = Org.Apache.Rocketmq.Endpoints;
+using Proto = Apache.Rocketmq.V2;
+using grpcLib = Grpc.Core;
+
+namespace tests
+{
+ /// <summary>
+ /// Exercises recovery under repeated contention and telemetry creation failures without changing process-wide
+ /// thread pool settings.
+ /// </summary>
+ [TestClass]
+ public class TransportRecoveryConcurrencyTest
+ {
+ private const int Rounds = 50;
+ private const int Threads = 16;
+
+ [TestMethod]
+ [Timeout(120000)]
+ public async Task TestRepeatedConcurrentReconnectNeverResetsTheTransportTwiceAtTheSameTime()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ var clientManager = new ClientManager(client);
+ client.SetClientManager(clientManager);
+ var rpcClient = new Mock<IRpcClient>();
+ var overlap = new OverlapRecorder();
+ var resetCalls = 0;
+ rpcClient.Setup(c => c.ResetTransport()).Callback(() =>
+ {
+ overlap.Enter();
+ // Widened on purpose: without a window to overlap in, a broken guard would still look correct.
+ Thread.SpinWait(4000);
+ overlap.Leave();
+ Interlocked.Increment(ref resetCalls);
+ });
+
+ for (var round = 0; round < Rounds; round++)
+ {
+ using var barrier = new Barrier(Threads);
+ await Task.WhenAll(Enumerable.Range(0, Threads).Select(_ => Task.Run(() =>
+ {
+ barrier.SignalAndWait();
+ clientManager.Reconnect(RecoveryTestSupport.FakeEndpoints, rpcClient.Object);
+ })));
+ }
+
+ // The guard makes the recoveries mutually exclusive rather than singular: a server command which arrives
+ // once the previous recovery completed is entitled to reset again, and only a reset which overlaps
+ // another one is a defect.
+ Assert.AreEqual(1, overlap.Widest,
+ "transport resets ran at the same time, the widest overlap reached " + overlap.Widest);
+ Assert.IsTrue(Volatile.Read(ref resetCalls) >= Rounds,
+ "every round must have been able to recover again, only " + resetCalls + " resets happened");
+ Assert.AreEqual(client.ReconnectTelemetryCalls, Volatile.Read(ref resetCalls),
+ "every transport reset must be followed by exactly one telemetry rebuild");
+ }
+
+ [TestMethod]
+ [Timeout(120000)]
+ public async Task TestConcurrentHeartbeatFailuresAndServerCommandsRecoverWithoutDeadlock()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ var clientManager = new ClientManager(client);
+ client.SetClientManager(clientManager);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.State).Returns(grpcLib.ConnectivityState.Ready);
+ var resetCalls = 0;
+ rpcClient.Setup(c => c.ResetTransport()).Callback(() => Interlocked.Increment(ref resetCalls));
+
+ // Every source at once: heartbeat timeouts, heartbeat refusals, the server command and plain successes.
+ var workload = Enumerable.Range(0, 64).Select(index => Task.Run(async () =>
+ {
+ switch (index % 4)
+ {
+ case 0:
+ await clientManager.MonitorHeartbeat(RecoveryTestSupport.FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.DeadlineExceeded));
+ break;
+ case 1:
+ await clientManager.MonitorHeartbeat(RecoveryTestSupport.FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(grpcLib.StatusCode.Unavailable));
+ break;
+ case 2:
+ clientManager.Reconnect(RecoveryTestSupport.FakeEndpoints, rpcClient.Object);
+ break;
+ default:
+ await clientManager.MonitorHeartbeat(RecoveryTestSupport.FakeEndpoints, rpcClient.Object,
+ Task.FromResult(new Proto.HeartbeatResponse()));
+ break;
+ }
+ })).ToArray();
+
+ await Task.WhenAll(workload);
+
+ Assert.IsTrue(Volatile.Read(ref resetCalls) > 0,
+ "the contending triggers must recover the transport");
+ Assert.AreEqual(Volatile.Read(ref resetCalls), client.ReconnectTelemetryCalls,
+ "every transport reset must be followed by exactly one telemetry rebuild");
+ }
+
+ [TestMethod]
+ [Timeout(30000)]
+ public async Task TestTelemetryRenewalRetriesCreationFailuresAndWritesSettings()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ var endpoints = new Endpoints(client.GetClientConfig().Endpoints);
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ var attempts = 0;
+ var clientManager = new Mock<IClientManager>();
+ clientManager.Setup(m => m.Telemetry(endpoints)).Returns(() =>
+ {
+ if (Interlocked.Increment(ref attempts) <= 2)
+ {
+ throw new grpcLib.RpcException(
+ new grpcLib.Status(grpcLib.StatusCode.Unavailable, "no stream"));
+ }
+
+ return renewed.Call;
+ });
+ client.SetClientManager(clientManager.Object);
+
+ var session = new Session(endpoints, first.Call, client);
+ try
+ {
+ session.Reconnect();
+ await TestAwaiter.Until(() => renewed.WrittenCommands > 0 && 1 == first.DisposeCalls,
+ "settings to be written after two telemetry creation failures", TimeSpan.FromSeconds(10));
+
+ Assert.AreEqual(3, Volatile.Read(ref attempts));
+ Assert.IsNotNull(renewed.WrittenCommand(0).Settings);
+ Assert.AreEqual(0, renewed.DisposeCalls, "the replacement stream must remain usable");
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+ }
+
+ /// <summary>
+ /// Widest number of recoveries which were inside the guarded section at the same moment.
+ /// </summary>
+ private sealed class OverlapRecorder
+ {
+ private int _inProgress;
+ private int _widest;
+
+ internal int Widest => Volatile.Read(ref _widest);
+
+ internal void Enter()
+ {
+ var depth = Interlocked.Increment(ref _inProgress);
+ int widest;
+ do
+ {
+ widest = Volatile.Read(ref _widest);
+ }
+ while (depth > widest && depth != Interlocked.CompareExchange(ref _widest, depth, widest));
+ }
+
+ internal void Leave()
+ {
+ Interlocked.Decrement(ref _inProgress);
+ }
+ }
+ }
+}
diff --git a/csharp/tests/TransportRecoveryLeakTest.cs b/csharp/tests/TransportRecoveryLeakTest.cs
new file mode 100644
index 00000000..941dec4d
--- /dev/null
+++ b/csharp/tests/TransportRecoveryLeakTest.cs
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Linq;
+using System.Net;
+using System.Net.NetworkInformation;
+using System.Threading.Tasks;
+using Grpc.Core;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.Rocketmq;
+using Endpoints = Org.Apache.Rocketmq.Endpoints;
+using Proto = Apache.Rocketmq.V2;
+
+namespace tests
+{
+ /// <summary>
+ /// Transport recovery replaces the channel and its handler on every cycle, and the replaced ones own real sockets.
+ /// A recovery which ran forever must not accumulate them, so the descriptors held by the process are counted across
+ /// a long run of real recoveries against a live server.
+ /// </summary>
+ [TestClass]
+ public class TransportRecoveryLeakTest : GrpcServerIntegrationTest
+ {
+ private const int RecoveryCycles = 200;
+
+ /// <summary>
+ /// Disposing a handler closes its sockets asynchronously, so a handful of descriptors may still be in flight
+ /// when the count is taken. Anything beyond that means the replaced transports are being retained.
+ /// </summary>
+ private const int DescriptorTolerance = 12;
+
+ private Server _server;
+
+ [TestInitialize]
+ public void SetUp()
+ {
+ _server = SetUpServer(new MockServer("topic", "broker", new List<string>()));
+ }
+
+ [TestCleanup]
+ public async Task TearDown()
+ {
+ await _server.ShutdownAsync().WaitAsync(TimeSpan.FromSeconds(10));
+ }
+
+ [TestMethod]
+ [Timeout(180000)]
+ public async Task TestRepeatedTransportRecoveryDoesNotLeakDescriptors()
+ {
+ var rpcClient = new RpcClient(new Endpoints($"127.0.0.1:{Port}"), false);
+ var metadata = new Metadata();
+ var request = new Proto.HeartbeatRequest();
+ var timeout = TimeSpan.FromSeconds(5);
+
+ // Called for real so that the first channel owns a connection before the baseline is taken, otherwise the
+ // comparison would measure a client which never connected at all.
+ await rpcClient.Heartbeat(metadata, request, timeout);
+ Assert.AreEqual(ConnectivityState.Ready, rpcClient.State,
+ "the channel of a client which just completed a call must be ready");
+
+ var descriptorsBefore = CountOpenDescriptors();
+ var establishedBefore = CountEstablishedConnectionsToTheServer();
+ var metadataCount = metadata.Count;
+
+ for (var i = 0; i < RecoveryCycles; i++)
+ {
+ rpcClient.ResetTransport();
+ metadata = new Metadata();
+ // Called again so that every cycle really does open a connection on the replacement channel.
+ await rpcClient.Heartbeat(metadata, request, timeout);
+ Assert.AreEqual(metadataCount, metadata.Count, "Caller metadata must not accumulate between heartbeats");
+ }
+
+ RecoveryTestSupport.ForceFinalization();
+
+ var descriptorsAfter = CountOpenDescriptors();
+ var establishedAfter = CountEstablishedConnectionsToTheServer();
+ var descriptorGrowth = descriptorsAfter - descriptorsBefore;
+ var establishedGrowth = establishedAfter - establishedBefore;
+
+ Assert.IsTrue(descriptorGrowth <= DescriptorTolerance,
+ $"{RecoveryCycles} transport recoveries leaked descriptors: {descriptorsBefore} -> " +
+ $"{descriptorsAfter} (growth {descriptorGrowth}, tolerance {DescriptorTolerance})");
+ Assert.IsTrue(establishedGrowth <= DescriptorTolerance,
+ $"{RecoveryCycles} transport recoveries left connections behind: {establishedBefore} -> " +
+ $"{establishedAfter} (growth {establishedGrowth})");
+ Assert.AreEqual(ConnectivityState.Ready, rpcClient.State,
+ "the channel of the last replacement must still be usable");
+ }
+
+ /// <summary>
+ /// The invariant the whole recovery rests on: replacing the transport must make the streams of the replaced
+ /// channel unusable at once, because that is what releases a writer which is stuck on a half-open connection
+ /// and holding the lock the telemetry renewal needs.
+ /// </summary>
+ [TestMethod]
+ [Timeout(60000)]
+ public async Task TestResetTransportMakesTheStreamsOfTheReplacedChannelUnusable()
+ {
+ var rpcClient = new RpcClient(new Endpoints($"127.0.0.1:{Port}"), false);
+ var call = rpcClient.Telemetry(new Metadata());
+
+ // Completed for real so that the channel owns a connection, which is the state a half-open one is in.
+ await rpcClient.Heartbeat(new Metadata(), new Proto.HeartbeatRequest(), TimeSpan.FromSeconds(5));
+ Assert.AreEqual(ConnectivityState.Ready, rpcClient.State,
+ "the channel of a client which just completed a call must be ready");
+
+ await call.RequestStream.WriteAsync(new Proto.TelemetryCommand());
+
+ rpcClient.ResetTransport();
+ Assert.AreEqual(ConnectivityState.Idle, rpcClient.State,
+ "a replacement channel which has not been used yet must start out idle");
+
+ // Bounded: a write which hangs here instead of failing is exactly the defect this guards against.
+ Exception failure = null;
+ var succeeded = false;
+ try
+ {
+ await call.RequestStream.WriteAsync(new Proto.TelemetryCommand())
+ .WaitAsync(TimeSpan.FromSeconds(10));
+ succeeded = true;
+ }
+ catch (Exception e)
+ {
+ failure = e;
+ }
+
+ Assert.IsFalse(succeeded, "a write on the stream of a replaced channel must not succeed");
+ Assert.IsNotNull(failure, "a write on the stream of a replaced channel must not hang");
+ Assert.IsNotInstanceOfType(failure, typeof(TimeoutException),
+ "a write on the stream of a replaced channel hung for the whole bound instead of failing");
+ Assert.IsInstanceOfType(failure, typeof(ObjectDisposedException),
+ "the replaced channel must be disposed, got " + failure.GetType().Name + ": " + failure.Message);
+ }
+
+ /// <summary>
+ /// Descriptors held by this process. <see cref="Process.HandleCount"/> reports zero on Unix, where the
+ /// descriptor table of the calling process is exposed under /dev/fd instead.
+ /// </summary>
+ private static int CountOpenDescriptors()
+ {
+ return OperatingSystem.IsWindows()
+ ? Process.GetCurrentProcess().HandleCount
+ : Directory.GetFiles("/dev/fd").Length;
+ }
+
+ /// <summary>
+ /// Connections this client holds towards the mock server. Filtering on the server port keeps the connections of
+ /// every other process on the machine, and the server side of these very connections, out of the count.
+ /// </summary>
+ private int CountEstablishedConnectionsToTheServer()
+ {
+ return IPGlobalProperties.GetIPGlobalProperties().GetActiveTcpConnections()
+ .Count(connection => TcpState.Established == connection.State
+ && Port == connection.RemoteEndPoint.Port
+ && IPAddress.Loopback.Equals(connection.RemoteEndPoint.Address));
+ }
+ }
+}
diff --git a/csharp/tests/UnobservedTaskExceptionTest.cs b/csharp/tests/UnobservedTaskExceptionTest.cs
new file mode 100644
index 00000000..8a273ef2
--- /dev/null
+++ b/csharp/tests/UnobservedTaskExceptionTest.cs
@@ -0,0 +1,311 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Moq;
+using Org.Apache.Rocketmq;
+using Endpoints = Org.Apache.Rocketmq.Endpoints;
+using Proto = Apache.Rocketmq.V2;
+using grpcLib = Grpc.Core;
+
+namespace tests
+{
+ /// <summary>
+ /// Checks recovery's background exception boundaries in a dedicated testhost. Other tests must neither contribute
+ /// faults to this check nor have their faults cleared by it; caller-owned monitor and write tasks are awaited.
+ /// </summary>
+ [TestClass]
+ public class UnobservedTaskExceptionTest
+ {
+ private const string ChildEnvironmentVariable = "ROCKETMQ_RECOVERY_UNOBSERVED_TEST_CHILD";
+ private const string ChildSuccessMarker = "RocketMQ isolated recovery exception check passed";
+
+ [TestMethod]
+ [Timeout(90000)]
+ public async Task TestTransportRecoveryLeavesNoUnobservedTaskException()
+ {
+ var testName = typeof(UnobservedTaskExceptionTest).FullName + "." +
+ nameof(TestTransportRecoveryLeavesNoUnobservedTaskException);
+ if (Environment.GetEnvironmentVariable(ChildEnvironmentVariable) != testName)
+ {
+ await RunInIsolatedTestHost(testName);
+ return;
+ }
+
+ var unobserved = new ConcurrentQueue<string>();
+ var canary = new InvalidOperationException("isolated recovery exception-handler canary");
+ var canaryObserved = 0;
+ EventHandler<UnobservedTaskExceptionEventArgs> handler = (_, args) =>
+ {
+ var exceptions = args.Exception.Flatten().InnerExceptions;
+ // Only this exact, deliberately unobserved task is excluded. Never clear real recovery faults.
+ if (1 == exceptions.Count && ReferenceEquals(canary, exceptions[0]))
+ {
+ Interlocked.Increment(ref canaryObserved);
+ }
+ else
+ {
+ unobserved.Enqueue(args.Exception.ToString());
+ }
+
+ args.SetObserved();
+ };
+
+ TaskScheduler.UnobservedTaskException += handler;
+ try
+ {
+ CreateUnobservedCanary(canary);
+ await HeartbeatMonitoringWithFailedRecovery();
+ await TelemetryRenewalRetries();
+ await RenewalAfterReadLoopEof();
+ await CloseWhileAWriteIsStuck();
+
+ RecoveryTestSupport.ForceFinalization();
+ var recorded = unobserved.ToArray();
+ Assert.AreEqual(0, recorded.Length,
+ $"Unobserved recovery task exceptions:{Environment.NewLine}" +
+ string.Join(Environment.NewLine + "---" + Environment.NewLine, recorded));
+ Assert.AreEqual(1, Volatile.Read(ref canaryObserved),
+ "the isolated handler must observe its known canary during finalization");
+ Console.WriteLine(ChildSuccessMarker);
+ }
+ finally
+ {
+ TaskScheduler.UnobservedTaskException -= handler;
+ }
+ }
+
+ private static async Task RunInIsolatedTestHost(string testName)
+ {
+ var assemblyPath = typeof(UnobservedTaskExceptionTest).Assembly.Location;
+ var startInfo = new ProcessStartInfo(FindDotnetHost())
+ {
+ UseShellExecute = false,
+ RedirectStandardOutput = true,
+ RedirectStandardError = true,
+ CreateNoWindow = true,
+ WorkingDirectory = Path.GetDirectoryName(assemblyPath)
+ };
+ startInfo.ArgumentList.Add("vstest");
+ startInfo.ArgumentList.Add(assemblyPath);
+ startInfo.ArgumentList.Add("/TestCaseFilter:FullyQualifiedName=" + testName);
+ startInfo.ArgumentList.Add("/Logger:console;verbosity=detailed");
+ startInfo.Environment[ChildEnvironmentVariable] = testName;
+
+ using var process = new Process { StartInfo = startInfo };
+ Assert.IsTrue(process.Start(), "the isolated testhost must start");
+ var stdout = process.StandardOutput.ReadToEndAsync();
+ var stderr = process.StandardError.ReadToEndAsync();
+ using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(60));
+ var timedOut = false;
+ try
+ {
+ await process.WaitForExitAsync(timeout.Token);
+ }
+ catch (OperationCanceledException) when (timeout.IsCancellationRequested)
+ {
+ timedOut = true;
+ }
+ finally
+ {
+ if (!process.HasExited)
+ {
+ process.Kill(entireProcessTree: true);
+ await process.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(5));
+ }
+ }
+
+ var output = await stdout.WaitAsync(TimeSpan.FromSeconds(5));
+ var errors = await stderr.WaitAsync(TimeSpan.FromSeconds(5));
+ var diagnostics = output + Environment.NewLine + errors;
+ Assert.IsFalse(timedOut, "isolated recovery test timed out:" + Environment.NewLine + diagnostics);
+ Assert.AreEqual(0, process.ExitCode, "isolated recovery test failed:" + Environment.NewLine + diagnostics);
+ StringAssert.Contains(output, ChildSuccessMarker,
+ "the exact filtered test must run its child scenarios, not merely discover zero tests");
+ }
+
+ private static string FindDotnetHost()
+ {
+ var executable = OperatingSystem.IsWindows() ? "dotnet.exe" : "dotnet";
+ var runtimeHost = Path.GetFullPath(Path.Combine(RuntimeEnvironment.GetRuntimeDirectory(),
+ "..", "..", "..", executable));
+ var root = Environment.GetEnvironmentVariable("DOTNET_ROOT");
+ var candidates = new[]
+ {
+ Environment.GetEnvironmentVariable("DOTNET_HOST_PATH"),
+ runtimeHost,
+ string.IsNullOrEmpty(root) ? null : Path.Combine(root, executable)
+ };
+ foreach (var candidate in candidates)
+ {
+ if (!string.IsNullOrEmpty(candidate) && Path.IsPathFullyQualified(candidate) && File.Exists(candidate))
+ {
+ return candidate;
+ }
+ }
+
+ throw new FileNotFoundException("Cannot locate the current .NET host via DOTNET_HOST_PATH, the runtime " +
+ "directory or DOTNET_ROOT; refusing to use an unrelated SDK from PATH.");
+ }
+
+ [MethodImpl(MethodImplOptions.NoInlining)]
+ private static void CreateUnobservedCanary(Exception canary)
+ {
+ _ = Task.FromException(canary);
+ }
+
+ private static async Task HeartbeatMonitoringWithFailedRecovery()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig())
+ {
+ FailReconnectTelemetry = true
+ };
+ var clientManager = new ClientManager(client);
+ client.SetClientManager(clientManager);
+ var rpcClient = new Mock<IRpcClient>();
+ rpcClient.Setup(c => c.State).Returns(grpcLib.ConnectivityState.Ready);
+ var resetCalls = 0;
+ rpcClient.Setup(c => c.ResetTransport())
+ .Callback(() => Interlocked.Increment(ref resetCalls))
+ .Throws(new grpcLib.RpcException(new grpcLib.Status(grpcLib.StatusCode.Internal, "reset failed")));
+
+ // A distinct endpoint keeps the explicit server recovery out of the heartbeat endpoint's cooldown.
+ // Both reset and telemetry rebuilding throw, and neither may escape the recovery boundary.
+ clientManager.Reconnect(new Endpoints("127.0.0.1:8081"), rpcClient.Object);
+ Assert.AreEqual(1, Volatile.Read(ref resetCalls));
+
+ var statuses = new[]
+ {
+ grpcLib.StatusCode.DeadlineExceeded, grpcLib.StatusCode.Unavailable,
+ grpcLib.StatusCode.ResourceExhausted, grpcLib.StatusCode.Internal, grpcLib.StatusCode.Cancelled
+ };
+ var monitors = new List<Task>();
+ foreach (var status in statuses)
+ {
+ for (var i = 0; i < 4; i++)
+ {
+ monitors.Add(clientManager.MonitorHeartbeat(RecoveryTestSupport.FakeEndpoints, rpcClient.Object,
+ RecoveryTestSupport.FailedHeartbeat(status)));
+ monitors.Add(clientManager.MonitorHeartbeat(RecoveryTestSupport.FakeEndpoints, rpcClient.Object,
+ Task.FromResult(new Proto.HeartbeatResponse())));
+ }
+ }
+
+ // A pending monitor has not yet observed its raw heartbeat failure.
+ await Task.WhenAll(monitors);
+ Assert.IsTrue(Volatile.Read(ref resetCalls) >= 2,
+ "a heartbeat-driven recovery must also attempt the failing reset");
+ }
+
+ private static async Task TelemetryRenewalRetries()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ var endpoints = new Endpoints(client.GetClientConfig().Endpoints);
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ var attempts = 0;
+ var clientManager = new Mock<IClientManager>();
+ clientManager.Setup(m => m.Telemetry(endpoints)).Returns(() =>
+ {
+ if (Interlocked.Increment(ref attempts) <= 2)
+ {
+ throw new grpcLib.RpcException(new grpcLib.Status(grpcLib.StatusCode.Unavailable, "no stream"));
+ }
+
+ return renewed.Call;
+ });
+ client.SetClientManager(clientManager.Object);
+ var session = new Session(endpoints, first.Call, client);
+ try
+ {
+ session.Reconnect();
+ await TestAwaiter.Until(() => renewed.WrittenCommands > 0 && 1 == first.DisposeCalls,
+ "the retry to write settings on the replacement stream", TimeSpan.FromSeconds(10));
+ Assert.AreEqual(3, Volatile.Read(ref attempts));
+ Assert.IsNotNull(renewed.WrittenCommand(0).Settings);
+ Assert.AreEqual(0, renewed.DisposeCalls);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+ }
+
+ private static async Task RenewalAfterReadLoopEof()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ var endpoints = new Endpoints(client.GetClientConfig().Endpoints);
+ using var first = new FakeTelemetryCall();
+ using var renewed = new FakeTelemetryCall();
+ var clientManager = new Mock<IClientManager>();
+ clientManager.Setup(m => m.Telemetry(endpoints)).Returns(renewed.Call);
+ client.SetClientManager(clientManager.Object);
+ var session = new Session(endpoints, first.Call, client);
+ try
+ {
+ first.Reader.Close();
+ await TestAwaiter.Until(() => renewed.WrittenCommands > 0 && 1 == first.DisposeCalls,
+ "EOF to renew telemetry and write settings on a healthy stream");
+ Assert.IsNotNull(renewed.WrittenCommand(0).Settings);
+ Assert.AreEqual(0, renewed.DisposeCalls);
+ clientManager.Verify(m => m.Telemetry(endpoints), Times.Once);
+ }
+ finally
+ {
+ await session.CloseAsync();
+ }
+ }
+
+ private static async Task CloseWhileAWriteIsStuck()
+ {
+ await using var client = new RecoveryTestClient(RecoveryTestSupport.CreateClientConfig());
+ var endpoints = new Endpoints(client.GetClientConfig().Endpoints);
+ using var stream = new FakeTelemetryCall();
+ client.SetClientManager(new Mock<IClientManager>().Object);
+ var session = new Session(endpoints, stream.Call, client);
+ var gate = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+ stream.Writer.WriteGate = gate;
+ var write = session.WriteAsync(new Proto.TelemetryCommand());
+ try
+ {
+ await TestAwaiter.Until(() => 1 == stream.WrittenCommands, "the write to reach the stream");
+ Assert.IsFalse(write.IsCompleted, "the write must still be blocked when close begins");
+ await session.CloseAsync().WaitAsync(Session.CloseTimeout.Add(TimeSpan.FromSeconds(5)));
+ Assert.AreEqual(1, stream.DisposeCalls);
+ Assert.IsFalse(gate.Task.IsCompleted, "call disposal must not complete or fault the caller's gate");
+ }
+ finally
+ {
+ await session.CloseAsync();
+ // WriteAsync belongs to its caller, who must observe the real cancellation fault even during cleanup.
+ var exception = await Assert.ThrowsExactlyAsync<grpcLib.RpcException>(
+ () => write.WaitAsync(TimeSpan.FromSeconds(5)));
+ Assert.AreEqual(grpcLib.StatusCode.Cancelled, exception.StatusCode);
+ Assert.IsTrue(write.IsFaulted, "disposal must fault the pending write, not report success");
+ }
+ }
+ }
+}