You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2017/10/26 21:02:41 UTC
[GitHub] merlimat closed pull request #854: Allow to configure the default number of bundles for new namespaces
merlimat closed pull request #854: Allow to configure the default number of bundles for new namespaces
URL: https://github.com/apache/incubator-pulsar/pull/854
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/conf/broker.conf b/conf/broker.conf
index b672bf0eb..b842e6e29 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -92,6 +92,10 @@ brokerDeduplicationEntriesInterval=1000
# relative to a disconnected producer. Default is 6 hours.
brokerDeduplicationProducerInactivityTimeoutMinutes=360
+# When a namespace is created without specifying the number of bundle, this
+# value will be used as the default
+defaultNumberOfNamespaceBundles=4
+
# Enable check for minimum allowed client library version
clientLibraryVersionCheckEnabled=false
@@ -287,7 +291,7 @@ managedLedgerMaxUnackedRangesToPersist=10000
# Max number of "acknowledgment holes" that can be stored in Zookeeper. If number of unack message range is higher
# than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
-# zookeeper.
+# zookeeper.
managedLedgerMaxUnackedRangesToPersistInZooKeeper=1000
diff --git a/conf/standalone.conf b/conf/standalone.conf
index ccbb5c092..54e951dad 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -85,6 +85,9 @@ brokerDeduplicationEntriesInterval=1000
# relative to a disconnected producer. Default is 6 hours.
brokerDeduplicationProducerInactivityTimeoutMinutes=360
+# When a namespace is created without specifying the number of bundle, this
+# value will be used as the default
+defaultNumberOfNamespaceBundles=4
# Enable check for minimum allowed client library version
clientLibraryVersionCheckEnabled=false
@@ -260,7 +263,7 @@ managedLedgerMaxUnackedRangesToPersist=10000
# Max number of "acknowledgment holes" that can be stored in Zookeeper. If number of unack message range is higher
# than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
-# zookeeper.
+# zookeeper.
managedLedgerMaxUnackedRangesToPersistInZooKeeper=1000
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 587f6a1cc..eb4437270 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -100,6 +100,10 @@
// relative to a disconnected producer. Default is 6 hours.
private int brokerDeduplicationProducerInactivityTimeoutMinutes = 360;
+ // When a namespace is created without specifying the number of bundle, this
+ // value will be used as the default
+ private int defaultNumberOfNamespaceBundles = 4;
+
// Enable check for minimum allowed client library version
private boolean clientLibraryVersionCheckEnabled = false;
// Allow client libraries with no version information
@@ -269,7 +273,7 @@
private int managedLedgerMaxUnackedRangesToPersist = 10000;
// Max number of "acknowledgment holes" that can be stored in Zookeeper. If number of unack message range is higher
// than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
- // zookeeper.
+ // zookeeper.
private int managedLedgerMaxUnackedRangesToPersistInZooKeeper = 1000;
/*** --- Load balancer --- ****/
@@ -502,6 +506,14 @@ public void setBrokerDeduplicationProducerInactivityTimeoutMinutes(
this.brokerDeduplicationProducerInactivityTimeoutMinutes = brokerDeduplicationProducerInactivityTimeoutMinutes;
}
+ public int getDefaultNumberOfNamespaceBundles() {
+ return defaultNumberOfNamespaceBundles;
+ }
+
+ public void setDefaultNumberOfNamespaceBundles(int defaultNumberOfNamespaceBundles) {
+ this.defaultNumberOfNamespaceBundles = defaultNumberOfNamespaceBundles;
+ }
+
public long getBrokerDeleteInactiveTopicsFrequencySeconds() {
return brokerDeleteInactiveTopicsFrequencySeconds;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java
index f8397e359..a66ee1ec3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/Namespaces.java
@@ -217,6 +217,9 @@ public void createNamespace(@PathParam("property") String property, @PathParam("
} else {
policies.bundles = validateBundlesData(initialBundles);
}
+ } else {
+ int defaultNumberOfBundles = config().getDefaultNumberOfNamespaceBundles();
+ policies.bundles = getBundles(defaultNumberOfBundles);
}
zkCreateOptimistic(path(POLICIES, property, cluster, namespace),
@@ -1170,7 +1173,7 @@ private void validatePersistencePolicies(PersistencePolicies persistence) {
(persistence.getBookkeeperEnsemble() >= persistence.getBookkeeperWriteQuorum())
&& (persistence.getBookkeeperWriteQuorum() >= persistence.getBookkeeperAckQuorum()),
"Bookkeeper Ensemble (%s) >= WriteQuorum (%s) >= AckQuoru (%s)", persistence.getBookkeeperEnsemble(),
- persistence.getBookkeeperWriteQuorum(), persistence.getBookkeeperAckQuorum());
+ persistence.getBookkeeperWriteQuorum(), persistence.getBookkeeperAckQuorum());
}catch(NullPointerException | IllegalArgumentException e) {
throw new RestException(Status.PRECONDITION_FAILED, e.getMessage());
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
index 6029ed5af..e0d55a661 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
@@ -91,6 +91,7 @@ void setup() throws Exception {
config.setWebServicePort(brokerWebServicePorts[i]);
config.setZookeeperServers("127.0.0.1" + ":" + ZOOKEEPER_PORT);
config.setBrokerServicePort(brokerNativeBrokerPorts[i]);
+ config.setDefaultNumberOfNamespaceBundles(1);
configurations[i] = config;
pulsarServices[i] = new PulsarService(config);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index 849f5f343..29f2ef65f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -93,6 +93,7 @@ protected void resetConfig() {
this.conf.setAdvertisedAddress("localhost"); // there are TLS tests in here, they need to use localhost because of the certificate
this.conf.setManagedLedgerCacheSizeMB(8);
this.conf.setActiveConsumerFailoverDelayTimeMillis(0);
+ this.conf.setDefaultNumberOfNamespaceBundles(1);
}
protected final void internalSetup() throws Exception {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceCreateBundlesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceCreateBundlesTest.java
new file mode 100644
index 000000000..7bb6a8fa5
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceCreateBundlesTest.java
@@ -0,0 +1,55 @@
+/**
+ * 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.namespace;
+
+import static org.testng.Assert.assertEquals;
+
+import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.common.policies.data.Policies;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class NamespaceCreateBundlesTest extends BrokerTestBase {
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ conf.setDefaultNumberOfNamespaceBundles(16);
+ super.baseSetup();
+ }
+
+ @AfterMethod
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void testCreateNamespaceWithDefaultBundles() throws Exception {
+ String namespaceName = "prop/use/default-bundles";
+
+ admin.namespaces().createNamespace(namespaceName);
+
+ Policies policies = admin.namespaces().getPolicies(namespaceName);
+ assertEquals(policies.bundles.numBundles, 16);
+ assertEquals(policies.bundles.boundaries.size(), 17);
+ }
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
index 523c8c6e7..857104ab6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
@@ -333,6 +333,71 @@ public void testLoadReportDeserialize() throws Exception {
System.out.println(result2);
}
+ @Test
+ public void testCreateNamespaceWithDefaultNumberOfBundles() throws Exception {
+ OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache());
+ doNothing().when(MockOwnershipCache).disableOwnership(any(NamespaceBundle.class));
+ Field ownership = NamespaceService.class.getDeclaredField("ownershipCache");
+ ownership.setAccessible(true);
+ ownership.set(pulsar.getNamespaceService(), MockOwnershipCache);
+ NamespaceService namespaceService = pulsar.getNamespaceService();
+ NamespaceName nsname = new NamespaceName("pulsar/global/ns1");
+ DestinationName dn = DestinationName.get("persistent://pulsar/global/ns1/topic-1");
+ NamespaceBundles bundles = namespaceService.getNamespaceBundleFactory().getBundles(nsname);
+ NamespaceBundle originalBundle = bundles.findBundle(dn);
+
+ // Split bundle and take ownership of split bundles
+ CompletableFuture<Void> result = namespaceService.splitAndOwnBundle(originalBundle);
+
+ try {
+ result.get();
+ } catch (Exception e) {
+ // make sure: no failure
+ fail("split bundle faild", e);
+ }
+ NamespaceBundleFactory bundleFactory = this.pulsar.getNamespaceService().getNamespaceBundleFactory();
+ NamespaceBundles updatedNsBundles = bundleFactory.getBundles(nsname);
+
+ // new updated bundles shouldn't be null
+ assertNotNull(updatedNsBundles);
+ List<NamespaceBundle> bundleList = updatedNsBundles.getBundles();
+ assertNotNull(bundles);
+
+ NamespaceBundleFactory utilityFactory = NamespaceBundleFactory.createFactory(pulsar, Hashing.crc32());
+
+ // (1) validate bundleFactory-cache has newly split bundles and removed old parent bundle
+ Pair<NamespaceBundles, List<NamespaceBundle>> splitBundles = splitBundles(utilityFactory, nsname, bundles,
+ originalBundle);
+ assertNotNull(splitBundles);
+ Set<NamespaceBundle> splitBundleSet = new HashSet<>(splitBundles.getRight());
+ splitBundleSet.removeAll(bundleList);
+ assertTrue(splitBundleSet.isEmpty());
+
+ // (2) validate LocalZookeeper policies updated with newly created split
+ // bundles
+ String path = joinPath(LOCAL_POLICIES_ROOT, nsname.toString());
+ byte[] content = this.pulsar.getLocalZkCache().getZooKeeper().getData(path, null, new Stat());
+ Policies policies = ObjectMapperFactory.getThreadLocal().readValue(content, Policies.class);
+ NamespaceBundles localZkBundles = bundleFactory.getBundles(nsname, policies.bundles);
+ assertTrue(updatedNsBundles.equals(localZkBundles));
+ log.info("Policies: {}", policies);
+
+ // (3) validate ownership of new split bundles by local owner
+ bundleList.stream().forEach(b -> {
+ try {
+ byte[] data = this.pulsar.getLocalZkCache().getZooKeeper().getData(ServiceUnitZkUtils.path(b), null,
+ new Stat());
+ NamespaceEphemeralData node = ObjectMapperFactory.getThreadLocal().readValue(data,
+ NamespaceEphemeralData.class);
+ Assert.assertEquals(node.getNativeUrl(), this.pulsar.getBrokerServiceUrl());
+ } catch (Exception e) {
+ fail("failed to setup ownership", e);
+ }
+ });
+
+ }
+
+
@SuppressWarnings("unchecked")
private Pair<NamespaceBundles, List<NamespaceBundle>> splitBundles(NamespaceBundleFactory utilityFactory,
NamespaceName nsname, NamespaceBundles bundles, NamespaceBundle targetBundle) throws Exception {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
index ccd1ba4c4..f57bcea65 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
@@ -94,6 +94,7 @@
@BeforeMethod
@Override
protected void setup() throws Exception {
+ conf.setDefaultNumberOfNamespaceBundles(1);
super.init();
org.apache.pulsar.client.api.ClientConfiguration clientConf = new org.apache.pulsar.client.api.ClientConfiguration();
clientConf.setStatsInterval(0, TimeUnit.SECONDS);
@@ -108,15 +109,15 @@ protected void cleanup() throws Exception {
super.internalCleanup();
}
-
+
/**
* UsecaseL Multiple Broker => Lookup Redirection test
- *
+ *
* 1. Broker1 is a leader
* 2. Lookup request reaches to Broker2 which redirects to leader (Broker1) with authoritative = false
* 3. Leader (Broker1) finds out least loaded broker as Broker2 and redirects request to Broker2 with authoritative = true
* 4. Broker2 receives final request to own a bundle with authoritative = true and client connects to Broker2
- *
+ *
* @throws Exception
*/
@Test
@@ -134,8 +135,8 @@ public void testMultipleBrokerLookup() throws Exception {
PulsarService pulsar2 = startBroker(conf2);
pulsar.getLoadManager().get().writeLoadReportOnZookeeper();
pulsar2.getLoadManager().get().writeLoadReportOnZookeeper();
-
-
+
+
LoadManager loadManager1 = spy(pulsar.getLoadManager().get());
LoadManager loadManager2 = spy(pulsar2.getLoadManager().get());
Field loadManagerField = NamespaceService.class.getDeclaredField("loadManager");
@@ -144,23 +145,23 @@ public void testMultipleBrokerLookup() throws Exception {
// mock: redirect request to leader [2]
doReturn(true).when(loadManager2).isCentralized();
loadManagerField.set(pulsar2.getNamespaceService(), new AtomicReference<>(loadManager2));
-
- // mock: return Broker2 as a Least-loaded broker when leader receies request [3]
+
+ // mock: return Broker2 as a Least-loaded broker when leader receies request [3]
doReturn(true).when(loadManager1).isCentralized();
SimpleResourceUnit resourceUnit = new SimpleResourceUnit(pulsar2.getWebServiceAddress(), null);
doReturn(resourceUnit).when(loadManager1).getLeastLoaded(any(ServiceUnitId.class));
loadManagerField.set(pulsar.getNamespaceService(), new AtomicReference<>(loadManager1));
-
+
/**** started broker-2 ****/
URI brokerServiceUrl = new URI("pulsar://localhost:" + conf2.getBrokerServicePort());
PulsarClient pulsarClient2 = PulsarClient.create(brokerServiceUrl.toString(), new ClientConfiguration());
-
+
// load namespace-bundle by calling Broker2
Consumer consumer = pulsarClient2.subscribe("persistent://my-property/use/my-ns/my-topic1", "my-subscriber-name",
new ConsumerConfiguration());
Producer producer = pulsarClient.createProducer("persistent://my-property/use/my-ns/my-topic1", new ProducerConfiguration());
-
+
for (int i = 0; i < 10; i++) {
String message = "my-message-" + i;
producer.send(message.getBytes());
@@ -184,16 +185,16 @@ public void testMultipleBrokerLookup() throws Exception {
pulsar2.close();
loadManager1 = null;
loadManager2 = null;
-
+
}
-
+
/**
- * Usecase: Redirection due to different cluster
- * 1. Broker1 runs on cluster: "use" and Broker2 runs on cluster: "use2"
+ * Usecase: Redirection due to different cluster
+ * 1. Broker1 runs on cluster: "use" and Broker2 runs on cluster: "use2"
* 2. Broker1 receives "use2" cluster request => Broker1 reads "/clusters" from global-zookkeeper and
* redirects request to Broker2 whch serves "use2"
* 3. Broker2 receives redirect request and own namespace bundle
- *
+ *
* @throws Exception
*/
@Test
@@ -211,42 +212,42 @@ public void testMultipleBrokerDifferentClusterLookup() throws Exception {
conf2.setAdvertisedAddress("localhost");
conf2.setClusterName(newCluster); // Broker2 serves newCluster
String broker2ServiceUrl = "pulsar://localhost:" + conf2.getBrokerServicePort();
-
+
admin.clusters().createCluster(newCluster, new ClusterData("http://127.0.0.1:" + BROKER_WEBSERVICE_PORT, null, broker2ServiceUrl, null));
admin.properties().createProperty(property,
new PropertyAdmin(Lists.newArrayList("appid1", "appid2"), Sets.newHashSet(newCluster)));
admin.namespaces().createNamespace(property + "/" + newCluster + "/my-ns");
-
-
+
+
PulsarService pulsar2 = startBroker(conf2);
pulsar.getLoadManager().get().writeLoadReportOnZookeeper();
pulsar2.getLoadManager().get().writeLoadReportOnZookeeper();
-
+
URI brokerServiceUrl = new URI(broker2ServiceUrl);
PulsarClient pulsarClient2 = PulsarClient.create(brokerServiceUrl.toString(), new ClientConfiguration());
-
+
// enable authorization: so, broker can validate cluster and redirect if finds different cluster
pulsar.getConfiguration().setAuthorizationEnabled(true);
// restart broker with authorization enabled: it initialize AuthorizationManager
stopBroker();
startBroker();
-
+
LoadManager loadManager2 = spy(pulsar2.getLoadManager().get());
Field loadManagerField = NamespaceService.class.getDeclaredField("loadManager");
loadManagerField.setAccessible(true);
-
+
// mock: return Broker2 as a Least-loaded broker when leader receies request
doReturn(true).when(loadManager2).isCentralized();
SimpleResourceUnit resourceUnit = new SimpleResourceUnit(pulsar2.getWebServiceAddress(), null);
doReturn(resourceUnit).when(loadManager2).getLeastLoaded(any(ServiceUnitId.class));
loadManagerField.set(pulsar.getNamespaceService(), new AtomicReference<>(loadManager2));
/**** started broker-2 ****/
-
+
// load namespace-bundle by calling Broker2
Consumer consumer = pulsarClient.subscribe("persistent://my-property2/use2/my-ns/my-topic1", "my-subscriber-name",
new ConsumerConfiguration());
Producer producer = pulsarClient2.createProducer("persistent://my-property2/use2/my-ns/my-topic1", new ProducerConfiguration());
-
+
for (int i = 0; i < 10; i++) {
String message = "my-message-" + i;
producer.send(message.getBytes());
@@ -265,21 +266,21 @@ public void testMultipleBrokerDifferentClusterLookup() throws Exception {
consumer.acknowledgeCumulative(msg);
consumer.close();
producer.close();
-
- // disable authorization
+
+ // disable authorization
pulsar.getConfiguration().setAuthorizationEnabled(false);
pulsarClient2.close();
pulsar2.close();
loadManager2 = null;
-
+
}
-
+
/**
- * Create #PartitionedTopic and let it served by multiple brokers which requries
+ * Create #PartitionedTopic and let it served by multiple brokers which requries
* a. tcp partitioned-metadata-lookup
- * b. multiple topic-lookup
+ * b. multiple topic-lookup
* c. partitioned producer-consumer
- *
+ *
* @throws Exception
*/
@Test
@@ -305,17 +306,17 @@ public void testPartitionTopicLookup() throws Exception {
PulsarService pulsar2 = startBroker(conf2);
pulsar.getLoadManager().get().writeLoadReportOnZookeeper();
pulsar2.getLoadManager().get().writeLoadReportOnZookeeper();
-
-
+
+
LoadManager loadManager1 = spy(pulsar.getLoadManager().get());
LoadManager loadManager2 = spy(pulsar2.getLoadManager().get());
Field loadManagerField = NamespaceService.class.getDeclaredField("loadManager");
loadManagerField.setAccessible(true);
-
+
// mock: return Broker2 as a Least-loaded broker when leader receies request
doReturn(true).when(loadManager1).isCentralized();
loadManagerField.set(pulsar.getNamespaceService(), new AtomicReference<>(loadManager1));
-
+
// mock: redirect request to leader
doReturn(true).when(loadManager2).isCentralized();
loadManagerField.set(pulsar2.getNamespaceService(), new AtomicReference<>(loadManager2));
@@ -347,7 +348,7 @@ public void testPartitionTopicLookup() throws Exception {
consumer.unsubscribe();
consumer.close();
admin.persistentTopics().deletePartitionedTopic(dn.toString());
-
+
pulsar2.close();
loadManager2 = null;
@@ -356,8 +357,8 @@ public void testPartitionTopicLookup() throws Exception {
/**
* 1. Start broker1 and broker2 with tls enable
- * 2. Hit HTTPS lookup url at broker2 which redirects to HTTPS broker1
- *
+ * 2. Hit HTTPS lookup url at broker2 which redirects to HTTPS broker1
+ *
* @throws Exception
*/
@Test
@@ -415,7 +416,7 @@ public void testWebserviceServiceTls() throws Exception {
final String lookupResourceUrl = "/lookup/v2/destination/persistent/my-property/use/my-ns/my-topic1";
- // set client cert_key file
+ // set client cert_key file
KeyManager[] keyManagers = null;
Certificate[] tlsCert = SecurityUtility.loadCertificatesFromPemFile(TLS_CLIENT_CERT_FILE_PATH);
PrivateKey tlsKey = SecurityUtility.loadPrivateKeyFromPemFile(TLS_CLIENT_KEY_FILE_PATH);
@@ -450,13 +451,13 @@ public void testWebserviceServiceTls() throws Exception {
loadManager2 = null;
}
-
+
/**
* Discovery-Service lookup over binary-protocol
* 1. Start discovery service
* 2. start broker
* 3. Create Producer/Consumer: by calling Discovery service for partitionedMetadata and topic lookup
- *
+ *
* @throws Exception
*/
@Test
@@ -469,7 +470,7 @@ public void testDiscoveryLookup() throws Exception {
DiscoveryService discoveryService = spy(new DiscoveryService(config));
doReturn(mockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory();
discoveryService.start();
-
+
// (2) lookup using discovery service
final String discoverySvcUrl = discoveryService.getServiceUrl();
ClientConfiguration clientConfig = new ClientConfiguration();
@@ -477,7 +478,7 @@ public void testDiscoveryLookup() throws Exception {
Consumer consumer = pulsarClient2.subscribe("persistent://my-property2/use2/my-ns/my-topic1", "my-subscriber-name",
new ConsumerConfiguration());
Producer producer = pulsarClient2.createProducer("persistent://my-property2/use2/my-ns/my-topic1", new ProducerConfiguration());
-
+
for (int i = 0; i < 10; i++) {
String message = "my-message-" + i;
producer.send(message.getBytes());
@@ -496,13 +497,13 @@ public void testDiscoveryLookup() throws Exception {
consumer.acknowledgeCumulative(msg);
consumer.close();
producer.close();
-
+
}
-
-
+
+
/**
* Verify discovery-service binary-proto lookup using tls
- *
+ *
* @throws Exception
*/
@Test
@@ -512,7 +513,7 @@ public void testDiscoveryLookupTls() throws Exception {
final String TLS_SERVER_KEY_FILE_PATH = "./src/test/resources/certificate/server.key";
final String TLS_CLIENT_CERT_FILE_PATH = "./src/test/resources/certificate/client.crt";
final String TLS_CLIENT_KEY_FILE_PATH = "./src/test/resources/certificate/client.key";
-
+
// (1) restart broker1 with tls enabled
conf.setTlsAllowInsecureConnection(true);
conf.setTlsEnabled(true);
@@ -520,7 +521,7 @@ public void testDiscoveryLookupTls() throws Exception {
conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
stopBroker();
startBroker();
-
+
// (2) start discovery service
ServiceConfig config = new ServiceConfig();
config.setServicePort(nextFreePort());
@@ -532,11 +533,11 @@ public void testDiscoveryLookupTls() throws Exception {
DiscoveryService discoveryService = spy(new DiscoveryService(config));
doReturn(mockZooKeeperClientFactory).when(discoveryService).getZooKeeperClientFactory();
discoveryService.start();
-
+
// (3) lookup using discovery service
final String discoverySvcUrl = discoveryService.getServiceUrlTls();
ClientConfiguration clientConfig = new ClientConfiguration();
-
+
Map<String, String> authParams = new HashMap<>();
authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);
authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH);
@@ -545,13 +546,13 @@ public void testDiscoveryLookupTls() throws Exception {
clientConfig.setAuthentication(auth);
clientConfig.setUseTls(true);
clientConfig.setTlsAllowInsecureConnection(true);
-
-
+
+
PulsarClient pulsarClient2 = PulsarClient.create(discoverySvcUrl, clientConfig);
Consumer consumer = pulsarClient2.subscribe("persistent://my-property2/use2/my-ns/my-topic1", "my-subscriber-name",
new ConsumerConfiguration());
Producer producer = pulsarClient2.createProducer("persistent://my-property2/use2/my-ns/my-topic1", new ProducerConfiguration());
-
+
for (int i = 0; i < 10; i++) {
String message = "my-message-" + i;
producer.send(message.getBytes());
@@ -570,7 +571,7 @@ public void testDiscoveryLookupTls() throws Exception {
consumer.acknowledgeCumulative(msg);
consumer.close();
producer.close();
-
+
}
@Test
@@ -696,7 +697,7 @@ public void testDiscoveryLookupAuthorizationFailure() throws Exception {
ServiceConfig config = new ServiceConfig();
config.setServicePort(nextFreePort());
config.setBindOnLocalhost(true);
- // set Authentication provider which returns "invalid" appid so, authorization fails
+ // set Authentication provider which returns "invalid" appid so, authorization fails
Set<String> providersClassNames = Sets.newHashSet(MockAuthorizationProviderFail.class.getName());
config.setAuthenticationProviders(providersClassNames);
// enable authentication
@@ -741,11 +742,11 @@ public void start() throws PulsarClientException {
}
/**
- *
+ *
* <pre>
* When broker-1's load-manager splits the bundle and update local-policies, broker-2 should get watch of
* local-policies and update bundleCache so, new lookup can be redirected properly.
- *
+ *
* (1) Start broker-1 and broker-2
* (2) Make sure broker-2 always assign bundle to broker1
* (3) Broker-2 receives topic-1 request, creates local-policies and sets the watch
@@ -753,9 +754,9 @@ public void start() throws PulsarClientException {
* (5) Split the bundle for topic-1
* (6) Broker-2 should get the watch and update bundle cache
* (7) Make lookup request again to Broker-2 which should succeed.
- *
+ *
* </pre>
- *
+ *
* @throws Exception
*/
@Test(timeOut = 5000)
diff --git a/site/docs/latest/deployment/Kubernetes.md b/site/docs/latest/deployment/Kubernetes.md
index d880412e8..9ea4a8e71 100644
--- a/site/docs/latest/deployment/Kubernetes.md
+++ b/site/docs/latest/deployment/Kubernetes.md
@@ -237,8 +237,8 @@ $ bin/pulsar-admin properties create $MY_PROPERTY \
--admin-roles admin \
--allowed-clusters us-central
-# Create a namespace that can be spread across up to 16 brokers
-$ bin/pulsar-admin namespaces create $MY_NAMESPACE --bundles 16
+# Create a namespace
+$ bin/pulsar-admin namespaces create $MY_NAMESPACE
```
#### Experimenting with your cluster
diff --git a/site/ja/deployment/Kubernetes.md b/site/ja/deployment/Kubernetes.md
index 2b0638a18..873bf73c7 100644
--- a/site/ja/deployment/Kubernetes.md
+++ b/site/ja/deployment/Kubernetes.md
@@ -237,8 +237,8 @@ $ bin/pulsar-admin properties create $MY_PROPERTY \
--admin-roles admin \
--allowed-clusters us-central
-# 16?Broker????????????????
-$ bin/pulsar-admin namespaces create $MY_NAMESPACE --bundles 16
+# 4?Broker????????????????
+$ bin/pulsar-admin namespaces create $MY_NAMESPACE
```
#### ????????????
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services