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 2024/11/27 22:57:02 UTC
(pulsar) 02/04: [fix][broker] fix null lookup result when brokers are starting (#23642)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a6b93a646fb58632ec657c09a852953a67bb777e
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Nov 28 04:38:15 2024 +0800
[fix][broker] fix null lookup result when brokers are starting (#23642)
(cherry picked from commit bd3b3b82abc78db6641b8917e8a2d2c947459ea7)
---
.../java/org/apache/pulsar/broker/PulsarService.java | 6 +++---
.../apache/pulsar/broker/lookup/TopicLookupBase.java | 19 +++++++++++++++++++
.../pulsar/broker/namespace/OwnershipCache.java | 8 +++++++-
3 files changed, 29 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 4da5e114a08..da57319a441 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -852,6 +852,9 @@ public class PulsarService implements AutoCloseable, ShutdownService {
this.webSocketService.setLocalCluster(clusterData);
}
+ // Initialize namespace service, after service url assigned. Should init zk and refresh self owner info.
+ this.nsService.initialize();
+
// Start the leader election service
startLeaderElectionService();
@@ -863,9 +866,6 @@ public class PulsarService implements AutoCloseable, ShutdownService {
// (namespace service depends on load manager)
this.startLoadManagementService();
- // Initialize namespace service, after service url assigned. Should init zk and refresh self owner info.
- this.nsService.initialize();
-
// Start topic level policies service
if (config.isTopicLevelPoliciesEnabled() && config.isSystemTopicEnabled()) {
this.topicPoliciesService = new SystemTopicBasedTopicPoliciesService(this);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
index 9a05c3d992a..adc65db380d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
@@ -314,6 +314,7 @@ public class TopicLookupBase extends PulsarWebResource {
}
LookupData lookupData = lookupResult.get().getLookupData();
+ printWarnLogIfLookupResUnexpected(topicName, lookupData, options, pulsarService);
if (lookupResult.get().isRedirect()) {
boolean newAuthoritative = lookupResult.get().isAuthoritativeRedirect();
lookupfuture.complete(
@@ -338,6 +339,24 @@ public class TopicLookupBase extends PulsarWebResource {
return lookupfuture;
}
+ /**
+ * Check if a internal client will get a null lookup result.
+ */
+ private static void printWarnLogIfLookupResUnexpected(TopicName topic, LookupData lookupData, LookupOptions options,
+ PulsarService pulsar) {
+ if (!pulsar.getBrokerService().isSystemTopic(topic)) {
+ return;
+ }
+ boolean tlsEnabled = pulsar.getConfig().isBrokerClientTlsEnabled();
+ if (!tlsEnabled && StringUtils.isBlank(lookupData.getBrokerUrl())) {
+ log.warn("[{}] Unexpected lookup result: brokerUrl is required when TLS isn't enabled. options: {},"
+ + " result {}", topic, options, lookupData);
+ } else if (tlsEnabled && StringUtils.isBlank(lookupData.getBrokerUrlTls())) {
+ log.warn("[{}] Unexpected lookup result: brokerUrlTls is required when TLS is enabled. options: {},"
+ + " result {}", topic, options, lookupData);
+ }
+ }
+
private static void handleLookupError(CompletableFuture<ByteBuf> lookupFuture, String topicName, String clientAppId,
long requestId, Throwable ex){
Throwable unwrapEx = FutureUtil.unwrapCompletionException(ex);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
index 5833531f8de..3f0483f58ad 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
@@ -73,7 +73,7 @@ public class OwnershipCache {
/**
* The NamespaceEphemeralData objects that can be associated with the current owner, when the broker is disabled.
*/
- private final NamespaceEphemeralData selfOwnerInfoDisabled;
+ private NamespaceEphemeralData selfOwnerInfoDisabled;
private final LockManager<NamespaceEphemeralData> lockManager;
@@ -121,6 +121,9 @@ public class OwnershipCache {
this.pulsar = pulsar;
this.ownerBrokerUrl = pulsar.getBrokerServiceUrl();
this.ownerBrokerUrlTls = pulsar.getBrokerServiceUrlTls();
+ // At this moment, the variables "webServiceAddress" and "webServiceAddressTls" and so on have not been
+ // initialized, so we will get an empty "selfOwnerInfo" and an empty "selfOwnerInfoDisabled" here.
+ // But do not worry, these two fields will be set by the method "refreshSelfOwnerInfo" soon.
this.selfOwnerInfo = new NamespaceEphemeralData(ownerBrokerUrl, ownerBrokerUrlTls,
pulsar.getWebServiceAddress(), pulsar.getWebServiceAddressTls(),
false, pulsar.getAdvertisedListeners());
@@ -353,6 +356,9 @@ public class OwnershipCache {
this.selfOwnerInfo = new NamespaceEphemeralData(pulsar.getBrokerServiceUrl(),
pulsar.getBrokerServiceUrlTls(), pulsar.getWebServiceAddress(),
pulsar.getWebServiceAddressTls(), false, pulsar.getAdvertisedListeners());
+ this.selfOwnerInfoDisabled = new NamespaceEphemeralData(pulsar.getBrokerServiceUrl(),
+ pulsar.getBrokerServiceUrlTls(), pulsar.getWebServiceAddress(),
+ pulsar.getWebServiceAddressTls(), true, pulsar.getAdvertisedListeners());
return selfOwnerInfo.getNativeUrl() != null || selfOwnerInfo.getNativeUrlTls() != null;
}
}