[fix][broker] Set the ServiceUnitStateChannel topic compaction threshold explicitly and other bug fixes

[heesung-sn]

Motivation

We better set the compaction threshold of the serviceUnitStateChannel topic when initializing itself instead of relying on other components' init.

Also, I found the lookup and unload request often timeout by

Modifications

• Set the compaction threshold of the serviceUnitStateChannel topic in the monitor thread
• return timeout completed future from deferGetOwnerRequest
• Release metadataAndPayload buffer in RawBatchMessageContainerImpl.
• Automatically restart load data store producer or table view if inactive.
• Updated getActiveOwnerAsync logic for better synchronization of ownership checks and deferred lookup handling
• added minor retry logic in the ExtensibleLoadManager tests

Verifying this change

• Make sure that the change passes the CI checks.

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: https://github.com/heesung-sn/pulsar/pull/61/files

[dragosvictor]

Can we put this in a finally block to limit the blast radius of this test if it fails?

[heesung-sn]

I think this should always pass. I wonder in what cases this setCompactionThreshold might fail?

[dragosvictor]

I'm sorry for the confusion. It's not that setCompactionThreshold might fail, but anything in between the previous call to this "reset" call might (i.e., lines https://github.com/apache/pulsar/pull/22048/files#diff-0bb4e0f521ef681668abbb30387c7de6ebf6cfce7dcbdd3d2f318ac50f6cc8abR967-R986). So put the entire respective block in a try and this in a finally.

[heesung-sn]

oh I see. nice catch. updated.

[lhotari]

What happens if this flag returns false? Is that a valid way to configure Pulsar with this code? would it make sense to prevent such configuration in the first place, or at least log a warning?

[heesung-sn]

re-raised the pr here. #22064