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/10 23:56:27 UTC
(pulsar) 05/18: [improve][client] Avoid unused send-stat snapshots for default callbacks (#26519)
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 3441d33bf11803578da9f713b808f7e2d414a469
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Sep 10 17:30:51 2026 +0300
[improve][client] Avoid unused send-stat snapshots for default callbacks (#26519)
(cherry picked from commit 5b09e52cf57d78720fee5c67e07b70f2bf606255)
---
.../client/impl/SendCompletionStatsBenchmark.java | 109 +++++++++++++++++++++
.../apache/pulsar/client/impl/ProducerImpl.java | 31 +++---
.../pulsar/client/impl/ProducerImplTest.java | 52 ++++++++++
3 files changed, 181 insertions(+), 11 deletions(-)
diff --git a/microbench/src/main/java/org/apache/pulsar/client/impl/SendCompletionStatsBenchmark.java b/microbench/src/main/java/org/apache/pulsar/client/impl/SendCompletionStatsBenchmark.java
new file mode 100644
index 000000000000..1c243e327633
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/client/impl/SendCompletionStatsBenchmark.java
@@ -0,0 +1,109 @@
+/*
+ * 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.impl;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+/** Measures actual OpSendMsg completion for custom callbacks; guards the snapshot-preserving fallback. */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(2)
+public class SendCompletionStatsBenchmark {
+ @Param({"false", "true"})
+ public boolean retainStats;
+
+ private MessageImpl<byte[]> message;
+ private ProducerImpl.OpSendMsg operation;
+ private StatsCallback callback;
+
+ @Setup
+ public void setup() {
+ message = MessageImpl.create(new MessageMetadata(), ByteBuffer.wrap(new byte[128]), Schema.BYTES, "benchmark");
+ callback = new StatsCallback();
+ operation = ProducerImpl.OpSendMsg.create(LatencyHistogram.NOOP, message, null, 123, callback);
+ operation.updateSentTimestamp();
+ }
+
+ @Benchmark
+ public long complete() {
+ operation.sendComplete(null);
+ return callback.observed;
+ }
+
+ @TearDown
+ public void tearDown() {
+ operation.recycle();
+ message.getDataBuffer().release();
+ message.recycle();
+ }
+
+ private class StatsCallback implements SendCallback {
+ private long observed;
+ private OpSendMsgStats retained;
+
+ @Override
+ public void sendComplete(Throwable error, OpSendMsgStats stats) {
+ observed = stats.getSequenceId() + stats.getUncompressedSize() + stats.getRetryCount();
+ if (retainStats) {
+ retained = stats;
+ }
+ }
+
+ @Override
+ public void addCallback(MessageImpl<?> message, SendCallback callback) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SendCallback getNextSendCallback() {
+ return null;
+ }
+
+ @Override
+ public MessageImpl<?> getNextMessage() {
+ return null;
+ }
+
+ @Override
+ public CompletableFuture<MessageId> getFuture() {
+ return null;
+ }
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 130dfb64fea9..fdb3ebf1f854 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -445,6 +445,10 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
@Override
public void sendComplete(Throwable e, OpSendMsgStats opSendMsgStats) {
+ sendComplete(e);
+ }
+
+ private void sendComplete(Throwable e) {
SendCallback loopingCallback = this;
MessageImpl<?> loopingMsg = currentMsg;
while (loopingCallback != null) {
@@ -1708,17 +1712,22 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
rpcLatencyHistogram.recordFailure(now - this.lastSentAt);
}
- OpSendMsgStats opSendMsgStats = OpSendMsgStatsImpl.builder()
- .uncompressedSize(uncompressedSize)
- .sequenceId(sequenceId)
- .retryCount(retryCount)
- .batchSizeByte(batchSizeByte)
- .numMessagesInBatch(numMessagesInBatch)
- .highestSequenceId(highestSequenceId)
- .totalChunks(totalChunks)
- .chunkId(chunkId)
- .build();
- callback.sendComplete(finalEx, opSendMsgStats);
+ if (callback instanceof ProducerImpl<?>.DefaultSendMessageCallback defaultCallback) {
+ // The client's default callback does not use the per-operation snapshot.
+ defaultCallback.sendComplete(finalEx);
+ } else {
+ OpSendMsgStats opSendMsgStats = OpSendMsgStatsImpl.builder()
+ .uncompressedSize(uncompressedSize)
+ .sequenceId(sequenceId)
+ .retryCount(retryCount)
+ .batchSizeByte(batchSizeByte)
+ .numMessagesInBatch(numMessagesInBatch)
+ .highestSequenceId(highestSequenceId)
+ .totalChunks(totalChunks)
+ .chunkId(chunkId)
+ .build();
+ callback.sendComplete(finalEx, opSendMsgStats);
+ }
}
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
index 0b832e414d9d..15473df89c9c 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
@@ -33,10 +33,62 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.ByteBufPair;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class ProducerImplTest {
+ @DataProvider
+ public Object[][] completionErrors() {
+ return new Object[][] {{null}, {new PulsarClientException.TimeoutException("send timed out", 41)}};
+ }
+
+ @Test(dataProvider = "completionErrors")
+ public void testCustomCompletionStatsSurviveOperationRecycle(Exception error) {
+ MessageImpl<byte[]> message = MessageImpl.create(new MessageMetadata(),
+ ByteBuffer.wrap(new byte[128]), Schema.BYTES, "test-topic");
+ SendCallback callback = mock(SendCallback.class);
+ ProducerImpl.OpSendMsg operation = ProducerImpl.OpSendMsg.create(
+ LatencyHistogram.NOOP, message, null, 41, callback);
+ OpSendMsgStats snapshot;
+ Throwable completionError;
+ try {
+ operation.updateSentTimestamp();
+ operation.retryCount = 2;
+ operation.batchSizeByte = 1024;
+ operation.numMessagesInBatch = 5;
+ operation.highestSequenceId = 45;
+ operation.totalChunks = 2;
+ operation.chunkId = 1;
+ operation.sendComplete(error);
+ ArgumentCaptor<OpSendMsgStats> statsCaptor = ArgumentCaptor.forClass(OpSendMsgStats.class);
+ ArgumentCaptor<Throwable> errorCaptor = ArgumentCaptor.forClass(Throwable.class);
+ verify(callback).sendComplete(errorCaptor.capture(), statsCaptor.capture());
+ snapshot = statsCaptor.getValue();
+ completionError = errorCaptor.getValue();
+ } finally {
+ operation.recycle();
+ message.getDataBuffer().release();
+ message.recycle();
+ }
+ assertEquals(snapshot.getUncompressedSize(), 128L);
+ assertEquals(snapshot.getSequenceId(), 41L);
+ assertEquals(snapshot.getRetryCount(), 2);
+ assertEquals(snapshot.getBatchSizeByte(), 1024L);
+ assertEquals(snapshot.getNumMessagesInBatch(), 5);
+ assertEquals(snapshot.getHighestSequenceId(), 45L);
+ assertEquals(snapshot.getTotalChunks(), 2);
+ assertEquals(snapshot.getChunkId(), 1);
+ if (error == null) {
+ assertNull(completionError);
+ } else {
+ assertTrue(completionError instanceof PulsarClientException.TimeoutException);
+ assertEquals(((PulsarClientException.TimeoutException) completionError).getSequenceId(), 41L);
+ assertTrue(completionError.getMessage().contains("retryCount 2"));
+ }
+ }
+
@Test
public void testChunkedMessageCtxDeallocate() {
int totalChunks = 3;