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 2023/06/07 18:18:36 UTC
[pulsar] branch branch-3.0 updated (1e0ee11b685 -> 9385e3878bd)
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 1e0ee11b685 [improve][broker] Do not expose bucketDelayedIndexStats (#20383)
new c0bb12c324c [fix][ml] There are two same-named managed ledgers in the one broker (#18688)
new 90a82aea99f [fix][broker] Restore solution for certain topic unloading race conditions (#20527)
new 9385e3878bd [fix][test] Reduce flakiness of AdminApi2Test (#20529)
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:
.../mledger/impl/ManagedLedgerFactoryImpl.java | 16 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 9 +
.../mledger/impl/ManagedLedgerFactoryTest.java | 43 +++
.../pulsar/broker/service/BrokerService.java | 4 -
.../broker/service/persistent/PersistentTopic.java | 2 +-
.../apache/pulsar/broker/admin/AdminApi2Test.java | 397 ++++++++++++---------
.../broker/service/BrokerBkEnsemblesTests.java | 4 +-
.../broker/testcontext/PulsarTestContext.java | 4 +
8 files changed, 305 insertions(+), 174 deletions(-)
[pulsar] 02/03: [fix][broker] Restore solution for certain topic unloading race conditions (#20527)
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 90a82aea99fcacdd72fdead672d6e4787ad24760
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Jun 7 16:39:45 2023 +0300
[fix][broker] Restore solution for certain topic unloading race conditions (#20527)
(cherry picked from commit 03f916702ec2a887833543fdea5a2d5305a87302)
---
.../src/main/java/org/apache/pulsar/broker/service/BrokerService.java | 4 ----
.../org/apache/pulsar/broker/service/persistent/PersistentTopic.java | 2 +-
.../java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java | 4 ++--
3 files changed, 3 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 33e5500d623..4a01fec9aba 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -2218,10 +2218,6 @@ public class BrokerService implements Closeable {
return authorizationService;
}
- public CompletableFuture<Void> removeTopicFromCache(String topicName) {
- return removeTopicFutureFromCache(topicName, null);
- }
-
public CompletableFuture<Void> removeTopicFromCache(Topic topic) {
Optional<CompletableFuture<Optional<Topic>>> createTopicFuture = findTopicFutureInCache(topic);
if (createTopicFuture.isEmpty()){
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 98e51a2e3ed..2a0c229daf6 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
@@ -1507,7 +1507,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
private void disposeTopic(CompletableFuture<?> closeFuture) {
- brokerService.removeTopicFromCache(topic)
+ brokerService.removeTopicFromCache(PersistentTopic.this)
.thenRun(() -> {
replicatedSubscriptionsController.ifPresent(ReplicatedSubscriptionsController::close);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
index e69714e539b..9f19bda3647 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
@@ -127,7 +127,7 @@ public class BrokerBkEnsemblesTests extends BkEnsemblesTestBase {
// (3) remove topic and managed-ledger from broker which means topic is not closed gracefully
consumer.close();
producer.close();
- pulsar.getBrokerService().removeTopicFromCache(topic1);
+ pulsar.getBrokerService().removeTopicFromCache(topic);
ManagedLedgerFactoryImpl factory = (ManagedLedgerFactoryImpl) pulsar.getManagedLedgerFactory();
Field field = ManagedLedgerFactoryImpl.class.getDeclaredField("ledgers");
field.setAccessible(true);
@@ -252,7 +252,7 @@ public class BrokerBkEnsemblesTests extends BkEnsemblesTestBase {
// clean managed-ledger and recreate topic to clean any data from the cache
producer.close();
- pulsar.getBrokerService().removeTopicFromCache(topic1);
+ pulsar.getBrokerService().removeTopicFromCache(topic);
ManagedLedgerFactoryImpl factory = (ManagedLedgerFactoryImpl) pulsar.getManagedLedgerFactory();
Field field = ManagedLedgerFactoryImpl.class.getDeclaredField("ledgers");
field.setAccessible(true);
[pulsar] 03/03: [fix][test] Reduce flakiness of AdminApi2Test (#20529)
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 9385e3878bd4e58b35cb479787d22655cb25a81f
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Jun 7 20:25:03 2023 +0300
[fix][test] Reduce flakiness of AdminApi2Test (#20529)
(cherry picked from commit 60dba5d675f998a5108a35be65649edd57ce8596)
---
.../apache/pulsar/broker/admin/AdminApi2Test.java | 397 ++++++++++++---------
.../broker/testcontext/PulsarTestContext.java | 4 +
2 files changed, 237 insertions(+), 164 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
index fb4f880efff..e1800349bbb 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.admin;
import static org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.pulsar.broker.BrokerTestUtil.newUniqueName;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -58,7 +59,6 @@ import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.commons.lang3.reflect.FieldUtils;
-import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.admin.AdminApiTest.MockedPulsarService;
@@ -127,6 +127,7 @@ import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -135,6 +136,10 @@ import org.testng.annotations.Test;
public class AdminApi2Test extends MockedPulsarServiceBaseTest {
private MockedPulsarService mockPulsarSetup;
+ private boolean restartClusterAfterTest;
+ private int usageCount;
+ private String defaultNamespace;
+ private String defaultTenant;
@BeforeClass
@Override
@@ -152,10 +157,17 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Override
protected ServiceConfiguration getDefaultConf() {
ServiceConfiguration conf = super.getDefaultConf();
+ configureDefaults(conf);
+ return conf;
+ }
+
+ void configureDefaults(ServiceConfiguration conf) {
conf.setForceDeleteNamespaceAllowed(true);
conf.setLoadBalancerEnabled(true);
conf.setEnableNamespaceIsolationUpdateOnTime(true);
- return conf;
+ conf.setAllowOverrideEntryFilters(true);
+ conf.setEntryFilterNames(List.of());
+ conf.setMaxNumPartitionsPerPartitionedTopic(0);
}
@AfterClass(alwaysRun = true)
@@ -171,6 +183,20 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@AfterMethod(alwaysRun = true)
public void resetClusters() throws Exception {
+ if (restartClusterAfterTest) {
+ restartClusterAndResetUsageCount();
+ } else {
+ try {
+ cleanupCluster();
+ } catch (Exception e) {
+ log.error("Failed to clean up state by deleting namespaces and tenants after test. "
+ + "Restarting the test broker.", e);
+ restartClusterAndResetUsageCount();
+ }
+ }
+ }
+
+ private void cleanupCluster() throws Exception {
pulsar.getConfiguration().setForceDeleteTenantAllowed(true);
pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true);
for (String tenant : admin.tenants().getTenants()) {
@@ -178,22 +204,59 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
deleteNamespaceWithRetry(namespace, true, admin, pulsar,
mockPulsarSetup.getPulsar());
}
- admin.tenants().deleteTenant(tenant, true);
+ try {
+ admin.tenants().deleteTenant(tenant, true);
+ } catch (Exception e) {
+ log.error("Failed to delete tenant {} after test", tenant, e);
+ String zkDirectory = "/managed-ledgers/" + tenant;
+ try {
+ log.info("Listing {} to see if existing keys are preventing deletion.", zkDirectory);
+ pulsar.getPulsarResources().getLocalMetadataStore().get().getChildren(zkDirectory)
+ .get(5, TimeUnit.SECONDS).forEach(key -> log.info("Child key '{}'", key));
+ } catch (Exception ignore) {
+ log.error("Failed to list tenant {} ZK directory {} after test", tenant, zkDirectory, e);
+ }
+ throw e;
+ }
}
for (String cluster : admin.clusters().getClusters()) {
admin.clusters().deleteCluster(cluster);
}
- resetConfig();
+ configureDefaults(conf);
setupClusters();
}
+ private void restartClusterAfterTest() {
+ restartClusterAfterTest = true;
+ }
+
+ private void restartClusterAndResetUsageCount() throws Exception {
+ cleanup();
+ restartClusterAfterTest = false;
+ usageCount = 0;
+ setup();
+ }
+
+ private void restartClusterIfReused() throws Exception {
+ if (usageCount > 1) {
+ restartClusterAndResetUsageCount();
+ }
+ }
+
+ @BeforeMethod
+ public void increaseUsageCount() {
+ usageCount++;
+ }
+
private void setupClusters() throws PulsarAdminException {
admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
- admin.tenants().createTenant("prop-xyz", tenantInfo);
- admin.namespaces().createNamespace("prop-xyz/ns1", Set.of("test"));
+ defaultTenant = newUniqueName("prop-xyz");
+ admin.tenants().createTenant(defaultTenant, tenantInfo);
+ defaultNamespace = defaultTenant + "/ns1";
+ admin.namespaces().createNamespace(defaultNamespace, Set.of("test"));
}
@DataProvider(name = "topicType")
@@ -250,7 +313,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
final String subName2 = topicName + "-my-sub-2/encode";
final int startPartitions = 4;
final int newPartitions = 8;
- final String partitionedTopicName = "persistent://prop-xyz/ns1/" + topicName;
+ final String partitionedTopicName = "persistent://" + defaultNamespace + "/" + topicName;
URL pulsarUrl = new URL(pulsar.getWebServiceAddress());
@@ -299,7 +362,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
assertEquals(new HashSet<>(admin.topics().getSubscriptions(newPartitionTopicName)),
Set.of(subName1, subName2));
- assertEquals(new HashSet<>(admin.topics().getList("prop-xyz/ns1")).size(), newPartitions);
+ assertEquals(new HashSet<>(admin.topics().getList(defaultNamespace)).size(), newPartitions);
// test cumulative stats for partitioned topic
PartitionedTopicStats topicStats = admin.topics().getPartitionedStats(partitionedTopicName, false);
@@ -333,14 +396,15 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testTopicPoliciesWithMultiBroker() throws Exception {
+ restartClusterAfterTest();
+
//setup cluster with 3 broker
- cleanup();
- setup();
admin.clusters().updateCluster("test",
ClusterData.builder().serviceUrl((pulsar.getWebServiceAddress() + ",localhost:1026," + "localhost:2050")).build());
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
- admin.tenants().createTenant("prop-xyz2", tenantInfo);
- admin.namespaces().createNamespace("prop-xyz2/ns1", Set.of("test"));
+ String tenantName = newUniqueName("prop-xyz2");
+ admin.tenants().createTenant(tenantName, tenantInfo);
+ admin.namespaces().createNamespace(tenantName + "/ns1", Set.of("test"));
conf.setBrokerServicePort(Optional.of(1024));
conf.setBrokerServicePortTls(Optional.of(1025));
conf.setWebServicePort(Optional.of(1026));
@@ -361,14 +425,14 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
PulsarAdmin admin3 = PulsarAdmin.builder().serviceHttpUrl(pulsar3.getWebServiceAddress()).build();
//for partitioned topic, we can get topic policies from every broker
- final String topic = "persistent://prop-xyz2/ns1/" + BrokerTestUtil.newUniqueName("test");
+ final String topic = "persistent://" + tenantName + "/ns1/" + newUniqueName("test");
int partitionNum = 3;
admin.topics().createPartitionedTopic(topic, partitionNum);
pulsarClient.newConsumer().topic(topic).subscriptionName("sub").subscribe().close();
setTopicPoliciesAndValidate(admin2, admin3, topic);
//for non-partitioned topic, we can get topic policies from every broker
- final String topic2 = "persistent://prop-xyz2/ns1/" + BrokerTestUtil.newUniqueName("test");
+ final String topic2 = "persistent://" + tenantName + "/ns1/" + newUniqueName("test");
pulsarClient.newConsumer().topic(topic2).subscriptionName("sub").subscribe().close();
setTopicPoliciesAndValidate(admin2, admin3, topic2);
}
@@ -402,7 +466,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
public void nonPersistentTopics() throws Exception {
final String topicName = "nonPersistentTopic";
- final String nonPersistentTopicName = "non-persistent://prop-xyz/ns1/" + topicName;
+ final String nonPersistentTopicName = "non-persistent://" + defaultNamespace + "/" + topicName;
// Force to create a topic
publishMessagesOnTopic(nonPersistentTopicName, 0, 0);
@@ -434,7 +498,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
assertTrue(topicStats.getSubscriptions().containsKey("my-sub"));
assertEquals(topicStats.getPublishers().size(), 0);
// test partitioned-topic
- final String partitionedTopicName = "non-persistent://prop-xyz/ns1/paritioned";
+ final String partitionedTopicName = "non-persistent://" + defaultNamespace + "/paritioned";
admin.topics().createPartitionedTopic(partitionedTopicName, 5);
assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions, 5);
}
@@ -462,7 +526,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testSetPersistencePolicies() throws Exception {
- final String namespace = "prop-xyz/ns2";
+ final String namespace = newUniqueName(defaultTenant + "/ns2");
admin.namespaces().createNamespace(namespace, Set.of("test"));
assertNull(admin.namespaces().getPersistence(namespace));
@@ -500,7 +564,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testUpdatePersistencePolicyUpdateManagedCursor() throws Exception {
- final String namespace = "prop-xyz/ns2";
+ final String namespace = newUniqueName(defaultTenant + "/ns2");
final String topicName = "persistent://" + namespace + "/topic1";
admin.namespaces().createNamespace(namespace, Set.of("test"));
@@ -541,7 +605,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(dataProvider = "topicType")
public void testUnloadTopic(final String topicType) throws Exception {
- final String namespace = "prop-xyz/ns2";
+ final String namespace = newUniqueName(defaultTenant + "/ns2");
final String topicName = topicType + "://" + namespace + "/topic1";
admin.namespaces().createNamespace(namespace, Set.of("test"));
@@ -591,11 +655,12 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
*/
@Test(dataProvider = "namespaceNames", timeOut = 30000)
public void testResetCursorOnPosition(String namespaceName) throws Exception {
- final String topicName = "persistent://prop-xyz/use/" + namespaceName + "/resetPosition";
+ restartClusterAfterTest();
+ final String topicName = "persistent://" + defaultTenant + "/use/" + namespaceName + "/resetPosition";
final int totalProducedMessages = 50;
// set retention
- admin.namespaces().setRetention("prop-xyz/ns1", new RetentionPolicies(10, 10));
+ admin.namespaces().setRetention(defaultNamespace, new RetentionPolicies(10, 10));
// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName("my-sub")
@@ -674,14 +739,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
consumer.close();
-
- cleanup();
- setup();
}
@Test
public void shouldNotSupportResetOnPartitionedTopic() throws PulsarAdminException, PulsarClientException {
- final String partitionedTopicName = "persistent://prop-xyz/ns1/" + BrokerTestUtil.newUniqueName("parttopic");
+ final String partitionedTopicName = "persistent://" + defaultNamespace + "/" + newUniqueName("parttopic");
admin.topics().createPartitionedTopic(partitionedTopicName, 4);
@Cleanup
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(partitionedTopicName).subscriptionName("my-sub")
@@ -713,7 +775,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 20000)
public void testMaxConsumersOnSubApi() throws Exception {
- final String namespace = "prop-xyz/ns1";
+ final String namespace = newUniqueName(defaultTenant + "/ns");
+ admin.namespaces().createNamespace(namespace, Set.of("test"));
+
assertNull(admin.namespaces().getMaxConsumersPerSubscription(namespace));
admin.namespaces().setMaxConsumersPerSubscription(namespace, 10);
Awaitility.await().untilAsserted(() -> {
@@ -732,7 +796,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
*/
@Test
public void testLoadReportApi() throws Exception {
-
+ restartClusterAfterTest();
this.conf.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName());
@Cleanup("cleanup")
MockedPulsarService mockPulsarSetup1 = new MockedPulsarService(this.conf);
@@ -795,6 +859,8 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
*/
@Test
public void testReplicationPeerCluster() throws Exception {
+ restartClusterAfterTest();
+
admin.clusters().createCluster("us-west1",
ClusterData.builder().serviceUrl("http://broker.messaging.west1.example.com:8080").build());
admin.clusters().createCluster("us-west2",
@@ -814,7 +880,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
assertEquals(allClusters,
List.of("test", "us-east1", "us-east2", "us-west1", "us-west2", "us-west3", "us-west4"));
- final String property = "peer-prop";
+ final String property = newUniqueName("peer-prop");
Set<String> allowedClusters = Set.of("us-west1", "us-west2", "us-west3", "us-west4", "us-east1",
"us-east2", "global");
TenantInfoImpl propConfig = new TenantInfoImpl(Set.of("test"), allowedClusters);
@@ -848,9 +914,6 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
clusterIds = Set.of("us-west1", "us-west4");
// no peer coexist in replication clusters
admin.namespaces().setNamespaceReplicationClusters(namespace, clusterIds);
-
- cleanup();
- setup();
}
@Test
@@ -889,7 +952,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void namespaceAntiAffinity() throws PulsarAdminException {
- final String namespace = "prop-xyz/ns1";
+ final String namespace = defaultNamespace;
final String antiAffinityGroup = "group";
assertTrue(isBlank(admin.namespaces().getNamespaceAntiAffinityGroup(namespace)));
admin.namespaces().setNamespaceAntiAffinityGroup(namespace, antiAffinityGroup);
@@ -897,9 +960,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
admin.namespaces().deleteNamespaceAntiAffinityGroup(namespace);
assertTrue(isBlank(admin.namespaces().getNamespaceAntiAffinityGroup(namespace)));
- final String ns1 = "prop-xyz/antiAG1";
- final String ns2 = "prop-xyz/antiAG2";
- final String ns3 = "prop-xyz/antiAG3";
+ final String ns1 = defaultTenant + "/antiAG1";
+ final String ns2 = defaultTenant + "/antiAG2";
+ final String ns3 = defaultTenant + "/antiAG3";
admin.namespaces().createNamespace(ns1, Set.of("test"));
admin.namespaces().createNamespace(ns2, Set.of("test"));
admin.namespaces().createNamespace(ns3, Set.of("test"));
@@ -908,19 +971,19 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
admin.namespaces().setNamespaceAntiAffinityGroup(ns3, antiAffinityGroup);
Set<String> namespaces = new HashSet<>(
- admin.namespaces().getAntiAffinityNamespaces("prop-xyz", "test", antiAffinityGroup));
+ admin.namespaces().getAntiAffinityNamespaces(defaultTenant, "test", antiAffinityGroup));
assertEquals(namespaces.size(), 3);
assertTrue(namespaces.contains(ns1));
assertTrue(namespaces.contains(ns2));
assertTrue(namespaces.contains(ns3));
- List<String> namespaces2 = admin.namespaces().getAntiAffinityNamespaces("prop-xyz", "test", "invalid-group");
+ List<String> namespaces2 = admin.namespaces().getAntiAffinityNamespaces(defaultTenant, "test", "invalid-group");
assertEquals(namespaces2.size(), 0);
}
@Test
public void testPersistentTopicList() throws Exception {
- final String namespace = "prop-xyz/ns2";
+ final String namespace = newUniqueName(defaultTenant + "/ns2");
final String topicName = "non-persistent://" + namespace + "/bundle-topic";
admin.namespaces().createNamespace(namespace, 20);
admin.namespaces().setNamespaceReplicationClusters(namespace, Set.of("test"));
@@ -954,7 +1017,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testCreateAndGetTopicProperties() throws Exception {
- final String namespace = "prop-xyz/ns2";
+ final String namespace = newUniqueName(defaultTenant + "/ns2");
final String nonPartitionedTopicName = "persistent://" + namespace + "/non-partitioned-TopicProperties";
admin.namespaces().createNamespace(namespace, 20);
Map<String, String> nonPartitionedTopicProperties = new HashMap<>();
@@ -975,7 +1038,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testUpdatePartitionedTopicProperties() throws Exception {
- final String namespace = "prop-xyz/ns2";
+ final String namespace = newUniqueName(defaultTenant + "/ns2");
final String topicName = "persistent://" + namespace + "/testUpdatePartitionedTopicProperties";
final String topicNameTwo = "persistent://" + namespace + "/testUpdatePartitionedTopicProperties2";
admin.namespaces().createNamespace(namespace, 20);
@@ -1033,7 +1096,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testUpdateNonPartitionedTopicProperties() throws Exception {
- final String namespace = "prop-xyz/ns2";
+ final String namespace = newUniqueName(defaultTenant + "/ns2");
final String topicName = "persistent://" + namespace + "/testUpdateNonPartitionedTopicProperties";
admin.namespaces().createNamespace(namespace, 20);
@@ -1068,7 +1131,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testNonPersistentTopics() throws Exception {
- final String namespace = "prop-xyz/ns2";
+ final String namespace = newUniqueName(defaultTenant + "/ns2");
final String topicName = "non-persistent://" + namespace + "/topic";
admin.namespaces().createNamespace(namespace, 20);
admin.namespaces().setNamespaceReplicationClusters(namespace, Set.of("test"));
@@ -1096,7 +1159,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
public void testPublishConsumerStats() throws Exception {
final String topicName = "statTopic";
final String subscriberName = topicName + "-my-sub-1";
- final String topic = "persistent://prop-xyz/ns1/" + topicName;
+ final String topic = "persistent://" + defaultNamespace + "/" + topicName;
final String producerName = "myProducer";
@Cleanup
@@ -1143,6 +1206,8 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testTenantNameWithUnderscore() throws Exception {
+ restartClusterAfterTest();
+
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
admin.tenants().createTenant("prop_xyz", tenantInfo);
@@ -1150,15 +1215,12 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
String topic = "persistent://prop_xyz/use/my-namespace/my-topic";
+ @Cleanup
Producer<byte[]> producer = pulsarClient.newProducer().topic(topic)
.create();
TopicStats stats = admin.topics().getStats(topic);
assertEquals(stats.getPublishers().size(), 1);
- producer.close();
-
- cleanup();
- setup();
}
@Test
@@ -1200,11 +1262,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
assertFalse(admin.tenants().getTenants().contains("test-tenant"));
// Check existing tenant
- assertTrue(admin.tenants().getTenants().contains("prop-xyz"));
+ assertTrue(admin.tenants().getTenants().contains(defaultTenant));
// If we try to update existing tenant with nonexistent clusters, it should fail immediately
try {
- admin.tenants().updateTenant("prop-xyz", tenantInfo);
+ admin.tenants().updateTenant(defaultTenant, tenantInfo);
fail("Should have failed");
} catch (PulsarAdminException e) {
assertEquals(e.getStatusCode(), Status.PRECONDITION_FAILED.getStatusCode());
@@ -1264,7 +1326,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void brokerNamespaceIsolationPoliciesUpdateOnTime() throws Exception {
String brokerName = pulsar.getAdvertisedAddress();
- String ns1Name = "prop-xyz/test_ns1_iso_" + System.currentTimeMillis();
+ String ns1Name = defaultTenant + "/test_ns1_iso_" + System.currentTimeMillis();
admin.namespaces().createNamespace(ns1Name, Set.of("test"));
// 0. without isolation policy configured, lookup will success.
@@ -1335,15 +1397,16 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
*/
@Test
public void testClusterIsReadyBeforeCreateTopic() throws Exception {
+ restartClusterAfterTest();
final String topicName = "partitionedTopic";
final int partitions = 4;
- final String persistentPartitionedTopicName = "persistent://prop-xyz/ns2/" + topicName;
- final String NonPersistentPartitionedTopicName = "non-persistent://prop-xyz/ns2/" + topicName;
+ final String persistentPartitionedTopicName = "persistent://" + defaultTenant + "/ns2/" + topicName;
+ final String NonPersistentPartitionedTopicName = "non-persistent://" + defaultTenant + "/ns2/" + topicName;
- admin.namespaces().createNamespace("prop-xyz/ns2");
+ admin.namespaces().createNamespace(defaultTenant + "/ns2");
// By default the cluster will configure as configuration file. So the create topic operation
// will never throw exception except there is no cluster.
- admin.namespaces().setNamespaceReplicationClusters("prop-xyz/ns2", new HashSet<String>());
+ admin.namespaces().setNamespaceReplicationClusters(defaultTenant + "/ns2", new HashSet<String>());
try {
admin.topics().createPartitionedTopic(persistentPartitionedTopicName, partitions);
@@ -1356,15 +1419,12 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
Assert.fail("should have failed due to Namespace does not have any clusters configured");
} catch (PulsarAdminException.PreconditionFailedException ignored) {
}
-
- cleanup();
- setup();
}
@Test
public void testCreateNamespaceWithNoClusters() throws PulsarAdminException {
String localCluster = pulsar.getConfiguration().getClusterName();
- String namespace = "prop-xyz/test-ns-with-no-clusters";
+ String namespace = newUniqueName(defaultTenant + "/test-ns-with-no-clusters");
admin.namespaces().createNamespace(namespace);
// Global cluster, if there, should be omitted from the results
@@ -1377,7 +1437,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
long timestamp = System.currentTimeMillis();
final String topicName = "consumer-stats-" + timestamp;
final String subscribeName = topicName + "-test-stats-sub";
- final String topic = "persistent://prop-xyz/ns1/" + topicName;
+ final String topic = "persistent://" + defaultNamespace + "/" + topicName;
final String producerName = "producer-" + topicName;
@Cleanup
@@ -1480,10 +1540,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 30000)
public void testPreciseBacklog() throws Exception {
- cleanup();
- setup();
+ restartClusterIfReused();
- final String topic = "persistent://prop-xyz/ns1/precise-back-log";
+ final String topic = "persistent://" + defaultNamespace + "/precise-back-log";
final String subName = "sub-name";
@Cleanup
@@ -1533,6 +1592,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testDeleteTenant() throws Exception {
+ restartClusterAfterTest();
// Disabled conf: systemTopicEnabled. see: https://github.com/apache/pulsar/pull/17070
boolean originalSystemTopicEnabled = conf.isSystemTopicEnabled();
if (originalSystemTopicEnabled) {
@@ -1542,7 +1602,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false);
- String tenant = "test-tenant-1";
+ String tenant = newUniqueName("test-tenant-1");
assertFalse(admin.tenants().getTenants().contains(tenant));
// create tenant
@@ -1588,12 +1648,6 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
assertFalse(pulsar.getLocalMetadataStore().exists(partitionedTopicPath).join());
assertFalse(pulsar.getLocalMetadataStore().exists(localPoliciesPath).join());
assertFalse(pulsar.getLocalMetadataStore().exists(bundleDataPath).join());
- // Reset conf: systemTopicEnabled
- if (originalSystemTopicEnabled) {
- cleanup();
- conf.setSystemTopicEnabled(true);
- setup();
- }
}
@Data
@@ -1628,13 +1682,14 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(dataProvider = "namespaceAttrs")
public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception {
+ restartClusterAfterTest();
+
// Set conf.
cleanup();
- NamespaceAttr originalNamespaceAttr = markOriginalNamespaceAttr();
setNamespaceAttr(namespaceAttr);
setup();
- String tenant = "test-tenant";
+ String tenant = newUniqueName("test-tenant");
assertFalse(admin.tenants().getTenants().contains(tenant));
// create tenant
@@ -1675,16 +1730,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
final String bundleDataPath = "/loadbalance/bundle-data/" + namespace;
assertFalse(pulsar.getLocalMetadataStore().exists(bundleDataPath).join());
-
- // Reset config
- cleanup();
- setNamespaceAttr(originalNamespaceAttr);
- setup();
}
@Test
public void testDeleteNamespaceWithTopicPolicies() throws Exception {
- String tenant = "test-tenant";
+ String tenant = newUniqueName("test-tenant");
assertFalse(admin.tenants().getTenants().contains(tenant));
// create tenant
@@ -1732,7 +1782,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 30000)
public void testBacklogNoDelayed() throws PulsarClientException, PulsarAdminException, InterruptedException {
- final String topic = "persistent://prop-xyz/ns1/precise-back-log-no-delayed-" + UUID.randomUUID().toString();
+ final String topic = "persistent://" + defaultNamespace + "/precise-back-log-no-delayed-" + UUID.randomUUID().toString();
final String subName = "sub-name";
@Cleanup
@@ -1788,7 +1838,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testPreciseBacklogForPartitionedTopic() throws PulsarClientException, PulsarAdminException {
- final String topic = "persistent://prop-xyz/ns1/precise-back-log-for-partitioned-topic";
+ final String topic = "persistent://" + defaultNamespace + "/precise-back-log-for-partitioned-topic";
admin.topics().createPartitionedTopic(topic, 2);
final String subName = "sub-name";
@@ -1830,7 +1880,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 30000)
public void testBacklogNoDelayedForPartitionedTopic() throws PulsarClientException, PulsarAdminException, InterruptedException {
- final String topic = "persistent://prop-xyz/ns1/precise-back-log-no-delayed-partitioned-topic";
+ final String topic = "persistent://" + defaultNamespace + "/precise-back-log-no-delayed-partitioned-topic";
admin.topics().createPartitionedTopic(topic, 2);
final String subName = "sub-name";
@@ -1884,7 +1934,8 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testMaxNumPartitionsPerPartitionedTopicSuccess() {
- final String topic = "persistent://prop-xyz/ns1/max-num-partitions-per-partitioned-topic-success";
+ restartClusterAfterTest();
+ final String topic = "persistent://" + defaultNamespace + "/max-num-partitions-per-partitioned-topic-success";
pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(3);
try {
@@ -1899,7 +1950,8 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testMaxNumPartitionsPerPartitionedTopicFailure() {
- final String topic = "persistent://prop-xyz/ns1/max-num-partitions-per-partitioned-topic-failure";
+ restartClusterAfterTest();
+ final String topic = "persistent://" + defaultNamespace + "/max-num-partitions-per-partitioned-topic-failure";
pulsar.getConfiguration().setMaxNumPartitionsPerPartitionedTopic(2);
try {
@@ -1916,21 +1968,24 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testListOfNamespaceBundles() throws Exception {
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
- admin.tenants().createTenant("prop-xyz2", tenantInfo);
- admin.namespaces().createNamespace("prop-xyz2/ns1", 10);
- admin.namespaces().setNamespaceReplicationClusters("prop-xyz2/ns1", Set.of("test"));
- admin.namespaces().createNamespace("prop-xyz2/test/ns2", 10);
- assertEquals(admin.namespaces().getBundles("prop-xyz2/ns1").getNumBundles(), 10);
- assertEquals(admin.namespaces().getBundles("prop-xyz2/test/ns2").getNumBundles(), 10);
+ String tenantName = newUniqueName("prop-xyz2");
+ admin.tenants().createTenant(tenantName, tenantInfo);
+ admin.namespaces().createNamespace(tenantName + "/ns1", 10);
+ admin.namespaces().setNamespaceReplicationClusters(tenantName + "/ns1", Set.of("test"));
+ admin.namespaces().createNamespace(tenantName + "/test/ns2", 10);
+ assertEquals(admin.namespaces().getBundles(tenantName + "/ns1").getNumBundles(), 10);
+ assertEquals(admin.namespaces().getBundles(tenantName + "/test/ns2").getNumBundles(), 10);
- admin.namespaces().deleteNamespace("prop-xyz2/test/ns2");
+ admin.namespaces().deleteNamespace(tenantName + "/test/ns2");
}
@Test
public void testForceDeleteNamespace() throws Exception {
- final String namespaceName = "prop-xyz2/ns1";
+ restartClusterAfterTest();
+ String tenantName = newUniqueName("prop-xyz2");
+ final String namespaceName = tenantName + "/ns1";
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
- admin.tenants().createTenant("prop-xyz2", tenantInfo);
+ admin.tenants().createTenant(tenantName, tenantInfo);
admin.namespaces().createNamespace(namespaceName, 1);
final String topic = "persistent://" + namespaceName + "/test" + UUID.randomUUID();
pulsarClient.newProducer(Schema.DOUBLE).topic(topic).create().close();
@@ -1942,17 +1997,15 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
} catch (PulsarAdminException e) {
assertEquals(e.getStatusCode(), 404);
}
-
- cleanup();
- setup();
}
@Test
public void testForceDeleteNamespaceWithAutomaticTopicCreation() throws Exception {
conf.setForceDeleteNamespaceAllowed(true);
- final String namespaceName = "prop-xyz2/ns1";
+ String tenantName = newUniqueName("prop-xyz2");
+ final String namespaceName = tenantName + "/ns1";
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
- admin.tenants().createTenant("prop-xyz2", tenantInfo);
+ admin.tenants().createTenant(tenantName, tenantInfo);
admin.namespaces().createNamespace(namespaceName, 1);
admin.namespaces().setAutoTopicCreation(namespaceName,
AutoTopicCreationOverride.builder()
@@ -1977,7 +2030,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
// the consumer will try to re-create the partitions
admin.namespaces().deleteNamespace(namespaceName, true);
- assertFalse(admin.namespaces().getNamespaces("prop-xyz2").contains("ns1"));
+ assertFalse(admin.namespaces().getNamespaces(tenantName).contains("ns1"));
}
}
@@ -2000,6 +2053,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testMaxNamespacesPerTenant() throws Exception {
+ restartClusterAfterTest();
cleanup();
conf.setMaxNamespacesPerTenant(2);
setup();
@@ -2013,19 +2067,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
Assert.assertEquals(e.getStatusCode(), 412);
Assert.assertEquals(e.getHttpError(), "Exceed the maximum number of namespace in tenant :testTenant");
}
-
- //unlimited
- cleanup();
- conf.setMaxNamespacesPerTenant(0);
- setup();
- admin.tenants().createTenant("testTenant", tenantInfo);
- for (int i = 0; i < 10; i++) {
- admin.namespaces().createNamespace("testTenant/ns-" + i, Set.of("test"));
- }
}
@Test
public void testAutoTopicCreationOverrideWithMaxNumPartitionsLimit() throws Exception{
+ restartClusterAfterTest();
cleanup();
conf.setMaxNumPartitionsPerPartitionedTopic(10);
setup();
@@ -2067,6 +2113,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
@Test
public void testMaxTopicsPerNamespace() throws Exception {
+ restartClusterAfterTest();
cleanup();
conf.setMaxTopicsPerNamespace(10);
setup();
@@ -2158,16 +2205,12 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
} catch (PulsarClientException e) {
log.info("Exception: ", e);
}
-
- // reset configuration
- conf.setMaxTopicsPerNamespace(0);
- conf.setDefaultNumPartitions(1);
}
@Test
public void testInvalidBundleErrorResponse() throws Exception {
try {
- admin.namespaces().deleteNamespaceBundle("prop-xyz/ns1", "invalid-bundle");
+ admin.namespaces().deleteNamespaceBundle(defaultNamespace, "invalid-bundle");
fail("should have failed due to invalid bundle");
} catch (PreconditionFailedException e) {
assertTrue(e.getMessage().startsWith("Invalid bundle range"));
@@ -2176,6 +2219,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testMaxSubscriptionsPerTopic() throws Exception {
+ restartClusterAfterTest();
cleanup();
conf.setMaxSubscriptionsPerTopic(2);
setup();
@@ -2258,7 +2302,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 30000)
public void testMaxSubPerTopicApi() throws Exception {
- final String myNamespace = "prop-xyz/ns" + UUID.randomUUID();
+ final String myNamespace = newUniqueName(defaultTenant + "/ns");
admin.namespaces().createNamespace(myNamespace, Set.of("test"));
assertNull(admin.namespaces().getMaxSubscriptionsPerTopic(myNamespace));
@@ -2281,8 +2325,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 60000)
public void testSetNamespaceEntryFilters() throws Exception {
+ restartClusterAfterTest();
+ restartClusterIfReused();
+ @Cleanup
final MockEntryFilterProvider testEntryFilterProvider =
new MockEntryFilterProvider(conf);
@@ -2299,7 +2346,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
try {
EntryFilters entryFilters = new EntryFilters("test");
- final String myNamespace = "prop-xyz/ns" + UUID.randomUUID();
+ final String myNamespace = newUniqueName(defaultTenant + "/ns");
admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test"));
final String topicName = myNamespace + "/topic";
admin.topics().createNonPartitionedTopic(topicName);
@@ -2359,6 +2406,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(dataProvider = "topicType")
public void testSetTopicLevelEntryFilters(String topicType) throws Exception {
+ restartClusterAfterTest();
+ restartClusterIfReused();
+ @Cleanup
final MockEntryFilterProvider testEntryFilterProvider =
new MockEntryFilterProvider(conf);
@@ -2373,7 +2423,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
"entryFilterProvider", testEntryFilterProvider, true);
try {
EntryFilters entryFilters = new EntryFilters("test");
- final String topic = topicType + "://prop-xyz/ns1/test-schema-validation-enforced";
+ final String topic = topicType + "://" + defaultNamespace + "/test-schema-validation-enforced";
admin.topics().createPartitionedTopic(topic, 1);
final String fullTopicName = topic + TopicName.PARTITIONED_TOPIC_SUFFIX + 0;
@Cleanup
@@ -2431,13 +2481,13 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 60000)
public void testSetEntryFiltersHierarchy() throws Exception {
+ restartClusterAfterTest();
+ restartClusterIfReused();
+ @Cleanup
final MockEntryFilterProvider testEntryFilterProvider =
new MockEntryFilterProvider(conf);
- conf.setEntryFilterNames(List.of("test", "test1"));
- conf.setAllowOverrideEntryFilters(true);
-
testEntryFilterProvider.setMockEntryFilters(new EntryFilterDefinition(
"test",
null,
@@ -2450,9 +2500,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
final EntryFilterProvider oldEntryFilterProvider = pulsar.getBrokerService().getEntryFilterProvider();
FieldUtils.writeField(pulsar.getBrokerService(),
"entryFilterProvider", testEntryFilterProvider, true);
+ conf.setEntryFilterNames(List.of("test", "test1"));
+ conf.setAllowOverrideEntryFilters(true);
try {
- final String topic = "persistent://prop-xyz/ns1/test-schema-validation-enforced";
+ final String topic = "persistent://" + defaultNamespace + "/test-schema-validation-enforced";
admin.topics().createPartitionedTopic(topic, 1);
final String fullTopicName = topic + TopicName.PARTITIONED_TOPIC_SUFFIX + 0;
@Cleanup
@@ -2460,8 +2512,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
.topic(fullTopicName)
.create();
assertNull(admin.topicPolicies().getEntryFiltersPerTopic(topic, false));
- assertEquals(admin.topicPolicies().getEntryFiltersPerTopic(topic, true),
- new EntryFilters("test,test1"));
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(admin.topicPolicies().getEntryFiltersPerTopic(topic, true),
+ new EntryFilters("test,test1")));
assertEquals(pulsar
.getBrokerService()
.getTopic(fullTopicName, false)
@@ -2471,10 +2524,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
.size(), 2);
EntryFilters nsEntryFilters = new EntryFilters("test");
- admin.namespaces().setNamespaceEntryFilters("prop-xyz/ns1", nsEntryFilters);
- assertEquals(admin.namespaces().getNamespaceEntryFilters("prop-xyz/ns1"), nsEntryFilters);
- assertEquals(admin.topicPolicies().getEntryFiltersPerTopic(topic, true),
- new EntryFilters("test"));
+ admin.namespaces().setNamespaceEntryFilters(defaultNamespace, nsEntryFilters);
+ assertEquals(admin.namespaces().getNamespaceEntryFilters(defaultNamespace), nsEntryFilters);
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(admin.topicPolicies().getEntryFiltersPerTopic(topic, true),
+ new EntryFilters("test")));
Awaitility.await().untilAsserted(() -> {
assertEquals(pulsar
.getBrokerService()
@@ -2503,8 +2557,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
admin.topicPolicies().setEntryFiltersPerTopic(topic, topicEntryFilters);
Awaitility.await().untilAsserted(() -> assertEquals(admin.topicPolicies().getEntryFiltersPerTopic(topic,
false), topicEntryFilters));
- assertEquals(admin.topicPolicies().getEntryFiltersPerTopic(topic, true),
- new EntryFilters("test1"));
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(admin.topicPolicies().getEntryFiltersPerTopic(topic, true),
+ new EntryFilters("test1")));
Awaitility.await().untilAsserted(() -> {
assertEquals(pulsar
.getBrokerService()
@@ -2530,8 +2585,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 60000)
public void testValidateNamespaceEntryFilters() throws Exception {
+ restartClusterAfterTest();
+ restartClusterIfReused();
+ @Cleanup
final MockEntryFilterProvider testEntryFilterProvider =
new MockEntryFilterProvider(conf);
@@ -2546,7 +2604,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
"entryFilterProvider", testEntryFilterProvider, true);
try {
- final String myNamespace = "prop-xyz/ns" + UUID.randomUUID();
+ final String myNamespace = newUniqueName(defaultTenant + "/ns");
admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test"));
try {
admin.namespaces().setNamespaceEntryFilters(myNamespace, new EntryFilters("notexists"));
@@ -2585,8 +2643,11 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
}
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 60000)
public void testValidateTopicEntryFilters() throws Exception {
+ restartClusterAfterTest();
+ restartClusterIfReused();
+ @Cleanup
final MockEntryFilterProvider testEntryFilterProvider =
new MockEntryFilterProvider(conf);
@@ -2601,7 +2662,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
"entryFilterProvider", testEntryFilterProvider, true);
try {
- final String myNamespace = "prop-xyz/ns" + UUID.randomUUID();
+ final String myNamespace = newUniqueName(defaultTenant + "/ns");
admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test"));
final String topicName = myNamespace + "/topic";
admin.topics().createNonPartitionedTopic(topicName);
@@ -2648,8 +2709,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 30000)
public void testMaxSubPerTopic() throws Exception {
+ restartClusterAfterTest();
pulsar.getConfiguration().setMaxSubscriptionsPerTopic(0);
- final String myNamespace = "prop-xyz/ns" + UUID.randomUUID();
+ final String myNamespace = newUniqueName(defaultTenant + "/ns");
admin.namespaces().createNamespace(myNamespace, Set.of("test"));
final String topic = "persistent://" + myNamespace + "/testMaxSubPerTopic";
pulsarClient.newProducer().topic(topic).create().close();
@@ -2689,12 +2751,13 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 30000)
public void testMaxSubPerTopicPriority() throws Exception {
+ restartClusterAfterTest();
final int brokerLevelMaxSub = 2;
cleanup();
conf.setMaxSubscriptionsPerTopic(brokerLevelMaxSub);
setup();
- final String myNamespace = "prop-xyz/ns" + UUID.randomUUID();
+ final String myNamespace = newUniqueName(defaultTenant + "/ns");
admin.namespaces().createNamespace(myNamespace, Set.of("test"));
final String topic = "persistent://" + myNamespace + "/testMaxSubPerTopic";
//Create a client that can fail quickly
@@ -2746,12 +2809,13 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testMaxProducersPerTopicUnlimited() throws Exception {
+ restartClusterAfterTest();
final int maxProducersPerTopic = 1;
cleanup();
conf.setMaxProducersPerTopic(maxProducersPerTopic);
setup();
//init namespace
- final String myNamespace = "prop-xyz/ns" + UUID.randomUUID();
+ final String myNamespace = newUniqueName(defaultTenant + "/ns");
admin.namespaces().createNamespace(myNamespace, Set.of("test"));
final String topic = "persistent://" + myNamespace + "/testMaxProducersPerTopicUnlimited";
//the policy is set to 0, so there will be no restrictions
@@ -2797,12 +2861,13 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testMaxConsumersPerTopicUnlimited() throws Exception {
+ restartClusterAfterTest();
final int maxConsumersPerTopic = 1;
cleanup();
conf.setMaxConsumersPerTopic(maxConsumersPerTopic);
setup();
//init namespace
- final String myNamespace = "prop-xyz/ns" + UUID.randomUUID();
+ final String myNamespace = newUniqueName(defaultTenant + "/ns");
admin.namespaces().createNamespace(myNamespace, Set.of("test"));
final String topic = "persistent://" + myNamespace + "/testMaxConsumersPerTopicUnlimited";
@@ -2854,7 +2919,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testClearBacklogForTheSubscriptionThatNoConsumers() throws Exception {
- final String topic = "persistent://prop-xyz/ns1/clear_backlog_no_consumers" + UUID.randomUUID().toString();
+ final String topic = "persistent://" + defaultNamespace + "/clear_backlog_no_consumers" + UUID.randomUUID().toString();
final String sub = "my-sub";
admin.topics().createNonPartitionedTopic(topic);
admin.topics().createSubscription(topic, sub, MessageId.earliest);
@@ -2863,7 +2928,10 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 200000)
public void testCompactionApi() throws Exception {
- final String namespace = "prop-xyz/ns1";
+ final String namespace = newUniqueName(defaultTenant + "/ns");
+ admin.namespaces().createNamespace(namespace, Set.of("test"));
+
+
assertNull(admin.namespaces().getCompactionThreshold(namespace));
assertEquals(pulsar.getConfiguration().getBrokerServiceCompactionThresholdInBytes(), 0);
@@ -2878,11 +2946,13 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(timeOut = 200000)
public void testCompactionPriority() throws Exception {
+ restartClusterAfterTest();
cleanup();
conf.setBrokerServiceCompactionMonitorIntervalInSeconds(10000);
setup();
- final String topic = "persistent://prop-xyz/ns1/topic" + UUID.randomUUID();
- final String namespace = "prop-xyz/ns1";
+ final String namespace = newUniqueName(defaultTenant + "/ns");
+ admin.namespaces().createNamespace(namespace, Set.of("test"));
+ final String topic = "persistent://" + namespace + "/topic" + UUID.randomUUID();
pulsarClient.newProducer().topic(topic).create().close();
TopicName topicName = TopicName.get(topic);
PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get();
@@ -2921,7 +2991,8 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testProperties() throws Exception {
- final String namespace = "prop-xyz/ns1";
+ final String namespace = newUniqueName(defaultTenant + "/ns");
+ admin.namespaces().createNamespace(namespace, Set.of("test"));
admin.namespaces().setProperty(namespace, "a", "a");
assertEquals("a", admin.namespaces().getProperty(namespace, "a"));
assertNull(admin.namespaces().getProperty(namespace, "b"));
@@ -2944,7 +3015,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testGetListInBundle() throws Exception {
- final String namespace = "prop-xyz/ns11";
+ final String namespace = defaultTenant + "/ns11";
admin.namespaces().createNamespace(namespace, 3);
final String persistentTopicName = TopicName.get(
@@ -2978,7 +3049,8 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testGetTopicsWithDifferentMode() throws Exception {
- final String namespace = "prop-xyz/ns1";
+ final String namespace = newUniqueName(defaultTenant + "/ns");
+ admin.namespaces().createNamespace(namespace, Set.of("test"));
final String persistentTopicName = TopicName
.get("persistent", NamespaceName.get(namespace), "get_topics_mode_" + UUID.randomUUID().toString())
@@ -3021,16 +3093,14 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(dataProvider = "isV1")
public void testNonPartitionedTopic(boolean isV1) throws Exception {
- String tenant = "prop-xyz";
+ restartClusterAfterTest();
+ String tenant = defaultTenant;
String cluster = "test";
String namespace = tenant + "/" + (isV1 ? cluster + "/" : "") + "n1" + isV1;
String topic = "persistent://" + namespace + "/t1" + isV1;
admin.namespaces().createNamespace(namespace, Set.of(cluster));
admin.topics().createNonPartitionedTopic(topic);
assertTrue(admin.topics().getList(namespace).contains(topic));
-
- cleanup();
- setup();
}
/**
@@ -3043,7 +3113,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
final String subName1 = topicName + "-my-sub-1";
final int startPartitions = 4;
final int newPartitions = 8;
- final String partitionedTopicName = "persistent://prop-xyz/ns1/" + topicName;
+ final String partitionedTopicName = "persistent://" + defaultNamespace + "/" + topicName;
URL pulsarUrl = new URL(pulsar.getWebServiceAddress());
@@ -3079,11 +3149,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(dataProvider = "topicType")
public void testPartitionedStatsAggregationByProducerName(String topicType) throws Exception {
- cleanup();
- setup();
-
+ restartClusterIfReused();
conf.setAggregatePublisherStatsByProducerName(true);
- final String topic = topicType + "://prop-xyz/ns1/test-partitioned-stats-aggregation-by-producer-name";
+ final String topic = topicType + "://" + defaultNamespace + "/test-partitioned-stats-aggregation-by-producer-name";
admin.topics().createPartitionedTopic(topic, 10);
@Cleanup
@@ -3137,8 +3205,9 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(dataProvider = "topicType")
public void testPartitionedStatsAggregationByProducerNamePerPartition(String topicType) throws Exception {
+ restartClusterIfReused();
conf.setAggregatePublisherStatsByProducerName(true);
- final String topic = topicType + "://prop-xyz/ns1/test-partitioned-stats-aggregation-by-producer-name-per-pt";
+ final String topic = topicType + "://" + defaultNamespace + "/test-partitioned-stats-aggregation-by-producer-name-per-pt";
admin.topics().createPartitionedTopic(topic, 2);
@Cleanup
@@ -3161,7 +3230,7 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test(dataProvider = "topicType")
public void testSchemaValidationEnforced(String topicType) throws Exception {
- final String topic = topicType + "://prop-xyz/ns1/test-schema-validation-enforced";
+ final String topic = topicType + "://" + defaultNamespace + "/test-schema-validation-enforced";
admin.topics().createPartitionedTopic(topic, 1);
@Cleanup
Producer<byte[]> producer1 = pulsarClient.newProducer()
@@ -3177,31 +3246,31 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
public void testGetNamespaceTopicList() throws Exception {
- final String persistentTopic = "persistent://prop-xyz/ns1/testGetNamespaceTopicList";
- final String nonPersistentTopic = "non-persistent://prop-xyz/ns1/non-testGetNamespaceTopicList";
- final String eventTopic = "persistent://prop-xyz/ns1/__change_events";
+ final String persistentTopic = "persistent://" + defaultNamespace + "/testGetNamespaceTopicList";
+ final String nonPersistentTopic = "non-persistent://" + defaultNamespace + "/non-testGetNamespaceTopicList";
+ final String eventTopic = "persistent://" + defaultNamespace + "/__change_events";
admin.topics().createNonPartitionedTopic(persistentTopic);
Awaitility.await().untilAsserted(() ->
- admin.namespaces().getTopics("prop-xyz/ns1",
+ admin.namespaces().getTopics(defaultNamespace,
ListNamespaceTopicsOptions.builder().mode(Mode.PERSISTENT).includeSystemTopic(true).build())
.contains(eventTopic));
- List<String> notIncludeSystemTopics = admin.namespaces().getTopics("prop-xyz/ns1",
+ List<String> notIncludeSystemTopics = admin.namespaces().getTopics(defaultNamespace,
ListNamespaceTopicsOptions.builder().includeSystemTopic(false).build());
Assert.assertFalse(notIncludeSystemTopics.contains(eventTopic));
@Cleanup
Producer<byte[]> producer = pulsarClient.newProducer()
.topic(nonPersistentTopic)
.create();
- List<String> notPersistentTopics = admin.namespaces().getTopics("prop-xyz/ns1",
+ List<String> notPersistentTopics = admin.namespaces().getTopics(defaultNamespace,
ListNamespaceTopicsOptions.builder().mode(Mode.NON_PERSISTENT).build());
Assert.assertTrue(notPersistentTopics.contains(nonPersistentTopic));
}
@Test
private void testTerminateSystemTopic() throws Exception {
- final String topic = "persistent://prop-xyz/ns1/testTerminateSystemTopic";
+ final String topic = "persistent://" + defaultNamespace + "/testTerminateSystemTopic";
admin.topics().createNonPartitionedTopic(topic);
- final String eventTopic = "persistent://prop-xyz/ns1/__change_events";
+ final String eventTopic = "persistent://" + defaultNamespace + "/__change_events";
admin.topicPolicies().setMaxConsumers(topic, 2);
Awaitility.await().untilAsserted(() -> {
Assert.assertEquals(admin.topicPolicies().getMaxConsumers(topic), Integer.valueOf(2));
@@ -3213,12 +3282,12 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
@Test
private void testDeleteNamespaceForciblyWithManyTopics() throws Exception {
- final String ns = "prop-xyz/ns-testDeleteNamespaceForciblyWithManyTopics";
+ final String ns = defaultTenant + "/ns-testDeleteNamespaceForciblyWithManyTopics";
admin.namespaces().createNamespace(ns, 2);
for (int i = 0; i < 100; i++) {
admin.topics().createPartitionedTopic(String.format("persistent://%s", ns + "/topic" + i), 3);
}
admin.namespaces().deleteNamespace(ns, true);
- Assert.assertFalse(admin.namespaces().getNamespaces("prop-xyz").contains(ns));
+ Assert.assertFalse(admin.namespaces().getNamespaces(defaultTenant).contains(ns));
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java
index e170425ffe4..062047c7133 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java
@@ -290,6 +290,10 @@ public class PulsarTestContext implements AutoCloseable {
if (svcConfig.getManagedLedgerCacheSizeMB() == unconfiguredDefaults.getManagedLedgerCacheSizeMB()) {
svcConfig.setManagedLedgerCacheSizeMB(8);
}
+
+ if (svcConfig.getTopicLoadTimeoutSeconds() == unconfiguredDefaults.getTopicLoadTimeoutSeconds()) {
+ svcConfig.setTopicLoadTimeoutSeconds(10);
+ }
}
/**
[pulsar] 01/03: [fix][ml] There are two same-named managed ledgers in the one broker (#18688)
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 c0bb12c324c1dc1afd945ac927758154d6025415
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Wed Jun 7 18:38:18 2023 +0800
[fix][ml] There are two same-named managed ledgers in the one broker (#18688)
(cherry picked from commit d7186a67fc13ae972a76fbd9ba59b96d8bc7daae)
---
.../mledger/impl/ManagedLedgerFactoryImpl.java | 16 ++++++--
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 9 +++++
.../mledger/impl/ManagedLedgerFactoryTest.java | 43 ++++++++++++++++++++++
3 files changed, 65 insertions(+), 3 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
index 9f3fe9bb0c4..cabd70ce427 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
@@ -504,9 +504,19 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
}
void close(ManagedLedger ledger) {
- // Remove the ledger from the internal factory cache
- ledgers.remove(ledger.getName());
- entryCacheManager.removeEntryCache(ledger.getName());
+ // If the future in map is not done or has exceptionally complete, it means that @param-ledger is not in the
+ // map.
+ CompletableFuture<ManagedLedgerImpl> ledgerFuture = ledgers.get(ledger.getName());
+ if (ledgerFuture == null || !ledgerFuture.isDone() || ledgerFuture.isCompletedExceptionally()){
+ return;
+ }
+ if (ledgerFuture.join() != ledger){
+ return;
+ }
+ // Remove the ledger from the internal factory cache.
+ if (ledgers.remove(ledger.getName(), ledgerFuture)) {
+ entryCacheManager.removeEntryCache(ledger.getName());
+ }
}
public CompletableFuture<Void> shutdownAsync() throws ManagedLedgerException {
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 9b3d7e46aaa..10f7948f553 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1528,6 +1528,15 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public synchronized void createComplete(int rc, final LedgerHandle lh, Object ctx) {
+ if (STATE_UPDATER.get(this) == State.Closed) {
+ if (lh != null) {
+ log.warn("[{}] ledger create completed after the managed ledger is closed rc={} ledger={}, so just"
+ + " close this ledger handle.", name, rc, lh != null ? lh.getId() : -1);
+ lh.closeAsync();
+ }
+ return;
+ }
+
if (log.isDebugEnabled()) {
log.debug("[{}] createComplete rc={} ledger={}", name, rc, lh != null ? lh.getId() : -1);
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
index d49d9ab3e2b..8695759c99f 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
@@ -20,12 +20,16 @@ package org.apache.bookkeeper.mledger.impl;
import static org.testng.Assert.assertEquals;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.CursorInfo;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.MessageRangeInfo;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.awaitility.Awaitility;
+import org.testng.Assert;
import org.testng.annotations.Test;
public class ManagedLedgerFactoryTest extends MockedBookKeeperTestCase {
@@ -71,4 +75,43 @@ public class ManagedLedgerFactoryTest extends MockedBookKeeperTestCase {
assertEquals(mri.to.entryId, 0);
}
+ /**
+ * see: https://github.com/apache/pulsar/pull/18688
+ */
+ @Test
+ public void testConcurrentCloseLedgerAndSwitchLedgerForReproduceIssue() throws Exception {
+ String managedLedgerName = "lg_" + UUID.randomUUID().toString().replaceAll("-", "_");
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setThrottleMarkDelete(1);
+ config.setMaximumRolloverTime(Integer.MAX_VALUE, TimeUnit.SECONDS);
+ config.setMaxEntriesPerLedger(5);
+
+ // create managedLedger once and close it.
+ ManagedLedgerImpl managedLedger1 = (ManagedLedgerImpl) factory.open(managedLedgerName, config);
+ waitManagedLedgerStateEquals(managedLedger1, ManagedLedgerImpl.State.LedgerOpened);
+ managedLedger1.close();
+
+ // create managedLedger the second time.
+ ManagedLedgerImpl managedLedger2 = (ManagedLedgerImpl) factory.open(managedLedgerName, config);
+ waitManagedLedgerStateEquals(managedLedger2, ManagedLedgerImpl.State.LedgerOpened);
+
+ // Mock the task create ledger complete now, it will change the state to another value which not is Closed.
+ // Close managedLedger1 the second time.
+ managedLedger1.createComplete(1, null, null);
+ managedLedger1.close();
+
+ // Verify managedLedger2 is still there.
+ Assert.assertFalse(factory.ledgers.isEmpty());
+ Assert.assertEquals(factory.ledgers.get(managedLedger2.getName()).join(), managedLedger2);
+
+ // cleanup.
+ managedLedger2.close();
+ }
+
+ private void waitManagedLedgerStateEquals(ManagedLedgerImpl managedLedger, ManagedLedgerImpl.State expectedStat){
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertTrue(managedLedger.getState() == expectedStat));
+ }
+
}