[fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage

[geniusjoe]

Fixes #18292
Related #18701

Motivation

When multiple requests concurrently create a schema for a brand-new topic (i.e., the schema locator z-node does not yet exist), each request first creates a BookKeeper ledger via addNewSchemaEntryToStore, then attempts to create the schema locator z-node via CAS (createSchemaLocator with expectedVersion = -1L).

Only one request succeeds; the others fail with BadVersionException (because ZK MetadataStore translates NODEEXISTS to BadVersionException when expectedVersion == -1). However, the ledgers created by the failed requests were never cleaned up, resulting in orphan ledgers (dirty data) in BookKeeper.

Note that the existing updateSchemaLocator method already has cleanup logic for this scenario (deleting the orphan ledger when CAS fails with BadVersionException) in #18701, but the createNewSchema method — which handles the initial schema creation path — was missing this cleanup.

Modifications

• BookkeeperSchemaStorage.createNewSchema: Added a whenComplete callback after createSchemaLocator. When the CAS operation fails due to AlreadyExistsException or BadVersionException, the orphan BookKeeper ledger is asynchronously deleted, consistent with the existing cleanup logic in updateSchemaLocator. The method is also refactored into three clearly commented steps for better readability.

• PulsarMockLedgerHandle: Added a new constructor that accepts Map<String, byte[]> customMetadata and passes it to LedgerMetadataBuilder.withCustomMetadata(), so that mock ledgers can retain custom metadata set during creation.

• PulsarMockBookKeeper: Updated asyncCreateLedger to forward the properties parameter to the new PulsarMockLedgerHandle constructor, enabling tests to inspect ledger custom metadata.

• SchemaTest: Added testConcurrentCreateSchemaNoOrphanLedger test that verifies orphan ledgers are cleaned up when 16 producers concurrently create schema on a brand-new topic. The test inspects surviving BK ledgers via PulsarMockBookKeeper.getLedgerMap() and asserts that only 1 ledger with matching pulsar/schemaId custom metadata exists.

Verifying this change

This change added tests and can be verified as follows:

• Added testConcurrentCreateSchemaNoOrphanLedger in SchemaTest that concurrently creates 16 producers with the same AVRO schema on a brand-new topic, then verifies:
  1. Only 1 schema version exists via admin.schemas().getAllSchemas()
  2. Only 1 surviving BK ledger has customMetadata["pulsar/schemaId"] matching the topic's schema name (orphan ledgers from failed concurrent creations were deleted)

Does this pull request potentially affect one of the following parts:

• 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

[dao-jun]

LGTM

[hanmz]

lgtm

[hanmz]

/pulsarbot rerun-failure-checks

[Denovo1998]

@geniusjoe @dao-jun @codelipenghui @hanmz

Seems like there are some issues?

Request A and Request B create a schema for a new topic simultaneously.

T1: A finds the locator does not exist.

T2: B also finds the locator does not exist.

T3: A creates BK ledger L1 and writes the schema entry.

T4: B creates BK ledger L2 and writes the schema entry.

T5: A's createSchemaLocator is successful, and the locator points to L1.

T6: B's createSchemaLocator fails, receiving a BadVersionException / AlreadyExistsException.

T7: B's whenComplete calls asyncDeleteLedger(L2).

T8: asyncDeleteLedger only asynchronously commits deletion and returns immediately.

T9: B's createNewSchema future still completes with a BadVersionException.

T10: The upper BookkeeperSchemaStorage#put catches the BadVersionException and immediately retries.

T11: The retry re-reads the locator, finds that the schema already exists, and the request succeeds and returns.

T12: But the deletion callback of L2 may not have finished executing.

whenComplete only waits for this callback to return, but the callback starts a callback-based bookKeeper.asyncDeleteLedger(...) and returns before the ledger deletion is confirmed. As a result, the outer put(...) can observe the original BadVersionException / AlreadyExistsException, retry immediately, and complete the schema creation before the orphan ledger is actually removed. This makes the new getLedgerMap() assertion timing-sensitive, and also weakens the cleanup guarantee exposed to callers. Could we wrap asyncDeleteLedger in a CompletableFuture and compose it before rethrowing the original CAS exception so the existing retry happens after the delete callback completes?

[geniusjoe]

@Denovo1998
Thank you very much for your review. I initially considered removing the redundant schema ledger as an asynchronous task to avoid blocking the main schema creation logic.

However, the current approach could indeed introduce race condition issues in unit tests. I think your proposed solution makes sense. If you have time, you could go ahead and implement it as you suggested. Perhaps the cleanup semantics for both createNewSchema and updateSchemaLocator need to be updated consistently.