You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/09/18 22:18:05 UTC
(pulsar) branch master updated: [fix][broker] Release the compaction buffers and permits on flush failures (#26576)
This is an automated email from the ASF dual-hosted git repository.
merlimat 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 52ddf4a3bcb7 [fix][broker] Release the compaction buffers and permits on flush failures (#26576)
52ddf4a3bcb7 is described below
commit 52ddf4a3bcb7452a283265dbbf2f398dfeadbcc3
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Sat Sep 19 06:17:59 2026 +0800
[fix][broker] Release the compaction buffers and permits on flush failures (#26576)
---
.../client/impl/RawBatchMessageContainerImpl.java | 12 +++---
.../compaction/StrategicTwoPhaseCompactor.java | 25 ++++++++++-
.../impl/RawBatchMessageContainerImplTest.java | 48 ++++++++++++++++++++++
3 files changed, 78 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java
index 962aa4e1d2bc..be080b7faf3f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImpl.java
@@ -29,7 +29,6 @@ import org.apache.pulsar.client.api.MessageCrypto;
import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
-import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.api.EncryptionContext;
import org.apache.pulsar.common.api.proto.CompressionType;
import org.apache.pulsar.common.api.proto.MessageIdData;
@@ -196,6 +195,7 @@ public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
ByteBuf encryptedPayload = encrypt(getCompressedBatchMetadataAndPayload(false));
ByteBuf metadataAndPayload = null;
+ ByteBuf buf = null;
try {
updateAndReserveBatchAllocatedSize(encryptedPayload.capacity());
metadataAndPayload = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c,
@@ -211,15 +211,17 @@ public class RawBatchMessageContainerImpl extends BatchMessageContainerImpl {
int idSize = idData.getSerializedSize();
int headerSize = 4 /* IdSize */ + idSize + 4 /* metadataAndPayloadSize */;
int totalSize = headerSize + metadataAndPayload.readableBytes();
- ByteBuf buf = PulsarByteBufAllocator.DEFAULT.buffer(totalSize);
+ buf = allocator.buffer(totalSize);
buf.writeInt(idSize);
idData.writeTo(buf);
buf.writeInt(metadataAndPayload.readableBytes());
buf.writeBytes(metadataAndPayload);
- return buf;
+ ByteBuf result = buf;
+ buf = null;
+ return result;
} finally {
- // Release everything allocated for this serialization on both success and failure, so a failure after
- // the batch buffer was built (e.g. an OOM) cannot orphan it.
+ // buf is nulled on the success path, where its ownership moved to the returned buffer.
+ ReferenceCountUtil.safeRelease(buf);
ReferenceCountUtil.safeRelease(metadataAndPayload);
ReferenceCountUtil.safeRelease(encryptedPayload);
clear();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java
index 2c5915375123..5747fe292055 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.compaction;
import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
import java.time.Duration;
import java.util.Iterator;
import java.util.LinkedHashMap;
@@ -452,23 +453,43 @@ public class StrategicTwoPhaseCompactor extends PublishingOrderCompactor {
return CompletableFuture.completedFuture(false);
}
CompletableFuture<Boolean> bkf = new CompletableFuture<>();
+ ByteBuf serialized = null;
+ boolean acquired = false;
try {
- ByteBuf serialized = batchMessageContainer.toByteBuf();
+ serialized = batchMessageContainer.toByteBuf();
outstanding.acquire();
+ acquired = true;
mxBean.addCompactionWriteOp(topic, serialized.readableBytes());
long start = System.nanoTime();
lh.asyncAddEntry(serialized,
(rc, ledger, eid, ctx) -> {
outstanding.release();
- mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
+ try {
+ mxBean.addCompactionLatencyOp(topic, System.nanoTime() - start, TimeUnit.NANOSECONDS);
+ } catch (Throwable t) {
+ // This callback can run synchronously inside asyncAddEntry: an exception escaping
+ // it would reach the caller's catch, which would release the permit and the
+ // buffer a second time - both already belong to the completed write here.
+ log.warn().exception(t).log("Failed to record the compaction write latency");
+ }
if (rc != BKException.Code.OK) {
bkf.completeExceptionally(BKException.create(rc));
} else {
bkf.complete(true);
}
}, null);
+ // Ownership transferred to BookKeeper's write; it releases the buffer when done.
+ serialized = null;
} catch (Throwable t) {
+ // toByteBuf() already cleared the container: nobody else holds the buffer if the hand-off
+ // never happened.
+ ReferenceCountUtil.safeRelease(serialized);
+ // Release the permit only when it was taken: a lost one stalls the next flush on acquire(),
+ // a surplus one raises the outstanding-writes limit.
+ if (acquired) {
+ outstanding.release();
+ }
log.error().exception(t).log("Failed to add entry");
// discard(Exception) cannot take an Error: wrap non-Exception Throwables so the batch container is
// always cleared here and stays reusable after a failed flush.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImplTest.java
index 2fe8b8406693..433d68bfaaed 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawBatchMessageContainerImplTest.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.client.impl;
import static org.apache.pulsar.common.api.proto.CompressionType.NONE;
import static org.apache.pulsar.common.api.proto.CompressionType.ZSTD;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anySet;
@@ -33,6 +34,7 @@ import static org.testng.AssertJUnit.assertTrue;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;
+import io.netty.buffer.WrappedByteBuf;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
@@ -41,6 +43,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.pulsar.client.api.CryptoKeyReader;
import org.apache.pulsar.client.api.EncryptionKeyInfo;
import org.apache.pulsar.client.api.MessageCrypto;
@@ -225,6 +228,51 @@ public class RawBatchMessageContainerImplTest {
buf.release();
}
+ /**
+ * The output buffer of {@code toByteBuf()} is allocated before the header writes; a failure between the
+ * allocation and the return (here: a failing final write) must not orphan it - the finally-block releases
+ * it, since its ownership never moved to the caller.
+ */
+ @Test
+ public void testToByteBufReleasesTheOutputBufferWhenSerializationFails() {
+ List<ByteBuf> outputs = new ArrayList<>();
+ AtomicInteger allocations = new AtomicInteger();
+ ByteBufAllocator allocator = mock(ByteBufAllocator.class);
+ doAnswer(invocation -> {
+ ByteBuf real = Unpooled.buffer((int) invocation.getArgument(0));
+ // The second allocation through the instance allocator is toByteBuf()'s output buffer (the first
+ // is the batch buffer from add()).
+ if (allocations.incrementAndGet() == 2) {
+ ByteBuf failing = new FailingWriteByteBuf(real);
+ outputs.add(failing);
+ return failing;
+ }
+ return real;
+ }).when(allocator).buffer(anyInt());
+
+ RawBatchMessageContainerImpl container = new RawBatchMessageContainerImpl(allocator);
+ container.add(createMessage("my-topic", "hi", 0), null);
+
+ assertThatThrownBy(container::toByteBuf)
+ .hasMessageContaining("mocked write failure");
+ Assert.assertEquals(outputs.size(), 1);
+ Assert.assertEquals(outputs.get(0).refCnt(), 0,
+ "the output buffer must be released when the serialization fails after its allocation");
+ }
+
+ /** Delegates everything but fails the final {@code writeBytes}, standing in for a write that throws. */
+ private static final class FailingWriteByteBuf extends WrappedByteBuf {
+
+ FailingWriteByteBuf(ByteBuf buffer) {
+ super(buffer);
+ }
+
+ @Override
+ public ByteBuf writeBytes(ByteBuf src) {
+ throw new RuntimeException("mocked write failure");
+ }
+ }
+
@Test
public void testToByteBufWithSingleMessage() throws IOException {
RawBatchMessageContainerImpl container = new RawBatchMessageContainerImpl();