[improve][offload]use dedicated bookkeeper-ml-offload-scheduler to avoid block core service

[ethqunzhong]

Motivation

We use HDFS as the offload remote storage in our online scenarios. During an HDFS namenode switchover, the normal service of the broker was disrupted due to the failure and retry of a large number of offload operations. The logs are as follows:

Offload operations in Pulsar's managed ledger share the same scheduler thread pool with core services. When offload operations block or take too long, they can impact critical managed ledger operations like data read/write and metadata management, causing system performance degradation.

This change introduces a dedicated offload scheduler to isolate offload operations from core services, preventing potential blocking issues.

Modifications

1. Added configuration support:

   • New managedLedgerNumOffloadSchedulerThreads parameter in ServiceConfiguration and broker.conf
   • Default thread count set to available CPU cores

2. Created dedicated offload scheduler:

   • Added offloadScheduler field in ManagedLedgerFactoryImpl using OrderedScheduler
   • Proper shutdown handling in factory cleanup

3. Isolated offload operations:

   • Migrated all offload-related async operations from scheduledExecutor to dedicated offloadScheduler
   • Affected operations: maybeOffloadInBackground(), maybeOffload(), cleanupOffloaded(), and tryTransformLedgerInfo()

This ensures offload operations don't interfere with core managed ledger functionality, improving overall system stability and performance.

Documentation

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

Matching PR in forked repository

PR in forked repository: ethqunzhong#11

[github-actions]

@ethqunzhong Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[dao-jun]

Offloader already have a separate threadpool, the configuration key is managedLedgerOffloadMaxThreads, and it's default value is 2. Why add a new threadpool?

[ethqunzhong]

Offloader already have a separate threadpool, the configuration key is managedLedgerOffloadMaxThreads, and it's default value is 2. Why add a new threadpool?

Strictly speaking, the thread pool configured by managedLedgerOffloadMaxThreads(use at FSOffloader.scheduler) is used in the offloader instance to perform cold storage operations for the sequential reading and writing of specific ledgers. The thread pool newly added in this modification is used for scheduling the offload operations of all ManagedLedgers at the Broker level—a capability that was previously handled by ManagedLedgerImpl#scheduledExecutor.

[dao-jun]

Maybe add a Backoff to https://github.com/apache/pulsar/blob/master/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L3702-L3704 is better? To avoid frequent retries.