Skip to content

Commit 4e9acda

Browse files
BewareMyPowerBewareMyPower
committed
[improve][broker] PIP-429: Optimize Handling of Compacted Last Entry by Skipping Payload Buffer Parsing
1 parent 994b1be commit 4e9acda

10 files changed

Lines changed: 177 additions & 160 deletions

File tree

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java‎

Lines changed: 8 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -2837,31 +2837,16 @@ protected CompletableFuture<MessageId> internalGetMessageIdByTimestampAsync(long
28372837
"Get message ID by timestamp on a non-persistent topic is not allowed");
28382838
}
28392839
final PersistentTopic persistentTopic = (PersistentTopic) topic;
2840+
final var compactionService = persistentTopic.getTopicCompactionService();
28402841

2841-
return persistentTopic.getTopicCompactionService().readLastCompactedEntry().thenCompose(lastEntry -> {
2842-
if (lastEntry == null) {
2843-
return findMessageIdByPublishTime(timestamp, persistentTopic.getManagedLedger());
2844-
}
2845-
MessageMetadata metadata;
2846-
Position position = lastEntry.getPosition();
2847-
try {
2848-
metadata = Commands.parseMessageMetadata(lastEntry.getDataBuffer());
2849-
} finally {
2850-
lastEntry.release();
2851-
}
2852-
if (timestamp == metadata.getPublishTime()) {
2853-
return CompletableFuture.completedFuture(new MessageIdImpl(position.getLedgerId(),
2854-
position.getEntryId(), topicName.getPartitionIndex()));
2855-
} else if (timestamp < metadata.getPublishTime()) {
2842+
return compactionService.getLastMessagePosition().thenCompose(messagePosition -> {
2843+
if (timestamp == messagePosition.publishTime()) {
2844+
return CompletableFuture.completedFuture(new MessageIdImpl(messagePosition.ledgerId(),
2845+
messagePosition.entryId(), topicName.getPartitionIndex()));
2846+
} else if (timestamp < messagePosition.publishTime()) {
28562847
return persistentTopic.getTopicCompactionService().findEntryByPublishTime(timestamp)
2857-
.thenApply(compactedEntry -> {
2858-
try {
2859-
return new MessageIdImpl(compactedEntry.getLedgerId(),
2860-
compactedEntry.getEntryId(), topicName.getPartitionIndex());
2861-
} finally {
2862-
compactedEntry.release();
2863-
}
2864-
});
2848+
.thenApply(__ -> new MessageIdImpl(__.getLedgerId(), __.getEntryId(),
2849+
topicName.getPartitionIndex()));
28652850
} else {
28662851
return findMessageIdByPublishTime(timestamp, persistentTopic.getManagedLedger());
28672852
}

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java‎

Lines changed: 8 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,6 @@
4242
import io.netty.util.concurrent.FastThreadLocal;
4343
import io.netty.util.concurrent.Promise;
4444
import io.netty.util.concurrent.ScheduledFuture;
45-
import java.io.IOException;
4645
import java.net.InetSocketAddress;
4746
import java.net.SocketAddress;
4847
import java.util.ArrayList;
@@ -138,7 +137,6 @@
138137
import org.apache.pulsar.common.api.proto.CommandUnsubscribe;
139138
import org.apache.pulsar.common.api.proto.CommandWatchTopicList;
140139
import org.apache.pulsar.common.api.proto.CommandWatchTopicListClose;
141-
import org.apache.pulsar.common.api.proto.CompressionType;
142140
import org.apache.pulsar.common.api.proto.FeatureFlags;
143141
import org.apache.pulsar.common.api.proto.KeySharedMeta;
144142
import org.apache.pulsar.common.api.proto.KeySharedMode;
@@ -149,10 +147,7 @@
149147
import org.apache.pulsar.common.api.proto.ProtocolVersion;
150148
import org.apache.pulsar.common.api.proto.Schema;
151149
import org.apache.pulsar.common.api.proto.ServerError;
152-
import org.apache.pulsar.common.api.proto.SingleMessageMetadata;
153150
import org.apache.pulsar.common.api.proto.TxnAction;
154-
import org.apache.pulsar.common.compression.CompressionCodec;
155-
import org.apache.pulsar.common.compression.CompressionCodecProvider;
156151
import org.apache.pulsar.common.intercept.InterceptException;
157152
import org.apache.pulsar.common.lookup.data.LookupData;
158153
import org.apache.pulsar.common.naming.Metadata;
@@ -2283,7 +2278,7 @@ protected void handleGetLastMessageId(CommandGetLastMessageId getLastMessageId)
22832278
.thenApply(lastPosition -> {
22842279
int partitionIndex = TopicName.getPartitionIndex(topic.getName());
22852280

2286-
Position markDeletePosition = null;
2281+
Position markDeletePosition = PositionFactory.EARLIEST;
22872282
if (consumer.getSubscription() instanceof PersistentSubscription) {
22882283
markDeletePosition = ((PersistentSubscription) consumer.getSubscription()).getCursor()
22892284
.getMarkDeletedPosition();
@@ -2344,8 +2339,7 @@ private void getLargestBatchIndexWhenPossible(
23442339
} else {
23452340
// if readCompacted is false, we need to return MessageId.earliest
23462341
writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, -1, -1, partitionIndex, -1,
2347-
markDeletePosition != null ? markDeletePosition.getLedgerId() : -1,
2348-
markDeletePosition != null ? markDeletePosition.getEntryId() : -1));
2342+
markDeletePosition.getLedgerId(), markDeletePosition.getEntryId()));
23492343
}
23502344
return;
23512345
}
@@ -2404,47 +2398,19 @@ public String toString() {
24042398

24052399
writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, lastPosition.getLedgerId(),
24062400
lastPosition.getEntryId(), partitionIndex, largestBatchIndex,
2407-
markDeletePosition != null ? markDeletePosition.getLedgerId() : -1,
2408-
markDeletePosition != null ? markDeletePosition.getEntryId() : -1));
2401+
markDeletePosition.getLedgerId(), markDeletePosition.getEntryId()));
24092402
}
24102403
});
24112404
});
24122405
}
24132406

24142407
private void handleLastMessageIdFromCompactionService(PersistentTopic persistentTopic, long requestId,
24152408
int partitionIndex, Position markDeletePosition) {
2416-
persistentTopic.getTopicCompactionService().readLastCompactedEntry().thenAccept(entry -> {
2417-
if (entry != null) {
2418-
try {
2419-
// in this case, all the data has been compacted, so return the last position
2420-
// in the compacted ledger to the client
2421-
ByteBuf payload = entry.getDataBuffer();
2422-
MessageMetadata metadata = Commands.parseMessageMetadata(payload);
2423-
int largestBatchIndex;
2424-
try {
2425-
largestBatchIndex = calculateTheLastBatchIndexInBatch(metadata, payload);
2426-
} catch (IOException ioEx) {
2427-
writeAndFlush(Commands.newError(requestId, ServerError.MetadataError,
2428-
"Failed to deserialize batched message from the last entry of the compacted Ledger: "
2429-
+ ioEx.getMessage()));
2430-
return;
2431-
}
2432-
writeAndFlush(Commands.newGetLastMessageIdResponse(requestId,
2433-
entry.getLedgerId(), entry.getEntryId(), partitionIndex, largestBatchIndex,
2434-
markDeletePosition != null ? markDeletePosition.getLedgerId() : -1,
2435-
markDeletePosition != null ? markDeletePosition.getEntryId() : -1));
2436-
} finally {
2437-
entry.release();
2438-
}
2439-
} else {
2440-
// in this case, the ledgers been removed except the current ledger
2441-
// and current ledger without any data
2442-
writeAndFlush(Commands.newGetLastMessageIdResponse(requestId,
2443-
-1, -1, partitionIndex, -1,
2444-
markDeletePosition != null ? markDeletePosition.getLedgerId() : -1,
2445-
markDeletePosition != null ? markDeletePosition.getEntryId() : -1));
2446-
}
2447-
}).exceptionally(ex -> {
2409+
persistentTopic.getTopicCompactionService().getLastMessagePosition().thenAccept(position ->
2410+
writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, position.ledgerId(), position.entryId(),
2411+
partitionIndex, position.batchIndex(), markDeletePosition.getLedgerId(),
2412+
markDeletePosition.getEntryId()))
2413+
).exceptionally(ex -> {
24482414
writeAndFlush(Commands.newError(
24492415
requestId, ServerError.MetadataError,
24502416
"Failed to read last entry of the compacted Ledger "
@@ -2453,33 +2419,6 @@ private void handleLastMessageIdFromCompactionService(PersistentTopic persistent
24532419
});
24542420
}
24552421

2456-
private int calculateTheLastBatchIndexInBatch(MessageMetadata metadata, ByteBuf payload) throws IOException {
2457-
int batchSize = metadata.getNumMessagesInBatch();
2458-
if (batchSize <= 1){
2459-
return -1;
2460-
}
2461-
if (metadata.hasCompression()) {
2462-
var tmp = payload;
2463-
CompressionType compressionType = metadata.getCompression();
2464-
CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(compressionType);
2465-
int uncompressedSize = metadata.getUncompressedSize();
2466-
payload = codec.decode(payload, uncompressedSize);
2467-
tmp.release();
2468-
}
2469-
SingleMessageMetadata singleMessageMetadata = new SingleMessageMetadata();
2470-
int lastBatchIndexInBatch = -1;
2471-
for (int i = 0; i < batchSize; i++){
2472-
ByteBuf singleMessagePayload =
2473-
Commands.deSerializeSingleMessageInBatch(payload, singleMessageMetadata, i, batchSize);
2474-
singleMessagePayload.release();
2475-
if (singleMessageMetadata.isCompactedOut()){
2476-
continue;
2477-
}
2478-
lastBatchIndexInBatch = i;
2479-
}
2480-
return lastBatchIndexInBatch;
2481-
}
2482-
24832422
private CompletableFuture<Boolean> isNamespaceOperationAllowed(NamespaceName namespaceName,
24842423
NamespaceOperation operation) {
24852424
if (!service.isAuthorizationEnabled()) {

‎pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -151,7 +151,7 @@ public static Optional<RawMessage> rebatchMessage(RawMessage msg,
151151
ByteBuf uncompressedPayload = codec.decode(payload, uncompressedSize);
152152
try {
153153
int batchSize = metadata.getNumMessagesInBatch();
154-
int messagesRetained = 0;
154+
final var retainedBatchIndexes = new ArrayList<Integer>();
155155

156156
SingleMessageMetadata emptyMetadata = new SingleMessageMetadata().setCompactedOut(true);
157157
SingleMessageMetadata singleMessageMetadata = new SingleMessageMetadata();
@@ -169,7 +169,7 @@ public static Optional<RawMessage> rebatchMessage(RawMessage msg,
169169
Unpooled.EMPTY_BUFFER, batchBuffer);
170170
} else if (!singleMessageMetadata.hasPartitionKey()) {
171171
if (retainNullKey) {
172-
messagesRetained++;
172+
retainedBatchIndexes.add(i);
173173
Commands.serializeSingleMessageInBatchWithPayload(singleMessageMetadata,
174174
singleMessagePayload, batchBuffer);
175175
} else {
@@ -178,7 +178,7 @@ public static Optional<RawMessage> rebatchMessage(RawMessage msg,
178178
}
179179
} else if (filter.test(singleMessageMetadata.getPartitionKey(), id)
180180
&& singleMessagePayload.readableBytes() > 0) {
181-
messagesRetained++;
181+
retainedBatchIndexes.add(i);
182182
Commands.serializeSingleMessageInBatchWithPayload(singleMessageMetadata,
183183
singleMessagePayload, batchBuffer);
184184
} else {
@@ -189,10 +189,14 @@ public static Optional<RawMessage> rebatchMessage(RawMessage msg,
189189
singleMessagePayload.release();
190190
}
191191

192-
if (messagesRetained > 0) {
192+
if (!retainedBatchIndexes.isEmpty()) {
193193
int newUncompressedSize = batchBuffer.readableBytes();
194194
ByteBuf compressedPayload = codec.encode(batchBuffer);
195195

196+
metadata.clearCompactedBatchIndexes();
197+
for (int index : retainedBatchIndexes) {
198+
metadata.addCompactedBatchIndexe(index);
199+
}
196200
metadata.setUncompressedSize(newUncompressedSize);
197201

198202
ByteBuf metadataAndPayload = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c,

‎pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -345,18 +345,23 @@ CompletableFuture<Entry> findFirstMatchEntry(final Predicate<Entry> predicate) {
345345
var compactedTopicContextFuture = this.getCompactedTopicContextFuture();
346346

347347
if (compactedTopicContextFuture == null) {
348-
return CompletableFuture.completedFuture(null);
348+
return CompletableFuture.failedFuture(new IllegalStateException(
349+
"CompactedTopicContext is not initialized"));
349350
}
350351
return compactedTopicContextFuture.thenCompose(compactedTopicContext -> {
351352
LedgerHandle lh = compactedTopicContext.getLedger();
352353
CompletableFuture<Long> promise = new CompletableFuture<>();
353354
findFirstMatchIndexLoop(predicate, 0L, lh.getLastAddConfirmed(), promise, null, lh);
354-
return promise.thenCompose(index -> {
355-
if (index == null) {
356-
return CompletableFuture.completedFuture(null);
355+
return promise.thenCompose(index -> readEntries(lh, index, index).thenApply(entries -> {
356+
if (entries.size() != 1) {
357+
for (final var entry : entries) {
358+
entry.release();
359+
}
360+
throw new IllegalStateException("Read " + entries.size() + " entries from the compacted ledger "
361+
+ lh + " entry " + index);
357362
}
358-
return readEntries(lh, index, index).thenApply(entries -> entries.get(0));
359-
});
363+
return entries.get(0);
364+
}));
360365
});
361366
}
362367
private static void findFirstMatchIndexLoop(final Predicate<Entry> predicate,

0 commit comments

Comments
 (0)