You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2022/12/14 11:39:17 UTC
[pulsar] 02/02: [fix][broker] Fix the breaking change of standalone metadata initialization (#18909)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e1ea99c641951e3c76afcbbf08ffd399023bffd4
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed Dec 14 19:35:24 2022 +0800
[fix][broker] Fix the breaking change of standalone metadata initialization (#18909)
---
.../java/org/apache/pulsar/PulsarStandalone.java | 58 +++++----
.../pulsar/MockTokenAuthenticationProvider.java | 104 ++++++++++++++++
.../org/apache/pulsar/PulsarStandaloneTest.java | 133 ++++++++++-----------
.../configurations/standalone_no_client_auth.conf | 31 +++++
.../tests/integration/standalone/SmokeTest.java | 15 ---
.../topologies/PulsarStandaloneTestBase.java | 3 -
6 files changed, 225 insertions(+), 119 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
index b5cc3462691..6317d02eb58 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java
@@ -25,6 +25,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import java.io.File;
import java.nio.file.Paths;
+import java.util.List;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -32,9 +33,9 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.broker.resources.ClusterResources;
import org.apache.pulsar.broker.resources.NamespaceResources;
-import org.apache.pulsar.broker.resources.TenantResources;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
@@ -367,36 +368,31 @@ public class PulsarStandalone implements AutoCloseable {
log.debug("--- setup completed ---");
}
- @VisibleForTesting
- void createNameSpace(String cluster, String publicTenant, NamespaceName ns) throws Exception {
- ClusterResources cr = broker.getPulsarResources().getClusterResources();
- TenantResources tr = broker.getPulsarResources().getTenantResources();
- NamespaceResources nsr = broker.getPulsarResources().getNamespaceResources();
-
- if (!cr.clusterExists(cluster)) {
- cr.createCluster(cluster,
- ClusterData.builder()
- .serviceUrl(broker.getWebServiceAddress())
- .serviceUrlTls(broker.getWebServiceAddressTls())
- .brokerServiceUrl(broker.getBrokerServiceUrl())
- .brokerServiceUrlTls(broker.getBrokerServiceUrlTls())
- .build());
- }
-
- if (!tr.tenantExists(publicTenant)) {
- tr.createTenant(publicTenant,
- TenantInfo.builder()
- .adminRoles(Sets.newHashSet(config.getSuperUserRoles()))
- .allowedClusters(Sets.newHashSet(cluster))
- .build());
- }
-
- if (!nsr.namespaceExists(ns)) {
- try {
- broker.getAdminClient().namespaces().createNamespace(ns.toString());
- } catch (Exception e) {
- log.warn("Failed to create the default namespace {}: {}", ns, e.getMessage());
+ private void createNameSpace(String cluster, String publicTenant, NamespaceName ns) throws Exception {
+ PulsarAdmin admin = broker.getAdminClient();
+ try {
+ final List<String> clusters = admin.clusters().getClusters();
+ if (!clusters.contains(cluster)) {
+ admin.clusters().createCluster(cluster, ClusterData.builder()
+ .serviceUrl(broker.getWebServiceAddress())
+ .serviceUrlTls(broker.getWebServiceAddressTls())
+ .brokerServiceUrl(broker.getBrokerServiceUrl())
+ .brokerServiceUrlTls(broker.getBrokerServiceUrlTls())
+ .build());
+ }
+ final List<String> tenants = admin.tenants().getTenants();
+ if (!tenants.contains(publicTenant)) {
+ admin.tenants().createTenant(publicTenant, TenantInfo.builder()
+ .adminRoles(Sets.newHashSet(config.getSuperUserRoles()))
+ .allowedClusters(Sets.newHashSet(cluster))
+ .build());
+ }
+ final List<String> namespaces = admin.namespaces().getNamespaces(publicTenant);
+ if (!namespaces.contains(ns.toString())) {
+ admin.namespaces().createNamespace(ns.toString(), config.getDefaultNumberOfNamespaceBundles());
}
+ } catch (PulsarAdminException e) {
+ log.error("Failed to create namespace {} on cluster {} and tenant {}", ns, cluster, publicTenant, e);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/MockTokenAuthenticationProvider.java b/pulsar-broker/src/test/java/org/apache/pulsar/MockTokenAuthenticationProvider.java
new file mode 100644
index 00000000000..e2593c9fe40
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/MockTokenAuthenticationProvider.java
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import javax.naming.AuthenticationException;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.authentication.AuthenticationProvider;
+import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.AuthenticationDataProvider;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+public class MockTokenAuthenticationProvider implements AuthenticationProvider {
+
+ public static final String KEY = "role";
+
+ @Override
+ public void initialize(ServiceConfiguration config) throws IOException {
+ // No ops
+ }
+
+ @Override
+ public String getAuthMethodName() {
+ return "mock";
+ }
+
+ @Override
+ public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
+ if (!authData.hasDataFromHttp()) {
+ throw new AuthenticationException("No HTTP data in " + authData);
+ }
+ String value = authData.getHttpHeader(KEY);
+ if (value == null) {
+ throw new AuthenticationException("No HTTP header for " + KEY);
+ }
+ return value;
+ }
+
+ @Override
+ public void close() throws IOException {
+ // No ops
+ }
+
+ public static class MockAuthentication implements Authentication {
+
+ @Override
+ public AuthenticationDataProvider getAuthData(String brokerHostName) throws PulsarClientException {
+ return new MockAuthenticationData();
+ }
+
+ @Override
+ public String getAuthMethodName() {
+ return "mock";
+ }
+
+ @Override
+ public void configure(Map<String, String> authParams) {
+ // No ops
+ }
+
+ @Override
+ public void start() throws PulsarClientException {
+ // No ops
+ }
+
+ @Override
+ public void close() throws IOException {
+ // No ops
+ }
+ }
+
+ private static class MockAuthenticationData implements AuthenticationDataProvider {
+
+ @Override
+ public boolean hasDataForHttp() {
+ return true;
+ }
+
+ @Override
+ public Set<Map.Entry<String, String>> getHttpHeaders() throws Exception {
+ return Collections.singletonMap(KEY, "admin").entrySet();
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/PulsarStandaloneTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/PulsarStandaloneTest.java
index f62ce1c5e46..a3bca0a8f75 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/PulsarStandaloneTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/PulsarStandaloneTest.java
@@ -19,89 +19,30 @@
package org.apache.pulsar;
import static org.apache.commons.io.FileUtils.cleanDirectory;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
import java.io.File;
+import java.util.Collections;
import java.util.List;
+import lombok.Cleanup;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.util.IOUtils;
-import org.apache.pulsar.client.admin.Namespaces;
import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.common.naming.NamespaceName;
-import org.apache.pulsar.broker.PulsarService;
-import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.broker.resources.ClusterResources;
-import org.apache.pulsar.broker.resources.NamespaceResources;
-import org.apache.pulsar.broker.resources.PulsarResources;
-import org.apache.pulsar.broker.resources.TenantResources;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
import org.testng.Assert;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Test(groups = "broker")
public class PulsarStandaloneTest {
- @Test
- public void testCreateNameSpace() throws Exception {
- final String cluster = "cluster1";
- final String tenant = "tenant1";
- final NamespaceName ns = NamespaceName.get(tenant, "ns1");
-
- ClusterResources cr = mock(ClusterResources.class);
- when(cr.clusterExists(cluster)).thenReturn(false).thenReturn(true);
- doNothing().when(cr).createCluster(eq(cluster), any());
-
- TenantResources tr = mock(TenantResources.class);
- when(tr.tenantExists(tenant)).thenReturn(false).thenReturn(true);
- doNothing().when(tr).createTenant(eq(tenant), any());
-
- NamespaceResources nsr = mock(NamespaceResources.class);
- when(nsr.namespaceExists(ns)).thenReturn(false).thenReturn(true).thenReturn(false);
- doNothing().when(nsr).createPolicies(eq(ns), any());
-
- PulsarResources resources = mock(PulsarResources.class);
- when(resources.getClusterResources()).thenReturn(cr);
- when(resources.getTenantResources()).thenReturn(tr);
- when(resources.getNamespaceResources()).thenReturn(nsr);
-
- Namespaces namespaces = mock(Namespaces.class);
- doNothing().when(namespaces).createNamespace(any());
- PulsarAdmin admin = mock(PulsarAdmin.class);
- when(admin.namespaces()).thenReturn(namespaces);
-
- PulsarService broker = mock(PulsarService.class);
- when(broker.getPulsarResources()).thenReturn(resources);
- when(broker.getWebServiceAddress()).thenReturn("pulsar://localhost:8080");
- when(broker.getWebServiceAddressTls()).thenReturn(null);
- when(broker.getBrokerServiceUrl()).thenReturn("pulsar://localhost:6650");
- when(broker.getBrokerServiceUrlTls()).thenReturn(null);
- when(broker.getAdminClient()).thenReturn(admin);
-
- ServiceConfiguration config = new ServiceConfiguration();
- config.setClusterName(cluster);
-
- PulsarStandalone standalone = new PulsarStandalone();
- standalone.setBroker(broker);
- standalone.setConfig(config);
-
- standalone.createNameSpace(cluster, tenant, ns);
- standalone.createNameSpace(cluster, tenant, ns);
- verify(cr, times(1)).createCluster(eq(cluster), any());
- verify(tr, times(1)).createTenant(eq(tenant), any());
- verify(admin, times(1)).namespaces();
- verify(admin.namespaces(), times(1)).createNamespace(eq(ns.toString()));
-
- doThrow(new PulsarAdminException("No permission")).when(namespaces).createNamespace(any());
- standalone.createNameSpace(cluster, tenant, ns);
+ @DataProvider
+ public Object[][] enableBrokerClientAuth() {
+ return new Object[][] { { true }, { false } };
}
- @Test(groups = "broker")
+ @Test
public void testStandaloneWithRocksDB() throws Exception {
String[] args = new String[]{"--config",
"./src/test/resources/configurations/pulsar_broker_test_standalone_with_rocksdb.conf"};
@@ -131,4 +72,56 @@ public class PulsarStandaloneTest {
cleanDirectory(tempDir);
}
+ @Test(dataProvider = "enableBrokerClientAuth")
+ public void testMetadataInitialization(boolean enableBrokerClientAuth) throws Exception {
+ final File metadataDir = IOUtils.createTempDir("standalone", "metadata");
+ @Cleanup final PulsarStandaloneStarter standalone = new PulsarStandaloneStarter(new String[]{
+ "--config",
+ "./src/test/resources/configurations/standalone_no_client_auth.conf",
+ "-nss",
+ "-nfw",
+ "--metadata-url",
+ "rocksdb://" + metadataDir.getAbsolutePath()
+ });
+ if (enableBrokerClientAuth) {
+ standalone.getConfig().setBrokerClientAuthenticationPlugin(
+ MockTokenAuthenticationProvider.MockAuthentication.class.getName());
+ }
+ final File bkDir = IOUtils.createTempDir("standalone", "bk");
+ standalone.setNumOfBk(1);
+ standalone.setBkDir(bkDir.getAbsolutePath());
+ standalone.start();
+
+ @Cleanup PulsarAdmin admin = PulsarAdmin.builder()
+ .serviceHttpUrl("http://localhost:8080")
+ .authentication(new MockTokenAuthenticationProvider.MockAuthentication())
+ .build();
+ if (enableBrokerClientAuth) {
+ assertTrue(admin.clusters().getClusters().contains("test_cluster"));
+ assertTrue(admin.tenants().getTenants().contains("public"));
+ assertTrue(admin.namespaces().getNamespaces("public").contains("public/default"));
+ } else {
+ assertTrue(admin.clusters().getClusters().isEmpty());
+ admin.clusters().createCluster("test_cluster", ClusterData.builder()
+ .serviceUrl("http://localhost:8080/")
+ .brokerServiceUrl("pulsar://localhost:6650/")
+ .build());
+ assertTrue(admin.tenants().getTenants().isEmpty());
+ admin.tenants().createTenant("public", TenantInfo.builder()
+ .adminRoles(Collections.singleton("admin"))
+ .allowedClusters(Collections.singleton("test_cluster"))
+ .build());
+
+ assertTrue(admin.namespaces().getNamespaces("public").isEmpty());
+ admin.namespaces().createNamespace("public/default");
+ }
+
+ String topic = "test-get-topic-bundle-range";
+ admin.topics().createNonPartitionedTopic(topic);
+ assertEquals(admin.lookups().getBundleRange(topic), "0xc0000000_0xffffffff");
+
+ standalone.close();
+ cleanDirectory(bkDir);
+ cleanDirectory(metadataDir);
+ }
}
diff --git a/pulsar-broker/src/test/resources/configurations/standalone_no_client_auth.conf b/pulsar-broker/src/test/resources/configurations/standalone_no_client_auth.conf
new file mode 100644
index 00000000000..573d6cdba99
--- /dev/null
+++ b/pulsar-broker/src/test/resources/configurations/standalone_no_client_auth.conf
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+brokerServicePort=6650
+webServicePort=8080
+allowLoopback=true
+clusterName=test_cluster
+superUserRoles=admin
+managedLedgerDefaultEnsembleSize=1
+managedLedgerDefaultWriteQuorum=1
+managedLedgerDefaultAckQuorum=1
+authenticationEnabled=true
+authenticationProviders=org.apache.pulsar.MockTokenAuthenticationProvider
+brokerClientAuthenticationPlugin=
+brokerClientAuthenticationParameters=
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/standalone/SmokeTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/standalone/SmokeTest.java
index 9fb90f172c5..438d0e42353 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/standalone/SmokeTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/standalone/SmokeTest.java
@@ -18,12 +18,7 @@
*/
package org.apache.pulsar.tests.integration.standalone;
-import static org.testng.Assert.assertEquals;
import java.util.function.Supplier;
-import lombok.Cleanup;
-import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.tests.integration.suites.PulsarStandaloneTestSuite;
import org.testng.annotations.Test;
@@ -33,14 +28,4 @@ public class SmokeTest extends PulsarStandaloneTestSuite {
public void testPublishAndConsume(Supplier<String> serviceUrl, boolean isPersistent) throws Exception {
super.testPublishAndConsume(serviceUrl.get(), isPersistent);
}
-
- @Test
- public void testGetBundleRange() throws PulsarClientException, PulsarAdminException {
- @Cleanup
- PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build();
-
- String topic = "test-get-topic-bundle-range";
- admin.topics().createNonPartitionedTopic(topic);
- assertEquals(admin.lookups().getBundleRange(topic), "0xc0000000_0xffffffff");
- }
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarStandaloneTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarStandaloneTestBase.java
index 1261e76c0ba..85d7b46c711 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarStandaloneTestBase.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarStandaloneTestBase.java
@@ -121,7 +121,4 @@ public abstract class PulsarStandaloneTestBase extends PulsarTestBase {
}
}
- protected String getHttpServiceUrl() {
- return container.getHttpServiceUrl();
- }
}