From dc45c82018411c6fce6f05784e96894642524754 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Fri, 6 Jan 2023 17:23:30 +0800 Subject: [PATCH 1/8] [improve][broker] Add parameter check for cluster data. --- .../broker/admin/impl/ClustersBase.java | 17 +++- .../broker/admin/AdminApiClusterTest.java | 14 +++ .../common/policies/data/ClusterDataImpl.java | 28 ++++++ .../pulsar/common/util/URIPreconditions.java | 98 +++++++++++++++++++ .../common/policies/data/ClusterDataTest.java | 77 +++++++++++++++ .../common/util/URIPreconditionsTest.java | 59 +++++++++++ 6 files changed, 291 insertions(+), 2 deletions(-) create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/common/util/URIPreconditionsTest.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java index 391cfe5ebbfcc..e61a7f20d1d67 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java @@ -133,6 +133,7 @@ public void getCluster(@Suspended AsyncResponse asyncResponse, ) @ApiResponses(value = { @ApiResponse(code = 204, message = "Cluster has been created."), + @ApiResponse(code = 400, message = "Bad request parameter."), @ApiResponse(code = 403, message = "You don't have admin permission to create the cluster."), @ApiResponse(code = 409, message = "Cluster already exists."), @ApiResponse(code = 412, message = "Cluster name is not valid."), @@ -161,6 +162,11 @@ public void createCluster( .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync()) .thenCompose(__ -> { NamedEntity.checkName(cluster); + try { + clusterData.checkPropertiesIfPresent(); + } catch (IllegalArgumentException ex) { + throw new RestException(Status.BAD_REQUEST, ex.getMessage()); + } return clusterResources().getClusterAsync(cluster); }).thenCompose(clusterOpt -> { if (clusterOpt.isPresent()) { @@ -190,6 +196,7 @@ public void createCluster( notes = "This operation requires Pulsar superuser privileges.") @ApiResponses(value = { @ApiResponse(code = 204, message = "Cluster has been updated."), + @ApiResponse(code = 400, message = "Bad request parameter."), @ApiResponse(code = 403, message = "Don't have admin permission or policies are read-only."), @ApiResponse(code = 404, message = "Cluster doesn't exist."), @ApiResponse(code = 500, message = "Internal server error.") @@ -215,8 +222,14 @@ public void updateCluster( ) ClusterDataImpl clusterData) { validateSuperUserAccessAsync() .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync()) - .thenCompose(__ -> clusterResources().updateClusterAsync(cluster, old -> clusterData)) - .thenAccept(__ -> { + .thenCompose(__ -> { + try { + clusterData.checkPropertiesIfPresent(); + } catch (IllegalArgumentException ex) { + throw new RestException(Status.BAD_REQUEST, ex.getMessage()); + } + return clusterResources().updateClusterAsync(cluster, old -> clusterData); + }).thenAccept(__ -> { log.info("[{}] Updated cluster {}", clientAppId(), cluster); asyncResponse.resume(Response.ok().build()); }).exceptionally(ex -> { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiClusterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiClusterTest.java index 13291eb962856..c37cc234b1213 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiClusterTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiClusterTest.java @@ -18,8 +18,10 @@ */ package org.apache.pulsar.broker.admin; +import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertThrows; +import static org.testng.Assert.fail; import java.util.Set; import java.util.UUID; import lombok.extern.slf4j.Slf4j; @@ -51,6 +53,18 @@ public void cleanup() throws Exception { super.internalCleanup(); } + @Test + public void testCreateClusterBadRequest() { + try { + admin.clusters() + .createCluster("bad_request", ClusterData.builder() + .serviceUrl("pulsar://example.com").build()); + fail("Unexpected behaviour"); + } catch (PulsarAdminException ex) { + assertEquals(ex.getStatusCode(), 400); + } + } + @Test public void testDeleteNonExistCluster() { String cluster = "test-non-exist-cluster-" + UUID.randomUUID(); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterDataImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterDataImpl.java index 73c5b074f8434..3cf0244b2e78b 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterDataImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterDataImpl.java @@ -21,10 +21,12 @@ import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; import java.util.LinkedHashSet; +import java.util.Objects; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import org.apache.pulsar.client.api.ProxyProtocol; +import org.apache.pulsar.common.util.URIPreconditions; /** * The configuration data for a cluster. @@ -398,4 +400,30 @@ public ClusterDataImpl build() { migratedClusterUrl); } } + + /** + * Check cluster data properties by rule, if some property is illegal, it will throw + * {@link IllegalArgumentException}. + * + * @throws IllegalArgumentException exist illegal property. + */ + public void checkPropertiesIfPresent() throws IllegalArgumentException { + URIPreconditions.checkURIIfPresent(getServiceUrl(), + uri -> Objects.equals(uri.getScheme(), "http"), + "Illegal service url, example: http://pulsar.example.com:8080"); + URIPreconditions.checkURIIfPresent(getServiceUrlTls(), + uri -> Objects.equals(uri.getScheme(), "https"), + "Illegal service tls url, example: https://pulsar.example.com:8443"); + URIPreconditions.checkURIIfPresent(getBrokerServiceUrl(), + uri -> Objects.equals(uri.getScheme(), "pulsar"), + "Illegal broker service url, example: pulsar://pulsar.example.com:6650"); + URIPreconditions.checkURIIfPresent(getBrokerServiceUrlTls(), + uri -> Objects.equals(uri.getScheme(), "pulsar+ssl"), + "Illegal broker service tls url, example: pulsar+ssl://pulsar.example.com:6651"); + URIPreconditions.checkURIIfPresent(getProxyServiceUrl(), + uri -> Objects.equals(uri.getScheme(), "pulsar") + || Objects.equals(uri.getScheme(), "pulsar+ssl"), + "Illegal proxy service url, example: pulsar+ssl://ats-proxy.example.com:4443 " + + "or pulsar://ats-proxy.example.com:4080"); + } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java new file mode 100644 index 0000000000000..1be4dca9a102e --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java @@ -0,0 +1,98 @@ +/* + * 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.common.util; + +import com.google.common.base.Preconditions; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.function.Predicate; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import javax.annotation.concurrent.ThreadSafe; + +/** + * Static convenience URI checker. + */ +@ThreadSafe +public class URIPreconditions { + + /** + * Check whether the given string is a legal URI and passes the user's check. + * + * @param uri URI String + * @param predicate User defined rule + * @throws IllegalArgumentException Illegal URI or failed in the user's rules + */ + public static void checkURI(@Nonnull String uri, + @Nonnull Predicate predicate) throws IllegalArgumentException { + checkURI(uri, predicate, null); + } + + /** + * Check whether the given string is a legal URI and passes the user's check. + * + * @param uri URI String + * @param predicate User defined rule + * @throws IllegalArgumentException Illegal URI or failed in the user's rules + */ + public static void checkURIIfPresent(@Nullable String uri, + @Nonnull Predicate predicate) throws IllegalArgumentException { + checkURIIfPresent(uri, predicate, null); + } + + /** + * Check whether the given string is a legal URI and passes the user's check. + * + * @param uri URI String + * @param predicate User defined rule + * @param errorMessage Error message + * @throws IllegalArgumentException Illegal URI or failed in the user's rules + */ + public static void checkURIIfPresent(@Nullable String uri, + @Nonnull Predicate predicate, + @Nullable String errorMessage) throws IllegalArgumentException { + if (uri == null) { + return; + } + checkURI(uri, predicate, errorMessage); + } + + /** + * Check whether the given string is a legal URI and passes the user's check. + * + * @param uri URI String + * @param predicate User defined rule + * @param errorMessage Error message + * @throws IllegalArgumentException Illegal URI or failed in the user's rules + */ + public static void checkURI(@Nonnull String uri, + @Nonnull Predicate predicate, + @Nullable String errorMessage) throws IllegalArgumentException { + Preconditions.checkNotNull(uri, "uri"); + Preconditions.checkNotNull(predicate, "predicate"); + try { + URI u = new URI(uri); + if (!predicate.test(u)) { + throw new IllegalArgumentException(errorMessage == null ? "Illegal syntax: " + uri : errorMessage); + } + } catch (URISyntaxException e) { + throw new IllegalArgumentException(errorMessage == null ? "Illegal syntax: " + uri : errorMessage); + } + } +} diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/ClusterDataTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/ClusterDataTest.java index 9463129cfc4af..ba6c7cd0b7aeb 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/ClusterDataTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/ClusterDataTest.java @@ -21,6 +21,7 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotEquals; +import org.testng.Assert; import org.testng.annotations.Test; public class ClusterDataTest { @@ -109,4 +110,80 @@ public void simple() { .build().hashCode()); } + + @Test + public void testCheckProperties() { + String url1 = "/broker.messaging.c1.example.com:8080"; + String url2 = "broker.messaging.c2.example.com:8080"; + String url3 = "fdsafasfasdf"; + String url4 = "pulsar://broker.messaging.c2.example.com:8080"; + String url5 = "pulsar+ssl://broker.messaging.c2.example.com:8080"; + String url6 = "http://broker.messaging.c2.example.com:8080"; + String url7 = "https://broker.messaging.c2.example.com:8080"; + + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrl(url1).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrlTls(url1).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrl(url1).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrlTls(url1).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().proxyServiceUrl(url1).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrl(url2).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrlTls(url2).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrl(url2).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrlTls(url2).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().proxyServiceUrl(url2).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrl(url3).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrlTls(url3).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrl(url3).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrlTls(url3).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().proxyServiceUrl(url3).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrl(url4).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrlTls(url4).build().checkPropertiesIfPresent()); + ClusterDataImpl.builder().brokerServiceUrl(url4).build().checkPropertiesIfPresent(); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrlTls(url4).build().checkPropertiesIfPresent()); + ClusterDataImpl.builder().proxyServiceUrl(url4).build().checkPropertiesIfPresent(); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrl(url5).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrlTls(url5).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrl(url5).build().checkPropertiesIfPresent()); + ClusterDataImpl.builder().brokerServiceUrlTls(url5).build().checkPropertiesIfPresent(); + ClusterDataImpl.builder().proxyServiceUrl(url5).build().checkPropertiesIfPresent(); + ClusterDataImpl.builder().serviceUrl(url6).build().checkPropertiesIfPresent(); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrlTls(url6).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrl(url6).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrlTls(url6).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().proxyServiceUrl(url6).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().serviceUrl(url7).build().checkPropertiesIfPresent()); + ClusterDataImpl.builder().serviceUrlTls(url7).build().checkPropertiesIfPresent(); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrl(url7).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().brokerServiceUrlTls(url7).build().checkPropertiesIfPresent()); + Assert.assertThrows(IllegalArgumentException.class, () -> + ClusterDataImpl.builder().proxyServiceUrl(url7).build().checkPropertiesIfPresent()); + } } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/URIPreconditionsTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/URIPreconditionsTest.java new file mode 100644 index 0000000000000..d5809e8fdd0d3 --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/URIPreconditionsTest.java @@ -0,0 +1,59 @@ +/* + * 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.common.util; + +import static org.apache.pulsar.common.util.URIPreconditions.*; +import org.testng.Assert; +import org.testng.annotations.Test; +import java.util.Objects; + +public class URIPreconditionsTest { + + @Test + public void testCheckURI() { + // normal + checkURI("http://pulsar.apache.org", uri -> true); + checkURI("http://pulsar.apache.org", uri -> Objects.equals(uri.getScheme(), "http")); + // illegal + try { + checkURI("pulsar.apache.org", uri -> Objects.equals(uri.getScheme(), "http")); + Assert.fail("Unexpected behaviour"); + } catch (IllegalArgumentException ex) { + // Ok + } + try { + checkURI("pulsar.apache.org", uri -> Objects.equals(uri.getScheme(), "http"), "oops"); + Assert.fail("Unexpected behaviour"); + } catch (IllegalArgumentException ex) { + Assert.assertEquals(ex.getMessage(), "oops"); + } + } + + @Test + public void testCheckURIIfPresent() { + checkURIIfPresent(null, uri -> false); + checkURIIfPresent("http://pulsar.apache.org", uri -> true); + try { + checkURIIfPresent("http/pulsar.apache.org", uri -> uri.getScheme() != null, "Error"); + Assert.fail("Unexpected behaviour"); + } catch (IllegalArgumentException ex) { + // Ok + } + } +} From 7ca2dee6d6aae96d87ddaba1da3abab36d7a361d Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Fri, 6 Jan 2023 19:38:39 +0800 Subject: [PATCH 2/8] Fix checkstyle --- .../org/apache/pulsar/common/util/URIPreconditions.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java index 1be4dca9a102e..1db2660a47bc6 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java @@ -18,7 +18,7 @@ */ package org.apache.pulsar.common.util; -import com.google.common.base.Preconditions; +import static com.google.common.base.Preconditions.checkNotNull; import java.net.URI; import java.net.URISyntaxException; import java.util.function.Predicate; @@ -84,8 +84,8 @@ public static void checkURIIfPresent(@Nullable String uri, public static void checkURI(@Nonnull String uri, @Nonnull Predicate predicate, @Nullable String errorMessage) throws IllegalArgumentException { - Preconditions.checkNotNull(uri, "uri"); - Preconditions.checkNotNull(predicate, "predicate"); + checkNotNull(uri, "uri"); + checkNotNull(predicate, "predicate"); try { URI u = new URI(uri); if (!predicate.test(u)) { From fe5c8a1ba1035c8df6749bf5cacb35c6926576a5 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Fri, 6 Jan 2023 19:43:58 +0800 Subject: [PATCH 3/8] Fix error --- .../org/apache/pulsar/common/util/URIPreconditions.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java index 1db2660a47bc6..f68ed5e41a430 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/URIPreconditions.java @@ -18,7 +18,7 @@ */ package org.apache.pulsar.common.util; -import static com.google.common.base.Preconditions.checkNotNull; +import static java.util.Objects.requireNonNull; import java.net.URI; import java.net.URISyntaxException; import java.util.function.Predicate; @@ -84,8 +84,8 @@ public static void checkURIIfPresent(@Nullable String uri, public static void checkURI(@Nonnull String uri, @Nonnull Predicate predicate, @Nullable String errorMessage) throws IllegalArgumentException { - checkNotNull(uri, "uri"); - checkNotNull(predicate, "predicate"); + requireNonNull(uri, "uri"); + requireNonNull(predicate, "predicate"); try { URI u = new URI(uri); if (!predicate.test(u)) { From 06782d80cb12470d19cdbc25603959d616f9b496 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Fri, 6 Jan 2023 22:05:55 +0800 Subject: [PATCH 4/8] Fix conflict --- .../zookeeper/ZooKeeperSessionExpireRecoveryTest.java | 9 ++++++--- .../java/org/apache/pulsar/io/AbstractPulsarE2ETest.java | 7 ++++++- .../pulsar/proxy/server/ProxyWithAuthorizationTest.java | 3 ++- 3 files changed, 14 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ZooKeeperSessionExpireRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ZooKeeperSessionExpireRecoveryTest.java index e29f6f4116ac9..f6ccee316edb4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ZooKeeperSessionExpireRecoveryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/zookeeper/ZooKeeperSessionExpireRecoveryTest.java @@ -52,7 +52,8 @@ protected void cleanup() throws Exception { */ @Test public void testSessionExpired() throws Exception { - admin.clusters().createCluster("my-cluster", ClusterData.builder().serviceUrl("test-url").build()); + admin.clusters().createCluster("my-cluster", ClusterData.builder() + .serviceUrl("http://test-url").build()); assertTrue(Sets.newHashSet(admin.clusters().getClusters()).contains("my-cluster")); @@ -64,12 +65,14 @@ public void testSessionExpired() throws Exception { assertTrue(Sets.newHashSet(admin.clusters().getClusters()).contains("my-cluster")); try { - admin.clusters().createCluster("my-cluster-2", ClusterData.builder().serviceUrl("test-url").build()); + admin.clusters().createCluster("my-cluster-2", ClusterData.builder() + .serviceUrl("http://test-url").build()); fail("Should have failed, because global zk is down"); } catch (PulsarAdminException e) { // Ok } - admin.clusters().createCluster("cluster-2", ClusterData.builder().serviceUrl("test-url").build()); + admin.clusters().createCluster("cluster-2", ClusterData.builder() + .serviceUrl("http://test-url").build()); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java index 6e5a61791755a..6adb0acab9423 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java @@ -186,7 +186,12 @@ public void setup(Method method) throws Exception { primaryHost = String.format("http://%s:%d", "localhost", pulsar.getListenPortHTTP().get()); // update cluster metadata - ClusterData clusterData = ClusterData.builder().serviceUrl(pulsar.getBrokerServiceUrlTls()).build(); + ClusterData clusterData = ClusterData.builder() + .serviceUrl(pulsar.getWebServiceAddress()) + .serviceUrlTls(pulsar.getWebServiceAddressTls()) + .brokerServiceUrl(pulsar.getBrokerServiceUrl()) + .brokerServiceUrlTls(pulsar.getBrokerServiceUrlTls()) + .build(); admin.clusters().updateCluster(config.getClusterName(), clusterData); ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl()); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java index 6d5381bdcd5c6..16039cb16978e 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java @@ -319,7 +319,8 @@ public void testTlsHostVerificationProxyToClient(boolean hostnameVerificationEna String namespaceName = "my-tenant/my-ns"; - admin.clusters().createCluster("proxy-authorization", ClusterData.builder().serviceUrl(brokerUrlTls.toString()).build()); + admin.clusters().createCluster("proxy-authorization", ClusterData.builder() + .serviceUrlTls(brokerUrlTls.toString()).build()); admin.tenants().createTenant("my-tenant", new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("proxy-authorization"))); From ad173d5382bf125ac78b21dc9d33d95c8f1ac051 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Fri, 6 Jan 2023 22:09:15 +0800 Subject: [PATCH 5/8] Fix test --- .../test/java/org/apache/pulsar/broker/admin/AdminTest.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java index 39d20990493e3..c293bbea841fd 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java @@ -444,9 +444,7 @@ public void clusters() throws Exception { try { asyncRequests(ctx -> clusters.createCluster(ctx, "auth", ClusterDataImpl.builder() .serviceUrl("http://dummy.web.example.com") - .serviceUrlTls("") - .brokerServiceUrl("http://dummy.messaging.example.com") - .brokerServiceUrlTls("") + .brokerServiceUrl("pulsar://dummy.messaging.example.com") .authenticationPlugin("authenticationPlugin") .authenticationParameters("authenticationParameters") .listenerName("listenerName") From a0846d836fc1d26c0b4b573728608f88b7a5a686 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 8 Jan 2023 11:59:36 +0800 Subject: [PATCH 6/8] Fix failed test --- .../java/org/apache/pulsar/broker/EmbeddedPulsarCluster.java | 2 +- .../test/java/org/apache/pulsar/broker/admin/AdminTest.java | 2 +- .../apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java | 3 ++- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/EmbeddedPulsarCluster.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/EmbeddedPulsarCluster.java index 1bb985e20b921..eac76a3a80ede 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/EmbeddedPulsarCluster.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/EmbeddedPulsarCluster.java @@ -78,7 +78,7 @@ private EmbeddedPulsarCluster(int numBrokers, int numBookies, String metadataSto .serviceHttpUrl(adminUrl) .build(); - admin.clusters().createCluster(CLUSTER_NAME, ClusterData.builder().serviceUrl(serviceUrl).build()); + admin.clusters().createCluster(CLUSTER_NAME, ClusterData.builder().brokerServiceUrl(serviceUrl).build()); admin.tenants().createTenant("public", TenantInfo.builder().allowedClusters(Collections.singleton(CLUSTER_NAME)).build()); admin.namespaces().createNamespace("public/default", Collections.singleton(CLUSTER_NAME)); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java index c293bbea841fd..5eb39a9c2832f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java @@ -773,7 +773,7 @@ public void resourceQuotas() throws Exception { TenantInfoImpl admin = TenantInfoImpl.builder() .allowedClusters(Collections.singleton(cluster)) .build(); - ClusterDataImpl clusterData = ClusterDataImpl.builder().serviceUrl(cluster).build(); + ClusterDataImpl clusterData = ClusterDataImpl.builder().serviceUrl("http://example.pulsar").build(); asyncRequests(ctx -> clusters.createCluster(ctx, cluster, clusterData )); asyncRequests(ctx -> properties.createTenant(ctx, property, admin)); diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java index 16039cb16978e..e2362478e4782 100644 --- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java +++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationTest.java @@ -411,7 +411,8 @@ public void tlsCiphersAndProtocols(Set tlsCiphers, Set tlsProtoc String namespaceName = "my-tenant/my-ns"; createAdminClient(); - admin.clusters().createCluster("proxy-authorization", ClusterData.builder().serviceUrl(brokerUrlTls.toString()).build()); + admin.clusters().createCluster("proxy-authorization", ClusterData.builder() + .serviceUrlTls(brokerUrlTls.toString()).build()); admin.tenants().createTenant("my-tenant", new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("proxy-authorization"))); From d892a74adc0c4354eb65c4b6c087b517769b44c8 Mon Sep 17 00:00:00 2001 From: mattisonchao Date: Sun, 8 Jan 2023 17:03:31 +0800 Subject: [PATCH 7/8] Continue fix illgeal usage --- .../org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java | 1 + .../java/org/apache/pulsar/broker/admin/AdminApi2Test.java | 2 +- .../client/api/MultiRolesTokenAuthorizationProviderTest.java | 3 +-- .../pulsar/functions/worker/PulsarFunctionPublishTest.java | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java index 42ef264b6db04..b77a7cad7e4ed 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java @@ -79,6 +79,7 @@ public static boolean isCGroupEnabled() { * @return Total cpu limit */ public static double getTotalCpuLimit(boolean isCGroupsEnabled) { + if (isCGroupsEnabled) { try { long quota = readLongFromFile(Paths.get(CGROUPS_CPU_LIMIT_QUOTA_PATH)); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index 11c84d990f68d..03622213b69eb 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -1941,7 +1941,7 @@ public void testUpdateClusterWithProxyUrl() throws Exception { // update cluster = ClusterData.builder() .serviceUrl(pulsar.getWebServiceAddress()) - .proxyServiceUrl("proxy") + .proxyServiceUrl("pulsar://example.com") .proxyProtocol(ProxyProtocol.SNI) .build(); admin.clusters().updateCluster(clusterName, cluster); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiRolesTokenAuthorizationProviderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiRolesTokenAuthorizationProviderTest.java index 02b58a8667dd1..0445ad27ca8e7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiRolesTokenAuthorizationProviderTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiRolesTokenAuthorizationProviderTest.java @@ -111,8 +111,7 @@ protected void setup() throws Exception { admin.clusters().createCluster(configClusterName, ClusterData.builder() - .brokerServiceUrl(brokerUrl.toString()) - .serviceUrl(getPulsar().getWebServiceAddress()) + .serviceUrl(brokerUrl.toString()) .build() ); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java index f8471de5a5f84..a5d45652888a8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java @@ -179,7 +179,7 @@ void setup(Method method) throws Exception { primaryHost = pulsar.getWebServiceAddress(); // update cluster metadata - ClusterData clusterData = ClusterData.builder().serviceUrl(urlTls.toString()).build(); + ClusterData clusterData = ClusterData.builder().serviceUrlTls(urlTls.toString()).build(); admin.clusters().updateCluster(config.getClusterName(), clusterData); ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl()); From 3e63d5494b46e2d9c7f1f3ef3f1edaddbd837346 Mon Sep 17 00:00:00 2001 From: Qiang Zhao Date: Mon, 9 Jan 2023 09:41:54 +0800 Subject: [PATCH 8/8] Update pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java --- .../org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java index b77a7cad7e4ed..42ef264b6db04 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LinuxInfoUtils.java @@ -79,7 +79,6 @@ public static boolean isCGroupEnabled() { * @return Total cpu limit */ public static double getTotalCpuLimit(boolean isCGroupsEnabled) { - if (isCGroupsEnabled) { try { long quota = readLongFromFile(Paths.get(CGROUPS_CPU_LIMIT_QUOTA_PATH));