You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2025/07/07 09:42:50 UTC
(pulsar) branch branch-4.0 updated: [fix][txn] Fix negative unacknowledged messages in transactions by ensuring that the batch size is added into CommandAck (#24443)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 50da3cc2a62 [fix][txn] Fix negative unacknowledged messages in transactions by ensuring that the batch size is added into CommandAck (#24443)
50da3cc2a62 is described below
commit 50da3cc2a62be642a9e638d7aadbcb315c97731b
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Tue Jun 24 15:51:54 2025 +0800
[fix][txn] Fix negative unacknowledged messages in transactions by ensuring that the batch size is added into CommandAck (#24443)
(cherry picked from commit 23a2ef29a2de488375d2607600bfadec3e51b330)
---
.../client/impl/TransactionEndToEndTest.java | 71 +++++++++++++++++++++-
...ransactionEndToEndWithoutBatchIndexAckTest.java | 13 ++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 1 +
3 files changed, 84 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
index 05245abf374..afdd0d44c4c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
@@ -46,6 +46,7 @@ import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.TransactionMetadataStoreService;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Topic;
@@ -455,8 +456,76 @@ public class TransactionEndToEndTest extends TransactionTestBase {
log.info("finished test partitionAbortTest");
}
+ @DataProvider
+ public Object[][] unackMessagesCountParams() {
+ return new Object[][] {
+ // batchSend, batchAck, asyncAck
+ {Boolean.TRUE, Boolean.TRUE, Boolean.TRUE},
+ {Boolean.TRUE, Boolean.TRUE, Boolean.FALSE},
+ {Boolean.TRUE, Boolean.FALSE, Boolean.TRUE},
+ {Boolean.TRUE, Boolean.FALSE, Boolean.FALSE},
+ {Boolean.FALSE, Boolean.TRUE, Boolean.TRUE},
+ {Boolean.FALSE, Boolean.TRUE, Boolean.FALSE},
+ {Boolean.FALSE, Boolean.FALSE, Boolean.TRUE},
+ {Boolean.FALSE, Boolean.FALSE, Boolean.FALSE}
+ };
+ }
+
+ @Test(dataProvider= "unackMessagesCountParams")
+ public void testUnackMessageAfterAckAllMessages(boolean batchSend, boolean batchAck, boolean asyncAck)
+ throws Exception {
+ final int messageCount = 50;
+ final String subName = "s1";
+ final String topicName = BrokerTestUtil.newUniqueName(NAMESPACE1 + "/tp");
+ // Create producer and consumer.
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
+ .subscriptionType(SubscriptionType.Shared).isAckReceiptEnabled(true).subscribe();
+ Awaitility.await().until(consumer::isConnected);
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).enableBatching(batchSend)
+ .batchingMaxMessages(10).create();
+ Awaitility.await().until(producer::isConnected);
+ // Publish messages.
+ CompletableFuture<MessageId> latestSendFuture = null;
+ for (int i = 0; i < messageCount; i++) {
+ latestSendFuture = producer.sendAsync((i + "").getBytes());
+ }
+ latestSendFuture.join();
+
+ // Ack messages with TXN.
+ Transaction txn = getTxn();
+ List<MessageId> msgIds = new ArrayList<>();
+ for (int i = 0; i < messageCount; i++) {
+ Message<byte[]> message = consumer.receive(waitTimeForCanReceiveMsgInSec, TimeUnit.SECONDS);
+ msgIds.add(message.getMessageId());
+ }
+ if (batchAck) {
+ consumer.acknowledgeAsync(msgIds, txn).get();
+ } else if (asyncAck) {
+ CompletableFuture<Void> latestAckFuture = null;
+ for (MessageId msgId : msgIds) {
+ latestAckFuture = consumer.acknowledgeAsync(msgId, txn);
+ }
+ latestAckFuture.join();
+ } else {
+ for (MessageId msgId : msgIds) {
+ consumer.acknowledgeAsync(msgId, txn).get();
+ }
+ }
+ txn.commit().get();
+
+ // Verify: the quantity of un-ack messages is 0.
+ int unAckMsgs = admin.topics().getStats(topicName).getSubscriptions().get(subName).getConsumers().get(0)
+ .getUnackedMessages();
+ assertEquals(unAckMsgs, 0);
+
+ // cleanup.
+ producer.close();
+ consumer.close();
+ admin.topics().delete(topicName, true);
+ }
+
@Test(dataProvider="enableBatch")
- private void testAckWithTransactionReduceUnAckMessageCount(boolean enableBatch) throws Exception {
+ public void testAckWithTransactionReduceUnAckMessageCount(boolean enableBatch) throws Exception {
final int messageCount = 50;
final String subName = "testAckWithTransactionReduceUnAckMessageCount";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java
index df4ad32b6a8..b8c1bd10763 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java
@@ -43,4 +43,17 @@ public class TransactionEndToEndWithoutBatchIndexAckTest extends TransactionEndT
conf.setAcknowledgmentAtBatchIndexLevelEnabled(true);
txnAckTest(true, 200, SubscriptionType.Failover);
}
+
+ @Override
+ @Test(dataProvider="unackMessagesCountParams", enabled = false)
+ public void testUnackMessageAfterAckAllMessages(boolean batchSend, boolean batchAck, boolean asyncAck)
+ throws Exception {
+ super.testUnackMessageAfterAckAllMessages(batchSend, batchAck, asyncAck);
+ }
+
+ @Override
+ @Test(dataProvider="enableBatch", enabled = false)
+ public void testAckWithTransactionReduceUnAckMessageCount(boolean enableBatch) throws Exception {
+ super.testAckWithTransactionReduceUnAckMessageCount(enableBatch);
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index dfc217cf574..c502928a4df 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -3251,6 +3251,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (MessageIdAdvUtils.isBatch(messageIdAdv)) {
final BitSetRecyclable bitSetRecyclable = BitSetRecyclable.create();
bitSetRecyclable.set(0, messageIdAdv.getBatchSize());
+ messageIdData.setBatchSize(messageIdAdv.getBatchSize());
if (ackType == AckType.Cumulative) {
MessageIdAdvUtils.acknowledge(messageIdAdv, false);
bitSetRecyclable.clear(0, messageIdAdv.getBatchIndex() + 1);