You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/06/08 16:12:57 UTC
(pulsar) branch master updated: [improve][test] PIP-473: extensive v5 transaction tests on scalable topics (#25958)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 0daa5aaeeb0 [improve][test] PIP-473: extensive v5 transaction tests on scalable topics (#25958)
0daa5aaeeb0 is described below
commit 0daa5aaeeb0411c775cd7212a52df1737f7e0d75
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Jun 8 09:12:49 2026 -0700
[improve][test] PIP-473: extensive v5 transaction tests on scalable topics (#25958)
---
.../client/api/v5/V5TransactionRecoveryTest.java | 219 ++++++++++++
.../client/api/v5/V5TransactionScalableTest.java | 382 +++++++++++++++++++++
.../pulsar/client/api/v5/V5TransactionTest.java | 13 +-
.../transaction/TcMetadataDiscoveryTestBase.java | 10 +-
.../V5ScalableTopicTransactionOnOxiaTest.java | 46 +++
.../V5ScalableTopicTransactionTest.java | 346 +++++++++++++++++++
.../src/test/resources/pulsar-transaction.xml | 2 +
7 files changed, 1006 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionRecoveryTest.java
new file mode 100644
index 00000000000..64fd9e88544
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionRecoveryTest.java
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import com.google.common.collect.Sets;
+import java.time.Duration;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.v5.config.TransactionPolicy;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.SystemTopicNames;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Recovery coverage for V5 transactions on scalable topics. Uses a single restartable broker
+ * (the mock BookKeeper + metadata stores are reused across {@link #restartBroker()}, so segment
+ * data and the {@code /txn} metadata layout survive a restart). After restart a segment topic is
+ * reloaded cold, exercising {@code MetadataTransactionBuffer} / {@code MetadataPendingAckStore}
+ * recovery: committed data must stay visible, aborted data must stay filtered, and committed
+ * transactional acks must not be redelivered. Also covers the timeout sweep aborting a dangling
+ * transaction.
+ */
+public class V5TransactionRecoveryTest extends MockedPulsarServiceBaseTest {
+
+ private final String myNamespace = "pulsar/txn-recovery";
+
+ private PulsarClient v5Client;
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ ServiceConfiguration config = getDefaultConf();
+ config.setTransactionCoordinatorEnabled(true);
+ config.setTopicLevelPoliciesEnabled(false);
+ super.internalSetup(config);
+
+ admin.clusters().createCluster("test",
+ ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+ // SYSTEM_NAMESPACE's tenant IS "pulsar" (pulsar/system), which also owns myNamespace below.
+ admin.tenants().createTenant(NamespaceName.SYSTEM_NAMESPACE.getTenant(),
+ new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
+ admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString());
+ pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .createPartitionedTopic(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN,
+ new PartitionedTopicMetadata(1));
+
+ v5Client = newV5Client(Duration.ofMinutes(2));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (v5Client != null) {
+ v5Client.close();
+ v5Client = null;
+ }
+ super.internalCleanup();
+ }
+
+ private PulsarClient newV5Client(Duration txnTimeout) throws Exception {
+ return PulsarClient.builder()
+ .serviceUrl(pulsar.getBrokerServiceUrl())
+ .transactionPolicy(TransactionPolicy.builder().timeout(txnTimeout).build())
+ .build();
+ }
+
+ private String newScalableTopic(int numInitialSegments) throws Exception {
+ String name = "topic://" + myNamespace + "/scalable-" + UUID.randomUUID().toString().substring(0, 8);
+ admin.scalableTopics().createScalableTopic(name, numInitialSegments);
+ return name;
+ }
+
+ private QueueConsumer<String> subscribe(String topic, String sub) throws Exception {
+ return v5Client.newQueueConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName(sub)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ }
+
+ /** Restart the broker (segment data + metadata survive) and rebuild the V5 client. */
+ private void restartBrokerAndReconnect() throws Exception {
+ v5Client.close();
+ restartBroker();
+ v5Client = newV5Client(Duration.ofMinutes(2));
+ }
+
+ @Test
+ public void testCommittedMessagesVisibleAfterBrokerRestart() throws Exception {
+ // Commit a transaction, then restart so the segment topic is reloaded cold. The transaction
+ // buffer recovers from the /txn metadata and re-exposes the committed messages.
+ String topic = newScalableTopic(2);
+ @Cleanup
+ Producer<String> producer = v5Client.newProducer(Schema.string()).topic(topic).create();
+
+ Transaction txn = v5Client.newTransaction();
+ Set<String> sent = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ String v = "v-" + i;
+ producer.newMessage().key("k-" + i).transaction(txn).value(v).send();
+ sent.add(v);
+ }
+ txn.commit();
+ producer.close();
+
+ restartBrokerAndReconnect();
+
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(topic, "after-restart-sub");
+ Set<String> got = new HashSet<>();
+ for (int i = 0; i < sent.size(); i++) {
+ Message<String> msg = consumer.receive(Duration.ofSeconds(15));
+ assertNotNull(msg, "committed message #" + i + " must survive restart");
+ got.add(msg.value());
+ consumer.acknowledge(msg.id());
+ }
+ assertEquals(got, sent, "every committed message must be visible after recovery");
+ assertNull(consumer.receive(Duration.ofMillis(500)), "no extra messages");
+ }
+
+ @Test
+ public void testAbortedMessagesFilteredAfterBrokerRestart() throws Exception {
+ // Abort a transaction and publish a non-transactional sentinel, then restart. Recovery must
+ // rebuild the aborted set (from the durable ABORTED header / aborted records) so the aborted
+ // messages stay filtered and only the sentinel is delivered.
+ String topic = newScalableTopic(1);
+ @Cleanup
+ Producer<String> producer = v5Client.newProducer(Schema.string()).topic(topic).create();
+
+ Transaction txn = v5Client.newTransaction();
+ for (int i = 0; i < 5; i++) {
+ producer.newMessage().transaction(txn).value("aborted-" + i).send();
+ }
+ txn.abort();
+ producer.newMessage().value("sentinel").send();
+ producer.close();
+
+ restartBrokerAndReconnect();
+
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(topic, "after-restart-abort-sub");
+ Message<String> msg = consumer.receive(Duration.ofSeconds(15));
+ assertNotNull(msg, "the non-transactional sentinel must be delivered after restart");
+ assertEquals(msg.value(), "sentinel", "aborted messages must stay filtered after recovery");
+ consumer.acknowledge(msg.id());
+ assertNull(consumer.receive(Duration.ofMillis(500)), "no aborted messages after the sentinel");
+ }
+
+ @Test
+ public void testTransactionalAcksSurviveBrokerRestart() throws Exception {
+ // Acknowledge messages inside a committed transaction, then restart. The pending-ack store
+ // recovery (and the durable cursor) must keep them acknowledged — no redelivery.
+ String topic = newScalableTopic(1);
+ @Cleanup
+ Producer<String> producer = v5Client.newProducer(Schema.string()).topic(topic).create();
+ int n = 10;
+ for (int i = 0; i < n; i++) {
+ producer.newMessage().value("m-" + i).send();
+ }
+
+ // Acknowledge all inside the transaction, keeping the consumer open through commit so the
+ // async ack operations complete before commit (closing it early fails them).
+ Transaction txn = v5Client.newTransaction();
+ QueueConsumer<String> consumer = subscribe(topic, "ack-sub");
+ for (int i = 0; i < n; i++) {
+ Message<String> m = consumer.receive(Duration.ofSeconds(5));
+ assertNotNull(m, "delivery #" + i);
+ consumer.acknowledge(m.id(), txn);
+ }
+ txn.commit();
+ consumer.close();
+
+ // Pre-restart: confirm the committed acks have materialised (a fresh consumer on the same
+ // subscription sees nothing), so the restart genuinely exercises recovery of acked state.
+ {
+ @Cleanup
+ QueueConsumer<String> check = subscribe(topic, "ack-sub");
+ assertNull(check.receive(Duration.ofSeconds(10)), "committed acks must materialise before restart");
+ }
+
+ restartBrokerAndReconnect();
+
+ @Cleanup
+ QueueConsumer<String> after = subscribe(topic, "ack-sub");
+ assertNull(after.receive(Duration.ofSeconds(10)),
+ "committed transactional acks must not be redelivered after restart");
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionScalableTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionScalableTest.java
new file mode 100644
index 00000000000..bf8ef8e84eb
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionScalableTest.java
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api.v5;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import lombok.Cleanup;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.v5.config.TransactionPolicy;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.policies.data.ScalableTopicMetadata;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * Extended coverage for V5 transactions on scalable topics: multi-segment producers, many
+ * concurrent transactions with mixed commit/abort, abort across layout changes (split / merge),
+ * large transactions, transactional acknowledgement lifecycle, per-subscription isolation, and
+ * read-committed visibility while a transaction is still open.
+ *
+ * <p>Complements {@link V5TransactionTest} (single-segment happy paths + commit-across-split/merge).
+ */
+public class V5TransactionScalableTest extends V5ClientBaseTest {
+
+ private PulsarClient newTxnClient() throws Exception {
+ return track(PulsarClient.builder()
+ .serviceUrl(getBrokerServiceUrl())
+ .transactionPolicy(TransactionPolicy.builder().timeout(Duration.ofMinutes(2)).build())
+ .build());
+ }
+
+ private QueueConsumer<String> subscribe(PulsarClient client, String topic, String sub) throws Exception {
+ return client.newQueueConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName(sub)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ }
+
+ /** Receive exactly {@code count} messages (acking each), failing if any is missing. */
+ private Set<String> receiveValues(QueueConsumer<String> consumer, int count, Duration perMessage)
+ throws Exception {
+ Set<String> values = new HashSet<>();
+ for (int i = 0; i < count; i++) {
+ Message<String> msg = consumer.receive(perMessage);
+ assertNotNull(msg, "missing message #" + i + " (received so far: " + values.size() + ")");
+ values.add(msg.value());
+ consumer.acknowledge(msg.id());
+ }
+ return values;
+ }
+
+ private long firstActiveSegment(String topic) throws Exception {
+ ScalableTopicMetadata meta = admin.scalableTopics().getMetadata(topic);
+ for (ScalableTopicMetadata.SegmentInfo seg : meta.getSegments().values()) {
+ if (seg.isActive()) {
+ return seg.getSegmentId();
+ }
+ }
+ throw new IllegalStateException("no active segment for " + topic);
+ }
+
+ private List<Long> activeSegments(String topic) throws Exception {
+ ScalableTopicMetadata meta = admin.scalableTopics().getMetadata(topic);
+ List<Long> ids = new ArrayList<>();
+ for (ScalableTopicMetadata.SegmentInfo seg : meta.getSegments().values()) {
+ if (seg.isActive()) {
+ ids.add(seg.getSegmentId());
+ }
+ }
+ return ids;
+ }
+
+ private void awaitActiveSegmentCount(String topic, int expected) {
+ Awaitility.await().untilAsserted(() -> assertEquals(activeSegments(topic).size(), expected));
+ }
+
+ @Test
+ public void testMultiSegmentTransactionCommit() throws Exception {
+ // A single transaction whose keyed writes spread across all segments of a multi-segment
+ // topic commits atomically: every message becomes visible together.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(3);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "multi-seg-sub");
+
+ Transaction txn = client.newTransaction();
+ Set<String> sent = new HashSet<>();
+ for (int i = 0; i < 60; i++) {
+ String v = "v-" + i;
+ producer.newMessage().key("k-" + i).transaction(txn).value(v).send();
+ sent.add(v);
+ }
+ assertNull(consumer.receive(Duration.ofMillis(500)), "nothing visible before commit");
+
+ txn.commit();
+ assertEquals(receiveValues(consumer, sent.size(), Duration.ofSeconds(10)), sent,
+ "every committed message across all segments must be delivered");
+ assertNull(consumer.receive(Duration.ofMillis(500)), "no extra messages");
+ }
+
+ @Test
+ public void testConcurrentTransactionsMixedCommitAbort() throws Exception {
+ // Many open transactions on one topic, interleaved; half commit, half abort. Only the
+ // committed transactions' messages are ever delivered. While any transaction stays open the
+ // buffer pins visibility, so nothing is visible until all resolve.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(2);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "concurrent-sub");
+
+ int numTxns = 6;
+ int perTxn = 10;
+ List<Transaction> txns = new ArrayList<>();
+ for (int t = 0; t < numTxns; t++) {
+ txns.add(client.newTransaction());
+ }
+ Set<String> committedExpected = new HashSet<>();
+ for (int i = 0; i < perTxn; i++) {
+ for (int t = 0; t < numTxns; t++) {
+ String v = "t" + t + "-m" + i;
+ producer.newMessage().key("t" + t).transaction(txns.get(t)).value(v).send();
+ if (t % 2 == 0) {
+ committedExpected.add(v);
+ }
+ }
+ }
+ assertNull(consumer.receive(Duration.ofMillis(500)), "nothing visible while transactions are open");
+
+ for (int t = 0; t < numTxns; t++) {
+ if (t % 2 == 0) {
+ txns.get(t).commit();
+ } else {
+ txns.get(t).abort();
+ }
+ }
+
+ assertEquals(receiveValues(consumer, committedExpected.size(), Duration.ofSeconds(10)),
+ committedExpected, "exactly the committed transactions' messages must be delivered");
+ assertNull(consumer.receive(Duration.ofSeconds(1)), "aborted transactions' messages must never appear");
+ }
+
+ @Test
+ public void testAbortSpansSplit() throws Exception {
+ // A transaction spanning a split that aborts must leave nothing visible from either the
+ // sealed parent or the new children. A non-transactional sentinel proves the consumer is
+ // live and that only it is delivered.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(1);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "abort-split-sub");
+
+ Transaction txn = client.newTransaction();
+ for (int i = 0; i < 5; i++) {
+ producer.newMessage().key("k-" + i).transaction(txn).value("before-split-" + i).send();
+ }
+ admin.scalableTopics().splitSegment(topic, firstActiveSegment(topic));
+ awaitActiveSegmentCount(topic, 2);
+ for (int i = 0; i < 5; i++) {
+ producer.newMessage().key("k-after-" + i).transaction(txn).value("after-split-" + i).send();
+ }
+ txn.abort();
+
+ producer.newMessage().value("sentinel").send();
+ Message<String> msg = consumer.receive(Duration.ofSeconds(5));
+ assertNotNull(msg, "non-transactional sentinel must be delivered");
+ assertEquals(msg.value(), "sentinel", "aborted txn messages must not precede the sentinel");
+ consumer.acknowledge(msg.id());
+ assertNull(consumer.receive(Duration.ofMillis(500)), "no aborted messages after the sentinel");
+ }
+
+ @Test
+ public void testAbortSpansMerge() throws Exception {
+ // Same as the split case but across a merge: an aborted transaction whose lifetime spans a
+ // merge leaves nothing visible.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(2);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "abort-merge-sub");
+
+ Transaction txn = client.newTransaction();
+ for (int i = 0; i < 10; i++) {
+ producer.newMessage().key("k-" + i).transaction(txn).value("before-merge-" + i).send();
+ }
+ List<Long> active = activeSegments(topic);
+ assertEquals(active.size(), 2, "expected 2 active segments before merge");
+ admin.scalableTopics().mergeSegments(topic, active.get(0), active.get(1));
+ awaitActiveSegmentCount(topic, 1);
+ for (int i = 0; i < 5; i++) {
+ producer.newMessage().key("k-after-" + i).transaction(txn).value("after-merge-" + i).send();
+ }
+ txn.abort();
+
+ producer.newMessage().value("sentinel").send();
+ Message<String> msg = consumer.receive(Duration.ofSeconds(5));
+ assertNotNull(msg, "non-transactional sentinel must be delivered");
+ assertEquals(msg.value(), "sentinel");
+ consumer.acknowledge(msg.id());
+ assertNull(consumer.receive(Duration.ofMillis(500)), "no aborted messages after the sentinel");
+ }
+
+ @Test
+ public void testLargeTransaction() throws Exception {
+ // A large transaction commits atomically and every message is delivered exactly once.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(2);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "large-txn-sub");
+
+ int n = 500;
+ Transaction txn = client.newTransaction();
+ Set<String> sent = new HashSet<>();
+ for (int i = 0; i < n; i++) {
+ String v = "v-" + i;
+ producer.newMessage().key("k-" + (i % 50)).transaction(txn).value(v).send();
+ sent.add(v);
+ }
+ txn.commit();
+
+ assertEquals(receiveValues(consumer, n, Duration.ofSeconds(15)), sent,
+ "all messages in the large transaction must be delivered");
+ assertNull(consumer.receive(Duration.ofSeconds(1)), "no extra messages");
+ }
+
+ @Test
+ public void testTransactionalAckAbortRedeliversThenCommitSticks() throws Exception {
+ // Acknowledge messages inside a transaction that aborts -> the acks roll back and the
+ // messages are redelivered. Acknowledging the redelivered batch in a committed transaction
+ // makes the acks durable -> no further redelivery.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(1);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ int n = 10;
+ Set<String> produced = new HashSet<>();
+ for (int i = 0; i < n; i++) {
+ String v = "m-" + i;
+ producer.newMessage().value(v).send();
+ produced.add(v);
+ }
+
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "ack-lifecycle-sub");
+
+ // Consume all, ack inside a transaction we then abort.
+ List<Message<String>> first = new ArrayList<>();
+ for (int i = 0; i < n; i++) {
+ Message<String> m = consumer.receive(Duration.ofSeconds(5));
+ assertNotNull(m, "initial delivery #" + i);
+ first.add(m);
+ }
+ Transaction abortTxn = client.newTransaction();
+ for (Message<String> m : first) {
+ consumer.acknowledge(m.id(), abortTxn);
+ }
+ abortTxn.abort();
+
+ // After abort the messages must be redelivered. Collect them in a committed transaction.
+ Transaction commitTxn = client.newTransaction();
+ Set<String> redelivered = new HashSet<>();
+ long deadline = System.currentTimeMillis() + 30_000;
+ while (redelivered.size() < n && System.currentTimeMillis() < deadline) {
+ Message<String> m = consumer.receive(Duration.ofSeconds(2));
+ if (m == null) {
+ continue;
+ }
+ redelivered.add(m.value());
+ consumer.acknowledge(m.id(), commitTxn);
+ }
+ assertEquals(redelivered, produced, "aborting a transactional ack must redeliver every message");
+ commitTxn.commit();
+
+ assertNull(consumer.receive(Duration.ofSeconds(2)),
+ "messages acked in a committed transaction must not be redelivered");
+ }
+
+ @Test
+ public void testMultipleSubscriptionsIndependentTransactionalAcks() throws Exception {
+ // A committed transactional ack on one subscription must not affect another subscription on
+ // the same scalable topic.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(1);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ int n = 10;
+ Set<String> produced = new HashSet<>();
+ for (int i = 0; i < n; i++) {
+ String v = "m-" + i;
+ producer.newMessage().value(v).send();
+ produced.add(v);
+ }
+
+ @Cleanup
+ QueueConsumer<String> subA = subscribe(client, topic, "sub-a");
+ @Cleanup
+ QueueConsumer<String> subB = subscribe(client, topic, "sub-b");
+
+ Transaction txn = client.newTransaction();
+ for (int i = 0; i < n; i++) {
+ Message<String> m = subA.receive(Duration.ofSeconds(5));
+ assertNotNull(m, "sub-a delivery #" + i);
+ subA.acknowledge(m.id(), txn);
+ }
+ txn.commit();
+ assertNull(subA.receive(Duration.ofSeconds(2)), "sub-a committed acks must stick");
+
+ // sub-b is independent and must still see everything.
+ assertEquals(receiveValues(subB, n, Duration.ofSeconds(5)), produced,
+ "the other subscription must be unaffected by sub-a's transactional acks");
+ }
+
+ @Test
+ public void testOpenTransactionPinsLaterNonTransactionalWrites() throws Exception {
+ // Read-committed visibility: while a transaction is open, the buffer's max-read position is
+ // pinned below its first write, so even a non-transactional message published afterwards
+ // stays invisible until the transaction resolves. After abort, the non-transactional message
+ // becomes visible and the aborted ones are filtered.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(1);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "pin-sub");
+
+ Transaction txn = client.newTransaction();
+ for (int i = 0; i < 5; i++) {
+ producer.newMessage().transaction(txn).value("txn-" + i).send();
+ }
+ assertNull(consumer.receive(Duration.ofSeconds(1)), "uncommitted writes are invisible");
+
+ producer.newMessage().value("sentinel").send();
+ assertNull(consumer.receive(Duration.ofSeconds(1)),
+ "a non-transactional write after an open txn is pinned until the txn resolves");
+
+ txn.abort();
+ Message<String> msg = consumer.receive(Duration.ofSeconds(5));
+ assertNotNull(msg, "after abort the non-transactional write becomes visible");
+ assertEquals(msg.value(), "sentinel", "only the non-transactional write is delivered");
+ consumer.acknowledge(msg.id());
+ assertNull(consumer.receive(Duration.ofMillis(500)), "aborted txn messages must stay filtered");
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionTest.java
index 6cdc197709f..59057de8bf3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionTest.java
@@ -191,13 +191,7 @@ public class V5TransactionTest extends V5ClientBaseTest {
assertEquals(mb.value(), "b-1");
}
- // Disabled: documents a real broker gap in the per-segment TransactionBuffer model.
- // After split, the parent segment is sealed; the txn coordinator's COMMIT END-TXN to
- // the sealed parent never returns and the commit RPC times out (~30s). The fix
- // requires moving transaction tracking up to the scalable-topic level (so layout
- // changes don't strand in-flight transactions on now-sealed segments) — that's
- // beyond the scope of this commit.
- @Test(enabled = false)
+ @Test
public void testCommitSpansSplit() throws Exception {
// A single transaction whose lifetime spans a layout-changing split must commit
// atomically: pre-split writes (on the now-sealed parent) and post-split writes
@@ -270,10 +264,7 @@ public class V5TransactionTest extends V5ClientBaseTest {
assertEquals(received, sent, "all txn messages across the split must be delivered after commit");
}
- // Disabled: same broker gap as testCommitSpansSplit. After merge, both source
- // segments are sealed and the COMMIT marker can't be delivered, so the END-TXN
- // request times out.
- @Test(enabled = false)
+ @Test
public void testCommitSpansMerge() throws Exception {
// A single transaction whose lifetime spans a layout-changing merge must commit
// atomically: writes to the two pre-merge segments and writes to the post-merge
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/TcMetadataDiscoveryTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/TcMetadataDiscoveryTestBase.java
index 8af84ab229e..05ed5747f64 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/TcMetadataDiscoveryTestBase.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/TcMetadataDiscoveryTestBase.java
@@ -64,8 +64,16 @@ public abstract class TcMetadataDiscoveryTestBase extends PulsarTestSuite {
BrokerContainer brokerContainer = pulsarCluster.getBrokers().iterator().next();
brokerContainer.execCmd(
"/pulsar/bin/pulsar", "initialize-transaction-coordinator-metadata",
- "-cs", ZKContainer.NAME,
+ "-cs", configurationStoreConnectionString(),
"-c", pulsarCluster.getClusterName(),
"--initial-num-transaction-coordinators", Integer.toString(TC_PARALLELISM));
}
+
+ /**
+ * Configuration-store connection string used to initialize the transaction-coordinator metadata.
+ * Defaults to the ZooKeeper container; an Oxia-backed subclass overrides this with the Oxia URL.
+ */
+ protected String configurationStoreConnectionString() {
+ return ZKContainer.NAME;
+ }
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/V5ScalableTopicTransactionOnOxiaTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/V5ScalableTopicTransactionOnOxiaTest.java
new file mode 100644
index 00000000000..e28030b4a1d
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/V5ScalableTopicTransactionOnOxiaTest.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.transaction;
+
+import lombok.CustomLog;
+import org.apache.pulsar.tests.integration.oxia.OxiaContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec;
+
+/**
+ * Runs the full {@link V5ScalableTopicTransactionTest} suite against a cluster whose metadata store
+ * is a containerized Oxia instead of ZooKeeper. Oxia is the backend PIP-473 targets natively — its
+ * {@code scanByIndex} / {@code subscribeSequence} / partition-key / sequence-delta primitives back the
+ * {@code /txn} layout directly, rather than the scan-and-filter fallback ZooKeeper uses — so this is
+ * the closest the test suite gets to the production transaction path.
+ */
+@CustomLog
+public class V5ScalableTopicTransactionOnOxiaTest extends V5ScalableTopicTransactionTest {
+
+ @Override
+ protected PulsarClusterSpec.PulsarClusterSpecBuilder beforeSetupCluster(
+ String clusterName, PulsarClusterSpec.PulsarClusterSpecBuilder specBuilder) {
+ specBuilder.enableOxia(true);
+ return specBuilder;
+ }
+
+ @Override
+ protected String configurationStoreConnectionString() {
+ return "oxia://" + OxiaContainer.NAME + ":" + OxiaContainer.OXIA_PORT;
+ }
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/V5ScalableTopicTransactionTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/V5ScalableTopicTransactionTest.java
new file mode 100644
index 00000000000..0b5b62acd6c
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/transaction/V5ScalableTopicTransactionTest.java
@@ -0,0 +1,346 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.transaction;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import lombok.Cleanup;
+import lombok.CustomLog;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.v5.Message;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.PulsarClient;
+import org.apache.pulsar.client.api.v5.QueueConsumer;
+import org.apache.pulsar.client.api.v5.Transaction;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.v5.config.TransactionPolicy;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.policies.data.ScalableTopicMetadata;
+import org.apache.pulsar.tests.integration.containers.BrokerContainer;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * End-to-end integration coverage for V5 transactions producing and consuming data on scalable
+ * ({@code topic://}) topics across a multi-broker docker cluster: commit visibility, abort filtering,
+ * multi-topic atomic commit, consume-transform-produce, survival of a broker failover mid-transaction,
+ * and transactions whose lifetime spans a layout-changing split / merge.
+ */
+@CustomLog
+public class V5ScalableTopicTransactionTest extends TcMetadataDiscoveryTestBase {
+
+ private static final int OP_TIMEOUT_SECONDS = 30;
+
+ private PulsarAdmin newAdmin() throws Exception {
+ return PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build();
+ }
+
+ private PulsarClient newTxnClient() throws Exception {
+ return PulsarClient.builder()
+ .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
+ .transactionPolicy(TransactionPolicy.builder().timeout(Duration.ofMinutes(5)).build())
+ .build();
+ }
+
+ private String scalableTopicName() {
+ return "topic://public/default/scalable-" + randomName(8);
+ }
+
+ private QueueConsumer<String> subscribe(PulsarClient client, String topic, String sub) throws Exception {
+ return client.newQueueConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName(sub)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ }
+
+ private Set<String> drain(QueueConsumer<String> consumer, int count) throws Exception {
+ Set<String> values = new HashSet<>();
+ for (int i = 0; i < count; i++) {
+ Message<String> msg = consumer.receive(Duration.ofSeconds(OP_TIMEOUT_SECONDS));
+ assertNotNull(msg, "missing message #" + i + " (received so far: " + values.size() + ")");
+ values.add(msg.value());
+ consumer.acknowledge(msg.id());
+ }
+ return values;
+ }
+
+ private List<Long> activeSegments(PulsarAdmin admin, String topic) throws Exception {
+ ScalableTopicMetadata meta = admin.scalableTopics().getMetadata(topic);
+ List<Long> ids = new ArrayList<>();
+ for (ScalableTopicMetadata.SegmentInfo seg : meta.getSegments().values()) {
+ if (seg.isActive()) {
+ ids.add(seg.getSegmentId());
+ }
+ }
+ return ids;
+ }
+
+ @Test(timeOut = 300_000)
+ public void testProduceConsumeCommitAndAbort() throws Exception {
+ @Cleanup
+ PulsarAdmin admin = newAdmin();
+ @Cleanup
+ PulsarClient client = newTxnClient();
+ String topic = scalableTopicName();
+ admin.scalableTopics().createScalableTopic(topic, 2);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "ce-sub");
+
+ // Commit path.
+ Transaction commit = client.newTransaction();
+ Set<String> committed = new HashSet<>();
+ for (int i = 0; i < 20; i++) {
+ String v = "commit-" + i;
+ producer.newMessage().key("k-" + i).transaction(commit).value(v).send();
+ committed.add(v);
+ }
+ assertNull(consumer.receive(Duration.ofSeconds(2)), "nothing visible before commit");
+ commit.commit();
+ assertEquals(drain(consumer, committed.size()), committed, "committed messages must all be delivered");
+
+ // Abort path: aborted messages never delivered; a sentinel proves the consumer is live.
+ Transaction abort = client.newTransaction();
+ for (int i = 0; i < 20; i++) {
+ producer.newMessage().key("k-" + i).transaction(abort).value("abort-" + i).send();
+ }
+ abort.abort();
+ producer.newMessage().value("sentinel").send();
+ Message<String> msg = consumer.receive(Duration.ofSeconds(OP_TIMEOUT_SECONDS));
+ assertNotNull(msg, "sentinel must be delivered");
+ assertEquals(msg.value(), "sentinel", "aborted messages must not precede the sentinel");
+ consumer.acknowledge(msg.id());
+ assertNull(consumer.receive(Duration.ofSeconds(2)), "no aborted messages after the sentinel");
+ }
+
+ @Test(timeOut = 300_000)
+ public void testMultiTopicAtomicCommit() throws Exception {
+ @Cleanup
+ PulsarAdmin admin = newAdmin();
+ @Cleanup
+ PulsarClient client = newTxnClient();
+ String topicA = scalableTopicName();
+ String topicB = scalableTopicName();
+ admin.scalableTopics().createScalableTopic(topicA, 2);
+ admin.scalableTopics().createScalableTopic(topicB, 2);
+
+ @Cleanup
+ Producer<String> prodA = client.newProducer(Schema.string()).topic(topicA).create();
+ @Cleanup
+ Producer<String> prodB = client.newProducer(Schema.string()).topic(topicB).create();
+ @Cleanup
+ QueueConsumer<String> consA = subscribe(client, topicA, "multi-a");
+ @Cleanup
+ QueueConsumer<String> consB = subscribe(client, topicB, "multi-b");
+
+ Transaction txn = client.newTransaction();
+ prodA.newMessage().value("a-1").send();
+ prodB.newMessage().value("b-1").send();
+ // The above are non-transactional sends to confirm the topics work; now the transactional ones:
+ prodA.newMessage().transaction(txn).value("a-txn").send();
+ prodB.newMessage().transaction(txn).value("b-txn").send();
+
+ // The non-transactional a-1/b-1 are pinned behind the open transaction's first write only if
+ // they were published after it; here they were published before, so they are visible.
+ assertEquals(consA.receive(Duration.ofSeconds(OP_TIMEOUT_SECONDS)).value(), "a-1");
+ assertEquals(consB.receive(Duration.ofSeconds(OP_TIMEOUT_SECONDS)).value(), "b-1");
+ assertNull(consA.receive(Duration.ofSeconds(2)), "txn message on A invisible pre-commit");
+ assertNull(consB.receive(Duration.ofSeconds(2)), "txn message on B invisible pre-commit");
+
+ txn.commit();
+ assertEquals(consA.receive(Duration.ofSeconds(OP_TIMEOUT_SECONDS)).value(), "a-txn");
+ assertEquals(consB.receive(Duration.ofSeconds(OP_TIMEOUT_SECONDS)).value(), "b-txn");
+ }
+
+ @Test(timeOut = 300_000)
+ public void testConsumeTransformProduce() throws Exception {
+ @Cleanup
+ PulsarAdmin admin = newAdmin();
+ @Cleanup
+ PulsarClient client = newTxnClient();
+ String input = scalableTopicName();
+ String output = scalableTopicName();
+ admin.scalableTopics().createScalableTopic(input, 1);
+ admin.scalableTopics().createScalableTopic(output, 1);
+
+ @Cleanup
+ Producer<String> seed = client.newProducer(Schema.string()).topic(input).create();
+ seed.newMessage().value("hello").send();
+
+ @Cleanup
+ QueueConsumer<String> in = subscribe(client, input, "in-sub");
+ @Cleanup
+ Producer<String> out = client.newProducer(Schema.string()).topic(output).create();
+ @Cleanup
+ QueueConsumer<String> verify = subscribe(client, output, "verify-sub");
+
+ Message<String> seedMsg = in.receive(Duration.ofSeconds(OP_TIMEOUT_SECONDS));
+ assertNotNull(seedMsg, "seed message must be delivered");
+
+ Transaction txn = client.newTransaction();
+ out.newMessage().transaction(txn).value(seedMsg.value().toUpperCase()).send();
+ in.acknowledge(seedMsg.id(), txn);
+ assertNull(verify.receive(Duration.ofSeconds(2)), "output invisible before commit");
+
+ txn.commit();
+ Message<String> outMsg = verify.receive(Duration.ofSeconds(OP_TIMEOUT_SECONDS));
+ assertNotNull(outMsg, "committed output must be delivered");
+ assertEquals(outMsg.value(), "HELLO");
+ }
+
+ @Test(timeOut = 360_000)
+ public void testTransactionSurvivesBrokerFailover() throws Exception {
+ @Cleanup
+ PulsarAdmin admin = newAdmin();
+ @Cleanup
+ PulsarClient client = newTxnClient();
+ String topic = scalableTopicName();
+ // Several segments so they spread across both brokers.
+ admin.scalableTopics().createScalableTopic(topic, 4);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "failover-sub");
+
+ Transaction txn = client.newTransaction();
+ Set<String> sent = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ String v = "before-kill-" + i;
+ producer.newMessage().key("k-" + i).transaction(txn).value(v).send();
+ sent.add(v);
+ }
+
+ // Kill one broker mid-transaction; coordinator partitions and segments it led are reassigned.
+ BrokerContainer victim = pulsarCluster.getBrokers().iterator().next();
+ log.info().attr("broker", victim.getContainerName()).log("Stopping broker mid-transaction");
+ victim.stop();
+
+ // Continue producing in the same transaction; the producer reconnects to the survivor. Retry
+ // each send within a bounded window while reassignment/reconnection settles.
+ for (int i = 0; i < 10; i++) {
+ String v = "after-kill-" + i;
+ sendWithRetry(producer, txn, "k-after-" + i, v);
+ sent.add(v);
+ }
+
+ txn.commit();
+ assertEquals(drain(consumer, sent.size()), sent,
+ "every message in a transaction that spanned a broker failover must be delivered");
+ }
+
+ @Test(timeOut = 360_000)
+ public void testCommitSpansSplit() throws Exception {
+ @Cleanup
+ PulsarAdmin admin = newAdmin();
+ @Cleanup
+ PulsarClient client = newTxnClient();
+ String topic = scalableTopicName();
+ admin.scalableTopics().createScalableTopic(topic, 1);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "split-sub");
+
+ Transaction txn = client.newTransaction();
+ Set<String> sent = new HashSet<>();
+ for (int i = 0; i < 5; i++) {
+ String v = "before-split-" + i;
+ producer.newMessage().key("k-" + i).transaction(txn).value(v).send();
+ sent.add(v);
+ }
+ List<Long> active = activeSegments(admin, topic);
+ assertEquals(active.size(), 1, "expected one active segment before split");
+ admin.scalableTopics().splitSegment(topic, active.get(0));
+ Awaitility.await().untilAsserted(() -> assertEquals(activeSegments(admin, topic).size(), 2));
+
+ for (int i = 0; i < 5; i++) {
+ String v = "after-split-" + i;
+ sendWithRetry(producer, txn, "k-after-" + i, v);
+ sent.add(v);
+ }
+ assertNull(consumer.receive(Duration.ofSeconds(2)), "nothing visible before commit");
+ txn.commit();
+ assertEquals(drain(consumer, sent.size()), sent, "all messages across the split must be delivered");
+ }
+
+ @Test(timeOut = 360_000)
+ public void testCommitSpansMerge() throws Exception {
+ @Cleanup
+ PulsarAdmin admin = newAdmin();
+ @Cleanup
+ PulsarClient client = newTxnClient();
+ String topic = scalableTopicName();
+ admin.scalableTopics().createScalableTopic(topic, 2);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create();
+ @Cleanup
+ QueueConsumer<String> consumer = subscribe(client, topic, "merge-sub");
+
+ Transaction txn = client.newTransaction();
+ Set<String> sent = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ String v = "before-merge-" + i;
+ producer.newMessage().key("k-" + i).transaction(txn).value(v).send();
+ sent.add(v);
+ }
+ List<Long> active = activeSegments(admin, topic);
+ assertEquals(active.size(), 2, "expected two active segments before merge");
+ admin.scalableTopics().mergeSegments(topic, active.get(0), active.get(1));
+ Awaitility.await().untilAsserted(() -> assertEquals(activeSegments(admin, topic).size(), 1));
+
+ for (int i = 0; i < 5; i++) {
+ String v = "after-merge-" + i;
+ sendWithRetry(producer, txn, "k-after-" + i, v);
+ sent.add(v);
+ }
+ assertNull(consumer.receive(Duration.ofSeconds(2)), "nothing visible before commit");
+ txn.commit();
+ assertEquals(drain(consumer, sent.size()), sent, "all messages across the merge must be delivered");
+ }
+
+ /** Send a transactional message, retrying within a bounded window while the cluster re-settles. */
+ private void sendWithRetry(Producer<String> producer, Transaction txn, String key, String value)
+ throws Exception {
+ long deadline = System.currentTimeMillis() + 60_000;
+ Exception last = null;
+ while (System.currentTimeMillis() < deadline) {
+ try {
+ producer.newMessage().key(key).transaction(txn).value(value).send();
+ return;
+ } catch (Exception e) {
+ last = e;
+ Thread.sleep(1000);
+ }
+ }
+ assertTrue(false, "send did not succeed within the retry window: " + (last == null ? "?" : last));
+ }
+}
diff --git a/tests/integration/src/test/resources/pulsar-transaction.xml b/tests/integration/src/test/resources/pulsar-transaction.xml
index 0c23b9e93ab..b69d5f31288 100644
--- a/tests/integration/src/test/resources/pulsar-transaction.xml
+++ b/tests/integration/src/test/resources/pulsar-transaction.xml
@@ -24,6 +24,8 @@
<classes>
<class name="org.apache.pulsar.tests.integration.transaction.TransactionTest" />
<class name="org.apache.pulsar.tests.integration.transaction.TcMetadataDiscoveryTest" />
+ <class name="org.apache.pulsar.tests.integration.transaction.V5ScalableTopicTransactionTest" />
+ <class name="org.apache.pulsar.tests.integration.transaction.V5ScalableTopicTransactionOnOxiaTest" />
</classes>
</test>
</suite>