You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "SongOf (via GitHub)" <gi...@apache.org> on 2026/09/11 15:53:14 UTC
[PR] [fix][client] Keep transactional and non-transactional messages in separate batches [pulsar]
SongOf opened a new pull request, #26547:
URL: https://github.com/apache/pulsar/pull/26547
### Motivation
A batch carries a single transaction id in its metadata, so every message inside it inherits that
transaction — there is no per-message override:
```java
// BatchMessageContainerImpl.createOpSendMsg
if (currentTxnidMostBits != -1) {
messageMetadata.setTxnidMostBits(currentTxnidMostBits);
}
```
`hasSameTxn`, the admission check meant to enforce that invariant, cannot express it:
```java
public boolean hasSameTxn(MessageImpl<?> msg) {
if (!msg.getMessageBuilder().hasTxnidMostBits() || !msg.getMessageBuilder().hasTxnidLeastBits()) {
return true; // (1)
}
if (currentTxnidMostBits == -1 || currentTxnidLeastBits == -1) {
currentTxnidMostBits = msg.getMessageBuilder().getTxnidMostBits(); // (2) mutates the batch
currentTxnidLeastBits = msg.getMessageBuilder().getTxnidLeastBits();
return true;
}
return currentTxnidMostBits == msg.getMessageBuilder().getTxnidMostBits() // (3)
&& currentTxnidLeastBits == msg.getMessageBuilder().getTxnidLeastBits();
}
```
Branch (1) inspects only the message, never the batch, so a plain message is admitted into any batch
including a transactional one. Branch (2) performs a write inside a `hasSameXxx` query, and makes a
plain batch adopt the transaction of the first transactional message merely *offered* to it. Only
branch (3) actually compares, and it is reachable only when both sides carry a transaction. The check
can answer "are these two transactions the same?" but never "is one of them absent?".
Three consequences, all reachable from a single batching producer used for both `newMessage(txn)` and
`newMessage()` — legal API usage, with batching enabled by default:
1. A plain message batched after transactional ones inherits the transaction: it stays invisible until
commit and is **silently discarded if the transaction aborts**, although the application never sent
it inside a transaction.
2. A transactional message joining a batch that already holds plain messages stamps its transaction
onto the whole batch, retroactively enrolling those plain messages in it.
3. The side effect in branch (2) also fires on the duplicate-sequence-id path in `ProducerImpl`:
`canAddToCurrentBatch` adopts the transaction, and the caller then flushes the **current** batch via
`doBatchSendAndAdd` and puts the new message into a fresh one — so a batch of plain messages is
published stamped with a transaction that none of its messages belongs to.
`BatchMessageKeyBasedContainer` fails the other way round. The guard reads the outer container's
transaction state, while the batch metadata is built by the per-key inner container from *its own*
first message. A transactional message landing in a key bucket whose first message was plain is
therefore published with no transaction id at all — outside the transaction, and not rolled back when
it aborts.
For reference, the sibling admission check on the same line of `canAddToCurrentBatch` is already the
correct shape: `hasSameSchema` is a pure query that starts with `numMessagesInBatch == 0` and treats
"absent" as an identity that must match.
### Modifications
- `AbstractBatchMessageContainer.hasSameTxn` is now a pure query comparing transaction identities, with
"no transaction" as an identity of its own that is incompatible with any transaction. An empty batch
(`numMessagesInBatch == 0`) still accepts anything; testing the message count rather than the `-1`
sentinel is what distinguishes "no identity yet" from "identity is *no transaction*". Adopting the
identity is left solely to `add`, where it already happened — which also removes consequence 3 above.
- `BatchMessageKeyBasedContainer.add` captures the transaction id from the container's first message.
This is required rather than cosmetic: with the side effect gone, the outer container's transaction
state would otherwise stay unset forever, `batchHasTxn` would always be false, and plain messages
would again be admitted into a transactional container.
Nothing changes at the call site. When the guard rejects a message, `ProducerImpl.doBatchSendAndAdd`
already flushes the current batch and opens a new one with that message, exactly as it does today when
the batch is full or the schema differs.
Applications that mix both kinds of send on one producer will see batches split more often. That is the
necessary cost: the two cannot share a batch without one of them inheriting the wrong transaction
identity. There is no wire format or API change, and no behaviour change for applications that do not
mix.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- *Added `BatchMessageContainerImplTest.testPlainMessageIsNotBatchedWithTransactionalMessages` and
`...testTransactionalMessageIsNotBatchedWithPlainMessages`, covering both directions, and
`...testKeyBasedContainerKeepsTransactionalAndPlainMessagesApart` for the key-based batcher. All
three fail before the change.*
- *Added `BatchMessageContainerImplTest.testSameTransactionStillBatchesTogether` and
`...testPlainMessagesStillBatchTogether` as guards that messages of one transaction still share a
batch, that two different transactions still do not, and that plain messages still batch together.*
- *Added `TransactionProduceTest.testAbortedTransactionDoesNotDiscardPlainMessageBatchedWithIt`: a
batching producer sends a transactional and a plain message into the same batch, the transaction is
aborted, and the plain message must still be readable. It fails before the change with "the plain
message was discarded by the aborted transaction".*
- *`BatchMessageContainerImplTest` passes 9/9, the `pulsar-client` module suite shows no new failures,
and `./gradlew quickCheck` is clean.*
### Does this pull request potentially affect one of the following parts:
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Keep transactional and non-transactional messages in separate batches [pulsar]
Posted by "SongOf (via GitHub)" <gi...@apache.org>.
SongOf commented on code in PR #26547:
URL: https://github.com/apache/pulsar/pull/26547#discussion_r3996099363
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java:
##########
@@ -46,6 +46,14 @@ public boolean add(MessageImpl<?> msg, SendCallback callback) {
.attr("producerName", producer.getProducerName())
.attr("numMessagesInBatch", numMessagesInBatch)
.log("add message to batch");
+ if (numMessagesInBatch == 0) {
+ // The whole container shares one transaction identity; hasSameTxn keeps later messages consistent
+ // with it. Capturing it here rather than in hasSameTxn keeps that check side-effect free.
+ if (msg.getMessageBuilder().hasTxnidMostBits() && msg.getMessageBuilder().hasTxnidLeastBits()) {
+ currentTxnidMostBits = msg.getMessageBuilder().getTxnidMostBits();
+ currentTxnidLeastBits = msg.getMessageBuilder().getTxnidLeastBits();
+ }
Review Comment:
Confirmed and fixed in d18e8a1b2f9. The identity is now assigned unconditionally for every first message, so a plain one resets it to -1.
The path is exactly as you describe, and the existing comment in `add` documents the shape of it: when the inner `add` fails it clears itself and reports empty, so the outer never increments its count and is never cleared either, while the identity had already been captured before the delegation. The next message then enters through the empty-container check, and if it is plain the stale id survives and admits a later message of that transaction into a bucket whose inner metadata carries no transaction id. Note the single-message container does not have this problem, because its own `clear()` resets the count and the identity together.
Added `testKeyBasedContainerDropsTheTransactionIdentityOfAFailedFirstAdd`, which fails the first inner allocation through `forceReserveMemory` and then asserts the sequence you asked for: failed transactional add, successful plain add, same-transaction admission refused. It fails without the change with "a transactional message was admitted into a plain batch through the transaction identity left behind by a failed add".
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java:
##########
@@ -230,4 +231,84 @@ private void addMessagesAndCreateOpSendMsg(BatchMessageContainerImpl batchMessag
batchMessageContainer.clear();
messages.forEach(ReferenceCountUtil::safeRelease);
}
+
+ /**
+ * A batch carries one transaction id in its metadata, so every message in it inherits that transaction.
+ * Mixing a plain message into a transactional batch therefore silently enrolls it in the transaction: it
+ * stays invisible until commit and is dropped on abort, even though the application never sent it inside
+ * a transaction.
+ */
+ @Test
+ public void testPlainMessageIsNotBatchedWithTransactionalMessages() throws Exception {
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(createTestProducer());
+ container.add(newMessage(1L, 7L, 13L), null);
+
+ assertFalse(container.hasSameTxn(newMessage(2L, null, null)),
+ "a plain message was accepted into a transactional batch");
Review Comment:
Done in the same commit. All five tests now discard their container in a `finally`, so the batch buffer allocated by the first `add` is released even when an assertion fails. The new recovery test does the same.
`BatchMessageContainerImplTest` passes 10/10, the `pulsar-client` module suite (417 tests) shows no new failures, and `quickCheck` is clean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Keep transactional and non-transactional messages in separate batches [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #26547:
URL: https://github.com/apache/pulsar/pull/26547
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Keep transactional and non-transactional messages in separate batches [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26547:
URL: https://github.com/apache/pulsar/pull/26547#discussion_r3997180021
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java:
##########
@@ -46,6 +46,14 @@ public boolean add(MessageImpl<?> msg, SendCallback callback) {
.attr("producerName", producer.getProducerName())
.attr("numMessagesInBatch", numMessagesInBatch)
.log("add message to batch");
+ if (numMessagesInBatch == 0) {
+ // The whole container shares one transaction identity; hasSameTxn keeps later messages consistent
+ // with it. Capturing it here rather than in hasSameTxn keeps that check side-effect free.
+ if (msg.getMessageBuilder().hasTxnidMostBits() && msg.getMessageBuilder().hasTxnidLeastBits()) {
+ currentTxnidMostBits = msg.getMessageBuilder().getTxnidMostBits();
+ currentTxnidLeastBits = msg.getMessageBuilder().getTxnidLeastBits();
+ }
Review Comment:
Thanks, this addresses the recovery path. [`BatchMessageKeyBasedContainer.java:49-60`](https://github.com/apache/pulsar/blob/d18e8a1b2f97014b54fa35ed8a71a7a37a2aeff8/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java#L49-L60) now assigns both transaction-id fields on every first add, so a plain message resets the identity left by a failed allocation. `testKeyBasedContainerDropsTheTransactionIdentityOfAFailedFirstAdd` covers the requested sequence.
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java:
##########
@@ -230,4 +231,84 @@ private void addMessagesAndCreateOpSendMsg(BatchMessageContainerImpl batchMessag
batchMessageContainer.clear();
messages.forEach(ReferenceCountUtil::safeRelease);
}
+
+ /**
+ * A batch carries one transaction id in its metadata, so every message in it inherits that transaction.
+ * Mixing a plain message into a transactional batch therefore silently enrolls it in the transaction: it
+ * stays invisible until commit and is dropped on abort, even though the application never sent it inside
+ * a transaction.
+ */
+ @Test
+ public void testPlainMessageIsNotBatchedWithTransactionalMessages() throws Exception {
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(createTestProducer());
+ container.add(newMessage(1L, 7L, 13L), null);
+
+ assertFalse(container.hasSameTxn(newMessage(2L, null, null)),
+ "a plain message was accepted into a transactional batch");
Review Comment:
Thanks, the five original tests now discard their containers in `finally`, and [`BatchMessageContainerImplTest.java:343-365`](https://github.com/apache/pulsar/blob/d18e8a1b2f97014b54fa35ed8a71a7a37a2aeff8/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java#L343-L365) gives the new recovery test the same cleanup. The allocated batch buffers are released even when an assertion fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Keep transactional and non-transactional messages in separate batches [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26547:
URL: https://github.com/apache/pulsar/pull/26547#discussion_r3993268825
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java:
##########
@@ -230,4 +231,84 @@ private void addMessagesAndCreateOpSendMsg(BatchMessageContainerImpl batchMessag
batchMessageContainer.clear();
messages.forEach(ReferenceCountUtil::safeRelease);
}
+
+ /**
+ * A batch carries one transaction id in its metadata, so every message in it inherits that transaction.
+ * Mixing a plain message into a transactional batch therefore silently enrolls it in the transaction: it
+ * stays invisible until commit and is dropped on abort, even though the application never sent it inside
+ * a transaction.
+ */
+ @Test
+ public void testPlainMessageIsNotBatchedWithTransactionalMessages() throws Exception {
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(createTestProducer());
+ container.add(newMessage(1L, 7L, 13L), null);
+
+ assertFalse(container.hasSameTxn(newMessage(2L, null, null)),
+ "a plain message was accepted into a transactional batch");
Review Comment:
**[QUALITY] Release the batch buffers allocated by the new tests**
Please discard each container in a `finally` block in these five new tests. The first `add` allocates a batch buffer, and these tests neither send nor discard it, so each leaves that buffer unreleased. `discard(null)` releases it even when an assertion fails; `clear()` alone does not.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java:
##########
@@ -46,6 +46,14 @@ public boolean add(MessageImpl<?> msg, SendCallback callback) {
.attr("producerName", producer.getProducerName())
.attr("numMessagesInBatch", numMessagesInBatch)
.log("add message to batch");
+ if (numMessagesInBatch == 0) {
+ // The whole container shares one transaction identity; hasSameTxn keeps later messages consistent
+ // with it. Capturing it here rather than in hasSameTxn keeps that check side-effect free.
+ if (msg.getMessageBuilder().hasTxnidMostBits() && msg.getMessageBuilder().hasTxnidLeastBits()) {
+ currentTxnidMostBits = msg.getMessageBuilder().getTxnidMostBits();
+ currentTxnidLeastBits = msg.getMessageBuilder().getTxnidLeastBits();
+ }
Review Comment:
**[INTENT MISMATCH] Reset the transaction identity when the first message is plain**
Could this initialize both fields for every first message, including setting them to `-1L` for a plain message? If the first transactional message fails to allocate its inner batch buffer, the inner container catches the failure and clears itself, but the outer container retains this transaction id with a message count of zero. Before the batch timer runs, a plain message can then enter through the empty-container check without resetting that id. A subsequent message from the same transaction passes `hasSameTxn` and joins the plain message's key bucket, whose inner metadata has no transaction id, so it is published outside the transaction. This leaves an existing recovery case unfixed; the client already handles allocation failure as recoverable in `recoveryAfterOom`. Please cover failed transactional add → successful plain add → same-transaction admission in the key-based regression tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org