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/10/17 20:20:12 UTC
(pulsar) 18/22: [fix][broker] Use the uncompressed payload when finding the last message position of a compacted batch (#26962)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4fd88619bfdcc0218cb4ba2d1c7ae81739129ffe
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Oct 17 09:42:12 2026 -0700
[fix][broker] Use the uncompressed payload when finding the last message position of a compacted batch (#26962)
(cherry picked from commit a2c5e03f894229daf465cf9965bbd3027d08e697)
---
.../compaction/PulsarTopicCompactionService.java | 2 +-
.../apache/pulsar/compaction/CompactorTest.java | 42 ++++++++++++++++++++++
2 files changed, 43 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
index ea724cb6480e..15712c502cfe 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
@@ -156,7 +156,7 @@ public class PulsarTopicCompactionService implements TopicCompactionService {
SingleMessageMetadata singleMessageMetadata = new SingleMessageMetadata();
int batchIndex = -1;
for (int i = 0; i < batchSize; i++){
- final var singleMessagePayload = Commands.deSerializeSingleMessageInBatch(payload,
+ final var singleMessagePayload = Commands.deSerializeSingleMessageInBatch(uncompressedPayload,
singleMessageMetadata, i, batchSize);
singleMessagePayload.release();
if (singleMessageMetadata.isCompactedOut()){
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
index 108ef985f7fc..97b31f293011 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
@@ -54,6 +54,7 @@ import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.stats.OpenTelemetryTopicStats;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
import org.apache.pulsar.client.admin.LongRunningProcessStatus;
+import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRoutingMode;
@@ -68,6 +69,7 @@ import org.apache.pulsar.client.impl.RawMessageImpl;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
import org.awaitility.Awaitility;
import org.mockito.Mockito;
@@ -408,6 +410,46 @@ public class CompactorTest extends MockedPulsarServiceBaseTest {
future.join();
}
+ @Test
+ public void testLastMessagePositionOfCompressedBatch() throws Exception {
+ String topic = BrokerTestUtil.newUniqueName(
+ "persistent://my-property/my-ns/testLastMessagePositionOfCompressedBatch");
+ final int numMessages = 5;
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic)
+ .compressionType(CompressionType.LZ4)
+ .compressionMinMsgBodySize(0)
+ .enableBatching(true)
+ .batchingMaxMessages(numMessages)
+ .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>();
+ for (int i = 0; i < numMessages; i++) {
+ // a compressible value so that the compressed batch payload differs from the uncompressed one
+ futures.add(producer.newMessage().key("key-" + i).value(("value-" + i).repeat(20).getBytes())
+ .sendAsync());
+ }
+ FutureUtil.waitForAll(futures).get();
+
+ PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
+ PulsarTopicCompactionService topicCompactionService =
+ (PulsarTopicCompactionService) persistentTopic.getTopicCompactionService();
+
+ long compactedLedgerId = compact(topic);
+ Awaitility.await().untilAsserted(() -> {
+ Optional<CompactedTopicContext> compactedTopicContext = topicCompactionService.getCompactedTopic()
+ .getCompactedTopicContext();
+ assertThat(compactedTopicContext).isPresent();
+ assertThat(compactedTopicContext.get().ledger.getId()).isEqualTo(compactedLedgerId);
+ });
+
+ // the last entry of the compacted ledger is a compressed batch of numMessages messages
+ final var lastMessagePosition = topicCompactionService.getLastMessagePosition().get();
+ assertThat(lastMessagePosition.batchIndex()).isEqualTo(numMessages - 1);
+ }
+
public ByteBuf extractPayload(RawMessage m) throws Exception {
ByteBuf payloadAndMetadata = m.getHeadersAndPayload();
Commands.skipChecksumIfPresent(payloadAndMetadata);