Skip to content

Commit d586630

Browse files
BewareMyPowerKannarFr
BewareMyPower
authored and
KannarFr
committed
[improve][broker] PIP-429: Optimize Handling of Compacted Last Entry by Skipping Payload Buffer Parsing (apache#24523)
1 parent 1c84bd9 commit d586630

File tree

10 files changed

+177
-165
lines changed

10 files changed

+177
-165
lines changed

‎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
@@ -41,7 +41,6 @@
4141
import io.netty.util.concurrent.FastThreadLocal;
4242
import io.netty.util.concurrent.Promise;
4343
import io.netty.util.concurrent.ScheduledFuture;
44-
import java.io.IOException;
4544
import java.net.InetSocketAddress;
4645
import java.net.SocketAddress;
4746
import java.util.ArrayList;
@@ -137,7 +136,6 @@
137136
import org.apache.pulsar.common.api.proto.CommandUnsubscribe;
138137
import org.apache.pulsar.common.api.proto.CommandWatchTopicList;
139138
import org.apache.pulsar.common.api.proto.CommandWatchTopicListClose;
140-
import org.apache.pulsar.common.api.proto.CompressionType;
141139
import org.apache.pulsar.common.api.proto.FeatureFlags;
142140
import org.apache.pulsar.common.api.proto.KeySharedMeta;
143141
import org.apache.pulsar.common.api.proto.KeySharedMode;
@@ -148,10 +146,7 @@
148146
import org.apache.pulsar.common.api.proto.ProtocolVersion;
149147
import org.apache.pulsar.common.api.proto.Schema;
150148
import org.apache.pulsar.common.api.proto.ServerError;
151-
import org.apache.pulsar.common.api.proto.SingleMessageMetadata;
152149
import org.apache.pulsar.common.api.proto.TxnAction;
153-
import org.apache.pulsar.common.compression.CompressionCodec;
154-
import org.apache.pulsar.common.compression.CompressionCodecProvider;
155150
import org.apache.pulsar.common.intercept.InterceptException;
156151
import org.apache.pulsar.common.lookup.data.LookupData;
157152
import org.apache.pulsar.common.naming.Metadata;
@@ -2282,7 +2277,7 @@ protected void handleGetLastMessageId(CommandGetLastMessageId getLastMessageId)
22822277
.thenApply(lastPosition -> {
22832278
int partitionIndex = TopicName.getPartitionIndex(topic.getName());
22842279

2285-
Position markDeletePosition = null;
2280+
Position markDeletePosition = PositionFactory.EARLIEST;
22862281
if (consumer.getSubscription() instanceof PersistentSubscription) {
22872282
markDeletePosition = ((PersistentSubscription) consumer.getSubscription()).getCursor()
22882283
.getMarkDeletedPosition();
@@ -2343,8 +2338,7 @@ private void getLargestBatchIndexWhenPossible(
23432338
} else {
23442339
// if readCompacted is false, we need to return MessageId.earliest
23452340
writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, -1, -1, partitionIndex, -1,
2346-
markDeletePosition != null ? markDeletePosition.getLedgerId() : -1,
2347-
markDeletePosition != null ? markDeletePosition.getEntryId() : -1));
2341+
markDeletePosition.getLedgerId(), markDeletePosition.getEntryId()));
23482342
}
23492343
return;
23502344
}
@@ -2403,47 +2397,19 @@ public String toString() {
24032397

24042398
writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, lastPosition.getLedgerId(),
24052399
lastPosition.getEntryId(), partitionIndex, largestBatchIndex,
2406-
markDeletePosition != null ? markDeletePosition.getLedgerId() : -1,
2407-
markDeletePosition != null ? markDeletePosition.getEntryId() : -1));
2400+
markDeletePosition.getLedgerId(), markDeletePosition.getEntryId()));
24082401
}
24092402
});
24102403
});
24112404
}
24122405

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

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