You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2025/06/27 01:34:02 UTC
(pulsar) branch branch-3.0 updated (2061f681444 -> 949a593738d)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 2061f681444 [fix][build] Fix pulsar-sql license
new 39500e06755 [improve][broker] Improve the log when namespace bundle is not available (#24434)
new 949a593738d [improve][broker] change to warn log level for ack validation error (#24459)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/pulsar/broker/namespace/NamespaceService.java | 13 ++++++++-----
.../org/apache/pulsar/broker/service/BrokerService.java | 12 ++++++------
.../java/org/apache/pulsar/broker/service/Consumer.java | 2 +-
.../service/PersistentDispatcherFailoverConsumerTest.java | 3 ++-
.../broker/service/PersistentTopicConcurrentTest.java | 3 ++-
.../apache/pulsar/broker/service/PersistentTopicTest.java | 3 ++-
.../org/apache/pulsar/broker/service/ServerCnxTest.java | 6 ++++--
7 files changed, 25 insertions(+), 17 deletions(-)
(pulsar) 01/02: [improve][broker] Improve the log when namespace bundle is not available (#24434)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 39500e06755bec8be5dbaf4d0be84b177fddb648
Author: Yunze Xu <xy...@163.com>
AuthorDate: Sat Jun 21 20:14:44 2025 +0800
[improve][broker] Improve the log when namespace bundle is not available (#24434)
(cherry picked from commit 19b4a05f888da35a584cf7ff6d594dd9df5e03d1)
---
.../apache/pulsar/broker/namespace/NamespaceService.java | 13 ++++++++-----
.../org/apache/pulsar/broker/service/BrokerService.java | 12 ++++++------
.../service/PersistentDispatcherFailoverConsumerTest.java | 3 ++-
.../broker/service/PersistentTopicConcurrentTest.java | 3 ++-
.../apache/pulsar/broker/service/PersistentTopicTest.java | 3 ++-
.../org/apache/pulsar/broker/service/ServerCnxTest.java | 6 ++++--
6 files changed, 24 insertions(+), 16 deletions(-)
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 e10134e4baa..1fc13f69891 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
@@ -1255,13 +1255,16 @@ public class NamespaceService implements AutoCloseable {
}
public CompletableFuture<Boolean> checkTopicOwnership(TopicName topicName) {
- // TODO: Add unit tests cover it.
+ return getBundleAsync(topicName).thenCompose(bundle -> checkBundleOwnership(topicName, bundle));
+ }
+
+ public CompletableFuture<Boolean> checkBundleOwnership(TopicName topicName, NamespaceBundle bundle) {
if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
- return getBundleAsync(topicName)
- .thenCompose(bundle -> loadManager.get().checkOwnershipAsync(Optional.of(topicName), bundle));
+ // TODO: Add unit tests cover it.
+ return loadManager.get().checkOwnershipAsync(Optional.of(topicName), bundle);
+ } else {
+ return ownershipCache.checkOwnershipAsync(bundle);
}
- return getBundleAsync(topicName)
- .thenCompose(ownershipCache::checkOwnershipAsync);
}
public CompletableFuture<Void> removeOwnedServiceUnitAsync(NamespaceBundle nsBundle) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 805f13247b6..685226779c3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -2320,19 +2320,19 @@ public class BrokerService implements Closeable {
public CompletableFuture<Void> checkTopicNsOwnership(final String topic) {
TopicName topicName = TopicName.get(topic);
+ final var namespaceService = pulsar.getNamespaceService();
- return pulsar.getNamespaceService().checkTopicOwnership(topicName)
- .thenCompose(ownedByThisInstance -> {
+ return namespaceService.getBundleAsync(topicName).thenCompose(bundle ->
+ namespaceService.checkBundleOwnership(topicName, bundle).thenCompose(ownedByThisInstance -> {
if (ownedByThisInstance) {
return CompletableFuture.completedFuture(null);
} else {
- String msg = String.format("Namespace bundle for topic (%s) not served by this instance:%s. "
- + "Please redo the lookup. Request is denied: namespace=%s",
- topic, pulsar.getBrokerId(), topicName.getNamespace());
+ String msg = String.format("Namespace bundle (%s) for topic (%s) not served by this instance:"
+ + "%s. Please redo the lookup.", bundle, topic, pulsar.getBrokerId());
log.warn(msg);
return FutureUtil.failedFuture(new ServiceUnitNotReadyException(msg));
}
- });
+ }));
}
public CompletableFuture<Integer> unloadServiceUnit(NamespaceBundle serviceUnit,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
index 2718323f1bb..f69546d9dc6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
@@ -161,7 +161,8 @@ public class PersistentDispatcherFailoverConsumerTest {
doReturn(nsSvc).when(pulsarTestContext.getPulsarService()).getNamespaceService();
doReturn(true).when(nsSvc).isServiceUnitOwned(any(NamespaceBundle.class));
doReturn(true).when(nsSvc).isServiceUnitActive(any(TopicName.class));
- doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkTopicOwnership(any(TopicName.class));
+ doReturn(CompletableFuture.completedFuture(mock(NamespaceBundle.class))).when(nsSvc).getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkBundleOwnership(any(), any());
setupMLAsyncCallbackMocks();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
index 6ce6de1916f..57390681522 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicConcurrentTest.java
@@ -105,7 +105,8 @@ public class PersistentTopicConcurrentTest extends MockedBookKeeperTestCase {
doReturn(nsSvc).when(pulsar).getNamespaceService();
doReturn(true).when(nsSvc).isServiceUnitOwned(any(NamespaceBundle.class));
doReturn(true).when(nsSvc).isServiceUnitActive(any(TopicName.class));
- doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkTopicOwnership(any(TopicName.class));
+ doReturn(CompletableFuture.completedFuture(mock(NamespaceBundle.class))).when(nsSvc).getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkBundleOwnership(any(), any());
final List<Position> addedEntries = new ArrayList<>();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index aa07b56d569..1938567387a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -216,7 +216,8 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
doReturn(true).when(nsSvc).isServiceUnitOwned(any());
doReturn(true).when(nsSvc).isServiceUnitActive(any());
doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).isServiceUnitActiveAsync(any());
- doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkTopicOwnership(any());
+ doReturn(CompletableFuture.completedFuture(mock(NamespaceBundle.class))).when(nsSvc).getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(true)).when(nsSvc).checkBundleOwnership(any(), any());
setupMLAsyncCallbackMocks();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 59215ab2316..760a9d74df2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -142,6 +142,7 @@ import org.apache.pulsar.common.api.proto.ProtocolVersion;
import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.api.proto.Subscription;
import org.apache.pulsar.common.api.proto.TxnAction;
+import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
@@ -223,11 +224,12 @@ public class ServerCnxTest {
brokerService = pulsarTestContext.getBrokerService();
namespaceService = pulsar.getNamespaceService();
- doReturn(CompletableFuture.completedFuture(null)).when(namespaceService).getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(mock(NamespaceBundle.class))).when(namespaceService)
+ .getBundleAsync(any());
+ doReturn(CompletableFuture.completedFuture(true)).when(namespaceService).checkBundleOwnership(any(), any());
doReturn(true).when(namespaceService).isServiceUnitOwned(any());
doReturn(true).when(namespaceService).isServiceUnitActive(any());
doReturn(CompletableFuture.completedFuture(true)).when(namespaceService).isServiceUnitActiveAsync(any());
- doReturn(CompletableFuture.completedFuture(true)).when(namespaceService).checkTopicOwnership(any());
doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfTopics(
NamespaceName.get("use", "ns-abc"), CommandGetTopicsOfNamespace.Mode.ALL);
doReturn(CompletableFuture.completedFuture(topics)).when(namespaceService).getListOfUserTopics(
(pulsar) 02/02: [improve][broker] change to warn log level for ack validation error (#24459)
Posted by xy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 949a593738d10e03a7de12a36e78a4393f9c1250
Author: Zike Yang <zi...@apache.org>
AuthorDate: Wed Jun 25 20:06:33 2025 +0800
[improve][broker] change to warn log level for ack validation error (#24459)
## Motivation
The error log https://github.com/apache/pulsar/blob/e7eb7b37a779f9b74b8abf43cf622957663495d1/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java#L757-L758 mostly comes from the client.
The client could already receive this error, as seen at https://github.com/apache/pulsar/blob/82237d3684fe506bcb6426b3b23f413422e6e4fb/pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/MessageParser.java#L151-L152
We do not need to print this error log in the broker. Doing so would fill up the broker error logs and make them hard to manage.
## Modification
- Change to warn log level for ack validation error
(cherry picked from commit 3f9e21a5fa1a8a03bbc20f041d8e2590e6e9d742)
---
.../src/main/java/org/apache/pulsar/broker/service/Consumer.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index 001ce6681da..c8566348571 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -678,7 +678,7 @@ public class Consumer {
private void checkAckValidationError(CommandAck ack, PositionImpl position) {
if (ack.hasValidationError()) {
- log.error("[{}] [{}] Received ack for corrupted message at {} - Reason: {}", subscription,
+ log.warn("[{}] [{}] Received ack for corrupted message at {} - Reason: {}", subscription,
consumerId, position, ack.getValidationError());
}
}