Google Git
Sign in
apache / pulsar / refs/heads/v2.11 / . / pulsar-broker / src / test / java / org / apache / pulsar / broker / admin / v1 / V1_AdminApiTest.java
blob: ab83c8fec032e86bb4f67fa992b820eabfa48b9a [file] [log] [blame]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.pulsar.broker.admin.v1;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.collect.BoundType;
import com.google.common.collect.Lists;
import com.google.common.collect.Range;
import com.google.common.hash.Hashing;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.client.InvocationCallback;
import javax.ws.rs.client.WebTarget;
import lombok.Builder;
import lombok.Cleanup;
import lombok.Value;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.namespace.NamespaceEphemeralData;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
import org.apache.pulsar.broker.testcontext.SpyConfig;
import org.apache.pulsar.client.admin.LongRunningProcessStatus;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException;
import org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException;
import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
import org.apache.pulsar.client.admin.internal.LookupImpl;
import org.apache.pulsar.client.admin.internal.TenantsImpl;
import org.apache.pulsar.client.admin.internal.TopicsImpl;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.lookup.data.LookupData;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.AuthAction;
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType;
import org.apache.pulsar.common.policies.data.BacklogQuota.RetentionPolicy;
import org.apache.pulsar.common.policies.data.BrokerAssignment;
import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationData;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.NamespaceIsolationData;
import org.apache.pulsar.common.policies.data.NamespaceIsolationDataImpl;
import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus;
import org.apache.pulsar.common.policies.data.PartitionedTopicStats;
import org.apache.pulsar.common.policies.data.PersistencePolicies;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.PoliciesUtil;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.compaction.Compactor;
import org.apache.pulsar.compaction.PulsarCompactionServiceFactory;
import org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Slf4j
@Test(groups = "broker-admin")
public class V1_AdminApiTest extends MockedPulsarServiceBaseTest {
private static final Logger LOG = LoggerFactory.getLogger(V1_AdminApiTest.class);
private MockedPulsarService mockPulsarSetup;
private PulsarService otherPulsar;
private PulsarAdmin adminTls;
private PulsarAdmin otheradmin;
private NamespaceBundleFactory bundleFactory;
@BeforeClass
@Override
public void setup() throws Exception {
conf.setTopicLevelPoliciesEnabled(false);
conf.setSystemTopicEnabled(false);
conf.setLoadBalancerEnabled(true);
conf.setBrokerServicePortTls(Optional.of(0));
conf.setWebServicePortTls(Optional.of(0));
conf.setTlsCertificateFilePath(BROKER_CERT_FILE_PATH);
conf.setTlsKeyFilePath(BROKER_KEY_FILE_PATH);
conf.setNumExecutorThreadPoolSize(5);
super.internalSetup();
bundleFactory = new NamespaceBundleFactory(pulsar, Hashing.crc32());
adminTls = spy(PulsarAdmin.builder().tlsTrustCertsFilePath(CA_CERT_FILE_PATH)
.serviceHttpUrl(brokerUrlTls.toString()).build());
// create otherbroker to test redirect on calls that need
// namespace ownership
mockPulsarSetup = new MockedPulsarService(this.conf);
mockPulsarSetup.setup();
otherPulsar = mockPulsarSetup.getPulsar();
otheradmin = mockPulsarSetup.getAdmin();
// Setup namespaces
admin.clusters().createCluster("use", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("use"));
admin.tenants().createTenant("prop-xyz", tenantInfo);
admin.namespaces().createNamespace("prop-xyz/use/ns1");
}
@AfterClass(alwaysRun = true)
@Override
public void cleanup() throws Exception {
adminTls.close();
super.internalCleanup();
mockPulsarSetup.cleanup();
}
@Override
protected void customizeMainPulsarTestContextBuilder(PulsarTestContext.Builder pulsarTestContextBuilder) {
pulsarTestContextBuilder.spyConfigCustomizer(
// verify(compactor) is used in this test class
builder -> builder.compactor(SpyConfig.SpyType.SPY_ALSO_INVOCATIONS));
}
@AfterMethod(alwaysRun = true)
public void reset() throws Exception {
pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true);
for (String tenant : admin.tenants().getTenants()) {
for (String namespace : admin.namespaces().getNamespaces(tenant)) {
deleteNamespaceWithRetry(namespace, true, admin, pulsar,
mockPulsarSetup.getPulsar());
}
}
pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false);
resetConfig();
if (!admin.clusters().getClusters().contains("use")) {
admin.clusters().createCluster("use",
ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
}
if (!admin.tenants().getTenants().contains("prop-xyz")) {
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("use"));
admin.tenants().createTenant("prop-xyz", tenantInfo);
}
admin.namespaces().createNamespace("prop-xyz/use/ns1");
}
@DataProvider(name = "numBundles")
public static Object[][] numBundles() {
return new Object[][] { { 1 }, { 4 } };
}
@DataProvider(name = "bundling")
public static Object[][] bundling() {
return new Object[][] { { 0 }, { 4 } };
}
@DataProvider(name = "topicName")
public Object[][] topicNamesProvider() {
return new Object[][] { { "topic_+&*%{}() \\/$@#^%" }, { "simple-topicName" } };
}
@DataProvider(name = "topicType")
public Object[][] topicTypeProvider() {
return new Object[][] { { TopicDomain.persistent.value() }, { TopicDomain.non_persistent.value() } };
}
@Test
public void clusters() throws Exception {
admin.clusters().createCluster("usw",
ClusterData.builder().serviceUrl("http://broker.messaging.use.example.com:8080").build());
// "test" cluster is part of config-default cluster and it's znode gets created when PulsarService creates
// failure-domain znode of this default cluster
assertEquals(admin.clusters().getClusters(), List.of("use", "usw"));
assertEquals(admin.clusters().getCluster("use"),
ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
admin.clusters().updateCluster("usw",
ClusterData.builder().serviceUrl("http://new-broker.messaging.use.example.com:8080").build());
assertEquals(admin.clusters().getClusters(), List.of("use", "usw"));
assertEquals(admin.clusters().getCluster("usw"),
ClusterData.builder().serviceUrl("http://new-broker.messaging.use.example.com:8080").build());
admin.clusters().updateCluster("usw",
ClusterData.builder()
.serviceUrl("http://new-broker.messaging.usw.example.com:8080")
.serviceUrlTls("https://new-broker.messaging.usw.example.com:4443")
.build());
assertEquals(admin.clusters().getClusters(), List.of("use", "usw"));
assertEquals(admin.clusters().getCluster("usw"),
ClusterData.builder()
.serviceUrl("http://new-broker.messaging.usw.example.com:8080")
.serviceUrlTls("https://new-broker.messaging.usw.example.com:4443")
.build());
admin.clusters().deleteCluster("usw");
Thread.sleep(300);
assertEquals(admin.clusters().getClusters(), List.of("use"));
admin.namespaces().deleteNamespace("prop-xyz/use/ns1");
admin.clusters().deleteCluster("use");
assertEquals(admin.clusters().getClusters(), new ArrayList<>());
// Check name validation
try {
admin.clusters().createCluster("bf!", ClusterData.builder().serviceUrl("http://dummy.messaging.example.com").build());
fail("should have failed");
} catch (PulsarAdminException e) {
assertTrue(e instanceof PreconditionFailedException);
}
}
@Test
public void clusterNamespaceIsolationPolicies() throws PulsarAdminException {
try {
// create
String policyName1 = "policy-1";
Map<String, String> parameters1 = new HashMap<>();
parameters1.put("min_limit", "1");
parameters1.put("usage_threshold", "100");
NamespaceIsolationData nsPolicyData1 = NamespaceIsolationData.builder()
.namespaces(Lists.newArrayList("other/use/other.*"))
.primary(Lists.newArrayList(".*")) // match all broker. make it easy to verify `getBrokersWithNamespaceIsolationPolicy` later
.secondary(Lists.newArrayList("prod1-broker.*.messaging.use.example.com"))
.autoFailoverPolicy(AutoFailoverPolicyData.builder()
.policyType(AutoFailoverPolicyType.min_available)
.parameters(parameters1)
.build())
.build();
admin.clusters().createNamespaceIsolationPolicy("use", policyName1, nsPolicyData1);
String policyName2 = "policy-2";
Map<String, String> parameters2 = new HashMap<>();
parameters2.put("min_limit", "1");
parameters2.put("usage_threshold", "100");
NamespaceIsolationData nsPolicyData2 = NamespaceIsolationData.builder()
.namespaces(Lists.newArrayList("other/use/other.*"))
.primary(Lists.newArrayList("prod1-broker[4-6].messaging.use.example.com"))
.secondary(Lists.newArrayList("prod1-broker.*.messaging.use.example.com"))
.autoFailoverPolicy(AutoFailoverPolicyData.builder()
.policyType(AutoFailoverPolicyType.min_available)
.parameters(parameters1)
.build())
.build();
admin.clusters().createNamespaceIsolationPolicy("use", policyName2, nsPolicyData2);
// verify create indirectly with get
Map<String, ? extends NamespaceIsolationData> policiesMap = admin.clusters().getNamespaceIsolationPolicies("use");
assertEquals(policiesMap.get(policyName1), nsPolicyData1);
assertEquals(policiesMap.get(policyName2), nsPolicyData2);
// verify local broker get matched.
List<BrokerNamespaceIsolationData> isoList = admin.clusters().getBrokersWithNamespaceIsolationPolicy("use");
assertEquals(isoList.size(), 1);
assertTrue(isoList.get(0).isPrimary());
assertEquals(isoList.get(0).getPolicyName(), policyName1);
// verify update of primary
nsPolicyData1.getPrimary().remove(0);
nsPolicyData1.getPrimary().add("prod1-broker[1-2].messaging.use.example.com");
admin.clusters().updateNamespaceIsolationPolicy("use", policyName1, nsPolicyData1);
// verify primary change
policiesMap = admin.clusters().getNamespaceIsolationPolicies("use");
assertEquals(policiesMap.get(policyName1), nsPolicyData1);
// verify update of secondary
nsPolicyData1.getSecondary().remove(0);
nsPolicyData1.getSecondary().add("prod1-broker[3-4].messaging.use.example.com");
admin.clusters().updateNamespaceIsolationPolicy("use", policyName1, nsPolicyData1);
// verify secondary change
policiesMap = admin.clusters().getNamespaceIsolationPolicies("use");
assertEquals(policiesMap.get(policyName1), nsPolicyData1);
// verify update of failover policy limit
nsPolicyData1.getAutoFailoverPolicy().getParameters().put("min_limit", "10");
admin.clusters().updateNamespaceIsolationPolicy("use", policyName1, nsPolicyData1);
// verify min_limit change
policiesMap = admin.clusters().getNamespaceIsolationPolicies("use");
assertEquals(policiesMap.get(policyName1), nsPolicyData1);
// verify update of failover usage_threshold limit
nsPolicyData1.getAutoFailoverPolicy().getParameters().put("usage_threshold", "80");
admin.clusters().updateNamespaceIsolationPolicy("use", policyName1, nsPolicyData1);
// verify usage_threshold change
policiesMap = admin.clusters().getNamespaceIsolationPolicies("use");
assertEquals(policiesMap.get(policyName1), nsPolicyData1);
// verify single get
NamespaceIsolationDataImpl policy1Data =
(NamespaceIsolationDataImpl) admin.clusters().getNamespaceIsolationPolicy("use", policyName1);
assertEquals(policy1Data, nsPolicyData1);
// verify creation of more than one policy
admin.clusters().createNamespaceIsolationPolicy("use", policyName2, nsPolicyData1);
try {
admin.clusters().getNamespaceIsolationPolicy("use", "no-such-policy");
fail("should have raised exception");
} catch (PulsarAdminException e) {
assertTrue(e instanceof NotFoundException);
}
// verify delete cluster failed
try {
admin.clusters().deleteCluster("use");
fail("should have raised exception");
} catch (PulsarAdminException e) {
assertTrue(e instanceof PreconditionFailedException);
}
// verify delete
admin.clusters().deleteNamespaceIsolationPolicy("use", policyName1);
admin.clusters().deleteNamespaceIsolationPolicy("use", policyName2);
try {
admin.clusters().getNamespaceIsolationPolicy("use", policyName1);
fail("should have raised exception");
} catch (PulsarAdminException e) {
assertTrue(e instanceof NotFoundException);
}
try {
admin.clusters().getNamespaceIsolationPolicy("use", policyName2);
fail("should have raised exception");
} catch (PulsarAdminException e) {
assertTrue(e instanceof NotFoundException);
}
try {
admin.clusters().getNamespaceIsolationPolicies("usc");
fail("should have raised exception");
} catch (PulsarAdminException e) {
assertTrue(e instanceof NotFoundException);
}
try {
admin.clusters().getNamespaceIsolationPolicy("usc", "no-such-cluster");
fail("should have raised exception");
} catch (PulsarAdminException e) {
assertTrue(e instanceof PreconditionFailedException);
}
try {
admin.clusters().createNamespaceIsolationPolicy("usc", "no-such-cluster", nsPolicyData1);
fail("should have raised exception");
} catch (PulsarAdminException e) {
assertTrue(e instanceof PreconditionFailedException);
}
try {
admin.clusters().updateNamespaceIsolationPolicy("usc", "no-such-cluster", policy1Data);
fail("should have raised exception");
} catch (PulsarAdminException e) {
assertTrue(e instanceof PreconditionFailedException);
}
} catch (PulsarAdminException e) {
LOG.warn("TEST FAILED [{}]", e.getMessage());
throw e;
}
}
@Test
public void brokers() throws Exception {
List<String> list = admin.brokers().getActiveBrokers("use");
Assert.assertNotNull(list);
Assert.assertEquals(list.size(), 1);
List<String> list2 = otheradmin.brokers().getActiveBrokers("test");
Assert.assertNotNull(list2);
Assert.assertEquals(list2.size(), 1);
Map<String, NamespaceOwnershipStatus> nsMap = admin.brokers().getOwnedNamespaces("use", list.get(0));
// since sla-monitor ns is not created nsMap.size() == 1 (for HeartBeat Namespace)
Assert.assertEquals(nsMap.size(), 2);
for (String ns : nsMap.keySet()) {
NamespaceOwnershipStatus nsStatus = nsMap.get(ns);
if (ns.equals(
NamespaceService.getHeartbeatNamespace(pulsar.getAdvertisedAddress(), pulsar.getConfiguration())
+ "/0x00000000_0xffffffff")) {
assertEquals(nsStatus.broker_assignment, BrokerAssignment.shared);
assertFalse(nsStatus.is_controlled);
assertTrue(nsStatus.is_active);
}
}
String[] parts = list.get(0).split(":");
Assert.assertEquals(parts.length, 2);
Map<String, NamespaceOwnershipStatus> nsMap2 = adminTls.brokers().getOwnedNamespaces("use",
String.format("%s:%d", parts[0], pulsar.getListenPortHTTPS().get()));
Assert.assertEquals(nsMap2.size(), 2);
admin.namespaces().deleteNamespace("prop-xyz/use/ns1");
admin.clusters().deleteCluster("use");
assertEquals(admin.clusters().getClusters(), new ArrayList<>());
}
/**
* <pre>
* Verifies: zk-update configuration updates service-config
* 1. create znode for dynamic-config
* 2. start pulsar service so, pulsar can set the watch on that znode
* 3. update the configuration with new value
* 4. wait and verify that new value has been updated
* </pre>
*
* @throws Exception
*/
@Test
public void testUpdateDynamicConfigurationWithZkWatch() throws Exception {
final int initValue = 30000;
long defaultValue = pulsar.getConfiguration().getBrokerShutdownTimeoutMs();
pulsar.getConfiguration().setBrokerShutdownTimeoutMs(initValue);
// (1) try to update dynamic field
final long shutdownTime = 10;
// update configuration
admin.brokers().updateDynamicConfiguration("brokerShutdownTimeoutMs", Long.toString(shutdownTime));
// sleep incrementally as zk-watch notification is async and may take some time
for (int i = 0; i < 5; i++) {
if (pulsar.getConfiguration().getBrokerShutdownTimeoutMs() != initValue) {
Thread.sleep(50 + (i * 10));
}
}
// wait config to be updated
for (int i = 0; i < 5; i++) {
if (pulsar.getConfiguration().getBrokerShutdownTimeoutMs() != shutdownTime) {
Thread.sleep(100 + (i * 10));
} else {
break;
}
}
// verify value is updated
assertEquals(pulsar.getConfiguration().getBrokerShutdownTimeoutMs(), shutdownTime);
// (2) try to update non-dynamic field
try {
admin.brokers().updateDynamicConfiguration("metadataStoreUrl", "zk:test-zk:1234");
} catch (Exception e) {
assertTrue(e instanceof PreconditionFailedException);
}
// (3) try to update non-existent field
try {
admin.brokers().updateDynamicConfiguration("test", Long.toString(shutdownTime));
} catch (Exception e) {
assertTrue(e instanceof PreconditionFailedException);
}
pulsar.getConfiguration().setBrokerShutdownTimeoutMs(defaultValue);
}
/**
* Verifies broker sets watch on dynamic-configuration map even with invalid init json data
*
* <pre>
* 1. Set invalid json at dynamic-config znode
* 2. Broker fails to deserialize znode content but sets the watch on znode
* 3. Update znode with valid json map
* 4. Broker should get watch and update the dynamic-config map
* </pre>
*
* @throws Exception
*/
@Test
public void testInvalidDynamicConfigContentInZK() throws Exception {
final int newValue = 10;
// set invalid data into dynamic-config znode so, broker startup fail to deserialize data
pulsar.getLocalMetadataStore().put("/admin/configuration", "$".getBytes(),
Optional.empty()).join();
stopBroker();
// start broker: it should have set watch even if with failure of deserialization
startBroker();
Assert.assertNotEquals(pulsar.getConfiguration().getBrokerShutdownTimeoutMs(), newValue);
// update zk with config-value which should fire watch and broker should update the config value
Map<String, String> configMap = new HashMap<>();
configMap.put("brokerShutdownTimeoutMs", Integer.toString(newValue));
pulsar.getLocalMetadataStore().put("/admin/configuration",
ObjectMapperFactory.getMapper().writer().writeValueAsBytes(configMap),
Optional.empty()).join();
// wait config to be updated
for (int i = 0; i < 5; i++) {
if (pulsar.getConfiguration().getBrokerShutdownTimeoutMs() != newValue) {
Thread.sleep(100 + (i * 10));
} else {
break;
}
}
// verify value is updated
assertEquals(pulsar.getConfiguration().getBrokerShutdownTimeoutMs(), newValue);
cleanup();
setup();
}
/**
* <pre>
* verifies: that registerListener updates pulsar.config value with newly updated zk-dynamic config
* 1.start pulsar
* 2.update zk-config with admin api
* 3. trigger watch and listener
* 4. verify that config is updated
* </pre>
*
* @throws Exception
*/
@Test
public void testUpdateDynamicLocalConfiguration() throws Exception {
// (1) try to update dynamic field
final long initValue = 30000;
final long shutdownTime = 10;
long defaultValue = pulsar.getConfiguration().getBrokerShutdownTimeoutMs();
pulsar.getConfiguration().setBrokerShutdownTimeoutMs(initValue);
// update configuration
admin.brokers().updateDynamicConfiguration("brokerShutdownTimeoutMs", Long.toString(shutdownTime));
// sleep incrementally as zk-watch notification is async and may take some time
for (int i = 0; i < 5; i++) {
if (pulsar.getConfiguration().getBrokerShutdownTimeoutMs() == initValue) {
Thread.sleep(50 + (i * 10));
}
}
// verify value is updated
assertEquals(pulsar.getConfiguration().getBrokerShutdownTimeoutMs(), shutdownTime);
pulsar.getConfiguration().setBrokerShutdownTimeoutMs(defaultValue);
}
@Test
public void testUpdatableConfigurationName() throws Exception {
// (1) try to update dynamic field
final String configName = "brokerShutdownTimeoutMs";
assertTrue(admin.brokers().getDynamicConfigurationNames().contains(configName));
}
@Test
public void testGetDynamicLocalConfiguration() throws Exception {
// (1) try to update dynamic field
final String configName = "brokerShutdownTimeoutMs";
final long shutdownTime = 10;
long defaultValue = pulsar.getConfiguration().getBrokerShutdownTimeoutMs();
pulsar.getConfiguration().setBrokerShutdownTimeoutMs(30000);
Map<String, String> configs = admin.brokers().getAllDynamicConfigurations();
assertTrue(configs.isEmpty());
assertNotEquals(pulsar.getConfiguration().getBrokerShutdownTimeoutMs(), shutdownTime);
// update configuration
admin.brokers().updateDynamicConfiguration(configName, Long.toString(shutdownTime));
// Now, znode is created: updateConfigurationAndregisterListeners and check if configuration updated
assertEquals(Long.parseLong(admin.brokers().getAllDynamicConfigurations().get(configName)), shutdownTime);
pulsar.getConfiguration().setBrokerShutdownTimeoutMs(defaultValue);
}
@Test
public void testTenant() throws Exception {
Set<String> allowedClusters = Set.of("use");
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), allowedClusters);
admin.tenants().createTenant("prop-xyz2", tenantInfo);
admin.namespaces().createNamespace("prop-xyz2/use/ns1");
assertEquals(admin.tenants().getTenants(), List.of("prop-xyz", "prop-xyz2"));
assertEquals(admin.tenants().getTenantInfo("prop-xyz2"), tenantInfo);
TenantInfoImpl newPropertyAdmin = new TenantInfoImpl(Set.of("role3", "role4"), allowedClusters);
admin.tenants().updateTenant("prop-xyz2", newPropertyAdmin);
assertEquals(admin.tenants().getTenantInfo("prop-xyz2"), newPropertyAdmin);
admin.namespaces().deleteNamespace("prop-xyz2/use/ns1");
admin.tenants().deleteTenant("prop-xyz2");
assertEquals(admin.tenants().getTenants(), List.of("prop-xyz"));
// Check name validation
try {
admin.tenants().createTenant("prop-xyz&", tenantInfo);
fail("should have failed");
} catch (PulsarAdminException e) {
assertTrue(e instanceof PreconditionFailedException);
}
}
@Test
public void namespaces() throws Exception {
admin.clusters().createCluster("usw", ClusterData.builder().build());
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"),
Set.of("use", "usw"));
admin.tenants().updateTenant("prop-xyz", tenantInfo);
assertEquals(admin.namespaces().getPolicies("prop-xyz/use/ns1").bundles, PoliciesUtil.defaultBundle());
admin.namespaces().createNamespace("prop-xyz/use/ns2");
admin.namespaces().createNamespace("prop-xyz/use/ns3", 4);
assertEquals(admin.namespaces().getPolicies("prop-xyz/use/ns3").bundles.getNumBundles(), 4);
assertEquals(admin.namespaces().getPolicies("prop-xyz/use/ns3").bundles.getBoundaries().size(), 5);
admin.namespaces().deleteNamespace("prop-xyz/use/ns3");
try {
admin.namespaces().createNamespace("non-existing/usw/ns1");
fail("Should not have passed");
} catch (NotFoundException e) {
// Ok
}
assertEquals(admin.namespaces().getNamespaces("prop-xyz"),
List.of("prop-xyz/use/ns1", "prop-xyz/use/ns2"));
assertEquals(admin.namespaces().getNamespaces("prop-xyz", "use"),
List.of("prop-xyz/use/ns1", "prop-xyz/use/ns2"));
try {
admin.namespaces().createNamespace("prop-xyz/usc/ns1");
fail("Should not have passed");
} catch (NotAuthorizedException e) {
// Ok, got the non authorized exception since usc cluster is not in the allowed clusters list.
}
// no need to clear cache once authorization-provide also start using metadata-store
clearCache();
admin.namespaces().grantPermissionOnNamespace("prop-xyz/use/ns1", "my-role", EnumSet.allOf(AuthAction.class));
Policies policies = new Policies();
policies.bundles = PoliciesUtil.defaultBundle();
policies.auth_policies.getNamespaceAuthentication().put("my-role", EnumSet.allOf(AuthAction.class));
policies.is_allow_auto_update_schema = conf.isAllowAutoUpdateSchemaEnabled();
assertEquals(admin.namespaces().getPolicies("prop-xyz/use/ns1"), policies);
assertEquals(admin.namespaces().getPermissions("prop-xyz/use/ns1"), policies.auth_policies.getNamespaceAuthentication());
assertEquals(admin.namespaces().getTopics("prop-xyz/use/ns1"), new ArrayList<>());
admin.namespaces().revokePermissionsOnNamespace("prop-xyz/use/ns1", "my-role");
policies.auth_policies.getNamespaceAuthentication().remove("my-role");
policies.is_allow_auto_update_schema = conf.isAllowAutoUpdateSchemaEnabled();
assertEquals(admin.namespaces().getPolicies("prop-xyz/use/ns1"), policies);
assertNull(admin.namespaces().getPersistence("prop-xyz/use/ns1"));
admin.namespaces().setPersistence("prop-xyz/use/ns1", new PersistencePolicies(3, 2, 1, 10.0));
assertEquals(admin.namespaces().getPersistence("prop-xyz/use/ns1"), new PersistencePolicies(3, 2, 1, 10.0));
// Force topic creation and namespace being loaded
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES).topic("persistent://prop-xyz/use/ns1/my-topic").create();
producer.close();
admin.topics().delete("persistent://prop-xyz/use/ns1/my-topic");
admin.namespaces().unloadNamespaceBundle("prop-xyz/use/ns1", "0x00000000_0xffffffff");
NamespaceName ns = NamespaceName.get("prop-xyz/use/ns1");
// Now, w/ bundle policies, we will use default bundle
NamespaceBundle defaultBundle = bundleFactory.getFullBundle(ns);
int i = 0;
for (; i < 10; i++) {
Optional<NamespaceEphemeralData> data1 = pulsar.getNamespaceService().getOwnershipCache()
.getOwnerAsync(defaultBundle).get();
if (!data1.isPresent()) {
// Already unloaded
break;
}
LOG.info("Waiting for unload namespace {} to complete. Current service unit isDisabled: {}", defaultBundle,
data1.get().isDisabled());
Thread.sleep(1000);
}
assertTrue(i < 10);
admin.namespaces().deleteNamespace("prop-xyz/use/ns1");
assertEquals(admin.namespaces().getNamespaces("prop-xyz", "use"), List.of("prop-xyz/use/ns2"));
try {
admin.namespaces().unload("prop-xyz/use/ns1");
fail("should have raised exception");
} catch (Exception e) {
// OK excepted
}
// Force topic creation and namespace being loaded
producer = pulsarClient.newProducer(Schema.BYTES).topic("persistent://prop-xyz/use/ns2/my-topic").create();
producer.close();
admin.topics().delete("persistent://prop-xyz/use/ns2/my-topic");
// both unload and delete should succeed for ns2 on other broker with a redirect
// otheradmin.namespaces().unload("prop-xyz/use/ns2");
tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("use"));
admin.tenants().updateTenant("prop-xyz", tenantInfo);
}
@Test(dataProvider = "topicName")
public void persistentTopics(String topicName) throws Exception {
assertEquals(admin.topics().getList("prop-xyz/use/ns1"), new ArrayList<>());
final String persistentTopicName = "persistent://prop-xyz/use/ns1/" + topicName;
// Force to create a topic
publishMessagesOnPersistentTopic("persistent://prop-xyz/use/ns1/" + topicName, 0);
assertEquals(admin.topics().getList("prop-xyz/use/ns1"),
List.of("persistent://prop-xyz/use/ns1/" + topicName));
// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
.build();
Consumer<byte[]> consumer = client.newConsumer().topic(persistentTopicName).subscriptionName("my-sub")
.subscriptionType(SubscriptionType.Exclusive).subscribe();
assertEquals(admin.topics().getSubscriptions(persistentTopicName), List.of("my-sub"));
publishMessagesOnPersistentTopic("persistent://prop-xyz/use/ns1/" + topicName, 10);
TopicStats topicStats = admin.topics().getStats(persistentTopicName);
assertEquals(topicStats.getSubscriptions().keySet(), new TreeSet<>(List.of("my-sub")));
assertEquals(topicStats.getSubscriptions().get("my-sub").getConsumers().size(), 1);
assertEquals(topicStats.getSubscriptions().get("my-sub").getMsgBacklog(), 10);
assertEquals(topicStats.getPublishers().size(), 0);
PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(persistentTopicName, false);
assertEquals(internalStats.cursors.keySet(), new TreeSet<>(List.of("my-sub")));
List<Message<byte[]>> messages = admin.topics().peekMessages(persistentTopicName, "my-sub", 3);
assertEquals(messages.size(), 3);
for (int i = 0; i < 3; i++) {
String expectedMessage = "message-" + i;
assertEquals(messages.get(i).getData(), expectedMessage.getBytes());
}
messages = admin.topics().peekMessages(persistentTopicName, "my-sub", 15);
assertEquals(messages.size(), 10);
for (int i = 0; i < 10; i++) {
String expectedMessage = "message-" + i;
assertEquals(messages.get(i).getData(), expectedMessage.getBytes());
}
admin.topics().skipMessages(persistentTopicName, "my-sub", 5);
topicStats = admin.topics().getStats(persistentTopicName);
assertEquals(topicStats.getSubscriptions().get("my-sub").getMsgBacklog(), 5);
admin.topics().skipAllMessages(persistentTopicName, "my-sub");
topicStats = admin.topics().getStats(persistentTopicName);
assertEquals(topicStats.getSubscriptions().get("my-sub").getMsgBacklog(), 0);
consumer.close();
client.close();
admin.topics().deleteSubscription(persistentTopicName, "my-sub");
assertEquals(admin.topics().getSubscriptions(persistentTopicName), new ArrayList<>());
topicStats = admin.topics().getStats(persistentTopicName);
assertEquals(topicStats.getSubscriptions().keySet(), new TreeSet<>());
assertEquals(topicStats.getPublishers().size(), 0);
try {
admin.topics().skipAllMessages(persistentTopicName, "my-sub");
} catch (NotFoundException e) {
}
admin.topics().delete(persistentTopicName);
try {
admin.topics().delete(persistentTopicName);
fail("Should have received 404");
} catch (NotFoundException e) {
}
assertEquals(admin.topics().getList("prop-xyz/use/ns1"), new ArrayList<>());
}
@Test(dataProvider = "topicName")
public void partitionedTopics(String topicName) throws Exception {
assertEquals(admin.topics().getPartitionedTopicList("prop-xyz/use/ns1"), new ArrayList<>());
final String partitionedTopicName = "persistent://prop-xyz/use/ns1/" + topicName;
admin.topics().createPartitionedTopic(partitionedTopicName, 4);
assertEquals(admin.topics().getPartitionedTopicList("prop-xyz/use/ns1"),
List.of(partitionedTopicName));
assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions, 4);
List<String> topics = admin.topics().getList("prop-xyz/use/ns1");
assertEquals(topics.size(), 4);
try {
admin.topics().getPartitionedTopicMetadata("persistent://prop-xyz/use/ns1/ds2");
fail("getPartitionedTopicMetadata of persistent://prop-xyz/use/ns1/ds2 should not succeed");
} catch (NotFoundException expected) {
}
// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
.build();
Consumer<byte[]> consumer = client.newConsumer().topic(partitionedTopicName).subscriptionName("my-sub")
.subscriptionType(SubscriptionType.Exclusive).subscribe();
assertEquals(admin.topics().getSubscriptions(partitionedTopicName), List.of("my-sub"));
try {
admin.topics().deleteSubscription(partitionedTopicName, "my-sub");
fail("should have failed");
} catch (PulsarAdminException.PreconditionFailedException e) {
// ok
} catch (Exception e) {
fail(e.getMessage());
}
List<String> subscriptions = admin.topics().getSubscriptions(partitionedTopicName);
assertEquals(subscriptions.size(), 1);
Consumer<byte[]> consumer1 = client.newConsumer().topic(partitionedTopicName).subscriptionName("my-sub-1")
.subscribe();
assertEquals(new HashSet<>(admin.topics().getSubscriptions(partitionedTopicName)),
Set.of("my-sub", "my-sub-1"));
consumer1.close();
admin.topics().deleteSubscription(partitionedTopicName, "my-sub-1");
assertEquals(admin.topics().getSubscriptions(partitionedTopicName), List.of("my-sub"));
Producer<byte[]> producer = client.newProducer(Schema.BYTES)
.topic(partitionedTopicName)
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
for (int i = 0; i < 10; i++) {
String message = "message-" + i;
producer.send(message.getBytes());
}
assertEquals(new HashSet<>(admin.topics().getList("prop-xyz/use/ns1")),
Set.of(partitionedTopicName + "-partition-0", partitionedTopicName + "-partition-1",
partitionedTopicName + "-partition-2", partitionedTopicName + "-partition-3"));
// test cumulative stats for partitioned topic
PartitionedTopicStats topicStats = admin.topics().getPartitionedStats(partitionedTopicName, false);
assertEquals(topicStats.getSubscriptions().keySet(), new TreeSet<>(List.of("my-sub")));
assertEquals(topicStats.getSubscriptions().get("my-sub").getConsumers().size(), 1);
assertEquals(topicStats.getSubscriptions().get("my-sub").getMsgBacklog(), 10);
assertEquals(topicStats.getPublishers().size(), 1);
assertEquals(topicStats.getPartitions(), new HashMap<>());
// test per partition stats for partitioned topic
topicStats = admin.topics().getPartitionedStats(partitionedTopicName, true);
assertEquals(topicStats.getMetadata().partitions, 4);
assertEquals(topicStats.getPartitions().keySet(),
Set.of(partitionedTopicName + "-partition-0", partitionedTopicName + "-partition-1",
partitionedTopicName + "-partition-2", partitionedTopicName + "-partition-3"));
TopicStats partitionStats = topicStats.getPartitions().get(partitionedTopicName + "-partition-0");
assertEquals(partitionStats.getPublishers().size(), 1);
assertEquals(partitionStats.getSubscriptions().get("my-sub").getConsumers().size(), 1);
assertEquals(partitionStats.getSubscriptions().get("my-sub").getMsgBacklog(), 3, 1);
try {
admin.topics().skipMessages(partitionedTopicName, "my-sub", 5);
fail("skip messages for partitioned topics should fail");
} catch (Exception e) {
// ok
}
admin.topics().skipAllMessages(partitionedTopicName, "my-sub");
topicStats = admin.topics().getPartitionedStats(partitionedTopicName, false);
assertEquals(topicStats.getSubscriptions().get("my-sub").getMsgBacklog(), 0);
producer.close();
consumer.close();
admin.topics().deleteSubscription(partitionedTopicName, "my-sub");
assertEquals(admin.topics().getSubscriptions(partitionedTopicName), new ArrayList<>());
try {
admin.topics().createPartitionedTopic(partitionedTopicName, 32);
fail("Should have failed as the partitioned topic exists with its partition created");
} catch (ConflictException ignore) {
}
producer = client.newProducer(Schema.BYTES)
.topic(partitionedTopicName)
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
topics = admin.topics().getList("prop-xyz/use/ns1");
assertEquals(topics.size(), 4);
try {
admin.topics().deletePartitionedTopic(partitionedTopicName);
fail("The topic is busy");
} catch (PreconditionFailedException pfe) {
// ok
}
producer.close();
client.close();
admin.topics().deletePartitionedTopic(partitionedTopicName);
try {
admin.topics().getPartitionedTopicMetadata(partitionedTopicName);
fail("getPartitionedTopicMetadata of " + partitionedTopicName + " should not succeed");
} catch (NotFoundException expected) {
}
admin.topics().createPartitionedTopic(partitionedTopicName, 32);
assertEquals(admin.topics().getPartitionedTopicMetadata(partitionedTopicName).partitions, 32);
try {
admin.topics().deletePartitionedTopic("persistent://prop-xyz/use/ns1/ds2");
fail("Should have failed as the partitioned topic was not created");
} catch (NotFoundException nfe) {
}
admin.topics().deletePartitionedTopic(partitionedTopicName);
// delete a partitioned topic in a global namespace
admin.topics().createPartitionedTopic(partitionedTopicName, 4);
admin.topics().deletePartitionedTopic(partitionedTopicName);
}
@Test(dataProvider = "numBundles")
public void testDeleteNamespaceBundle(Integer numBundles) throws Exception {
admin.namespaces().deleteNamespace("prop-xyz/use/ns1");
admin.namespaces().createNamespace("prop-xyz/use/ns1-bundles", numBundles);
// since we have 2 brokers running, we try to let both of them acquire bundle ownership
admin.lookups().lookupTopic("persistent://prop-xyz/use/ns1-bundles/ds1");
admin.lookups().lookupTopic("persistent://prop-xyz/use/ns1-bundles/ds2");
admin.lookups().lookupTopic("persistent://prop-xyz/use/ns1-bundles/ds3");
admin.lookups().lookupTopic("persistent://prop-xyz/use/ns1-bundles/ds4");
assertEquals(admin.namespaces().getTopics("prop-xyz/use/ns1-bundles"), new ArrayList<>());
admin.namespaces().deleteNamespace("prop-xyz/use/ns1-bundles");
assertEquals(admin.namespaces().getNamespaces("prop-xyz", "use"), new ArrayList<>());
}
@Test
public void testNamespaceSplitBundle() throws Exception {
// Force to create a topic
final String namespace = "prop-xyz/use/ns1";
final String topicName = "persistent://" + namespace + "/ds2";
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic(topicName)
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
producer.send("message".getBytes());
publishMessagesOnPersistentTopic(topicName, 0);
assertEquals(admin.topics().getList(namespace), List.of(topicName));
try {
admin.namespaces().splitNamespaceBundle(namespace, "0x00000000_0xffffffff", true, null);
} catch (Exception e) {
fail("split bundle shouldn't have thrown exception");
}
// bundle-factory cache must have updated split bundles
NamespaceBundles bundles = bundleFactory.getBundles(NamespaceName.get(namespace));
String[] splitRange = { namespace + "/0x00000000_0x7fffffff", namespace + "/0x7fffffff_0xffffffff" };
for (int i = 0; i < bundles.getBundles().size(); i++) {
assertEquals(bundles.getBundles().get(i).toString(), splitRange[i]);
}
producer.close();
}
@Test
public void testNamespaceSplitBundleConcurrent() throws Exception {
// Force to create a topic
final String namespace = "prop-xyz/use/ns1";
final String topicName = "persistent://" + namespace + "/ds2";
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic(topicName)
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
producer.send("message".getBytes());
publishMessagesOnPersistentTopic(topicName, 0);
assertEquals(admin.topics().getList(namespace), List.of(topicName));
try {
admin.namespaces().splitNamespaceBundle(namespace, "0x00000000_0xffffffff", false, null);
} catch (Exception e) {
fail("split bundle shouldn't have thrown exception");
}
// bundle-factory cache must have updated split bundles
NamespaceBundles bundles = bundleFactory.getBundles(NamespaceName.get(namespace));
String[] splitRange = {namespace + "/0x00000000_0x7fffffff", namespace + "/0x7fffffff_0xffffffff"};
for (int i = 0; i < bundles.getBundles().size(); i++) {
assertEquals(bundles.getBundles().get(i).toString(), splitRange[i]);
}
@Cleanup("shutdownNow")
ExecutorService executorService = Executors.newCachedThreadPool();
try {
executorService.invokeAll(
Arrays.asList(
() ->
{
log.info("split 2 bundles at the same time. spilt: 0x00000000_0x7fffffff ");
admin.namespaces().splitNamespaceBundle(namespace, "0x00000000_0x7fffffff", false, null);
return null;
},
() ->
{
log.info("split 2 bundles at the same time. spilt: 0x7fffffff_0xffffffff ");
admin.namespaces().splitNamespaceBundle(namespace, "0x7fffffff_0xffffffff", false, null);
return null;
}
)
);
} catch (Exception e) {
fail("split bundle shouldn't have thrown exception");
}
String[] splitRange4 = {
namespace + "/0x00000000_0x3fffffff",
namespace + "/0x3fffffff_0x7fffffff",
namespace + "/0x7fffffff_0xbfffffff",
namespace + "/0xbfffffff_0xffffffff"};
bundles = bundleFactory.getBundles(NamespaceName.get(namespace));
assertEquals(bundles.getBundles().size(), 4);
for (int i = 0; i < bundles.getBundles().size(); i++) {
assertEquals(bundles.getBundles().get(i).toString(), splitRange4[i]);
}
try {
executorService.invokeAll(
Arrays.asList(
() ->
{
log.info("split 4 bundles at the same time. spilt: 0x00000000_0x3fffffff ");
admin.namespaces().splitNamespaceBundle(namespace, "0x00000000_0x3fffffff", false, null);
return null;
},
() ->
{
log.info("split 4 bundles at the same time. spilt: 0x3fffffff_0x7fffffff ");
admin.namespaces().splitNamespaceBundle(namespace, "0x3fffffff_0x7fffffff", false, null);
return null;
},
() ->
{
log.info("split 4 bundles at the same time. spilt: 0x7fffffff_0xbfffffff ");
admin.namespaces().splitNamespaceBundle(namespace, "0x7fffffff_0xbfffffff", false, null);
return null;
},
() ->
{
log.info("split 4 bundles at the same time. spilt: 0xbfffffff_0xffffffff ");
admin.namespaces().splitNamespaceBundle(namespace, "0xbfffffff_0xffffffff", false, null);
return null;
}
)
);
} catch (Exception e) {
fail("split bundle shouldn't have thrown exception");
}
String[] splitRange8 = {
namespace + "/0x00000000_0x1fffffff",
namespace + "/0x1fffffff_0x3fffffff",
namespace + "/0x3fffffff_0x5fffffff",
namespace + "/0x5fffffff_0x7fffffff",
namespace + "/0x7fffffff_0x9fffffff",
namespace + "/0x9fffffff_0xbfffffff",
namespace + "/0xbfffffff_0xdfffffff",
namespace + "/0xdfffffff_0xffffffff"};
bundles = bundleFactory.getBundles(NamespaceName.get(namespace));
assertEquals(bundles.getBundles().size(), 8);
for (int i = 0; i < bundles.getBundles().size(); i++) {
assertEquals(bundles.getBundles().get(i).toString(), splitRange8[i]);
}
producer.close();
}
@Test
public void testNamespaceUnloadBundle() throws Exception {
assertEquals(admin.topics().getList("prop-xyz/use/ns1"), new ArrayList<>());
// Force to create a topic
publishMessagesOnPersistentTopic("persistent://prop-xyz/use/ns1/ds2", 0);
assertEquals(admin.topics().getList("prop-xyz/use/ns1"),
List.of("persistent://prop-xyz/use/ns1/ds2"));
// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1/ds2")
.subscriptionName("my-sub").subscribe();
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/use/ns1/ds2"),
List.of("my-sub"));
// Create producer
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic("persistent://prop-xyz/use/ns1/ds2")
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
for (int i = 0; i < 10; i++) {
String message = "message-" + i;
producer.send(message.getBytes());
}
consumer.close();
producer.close();
try {
admin.namespaces().unloadNamespaceBundle("prop-xyz/use/ns1", "0x00000000_0xffffffff");
} catch (Exception e) {
fail("Unload shouldn't have throw exception");
}
// check that no one owns the namespace
NamespaceBundle bundle = bundleFactory.getBundle(NamespaceName.get("prop-xyz/use/ns1"),
Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED));
assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle));
assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle));
pulsarClient.shutdown();
LOG.info("--- RELOAD ---");
// Force reload of namespace and wait for topic to be ready
for (int i = 0; i < 30; i++) {
try {
admin.topics().getStats("persistent://prop-xyz/use/ns1/ds2");
break;
} catch (PulsarAdminException e) {
LOG.warn("Failed to get topic stats.. {}", e.getMessage());
Thread.sleep(1000);
}
}
admin.topics().deleteSubscription("persistent://prop-xyz/use/ns1/ds2", "my-sub");
admin.topics().delete("persistent://prop-xyz/use/ns1/ds2");
}
@Test(dataProvider = "numBundles")
public void testNamespaceBundleUnload(Integer numBundles) throws Exception {
admin.namespaces().createNamespace("prop-xyz/use/ns1-bundles", numBundles);
assertEquals(admin.topics().getList("prop-xyz/use/ns1-bundles"), new ArrayList<>());
// Force to create a topic
publishMessagesOnPersistentTopic("persistent://prop-xyz/use/ns1-bundles/ds2", 0);
assertEquals(admin.topics().getList("prop-xyz/use/ns1-bundles"),
List.of("persistent://prop-xyz/use/ns1-bundles/ds2"));
// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds2")
.subscriptionName("my-sub").subscribe();
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/use/ns1-bundles/ds2"),
List.of("my-sub"));
// Create producer
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic("persistent://prop-xyz/use/ns1-bundles/ds2")
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
for (int i = 0; i < 10; i++) {
String message = "message-" + i;
producer.send(message.getBytes());
}
NamespaceBundle bundle = pulsar.getNamespaceService()
.getBundle(TopicName.get("persistent://prop-xyz/use/ns1-bundles/ds2"));
consumer.close();
producer.close();
admin.namespaces().unloadNamespaceBundle("prop-xyz/use/ns1-bundles", bundle.getBundleRange());
// check that no one owns the namespace bundle
assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle));
assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle));
LOG.info("--- RELOAD ---");
// Force reload of namespace and wait for topic to be ready
for (int i = 0; i < 30; i++) {
try {
admin.topics().getStats("persistent://prop-xyz/use/ns1-bundles/ds2");
break;
} catch (PulsarAdminException e) {
LOG.warn("Failed to get topic stats.. {}", e.getMessage());
Thread.sleep(1000);
}
}
admin.topics().deleteSubscription("persistent://prop-xyz/use/ns1-bundles/ds2", "my-sub");
admin.topics().delete("persistent://prop-xyz/use/ns1-bundles/ds2");
}
@Test(dataProvider = "bundling")
public void testClearBacklogOnNamespace(Integer numBundles) throws Exception {
admin.namespaces().createNamespace("prop-xyz/use/ns1-bundles", numBundles);
// create consumer and subscription
@Cleanup
Consumer<byte[]> subscribe =
pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds2").subscriptionName("my-sub")
.subscribe();
@Cleanup
Consumer<byte[]> subscribe1 = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds2")
.subscriptionName("my-sub-1")
.subscribe();
@Cleanup
Consumer<byte[]> subscribe2 = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds2")
.subscriptionName("my-sub-2")
.subscribe();
@Cleanup
Consumer<byte[]> subscribe3 =
pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds1").subscriptionName("my-sub")
.subscribe();
@Cleanup
Consumer<byte[]> subscribe4 = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds1")
.subscriptionName("my-sub-1")
.subscribe();
// Create producer
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic("persistent://prop-xyz/use/ns1-bundles/ds2")
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
for (int i = 0; i < 10; i++) {
String message = "message-" + i;
producer.send(message.getBytes());
}
producer.close();
// Create producer
Producer<byte[]> producer1 = pulsarClient.newProducer(Schema.BYTES)
.topic("persistent://prop-xyz/use/ns1-bundles/ds1")
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
for (int i = 0; i < 10; i++) {
String message = "message-" + i;
producer1.send(message.getBytes());
}
producer1.close();
admin.namespaces().clearNamespaceBacklogForSubscription("prop-xyz/use/ns1-bundles", "my-sub");
long backlog = admin.topics().getStats("persistent://prop-xyz/use/ns1-bundles/ds2").getSubscriptions()
.get("my-sub").getMsgBacklog();
assertEquals(backlog, 0);
backlog = admin.topics().getStats("persistent://prop-xyz/use/ns1-bundles/ds1").getSubscriptions()
.get("my-sub").getMsgBacklog();
assertEquals(backlog, 0);
backlog = admin.topics().getStats("persistent://prop-xyz/use/ns1-bundles/ds1").getSubscriptions()
.get("my-sub-1").getMsgBacklog();
assertEquals(backlog, 10);
admin.namespaces().clearNamespaceBacklog("prop-xyz/use/ns1-bundles");
backlog = admin.topics().getStats("persistent://prop-xyz/use/ns1-bundles/ds1").getSubscriptions()
.get("my-sub-1").getMsgBacklog();
assertEquals(backlog, 0);
backlog = admin.topics().getStats("persistent://prop-xyz/use/ns1-bundles/ds2").getSubscriptions()
.get("my-sub-1").getMsgBacklog();
assertEquals(backlog, 0);
backlog = admin.topics().getStats("persistent://prop-xyz/use/ns1-bundles/ds2").getSubscriptions()
.get("my-sub-2").getMsgBacklog();
assertEquals(backlog, 0);
}
@Test(dataProvider = "bundling")
public void testUnsubscribeOnNamespace(Integer numBundles) throws Exception {
admin.namespaces().createNamespace("prop-xyz/use/ns1-bundles", numBundles);
// create consumer and subscription
Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds2")
.subscriptionName("my-sub").subscribe();
Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds2")
.subscriptionName("my-sub-1").subscribe();
/* Consumer consumer3 = */ pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds2")
.subscriptionName("my-sub-2").subscribe().close();
Consumer<byte[]> consumer4 = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds1")
.subscriptionName("my-sub").subscribe();
Consumer<byte[]> consumer5 = pulsarClient.newConsumer().topic("persistent://prop-xyz/use/ns1-bundles/ds1")
.subscriptionName("my-sub-1").subscribe();
try {
admin.namespaces().unsubscribeNamespace("prop-xyz/use/ns1-bundles", "my-sub");
fail("should have failed");
} catch (PulsarAdminException.PreconditionFailedException e) {
// ok
}
consumer1.close();
try {
admin.namespaces().unsubscribeNamespace("prop-xyz/use/ns1-bundles", "my-sub");
fail("should have failed");
} catch (PulsarAdminException.PreconditionFailedException e) {
// ok
}
consumer4.close();
admin.namespaces().unsubscribeNamespace("prop-xyz/use/ns1-bundles", "my-sub");
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/use/ns1-bundles/ds2"),
List.of("my-sub-1", "my-sub-2"));
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/use/ns1-bundles/ds1"),
List.of("my-sub-1"));
consumer2.close();
consumer5.close();
admin.namespaces().unsubscribeNamespace("prop-xyz/use/ns1-bundles", "my-sub-1");
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/use/ns1-bundles/ds2"),
List.of("my-sub-2"));
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/use/ns1-bundles/ds1"),
new ArrayList<>());
}
private void publishMessagesOnPersistentTopic(String topicName, int messages) throws Exception {
publishMessagesOnPersistentTopic(topicName, messages, 0);
}
private void publishMessagesOnPersistentTopic(String topicName, int messages, int startIdx) throws Exception {
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic(topicName)
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
for (int i = startIdx; i < (messages + startIdx); i++) {
String message = "message-" + i;
producer.send(message.getBytes());
}
producer.close();
}
@Test
public void backlogQuotas() throws Exception {
assertEquals(admin.namespaces().getBacklogQuotaMap("prop-xyz/use/ns1"),
new HashMap<>());
Map<BacklogQuotaType, BacklogQuota> quotaMap = admin.namespaces().getBacklogQuotaMap("prop-xyz/use/ns1");
assertEquals(quotaMap.size(), 0);
assertNull(quotaMap.get(BacklogQuotaType.destination_storage));
admin.namespaces().setBacklogQuota("prop-xyz/use/ns1",
BacklogQuota.builder()
.limitSize(1 * 1024 * 1024)
.retentionPolicy(RetentionPolicy.producer_exception)
.build());
quotaMap = admin.namespaces().getBacklogQuotaMap("prop-xyz/use/ns1");
assertEquals(quotaMap.size(), 1);
assertEquals(quotaMap.get(BacklogQuotaType.destination_storage), BacklogQuota.builder()
.limitSize(1 * 1024 * 1024)
.retentionPolicy(RetentionPolicy.producer_exception)
.build());
admin.namespaces().removeBacklogQuota("prop-xyz/use/ns1");
quotaMap = admin.namespaces().getBacklogQuotaMap("prop-xyz/use/ns1");
assertEquals(quotaMap.size(), 0);
assertNull(quotaMap.get(BacklogQuotaType.destination_storage));
}
@Test
public void statsOnNonExistingTopics() throws Exception {
try {
admin.topics().getStats("persistent://prop-xyz/use/ns1/ghostTopic");
fail("The topic doesn't exist");
} catch (NotFoundException e) {
// OK
}
}
@Test
public void testDeleteFailedReturnCode() throws Exception {
String topicName = "persistent://prop-xyz/use/ns1/my-topic";
Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
.topic(topicName)
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.SinglePartition)
.create();
try {
admin.topics().delete(topicName);
fail("The topic is busy");
} catch (PreconditionFailedException e) {
// OK
}
producer.close();
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName("sub").subscribe();
try {
admin.topics().delete(topicName);
fail("The topic is busy");
} catch (PreconditionFailedException e) {
// OK
}
try {
admin.topics().deleteSubscription(topicName, "sub");
fail("The topic is busy");
} catch (PreconditionFailedException e) {
// Ok
}
consumer.close();
// Now should succeed
admin.topics().delete(topicName);
}
private static class IncompatiblePropertyAdmin {
public Set<String> allowedClusters;
public int someNewIntField;
public String someNewString;
}
@Test
public void testJacksonWithTypeDifferencies() throws Exception {
String expectedJson = "{\"adminRoles\":[\"role1\",\"role2\"],\"allowedClusters\":[\"usw\",\"use\"]}";
IncompatiblePropertyAdmin r1 = ObjectMapperFactory.getMapper().reader().forType(IncompatiblePropertyAdmin.class)
.readValue(expectedJson);
assertEquals(r1.allowedClusters, Set.of("use", "usw"));
assertEquals(r1.someNewIntField, 0);
assertNull(r1.someNewString);
}
@Test
public void testBackwardCompatiblity() throws Exception {
Set<String> allowedClusters = Set.of("use");
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), allowedClusters);
admin.tenants().createTenant("prop-xyz2", tenantInfo);
admin.namespaces().createNamespace("prop-xyz2/use/ns1");
assertEquals(admin.tenants().getTenants(), List.of("prop-xyz", "prop-xyz2"));
assertEquals(admin.tenants().getTenantInfo("prop-xyz2").getAdminRoles(),
List.of("role1", "role2"));
assertEquals(admin.tenants().getTenantInfo("prop-xyz2").getAllowedClusters(), Set.of("use"));
// Try to deserialize property JSON with IncompatiblePropertyAdmin format
// it should succeed ignoring missing fields
TenantsImpl properties = (TenantsImpl) admin.tenants();
IncompatiblePropertyAdmin result = properties.request(properties.getWebTarget().path("prop-xyz2"))
.get(IncompatiblePropertyAdmin.class);
assertEquals(result.allowedClusters, Set.of("use"));
assertEquals(result.someNewIntField, 0);
assertNull(result.someNewString);
admin.namespaces().deleteNamespace("prop-xyz2/use/ns1");
admin.tenants().deleteTenant("prop-xyz2");
assertEquals(admin.tenants().getTenants(), Set.of("prop-xyz"));
}
@Test(dataProvider = "topicName")
public void persistentTopicsCursorReset(String topicName) throws Exception {
admin.namespaces().setRetention("prop-xyz/use/ns1", new RetentionPolicies(10, 10));
assertEquals(admin.topics().getList("prop-xyz/use/ns1"), new ArrayList<>());
topicName = "persistent://prop-xyz/use/ns1/" + topicName;
// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
.subscriptionName("my-sub").startMessageIdInclusive()
.subscriptionType(SubscriptionType.Exclusive)
.acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
assertEquals(admin.topics().getSubscriptions(topicName), List.of("my-sub"));
publishMessagesOnPersistentTopic(topicName, 5, 0);
// Allow at least 1ms for messages to have different timestamps
Thread.sleep(1);
long messageTimestamp = System.currentTimeMillis();
publishMessagesOnPersistentTopic(topicName, 5, 5);
List<Message<byte[]>> messages = admin.topics().peekMessages(topicName, "my-sub", 10);
assertEquals(messages.size(), 10);
for (int i = 0; i < 10; i++) {
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message);
}
// messages should still be available due to retention
admin.topics().resetCursor(topicName, "my-sub", messageTimestamp);
int receivedAfterReset = 0;
for (int i = 5; i < 10; i++) {
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message);
++receivedAfterReset;
String expected = "message-" + i;
assertEquals(message.getData(), expected.getBytes());
}
assertEquals(receivedAfterReset, 5);
consumer.close();
admin.topics().deleteSubscription(topicName, "my-sub");
assertEquals(admin.topics().getSubscriptions(topicName), new ArrayList<>());
admin.topics().delete(topicName);
}
@Test(dataProvider = "topicName")
public void persistentTopicsCursorResetAfterReset(String topicName) throws Exception {
admin.namespaces().setRetention("prop-xyz/use/ns1", new RetentionPolicies(10, 10));
assertEquals(admin.topics().getList("prop-xyz/use/ns1"), new ArrayList<>());
topicName = "persistent://prop-xyz/use/ns1/" + topicName;
// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
.subscriptionName("my-sub").startMessageIdInclusive()
.subscriptionType(SubscriptionType.Exclusive)
.acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
assertEquals(admin.topics().getSubscriptions(topicName), List.of("my-sub"));
publishMessagesOnPersistentTopic(topicName, 5, 0);
// Allow at least 1ms for messages to have different timestamps
Thread.sleep(1);
long firstTimestamp = System.currentTimeMillis();
publishMessagesOnPersistentTopic(topicName, 3, 5);
Thread.sleep(1);
long secondTimestamp = System.currentTimeMillis();
publishMessagesOnPersistentTopic(topicName, 2, 8);
List<Message<byte[]>> messages = admin.topics().peekMessages(topicName, "my-sub", 10);
assertEquals(messages.size(), 10);
messages.forEach(message -> {
LOG.info("Peeked message: {}", new String(message.getData()));
});
for (int i = 0; i < 10; i++) {
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message);
}
admin.topics().resetCursor(topicName, "my-sub", firstTimestamp);
int receivedAfterReset = 0;
// Should received messages from 5-9
for (int i = 5; i < 10; i++) {
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message);
++receivedAfterReset;
String expected = "message-" + i;
assertEquals(new String(message.getData()), expected);
}
assertEquals(receivedAfterReset, 5);
// Reset at 2nd timestamp
receivedAfterReset = 0;
admin.topics().resetCursor(topicName, "my-sub", secondTimestamp);
// Should received messages from 8-9
for (int i = 8; i < 10; i++) {
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message);
++receivedAfterReset;
String expected = "message-" + i;
assertEquals(new String(message.getData()), expected);
}
assertEquals(receivedAfterReset, 2);
consumer.close();
admin.topics().deleteSubscription(topicName, "my-sub");
assertEquals(admin.topics().getSubscriptions(topicName), new ArrayList<>());
admin.topics().delete(topicName);
}
@Test(dataProvider = "topicName")
public void partitionedTopicsCursorReset(String topicName) throws Exception {
admin.namespaces().setRetention("prop-xyz/use/ns1", new RetentionPolicies(10, 10));
topicName = "persistent://prop-xyz/use/ns1/" + topicName;
admin.topics().createPartitionedTopic(topicName, 4);
// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
.subscriptionName("my-sub").startMessageIdInclusive()
.subscriptionType(SubscriptionType.Exclusive)
.acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
List<String> topics = admin.topics().getList("prop-xyz/use/ns1");
assertEquals(topics.size(), 4);
assertEquals(admin.topics().getSubscriptions(topicName), List.of("my-sub"));
publishMessagesOnPersistentTopic(topicName, 5, 0);
Thread.sleep(1);
long timestamp = System.currentTimeMillis();
publishMessagesOnPersistentTopic(topicName, 5, 5);
for (int i = 0; i < 10; i++) {
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message);
}
// messages should still be available due to retention
admin.topics().resetCursor(topicName, "my-sub", timestamp);
Set<String> expectedMessages = new HashSet<>();
Set<String> receivedMessages = new HashSet<>();
for (int i = 5; i < 10; i++) {
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message);
expectedMessages.add("message-" + i);
receivedMessages.add(new String(message.getData()));
}
receivedMessages.removeAll(expectedMessages);
assertEquals(receivedMessages.size(), 0);
consumer.close();
admin.topics().deleteSubscription(topicName, "my-sub");
admin.topics().deletePartitionedTopic(topicName);
}
@Test
public void persistentTopicsInvalidCursorReset() throws Exception {
admin.namespaces().setRetention("prop-xyz/use/ns1", new RetentionPolicies(10, 10));
assertEquals(admin.topics().getList("prop-xyz/use/ns1"), new ArrayList<>());
String topicName = "persistent://prop-xyz/use/ns1/invalidcursorreset";
// Force to create a topic
publishMessagesOnPersistentTopic(topicName, 0);
assertEquals(admin.topics().getList("prop-xyz/use/ns1"), List.of(topicName));
// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
.build();
Consumer<byte[]> consumer = client.newConsumer().topic(topicName).subscriptionName("my-sub")
.subscriptionType(SubscriptionType.Exclusive).subscribe();
assertEquals(admin.topics().getSubscriptions(topicName), List.of("my-sub"));
publishMessagesOnPersistentTopic(topicName, 10);
List<Message<byte[]>> messages = admin.topics().peekMessages(topicName, "my-sub", 10);
assertEquals(messages.size(), 10);
for (int i = 0; i < 10; i++) {
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message);
}
// use invalid timestamp
try {
admin.topics().resetCursor(topicName, "my-sub", System.currentTimeMillis() - 190000);
} catch (Exception e) {
// fail the test
throw e;
}
admin.topics().resetCursor(topicName, "my-sub", System.currentTimeMillis() + 90000);
consumer = client.newConsumer().topic(topicName).subscriptionName("my-sub").subscribe();
consumer.close();
client.close();
admin.topics().deleteSubscription(topicName, "my-sub");
assertEquals(admin.topics().getSubscriptions(topicName), new ArrayList<>());
admin.topics().delete(topicName);
}
@Value
@Builder
static class CustomTenantAdmin implements TenantInfo {
private final int newTenant;
private final Set<String> adminRoles;
private final Set<String> allowedClusters;
}
@Test
public void testObjectWithUnknownProperties() {
TenantInfo pa = TenantInfo.builder()
.adminRoles(Set.of("test_appid1", "test_appid2"))
.allowedClusters(Set.of("use"))
.build();
CustomTenantAdmin cpa = CustomTenantAdmin.builder()
.adminRoles(pa.getAdminRoles())
.allowedClusters(pa.getAllowedClusters())
.newTenant(100)
.build();
try {
admin.tenants().createTenant("test-property", cpa);
} catch (Exception e) {
fail("Should not happen : ", e);
}
}
/**
* <pre>
* Verify: PersistentTopicsBase.expireMessages()/expireMessagesForAllSubscriptions()
* 1. Created multiple shared subscriptions and publisher on topic
* 2. Publish messages on the topic
* 3. expire message on sub-1 : backlog for sub-1 must be 0
* 4. expire message on all subscriptions: backlog for all subscription must be 0
* </pre>
*
* @throws Exception
*/
@Test
public void testPersistentTopicsExpireMessages() throws Exception {
cleanup();
setup();
// Force to create a topic
publishMessagesOnPersistentTopic("persistent://prop-xyz/use/ns1/ds2", 0);
assertEquals(admin.topics().getList("prop-xyz/use/ns1"),
List.of("persistent://prop-xyz/use/ns1/ds2"));
// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
.build();
ConsumerBuilder<byte[]> consumerBuilder = client.newConsumer().topic("persistent://prop-xyz/use/ns1/ds2")
.subscriptionType(SubscriptionType.Shared);
Consumer<byte[]> consumer1 = consumerBuilder.clone().subscriptionName("my-sub1").subscribe();
Consumer<byte[]> consumer2 = consumerBuilder.clone().subscriptionName("my-sub2").subscribe();
Consumer<byte[]> consumer3 = consumerBuilder.clone().subscriptionName("my-sub3").subscribe();
assertEquals(admin.topics().getSubscriptions("persistent://prop-xyz/use/ns1/ds2").size(), 3);
publishMessagesOnPersistentTopic("persistent://prop-xyz/use/ns1/ds2", 10);
TopicStats topicStats = admin.topics().getStats("persistent://prop-xyz/use/ns1/ds2");
assertEquals(topicStats.getSubscriptions().get("my-sub1").getMsgBacklog(), 10);
assertEquals(topicStats.getSubscriptions().get("my-sub2").getMsgBacklog(), 10);
assertEquals(topicStats.getSubscriptions().get("my-sub3").getMsgBacklog(), 10);
Thread.sleep(1000); // wait for 1 seconds to expire message
admin.topics().expireMessages("persistent://prop-xyz/use/ns1/ds2", "my-sub1", 1);
Thread.sleep(1000); // wait for 1 seconds to execute expire message as it is async
topicStats = admin.topics().getStats("persistent://prop-xyz/use/ns1/ds2");
assertEquals(topicStats.getSubscriptions().get("my-sub1").getMsgBacklog(), 0);
assertEquals(topicStats.getSubscriptions().get("my-sub2").getMsgBacklog(), 10);
assertEquals(topicStats.getSubscriptions().get("my-sub3").getMsgBacklog(), 10);
try {
admin.topics().expireMessagesForAllSubscriptions("persistent://prop-xyz/use/ns1/ds2", 1);
} catch (Exception e) {
// my-sub1 has no msg backlog, so expire message won't be issued on that subscription
assertTrue(e.getMessage().startsWith("Expire message by timestamp not issued on topic"));
}
Thread.sleep(1000); // wait for 1 seconds to execute expire message as it is async
topicStats = admin.topics().getStats("persistent://prop-xyz/use/ns1/ds2");
assertEquals(topicStats.getSubscriptions().get("my-sub1").getMsgBacklog(), 0);
assertEquals(topicStats.getSubscriptions().get("my-sub2").getMsgBacklog(), 0);
assertEquals(topicStats.getSubscriptions().get("my-sub3").getMsgBacklog(), 0);
consumer1.close();
consumer2.close();
consumer3.close();
}
/**
* Verify: PersistentTopicsBase.expireMessages()/expireMessagesForAllSubscriptions() for PartitionTopic
*
* @throws Exception
*/
@Test
public void testPersistentTopicExpireMessageOnPartitionTopic() throws Exception {
admin.topics().createPartitionedTopic("persistent://prop-xyz/use/ns1/ds1", 4);
// create consumer and subscription
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsar.getWebServiceAddress())
.statsInterval(0, TimeUnit.SECONDS)
.build();
Consumer<byte[]> consumer = client.newConsumer().topic("persistent://prop-xyz/use/ns1/ds1")
.subscriptionName("my-sub").subscribe();
Producer<byte[]> producer = client.newProducer(Schema.BYTES)
.topic("persistent://prop-xyz/use/ns1/ds1")
.enableBatching(false)
.messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
.create();
for (int i = 0; i < 10; i++) {
String message = "message-" + i;
producer.send(message.getBytes());
}
PartitionedTopicStats topicStats = admin.topics()
.getPartitionedStats("persistent://prop-xyz/use/ns1/ds1", true);
assertEquals(topicStats.getSubscriptions().get("my-sub").getMsgBacklog(), 10);
TopicStats partitionStatsPartition0 = topicStats.getPartitions()
.get("persistent://prop-xyz/use/ns1/ds1-partition-0");
TopicStats partitionStatsPartition1 = topicStats.getPartitions()
.get("persistent://prop-xyz/use/ns1/ds1-partition-1");
assertEquals(partitionStatsPartition0.getSubscriptions().get("my-sub").getMsgBacklog(), 3, 1);
assertEquals(partitionStatsPartition1.getSubscriptions().get("my-sub").getMsgBacklog(), 3, 1);
Thread.sleep(1000);
admin.topics().expireMessagesForAllSubscriptions("persistent://prop-xyz/use/ns1/ds1", 1);
Thread.sleep(1000);
topicStats = admin.topics().getPartitionedStats("persistent://prop-xyz/use/ns1/ds1", true);
partitionStatsPartition0 = topicStats.getPartitions().get("persistent://prop-xyz/use/ns1/ds1-partition-0");
partitionStatsPartition1 = topicStats.getPartitions().get("persistent://prop-xyz/use/ns1/ds1-partition-1");
assertEquals(partitionStatsPartition0.getSubscriptions().get("my-sub").getMsgBacklog(), 0);
assertEquals(partitionStatsPartition1.getSubscriptions().get("my-sub").getMsgBacklog(), 0);
producer.close();
consumer.close();
}
/**
* This test-case verifies that broker should support both url/uri encoding for topic-name. It calls below api with
* url-encoded and also uri-encoded topic-name in http request: a. PartitionedMetadataLookup b. TopicLookupBase c. Topic
* Stats
*
* @param topicName
* @throws Exception
*/
@Test(dataProvider = "topicName")
public void testPulsarAdminForUriAndUrlEncoding(String topicName) throws Exception {
final String ns1 = "prop-xyz/use/ns1";
final String topic1 = "persistent://" + ns1 + "/" + topicName;
final String urlEncodedTopic = Codec.encode(topicName);
final String uriEncodedTopic = urlEncodedTopic.replaceAll("\\+", "%20");
final int numOfPartitions = 4;
admin.topics().createPartitionedTopic(topic1, numOfPartitions);
// Create a consumer to get stats on this topic
pulsarClient.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe().close();
TopicsImpl persistent = (TopicsImpl) admin.topics();
Field field = TopicsImpl.class.getDeclaredField("adminTopics");
field.setAccessible(true);
WebTarget persistentTopics = ((WebTarget) field.get(persistent)).path("persistent");
// (1) Get PartitionedMetadata : with Url and Uri encoding
final CompletableFuture<PartitionedTopicMetadata> urlEncodedPartitionedMetadata = new CompletableFuture<>();
// (a) Url encoding
persistent.asyncGetRequest(persistentTopics.path(ns1).path(urlEncodedTopic).path("partitions"),
new InvocationCallback<PartitionedTopicMetadata>() {
@Override
public void completed(PartitionedTopicMetadata response) {
urlEncodedPartitionedMetadata.complete(response);
}
@Override
public void failed(Throwable e) {
urlEncodedPartitionedMetadata.completeExceptionally(e);
}
});
final CompletableFuture<PartitionedTopicMetadata> uriEncodedPartitionedMetadata = new CompletableFuture<>();
// (b) Uri encoding
persistent.asyncGetRequest(persistentTopics.path(ns1).path(uriEncodedTopic).path("partitions"),
new InvocationCallback<PartitionedTopicMetadata>() {
@Override
public void completed(PartitionedTopicMetadata response) {
uriEncodedPartitionedMetadata.complete(response);
}
@Override
public void failed(Throwable e) {
uriEncodedPartitionedMetadata.completeExceptionally(e);
}
});
assertEquals(urlEncodedPartitionedMetadata.get().partitions, numOfPartitions);
assertEquals(urlEncodedPartitionedMetadata.get().partitions, (uriEncodedPartitionedMetadata.get().partitions));
// (2) Get Topic Lookup
LookupImpl lookup = (LookupImpl) admin.lookups();
Field field2 = LookupImpl.class.getDeclaredField("v2lookup");
field2.setAccessible(true);
WebTarget target2 = (WebTarget) field2.get(lookup);
// (a) Url encoding
LookupData urlEncodedLookupData = lookup
.request(target2.path("/destination/persistent").path(ns1 + "/" + urlEncodedTopic))
.get(LookupData.class);
// (b) Uri encoding
LookupData uriEncodedLookupData = lookup
.request(target2.path("/destination/persistent").path(ns1 + "/" + uriEncodedTopic))
.get(LookupData.class);
Assert.assertNotNull(urlEncodedLookupData.getBrokerUrl());
assertEquals(urlEncodedLookupData.getBrokerUrl(), uriEncodedLookupData.getBrokerUrl());
// (3) Get Topic Stats
final CompletableFuture<TopicStats> urlStats = new CompletableFuture<>();
// (a) Url encoding
persistent.asyncGetRequest(persistentTopics.path(ns1).path(urlEncodedTopic + "-partition-1").path("stats"),
new InvocationCallback<TopicStats>() {
@Override
public void completed(TopicStats response) {
urlStats.complete(response);
}
@Override
public void failed(Throwable e) {
urlStats.completeExceptionally(e);
}
});
// (b) Uri encoding
final CompletableFuture<TopicStats> uriStats = new CompletableFuture<>();
persistent.asyncGetRequest(persistentTopics.path(ns1).path(uriEncodedTopic + "-partition-1").path("stats"),
new InvocationCallback<TopicStats>() {
@Override
public void completed(TopicStats response) {
uriStats.complete(response);
}
@Override
public void failed(Throwable e) {
uriStats.completeExceptionally(e);
}
});
assertEquals(urlStats.get().getSubscriptions().size(), 1);
assertEquals(uriStats.get().getSubscriptions().size(), 1);
}
static class MockedPulsarService extends MockedPulsarServiceBaseTest {
private final ServiceConfiguration conf;
public MockedPulsarService(ServiceConfiguration conf) {
super();
this.conf = conf;
}
@Override
protected void setup() throws Exception {
super.conf.setLoadManagerClassName(conf.getLoadManagerClassName());
super.internalSetup();
}
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
}
public PulsarService getPulsar() {
return pulsar;
}
public PulsarAdmin getAdmin() {
return admin;
}
}
@Test
public void testTopicBundleRangeLookup() throws Exception {
admin.clusters().createCluster("usw", ClusterData.builder().build());
TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"),
Set.of("use", "usw"));
admin.tenants().updateTenant("prop-xyz", tenantInfo);
admin.namespaces().createNamespace("prop-xyz/use/getBundleNs", 100);
assertEquals(admin.namespaces().getPolicies("prop-xyz/use/getBundleNs").bundles.getNumBundles(), 100);
// (1) create a topic
final String topicName = "persistent://prop-xyz/use/getBundleNs/topic1";
String bundleRange = admin.lookups().getBundleRange(topicName);
assertEquals(bundleRange, pulsar.getNamespaceService().getBundle(TopicName.get(topicName)).getBundleRange());
admin.tenants().updateTenant("prop-xyz", new TenantInfoImpl(Set.of("role1", "role2"),
Set.of("use")));
}
@Test
public void testTriggerCompaction() throws Exception {
String topicName = "persistent://prop-xyz/use/ns1/topic1";
// create a topic by creating a producer
pulsarClient.newProducer(Schema.BYTES).topic(topicName).create().close();
assertNotNull(pulsar.getBrokerService().getTopicReference(topicName));
// mock actual compaction, we don't need to really run it
CompletableFuture<Long> promise = new CompletableFuture<>();
Compactor compactor = ((PulsarCompactionServiceFactory)pulsar.getCompactionServiceFactory()).getCompactor();
doReturn(promise).when(compactor).compact(topicName);
admin.topics().triggerCompaction(topicName);
// verify compact called once
verify(compactor).compact(topicName);
try {
admin.topics().triggerCompaction(topicName);
fail("Shouldn't be able to run while already running");
} catch (ConflictException e) {
// expected
}
// compact shouldn't have been called again
verify(compactor).compact(topicName);
// complete first compaction, and trigger again
promise.complete(1L);
admin.topics().triggerCompaction(topicName);
// verify compact was called again
verify(compactor, times(2)).compact(topicName);
}
@Test
public void testCompactionStatus() throws Exception {
String topicName = "persistent://prop-xyz/use/ns1/topic1";
// create a topic by creating a producer
pulsarClient.newProducer(Schema.BYTES).topic(topicName).create().close();
assertNotNull(pulsar.getBrokerService().getTopicReference(topicName));
assertEquals(admin.topics().compactionStatus(topicName).status,
LongRunningProcessStatus.Status.NOT_RUN);
// mock actual compaction, we don't need to really run it
CompletableFuture<Long> promise = new CompletableFuture<>();
Compactor compactor = ((PulsarCompactionServiceFactory)pulsar.getCompactionServiceFactory()).getCompactor();
doReturn(promise).when(compactor).compact(topicName);
admin.topics().triggerCompaction(topicName);
assertEquals(admin.topics().compactionStatus(topicName).status,
LongRunningProcessStatus.Status.RUNNING);
promise.complete(1L);
assertEquals(admin.topics().compactionStatus(topicName).status,
LongRunningProcessStatus.Status.SUCCESS);
CompletableFuture<Long> errorPromise = new CompletableFuture<>();
doReturn(errorPromise).when(compactor).compact(topicName);
admin.topics().triggerCompaction(topicName);
errorPromise.completeExceptionally(new Exception("Failed at something"));
assertEquals(admin.topics().compactionStatus(topicName).status,
LongRunningProcessStatus.Status.ERROR);
assertTrue(admin.topics().compactionStatus(topicName)
.lastError.contains("Failed at something"));
}
private void clearCache() {
((MetadataCacheImpl) pulsar.getPulsarResources().getNamespaceResources().getCache()).invalidateAll();
}
}