BewareMyPower commented on code in PR #24431:
URL: https://github.com/apache/pulsar/pull/24431#discussion_r2158558605
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -2453,33 +2450,6 @@ private void
handleLastMessageIdFromCompactionService(PersistentTopic persistent
});
}
- private int calculateTheLastBatchIndexInBatch(MessageMetadata metadata,
ByteBuf payload) throws IOException {
- int batchSize = metadata.getNumMessagesInBatch();
- if (batchSize <= 1){
- return -1;
- }
- if (metadata.hasCompression()) {
- var tmp = payload;
- CompressionType compressionType = metadata.getCompression();
- CompressionCodec codec =
CompressionCodecProvider.getCompressionCodec(compressionType);
- int uncompressedSize = metadata.getUncompressedSize();
- payload = codec.decode(payload, uncompressedSize);
- tmp.release();
- }
- SingleMessageMetadata singleMessageMetadata = new
SingleMessageMetadata();
- int lastBatchIndexInBatch = -1;
- for (int i = 0; i < batchSize; i++){
- ByteBuf singleMessagePayload =
- Commands.deSerializeSingleMessageInBatch(payload,
singleMessageMetadata, i, batchSize);
- singleMessagePayload.release();
- if (singleMessageMetadata.isCompactedOut()){
- continue;
- }
- lastBatchIndexInBatch = i;
Review Comment:
The key point is here assumes the compaction batch should have
`SingleMessageMetadata` for each single message. The behavior was introduced
from https://github.com/apache/pulsar/pull/18877. However, it requires the
message payload can always have a serialized `SingleMessageMetadata` if the
batch size is greater than 1, which could be too strict.
This comment also makes sense:
https://github.com/apache/pulsar/pull/18877/files#r1165000498
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
