You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/16 22:31:32 UTC
(pulsar) 09/24: [fix][broker] Match namespace isolation policy brokers given as host:port (#26829)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8dd562f1d4e3bfeb0d6b9fcc7c8c6c77a41811ad
Author: Roman Melnyk <ay...@users.noreply.github.com>
AuthorDate: Thu Oct 15 07:52:51 2026 +0200
[fix][broker] Match namespace isolation policy brokers given as host:port (#26829)
(cherry picked from commit 25fe96c42eea6ccc5e5a38bad886a62e86710b51)
---
.../pulsar/broker/admin/impl/ClustersBase.java | 10 ++-
.../broker/loadbalance/impl/LoadManagerShared.java | 31 +++++--
.../impl/SimpleResourceAllocationPolicies.java | 16 ++++
.../pulsar/broker/namespace/NamespaceService.java | 6 +-
.../apache/pulsar/broker/admin/AdminApi2Test.java | 57 +++++++++++++
.../loadbalance/impl/LoadManagerSharedTest.java | 95 ++++++++++++++++++++++
.../admin/cli/CmdNamespaceIsolationPolicy.java | 8 +-
.../common/policies/NamespaceIsolationPolicy.java | 26 ++++++
.../policies/impl/NamespaceIsolationPolicies.java | 17 ++++
.../impl/NamespaceIsolationPolicyImpl.java | 67 ++++++++++++++-
.../impl/NamespaceIsolationPolicyImplTest.java | 83 +++++++++++++++++++
11 files changed, 403 insertions(+), 13 deletions(-)
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 0ae18fc2e54f..bc977569d37e 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
@@ -648,11 +648,17 @@ public class ClustersBase extends AdminResource {
return brokerIsolationData.build();
}
List<String> namespaceRegexes = new ArrayList<>();
+ // the broker is given as a broker id (host:port), as listed by `pulsar-admin brokers list`;
+ // match host entries against the host and port-specific entries against the broker id, like placement does
+ int portIdx = broker.lastIndexOf(':');
+ String brokerHost = portIdx > 0 ? broker.substring(0, portIdx) : broker;
policies.forEach((name, policyData) -> {
NamespaceIsolationPolicyImpl nsPolicyImpl = new NamespaceIsolationPolicyImpl(policyData);
- if (nsPolicyImpl.isPrimaryBroker(broker) || nsPolicyImpl.isSecondaryBroker(broker)) {
+ boolean primary = nsPolicyImpl.isPrimaryBroker(brokerHost) || nsPolicyImpl.isPrimaryBrokerId(broker);
+ boolean secondary = nsPolicyImpl.isSecondaryBroker(brokerHost) || nsPolicyImpl.isSecondaryBrokerId(broker);
+ if (primary || secondary) {
namespaceRegexes.addAll(policyData.getNamespaces());
- brokerIsolationData.primary(nsPolicyImpl.isPrimaryBroker(broker));
+ brokerIsolationData.primary(primary);
brokerIsolationData.policyName(name);
}
});
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerShared.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerShared.java
index 591b061253d3..7bfc93aa7b0f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerShared.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerShared.java
@@ -113,13 +113,13 @@ public class LoadManagerShared {
// todo: in future check if the resource unit has resources to take the namespace
if (isIsolationPoliciesPresent) {
// note: serviceUnitID is namespace name and ResourceID is brokerName
- if (policies.isPrimaryBroker(namespace, brokerHost)) {
+ if (isPrimaryBroker(policies, namespace, brokerHost, brokerId)) {
primariesCache.add(brokerId);
if (LOG.isDebugEnabled()) {
LOG.debug("Added Primary Broker - [{}] as possible Candidates for"
+ " namespace - [{}] with policies", brokerHost, namespace.toString());
}
- } else if (policies.isSecondaryBroker(namespace, brokerHost)) {
+ } else if (isSecondaryBroker(policies, namespace, brokerHost, brokerId)) {
secondaryCache.add(brokerId);
if (LOG.isDebugEnabled()) {
LOG.debug(
@@ -147,7 +147,7 @@ public class LoadManagerShared {
LOG.debug("Filter broker- [{}] because broker only supports non-persistent namespace - [{}]",
brokerHost, namespace.toString());
}
- } else if (policies.isSharedBroker(brokerHost)) {
+ } else if (isSharedBroker(policies, brokerHost, brokerId)) {
secondaryCache.add(brokerId);
if (LOG.isDebugEnabled()) {
LOG.debug("Added Shared Broker - [{}] as possible Candidates for namespace - [{}]",
@@ -174,6 +174,25 @@ public class LoadManagerShared {
}
}
+ // Policy entries are matched against the broker host. Entries that contain a ':' may name a port
+ // (host:port, as brokers are listed by `pulsar-admin brokers list`) and are also matched against the broker id.
+ private static boolean isPrimaryBroker(SimpleResourceAllocationPolicies policies, NamespaceName namespace,
+ String brokerHost, String brokerId) {
+ return policies.isPrimaryBroker(namespace, brokerHost) || policies.isPrimaryBrokerId(namespace, brokerId);
+ }
+
+ private static boolean isSecondaryBroker(SimpleResourceAllocationPolicies policies, NamespaceName namespace,
+ String brokerHost, String brokerId) {
+ return policies.isSecondaryBroker(namespace, brokerHost)
+ || policies.isSecondaryBrokerId(namespace, brokerId);
+ }
+
+ // A broker reserved as primary by a policy entry with a port is not part of the shared broker pool.
+ private static boolean isSharedBroker(SimpleResourceAllocationPolicies policies, String brokerHost,
+ String brokerId) {
+ return policies.isSharedBroker(brokerHost) && !policies.isPrimaryBrokerIdInAnyPolicy(brokerId);
+ }
+
private static String parseBrokerHost(String brokerId) {
// use last index to support ipv6 addresses
int lastIdx = brokerId.lastIndexOf(':');
@@ -213,13 +232,13 @@ public class LoadManagerShared {
// todo: in future check if the resource unit has resources to take the namespace
if (isIsolationPoliciesPresent) {
// note: serviceUnitID is namespace name and ResourceID is brokerName
- if (policies.isPrimaryBroker(namespace, brokerHost)) {
+ if (isPrimaryBroker(policies, namespace, brokerHost, brokerId)) {
primariesCache.add(brokerId);
if (LOG.isDebugEnabled()) {
LOG.debug("Added Primary Broker - [{}] as possible Candidates for"
+ " namespace - [{}] with policies", brokerHost, namespace.toString());
}
- } else if (policies.isSecondaryBroker(namespace, brokerHost)) {
+ } else if (isSecondaryBroker(policies, namespace, brokerHost, brokerId)) {
secondaryCache.add(brokerId);
if (LOG.isDebugEnabled()) {
LOG.debug(
@@ -248,7 +267,7 @@ public class LoadManagerShared {
LOG.debug("Filter broker- [{}] because broker only supports non-persistent "
+ "namespace - [{}]", brokerId, namespace.toString());
}
- } else if (policies.isSharedBroker(brokerHost)) {
+ } else if (isSharedBroker(policies, brokerHost, brokerId)) {
secondaryCache.add(brokerId);
if (LOG.isDebugEnabled()) {
LOG.debug("Added Shared Broker - [{}] as possible Candidates for namespace - [{}]",
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleResourceAllocationPolicies.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleResourceAllocationPolicies.java
index 25fbc152b116..6742c931ced8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleResourceAllocationPolicies.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleResourceAllocationPolicies.java
@@ -102,6 +102,16 @@ public class SimpleResourceAllocationPolicies {
return nsPolicy.isPresent() && nsPolicy.get().isSecondaryBroker(broker);
}
+ public boolean isPrimaryBrokerId(NamespaceName namespace, String brokerId) {
+ Optional<NamespaceIsolationPolicy> nsPolicy = getNamespaceIsolationPolicy(namespace);
+ return nsPolicy.isPresent() && nsPolicy.get().isPrimaryBrokerId(brokerId);
+ }
+
+ public boolean isSecondaryBrokerId(NamespaceName namespace, String brokerId) {
+ Optional<NamespaceIsolationPolicy> nsPolicy = getNamespaceIsolationPolicy(namespace);
+ return nsPolicy.isPresent() && nsPolicy.get().isSecondaryBrokerId(brokerId);
+ }
+
public boolean isSharedBroker(String broker) {
try {
Optional<NamespaceIsolationPolicies> policies =
@@ -114,6 +124,12 @@ public class SimpleResourceAllocationPolicies {
return false;
}
+ public boolean isPrimaryBrokerIdInAnyPolicy(String brokerId) {
+ return getIsolationPolicies(pulsar.getConfiguration().getClusterName())
+ .map(isolationPolicies -> isolationPolicies.isPrimaryBrokerIdInAnyPolicy(brokerId))
+ .orElse(false);
+ }
+
public boolean shouldFailoverToSecondaries(NamespaceName namespace, int totalPrimaryCandidates) {
Optional<NamespaceIsolationPolicy> nsPolicy = getNamespaceIsolationPolicy(namespace);
return nsPolicy.isPresent() && nsPolicy.get().shouldFailover(totalPrimaryCandidates);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index 14cf7233c405..caef15d803c7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -946,9 +946,11 @@ public class NamespaceService implements AutoCloseable {
}
// found corresponding policy, set the status to controlled
nsOwnedStatus.is_controlled = true;
- if (nsIsolationPolicy.isPrimaryBroker(pulsar.getAdvertisedAddress())) {
+ if (nsIsolationPolicy.isPrimaryBroker(pulsar.getAdvertisedAddress())
+ || nsIsolationPolicy.isPrimaryBrokerId(pulsar.getBrokerId())) {
nsOwnedStatus.broker_assignment = BrokerAssignment.primary;
- } else if (nsIsolationPolicy.isSecondaryBroker(pulsar.getAdvertisedAddress())) {
+ } else if (nsIsolationPolicy.isSecondaryBroker(pulsar.getAdvertisedAddress())
+ || nsIsolationPolicy.isSecondaryBrokerId(pulsar.getBrokerId())) {
nsOwnedStatus.broker_assignment = BrokerAssignment.secondary;
}
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 c5b67075ea36..8e5eb4dc8628 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
@@ -63,6 +63,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
import javax.ws.rs.NotAcceptableException;
import javax.ws.rs.core.Response.Status;
import lombok.AllArgsConstructor;
@@ -126,6 +127,7 @@ import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
import org.apache.pulsar.common.policies.data.BacklogQuota;
+import org.apache.pulsar.common.policies.data.BrokerAssignment;
import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationData;
import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationDataImpl;
import org.apache.pulsar.common.policies.data.BundlesData;
@@ -136,6 +138,7 @@ import org.apache.pulsar.common.policies.data.EntryFilters;
import org.apache.pulsar.common.policies.data.FailureDomain;
import org.apache.pulsar.common.policies.data.NamespaceIsolationData;
import org.apache.pulsar.common.policies.data.NamespaceIsolationPolicyUnloadScope;
+import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus;
import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
import org.apache.pulsar.common.policies.data.OffloadPolicies;
import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
@@ -1457,6 +1460,60 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
admin.clusters().deleteNamespaceIsolationPolicy(cluster, policyName1);
}
+ @Test
+ public void brokerNamespaceIsolationPoliciesWithBrokerId() throws Exception {
+ String cluster = pulsar.getConfiguration().getClusterName();
+ String brokerHost = pulsar.getAdvertisedAddress();
+ String brokerId = pulsar.getBrokerId();
+ String namespace = defaultTenant + "/test_ns_iso_broker_id_" + System.currentTimeMillis();
+ String policyName = "policy-broker-id";
+
+ // a primary entry that names the broker id (host:port), as listed by `pulsar-admin brokers list`
+ NamespaceIsolationData policyData = NamespaceIsolationData.builder()
+ .namespaces(Collections.singletonList(namespace))
+ .primary(new ArrayList<>(List.of(brokerId)))
+ .autoFailoverPolicy(AutoFailoverPolicyData.builder()
+ .policyType(AutoFailoverPolicyType.min_available)
+ .parameters(Map.of("min_limit", "1", "usage_threshold", "100"))
+ .build())
+ .build();
+ admin.clusters().createNamespaceIsolationPolicy(cluster, policyName, policyData);
+ try {
+ BrokerNamespaceIsolationDataImpl brokerIsolationData = (BrokerNamespaceIsolationDataImpl) admin.clusters()
+ .getBrokerWithNamespaceIsolationPolicy(cluster, brokerId);
+ assertTrue(brokerIsolationData.isPrimary());
+ assertEquals(brokerIsolationData.getNamespaceRegex(), List.of(namespace));
+ List<BrokerNamespaceIsolationData> brokerIsolationDataList =
+ admin.clusters().getBrokersWithNamespaceIsolationPolicy(cluster);
+ assertEquals(brokerIsolationDataList.size(), 1);
+ assertTrue(brokerIsolationDataList.get(0).isPrimary());
+
+ // the namespace is placed on the broker, and its ownership status is primary
+ admin.namespaces().createNamespace(namespace, Set.of(cluster));
+ assertTrue(admin.lookups().lookupTopic(namespace + "/topic1").contains(brokerHost));
+ Map<String, NamespaceOwnershipStatus> ownedBundles =
+ admin.brokers().getOwnedNamespaces(cluster, brokerId).entrySet().stream()
+ .filter(e -> e.getKey().startsWith(namespace + "/"))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ assertFalse(ownedBundles.isEmpty());
+ ownedBundles.values().forEach(status -> {
+ assertTrue(status.is_controlled);
+ assertEquals(status.broker_assignment, BrokerAssignment.primary);
+ });
+
+ // a host-only primary entry is reported for the broker id as well
+ policyData.getPrimary().clear();
+ policyData.getPrimary().add(brokerHost);
+ admin.clusters().updateNamespaceIsolationPolicy(cluster, policyName, policyData);
+ brokerIsolationData = (BrokerNamespaceIsolationDataImpl) admin.clusters()
+ .getBrokerWithNamespaceIsolationPolicy(cluster, brokerId);
+ assertTrue(brokerIsolationData.isPrimary());
+ assertEquals(brokerIsolationData.getNamespaceRegex(), List.of(namespace));
+ } finally {
+ admin.clusters().deleteNamespaceIsolationPolicy(cluster, policyName);
+ }
+ }
+
// create 1 namespace:
// 0. without isolation policy configured, lookup will success.
// 1. with matched isolation broker configured and matched, lookup will success.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerSharedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerSharedTest.java
index 465e8e2d8524..4219f0ab66dd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerSharedTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/LoadManagerSharedTest.java
@@ -18,9 +18,24 @@
*/
package org.apache.pulsar.broker.loadbalance.impl;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import com.google.common.collect.Sets;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.resources.NamespaceResources.IsolationPolicyResources;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
+import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
+import org.apache.pulsar.common.policies.data.NamespaceIsolationDataImpl;
+import org.apache.pulsar.common.policies.impl.NamespaceIsolationPolicies;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -73,4 +88,84 @@ public class LoadManagerSharedTest {
Assert.assertTrue(candidates.contains("broker1"));
Assert.assertTrue(candidates.contains("broker3"));
}
+
+ private static final LoadManagerShared.BrokerTopicLoadingPredicate ALL_TOPICS =
+ new LoadManagerShared.BrokerTopicLoadingPredicate() {
+ @Override
+ public boolean isEnablePersistentTopics(String brokerId) {
+ return true;
+ }
+
+ @Override
+ public boolean isEnableNonPersistentTopics(String brokerId) {
+ return true;
+ }
+ };
+
+ private static SimpleResourceAllocationPolicies policiesWithPrimary(NamespaceName namespace,
+ String... primary) throws Exception {
+ NamespaceIsolationDataImpl policyData = NamespaceIsolationDataImpl.builder()
+ .namespaces(List.of(namespace.toString()))
+ .primary(List.of(primary))
+ .secondary(List.of())
+ .autoFailoverPolicy(AutoFailoverPolicyData.builder()
+ .policyType(AutoFailoverPolicyType.min_available)
+ .parameters(Map.of("min_limit", "1", "usage_threshold", "100"))
+ .build())
+ .build();
+ Optional<NamespaceIsolationPolicies> policies =
+ Optional.of(new NamespaceIsolationPolicies(Map.of("policy1", policyData)));
+ PulsarService pulsar = mock(PulsarService.class, RETURNS_DEEP_STUBS);
+ when(pulsar.getConfiguration().getClusterName()).thenReturn("cluster1");
+ IsolationPolicyResources isolationPolicyResources =
+ pulsar.getPulsarResources().getNamespaceResources().getIsolationPolicies();
+ doReturn(policies).when(isolationPolicyResources).getIsolationDataPolicies("cluster1");
+ doReturn(CompletableFuture.completedFuture(policies))
+ .when(isolationPolicyResources).getIsolationDataPoliciesAsync("cluster1");
+ return new SimpleResourceAllocationPolicies(pulsar);
+ }
+
+ private static void assertPlacement(SimpleResourceAllocationPolicies policies, NamespaceName namespace,
+ Set<String> availableBrokers, Set<String> expected) throws Exception {
+ Set<String> candidates = new HashSet<>();
+ LoadManagerShared.applyNamespacePolicies(namespace, policies, candidates, availableBrokers, ALL_TOPICS);
+ Assert.assertEquals(candidates, expected);
+ Assert.assertEquals(LoadManagerShared.applyNamespacePoliciesAsync(namespace, policies, availableBrokers,
+ ALL_TOPICS).get(), expected);
+ }
+
+ @Test
+ public void testApplyNamespacePoliciesWithBrokerIdPolicy() throws Exception {
+ NamespaceName namespace = NamespaceName.get("tenant1/ns1");
+ assertPlacement(policiesWithPrimary(namespace, "broker1:8080"), namespace,
+ Set.of("broker1:8080", "broker1:8081", "broker2:8080"), Set.of("broker1:8080"));
+ assertPlacement(policiesWithPrimary(namespace, "broker1:8080|broker2:8080"), namespace,
+ Set.of("broker1:8080", "broker1:8081", "broker2:8080"), Set.of("broker1:8080", "broker2:8080"));
+ assertPlacement(policiesWithPrimary(namespace, "broker1:80[0-9]{2}"), namespace,
+ Set.of("broker1:8080", "broker1:9090", "broker2:8080"), Set.of("broker1:8080"));
+ // IPv6 broker ids are matched in the bracketed form
+ assertPlacement(policiesWithPrimary(namespace, "\\[2001:db8::1\\]:8080"), namespace,
+ Set.of("2001:db8::1:8080", "2001:db8::1:8081", "2001:db8::2:8080"), Set.of("2001:db8::1:8080"));
+ }
+
+ @Test
+ public void testApplyNamespacePoliciesKeepsHostRegexSemantics() throws Exception {
+ NamespaceName namespace = NamespaceName.get("tenant1/ns1");
+ // must not match broker2 through the ":8081" port suffix of its broker id
+ assertPlacement(policiesWithPrimary(namespace, "broker.*[13]"), namespace,
+ Set.of("broker1:8080", "broker2:8081", "broker3:8080"), Set.of("broker1:8080", "broker3:8080"));
+ // the last segment of an IPv6 host must not be taken for a port
+ assertPlacement(policiesWithPrimary(namespace, "2001:db8::1"), namespace,
+ Set.of("2001:db8::1:1234:8080", "2001:db8::1:8080"), Set.of("2001:db8::1:8080"));
+ }
+
+ @Test
+ public void testApplyNamespacePoliciesExcludesBrokerIdPrimaryFromSharedPool() throws Exception {
+ SimpleResourceAllocationPolicies policies =
+ policiesWithPrimary(NamespaceName.get("tenant1/isolated"), "broker1:8080");
+ // a namespace without a policy must not be placed on a broker reserved by a port-specific entry
+ NamespaceName uncontrolled = NamespaceName.get("tenant1/ns1");
+ assertPlacement(policies, uncontrolled, Set.of("broker1:8080", "broker1:8081", "broker2:8080"),
+ Set.of("broker1:8081", "broker2:8080"));
+ }
}
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaceIsolationPolicy.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaceIsolationPolicy.java
index ef36eb417136..08e7f6b16524 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaceIsolationPolicy.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaceIsolationPolicy.java
@@ -58,10 +58,14 @@ public class CmdNamespaceIsolationPolicy extends CmdBase {
+ "a secondary regex to select desired brokers. If no broker matches the specified regex, you cannot "
+ "create a topic. If there are not enough primary brokers, topics are assigned to secondary brokers. "
+ "If there are not enough secondary brokers, topics are assigned to other brokers which do not have "
- + "any isolation policies.", required = true, split = ",")
+ + "any isolation policies. A regex is matched against the broker host name. A regex that contains "
+ + "':' is also matched against host:port, as brokers are listed by `pulsar-admin brokers list` "
+ + "(e.g. broker1:8080). For an IPv6 broker, put the address in escaped brackets "
+ + "(e.g. \\[2001:db8::1\\]:8080).", required = true, split = ",")
private List<String> primary;
- @Option(names = "--secondary", description = "comma separated secondary-broker-regex list",
+ @Option(names = "--secondary", description = "comma separated secondary-broker-regex list. "
+ + "Regexes are matched in the same way as for --primary.",
required = false, split = ",")
private List<String> secondary = new ArrayList<String>(); // optional
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/NamespaceIsolationPolicy.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/NamespaceIsolationPolicy.java
index 52480d91eefa..5cea85074641 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/NamespaceIsolationPolicy.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/NamespaceIsolationPolicy.java
@@ -107,6 +107,32 @@ public interface NamespaceIsolationPolicy {
*/
boolean isSecondaryBroker(String brokerAddress);
+ /**
+ * Check to see whether the broker with the given broker id ({@code host:port}) is a primary broker
+ * because of a policy entry that names a port. Entries that contain a {@code :} are matched against
+ * {@code host:port}, with an IPv6 host in brackets (e.g. {@code [2001:db8::1]:8080}). Entries without
+ * a {@code :} are matched against the broker host only, see {@link #isPrimaryBroker(String)}.
+ *
+ * @param brokerId the broker id, in the form {@code host:port}
+ * @return true if a primary broker entry matches the broker id
+ */
+ default boolean isPrimaryBrokerId(String brokerId) {
+ return false;
+ }
+
+ /**
+ * Check to see whether the broker with the given broker id ({@code host:port}) is a secondary broker
+ * because of a policy entry that names a port. Entries that contain a {@code :} are matched against
+ * {@code host:port}, with an IPv6 host in brackets (e.g. {@code [2001:db8::1]:8080}). Entries without
+ * a {@code :} are matched against the broker host only, see {@link #isSecondaryBroker(String)}.
+ *
+ * @param brokerId the broker id, in the form {@code host:port}
+ * @return true if a secondary broker entry matches the broker id
+ */
+ default boolean isSecondaryBrokerId(String brokerId) {
+ return false;
+ }
+
/**
* According to the namespace isolation policy, find the allowed available primary brokers.
*
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicies.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicies.java
index d5025ab0f6bb..bea7be736d03 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicies.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicies.java
@@ -129,6 +129,23 @@ public class NamespaceIsolationPolicies {
return true;
}
+ /**
+ * Check to see whether a broker id ({@code host:port}) is the primary broker of any policy because of an
+ * entry that names a port, see {@link NamespaceIsolationPolicyImpl#isPrimaryBrokerId(String)}. Such a
+ * broker is not part of the shared broker pool.
+ *
+ * @param brokerId the broker id, in the form {@code host:port}
+ * @return true if any policy has a primary broker entry with a port that matches the broker id
+ */
+ public boolean isPrimaryBrokerIdInAnyPolicy(String brokerId) {
+ for (NamespaceIsolationData policyData : this.policies.values()) {
+ if (new NamespaceIsolationPolicyImpl(policyData).isPrimaryBrokerId(brokerId)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Get the broker assignment based on the namespace name.
*
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicyImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicyImpl.java
index 440282f29cb3..f7acce685e94 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicyImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicyImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.common.policies.impl;
+import com.google.common.net.HostAndPort;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
@@ -54,7 +55,14 @@ public class NamespaceIsolationPolicyImpl implements NamespaceIsolationPolicy {
private List<URL> getMatchedBrokers(List<String> brkRegexList, List<URL> availableBrokers) {
List<URL> matchedBrokers = new ArrayList<URL>();
for (URL brokerUrl : availableBrokers) {
- if (this.matchesBrokerRegex(brkRegexList, brokerUrl.getHost())) {
+ // URL#getHost keeps the brackets around an IPv6 host; match the bare host and, as before, the
+ // bracketed one
+ String host = brokerUrl.getHost();
+ String bareHost = stripBrackets(host);
+ if (this.matchesBrokerRegex(brkRegexList, bareHost)
+ || (!bareHost.equals(host) && this.matchesBrokerRegex(brkRegexList, host))
+ || (brokerUrl.getPort() != -1 && matchesHostAndPortRegex(brkRegexList,
+ HostAndPort.fromParts(bareHost, brokerUrl.getPort()).toString()))) {
matchedBrokers.add(brokerUrl);
}
}
@@ -117,6 +125,53 @@ public class NamespaceIsolationPolicyImpl implements NamespaceIsolationPolicy {
return false;
}
+ /**
+ * Checks whether the broker id matches any of the given regexes that contains a {@code :}, i.e. that may
+ * name a port. Regexes without a {@code :} are only ever matched against the broker host, so this doesn't
+ * change how they are applied.
+ */
+ private static boolean matchesBrokerIdRegex(List<String> brkRegexList, String brokerId) {
+ String hostAndPort = toHostAndPort(brokerId);
+ return hostAndPort != null && matchesHostAndPortRegex(brkRegexList, hostAndPort);
+ }
+
+ private static boolean matchesHostAndPortRegex(List<String> brkRegexList, String hostAndPort) {
+ for (String brkRegex : brkRegexList) {
+ if (brkRegex.indexOf(':') >= 0 && hostAndPort.matches(brkRegex)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Converts a broker id ({@code host:port}, where an IPv6 host has no brackets) to the {@code host:port} form
+ * that policy entries are matched against, with an IPv6 host in brackets, e.g. {@code [2001:db8::1]:8080}.
+ * Returns {@code null} if the value isn't a valid broker id.
+ */
+ private static String toHostAndPort(String brokerId) {
+ if (brokerId == null) {
+ return null;
+ }
+ // the port follows the last ':', as in LoadManagerShared#parseBrokerHost
+ int idx = brokerId.lastIndexOf(':');
+ if (idx <= 0) {
+ return null;
+ }
+ try {
+ String host = stripBrackets(brokerId.substring(0, idx));
+ int port = Integer.parseInt(brokerId.substring(idx + 1));
+ return HostAndPort.fromParts(host, port).toString();
+ } catch (IllegalArgumentException e) {
+ return null;
+ }
+ }
+
+ private static String stripBrackets(String host) {
+ return host.length() > 1 && host.startsWith("[") && host.endsWith("]")
+ ? host.substring(1, host.length() - 1) : host;
+ }
+
@Override
public boolean isPrimaryBroker(String broker) {
return this.matchesBrokerRegex(this.primary, broker);
@@ -127,6 +182,16 @@ public class NamespaceIsolationPolicyImpl implements NamespaceIsolationPolicy {
return this.matchesBrokerRegex(this.secondary, broker);
}
+ @Override
+ public boolean isPrimaryBrokerId(String brokerId) {
+ return matchesBrokerIdRegex(this.primary, brokerId);
+ }
+
+ @Override
+ public boolean isSecondaryBrokerId(String brokerId) {
+ return matchesBrokerIdRegex(this.secondary, brokerId);
+ }
+
@Override
public int hashCode() {
return Objects.hash(namespaces, primary, secondary,
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicyImplTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicyImplTest.java
index 4c2f78888679..e406b477f3fb 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicyImplTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/impl/NamespaceIsolationPolicyImplTest.java
@@ -18,12 +18,14 @@
*/
package org.apache.pulsar.common.policies.impl;
+import static java.util.Arrays.asList;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
@@ -103,6 +105,87 @@ public class NamespaceIsolationPolicyImplTest {
assertFalse(defaultPolicy.isSecondaryBroker("broker-X.messaging.use.example.com"));
}
+ private static NamespaceIsolationPolicyImpl policy(List<String> primary, List<String> secondary) {
+ return new NamespaceIsolationPolicyImpl(NamespaceIsolationData.builder()
+ .namespaces(Collections.singletonList("pulsar/test.*"))
+ .primary(primary)
+ .secondary(secondary)
+ .autoFailoverPolicy(AutoFailoverPolicyData.builder()
+ .policyType(AutoFailoverPolicyType.min_available)
+ .parameters(ImmutableMap.of("min_limit", "1", "usage_threshold", "100"))
+ .build())
+ .build());
+ }
+
+ @Test
+ public void testIsPrimaryOrSecondaryBrokerId() throws Exception {
+ // entries with a port are matched against the broker id
+ NamespaceIsolationPolicyImpl portPolicy = policy(asList("broker1:8080"), asList("broker[2-3]:8080"));
+ assertTrue(portPolicy.isPrimaryBrokerId("broker1:8080"));
+ assertFalse(portPolicy.isPrimaryBrokerId("broker1:8081"));
+ assertFalse(portPolicy.isPrimaryBroker("broker1"));
+ assertTrue(portPolicy.isSecondaryBrokerId("broker3:8080"));
+ assertFalse(portPolicy.isSecondaryBrokerId("broker4:8080"));
+
+ // any entry that contains ':' may name a port
+ for (String entry : asList("^broker1:8080$", "broker1:8080|broker2:8080", "broker1:(8080|8081)",
+ "broker1:80[0-9]{2}", "broker1:\\d+", "broker1:8080.*")) {
+ NamespaceIsolationPolicyImpl entryPolicy = policy(asList(entry), Collections.emptyList());
+ assertTrue(entryPolicy.isPrimaryBrokerId("broker1:8080"), entry);
+ assertFalse(entryPolicy.isPrimaryBrokerId("broker3:8080"), entry);
+ }
+
+ // entries without ':' are never matched against the broker id
+ NamespaceIsolationPolicyImpl hostPolicy = this.getDefaultPolicy();
+ assertTrue(hostPolicy.isPrimaryBroker("prod1-broker2.messaging.use.example.com"));
+ assertFalse(hostPolicy.isPrimaryBrokerId("prod1-broker2.messaging.use.example.com:8080"));
+ assertFalse(hostPolicy.isSecondaryBrokerId("prod1-broker5.messaging.use.example.com:8080"));
+
+ // a host regex must not match through the port of a broker id
+ NamespaceIsolationPolicyImpl suffixPolicy = policy(asList("broker.*[13]"), Collections.emptyList());
+ assertFalse(suffixPolicy.isPrimaryBroker("broker2"));
+ assertFalse(suffixPolicy.isPrimaryBrokerId("broker2:8081"));
+
+ // IPv6: the broker id is matched in its bracketed form, so the last address segment is never taken
+ // for a port
+ NamespaceIsolationPolicyImpl ipv6HostPolicy = policy(asList("2001:db8::1"), Collections.emptyList());
+ assertTrue(ipv6HostPolicy.isPrimaryBroker("2001:db8::1"));
+ assertFalse(ipv6HostPolicy.isPrimaryBroker("2001:db8::1:1234"));
+ assertFalse(ipv6HostPolicy.isPrimaryBrokerId("2001:db8::1:1234:8080"));
+ NamespaceIsolationPolicyImpl ipv6PortPolicy = policy(asList("\\[2001:db8::1\\]:8080"), Collections.emptyList());
+ assertTrue(ipv6PortPolicy.isPrimaryBrokerId("2001:db8::1:8080"));
+ assertFalse(ipv6PortPolicy.isPrimaryBrokerId("2001:db8::1:8081"));
+ assertFalse(ipv6PortPolicy.isPrimaryBrokerId("2001:db8::1:1234:8080"));
+
+ // values that aren't broker ids
+ assertFalse(portPolicy.isPrimaryBrokerId(null));
+ assertFalse(portPolicy.isPrimaryBrokerId("broker1"));
+ assertFalse(portPolicy.isPrimaryBrokerId("broker1:http"));
+ }
+
+ @Test
+ public void testFindBrokersWithPort() throws Exception {
+ NamespaceName ns = NamespaceName.get("pulsar/testns-1");
+ List<URL> brokers = asList(new URL("http://broker1:8080"), new URL("http://broker1:8081"),
+ new URL("http://broker2:8080"));
+ NamespaceIsolationPolicyImpl portPolicy = policy(asList("broker1:8080"), asList("broker[2-3]:8080"));
+ assertEquals(portPolicy.findPrimaryBrokers(brokers, ns), asList(new URL("http://broker1:8080")));
+ assertEquals(portPolicy.findSecondaryBrokers(brokers, ns), asList(new URL("http://broker2:8080")));
+
+ List<URL> ipv6Brokers = asList(new URL("http://[2001:db8::1]:8080"), new URL("http://[2001:db8::1]:8081"),
+ new URL("http://[2001:db8::2]:8080"));
+ // host-only IPv6 entry
+ assertEquals(policy(asList("2001:db8::1"), Collections.emptyList()).findPrimaryBrokers(ipv6Brokers, ns),
+ asList(new URL("http://[2001:db8::1]:8080"), new URL("http://[2001:db8::1]:8081")));
+ // IPv6 entry with a port
+ assertEquals(policy(asList("\\[2001:db8::1\\]:8080"), Collections.emptyList())
+ .findPrimaryBrokers(ipv6Brokers, ns),
+ asList(new URL("http://[2001:db8::1]:8080")));
+ // the bracketed host is still matched as before
+ assertEquals(policy(asList("\\[2001:db8::2\\]"), Collections.emptyList()).findPrimaryBrokers(ipv6Brokers, ns),
+ asList(new URL("http://[2001:db8::2]:8080")));
+ }
+
@Test
public void testFindBrokers() throws Exception {
NamespaceIsolationPolicyImpl defaultPolicy = this.getDefaultPolicy();