[improve][meta] PIP-453: Improve the metadata store threading model

[BewareMyPower]

PIP: #25173

Modifications

1. Trigger callbacks in the underlying metadata store's internal thread directly.
2. Add a dedicated scheduler for batch flushing tasks
3. Add a dedicated thread pool for MetadataCache's computing tasks
4. Add tests to verify serialization and deserialization tasks are executed in the pool whose thread count is specified by metadataStoreSerDesThreads

Documentation

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

Matching PR in forked repository

PR in forked repository:

[BewareMyPower]

@lhotari @tjiuming @nodece Could you take a look at this PR?

[nodece]

Overall LGTM

[Copilot]

Pull request overview

This PR implements PIP-453 to improve the metadata store threading model by separating concerns across multiple dedicated thread pools. The change aims to prevent a single metadata store thread from being overwhelmed by various types of tasks (event handling, scheduling, flushing, and serialization/deserialization).

Changes:

• Replaced single ScheduledExecutorService with four specialized executors: eventExecutor for notifications, schedulerExecutor for delayed tasks, flushExecutor for batch flushing, and serDesExecutor for serialization/deserialization
• Modified callback execution to use the underlying store's thread directly instead of switching to executor thread
• Removed batch metadata store queue size metrics since the dedicated flush executor only handles flushing

Reviewed changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
AbstractMetadataStore.java	Introduces new thread pool structure with separate executors for events, scheduling, and ser/des operations
AbstractBatchedMetadataStore.java	Adds dedicated flush executor and updates flush mechanism with synchronized method
BatchMetadataStoreStats.java	Removes executor queue size metrics and simplifies constructor
MetadataCacheImpl.java	Updates to use OrderedExecutor for ser/des operations and separate scheduler for retries
ZKMetadataStore.java	Updates callback execution to use new helper methods
EtcdMetadataStore.java	Updates async operations to execute callbacks in event executor
RocksdbMetadataStore.java, LocalMemoryMetadataStore.java, OxiaMetadataStore.java	Updates constructors to pass numSerDesThreads parameter
ServiceConfiguration.java, MetadataStoreConfig.java	Adds new configuration for ser/des thread pool size
PulsarService.java	Propagates new configuration to metadata store instances
PulsarServiceTest.java	Adds comprehensive test for ser/des thread pool behavior
OpenTelemetryMetadataStoreStatsTest.java	Updates test to reflect removed metrics
MetadataStoreFactoryImplTest.java	Updates test constructor call
pip-453.md	Updates PIP document to reflect actual implementation with 4 thread sets

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[BewareMyPower]

@nodece @lhotari @dao-jun Hi, the comments are all addressed and tests all passed eventually, PTAL again

[lhotari]

LGTM, great work @BewareMyPower

[BewareMyPower]

There is a bug with this PR:

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java

Line 157 in 9db31cc

if (cachedFuture != null && cachedFuture != future) {

So don't cherry-pick it for now, I will push a fix soon

[lhotari]

There is a bug with this PR:

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java

Line 157 in 9db31cc

if (cachedFuture != null && cachedFuture != future) {

So don't cherry-pick it for now, I will push a fix soon

@BewareMyPower It's already cherry-picked for 4.0.9 and 4.1.3 release candidate 1. Should I start a new release candidate after your fix is merged?

[BewareMyPower]

Yes, please cancel the existing releases. I will work on the fix ASAP

[BewareMyPower]

The fix is here: #25246