[PR] [fix][broker] Use the uncompressed payload when finding the last message position of a compacted batch [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari opened a new pull request, #26962:
URL: https://github.com/apache/pulsar/pull/26962

   ### Motivation
   
   `PulsarTopicCompactionService.getLastMessagePosition()` decompresses the last entry of the compacted ledger into `uncompressedPayload`, but then passes the still-compressed `payload` to `Commands.deSerializeSingleMessageInBatch()`. Parsing compressed bytes as `SingleMessageMetadata` fails with errors such as `Invalid unknonwn tag type: 4` or `Truncated protobuf message`.
   
   `ServerCnx.handleLastMessageIdFromCompactionService` uses this code path. It handles `getLastMessageId` for `readCompacted` consumers and readers when reading the topic's last entry fails with a `NonRecoverableLedgerException`. The bug only shows when the last compacted entry is a compressed batch of more than one message with no compacted batch indexes. `StrategicTwoPhaseCompactor` writes exactly that for the ExtensibleLoadManager's `loadbalancer-service-unit-state` topic, which uses ZSTD.
   
   Seen in branch-4.2 CI (LoadBalance integration tests, `ExtensibleLoadManagerTest.startBroker` timeout). After `testStopBroker`, the restarted broker kept failing to start with `ServiceUnitStateChannelImpl - Failed to start the channel`. The cause was a `BrokerMetadataException: Failed to read last entry of the compacted Ledger Invalid unknonwn tag type: 4` from the broker that owned the service-unit-state topic. The broker restarted in a loop until the test timed out.
   
   ### Modifications
   
   - Deserialize the single messages of the batch from `uncompressedPayload`.
   
   ### Verifying this change
   
   - [ ] Make sure that the change passes the CI checks.
   
   This change added tests and can be verified as follows:
   
   - Added `CompactorTest#testLastMessagePositionOfCompressedBatch`, which runs for all three compactors (`CompactorTest`, `StrategicCompactorTest`, `EventTimeOrderCompactorTest`). It publishes one LZ4-compressed batch of 5 messages, compacts the topic and checks that `getLastMessagePosition()` returns batch index 4. Without the fix, the test fails for `StrategicCompactorTest`.
   
   ### Does this pull request potentially affect one of the following parts:
   
   *If the box was checked, please highlight the changes*
   
   - [ ] Dependencies (add or upgrade a dependency)
   - [ ] The public API
   - [ ] The schema
   - [ ] The default values of configurations
   - [ ] The threading model
   - [ ] The binary protocol
   - [ ] The REST endpoints
   - [ ] The admin CLI options
   - [ ] The metrics
   - [ ] Anything that affects deployment
   


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][broker] Use the uncompressed payload when finding the last message position of a compacted batch [pulsar]

["merlimat (via GitHub)" <gi...@apache.org>]

merlimat merged PR #26962:
URL: https://github.com/apache/pulsar/pull/26962


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org