[improve][pip] PIP-429: Optimize Handling of Compacted Last Entry by Skipping Payload Buffer Parsing

[BewareMyPower]

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[gaoran10]

I prefer to make changes on the server side for several reasons.

• Broker can handle this issue well and provide a compatibility solution.
• Add a flag for the valid compacted position can avoid deserializing and uncompressing the entire batch messages metadata header.
• If making this change at the client-side, it increases the complexity for the client, and all language clients need to change.
• It's weird to return EntryBuffer when getting the last message ID.
• Actually, Pulsar only needs to retain valid message data in a compacted entry, but it retains all compacted messages with "empty header" and "empty payload".

[BewareMyPower]

If making this change at the client-side, it increases the complexity for the client, and all language clients need to change.

Please check the Backward & Forward Compatibility section

When an older client sends a GetLastMessageId request to a newer broker, the broker computes the last message ID using its standard process. To ensure backward compatibility, the broker excludes the payload buffer from the response, aligning with the expectations of older clients.

[BewareMyPower]

Actually, Pulsar only needs to retain valid message data in a compacted entry, but it retains all compacted messages with "empty header" and "empty payload".

Yes. Unfortunately, it's limited by client side's logic that the default logic to parse the payload buffer looks like:

int batchSize = msgMetadata.getNumMessagesInBatch();
for (int i = 0; i < batchSize; i++) {
    int batchIndex = i;
    final var singleMessageMetadata = parse(payload);
    if (singleMessageMetadata.isCompactedOut()) {
        break;
    }
    // Create a message, whose batch index is i, from the payload buffer
}

Add a flag for the valid compacted position can avoid deserializing and uncompressing the entire batch messages metadata header.

This makes sense to me. Adding a new field to MessageMetadata might be more simple. But we need to take care of the compatibility issue. We can simplify the payload format when this field is set. Let me try a demo first.

[coderzc]

I prefer to make changes on the server side for several reasons.

• Broker can handle this issue well and provide a compatibility solution.
• Add a flag for the valid compacted position can avoid deserializing and uncompressing the entire batch messages metadata header.
• If making this change at the client-side, it increases the complexity for the client, and all language clients need to change.
• It's weird to return EntryBuffer when getting the last message ID.
• Actually, Pulsar only needs to retain valid message data in a compacted entry, but it retains all compacted messages with "empty header" and "empty payload".

If we purpose are to ensure that the compaction task is successful, we only need to check lastCompactedPosition and lastDispatchablePosition before trigger compaction task. If lastDispatchablePosition <= lastCompactedPosition, then I just need to skip the compact task because there is no new data need to be compacted.

If we need make the Pulsar reader to read Kafka format data, then we need this change.

[BewareMyPower]

If we need make the Pulsar reader to read Kafka format data, then we need this change.

No. We need to ensure the GetLastMessageId RPC works even if the last entry is compacted with a non-default payload format.

The consumer is able to configure a MessagePayloadProcessor to parse the message of different format. The reader does not have the configuration and I will use another PR to add it.

[dao-jun]

LGTM, good change!

Addressed