[jira] [Updated] (QPID-8750) [Broker-J] AsyncAutoCommitTransaction can enqueue messages out of order when a queue's messageDurability forces storage of non-persistent messages

["Marco Geri (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/QPID-8750?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Marco Geri updated QPID-8750:
-----------------------------
    Summary: [Broker-J] AsyncAutoCommitTransaction can enqueue messages out of order when a queue's messageDurability forces storage of non-persistent messages  (was: AsyncAutoCommitTransaction can enqueue messages out of order when a queue's messageDurability forces storage of non-persistent messages)

> [Broker-J] AsyncAutoCommitTransaction can enqueue messages out of order when a queue's messageDurability forces storage of non-persistent messages
> --------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: QPID-8750
>                 URL: https://issues.apache.org/jira/browse/QPID-8750
>             Project: Qpid
>          Issue Type: Bug
>          Components: Broker-J
>    Affects Versions: qpid-java-broker-10.0.1
>            Reporter: Marco Geri
>            Priority: Major
>
> When a burst of non-persistent (transient) messages is published to a queue whose _messageDurability_ attribute is {_}ALWAYS{_}, the messages are occasionally enqueued in an order different from the order in which they were received on the connection.
> _AsyncAutoCommitTransaction_ decides whether to write to the message store using the queue's durability policy combined with the message's delivery mode:
> {code:java}
> if (queue.getMessageDurability().persist(message.isPersistent()))
> {
>     txn = _messageStore.newTransaction();
>     ...
>     future = txn.commitTranAsync(null);
> } {code}
> but it then tells _addEnqueueFuture()_ whether the operation was persisted using the message's delivery mode alone:
> {code:java}
> addEnqueueFuture(future, new Action() { ... }, message.isPersistent()); {code}
> _addEnqueueFuture()_ relies on that flag to decide whether it is safe to invoke _postCommit()_ (which performs the actual enqueue onto the queue) synchronously:
> {code:java}
> private void addEnqueueFuture(final CompletableFuture<Void> future, final Action action, boolean persistent)
> {
>     if (action != null)
>     {
>         // For persistent messages, do not synchronously invoke postCommit even if the future is completed.
>         // Otherwise, postCommit (which actually does the enqueuing) might be called on successive messages out of order.
>         if (future.isDone() && !persistent && !_strictOrderWithMixedDeliveryMode)
>         {
>             action.postCommit();
>         }
>         else
>         {
>             _futureRecorder.recordFuture(future, action);
>         }
>     }
> } {code}
> The two values diverge whenever the queue's durability policy overrides the message's delivery mode. With _MessageDurability.ALWAYS_ and a non-persistent message, _MessageDurability.persist(false)_ returns true, so a store transaction is created and _commitTranAsync()_ returns a genuinely asynchronous future (for the BDB store, one completed later by the coalescing committer thread). However, persistent is passed as false, so addEnqueueFuture() takes the synchronous fast path as soon as the future happens to be already done.
> The result is a race: message N+1, whose commit future happens to be complete at the moment it is checked, is enqueued immediately, while message N, whose future is still pending, is deferred to the FutureRecorder (e.g. {_}AMQChannel._unfinishedCommandsQueue{_}) and enqueued only when that future completes. Consumers therefore see N+1 before N.
> Note that this is not the documented mixed-delivery-mode trade-off that _qpid.strict_order_with_mixed_delivery_mode_ addresses: here reordering occurs within a uniform stream of non-persistent messages, purely because of the queue's durability configuration.
> The same defect exists in both _enqueue()_ overloads of {_}AsyncAutoCommitTransaction{_}. The _Collection<? extends BaseQueue>_ overload is the one used by the normal publishing path _(RoutingResult.send() → ServerTransaction.enqueue(Collection, ...)),_ so it is the more commonly hit of the two.
> The inverse mismatch is also wrong, though harmless: with _messageDurability=NEVER_ and a persistent message, no store write happens, but _persistent=true_ causes an already-completed future to be handed to the FutureRecorder, needlessly deferring _postCommit()_ until the next sync().
> Steps to reproduce:
>  # Create a durable queue with messageDurability=ALWAYS on a virtualhost backed by the BDB message store (default coalescing-sync commit behaviour).
>  # Publish a burst of non-persistent messages with a monotonically increasing sequence number in an application property, without waiting
> Occasionally the sequence is out of order. The frequency depends on store commit timing, so it is easier to reproduce under load or on slower storage.
> Suggested fix:
> Pass the value actually used to decide whether a store write was issued, rather than {_}message.isPersistent(){_}.
> For {_}*enqueue(TransactionLogResource, EnqueueableMessage, EnqueueAction)*{_}:
> {code:java}
> CompletableFuture<Void> future;
> final MessageEnqueueRecord enqueueRecord;
> final boolean persist = queue.getMessageDurability().persist(message.isPersistent());
> if (persist)
> {
>     LOGGER.debug("Enqueue of message number {} to transaction log. Queue : {}",
>                  message.getMessageNumber(), queue.getName());
>     txn = _messageStore.newTransaction();
>     enqueueRecord = txn.enqueueMessage(queue, message);
>     future = txn.commitTranAsync(null);
>     txn = null;
> }
> else
> {
>     future = CompletableFuture.completedFuture(null);
>     enqueueRecord = null;
> }
> final EnqueueAction underlying = postTransactionAction;
> addEnqueueFuture(future, new Action() { /* unchanged */ }, persist); {code}
> For {*}_enqueue(Collection<? extends BaseQueue>, EnqueueableMessage, EnqueueAction)_{*}, the equivalent condition is whether a store transaction was opened for any of the target queues:
> {code:java}
> CompletableFuture<Void> future;
> final boolean persist = txn != null;
> if (persist)
> {
>     future = txn.commitTranAsync(null);
>     txn = null;
> }
> else
> {
>     future = CompletableFuture.completedFuture(null);
> }
> final EnqueueAction underlying = postTransactionAction;
> addEnqueueFuture(future, new Action() { /* unchanged */ }, persist); {code}
>  



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org