[
https://issues.apache.org/jira/browse/QPID-8750?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Marco Geri updated QPID-8750:
-----------------------------
Description:
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}
was:
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}
> 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: [email protected]
For additional commands, e-mail: [email protected]
