Azure Service Bus Messages Marked as Completed Even When Persistence Fails

[oussama-smida]

Description:
In CAP 8.2, when using Azure Service Bus and SQL Server, messages are marked as completed even if the system fails to persist the received message to the database.
This issue has been observed as a rare occurrence but poses a considerable risk to the reliability of the system, as it can lead to message loss.

Analysis:
The issue occurs because the default setting for AutoCompleteMessages in CAP is true. When an exception occurs during message persistence (e.g., SQL timeout), CAP attempts to call the Reject method. However, the current implementation of Reject for Azure Service Bus is a no-op, resulting in the message being automatically completed despite the failure.

Trace Log:

An exception occurred when processing the received message. Message:'{0}'.
Source context: DotNetCore.CAP.Internal.ConsumerRegister

Exception:

Microsoft.Data.SqlClient.SqlException (0x80131904): Execution Timeout Expired. The timeout period elapsed prior to completion of the operation or the server is not responding.
 ---> System.ComponentModel.Win32Exception (258): Unknown error 258
   at Microsoft.Data.SqlClient.SqlConnection.OnError(SqlException exception, Boolean breakConnection, Action`1 wrapCloseInAction)
...

Sql Command:

INSERT INTO cap.Received([Id],[Version],[Name],[Group],[Content],[Retries],[Added],[ExpiresAt],[StatusName]) VALUES(@Id,'v1',@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);

Relevant source code
DotNetCore.CAP.Internal.ConsumerRegister

try
{
...
}
catch (Exception e)
{
    _logger.LogError(e, "An exception occurred when process received message. Message:'{0}'.",
        transportMessage);

    client.Reject(sender); // NO-OP in the Azure Service Bus IConsumerClient implementation

    TracingError(tracingTimestamp, transportMessage, client.BrokerAddress, e);
}

Proposed Fix:
DotNetCore.CAP.AzureServiceBus.AzureServiceBusConsumerClient

public void Reject(object? sender)
{
    var commitInput = (AzureServiceBusConsumerCommitInput)sender!;
    commitInput.AbandonMessageAsync().GetAwaiter().GetResult();
}

[yang-xiaodong]

Fixed in version 8.3.0-preview-243613753