You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/04/29 19:57:55 UTC
(pulsar) branch branch-3.3 updated: [fix][broker] Fix ByteBuf memory leak in REST API for publishing messages (#24228)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new 5d90dbf18ff [fix][broker] Fix ByteBuf memory leak in REST API for publishing messages (#24228)
5d90dbf18ff is described below
commit 5d90dbf18ff2dca133ba92d972fa9e0d2a989f70
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Apr 29 18:19:35 2025 +0300
[fix][broker] Fix ByteBuf memory leak in REST API for publishing messages (#24228)
(cherry picked from commit 0c8e88cf983a8793a117f05a14c37c755464ca89)
---
.../main/java/org/apache/pulsar/broker/rest/TopicsBase.java | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
index 20c35b4f776..3205eed6993 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
@@ -61,6 +61,7 @@ import org.apache.pulsar.broker.authentication.AuthenticationParameters;
import org.apache.pulsar.broker.lookup.LookupResult;
import org.apache.pulsar.broker.namespace.LookupOptions;
import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.schema.SchemaRegistry;
import org.apache.pulsar.broker.service.schema.exceptions.SchemaException;
import org.apache.pulsar.broker.web.RestException;
@@ -280,8 +281,14 @@ public class TopicsBase extends PersistentTopicsBase {
.remove(topicName.getPartitionIndex());
} else {
try {
- t.get().publishMessage(messageToByteBuf(message),
- RestMessagePublishContext.get(publishResult, t.get(), System.nanoTime()));
+ ByteBuf headersAndPayload = messageToByteBuf(message);
+ try {
+ Topic topicObj = t.get();
+ topicObj.publishMessage(headersAndPayload,
+ RestMessagePublishContext.get(publishResult, topicObj, System.nanoTime()));
+ } finally {
+ headersAndPayload.release();
+ }
} catch (Exception e) {
if (log.isDebugEnabled()) {
log.debug("Fail to publish single messages to topic {}: {} ",