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 2026/09/23 08:34:01 UTC
(pulsar) 22/27: [fix][broker] Release the compaction buffers and permits on flush failures (#26576)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 38384a36bc0f4642e6ad6aaf577e961605332088
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)
(cherry picked from commit 52ddf4a3bcb7452a283265dbbf2f398dfeadbcc3)
---
.../client/impl/RawBatchMessageContainerImpl.java | 12 ++++---
.../compaction/StrategicTwoPhaseCompactor.java | 25 +++++++++++++--
.../impl/RawBatchMessageContainerImplTest.java | 37 ++++++++++++++++++++++
3 files changed, 67 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 c0cfd67ce4db..c051e48c422c 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;
@@ -440,23 +441,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("Failed to record the compaction write latency", t);
+ }
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("Failed to add entry", t);
// 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 96a69ca61616..5c417444d586 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,12 +21,14 @@ 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;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import static org.testng.AssertJUnit.assertFalse;
import static org.testng.AssertJUnit.assertTrue;
@@ -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;
@@ -223,6 +226,40 @@ 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) {
+ // Delegates everything but fails the final writeBytes, standing in for a write that throws.
+ ByteBuf failing = spy(real);
+ doThrow(new RuntimeException("mocked write failure")).when(failing).writeBytes(any(ByteBuf.class));
+ 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");
+ }
+
@Test
public void testToByteBufWithSingleMessage() throws IOException {
RawBatchMessageContainerImpl container = new RawBatchMessageContainerImpl();