BewareMyPower commented on code in PR #24431:
URL: https://github.com/apache/pulsar/pull/24431#discussion_r2158542182
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java:
##########
@@ -104,4 +114,60 @@ public static void
asyncReadCompactedEntries(TopicCompactionService topicCompact
return null;
});
}
+
+ public static int calculateTheLastBatchIndexInBatch(MessageMetadata
metadata, ByteBuf payload) throws IOException {
+ int retainedMessageCount = getRetainedMessageCount(metadata);
+ if (retainedMessageCount > 0) {
+ return retainedMessageCount;
+ }
+ if (retainedMessageCount == 0) {
+ return -1;
+ }
+
+ 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;
+ }
+ return lastBatchIndexInBatch;
+ }
+
+ /**
+ * Use `-1` as default value if the property is not set or cannot be
parsed.
+ */
+ private static int getRetainedMessageCount(MessageMetadata metadata) {
+ if (CollectionUtils.isEmpty(metadata.getPropertiesList())) {
+ return -1;
+ }
+ for (KeyValue kv : metadata.getPropertiesList()) {
+ if (RETAINED_MESSAGE_COUNT_PROPERTY.equals(kv.getKey())) {
+ try {
+ return Integer.parseInt(kv.getValue());
+ } catch (NumberFormatException e) {
+ // Ignore and return default value
+ return -1;
+ }
Review Comment:
IMO, this hacky behavior should be avoided. It gives a new meaning that if
the value of "RetainedMessageCount" property is a valid integer, the value will
represent the number of messages in the batch.
--
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]
