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/05/21 10:08:16 UTC
(pulsar) 04/09: Address review comment
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-support-webservice-bindaddresses-and-advertised-listeners
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 60e5a4cce161958e79f2bc23bb31a5677b69154a
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed May 20 19:31:22 2026 +0300
Address review comment
---
.../apache/pulsar/broker/lookup/LookupResult.java | 27 ++++++++++++++--------
.../broker/namespace/NamespaceEphemeralData.java | 6 +++++
.../pulsar/broker/namespace/NamespaceService.java | 17 +++++++-------
.../pulsar/broker/web/RestProducerContext.java | 2 +-
.../pulsar/broker/lookup/LookupResultTest.java | 6 ++++-
5 files changed, 38 insertions(+), 20 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/LookupResult.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/LookupResult.java
index 3d46ef16cda..c45614a8be3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/LookupResult.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/LookupResult.java
@@ -22,6 +22,8 @@ import static org.apache.pulsar.broker.lookup.v2.TopicLookup.LISTENERNAME_PARAM;
import java.net.URI;
import java.util.Map;
import java.util.Objects;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriBuilder;
import lombok.Builder;
import lombok.Getter;
@@ -176,14 +178,14 @@ public class LookupResult {
}
}
- // for backwards compatibility, parse the brokerId from webServiceUrl or webServiceUrlTls
- // this might be the case temporarily when the broker upgrade happens and there are mixed versions
- // of brokers in the cluster
+ // for backwards compatibility, derive the brokerId from webServiceUrl or webServiceUrlTls by
+ // parsing the URL and taking host:port. This is a transient state that may occur during a
+ // rolling upgrade when older brokers in the cluster do not yet publish a brokerId in their
+ // ephemeral data; once all brokers have been upgraded, the brokerId field is always populated.
if (brokerId == null && (webServiceUrl != null || webServiceUrlTls != null)) {
- if (webServiceUrl != null) {
- brokerId = webServiceUrl.substring("http://".length());
- } else {
- brokerId = webServiceUrlTls.substring("https://".length());
+ URI url = URI.create(webServiceUrl != null ? webServiceUrl : webServiceUrlTls);
+ if (url.getHost() != null && url.getPort() != -1) {
+ brokerId = url.getHost() + ":" + url.getPort();
}
}
@@ -246,9 +248,14 @@ public class LookupResult {
private URI toRedirectUriInternal(URI requestUri, boolean authoritativeRedirect) {
boolean requireHttps = "https".equalsIgnoreCase(requestUri.getScheme());
String webServiceUrl = requireHttps ? lookupData.getHttpUrlTls() : lookupData.getHttpUrl();
- Objects.requireNonNull(webServiceUrl, () -> "No "
- + (requireHttps ? "https" : "http")
- + " URL configured for broker " + lookupData.getBrokerId());
+ if (webServiceUrl == null) {
+ // Preserve the legacy 412 error semantics when the redirect target broker has no URL
+ // configured for the requested scheme.
+ throw new WebApplicationException(Response.status(Response.Status.PRECONDITION_FAILED)
+ .entity("No " + (requireHttps ? "https" : "http")
+ + " URL configured for broker " + lookupData.getBrokerId())
+ .build());
+ }
URI webServiceUri = URI.create(webServiceUrl);
UriBuilder uriBuilder =
UriBuilder.fromUri(requestUri) // use the path and query parameters from the request URI
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceEphemeralData.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceEphemeralData.java
index 4b4cb53ff0a..a580eb1314b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceEphemeralData.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceEphemeralData.java
@@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import lombok.Data;
+import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import org.apache.pulsar.policies.data.loadbalancer.AdvertisedListener;
import org.jspecify.annotations.NonNull;
@@ -29,6 +30,11 @@ import org.jspecify.annotations.NonNull;
@Data
@NoArgsConstructor
public class NamespaceEphemeralData {
+ // The brokerId field was added later. During a rolling upgrade, entries written by older brokers may not
+ // contain it (LookupResult derives it from the URL host:port in that case). Exclude it from equals/hashCode
+ // so that consumers of this object do not see spurious "changed" notifications when the same logical owner
+ // transitions from a missing brokerId to a populated one.
+ @EqualsAndHashCode.Exclude
private String brokerId;
private String nativeUrl;
private String nativeUrlTls;
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 b1d97a6a9cc..db6ce90c203 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
@@ -328,11 +328,14 @@ public class NamespaceService implements AutoCloseable {
}
/**
- * Return the LookupResult of the broker who's owning a particular service unit.
- * The LookupResult won't necessary be the broker who is owning the service unit. When the cluster contains
- * multiple brokers with different load manager implementations, the LookupResult will be
- * <p>
- * If the service unit is not owned, return an empty optional
+ * Return the LookupResult of the broker that owns a particular service unit.
+ *
+ * <p>The returned LookupResult will not necessarily point to the broker that currently owns the service unit.
+ * When the cluster contains multiple brokers with different load manager implementations (e.g. during a
+ * load-manager migration), the LookupResult may point to a broker running the expected load manager, so the
+ * caller redirects the request to a broker that can handle it.
+ *
+ * <p>If the service unit is not owned, return an empty optional.
*/
public Optional<LookupResult> getLookupResultForWebRequest(ServiceUnitId suName, LookupOptions options)
throws Exception {
@@ -573,9 +576,7 @@ public class NamespaceService implements AutoCloseable {
.get(config.getMetadataStoreOperationTimeoutSeconds(), SECONDS);
if (candidateBroker == null) {
- Optional<LeaderBroker> currentLeader = pulsar.getLeaderElectionService().getCurrentLeader()
- // make a copy to avoid races, lookups will tolerate stale leader information
- .map(Function.identity());
+ Optional<LeaderBroker> currentLeader = pulsar.getLeaderElectionService().getCurrentLeader();
if (options.isAuthoritative()) {
// leader broker already assigned the current broker as owner
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestProducerContext.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestProducerContext.java
index 6b9192845ab..488a8be2110 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestProducerContext.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/RestProducerContext.java
@@ -52,7 +52,7 @@ public class RestProducerContext implements TopicEventsListener {
public void handleEvent(String topicName, TopicEvent event, EventStage stage, Throwable t) {
// remove topic from owning topics when it's unloaded
if (event == TopicEvent.UNLOAD && stage == EventStage.SUCCESS) {
- addOrRemoveTopic(topicName, event == TopicEvent.UNLOAD);
+ addOrRemoveTopic(topicName, true);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/LookupResultTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/LookupResultTest.java
index 553186e24eb..917bd1a3128 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/LookupResultTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/LookupResultTest.java
@@ -25,10 +25,13 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
import java.net.URI;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Response;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
import org.apache.pulsar.broker.namespace.LookupOptions;
import org.apache.pulsar.broker.namespace.NamespaceEphemeralData;
@@ -370,7 +373,8 @@ public class LookupResultTest {
URI request = URI.create("https://original-host:1234/admin");
- assertThrows(NullPointerException.class, () -> result.toRedirectUri(request));
+ WebApplicationException e = expectThrows(WebApplicationException.class, () -> result.toRedirectUri(request));
+ assertEquals(e.getResponse().getStatus(), Response.Status.PRECONDITION_FAILED.getStatusCode());
}
@Test