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/02/11 21:45:41 UTC
(pulsar) branch branch-4.0 updated: [fix][ml] Fix memory leaks in ManagedCursorInfo and ManagedLedgerInfo decompression and compression (#23960)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 6ed9efeef2b [fix][ml] Fix memory leaks in ManagedCursorInfo and ManagedLedgerInfo decompression and compression (#23960)
6ed9efeef2b is described below
commit 6ed9efeef2bd45b7f6134e02e2b617fb8e5bc727
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Feb 11 23:39:13 2025 +0200
[fix][ml] Fix memory leaks in ManagedCursorInfo and ManagedLedgerInfo decompression and compression (#23960)
(cherry picked from commit 7f595cd9dc7522dd597ff58aa8b2a23737c9d841)
---
.../bookkeeper/mledger/impl/MetaStoreImpl.java | 34 ++++++++++++----------
1 file changed, 19 insertions(+), 15 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
index e47443e4e8f..611d9d60202 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java
@@ -456,8 +456,13 @@ public class MetaStoreImpl implements MetaStore, Consumer<Notification> {
try {
MLDataFormats.ManagedLedgerInfoMetadata metadata =
MLDataFormats.ManagedLedgerInfoMetadata.parseFrom(metadataBytes);
- return ManagedLedgerInfo.parseFrom(getCompressionCodec(metadata.getCompressionType())
- .decode(byteBuf, metadata.getUncompressedSize()).nioBuffer());
+ ByteBuf uncompressed = getCompressionCodec(metadata.getCompressionType())
+ .decode(byteBuf, metadata.getUncompressedSize());
+ try {
+ return ManagedLedgerInfo.parseFrom(uncompressed.nioBuffer());
+ } finally {
+ uncompressed.release();
+ }
} catch (Exception e) {
log.error("Failed to parse managedLedgerInfo metadata, "
+ "fall back to parse managedLedgerInfo directly.", e);
@@ -478,8 +483,13 @@ public class MetaStoreImpl implements MetaStore, Consumer<Notification> {
try {
MLDataFormats.ManagedCursorInfoMetadata metadata =
MLDataFormats.ManagedCursorInfoMetadata.parseFrom(metadataBytes);
- return ManagedCursorInfo.parseFrom(getCompressionCodec(metadata.getCompressionType())
- .decode(byteBuf, metadata.getUncompressedSize()).nioBuffer());
+ ByteBuf uncompressed = getCompressionCodec(metadata.getCompressionType())
+ .decode(byteBuf, metadata.getUncompressedSize());
+ try {
+ return ManagedCursorInfo.parseFrom(uncompressed.nioBuffer());
+ } finally {
+ uncompressed.release();
+ }
} catch (Exception e) {
log.error("Failed to parse ManagedCursorInfo metadata, "
+ "fall back to parse ManagedCursorInfo directly", e);
@@ -503,29 +513,23 @@ public class MetaStoreImpl implements MetaStore, Consumer<Notification> {
if (compressionType == null || compressionType.equals(CompressionType.NONE)) {
return info;
}
- ByteBuf metadataByteBuf = null;
- ByteBuf encodeByteBuf = null;
+
+ CompositeByteBuf compositeByteBuf = PulsarByteBufAllocator.DEFAULT.compositeBuffer();
try {
- metadataByteBuf = PulsarByteBufAllocator.DEFAULT.buffer(metadataSerializedSize + 6,
+ ByteBuf metadataByteBuf = PulsarByteBufAllocator.DEFAULT.buffer(metadataSerializedSize + 6,
metadataSerializedSize + 6);
metadataByteBuf.writeShort(MAGIC_MANAGED_INFO_METADATA);
metadataByteBuf.writeInt(metadataSerializedSize);
metadataByteBuf.writeBytes(metadata);
- encodeByteBuf = getCompressionCodec(compressionType)
+ ByteBuf encodeByteBuf = getCompressionCodec(compressionType)
.encode(Unpooled.wrappedBuffer(info));
- CompositeByteBuf compositeByteBuf = PulsarByteBufAllocator.DEFAULT.compositeBuffer();
compositeByteBuf.addComponent(true, metadataByteBuf);
compositeByteBuf.addComponent(true, encodeByteBuf);
byte[] dataBytes = new byte[compositeByteBuf.readableBytes()];
compositeByteBuf.readBytes(dataBytes);
return dataBytes;
} finally {
- if (metadataByteBuf != null) {
- metadataByteBuf.release();
- }
- if (encodeByteBuf != null) {
- encodeByteBuf.release();
- }
+ compositeByteBuf.release();
}
}