[fix][broker] Fix get outdated compactedTopicContext after compactionHorizon has been updated

[coderzc]

Motivation

In #20697, we remove the lock of get compactedTopicContext and asyncReadEntries, but compactionHorizon has been updated before compactedTopicContext is updated. In concurrent cases, this may lead to cursor skip messages of newly compacted or getting an NPE when reading compacted entries.

Modifications

Ensure compactionHorizon is updated after compactedTopicContext.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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

Merging #20984 (0995f30) into master (d6734b7) will increase coverage by 36.29%.
Report is 2 commits behind head on master.
The diff coverage is 100.00%.

@@              Coverage Diff              @@
##             master   #20984       +/-   ##
=============================================
+ Coverage     36.84%   73.13%   +36.29%     
- Complexity    12195    32257    +20062     
=============================================
  Files          1698     1875      +177     
  Lines        129852   139443     +9591     
  Branches      14161    15333     +1172     
=============================================
+ Hits          47843   101983    +54140     
+ Misses        75680    29398    -46282     
- Partials       6329     8062     +1733     

Flag	Coverage Δ
inttests	24.13% <100.00%> (+0.01%)	⬆️
systests	25.12% <100.00%> (+0.02%)	⬆️
unittests	72.42% <100.00%> (+40.36%)	⬆️

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

Files Changed	Coverage Δ
...g/apache/pulsar/compaction/CompactedTopicImpl.java	63.90% <100.00%> (+1.50%)	⬆️

... and 1437 files with indirect coverage changes

[poorbarcode]

We can not guarantee that the reading of compactionHorizon and compactedTopicContext safety even if this patch is merged. We should add some code-comment on the method getCompactedTopicContextFuture and getCompactedTopicContext like this:

/**
 * If you want to guarantee that `compactionHorizon` and `compactedTopicContext` match, you need to call this method in "synchronized(compartor){ ... }" lock block
 */

[mattisonchao]

I think we don't need to give it a strong guarantee. :)

[poorbarcode]

I think we don't need to give it a strong guarantee. :)

Yes, I agree. But a description is needed

[coderzc]

I have added some comments to explain this.