[fix][client] Fix consumer not returning encrypted messages on decryption failure with compression enabled

[RobertIndie]

Motivation

There is a problem when a consumer tries to decrypt a compressed message. If a consumer fails to decrypt a compressed message, it also fails to decompress it, leading to the message being discarded. The user won't receive the message, even if using CryptoFailureAction.CONSUME, because the failure occurs during decompression, not decryption. This issue could be easily reproduced by the test added in this PR: testE2EEncryptionWithCompression .

We already have logic to skip decompression for messages that can't be decrypted:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Lines 1439 to 1441 in 397b021

	// uncompress decryptedPayload and release decryptedPayload-ByteBuf
	ByteBuf uncompressedPayload = (isMessageUndecryptable || isChunkedMessage) ? decryptedPayload.retain()
	: uncompressPayloadIfNeeded(messageId, msgMetadata, decryptedPayload, cnx, true);

The main issue is if the consumer fails to decrypt the message here:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 1947 in 397b021

return handleCryptoFailure(payload, messageId, currentCnx, redeliveryCount, batchSize, false);

, it isn't treated as undecryptable and still tries to decompress.

Modification

• Refactor decryptPayloadIfNeeded to return a clearer result, DecryptResult, to guide the logic on whether the message was decrypted successfully or should be discarded.
• If decryption fails, treat the message as undecryptable and skip decompression.

Verifying this change

This change added tests

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

Documentation

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

Matching PR in forked repository

PR in forked repository:

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 94.44444% with 1 line in your changes missing coverage. Please review.

Project coverage is 74.25%. Comparing base (bbc6224) to head (ee0ee2c).
Report is 1122 commits behind head on master.

Files with missing lines	Patch %	Lines
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	94.44%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #24356      +/-   ##
============================================
+ Coverage     73.57%   74.25%   +0.67%     
- Complexity    32624    32643      +19     
============================================
  Files          1877     1866      -11     
  Lines        139502   145142    +5640     
  Branches      15299    16598    +1299     
============================================
+ Hits         102638   107771    +5133     
+ Misses        28908    28838      -70     
- Partials       7956     8533     +577     

Flag	Coverage Δ
inttests	26.69% <44.44%> (+2.11%)	⬆️
systests	23.28% <44.44%> (-1.05%)	⬇️
unittests	73.74% <94.44%> (+0.90%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	79.43% <94.44%> (+1.85%)	⬆️

... and 1085 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.