Question - Receiving Messages - acknowledge-mode "transacted"

[rizi]

Hi,
Our customer told us we have to use the acknowledge-mode "transacted" when receiving messages, can you tell me what this means?

I can't find a way to specify this mode when receiving a message, am I doing something wrong?

Here is a small sample which shows how I receive a message

  Endpoint endpoint = Endpoint.Create("localhost", 9898, "test", "test"); 
  ConnectionFactory connectionFactory = new ConnectionFactory();

  await using (IConnection connection = await connectionFactory.CreateAsync(endpoint))
  {

      IConsumer consumer = await connection.CreateConsumerAsync("Test", RoutingType.Anycast);
      Message message = await consumer.ReceiveAsync();

      await consumer.AcceptAsync(message);
  }

[Havret]

Hi @rizi,

It basically means to use transactions for message acknowledgment. It is especially useful when you consume multiple messages, and you want to be sure that they will be acknowledged in all or nothing manner.

Modifying slightly your example, it may look as follows:

await using (IConnection connection = await connectionFactory.CreateAsync(endpoint))
{
    IConsumer consumer = await connection.CreateConsumerAsync("Test", RoutingType.Anycast);
    Message msg1 = await consumer.ReceiveAsync();
    Message msg2 = await consumer.ReceiveAsync();

    // start transaction
    await using (var transaction = new Transaction())
    {
        // ack the first message
        await consumer.AcceptAsync(msg1, transaction);
        
        // ack the second message
        await consumer.AcceptAsync(msg2, transaction);

        // commit the transaction
        await transaction.CommitAsync();
        
        // at this point you're sure that both messages were acknowledged

    }
}

Of course, this is only a small example. You can do much more with transactions. For more use cases, please check out the transactions spec --> https://github.com/Havret/dotnet-activemq-artemis-client/blob/master/test/ActiveMQ.Artemis.Client.IntegrationTests/TransactionsSpec.cs

I hope this answers your question.

[rizi]

Hi @rizi,

It basically means to use transactions for message acknowledgment. It is especially useful when you consume multiple messages, and you want to be sure that they will be acknowledged in all or nothing manner.

Modifying slightly your example, it may look as follows:

await using (IConnection connection = await connectionFactory.CreateAsync(endpoint))
{
    IConsumer consumer = await connection.CreateConsumerAsync("Test", RoutingType.Anycast);
    Message msg1 = await consumer.ReceiveAsync();
    Message msg2 = await consumer.ReceiveAsync();

    // start transaction
    await using (var transaction = new Transaction())
    {
        // ack the first message
        await consumer.AcceptAsync(msg1, transaction);
        
        // ack the second message
        await consumer.AcceptAsync(msg2, transaction);

        // commit the transaction
        await transaction.CommitAsync();
        
        // at this point you're sure that both messages were acknowledged

    }
}

Of course, this is only a small example. You can do much more with transactions. For more use cases, please check out the transactions spec --> https://github.com/Havret/dotnet-activemq-artemis-client/blob/master/test/ActiveMQ.Artemis.Client.IntegrationTests/TransactionsSpec.cs

I hope this answers your question.

Thx for the answer, that makes things much more clearer!

Br

[yashakapkov]

Hello!
I have a question about acknowledgment. How does it work?

I thought that consumer.AcceptAsync says consumer that the message is processed properly and must be dequeued so it disappears from the queue. Thus if an exception is thrown between consumer.ReceiveAsync and consumer.AcceptAsync methods consumer would be able to get the same message again, because the message wasn't acknowledged. However message disappears from the queue right after consumer.ReceiveAsync method call. And after recovery I'm not able to receive unacknowledged message. Where is my mistake?

[Havret]

Hi @yashakapkov,

When you invoke ReceiveAsync, the broker assigns the message to your consumer, making it unavailable to other consumers on the same queue. Here's what can happen next:

1. Accepting the message removes it from the queue, preventing further consumption by others.
2. Rejecting the message returns it to the queue, potentially disrupting the order of messages.
3. If your consumer fails, any messages it consumed but didn't acknowledge will go back to the queue for reprocessing, preserving their order within any message groups.

It's important to note that if you call Receive but fail to acknowledge the message due to an exception in your processing logic, the broker will assume the message is still being processed. This can lead to a risky situation where your consumer might become blocked. For more details, you can refer to this article.

I hope this clarifies things for you.

Best,
Havret

[yashakapkov]

Thank you so much for the answer! It really helped.

Sincerely
Yakov