You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2026/04/10 03:29:10 UTC
(pulsar) branch master updated: [improve][client] PIP-234: Support sharing the memory limit controller across multiple isolated Pulsar client instances (#25477)
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new dfe03c4fc1b [improve][client] PIP-234: Support sharing the memory limit controller across multiple isolated Pulsar client instances (#25477)
dfe03c4fc1b is described below
commit dfe03c4fc1b0611fbc77698e5c0e328d7ef3242b
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Fri Apr 10 11:29:04 2026 +0800
[improve][client] PIP-234: Support sharing the memory limit controller across multiple isolated Pulsar client instances (#25477)
---
.../client/impl/ConsumerMemoryLimitTest.java | 105 +++++++++++++++++++++
.../client/impl/ProducerMemoryLimitTest.java | 85 +++++++++++++++++
.../pulsar/client/impl/PulsarTestClient.java | 2 +-
.../pulsar/client/api/MemoryLimitConfig.java | 38 ++++++++
.../pulsar/client/api/OpenTelemetryConfig.java | 40 ++++++++
.../client/api/PulsarClientSharedResources.java | 11 ++-
.../api/PulsarClientSharedResourcesBuilder.java | 16 ++++
.../pulsar/client/impl/MemoryLimitController.java | 29 ++++--
.../pulsar/client/impl/PulsarClientImpl.java | 43 ++++++---
.../impl/PulsarClientResourcesConfigurer.java | 23 +++++
.../PulsarClientSharedResourcesBuilderImpl.java | 48 ++++++++++
.../impl/PulsarClientSharedResourcesImpl.java | 34 +++++++
12 files changed, 448 insertions(+), 26 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerMemoryLimitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerMemoryLimitTest.java
index f3bf80a646c..df4874317c4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerMemoryLimitTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerMemoryLimitTest.java
@@ -22,8 +22,11 @@ import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientSharedResources;
import org.apache.pulsar.client.api.SizeUnit;
import org.awaitility.Awaitility;
import org.testng.Assert;
@@ -92,4 +95,106 @@ public class ConsumerMemoryLimitTest extends SharedPulsarBaseTest {
c1.receive();
}
}
+
+ @Test
+ public void testMultiPulsarClientConsumerShareMemoryLimitController() throws Exception {
+ int msgSize = 100;
+ int msgCount = 3;
+ int memoryLimit = msgSize * msgCount;
+ String topic1 = newTopicName();
+ String topic2 = newTopicName();
+ PulsarClientSharedResources sharedResources = PulsarClientSharedResources.builder()
+ .configureMemoryLimitController(
+ memoryLimitConfig -> memoryLimitConfig.memoryLimit(memoryLimit, SizeUnit.BYTES)).build();
+ @Cleanup
+ PulsarClientImpl pulsarClient =
+ ((PulsarClientImpl) PulsarClient.builder().serviceUrl(getBrokerServiceUrl()).build());
+ @Cleanup
+ PulsarClientImpl pulsarClient1 = ((PulsarClientImpl) PulsarClient.builder().serviceUrl(getBrokerServiceUrl())
+ .sharedResources(sharedResources).build());
+ @Cleanup
+ PulsarClientImpl pulsarClient2 = ((PulsarClientImpl) PulsarClient.builder().serviceUrl(getBrokerServiceUrl())
+ .sharedResources(sharedResources).build());
+
+ Assert.assertSame(pulsarClient1.getMemoryLimitController(), pulsarClient2.getMemoryLimitController());
+
+ @Cleanup
+ Producer<byte[]> topic1Producer =
+ pulsarClient.newProducer().topic(topic1).enableBatching(false).blockIfQueueFull(false).create();
+ @Cleanup
+ Producer<byte[]> topic2Producer =
+ pulsarClient.newProducer().topic(topic2).enableBatching(false).blockIfQueueFull(false).create();
+ ConsumerImpl<byte[]> topic1Consumer =
+ (ConsumerImpl<byte[]>) pulsarClient1.newConsumer().subscriptionName("topic1-sub").topic(topic1)
+ .autoScaledReceiverQueueSizeEnabled(true).subscribe();
+ ConsumerImpl<byte[]> topic2Consumer =
+ (ConsumerImpl<byte[]>) pulsarClient2.newConsumer().subscriptionName("topic2-sub").topic(topic2)
+ .autoScaledReceiverQueueSizeEnabled(true).subscribe();
+
+ MemoryLimitController memoryLimitController1 = topic1Consumer.getMemoryLimitController().get();
+ MemoryLimitController memoryLimitController2 = topic2Consumer.getMemoryLimitController().get();
+ Assert.assertSame(memoryLimitController1, memoryLimitController2);
+
+ Assert.assertEquals(topic1Consumer.getCurrentReceiverQueueSize(), 1);
+ Assert.assertEquals(topic2Consumer.getCurrentReceiverQueueSize(), 1);
+
+
+ topic1Producer.send(new byte[msgSize]);
+ Awaitility.await().until(topic1Consumer.scaleReceiverQueueHint::get);
+
+ Message<byte[]> topic1Message = topic1Consumer.receive();
+ Assert.assertNotNull(topic1Message);
+ Assert.assertEquals(topic1Consumer.getCurrentReceiverQueueSize(), 1);
+
+ // Trigger ConsumerBase.expectMoreIncomingMessages() method to expand receiverQueueSize.
+ topic1Consumer.receiveAsync();
+ Awaitility.await().until(() -> topic1Consumer.getCurrentReceiverQueueSize() == 2);
+
+
+ topic2Producer.send(new byte[msgSize]);
+ Awaitility.await().until(topic2Consumer.scaleReceiverQueueHint::get);
+
+ Message<byte[]> topic2Message = topic2Consumer.receive();
+ Assert.assertNotNull(topic2Message);
+ Assert.assertEquals(topic2Consumer.getCurrentReceiverQueueSize(), 1);
+
+ // Trigger ConsumerBase.expectMoreIncomingMessages() method to expand receiverQueueSize.
+ topic2Consumer.receiveAsync();
+ Awaitility.await().until(() -> topic2Consumer.getCurrentReceiverQueueSize() == 2);
+
+
+ // topic1Consumer.receiveAsync() will take one message, so we should send (msgCount + 1) messages.
+ // Trigger ConsumerBase.reduceCurrentReceiverQueueSize() method to reduce receiverQueueSize.
+ topic1Consumer.setCurrentReceiverQueueSize(msgCount);
+ for (int i = 0; i < msgCount + 1; i++) {
+ topic1Producer.send(new byte[msgSize]);
+ }
+ Awaitility.await().until(() -> memoryLimitController1.currentUsage() == memoryLimit);
+ Awaitility.await().until(() -> topic1Consumer.getCurrentReceiverQueueSize() == 1);
+ Awaitility.await().until(() -> topic2Consumer.getCurrentReceiverQueueSize() == 1);
+
+ // topic2Consumer.receiveAsync() will take one message, so we should send (msgCount + 1) messages.
+ for (int i = 0; i < msgCount + 1; i++) {
+ topic2Producer.send(new byte[msgSize]);
+ }
+ // topic2Consumer will not expand receiverQueueSize due to memory limit reached.
+ for (int i = 0; i < msgCount; i++) {
+ topic2Message = topic2Consumer.receive();
+ Assert.assertNotNull(topic2Message);
+ Assert.assertEquals(topic2Consumer.getCurrentReceiverQueueSize(), 1);
+ }
+
+ // Close topic1Consumer to clear release memory.
+ topic1Consumer.close();
+ Assert.assertEquals(memoryLimitController1.currentUsage(), 0);
+
+ // Trigger ConsumerBase.expectMoreIncomingMessages() method to expand receiverQueueSize.
+ topic2Consumer.receiveAsync();
+ Awaitility.await().until(() -> topic2Consumer.getCurrentReceiverQueueSize() == 2);
+
+ // Close topic1Consumer to clear release memory.
+ topic2Consumer.close();
+ Assert.assertEquals(memoryLimitController2.currentUsage(), 0);
+ }
+
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java
index 55a67ae644d..97c91c64667 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java
@@ -30,10 +30,13 @@ import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientSharedResources;
import org.apache.pulsar.client.api.SizeUnit;
+import org.awaitility.Awaitility;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -229,6 +232,88 @@ public class ProducerMemoryLimitTest extends ProducerConsumerBase {
Assert.assertEquals(memoryLimitController.currentUsage(), 0);
}
+ @Test(timeOut = 15000)
+ public void testMultiPulsarClientProducerShareMemoryLimitController() throws Exception {
+ int msgSize = 100;
+ int msgCount = 6;
+ int memoryLimit = msgSize * msgCount;
+ PulsarClientSharedResources sharedResources = PulsarClientSharedResources.builder()
+ .configureMemoryLimitController(
+ memoryLimitConfig -> memoryLimitConfig.memoryLimit(memoryLimit, SizeUnit.BYTES)).build();
+ @Cleanup
+ PulsarClientImpl pulsarClient1 = ((PulsarClientImpl) PulsarClient.builder().serviceUrl(lookupUrl.toString())
+ .sharedResources(sharedResources).build());
+ @Cleanup
+ PulsarClientImpl pulsarClient2 = ((PulsarClientImpl) PulsarClient.builder().serviceUrl(lookupUrl.toString())
+ .sharedResources(sharedResources).build());
+
+ Assert.assertSame(pulsarClient1.getMemoryLimitController(), pulsarClient2.getMemoryLimitController());
+
+ Producer<byte[]> producer1 = pulsarClient1.newProducer()
+ .topic("testProducerShareMemoryLimitController")
+ .sendTimeout(20, TimeUnit.SECONDS)
+ .maxPendingMessages(0)
+ .blockIfQueueFull(false)
+ .enableBatching(false)
+ .create();
+ Producer<byte[]> producer2 = pulsarClient2.newProducer()
+ .topic("testProducerShareMemoryLimitController")
+ .sendTimeout(20, TimeUnit.SECONDS)
+ .maxPendingMessages(0)
+ .blockIfQueueFull(false)
+ .enableBatching(false)
+ .create();
+
+ this.stopBroker();
+
+ byte[] msgBytes = new byte[msgSize];
+ int halfMsgCount = msgCount / 2;
+ int producer1MemoryUsage = halfMsgCount * msgSize;
+ for (int i = 0; i < msgCount / 2; i++) {
+ producer1.sendAsync(msgBytes);
+ }
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> pulsarClient1.getMemoryLimitController().currentUsage() == producer1MemoryUsage);
+
+ // The MemoryLimitController.tryReserveMemory() method returns false only when currentUsage > memoryLimit.
+ int producer2MsgCount = halfMsgCount + 1;
+ int producer2MemoryUsage = producer2MsgCount * msgSize;
+ int totalMemoryUsage = producer1MemoryUsage + producer2MemoryUsage;
+ for (int i = 0; i < producer2MsgCount; i++) {
+ producer2.sendAsync(msgBytes);
+ }
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> pulsarClient2.getMemoryLimitController().currentUsage() == totalMemoryUsage);
+
+ try {
+ producer1.send(msgBytes);
+ Assert.fail("producer can not send message due to memory limit exceeded");
+ } catch (PulsarClientException.MemoryBufferIsFullError ex) {
+ long currentUsage = pulsarClient1.getMemoryLimitController().currentUsage();
+ Assert.assertEquals(currentUsage, totalMemoryUsage);
+ }
+
+ try {
+ producer2.send(msgBytes);
+ Assert.fail("producer can not send message due to memory limit exceeded");
+ } catch (PulsarClientException.MemoryBufferIsFullError ex) {
+ long currentUsage = pulsarClient2.getMemoryLimitController().currentUsage();
+ Assert.assertEquals(currentUsage, totalMemoryUsage);
+ }
+
+ producer1.close();
+ Assert.assertEquals(pulsarClient1.getMemoryLimitController().currentUsage(), producer2MemoryUsage);
+
+ for (int i = 0; i < halfMsgCount; i++) {
+ producer2.sendAsync(msgBytes);
+ }
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> pulsarClient2.getMemoryLimitController().currentUsage() == totalMemoryUsage);
+
+ producer2.close();
+ Assert.assertEquals(pulsarClient1.getMemoryLimitController().currentUsage(), 0);
+ }
+
private void initClientWithMemoryLimit() throws PulsarClientException {
replacePulsarClient(PulsarClient.builder().
serviceUrl(lookupUrl.toString())
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
index db330951f74..d72426e3d8d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarTestClient.java
@@ -91,7 +91,7 @@ public class PulsarTestClient extends PulsarClientImpl {
private PulsarTestClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool,
AtomicReference<Supplier<ClientCnx>> clientCnxSupplierReference)
throws PulsarClientException {
- super(conf, eventLoopGroup, cnxPool, null, null, null, null, null, new DnsResolverGroupImpl(conf));
+ super(conf, eventLoopGroup, cnxPool, null, null, null, null, null, new DnsResolverGroupImpl(conf), null);
// workaround initialization order issue so that ClientCnx can be created in this class
clientCnxSupplierReference.set(this::createClientCnx);
}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MemoryLimitConfig.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MemoryLimitConfig.java
new file mode 100644
index 00000000000..14a85ee305e
--- /dev/null
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MemoryLimitConfig.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.client.api;
+
+/**
+ * Configuration interface for memory limit settings.
+ */
+public interface MemoryLimitConfig {
+
+ /**
+ * Configure a limit on the amount of direct memory that will be allocated by this shared client instance.
+ * <p>
+ * See also {@link ClientBuilder#memoryLimit(long, SizeUnit)}.
+ *
+ * @param memoryLimit the memory limit value, setting this to 0 will disable the limit
+ * @param unit the memory limit size unit
+ * @return the memory limit configuration instance for chained calls
+ */
+ MemoryLimitConfig memoryLimit(long memoryLimit, SizeUnit unit);
+
+}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/OpenTelemetryConfig.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/OpenTelemetryConfig.java
new file mode 100644
index 00000000000..0b05f960041
--- /dev/null
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/OpenTelemetryConfig.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.client.api;
+
+
+import io.opentelemetry.api.OpenTelemetry;
+
+/**
+ * Configuration interface for open telemetry settings.
+ */
+public interface OpenTelemetryConfig {
+
+ /**
+ * Configure OpenTelemetry for this shared client instance.
+ * <p>
+ * See also {@link ClientBuilder#openTelemetry(OpenTelemetry)}.
+ *
+ * @param openTelemetry the open telemetry instance
+ * @return the open telemetry configuration instance for chained calls
+ */
+ OpenTelemetryConfig openTelemetry(OpenTelemetry openTelemetry);
+
+}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResources.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResources.java
index 34a56ebd608..3b2db4e9233 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResources.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResources.java
@@ -71,7 +71,12 @@ public interface PulsarClientSharedResources extends AutoCloseable {
// pulsar-lookup threadpool
LookupExecutor(SharedResourceType.ThreadPool),
// DNS resolver and cache that must be shared together with eventLoopGroup
- DnsResolver(SharedResourceType.DnsResolver);
+ DnsResolver(SharedResourceType.DnsResolver),
+ // pulsar client global memory limit controller
+ MemoryLimitController(SharedResourceType.MemoryLimitController),
+ // pulsar client global open telemetry instance
+ OpenTelemetry(SharedResourceType.OpenTelemetry)
+ ;
private final SharedResourceType type;
@@ -88,7 +93,9 @@ public interface PulsarClientSharedResources extends AutoCloseable {
EventLoopGroup,
ThreadPool,
Timer,
- DnsResolver;
+ DnsResolver,
+ MemoryLimitController,
+ OpenTelemetry;
}
/**
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResourcesBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResourcesBuilder.java
index 3a94d5352b5..227f20da7ad 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResourcesBuilder.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientSharedResourcesBuilder.java
@@ -119,4 +119,20 @@ public interface PulsarClientSharedResourcesBuilder {
*/
PulsarClientSharedResourcesBuilder configureTimer(Consumer<TimerConfig> configurer);
+ /**
+ * Configures the memory limit settings.
+ *
+ * @param configurer a consumer that configures the memory limit settings
+ * @return this builder instance for method chaining
+ */
+ PulsarClientSharedResourcesBuilder configureMemoryLimitController(Consumer<MemoryLimitConfig> configurer);
+
+ /**
+ * Configures the open telemetry settings.
+ *
+ * @param configurer a consumer that configures the open telemetry settings
+ * @return this builder instance for method chaining
+ */
+ PulsarClientSharedResourcesBuilder configureOpenTelemetry(Consumer<OpenTelemetryConfig> configurer);
+
}
\ No newline at end of file
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MemoryLimitController.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MemoryLimitController.java
index 4f8a6fdb06a..489562278b2 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MemoryLimitController.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MemoryLimitController.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
@@ -29,22 +30,24 @@ public class MemoryLimitController {
private final long memoryLimit;
private final long triggerThreshold;
- private final Runnable trigger;
+ private final CopyOnWriteArraySet<Runnable> triggers = new CopyOnWriteArraySet<>();
private final AtomicLong currentUsage = new AtomicLong();
private final ReentrantLock mutex = new ReentrantLock(false);
private final Condition condition = mutex.newCondition();
private final AtomicBoolean triggerRunning = new AtomicBoolean(false);
public MemoryLimitController(long memoryLimitBytes) {
- this.memoryLimit = memoryLimitBytes;
- triggerThreshold = 0;
- trigger = null;
+ this(memoryLimitBytes, 0);
}
- public MemoryLimitController(long memoryLimitBytes, long triggerThreshold, Runnable trigger) {
+ public MemoryLimitController(long memoryLimitBytes, long triggerThreshold) {
this.memoryLimit = memoryLimitBytes;
this.triggerThreshold = triggerThreshold;
- this.trigger = trigger;
+ }
+
+ public MemoryLimitController(long memoryLimitBytes, long triggerThreshold, Runnable trigger) {
+ this(memoryLimitBytes, triggerThreshold);
+ this.triggers.add(trigger);
}
public void forceReserveMemory(long size) {
@@ -88,10 +91,12 @@ public class MemoryLimitController {
}
private void checkTrigger(long prevUsage, long newUsage) {
- if (newUsage >= triggerThreshold && prevUsage < triggerThreshold && trigger != null) {
+ if (newUsage >= triggerThreshold && prevUsage < triggerThreshold && !triggers.isEmpty()) {
if (triggerRunning.compareAndSet(false, true)) {
try {
- trigger.run();
+ for (Runnable trigger : triggers) {
+ trigger.run();
+ }
} finally {
triggerRunning.set(false);
}
@@ -152,4 +157,12 @@ public class MemoryLimitController {
public long memoryLimit() {
return memoryLimit;
}
+
+ public void registerTrigger(Runnable trigger) {
+ triggers.add(trigger);
+ }
+
+ public void deregisterTrigger(Runnable trigger) {
+ triggers.remove(trigger);
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index c1034925ce1..020786ee578 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -106,7 +106,7 @@ public class PulsarClientImpl implements PulsarClient {
private static final Logger log = LoggerFactory.getLogger(PulsarClientImpl.class);
private static final int CLOSE_TIMEOUT_SECONDS = 60;
- private static final double THRESHOLD_FOR_CONSUMER_RECEIVER_QUEUE_SIZE_SHRINKING = 0.95;
+ protected static final double THRESHOLD_FOR_CONSUMER_RECEIVER_QUEUE_SIZE_SHRINKING = 0.95;
// default limits for producers when memory limit controller is disabled
private static final int NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES = 1000;
@@ -171,23 +171,25 @@ public class PulsarClientImpl implements PulsarClient {
@Getter
private TransactionCoordinatorClientImpl tcClient;
+ private final Runnable memoryLimitTrigger = this::reduceConsumerReceiverQueueSize;
+
public PulsarClientImpl(ClientConfigurationData conf) throws PulsarClientException {
- this(conf, null, null, null, null, null, null, null, null);
+ this(conf, null, null, null, null, null, null, null, null, null);
}
public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) throws PulsarClientException {
- this(conf, eventLoopGroup, null, null, null, null, null, null, null);
+ this(conf, eventLoopGroup, null, null, null, null, null, null, null, null);
}
public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool)
throws PulsarClientException {
- this(conf, eventLoopGroup, cnxPool, null, null, null, null, null, null);
+ this(conf, eventLoopGroup, cnxPool, null, null, null, null, null, null, null);
}
public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool,
Timer timer)
throws PulsarClientException {
- this(conf, eventLoopGroup, cnxPool, timer, null, null, null, null, null);
+ this(conf, eventLoopGroup, cnxPool, timer, null, null, null, null, null, null);
}
public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool connectionPool,
@@ -196,7 +198,7 @@ public class PulsarClientImpl implements PulsarClient {
ScheduledExecutorProvider scheduledExecutorProvider)
throws PulsarClientException {
this(conf, eventLoopGroup, connectionPool, timer, externalExecutorProvider, internalExecutorProvider,
- scheduledExecutorProvider, null, null);
+ scheduledExecutorProvider, null, null, null);
}
@Builder(builderClassName = "PulsarClientImplBuilder")
@@ -205,8 +207,8 @@ public class PulsarClientImpl implements PulsarClient {
ExecutorProvider internalExecutorProvider,
ScheduledExecutorProvider scheduledExecutorProvider,
ExecutorProvider lookupExecutorProvider,
- DnsResolverGroupImpl dnsResolverGroup) throws PulsarClientException {
-
+ DnsResolverGroupImpl dnsResolverGroup,
+ MemoryLimitController memoryLimitController) throws PulsarClientException {
EventLoopGroup eventLoopGroupReference = null;
ConnectionPool connectionPoolReference = null;
try {
@@ -286,14 +288,19 @@ public class PulsarClientImpl implements PulsarClient {
}
}
- memoryLimitController = new MemoryLimitController(conf.getMemoryLimitBytes(),
- (long) (conf.getMemoryLimitBytes() * THRESHOLD_FOR_CONSUMER_RECEIVER_QUEUE_SIZE_SHRINKING),
- this::reduceConsumerReceiverQueueSize);
- // Only create memory buffer metrics if memory limiting is enabled
- if (memoryLimitController.isMemoryLimited()) {
- memoryBufferStats = new MemoryBufferStats(instrumentProvider, memoryLimitController);
+ if (memoryLimitController == null) {
+ this.memoryLimitController = new MemoryLimitController(conf.getMemoryLimitBytes(),
+ (long) (conf.getMemoryLimitBytes() * THRESHOLD_FOR_CONSUMER_RECEIVER_QUEUE_SIZE_SHRINKING),
+ this.memoryLimitTrigger);
+ } else {
+ this.memoryLimitController = memoryLimitController;
+ this.memoryLimitController.registerTrigger(this.memoryLimitTrigger);
+ }
+ // Only create memory buffer metrics if memory limit controller is local and memory limiting is enabled.
+ if (memoryLimitController == null && this.memoryLimitController.isMemoryLimited()) {
+ this.memoryBufferStats = new MemoryBufferStats(this.instrumentProvider, this.memoryLimitController);
} else {
- memoryBufferStats = null;
+ this.memoryBufferStats = null;
}
state.set(State.Open);
} catch (Throwable t) {
@@ -1003,6 +1010,7 @@ public class PulsarClientImpl implements PulsarClient {
} catch (PulsarClientException e) {
throwable = e;
}
+
if (memoryBufferStats != null) {
try {
memoryBufferStats.close();
@@ -1011,6 +1019,11 @@ public class PulsarClientImpl implements PulsarClient {
throwable = t;
}
}
+
+ if (memoryLimitController != null) {
+ memoryLimitController.deregisterTrigger(memoryLimitTrigger);
+ }
+
if (conf != null && conf.getAuthentication() != null) {
try {
conf.getAuthentication().close();
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientResourcesConfigurer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientResourcesConfigurer.java
index 28f97e92159..391cb93dba7 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientResourcesConfigurer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientResourcesConfigurer.java
@@ -20,10 +20,13 @@ package org.apache.pulsar.client.impl;
import io.netty.channel.EventLoopGroup;
import io.netty.util.HashedWheelTimer;
+import io.opentelemetry.api.OpenTelemetry;
+import java.util.Optional;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.client.util.ScheduledExecutorProvider;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
@@ -146,4 +149,24 @@ class PulsarClientResourcesConfigurer {
}
return new DnsResolverGroupImpl(resourceConfig);
}
+
+ static MemoryLimitController createMemoryLimitControllerWithResourceConfig(
+ PulsarClientSharedResourcesBuilderImpl.MemoryLimitResourceConfig resourceConfig) {
+ if (resourceConfig == null) {
+ resourceConfig = new PulsarClientSharedResourcesBuilderImpl.MemoryLimitResourceConfig();
+ }
+ long memoryLimit = resourceConfig.memoryLimit;
+ // Keep the same logic with PulsarClientImpl's memory limit settings.
+ long triggerThreshold =
+ (long) (memoryLimit * PulsarClientImpl.THRESHOLD_FOR_CONSUMER_RECEIVER_QUEUE_SIZE_SHRINKING);
+ return new MemoryLimitController(memoryLimit, triggerThreshold);
+ }
+
+ static InstrumentProvider createInstrumentProviderWithResourceConfig(
+ PulsarClientSharedResourcesBuilderImpl.OpenTelemetryResourceConfig resourceConfig) {
+ // Use global openTelemetry instance if not configured.
+ OpenTelemetry openTelemetry =
+ Optional.ofNullable(resourceConfig).map(config -> config.openTelemetry).orElse(null);
+ return new InstrumentProvider(openTelemetry);
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImpl.java
index f8ac98bb862..e2dc5995167 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesBuilderImpl.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.client.impl;
import com.google.common.collect.Lists;
+import io.opentelemetry.api.OpenTelemetry;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.HashMap;
@@ -30,8 +31,11 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import org.apache.pulsar.client.api.DnsResolverConfig;
import org.apache.pulsar.client.api.EventLoopGroupConfig;
+import org.apache.pulsar.client.api.MemoryLimitConfig;
+import org.apache.pulsar.client.api.OpenTelemetryConfig;
import org.apache.pulsar.client.api.PulsarClientSharedResources;
import org.apache.pulsar.client.api.PulsarClientSharedResourcesBuilder;
+import org.apache.pulsar.client.api.SizeUnit;
import org.apache.pulsar.client.api.ThreadPoolConfig;
import org.apache.pulsar.client.api.TimerConfig;
import org.apache.pulsar.common.util.netty.DnsResolverUtil;
@@ -202,6 +206,34 @@ public class PulsarClientSharedResourcesBuilderImpl implements PulsarClientShare
}
}
+ static class MemoryLimitResourceConfig implements ResourceConfig, MemoryLimitConfig {
+ long memoryLimit;
+
+ public MemoryLimitResourceConfig() {
+ }
+
+ @Override
+ public MemoryLimitConfig memoryLimit(long memoryLimit, SizeUnit unit) {
+ this.memoryLimit = unit.toBytes(memoryLimit);
+ return this;
+ }
+
+ }
+
+ static class OpenTelemetryResourceConfig implements ResourceConfig, OpenTelemetryConfig {
+ OpenTelemetry openTelemetry;
+
+ public OpenTelemetryResourceConfig() {
+ }
+
+ @Override
+ public OpenTelemetryConfig openTelemetry(OpenTelemetry openTelemetry) {
+ this.openTelemetry = openTelemetry;
+ return this;
+ }
+
+ }
+
@Override
public PulsarClientSharedResourcesBuilder resourceTypes(
PulsarClientSharedResources.SharedResource... sharedResource) {
@@ -242,6 +274,10 @@ public class PulsarClientSharedResourcesBuilderImpl implements PulsarClientShare
return new ThreadPoolResourceConfig();
case Timer:
return new TimerResourceConfig();
+ case MemoryLimitController:
+ return new MemoryLimitResourceConfig();
+ case OpenTelemetry:
+ return new OpenTelemetryResourceConfig();
default:
throw new IllegalArgumentException("Unknown resource type: " + sharedResource.getType());
}
@@ -277,6 +313,18 @@ public class PulsarClientSharedResourcesBuilderImpl implements PulsarClientShare
return this;
}
+ @Override
+ public PulsarClientSharedResourcesBuilder configureMemoryLimitController(Consumer<MemoryLimitConfig> configurer) {
+ configurer.accept(getOrCreateConfig(PulsarClientSharedResources.SharedResource.MemoryLimitController));
+ return this;
+ }
+
+ @Override
+ public PulsarClientSharedResourcesBuilder configureOpenTelemetry(Consumer<OpenTelemetryConfig> configurer) {
+ configurer.accept(getOrCreateConfig(PulsarClientSharedResources.SharedResource.OpenTelemetry));
+ return this;
+ }
+
@Override
public PulsarClientSharedResources build() {
return new PulsarClientSharedResourcesImpl(sharedResources, resourceConfigs, shareConfigured);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesImpl.java
index 29953784681..d6eb459e23e 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientSharedResourcesImpl.java
@@ -21,8 +21,10 @@ package org.apache.pulsar.client.impl;
import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createDnsResolverGroupWithResourceConfig;
import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createEventLoopGroupWithResourceConfig;
import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createExternalExecutorProviderWithResourceConfig;
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createInstrumentProviderWithResourceConfig;
import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createInternalExecutorProviderWithResourceConfig;
import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createLookupExecutorProviderWithResourceConfig;
+import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createMemoryLimitControllerWithResourceConfig;
import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createScheduledExecutorProviderWithResourceConfig;
import static org.apache.pulsar.client.impl.PulsarClientResourcesConfigurer.createTimer;
import io.netty.channel.EventLoopGroup;
@@ -31,12 +33,16 @@ import java.util.EnumSet;
import java.util.Map;
import java.util.Set;
import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientSharedResources;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.apache.pulsar.client.impl.metrics.MemoryBufferStats;
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.client.util.ScheduledExecutorProvider;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
+@Slf4j
@Getter
public class PulsarClientSharedResourcesImpl implements PulsarClientSharedResources {
Set<SharedResource> sharedResources;
@@ -47,6 +53,9 @@ public class PulsarClientSharedResourcesImpl implements PulsarClientSharedResour
private final Timer timer;
private final ExecutorProvider lookupExecutorProvider;
private final DnsResolverGroupImpl dnsResolverGroup;
+ private final MemoryLimitController memoryLimitController;
+ private final InstrumentProvider instrumentProvider;
+ private final MemoryBufferStats memoryBufferStats;
public PulsarClientSharedResourcesImpl(Set<SharedResource> sharedResources,
Map<SharedResource, PulsarClientSharedResourcesBuilderImpl.ResourceConfig>
@@ -91,6 +100,21 @@ public class PulsarClientSharedResourcesImpl implements PulsarClientSharedResour
? createDnsResolverGroupWithResourceConfig(
getResourceConfig(resourceConfigs, SharedResource.DnsResolver))
: null;
+ this.memoryLimitController = this.sharedResources.contains(SharedResource.MemoryLimitController)
+ ? createMemoryLimitControllerWithResourceConfig(
+ getResourceConfig(resourceConfigs, SharedResource.MemoryLimitController))
+ : null;
+ this.instrumentProvider = this.sharedResources.contains(SharedResource.OpenTelemetry)
+ ? createInstrumentProviderWithResourceConfig(
+ getResourceConfig(resourceConfigs, SharedResource.OpenTelemetry))
+ : null;
+ // Only create memory buffer metrics if memory limiting is enabled.
+ if (this.memoryLimitController != null && this.memoryLimitController.isMemoryLimited()
+ && this.instrumentProvider != null) {
+ this.memoryBufferStats = new MemoryBufferStats(this.instrumentProvider, this.memoryLimitController);
+ } else {
+ this.memoryBufferStats = null;
+ }
}
@SuppressWarnings("unchecked")
@@ -132,6 +156,13 @@ public class PulsarClientSharedResourcesImpl implements PulsarClientSharedResour
if (ioEventLoopGroup != null) {
EventLoopUtil.shutdownGracefully(ioEventLoopGroup);
}
+ if (memoryBufferStats != null) {
+ try {
+ memoryBufferStats.close();
+ } catch (Throwable t) {
+ log.warn("Failed to close shared memoryBufferStats", t);
+ }
+ }
}
public void applyTo(PulsarClientImpl.PulsarClientImplBuilder instanceBuilder) {
@@ -156,5 +187,8 @@ public class PulsarClientSharedResourcesImpl implements PulsarClientSharedResour
if (ioEventLoopGroup != null) {
instanceBuilder.eventLoopGroup(ioEventLoopGroup);
}
+ if (memoryLimitController != null) {
+ instanceBuilder.memoryLimitController(memoryLimitController);
+ }
}
}