You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2022/01/18 10:15:20 UTC
[pulsar] branch branch-2.8 updated (0f5b699 -> 8c86b3d)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 0f5b699 [Java Client] fixed Producer semaphore permit release issue (#13682)
new 4621722 Change ``ContextClassLoader`` to ``NarClassLoader`` in BrokerInterceptor (#13589)
new 4b19432 [Security] Upgrade protobuf to 3.16.1 to address CVE-2021-22569 (#13695)
new e375892 [pulsar-broker] The log prints NamespaceService#isServiceUnitActive exception stack information (#13553)
new 2e808b4 Fix bug :Infinity value for CPU or Bandwidth usage (#13609)
new 096b725 Fix the wrong multi-topic has message available behavior (#13634)
new 7a4691a Fix getInternalStats occasional lack of LeaderInfo (#13656)
new 5978edc Avoid call sync method in async rest API for delete subscription (#13666)
new 4f80414 Avoid call sync method in async rest API for force delete subscription (#13668)
new 9ab1aff Fix Flaky test BacklogQuotaManagerTest.testConsumerBacklogEvictionTimeQuotaWithEmptyLedger (#13724)
new d7117fb Fix the deadlock while using zookeeper thread to create ledger (#13744)
new 348ea61 Fix invalid rack name cause bookie join rack failed (#13683)
new 44b617b Fix inefficient forEach loop (#13742)
new 8c86b3d [Issuse 13640][broker] Fix non persistent topic subscription error. (#13685)
The 13 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:
distribution/server/src/assemble/LICENSE.bin.txt | 4 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 5 +-
pom.xml | 2 +-
.../broker/admin/impl/PersistentTopicsBase.java | 114 +++++++++++----------
.../pulsar/broker/admin/v1/PersistentTopics.java | 2 +-
.../pulsar/broker/admin/v2/PersistentTopics.java | 2 +-
.../BrokerInterceptorWithClassLoader.java | 31 ++++--
.../impl/GenericBrokerHostUsageImpl.java | 4 +-
.../loadbalance/impl/LinuxBrokerHostUsageImpl.java | 9 +-
.../pulsar/broker/namespace/NamespaceService.java | 8 +-
.../pulsar/broker/service/AbstractTopic.java | 14 +--
.../apache/pulsar/broker/service/ServerCnx.java | 3 +-
.../org/apache/pulsar/broker/service/Topic.java | 2 +
.../service/nonpersistent/NonPersistentTopic.java | 5 +
.../broker/service/persistent/PersistentTopic.java | 23 +++--
.../BrokerInterceptorWithClassLoaderTest.java | 97 +++++++++++++++++-
.../broker/service/BacklogQuotaManagerTest.java | 1 +
.../BrokerServiceAutoSubscriptionCreationTest.java | 14 +++
.../pulsar/client/impl/MultiTopicsReaderTest.java | 26 ++---
.../common/policies/data/impl/BookieInfoImpl.java | 10 ++
.../pulsar/admin/cli/PulsarAdminToolTest.java | 20 +++-
.../org/apache/pulsar/admin/cli/CmdBookies.java | 13 +++
.../client/impl/MultiTopicsConsumerImpl.java | 2 +-
.../collections/ConcurrentSortedLongPairSet.java | 14 +--
.../ConcurrentSortedLongPairSetTest.java | 12 ++-
pulsar-sql/presto-distribution/LICENSE | 2 +-
.../zookeeper/ZkBookieRackAffinityMapping.java | 5 +-
.../zookeeper/ZkBookieRackAffinityMappingTest.java | 27 +++++
28 files changed, 348 insertions(+), 123 deletions(-)
[pulsar] 13/13: [Issuse 13640][broker] Fix non persistent topic subscription error. (#13685)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8c86b3d6552d03f660056e9447e3ba9ce1d77ebb
Author: Shen Liu <li...@126.com>
AuthorDate: Tue Jan 18 12:10:07 2022 +0800
[Issuse 13640][broker] Fix non persistent topic subscription error. (#13685)
Fixes #13640
If pulsar broker started with `allowAutoSubscriptionCreation=false`, there are no way to subscribe non persistent topic.
Add `isPersistent` check in `ServerCnx`.
(cherry picked from commit 3dbe418b70e35d710fef7c8da6689e5ef6b8574a)
---
.../apache/pulsar/broker/admin/v1/PersistentTopics.java | 2 +-
.../apache/pulsar/broker/admin/v2/PersistentTopics.java | 2 +-
.../java/org/apache/pulsar/broker/service/ServerCnx.java | 3 ++-
.../main/java/org/apache/pulsar/broker/service/Topic.java | 2 ++
.../broker/service/nonpersistent/NonPersistentTopic.java | 5 +++++
.../pulsar/broker/service/persistent/PersistentTopic.java | 5 +++++
.../service/BrokerServiceAutoSubscriptionCreationTest.java | 14 ++++++++++++++
7 files changed, 30 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
index 112273feb..df4d553 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
@@ -605,7 +605,7 @@ public class PersistentTopics extends PersistentTopicsBase {
try {
validateTopicName(property, cluster, namespace, topic);
if (!topicName.isPersistent()) {
- throw new RestException(Response.Status.BAD_REQUEST, "Create subscription on non-persistent topic"
+ throw new RestException(Response.Status.BAD_REQUEST, "Create subscription on non-persistent topic "
+ "can only be done through client");
}
internalCreateSubscription(asyncResponse, decode(encodedSubName), messageId, authoritative, replicated);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index ca7e6cc..0fd307f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -1326,7 +1326,7 @@ public class PersistentTopics extends PersistentTopicsBase {
try {
validateTopicName(tenant, namespace, topic);
if (!topicName.isPersistent()) {
- throw new RestException(Response.Status.BAD_REQUEST, "Create subscription on non-persistent topic"
+ throw new RestException(Response.Status.BAD_REQUEST, "Create subscription on non-persistent topic "
+ "can only be done through client");
}
internalCreateSubscription(asyncResponse, decode(encodedSubName), messageId, authoritative, replicated);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 824a2e8..04d3ab5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1014,7 +1014,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
boolean rejectSubscriptionIfDoesNotExist = isDurable
&& !service.isAllowAutoSubscriptionCreation(topicName.toString())
- && !topic.getSubscriptions().containsKey(subscriptionName);
+ && !topic.getSubscriptions().containsKey(subscriptionName)
+ && topic.isPersistent();
if (rejectSubscriptionIfDoesNotExist) {
return FutureUtil
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
index 2b92560..e9638f3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
@@ -244,6 +244,8 @@ public interface Topic {
return false;
}
+ boolean isPersistent();
+
/* ------ Transaction related ------ */
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index 0180f5d..15b3d51 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -1075,4 +1075,9 @@ public class NonPersistentTopic extends AbstractTopic implements Topic {
protected boolean isTerminated() {
return false;
}
+
+ @Override
+ public boolean isPersistent() {
+ return false;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 11ec5f2..eee2706 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2927,6 +2927,11 @@ public class PersistentTopic extends AbstractTopic
return false;
}
+ @Override
+ public boolean isPersistent() {
+ return true;
+ }
+
private synchronized void fence() {
isFenced = true;
ScheduledFuture<?> monitoringTask = this.fencedTopicMonitoringTask;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoSubscriptionCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoSubscriptionCreationTest.java
index c635968..81f9521 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoSubscriptionCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoSubscriptionCreationTest.java
@@ -131,4 +131,18 @@ public class BrokerServiceAutoSubscriptionCreationTest extends BrokerTestBase {
assertFalse(admin.topics().getSubscriptions(topicName.toString()).contains(subscriptionName));
}
+ @Test
+ public void testNonPersistentTopicSubscriptionCreationWithAutoCreationDisable() throws Exception {
+ pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
+
+ final String topicName = "non-persistent://prop/ns-abc/test-subtopic-" + testId.getAndIncrement();
+ final String subscriptionName = "test-subtopic-sub";
+
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ // Subscribe operation should be successful
+ pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscribe();
+ assertTrue(admin.topics().getSubscriptions(topicName).contains(subscriptionName));
+ }
+
}
[pulsar] 10/13: Fix the deadlock while using zookeeper thread to create ledger (#13744)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d7117fba68a0dbdb992680103c6d4e2676336c2d
Author: lipenghui <pe...@apache.org>
AuthorDate: Fri Jan 14 17:47:13 2022 +0800
Fix the deadlock while using zookeeper thread to create ledger (#13744)
### Motivation
Fixes: #13736, the deadlock when using ZK thread to create a ledger.
### Modification
Use the executor of the managed ledger to create the ledger to avoid the deadlock.
(cherry picked from commit 1d4c374d9a87e5188c8b64677b8c6bea652a2d2a)
---
.../java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 9ee0a8f..b8a21f4 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1593,7 +1593,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
STATE_UPDATER.set(this, State.CreatingLedger);
this.lastLedgerCreationInitiationTimestamp = System.currentTimeMillis();
mbean.startDataLedgerCreateOp();
- asyncCreateLedger(bookKeeper, config, digestType, this, Collections.emptyMap());
+ // Use the executor here is to avoid use the Zookeeper thread to create the ledger which will lead
+ // to deadlock at the zookeeper client, details to see https://github.com/apache/pulsar/issues/13736
+ this.executor.execute(() ->
+ asyncCreateLedger(bookKeeper, config, digestType, this, Collections.emptyMap()));
}
}
[pulsar] 06/13: Fix getInternalStats occasional lack of LeaderInfo (#13656)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 7a4691a577996380a9ed6116f58724b85b7c31c7
Author: 包子 <wu...@gmail.com>
AuthorDate: Thu Jan 13 00:27:33 2022 +0800
Fix getInternalStats occasional lack of LeaderInfo (#13656)
AdminApiTest is flaky. The testGetPartitionedStatsInternal test method fails sporadically.
The root cause it that `stateFuture` is returned, but `availableBookiesFuture` is not accomplished, so we need ensure that `futures` is all completed before `stateFuture.complete`.
https://github.com/apache/pulsar/blob/5faae1c4e683f4796ad34c1c612de0bcfe754d2a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L2176-L2178
(cherry picked from commit dcf01e910a4f8d30cd86ab95ab800952af92692a)
---
.../org/apache/pulsar/broker/service/persistent/PersistentTopic.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 87728a8..52beacd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -1981,8 +1981,9 @@ public class PersistentTopic extends AbstractTopic
stats.state = ml.getState().toString();
stats.ledgers = Lists.newArrayList();
- List<CompletableFuture<String>> futures = includeLedgerMetadata ? Lists.newArrayList() : null;
+ List<CompletableFuture<String>> futures = Lists.newArrayList();
CompletableFuture<Set<String>> availableBookiesFuture = brokerService.pulsar().getAvailableBookiesAsync();
+ futures.add(availableBookiesFuture.handle((strings, throwable) -> null));
availableBookiesFuture.whenComplete((bookies, e) -> {
if (e != null) {
log.error("[{}] Failed to fetch available bookies.", topic, e);
@@ -1995,7 +1996,7 @@ public class PersistentTopic extends AbstractTopic
info.size = li.getSize();
info.offloaded = li.hasOffloadContext() && li.getOffloadContext().getComplete();
stats.ledgers.add(info);
- if (futures != null) {
+ if (includeLedgerMetadata) {
futures.add(ml.getLedgerMetadata(li.getLedgerId()).handle((lMetadata, ex) -> {
if (ex == null) {
info.metadata = lMetadata;
[pulsar] 09/13: Fix Flaky test BacklogQuotaManagerTest.testConsumerBacklogEvictionTimeQuotaWithEmptyLedger (#13724)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9ab1afffbdfe397cc75f3dfc1482eeb754f453c7
Author: gaozhangmin <ga...@qq.com>
AuthorDate: Fri Jan 14 11:44:56 2022 +0800
Fix Flaky test BacklogQuotaManagerTest.testConsumerBacklogEvictionTimeQuotaWithEmptyLedger (#13724)
(cherry picked from commit 1fd44908e76881cd8035f0772a9cdccc7bafb62d)
---
.../java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
index c0fc189..a0c8fd9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
@@ -497,6 +497,7 @@ public class BacklogQuotaManagerTest {
consumer.receive();
admin.topics().unload(topic);
+ Awaitility.await().until(consumer::isConnected);
PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topic);
assertEquals(internalStats.ledgers.size(), 2);
assertEquals(internalStats.ledgers.get(1).entries, 0);
[pulsar] 07/13: Avoid call sync method in async rest API for delete subscription (#13666)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5978edcb96dca08d66a1fdb711a66ecaa530652c
Author: lipenghui <pe...@apache.org>
AuthorDate: Wed Jan 12 21:57:34 2022 +0800
Avoid call sync method in async rest API for delete subscription (#13666)
(cherry picked from commit 8dc506c66c9d49e1393c20050ee39326d3788b7f)
---
.../broker/admin/impl/PersistentTopicsBase.java | 59 ++++++++++++----------
1 file changed, 31 insertions(+), 28 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 23f5742..ef2cee7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -1467,35 +1467,38 @@ public class PersistentTopicsBase extends AdminResource {
private void internalDeleteSubscriptionForNonPartitionedTopic(AsyncResponse asyncResponse,
String subName, boolean authoritative) {
- try {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE);
-
- Topic topic = getTopicReference(topicName);
- Subscription sub = topic.getSubscription(subName);
- if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
- return;
- }
- sub.delete().get();
- log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e.getCause() instanceof SubscriptionBusyException) {
- log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName, topicName, e);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Subscription has active connected consumers"));
- } else if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
- clientAppId(), topicName, e);
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenRun(() -> validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE))
+ .thenCompose(__ -> {
+ Topic topic = getTopicReference(topicName);
+ Subscription sub = topic.getSubscription(subName);
+ if (sub == null) {
+ throw new RestException(Status.NOT_FOUND, "Subscription not found");
}
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ return sub.delete();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ Throwable cause = e.getCause();
+ if (cause instanceof SubscriptionBusyException) {
+ log.error("[{}] Failed to delete subscription {} from topic {}", clientAppId(), subName,
+ topicName, cause);
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Subscription has active connected consumers"));
+ } else if (cause instanceof WebApplicationException) {
+ if (log.isDebugEnabled() && ((WebApplicationException) cause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, cause);
+ }
+ asyncResponse.resume(cause);
+ } else {
+ log.error("[{}] Failed to delete subscription {} {}", clientAppId(), topicName, subName, cause);
+ asyncResponse.resume(new RestException(cause));
+ }
+ return null;
+ });
}
protected void internalDeleteSubscriptionForcefully(AsyncResponse asyncResponse,
[pulsar] 04/13: Fix bug :Infinity value for CPU or Bandwidth usage (#13609)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2e808b4d53edeea363ccca6a8a149a62c46ed813
Author: chenlin <15...@qq.com>
AuthorDate: Wed Jan 12 11:10:52 2022 +0800
Fix bug :Infinity value for CPU or Bandwidth usage (#13609)
* fix bug :Infinity value for CPU or Bandwidth usage
* use executorService.schedule replace to executorService.schedule
* 1.use scheduleAtFixedDelay instead of scheduleAtFixedRate;
2.returns when elapsedSeconds <= 0 is true, skip this round of calculateBrokerHostUsage;
* 1.move the early return conditional block to before getTotalCpuUsage;
* 1.use scheduleWithFixedDelay instead of scheduleAtFixedRate in the class GenericBrokerHostUsageImpl;
* update log content: add a space in log
(cherry picked from commit 2a7515f9593a76b294bfe2835621a0ab8a904957)
---
.../broker/loadbalance/impl/GenericBrokerHostUsageImpl.java | 4 ++--
.../pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java | 9 +++++++--
2 files changed, 9 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java
index 72d36ab..1405e8e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java
@@ -54,9 +54,9 @@ public class GenericBrokerHostUsageImpl implements BrokerHostUsage {
this.totalCpuLimit = getTotalCpuLimit();
// Call now to initialize values before the constructor returns
calculateBrokerHostUsage();
- executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::checkCpuLoad), CPU_CHECK_MILLIS,
+ executorService.scheduleWithFixedDelay(catchingAndLoggingThrowables(this::checkCpuLoad), CPU_CHECK_MILLIS,
CPU_CHECK_MILLIS, TimeUnit.MILLISECONDS);
- executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::doCalculateBrokerHostUsage),
+ executorService.scheduleWithFixedDelay(catchingAndLoggingThrowables(this::doCalculateBrokerHostUsage),
hostUsageCheckIntervalMin,
hostUsageCheckIntervalMin, TimeUnit.MINUTES);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
index ccbe8e3..5bc7bf9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
@@ -87,7 +87,7 @@ public class LinuxBrokerHostUsageImpl implements BrokerHostUsage {
// Call now to initialize values before the constructor returns
calculateBrokerHostUsage();
- executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::calculateBrokerHostUsage),
+ executorService.scheduleWithFixedDelay(catchingAndLoggingThrowables(this::calculateBrokerHostUsage),
hostUsageCheckIntervalMin,
hostUsageCheckIntervalMin, TimeUnit.MINUTES);
}
@@ -105,9 +105,14 @@ public class LinuxBrokerHostUsageImpl implements BrokerHostUsage {
double totalNicUsageRx = getTotalNicUsageRxKb(nics);
double totalCpuLimit = getTotalCpuLimit();
- SystemResourceUsage usage = new SystemResourceUsage();
long now = System.currentTimeMillis();
double elapsedSeconds = (now - lastCollection) / 1000d;
+ if (elapsedSeconds <= 0) {
+ log.warn("elapsedSeconds {} is not expected, skip this round of calculateBrokerHostUsage", elapsedSeconds);
+ return;
+ }
+
+ SystemResourceUsage usage = new SystemResourceUsage();
double cpuUsage = getTotalCpuUsage(elapsedSeconds);
if (lastCollection == 0L) {
[pulsar] 11/13: Fix invalid rack name cause bookie join rack failed (#13683)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 348ea617f53df25dea1bb3176542b38024c92911
Author: Hang Chen <ch...@apache.org>
AuthorDate: Sun Jan 16 10:58:15 2022 +0800
Fix invalid rack name cause bookie join rack failed (#13683)
* Fix invalid rackname cause bookie join rack failed
(cherry picked from commit af761e2f8d3aee35c2cf081998f556972a92b574)
---
.../common/policies/data/impl/BookieInfoImpl.java | 10 ++++++++
.../pulsar/admin/cli/PulsarAdminToolTest.java | 20 ++++++++++++++--
.../org/apache/pulsar/admin/cli/CmdBookies.java | 13 +++++++++++
.../zookeeper/ZkBookieRackAffinityMapping.java | 5 +++-
.../zookeeper/ZkBookieRackAffinityMappingTest.java | 27 ++++++++++++++++++++++
5 files changed, 72 insertions(+), 3 deletions(-)
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BookieInfoImpl.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BookieInfoImpl.java
index de316e6..b584989 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BookieInfoImpl.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BookieInfoImpl.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.common.policies.data.impl;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
+import lombok.NonNull;
import org.apache.pulsar.common.policies.data.BookieInfo;
/**
@@ -40,6 +41,7 @@ public final class BookieInfoImpl implements BookieInfo {
public static class BookieInfoImplBuilder implements BookieInfo.Builder {
private String rack;
private String hostname;
+ private static final String PATH_SEPARATOR = "/";
public BookieInfoImplBuilder rack(String rack) {
this.rack = rack;
@@ -52,7 +54,15 @@ public final class BookieInfoImpl implements BookieInfo {
}
public BookieInfoImpl build() {
+ checkArgument(rack != null && !rack.isEmpty() && !rack.equals(PATH_SEPARATOR),
+ "rack name is invalid, it should not be null, empty or '/'");
return new BookieInfoImpl(rack, hostname);
}
+
+ public static void checkArgument(boolean expression, @NonNull Object errorMessage) {
+ if (!expression) {
+ throw new IllegalArgumentException(String.valueOf(errorMessage));
+ }
+ }
}
}
diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
index 4cbefb9..362e230 100644
--- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
+++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -75,11 +76,9 @@ import org.apache.pulsar.common.policies.data.BookiesClusterInfo;
import org.apache.pulsar.common.policies.data.BookiesRackConfiguration;
import org.apache.pulsar.common.policies.data.BundlesData;
import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.DelayedDeliveryPolicies;
import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.FailureDomain;
-import org.apache.pulsar.common.policies.data.FailureDomainImpl;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.InactiveTopicPolicies;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats.LedgerInfo;
@@ -1448,6 +1447,23 @@ public class PulsarAdminToolTest {
.rack("rack-1")
.hostname("host-1")
.build());
+
+ // test invalid rack name ""
+ try {
+ BookieInfo.builder().rack("").hostname("host-1").build();
+ fail();
+ } catch (IllegalArgumentException e) {
+ assertEquals(e.getMessage(), "rack name is invalid, it should not be null, empty or '/'");
+ }
+
+ // test invalid rack name "/"
+ try {
+ BookieInfo.builder().rack("/").hostname("host-1").build();
+ fail();
+ } catch (IllegalArgumentException e) {
+ assertEquals(e.getMessage(), "rack name is invalid, it should not be null, empty or '/'");
+ }
+
}
@Test
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBookies.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBookies.java
index c15e8db..38fd847 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBookies.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBookies.java
@@ -25,6 +25,8 @@ import com.beust.jcommander.Parameter;
import com.beust.jcommander.Parameters;
import java.util.function.Supplier;
+import com.google.common.base.Strings;
+import lombok.NonNull;
@Parameters(commandDescription = "Operations about bookies rack placement")
public class CmdBookies extends CmdBase {
@@ -73,6 +75,8 @@ public class CmdBookies extends CmdBase {
@Parameters(commandDescription = "Updates the rack placement information for a specific bookie in the cluster (note. bookie address format:`address:port`)")
private class UpdateBookie extends CliCommand {
+ private static final String PATH_SEPARATOR = "/";
+
@Parameter(names = { "-g", "--group" }, description = "Bookie group name", required = false)
private String group = "default";
@@ -87,12 +91,21 @@ public class CmdBookies extends CmdBase {
@Override
void run() throws Exception {
+ checkArgument(!Strings.isNullOrEmpty(bookieRack) && !bookieRack.trim().equals(PATH_SEPARATOR),
+ "rack name is invalid, it should not be null, empty or '/'");
+
getAdmin().bookies().updateBookieRackInfo(bookieAddress, group,
BookieInfo.builder()
.rack(bookieRack)
.hostname(bookieHost)
.build());
}
+
+ private void checkArgument(boolean expression, @NonNull Object errorMessage) {
+ if (!expression) {
+ throw new IllegalArgumentException(String.valueOf(errorMessage));
+ }
+ }
}
public CmdBookies(Supplier<PulsarAdmin> admin) {
diff --git a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkBookieRackAffinityMapping.java b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkBookieRackAffinityMapping.java
index caeba51..86d3988 100644
--- a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkBookieRackAffinityMapping.java
+++ b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkBookieRackAffinityMapping.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.zookeeper;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.api.client.util.Strings;
import java.net.InetAddress;
import java.net.URI;
import java.util.ArrayList;
@@ -212,7 +213,9 @@ public class ZkBookieRackAffinityMapping extends AbstractDNSToSwitchMapping
}
}
- if (bi != null) {
+ if (bi != null
+ && !Strings.isNullOrEmpty(bi.getRack())
+ && !bi.getRack().trim().equals("/")) {
String rack = bi.getRack();
if (!rack.startsWith("/")) {
rack = "/" + rack;
diff --git a/pulsar-zookeeper-utils/src/test/java/org/apache/pulsar/zookeeper/ZkBookieRackAffinityMappingTest.java b/pulsar-zookeeper-utils/src/test/java/org/apache/pulsar/zookeeper/ZkBookieRackAffinityMappingTest.java
index 3dde98b..2c276ed 100644
--- a/pulsar-zookeeper-utils/src/test/java/org/apache/pulsar/zookeeper/ZkBookieRackAffinityMappingTest.java
+++ b/pulsar-zookeeper-utils/src/test/java/org/apache/pulsar/zookeeper/ZkBookieRackAffinityMappingTest.java
@@ -36,6 +36,7 @@ import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
import org.apache.pulsar.common.policies.data.BookieInfo;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.testng.annotations.AfterMethod;
@@ -105,6 +106,32 @@ public class ZkBookieRackAffinityMappingTest {
assertEquals(racks2.get(2), null);
localZkc.delete(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, -1);
+ assertNull(racks1.get(2));
+ }
+
+ @Test
+ public void testInvalidRackName() throws InterruptedException, KeeperException {
+ String data = "{\"group1\": {\"" + BOOKIE1
+ + "\": {\"rack\": \"/\", \"hostname\": \"bookie1.example.com\"}, \"" + BOOKIE2
+ + "\": {\"rack\": \"\", \"hostname\": \"bookie2.example.com\"}}}";
+
+ ZkUtils.createFullPathOptimistic(localZkc, ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, data.getBytes(),
+ ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+ // Case1: ZKCache is given
+ ZkBookieRackAffinityMapping mapping1 = new ZkBookieRackAffinityMapping();
+ ClientConfiguration bkClientConf1 = new ClientConfiguration();
+ bkClientConf1.setProperty(ZooKeeperCache.ZK_CACHE_INSTANCE, new ZooKeeperCache("test", localZkc, 30) {
+ });
+
+ mapping1.setBookieAddressResolver(BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ mapping1.setConf(bkClientConf1);
+ List<String> racks1 = mapping1
+ .resolve(Lists.newArrayList(BOOKIE1.getHostName(), BOOKIE2.getHostName(), BOOKIE3.getHostName()));
+
+ assertNull(racks1.get(0));
+ assertNull(racks1.get(1));
+ assertNull(racks1.get(2));
}
@Test
[pulsar] 05/13: Fix the wrong multi-topic has message available behavior (#13634)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 096b725edc9824e5599fd4b0f6bf99fa6d5e3403
Author: Kai Wang <kw...@streamnative.io>
AuthorDate: Thu Jan 13 00:46:45 2022 +0800
Fix the wrong multi-topic has message available behavior (#13634)
Fixes #13605
### Motivation
Currently, the multiTopicReader `hasMessageAvailable` might get the wrong result, we must check `numMessagesInQueue() > 0` again after finish all consumer `hasMessageAvaliableAsync` future, bacause some message might already in `MultiTopicsConsumerImpl#incomingMessages`.
### Modifications
* Fix the wrong multi-topic has message available behavior.
* Use `reader.readNextAsync()` instead of block method `reader.readNext()`.
* Reduce the units test running time by changing `MultiTopicsReaderTest` to use `@BeforeClass`, `@AfterClass`.
(cherry picked from commit e57dc8faa2979b6d03410fa2cef5f1dd84e46785)
---
.../pulsar/client/impl/MultiTopicsReaderTest.java | 26 +++++++++++-----------
.../client/impl/MultiTopicsConsumerImpl.java | 2 +-
2 files changed, 14 insertions(+), 14 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java
index f6230e2..6b6bf95 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java
@@ -61,17 +61,17 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.Murmur3_32Hash;
import org.awaitility.Awaitility;
import org.testng.Assert;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-@Test(groups = "flaky")
@Slf4j
+@Test(groups = "flaky")
public class MultiTopicsReaderTest extends MockedPulsarServiceBaseTest {
private static final String subscription = "reader-multi-topics-sub";
- @BeforeMethod(alwaysRun = true)
+ @BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
super.internalSetup();
@@ -87,7 +87,7 @@ public class MultiTopicsReaderTest extends MockedPulsarServiceBaseTest {
admin.namespaces().createNamespace("my-property/my-ns", policies);
}
- @AfterMethod(alwaysRun = true)
+ @AfterClass(alwaysRun = true)
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
@@ -169,15 +169,15 @@ public class MultiTopicsReaderTest extends MockedPulsarServiceBaseTest {
private static <T> void readMessageUseAsync(Reader<T> reader, List<Message<T>> msgs, CountDownLatch latch) {
reader.hasMessageAvailableAsync().thenAccept(hasMessageAvailable -> {
if (hasMessageAvailable) {
- try {
- Message<T> msg = reader.readNext();
+ reader.readNextAsync().whenComplete((msg, ex) -> {
+ if (ex != null) {
+ log.error("Read message failed.", ex);
+ latch.countDown();
+ return;
+ }
msgs.add(msg);
- } catch (PulsarClientException e) {
- log.error("Read message failed.", e);
- latch.countDown();
- return;
- }
- readMessageUseAsync(reader, msgs, latch);
+ readMessageUseAsync(reader, msgs, latch);
+ });
} else {
latch.countDown();
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 177cc9f..fd98bb6 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -771,7 +771,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
if (exception != null) {
completableFuture.completeExceptionally(exception);
} else {
- completableFuture.complete(hasMessageAvailable.get());
+ completableFuture.complete(hasMessageAvailable.get() || numMessagesInQueue() > 0);
}
});
return completableFuture;
[pulsar] 08/13: Avoid call sync method in async rest API for force delete subscription (#13668)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4f804145e216c842dd4a174819426941b2465929
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Thu Jan 13 00:21:29 2022 +0800
Avoid call sync method in async rest API for force delete subscription (#13668)
(cherry picked from commit 416e7e5d110c35cbd51f7b082b70e944244a4938)
---
.../broker/admin/impl/PersistentTopicsBase.java | 55 +++++++++++-----------
1 file changed, 28 insertions(+), 27 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index ef2cee7..1c8aa8a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -1567,33 +1567,34 @@ public class PersistentTopicsBase extends AdminResource {
private void internalDeleteSubscriptionForNonPartitionedTopicForcefully(AsyncResponse asyncResponse,
String subName, boolean authoritative) {
- try {
- validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE);
-
- Topic topic = getTopicReference(topicName);
- Subscription sub = topic.getSubscription(subName);
- if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
- return;
- }
- sub.deleteForcefully().get();
- log.info("[{}][{}] Deleted subscription forcefully {}", clientAppId(), topicName, subName);
- asyncResponse.resume(Response.noContent().build());
- } catch (Exception e) {
- if (e instanceof WebApplicationException) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Failed to delete subscription forcefully from topic {},"
- + " redirecting to other brokers.",
- clientAppId(), topicName, e);
- }
- asyncResponse.resume(e);
- } else {
- log.error("[{}] Failed to delete subscription forcefully {} {}",
- clientAppId(), topicName, subName, e);
- asyncResponse.resume(new RestException(e));
- }
- }
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenRun(() -> validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE))
+ .thenCompose(__ -> {
+ Topic topic = getTopicReference(topicName);
+ Subscription sub = topic.getSubscription(subName);
+ if (sub == null) {
+ throw new RestException(Status.NOT_FOUND, "Subscription not found");
+ }
+ return sub.deleteForcefully();
+ }).thenRun(() -> {
+ log.info("[{}][{}] Deleted subscription forcefully {}", clientAppId(), topicName, subName);
+ asyncResponse.resume(Response.noContent().build());
+ }).exceptionally(e -> {
+ Throwable cause = e.getCause();
+ if (cause instanceof WebApplicationException) {
+ if (log.isDebugEnabled() && ((WebApplicationException) cause).getResponse().getStatus()
+ == Status.TEMPORARY_REDIRECT.getStatusCode()) {
+ log.debug("[{}] Failed to delete subscription from topic {}, redirecting to other brokers.",
+ clientAppId(), topicName, cause);
+ }
+ asyncResponse.resume(cause);
+ } else {
+ log.error("[{}] Failed to delete subscription forcefully {} {}",
+ clientAppId(), topicName, subName, cause);
+ asyncResponse.resume(new RestException(cause));
+ }
+ return null;
+ });
}
protected void internalSkipAllMessages(AsyncResponse asyncResponse, String subName, boolean authoritative) {
[pulsar] 01/13: Change ``ContextClassLoader`` to ``NarClassLoader`` in BrokerInterceptor (#13589)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4621722eaba3db189c2c78512fac133a9de05034
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Tue Jan 11 13:54:23 2022 +0800
Change ``ContextClassLoader`` to ``NarClassLoader`` in BrokerInterceptor (#13589)
It's ``BrokerInterceptor`` side change, like #13501
Change context class loader through Thread.currentThread().setContextClassLoader(classLoader) before every interceptor handler callback, and change it back to original class loader afterwards.
(cherry picked from commit afc241f02af4e8bf949d8d34b338b31c9fce5b9f)
---
.../BrokerInterceptorWithClassLoader.java | 31 +++++--
.../BrokerInterceptorWithClassLoaderTest.java | 97 +++++++++++++++++++++-
2 files changed, 116 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
index e1e23e4..6b06920 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
@@ -26,6 +26,7 @@ import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Entry;
+import org.apache.pulsar.broker.ClassLoaderSwitcher;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.Subscription;
@@ -50,39 +51,53 @@ public class BrokerInterceptorWithClassLoader implements BrokerInterceptor {
Entry entry,
long[] ackSet,
MessageMetadata msgMetadata) {
- this.interceptor.beforeSendMessage(
- subscription, entry, ackSet, msgMetadata);
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ this.interceptor.beforeSendMessage(
+ subscription, entry, ackSet, msgMetadata);
+ }
}
@Override
public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
- this.interceptor.onPulsarCommand(command, cnx);
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ this.interceptor.onPulsarCommand(command, cnx);
+ }
}
@Override
public void onConnectionClosed(ServerCnx cnx) {
- this.interceptor.onConnectionClosed(cnx);
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ this.interceptor.onConnectionClosed(cnx);
+ }
}
@Override
public void onWebserviceRequest(ServletRequest request) throws IOException, ServletException, InterceptException {
- this.interceptor.onWebserviceRequest(request);
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ this.interceptor.onWebserviceRequest(request);
+ }
}
@Override
public void onWebserviceResponse(ServletRequest request, ServletResponse response)
throws IOException, ServletException {
- this.interceptor.onWebserviceResponse(request, response);
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ this.interceptor.onWebserviceResponse(request, response);
+ }
}
@Override
public void initialize(PulsarService pulsarService) throws Exception {
- this.interceptor.initialize(pulsarService);
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ this.interceptor.initialize(pulsarService);
+ }
}
@Override
public void close() {
- interceptor.close();
+ try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
+ interceptor.close();
+ }
try {
classLoader.close();
} catch (IOException e) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoaderTest.java
index aa4a5bc..e4395b8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoaderTest.java
@@ -18,14 +18,30 @@
*/
package org.apache.pulsar.broker.intercept;
-import org.apache.pulsar.broker.PulsarService;
-import org.apache.pulsar.common.nar.NarClassLoader;
-import org.testng.annotations.Test;
-
import static org.mockito.ArgumentMatchers.same;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
+import com.google.common.collect.Maps;
+import io.netty.buffer.ByteBuf;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.Producer;
+import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandAck;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.intercept.InterceptException;
+import org.apache.pulsar.common.nar.NarClassLoader;
+import org.testng.annotations.Test;
+import javax.servlet.FilterChain;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import java.util.Map;
/**
* Unit test {@link BrokerInterceptorWithClassLoader}.
@@ -43,4 +59,77 @@ public class BrokerInterceptorWithClassLoaderTest {
verify(h, times(1)).initialize(same(pulsarService));
}
+
+ @Test
+ public void testClassLoaderSwitcher() throws Exception {
+ NarClassLoader narLoader = mock(NarClassLoader.class);
+ BrokerInterceptor interceptor = new BrokerInterceptor() {
+ @Override
+ public void beforeSendMessage(Subscription subscription, Entry entry, long[] ackSet, MessageMetadata msgMetadata) {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ @Override
+ public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ @Override
+ public void onConnectionClosed(ServerCnx cnx) {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ @Override
+ public void onWebserviceRequest(ServletRequest request) {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ @Override
+ public void onWebserviceResponse(ServletRequest request, ServletResponse response) {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ @Override
+ public void onFilter(ServletRequest request, ServletResponse response, FilterChain chain) {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ @Override
+ public void initialize(PulsarService pulsarService) throws Exception {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ @Override
+ public void close() {
+ assertEquals(Thread.currentThread().getContextClassLoader(), narLoader);
+ }
+ };
+
+ BrokerInterceptorWithClassLoader brokerInterceptorWithClassLoader =
+ new BrokerInterceptorWithClassLoader(interceptor, narLoader);
+ ClassLoader curClassLoader = Thread.currentThread().getContextClassLoader();
+ // test class loader
+ assertEquals(brokerInterceptorWithClassLoader.getClassLoader(), narLoader);
+ // test initialize
+ brokerInterceptorWithClassLoader.initialize(mock(PulsarService.class));
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test onFilter
+ brokerInterceptorWithClassLoader.onFilter(mock(ServletRequest.class)
+ , mock(ServletResponse.class), mock(FilterChain.class));
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test onWebserviceResponse
+ brokerInterceptorWithClassLoader.onWebserviceResponse(mock(ServletRequest.class)
+ , mock(ServletResponse.class));
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test onWebserviceRequest
+ brokerInterceptorWithClassLoader.onWebserviceRequest(mock(ServletRequest.class));
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test onConnectionClosed
+ brokerInterceptorWithClassLoader.onConnectionClosed(mock(ServerCnx.class));
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test onPulsarCommand
+ brokerInterceptorWithClassLoader.onPulsarCommand(null, mock(ServerCnx.class));
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test beforeSendMessage
+ brokerInterceptorWithClassLoader
+ .beforeSendMessage(mock(Subscription.class), mock(Entry.class), null, null);
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+ // test close
+ brokerInterceptorWithClassLoader.close();
+ assertEquals(Thread.currentThread().getContextClassLoader(), curClassLoader);
+
+ }
}
[pulsar] 02/13: [Security] Upgrade protobuf to 3.16.1 to address CVE-2021-22569 (#13695)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4b194327cdd2b3154491f5557a990041d9ca5aa0
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Jan 11 07:56:15 2022 +0200
[Security] Upgrade protobuf to 3.16.1 to address CVE-2021-22569 (#13695)
### Motivation
- protobuf < 3.16.1 contains DoS vulnerability CVE-2021-22569, https://nvd.nist.gov/vuln/detail/CVE-2021-22569.
### Modifications
- upgrade protobuf from 3.11.4 to 3.16.1
(cherry picked from commit 1a3688c936bb2320db34ccfbef08500f2c522591)
---
distribution/server/src/assemble/LICENSE.bin.txt | 4 ++--
pom.xml | 2 +-
pulsar-sql/presto-distribution/LICENSE | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index f5684e9..0c50f4c 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -543,8 +543,8 @@ MIT License
Protocol Buffers License
* Protocol Buffers
- - com.google.protobuf-protobuf-java-3.11.4.jar -- licenses/LICENSE-protobuf.txt
- - com.google.protobuf-protobuf-java-util-3.11.4.jar -- licenses/LICENSE-protobuf.txt
+ - com.google.protobuf-protobuf-java-3.16.1.jar -- licenses/LICENSE-protobuf.txt
+ - com.google.protobuf-protobuf-java-util-3.16.1.jar -- licenses/LICENSE-protobuf.txt
CDDL-1.1 -- licenses/LICENSE-CDDL-1.1.txt
* Java Annotations API
diff --git a/pom.xml b/pom.xml
index 8bafc50..84355ed 100644
--- a/pom.xml
+++ b/pom.xml
@@ -129,7 +129,7 @@ flexible messaging model and an intuitive client API.</description>
<puppycrawl.checkstyle.version>8.37</puppycrawl.checkstyle.version>
<dockerfile-maven.version>1.4.13</dockerfile-maven.version>
<typetools.version>0.5.0</typetools.version>
- <protobuf3.version>3.11.4</protobuf3.version>
+ <protobuf3.version>3.16.1</protobuf3.version>
<protoc3.version>${protobuf3.version}</protoc3.version>
<grpc.version>1.33.0</grpc.version>
<perfmark.version>0.19.0</perfmark.version>
diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE
index 0fa9952..9062e45 100644
--- a/pulsar-sql/presto-distribution/LICENSE
+++ b/pulsar-sql/presto-distribution/LICENSE
@@ -459,7 +459,7 @@ The Apache Software License, Version 2.0
Protocol Buffers License
* Protocol Buffers
- - protobuf-java-3.11.4.jar
+ - protobuf-java-3.16.1.jar
BSD 3-clause "New" or "Revised" License
* RE2J TD -- re2j-td-1.4.jar
[pulsar] 03/13: [pulsar-broker] The log prints NamespaceService#isServiceUnitActive exception stack information (#13553)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e375892af639194b48450389369129fc68be91b6
Author: mingyifei <96...@qq.com>
AuthorDate: Thu Jan 13 01:06:51 2022 +0800
[pulsar-broker] The log prints NamespaceService#isServiceUnitActive exception stack information (#13553)
(cherry picked from commit 58c281811b01dfc2bcf1906865964badbee4a952)
---
.../java/org/apache/pulsar/broker/namespace/NamespaceService.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 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 4fff4b0..1a42c19 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
@@ -991,9 +991,13 @@ public class NamespaceService implements AutoCloseable {
public boolean isServiceUnitActive(TopicName topicName) {
try {
- return ownershipCache.getOwnedBundle(getBundle(topicName)).isActive();
+ OwnedBundle ownedBundle = ownershipCache.getOwnedBundle(getBundle(topicName));
+ if (ownedBundle == null) {
+ return false;
+ }
+ return ownedBundle.isActive();
} catch (Exception e) {
- LOG.warn("Unable to find OwnedBundle for topic - [{}]", topicName);
+ LOG.warn("Unable to find OwnedBundle for topic - [{}]", topicName, e);
return false;
}
}
[pulsar] 12/13: Fix inefficient forEach loop (#13742)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 44b617b5830eed596fc3d6861231998aa9234d99
Author: 萧易客 <km...@live.com>
AuthorDate: Tue Jan 18 11:08:46 2022 +0800
Fix inefficient forEach loop (#13742)
### Motivation
There are some methods implemented with an inefficient forEach loop, so fix it to get better performance. It is similar to #12953
### Modifications
I rewrite it with the `for` loop.
(cherry picked from commit 9c94cd7803f3653e5aa170f80f5591fe68cc0366)
---
.../org/apache/pulsar/broker/service/AbstractTopic.java | 14 +++++++-------
.../pulsar/broker/service/persistent/PersistentTopic.java | 13 +++++++------
.../util/collections/ConcurrentSortedLongPairSet.java | 14 ++++++++------
.../util/collections/ConcurrentSortedLongPairSetTest.java | 12 ++++++++----
4 files changed, 30 insertions(+), 23 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 2576050..65296a7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -31,7 +31,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -264,14 +263,15 @@ public abstract class AbstractTopic implements Topic {
}
protected boolean hasLocalProducers() {
- AtomicBoolean foundLocal = new AtomicBoolean(false);
- producers.values().forEach(producer -> {
+ if (producers.isEmpty()) {
+ return false;
+ }
+ for (Producer producer : producers.values()) {
if (!producer.isRemote()) {
- foundLocal.set(true);
+ return true;
}
- });
-
- return foundLocal.get();
+ }
+ return false;
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 52beacd..11ec5f2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -596,14 +596,15 @@ public class PersistentTopic extends AbstractTopic
}
private boolean hasRemoteProducers() {
- AtomicBoolean foundRemote = new AtomicBoolean(false);
- producers.values().forEach(producer -> {
+ if (producers.isEmpty()) {
+ return false;
+ }
+ for (Producer producer : producers.values()) {
if (producer.isRemote()) {
- foundRemote.set(true);
+ return true;
}
- });
-
- return foundRemote.get();
+ }
+ return false;
}
public void startReplProducers() {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentSortedLongPairSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentSortedLongPairSet.java
index d3321f9..95e2302 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentSortedLongPairSet.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentSortedLongPairSet.java
@@ -166,13 +166,15 @@ public class ConcurrentSortedLongPairSet implements LongPairSet {
@Override
public boolean isEmpty() {
- AtomicBoolean isEmpty = new AtomicBoolean(true);
- longPairSets.forEach((item1, longPairSet) -> {
- if (isEmpty.get() && !longPairSet.isEmpty()) {
- isEmpty.set(false);
+ if (longPairSets.isEmpty()) {
+ return true;
+ }
+ for (ConcurrentLongPairSet subSet : longPairSets.values()) {
+ if (!subSet.isEmpty()) {
+ return false;
}
- });
- return isEmpty.get();
+ }
+ return true;
}
@Override
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentSortedLongPairSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentSortedLongPairSetTest.java
index 821bb88..fcb9884 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentSortedLongPairSetTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentSortedLongPairSetTest.java
@@ -22,7 +22,7 @@ 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 com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
@@ -30,13 +30,10 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-
import lombok.Cleanup;
import org.apache.pulsar.common.util.collections.ConcurrentLongPairSet.LongPair;
import org.testng.annotations.Test;
-import com.google.common.collect.Lists;
-
public class ConcurrentSortedLongPairSetTest {
@Test
@@ -241,4 +238,11 @@ public class ConcurrentSortedLongPairSetTest {
assertEquals(set.toString(), toString);
}
+ @Test
+ public void testIsEmpty() {
+ LongPairSet set = new ConcurrentSortedLongPairSet();
+ assertTrue(set.isEmpty());
+ set.add(1, 1);
+ assertFalse(set.isEmpty());
+ }
}