You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/14 09:56:35 UTC
(pulsar) 06/15: [fix][client] Keep transactional and non-transactional messages in separate batches (#26547)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 89184b53edb9ae9d5fbde66c5ed41528e1ed4aac
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Sun Sep 13 04:46:37 2026 +0800
[fix][client] Keep transactional and non-transactional messages in separate batches (#26547)
Co-authored-by: maxlisongsong <ma...@didiglobal.com>
(cherry picked from commit fa05c3e71d22dc4f4fd69a09385a6c0c5ae5f844)
---
.../broker/transaction/TransactionProduceTest.java | 55 ++++++++
.../client/impl/AbstractBatchMessageContainer.java | 22 ++-
.../client/impl/BatchMessageKeyBasedContainer.java | 13 ++
.../client/impl/BatchMessageContainerImplTest.java | 151 +++++++++++++++++++++
4 files changed, 237 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
index 757f86ec834d..5dc71db4d732 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
@@ -51,6 +51,7 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.api.transaction.Transaction;
@@ -92,6 +93,60 @@ public class TransactionProduceTest extends TransactionTestBase {
super.internalCleanup();
}
+ /**
+ * A batch carries one transaction id in its metadata, so every message in it inherits that transaction.
+ * A plain message batched together with transactional ones is therefore discarded when that transaction
+ * aborts, even though the application never sent it inside a transaction.
+ */
+ @Test
+ public void testAbortedTransactionDoesNotDiscardPlainMessageBatchedWithIt() throws Exception {
+ final String topic = NAMESPACE1 + "/txn-batch-isolation";
+ admin.topics().createNonPartitionedTopic(topic);
+
+ @Cleanup
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("txn-batch-isolation-sub")
+ .subscribe();
+
+ Transaction txn = pulsarClient.newTransaction()
+ .withTransactionTimeout(60, TimeUnit.SECONDS)
+ .build().get();
+
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .sendTimeout(0, TimeUnit.SECONDS)
+ .enableBatching(true)
+ .batchingMaxMessages(100)
+ // Long enough that the two sends under test cannot be split into separate batches by the
+ // timer, short enough that a batch still closes on its own when a flush cannot reach it.
+ .batchingMaxPublishDelay(5, TimeUnit.SECONDS)
+ .create();
+
+ // The first transactional send on a topic waits for the topic to be registered with the transaction
+ // coordinator before the message reaches the batch. Get that round trip out of the way, so the two
+ // sends under test are added to the container synchronously and therefore land in one batch.
+ // A flush issued here would run before the message reaches the batch, so let the batch timer close it.
+ producer.newMessage(txn).value("warm-up").sendAsync().get(30, TimeUnit.SECONDS);
+
+ producer.newMessage(txn).value("in-txn").sendAsync();
+ CompletableFuture<MessageId> plainSend = producer.newMessage().value("plain").sendAsync();
+ producer.flush();
+ plainSend.get(30, TimeUnit.SECONDS);
+
+ // The topic's max read position does not advance past an ongoing transaction, so nothing on this topic
+ // is readable until the transaction ends. Abort it: that discards the transactional messages, and the
+ // plain one must survive because it was never part of the transaction.
+ txn.abort().get(60, TimeUnit.SECONDS);
+
+ Message<String> received = consumer.receive(30, TimeUnit.SECONDS);
+ Assert.assertNotNull(received, "the plain message was discarded by the aborted transaction");
+ Assert.assertEquals(received.getValue(), "plain");
+
+ // Both transactional messages were aborted, so nothing else may arrive.
+ Assert.assertNull(consumer.receive(3, TimeUnit.SECONDS));
+ }
@Test
public void produceAndCommitTest() throws Exception {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractBatchMessageContainer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractBatchMessageContainer.java
index 3ba7866350a0..9f068dbb0969 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractBatchMessageContainer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractBatchMessageContainer.java
@@ -120,14 +120,28 @@ public abstract class AbstractBatchMessageContainer implements BatchMessageConta
this.maxBytesInBatch = producer.getConfiguration().getBatchingMaxBytes();
}
+ /**
+ * Whether {@code msg} belongs to the same transaction as the messages already in this batch.
+ *
+ * <p>A batch carries a single transaction id in its metadata, so every message in it inherits that
+ * transaction. "No transaction" is therefore an identity of its own and is not compatible with any
+ * transaction: mixing the two in one batch would either enroll a plain message in a transaction (invisible
+ * until commit, dropped on abort) or publish a transactional message outside its transaction.
+ *
+ * <p>This is a pure query. The batch adopts its transaction id from its first message in {@code add}.
+ */
@Override
public boolean hasSameTxn(MessageImpl<?> msg) {
- if (!msg.getMessageBuilder().hasTxnidMostBits() || !msg.getMessageBuilder().hasTxnidLeastBits()) {
+ if (numMessagesInBatch == 0) {
return true;
}
- if (currentTxnidMostBits == -1 || currentTxnidLeastBits == -1) {
- currentTxnidMostBits = msg.getMessageBuilder().getTxnidMostBits();
- currentTxnidLeastBits = msg.getMessageBuilder().getTxnidLeastBits();
+ boolean msgHasTxn = msg.getMessageBuilder().hasTxnidMostBits()
+ && msg.getMessageBuilder().hasTxnidLeastBits();
+ boolean batchHasTxn = currentTxnidMostBits != -1L && currentTxnidLeastBits != -1L;
+ if (msgHasTxn != batchHasTxn) {
+ return false;
+ }
+ if (!msgHasTxn) {
return true;
}
return currentTxnidMostBits == msg.getMessageBuilder().getTxnidMostBits()
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java
index f6e3f5a683ed..96074b43466f 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java
@@ -46,6 +46,19 @@ class BatchMessageKeyBasedContainer extends AbstractBatchMessageContainer {
log.debug("[{}] [{}] add message to batch, num messages in batch so far is {}", topicName,
producer.getProducerName(), numMessagesInBatch);
}
+ 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.
+ //
+ // Both fields are assigned unconditionally, so a plain first message resets them. An inner batch
+ // whose first add fails to allocate clears itself and stays empty, which leaves the count at zero
+ // without clearing this container: a leftover transaction id would then be inherited by the next,
+ // plain, first message and would wrongly admit later messages of that transaction.
+ boolean msgHasTxn = msg.getMessageBuilder().hasTxnidMostBits()
+ && msg.getMessageBuilder().hasTxnidLeastBits();
+ currentTxnidMostBits = msgHasTxn ? msg.getMessageBuilder().getTxnidMostBits() : -1L;
+ currentTxnidLeastBits = msgHasTxn ? msg.getMessageBuilder().getTxnidLeastBits() : -1L;
+ }
String key = getKey(msg);
final BatchMessageContainerImpl batchMessageContainer = batches.computeIfAbsent(key,
__ -> new BatchMessageContainerImpl(producer));
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java
index 3b2ac54b4bda..4b1aa23ac4f9 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BatchMessageContainerImplTest.java
@@ -20,10 +20,13 @@ package org.apache.pulsar.client.impl;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import io.netty.buffer.ByteBufAllocator;
@@ -199,4 +202,152 @@ public class BatchMessageContainerImplTest {
batchMessageContainer.clear();
messages.forEach(ReferenceCountUtil::safeRelease);
}
+
+ private ProducerImpl<?> createTestProducer() throws Exception {
+ ProducerImpl<?> producer = mock(ProducerImpl.class);
+ ProducerConfigurationData producerConfigurationData = new ProducerConfigurationData();
+ producerConfigurationData.setCompressionType(CompressionType.NONE);
+ PulsarClientImpl pulsarClient = mock(PulsarClientImpl.class);
+ when(pulsarClient.getCnxPool()).thenReturn(mock(ConnectionPool.class));
+ when(pulsarClient.getMemoryLimitController()).thenReturn(mock(MemoryLimitController.class));
+ Field clientField = HandlerState.class.getDeclaredField("client");
+ clientField.setAccessible(true);
+ clientField.set(producer, pulsarClient);
+ when(producer.getConfiguration()).thenReturn(producerConfigurationData);
+ when(producer.encryptMessage(any(), any())).thenAnswer(__ -> ByteBufAllocator.DEFAULT.buffer()
+ .writeBytes("payload".getBytes(StandardCharsets.UTF_8)));
+ return producer;
+ }
+
+ /**
+ * 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());
+ try {
+ container.add(newMessage(1L, 7L, 13L), null);
+
+ assertFalse(container.hasSameTxn(newMessage(2L, null, null)),
+ "a plain message was accepted into a transactional batch");
+ } finally {
+ container.discard(null);
+ }
+ }
+
+ /**
+ * The mirror image: a transactional message joining a batch that already holds plain messages stamps its
+ * transaction id onto the whole batch, retroactively pulling those plain messages into the transaction.
+ */
+ @Test
+ public void testTransactionalMessageIsNotBatchedWithPlainMessages() throws Exception {
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(createTestProducer());
+ try {
+ container.add(newMessage(1L, null, null), null);
+
+ assertFalse(container.hasSameTxn(newMessage(2L, 7L, 13L)),
+ "a transactional message was accepted into a plain batch");
+ } finally {
+ container.discard(null);
+ }
+ }
+
+ /**
+ * Key-based batching 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, so a transactional message landing in
+ * a key bucket whose first message was plain is published with no transaction id at all — outside the
+ * transaction, and not rolled back on abort.
+ */
+ @Test
+ public void testKeyBasedContainerKeepsTransactionalAndPlainMessagesApart() throws Exception {
+ BatchMessageKeyBasedContainer container = new BatchMessageKeyBasedContainer();
+ container.setProducer(createTestProducer());
+ try {
+ container.add(newMessage(1L, null, null), null);
+
+ assertFalse(container.hasSameTxn(newMessage(2L, 7L, 13L)),
+ "a transactional message was accepted into a plain key-based batch");
+ } finally {
+ container.discard(null);
+ }
+ }
+
+ /** Messages of the same transaction must still batch together, and two different transactions must not. */
+ @Test
+ public void testSameTransactionStillBatchesTogether() throws Exception {
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(createTestProducer());
+ try {
+ container.add(newMessage(1L, 7L, 13L), null);
+
+ assertTrue(container.hasSameTxn(newMessage(2L, 7L, 13L)),
+ "two messages of the same transaction were split across batches");
+ assertFalse(container.hasSameTxn(newMessage(3L, 7L, 14L)),
+ "messages of two different transactions were put in one batch");
+ } finally {
+ container.discard(null);
+ }
+ }
+
+ /** Plain messages must still batch with each other. */
+ @Test
+ public void testPlainMessagesStillBatchTogether() throws Exception {
+ BatchMessageContainerImpl container = new BatchMessageContainerImpl(createTestProducer());
+ try {
+ container.add(newMessage(1L, null, null), null);
+
+ assertTrue(container.hasSameTxn(newMessage(2L, null, null)),
+ "two plain messages were split across batches");
+ } finally {
+ container.discard(null);
+ }
+ }
+
+ private static MessageImpl<byte[]> newMessage(long sequenceId, Long txnIdMostBits, Long txnIdLeastBits) {
+ MessageMetadata metadata = new MessageMetadata();
+ metadata.setSequenceId(sequenceId);
+ metadata.setProducerName("producer1");
+ metadata.setPublishTime(System.currentTimeMillis());
+ if (txnIdMostBits != null) {
+ metadata.setTxnidMostBits(txnIdMostBits);
+ metadata.setTxnidLeastBits(txnIdLeastBits);
+ }
+ ByteBuffer payload = ByteBuffer.wrap(("payload-" + sequenceId).getBytes(StandardCharsets.UTF_8));
+ return MessageImpl.create(metadata, payload, Schema.BYTES, null);
+ }
+
+ /**
+ * An inner batch whose first add fails to allocate clears itself and stays empty, so the outer container's
+ * message count is never incremented and the outer is not cleared either. Any transaction identity captured
+ * for that failed message must therefore not survive into the next first message, or a later message of that
+ * transaction would be admitted into a plain batch and published outside its transaction. Allocation failure
+ * is a recoverable path the client already handles.
+ */
+ @Test
+ public void testKeyBasedContainerDropsTheTransactionIdentityOfAFailedFirstAdd() throws Exception {
+ ProducerImpl<?> producer = createTestProducer();
+ MemoryLimitController memoryLimitController = producer.client.getMemoryLimitController();
+ // fail only the first inner-batch allocation
+ doThrow(new OutOfMemoryError("test")).doNothing()
+ .when(memoryLimitController).forceReserveMemory(anyLong());
+
+ BatchMessageKeyBasedContainer container = new BatchMessageKeyBasedContainer();
+ container.setProducer(producer);
+ try {
+ container.add(newMessage(1L, 7L, 13L), null);
+ assertEquals(container.getNumMessagesInBatch(), 0,
+ "the failed add should have left the container empty");
+
+ container.add(newMessage(2L, null, null), null);
+
+ assertFalse(container.hasSameTxn(newMessage(3L, 7L, 13L)),
+ "a transactional message was admitted into a plain batch through the transaction identity"
+ + " left behind by a failed add");
+ } finally {
+ container.discard(null);
+ }
+ }
+
}