[fix][broker] Prevent BookkeeperSchemaStorage, BookkeeperBucketSnapshotStorage, and compactors from bypassing namespace bookie affinity

[void-ptr974]

Motivation

Managed-ledger topic data resolves an effective BookKeeper placement policy from the owner namespace's bookieAffinityGroup and strictBookieAffinityEnabled settings. It then uses a BookKeeper client initialized with that policy and stores the same EnsemblePlacementPolicyConfig in each ledger's custom metadata.

Several Pulsar-owned auxiliary-ledger creation paths did not follow that flow:

• BookkeeperSchemaStorage created schema ledgers through its default BookKeeper client.
• BookkeeperBucketSnapshotStorage created delayed-delivery bucket ledgers through its default BookKeeper client.
• PublishingOrderCompactor, EventTimeOrderCompactor, StrategicTwoPhaseCompactor, and the standalone CompactorTool created compacted ledgers through a fixed default client.

This matters when the owner namespace has a bookie-affinity group, or when strict bookie affinity selects IsolatedBookieEnsemblePlacementPolicy. In those cases, the auxiliary ledger could be created outside the bookie groups selected for the topic even though the namespace policy was configured successfully. The ledger also lacked placement-policy metadata, so a later BookKeeper ensemble replacement could not reliably preserve the intended placement.

The failure was silent: topic data ledgers honored the policy, while schema, delayed bucket, and compaction ledgers for the same topic could bypass it.

Modifications

Shared topic-aware BookKeeper context

• Extract the existing BrokerService affinity, strict-affinity, and system-topic decision matrix into BookKeeperPlacementPolicyConfigResolver and continue using the same resolver for managed-ledger configuration.
• Add asynchronous policy-aware client lookup to BookkeeperManagedLedgerStorageClass.
• Make ManagedLedgerClientFactory reuse its existing EnsemblePlacementPolicyConfig client cache for auxiliary ledgers.
• Add BookKeeperClientContext, which keeps the borrowed BookKeeper client and the matching encoded placement metadata together, preventing the client configuration and ledger metadata from diverging.
• Add PulsarService#getBookKeeperClientContext(TopicName) to resolve the owner namespace's local policy without requiring a loaded topic instance.
• Fail explicitly when a storage implementation cannot honor a custom placement policy instead of silently returning its default client.

Schema and delayed-delivery bucket ledgers

• Resolve the owner topic before BookkeeperSchemaStorage or BookkeeperBucketSnapshotStorage creates a ledger.
• Create the ledger with the topic's placement-aware BookKeeper client and add the same placement policy to its custom metadata.
• Support canonical schema keys with encoded local names and scalable-topic names.
• Preserve the existing behavior for arbitrary or legacy schema keys that cannot be mapped unambiguously to an owner topic.
• Preserve delayed-bucket retry behavior by translating placement lookup and client creation failures to BucketSnapshotPersistenceException.

Schema storage does not gain a separate placement-policy setting. A canonical schema ledger follows the policy of its owner topic's namespace.

Compaction ledgers

• Pass a topic-aware context provider through the publishing-order, event-time, and strategic broker compactor paths.
• Create compacted ledgers with the selected client and merge placement metadata without removing existing compaction metadata.
• Keep the existing compactor constructors so callers without a provider retain the previous default-client behavior.
• Make CompactorTool synchronously read the target namespace's local policy and create exactly one owned BookKeeper client configured with the matching policy.
• Keep compacted-ledger reads, deletion, and cleanup on their existing BookKeeper paths.
• Propagate placement lookup and client creation failures instead of falling back to a client that would ignore the configured policy.

Behavior and impact

• The change affects only creation of new schema, delayed-delivery bucket, and compacted ledgers. Existing ledgers are not moved or rewritten.
• Namespaces without an effective custom placement policy continue to use the existing default BookKeeper client and do not receive placement metadata.
• Broker paths do not create a client per topic. Clients remain cached and owned by ManagedLedgerClientFactory, one per distinct placement-policy configuration; auxiliary components only borrow them.
• CompactorTool still owns exactly one BookKeeper client for its single target topic.
• Reads and deletion of existing auxiliary ledgers are unchanged.
• When a configured policy cannot be resolved or its client cannot be created, ledger creation now fails rather than silently placing data outside the configured groups.
• Dynamic placement-policy refresh and pending-ack ledger creation are intentionally outside this change.

Commit structure

1. 314f70ccaab — introduce the shared resolver, cached client lookup, and BookKeeperClientContext.
2. a7b2ccb439f — apply the context to schema and delayed-delivery bucket ledger creation.
3. 1fa761e790b — apply the context to publishing-order, event-time, strategic, and CLI compaction ledger creation.

Tests

• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.ManagedLedgerClientFactoryTest --tests org.apache.pulsar.broker.storage.BookKeeperPlacementPolicyConfigResolverTest --tests org.apache.pulsar.broker.storage.BookKeeperClientContextTest
• ./gradlew :pulsar-broker:test --no-build-cache --tests org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorageTest --tests org.apache.pulsar.broker.service.schema.SchemaServiceTest --tests org.apache.pulsar.broker.delayed.BookkeeperBucketSnapshotStorageTest --tests org.apache.pulsar.broker.delayed.bucket.BookkeeperBucketSnapshotStoragePlacementTest
• ./gradlew :pulsar-broker:test --no-build-cache --tests org.apache.pulsar.compaction.CompactorPlacementPolicyTest --tests org.apache.pulsar.compaction.CompactorToolTest --tests org.apache.pulsar.compaction.CompactorToolProviderPinsTest
• ./gradlew :pulsar-broker:test --tests org.apache.pulsar.compaction.CompactorTest.testCompaction
• ./gradlew :pulsar-broker:clean :pulsar-broker:compileJava --no-build-cache --no-configuration-cache
• ./gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest

Assisted-by: Codex

[Denovo1998]

This looks good for selecting the right placement-aware client, but I think this PR still depends on #26704 for the guarantee it is trying to provide. On current master, IsolatedBookieEnsemblePlacementPolicy finishes client initialization before the initial /bookies mapping load completes, and getExcludedBookiesWithIsolationGroups() returns an empty exclusion set while cachedRackConfiguration is still null. In particular, CompactorTool creates a fresh isolated client and can immediately create the compacted ledger in that window.

Should we make #26704 an explicit dependency (or include the initialization fix here)? Otherwise #26705 alone can still place a newly created auxiliary ledger outside the configured affinity group.

[void-ptr974]

Agreed. #26704 (or an equivalent initialization fix) is required for the first-ledger isolation guarantee. This PR can be reviewed independently, but I consider that fix a merge dependency: otherwise a newly created client can still place an auxiliary ledger before the initial /bookies mapping is available.

[Denovo1998]

One thing I'm still worried about is the schema recovery path. We now create the schema ledger with the placement-aware client, but BookkeeperSchemaStorage#openLedger() still always uses the storage's default bookKeeper. BookKeeper#asyncOpenLedger() performs fencing/recovery when the ledger is still open, and ensemble replacement uses the placement policy of the client doing that recovery; the custom metadata alone does not switch a Rackaware client to IsolatedBookieEnsemblePlacementPolicy.

This seems reachable because addNewSchemaEntryToStore() calls ledgerHandle.closeAsync() without waiting for it before the new position can be published in the schema locator. If the broker dies in that window, the next read can recover this new placement-aware ledger through the default client and move an ensemble member outside the namespace affinity group.

Would it make sense to await closeAsync() before returning the position / publishing the locator? That looks simpler than making every schema open path resolve the owner placement context, and it would also make sure a locator only points to a fully closed ledger.

[void-ptr974]

Fixed in 6498854. addNewSchemaEntryToStore now composes closeAsync() before returning the position, so the locator is published only after the ledger is closed. If close fails, it does not publish the locator and makes a best-effort attempt to delete the orphan ledger. I added a deterministic test for both close outcomes.

[Denovo1998]

placementPolicyConfig can be null here, but we still require the default managed-ledger storage class to be BookkeeperManagedLedgerStorageClass. For schema/bucket this looks like a behavior regression unrelated to affinity: both components already own a BookKeeper client created through BookKeeperClientFactory, so they could work with a custom/non-BK ManagedLedgerStorage before this change.

Could we require a policy-aware storage class only when an effective custom placement policy is present, and preserve the existing component/default client path when there is no custom policy? The PR description says unsupported storage should fail when it cannot honor a custom placement policy, but the current check also fails the no-policy case.

[void-ptr974]

Fixed in 6498854. When there is no effective custom placement policy, the new overload uses the caller's existing BookKeeper client; schema and bucket pass the clients they already own. A custom policy still requires a BookkeeperManagedLedgerStorageClass so that we fail rather than silently ignore it. I added tests for the no-policy/non-BK-default case and both custom-policy cases.

[Denovo1998]

One related question: this always borrows the client from getDefaultStorageClass(), while normal topic loading honors PersistencePolicies.managedLedgerStorageClassName. With a multi-storage-class ManagedLedgerStorage, the owner topic can therefore use a different storage class/client than this "topic-aware" context.

Is the intended contract only "apply the namespace affinity to the broker's default BookKeeper storage", or should auxiliary ledgers follow the topic's actual storage class as well? If the former is intentional, I think it would be worth documenting that limitation explicitly.

[void-ptr974]

The intended contract is to apply namespace affinity within the existing BookKeeper auxiliary-ledger backend. This PR does not route auxiliary ledgers by the topic's managed-ledger storage class. With no custom policy, schema and bucket retain their existing clients; with a custom policy, the context borrows from the default BookKeeper storage class. Following the topic's actual storage class would also require persisting backend identity with each auxiliary-ledger reference and routing reads and deletes accordingly. I documented this boundary in the getBookKeeperClientContext Javadoc in 6498854.