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 2025/10/31 11:27:25 UTC
(pulsar) branch branch-3.0 updated (58f30d9f4b0 -> 60ea33a8ffe)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 58f30d9f4b0 [fix][ci][branch-3.0] Drop UBUNTU_MIRROR/UBUNTU_SECURITY_MIRROR support in Dockerfile since it's broken
new f02d2c93160 [fix][sec] Upgrade Spring to 6.2.12 to remediate CVE-2025-22233 and CVE-2025-41249 (#24903)
new ff2c9ece11b [improve][ci] Move replication tests to new group Broker Group 5 in Pulsar CI (#24917)
new 60ea33a8ffe [fix][broker] BacklogMessageAge is not reset when cursor mdPosition is on an open ledger (#24915)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/pulsar-ci.yaml | 2 +
build/run_unit_group.sh | 4 ++
pom.xml | 2 +-
.../broker/service/persistent/PersistentTopic.java | 2 +
...ameAuthParamsLookupAutoClusterFailoverTest.java | 1 +
.../broker/service/AbstractReplicatorTest.java | 2 +-
.../broker/service/BacklogQuotaManagerTest.java | 57 ++++++++++++++++------
...eateTopicToRemoteClusterForReplicationTest.java | 2 +-
.../service/OneWayReplicatorDeduplicationTest.java | 2 +-
.../broker/service/OneWayReplicatorTest.java | 2 +-
...OneWayReplicatorUsingGlobalPartitionedTest.java | 2 +-
.../service/OneWayReplicatorUsingGlobalZKTest.java | 2 +-
.../pulsar/broker/service/PeerReplicatorTest.java | 4 +-
.../broker/service/ReplicationTopicGcTest.java | 2 +-
.../ReplicationTopicGcUsingGlobalZKTest.java | 2 +-
.../pulsar/broker/service/ReplicationTxnTest.java | 2 +-
.../broker/service/ReplicatorAdminTlsTest.java | 2 +-
.../ReplicatorAdminTlsWithKeyStoreTest.java | 2 +-
.../broker/service/ReplicatorGlobalNSTest.java | 2 +-
.../broker/service/ReplicatorRateLimiterTest.java | 2 +-
.../service/ReplicatorRemoveClusterTest.java | 2 +-
.../broker/service/ReplicatorSubscriptionTest.java | 2 +-
.../ReplicatorSubscriptionWithTransactionTest.java | 2 +-
.../pulsar/broker/service/ReplicatorTest.java | 2 +-
.../pulsar/broker/service/ReplicatorTlsTest.java | 2 +-
.../service/ReplicatorTopicPoliciesTest.java | 2 +-
.../TransactionalReplicateSubscriptionTest.java | 1 +
.../PersistentReplicatorInflightTaskTest.java | 2 +-
28 files changed, 76 insertions(+), 37 deletions(-)
(pulsar) 03/03: [fix][broker] BacklogMessageAge is not reset when cursor mdPosition is on an open ledger (#24915)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 60ea33a8ffe53f434ce95312367eea1134fb9ed7
Author: Baodi Shi <ba...@apache.org>
AuthorDate: Fri Oct 31 18:29:02 2025 +0800
[fix][broker] BacklogMessageAge is not reset when cursor mdPosition is on an open ledger (#24915)
(cherry picked from commit 54da0c8cef54480b09272db34e770894ff0ef474)
---
.../broker/service/persistent/PersistentTopic.java | 2 +
.../broker/service/BacklogQuotaManagerTest.java | 57 ++++++++++++++++------
2 files changed, 45 insertions(+), 14 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 997e2a06b3f..5b64e7deedc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -3440,6 +3440,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
oldestMarkDeleteCursorInfo.getCursor().getName(),
checkResult.getEstimatedOldestUnacknowledgedMessageTimestamp(),
oldestMarkDeleteCursorInfo.getVersion()));
+ } else {
+ TIME_BASED_BACKLOG_QUOTA_CHECK_RESULT_UPDATER.set(this, null);
}
return CompletableFuture.completedFuture(null);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
index ce9e02dd3fe..f42f9f7a9b4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
@@ -42,7 +42,9 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.Cleanup;
+import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorContainer;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.PulsarService;
@@ -140,6 +142,7 @@ public class BacklogQuotaManagerTest {
config.setBacklogQuotaCheckIntervalInSeconds(TIME_TO_CHECK_BACKLOG_QUOTA);
config.setManagedLedgerMaxEntriesPerLedger(MAX_ENTRIES_PER_LEDGER);
config.setManagedLedgerMinLedgerRolloverTimeMinutes(0);
+ config.setManagedLedgerDefaultMarkDeleteRateLimit(1000);
config.setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED);
config.setSystemTopicEnabled(true);
config.setTopicLevelPoliciesEnabled(true);
@@ -677,7 +680,7 @@ public class BacklogQuotaManagerTest {
final String topic1 = "persistent://prop/ns-quota/topic2" + UUID.randomUUID();
final String subName1 = "c1";
- final int numMsgs = 5;
+ final int numMsgs = 7;
Consumer<byte[]> consumer1 = client.newConsumer().topic(topic1).subscriptionName(subName1)
.acknowledgmentGroupTime(0, SECONDS)
@@ -685,27 +688,35 @@ public class BacklogQuotaManagerTest {
Producer<byte[]> producer = createProducer(client, topic1);
byte[] content = new byte[1024];
+ // 1. Send messages
+ // The manager ledger max entries is 5, so we can send 7 messages to make sure we have multiple ledgers
+ // When send msg 4, the ledger closed.
+ // Second: 1 2 3 4 5 6 7
+ // msg idx: [0 1 2 3 4] [5 6]
for (int i = 0; i < numMsgs; i++) {
- Thread.sleep(3000); // Guarantees if we use wrong message in age, to show up in failed test
- producer.send(content);
+ Thread.sleep(1000);
+ MessageId send = producer.send(content);
}
+ long lastLedgerCloseTime = System.currentTimeMillis() - 2000;
+ // 2. Receive msg-0 and ack it.
+ String c1MarkDeletePositionBefore =
+ admin.topics().getInternalStats(topic1).cursors.get(subName1).markDeletePosition;
Message<byte[]> oldestMessage = consumer1.receive();
consumer1.acknowledge(oldestMessage);
- log.info("Moved subscription 1, by 1 message");
-
- // Unload topic to trigger the ledger close
- unloadAndLoadTopic(topic1, producer);
- long unloadTime = System.currentTimeMillis();
- waitForQuotaCheckToRunTwice();
-
- Metrics metrics = prometheusMetricsClient.getMetrics();
+ c1MarkDeletePositionBefore = waitForMarkDeletePositionToChange(topic1, subName1,
+ c1MarkDeletePositionBefore);
+ log.info("Moved subscription 1, by 1 message {}", oldestMessage.getMessageId());
+
+ // 3. Expected the oldestBacklogMessageAgeSeconds is based on last ledger close time
+ long expectedMessageAgeSeconds =
+ MILLISECONDS.toSeconds(System.currentTimeMillis() - lastLedgerCloseTime);
+ PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic1).get();
+ topicRef.updateOldPositionInfo();
TopicStats topicStats = getTopicStats(topic1);
-
- long expectedMessageAgeSeconds = MILLISECONDS.toSeconds(System.currentTimeMillis() - unloadTime);
assertThat(topicStats.getOldestBacklogMessageAgeSeconds())
.isCloseTo(expectedMessageAgeSeconds, within(1L));
-
+ Metrics metrics = prometheusMetricsClient.getMetrics();
Metric backlogAgeMetric =
metrics.findSingleMetricByNameAndLabels("pulsar_storage_backlog_age_seconds",
Pair.of("topic", topic1));
@@ -714,6 +725,24 @@ public class BacklogQuotaManagerTest {
entry("namespace", namespace),
entry("topic", topic1));
assertThat((long) backlogAgeMetric.value).isCloseTo(expectedMessageAgeSeconds, within(2L));
+
+ // 4. Move consumer to `end - 1`, then OldestBacklogMessageAgeSeconds should be `-1`, because the
+ // second ledger is not closed yet.
+ for (int i = 1; i < numMsgs - 1; i++) {
+ Message<byte[]> msg = consumer1.receive();
+ consumer1.acknowledge(msg);
+ }
+ waitForMarkDeletePositionToChange(topic1, subName1, c1MarkDeletePositionBefore);
+ ManagedCursorContainer cursors = (ManagedCursorContainer) topicRef.getManagedLedger().getCursors();
+ ManagedCursor subCursor = cursors.get(subName1);
+ Awaitility.await().pollInterval(100, MILLISECONDS).atMost(5, SECONDS).until(
+ () -> subCursor.getMarkDeletedPosition().equals(subCursor.getPersistentMarkDeletedPosition()));
+ topicRef.updateOldPositionInfo();
+ topicStats = getTopicStats(topic1, true);
+ assertThat(topicStats.getSubscriptions().get(subName1).getMsgBacklog())
+ .isEqualTo(1L);
+ assertThat(topicStats.getOldestBacklogMessageAgeSeconds())
+ .isEqualTo(-1L);
}
}
(pulsar) 01/03: [fix][sec] Upgrade Spring to 6.2.12 to remediate CVE-2025-22233 and CVE-2025-41249 (#24903)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f02d2c931602085d71aa8224e116edfb43f63f12
Author: guptas6est <sa...@est.tech>
AuthorDate: Wed Oct 29 07:53:25 2025 +0000
[fix][sec] Upgrade Spring to 6.2.12 to remediate CVE-2025-22233 and CVE-2025-41249 (#24903)
(cherry picked from commit 38e0fa1da6a3fa9e8f7ba9c079f3b61db8b1d075)
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index f23c02cf547..c1bb2d62c9e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -246,7 +246,7 @@ flexible messaging model and an intuitive client API.</description>
<kotlin-stdlib.version>1.6.0</kotlin-stdlib.version>
<nsq-client.version>1.0</nsq-client.version>
<cron-utils.version>9.1.6</cron-utils.version>
- <spring.version>6.1.13</spring.version>
+ <spring.version>6.2.12</spring.version>
<apache-http-client.version>4.5.13</apache-http-client.version>
<apache-httpcomponents.version>4.4.15</apache-httpcomponents.version>
<jetcd.version>0.7.7</jetcd.version>
(pulsar) 02/03: [improve][ci] Move replication tests to new group Broker Group 5 in Pulsar CI (#24917)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ff2c9ece11b0d5b2af6c34cea9eba6a5d592f045
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Oct 29 19:46:03 2025 +0200
[improve][ci] Move replication tests to new group Broker Group 5 in Pulsar CI (#24917)
(cherry picked from commit f55d45a51b9862240f2139104c16533a29636240)
---
.github/workflows/pulsar-ci.yaml | 2 ++
build/run_unit_group.sh | 4 ++++
.../pulsar/broker/SameAuthParamsLookupAutoClusterFailoverTest.java | 1 +
.../java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java | 2 +-
.../service/DisabledCreateTopicToRemoteClusterForReplicationTest.java | 2 +-
.../pulsar/broker/service/OneWayReplicatorDeduplicationTest.java | 2 +-
.../java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java | 2 +-
.../broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java | 2 +-
.../pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java | 2 +-
.../java/org/apache/pulsar/broker/service/PeerReplicatorTest.java | 4 ++--
.../java/org/apache/pulsar/broker/service/ReplicationTopicGcTest.java | 2 +-
.../pulsar/broker/service/ReplicationTopicGcUsingGlobalZKTest.java | 2 +-
.../java/org/apache/pulsar/broker/service/ReplicationTxnTest.java | 2 +-
.../java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java | 2 +-
.../pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java | 2 +-
.../java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java | 2 +-
.../org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java | 2 +-
.../org/apache/pulsar/broker/service/ReplicatorRemoveClusterTest.java | 2 +-
.../org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java | 2 +-
.../broker/service/ReplicatorSubscriptionWithTransactionTest.java | 2 +-
.../test/java/org/apache/pulsar/broker/service/ReplicatorTest.java | 2 +-
.../test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java | 2 +-
.../org/apache/pulsar/broker/service/ReplicatorTopicPoliciesTest.java | 2 +-
.../pulsar/broker/service/TransactionalReplicateSubscriptionTest.java | 1 +
.../service/persistent/PersistentReplicatorInflightTaskTest.java | 2 +-
25 files changed, 30 insertions(+), 22 deletions(-)
diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml
index 68d484ecd36..6128441cb1a 100644
--- a/.github/workflows/pulsar-ci.yaml
+++ b/.github/workflows/pulsar-ci.yaml
@@ -192,6 +192,8 @@ jobs:
group: BROKER_GROUP_3
- name: Brokers - Broker Group 4
group: BROKER_GROUP_4
+ - name: Brokers - Broker Group 5
+ group: BROKER_GROUP_5
- name: Brokers - Client Api
group: BROKER_CLIENT_API
- name: Brokers - Client Impl
diff --git a/build/run_unit_group.sh b/build/run_unit_group.sh
index 723653b2265..2ea2b0bafd0 100755
--- a/build/run_unit_group.sh
+++ b/build/run_unit_group.sh
@@ -93,6 +93,10 @@ function test_group_broker_group_4() {
mvn_test -pl pulsar-broker -Dgroups='cluster-migration'
}
+function test_group_broker_group_5() {
+ mvn_test -pl pulsar-broker -Dgroups='broker-replication'
+}
+
function test_group_broker_client_api() {
mvn_test -pl pulsar-broker -Dgroups='broker-api'
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/SameAuthParamsLookupAutoClusterFailoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/SameAuthParamsLookupAutoClusterFailoverTest.java
index 19859d17d50..0a2a74d9a22 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/SameAuthParamsLookupAutoClusterFailoverTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/SameAuthParamsLookupAutoClusterFailoverTest.java
@@ -42,6 +42,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "broker-replication")
public class SameAuthParamsLookupAutoClusterFailoverTest extends OneWayReplicatorTestBase {
public void setup() throws Exception {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
index 532c4b7858b..af501442fd4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java
@@ -50,7 +50,7 @@ import org.awaitility.reflect.WhiteboxImpl;
import org.testng.Assert;
import org.testng.annotations.Test;
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class AbstractReplicatorTest {
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/DisabledCreateTopicToRemoteClusterForReplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/DisabledCreateTopicToRemoteClusterForReplicationTest.java
index 0f8db4aaa73..75e2c548829 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/DisabledCreateTopicToRemoteClusterForReplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/DisabledCreateTopicToRemoteClusterForReplicationTest.java
@@ -44,7 +44,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class DisabledCreateTopicToRemoteClusterForReplicationTest extends OneWayReplicatorTestBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java
index 8a195a5f48f..3281bdb8c8f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java
@@ -90,7 +90,7 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class OneWayReplicatorDeduplicationTest extends OneWayReplicatorTestBase {
static final ObjectMapper JACKSON = new ObjectMapper();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index 798f4047bfe..4ffb34a2b7f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -122,7 +122,7 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
index 9f4a81b8d0b..c298b37a0e2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalPartitionedTest.java
@@ -46,7 +46,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class OneWayReplicatorUsingGlobalPartitionedTest extends OneWayReplicatorTest {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
index 12febb0620b..c33bda24b23 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
@@ -44,7 +44,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java
index 9f655921794..8de6a7a448d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PeerReplicatorTest.java
@@ -44,7 +44,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class PeerReplicatorTest extends ReplicatorTestBase {
@Override
@@ -187,7 +187,7 @@ public class PeerReplicatorTest extends ReplicatorTestBase {
*
* @throws Exception
*/
- @Test(groups = "broker")
+ @Test
public void testPeerClusterInReplicationClusterListChange() throws Exception {
// clean up peer-clusters
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcTest.java
index 01ace3672f6..cc255b52fe8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcTest.java
@@ -45,7 +45,7 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicationTopicGcTest extends OneWayReplicatorTestBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcUsingGlobalZKTest.java
index ce6eddc5698..fabfb3333e9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcUsingGlobalZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTopicGcUsingGlobalZKTest.java
@@ -25,7 +25,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicationTopicGcUsingGlobalZKTest extends ReplicationTopicGcTest {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTxnTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTxnTest.java
index 3caf4a1f239..aa67474207a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTxnTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicationTxnTest.java
@@ -59,7 +59,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicationTxnTest extends OneWayReplicatorTestBase {
private boolean transactionBufferSegmentedSnapshotEnabled = false;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java
index a5d14ca0487..eb3fce6b307 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsTest.java
@@ -31,7 +31,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicatorAdminTlsTest extends ReplicatorTestBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java
index 3d3eb3faa72..3c0af6b429f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorAdminTlsWithKeyStoreTest.java
@@ -31,7 +31,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicatorAdminTlsWithKeyStoreTest extends ReplicatorTestBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java
index c09809d4ad8..8492087f3f1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java
@@ -47,7 +47,7 @@ import org.testng.annotations.Test;
* a lot of topic deletion and makes namespace policies being incorrect.
*/
@Slf4j
-@Test(groups = "broker-impl")
+@Test(groups = "broker-replication")
public class ReplicatorGlobalNSTest extends ReplicatorTestBase {
protected String methodName;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
index d92bc6005fd..90254bd971d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
@@ -47,7 +47,7 @@ import org.testng.annotations.Test;
/**
* Starts 3 brokers that are in 3 different clusters
*/
-@Test(groups = "quarantine")
+@Test(groups = "broker-replication")
public class ReplicatorRateLimiterTest extends ReplicatorTestBase {
protected String methodName;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRemoveClusterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRemoveClusterTest.java
index 4a74966a3b6..5040947816b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRemoveClusterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRemoveClusterTest.java
@@ -39,7 +39,7 @@ import org.testng.annotations.Test;
/**
* Starts 3 brokers that are in 3 different clusters
*/
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicatorRemoveClusterTest extends ReplicatorTestBase {
protected String methodName;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java
index f038fee7504..e8856026a16 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java
@@ -78,7 +78,7 @@ import org.testng.annotations.Test;
/**
* Tests replicated subscriptions (PIP-33)
*/
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicatorSubscriptionTest extends ReplicatorTestBase {
private static final Logger log = LoggerFactory.getLogger(ReplicatorSubscriptionTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionWithTransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionWithTransactionTest.java
index 93a22a851f1..3907b4e8c6f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionWithTransactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionWithTransactionTest.java
@@ -26,7 +26,7 @@ import org.testng.annotations.Test;
/**
* Tests replicated subscriptions with transaction (PIP-33)
*/
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicatorSubscriptionWithTransactionTest extends ReplicatorSubscriptionTest {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index a5b7b4f3e6b..0467f885367 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -120,7 +120,7 @@ import org.testng.annotations.Test;
/**
* Starts 3 brokers that are in 3 different clusters
*/
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicatorTest extends ReplicatorTestBase {
protected String methodName;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java
index 49e4e795394..7a8e62d8e47 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java
@@ -30,7 +30,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicatorTlsTest extends ReplicatorTestBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTopicPoliciesTest.java
index 9a138ed583a..8879dbf80f0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTopicPoliciesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTopicPoliciesTest.java
@@ -58,7 +58,7 @@ import org.testng.annotations.Test;
/**
* Starts 3 brokers that are in 3 different clusters
*/
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class ReplicatorTopicPoliciesTest extends ReplicatorTestBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TransactionalReplicateSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TransactionalReplicateSubscriptionTest.java
index 2d348f82597..188c4d078d3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TransactionalReplicateSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TransactionalReplicateSubscriptionTest.java
@@ -49,6 +49,7 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+@Test(groups = "broker-replication")
public class TransactionalReplicateSubscriptionTest extends ReplicatorTestBase {
@Override
@BeforeClass(timeOut = 300000)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
index 31970f499fd..9eb3fff9676 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
@@ -46,7 +46,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "broker-replication")
public class PersistentReplicatorInflightTaskTest extends OneWayReplicatorTestBase {
private final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp_");