You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "void-ptr974 (via GitHub)" <gi...@apache.org> on 2026/09/24 16:23:03 UTC
[PR] [fix][broker] Prevent publishing, event-time, strategic, and CLI compaction ledgers from bypassing namespace bookie affinity [pulsar]
void-ptr974 opened a new pull request, #26705:
URL: https://github.com/apache/pulsar/pull/26705
### Motivation
When a namespace configures a BookKeeper ensemble placement policy or a bookie-affinity group, broker-managed topic ledgers use the matching placement-aware BookKeeper client. Compaction ledger creation did not: `PublishingOrderCompactor`, `EventTimeOrderCompactor`, `StrategicTwoPhaseCompactor`, and the standalone `CompactorTool` always created compacted ledgers through the default BookKeeper client and did not persist the configured placement policy in ledger metadata.
As a result, a newly compacted ledger could be placed outside the namespace's configured bookie groups. Because its metadata also omitted the policy, BookKeeper ensemble replacement could not consistently preserve that placement later.
### Modifications
- Pass a topic-aware `BookKeeperClientContext` provider into all broker compactor variants.
- Create each compacted ledger with the BookKeeper client selected for the topic's namespace and add the same placement policy to its custom metadata.
- Keep the existing compactor constructors for compatibility; callers that do not provide a context retain the previous default-client behavior.
- Make `CompactorTool` synchronously read the target namespace's local policy, resolve the effective placement policy, and create one owned BookKeeper client with matching policy configuration and metadata.
- Preserve existing compaction metadata and keep reads, deletes, and cleanup on the existing default BookKeeper client.
- Fail compaction when placement lookup or client creation fails instead of silently falling back and violating the configured policy.
### Impact
- Only creation of new compacted ledgers changes.
- Broker paths reuse the existing placement-aware client cache; this does not create a client per topic.
- The standalone CLI still owns exactly one BookKeeper client for its single target topic.
- Namespaces without a configured placement policy retain the existing default placement behavior.
- Existing compacted-ledger reads and deletion behavior are unchanged.
### Stacked PR
This PR depends on #26703, which depends on #26702. The branch contains those prerequisite commits because the upstream repository cannot use a fork-only branch as the base. The compaction-specific commit to review is `1fa761e790b`.
### Tests
- `./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:checkstyleMain :pulsar-broker:checkstyleTest`
### Scope
Dynamic policy refresh and pending-ack ledger creation are intentionally outside this series.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Prevent BookkeeperSchemaStorage, BookkeeperBucketSnapshotStorage, and compactors from bypassing namespace bookie affinity [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26705:
URL: https://github.com/apache/pulsar/pull/26705#issuecomment-6167634012
Pushed the review follow-up in [71142db824a](https://github.com/apache/pulsar/commit/71142db824ac6db213b32f986ef9a258691abe2d).
Changes:
- Validate canonical schema-key encoding without removing another partition suffix. This preserves the topic identity used for system-topic placement, adds DEBUG logging for legacy-key fallback, and covers repeated suffixes, encoded names, and scalable names.
- Add `BrokerBookieIsolationTest.testAuxiliaryLedgerBookieIsolation` with real bookies in disjoint default and namespace-affinity groups. It creates schema, broker compaction, and delayed-bucket ledgers, checks every ensemble and the stored metadata, and includes a no-affinity control. Schema registration happens before publishing topic data.
- Correct the Mockito verification to check the two-argument client-context overload.
- Expand the storage API Javadoc to describe the default implementation, borrowed-client ownership, the same-backend requirement, and third-party compatibility.
Compatibility note: existing storage implementations retain the default-client behavior when no custom policy is required. With namespace affinity or strict bookie affinity enabled, implementations must override `getBookKeeperClient(EnsemblePlacementPolicyConfig)`; otherwise auxiliary ledger creation completes exceptionally with `UnsupportedOperationException`.
Local validation: `quickCheck` passed, and all 42 scoped broker tests passed with `testRetryCount=0`, including the real BookKeeper placement test.
#26704, or an equivalent initialization fix, remains a merge prerequisite for the first-ledger isolation guarantee.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Prevent BookkeeperSchemaStorage, BookkeeperBucketSnapshotStorage, and compactors from bypassing namespace bookie affinity [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26705:
URL: https://github.com/apache/pulsar/pull/26705#discussion_r4156775400
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorTool.java:
##########
@@ -199,8 +252,15 @@ public static void main(String[] args) throws Exception {
EventLoopGroup eventLoopGroup = EventLoopUtil.newEventLoopGroup(1, false,
new DefaultThreadFactory("compactor-io"));
+ LocalPoliciesResources localPoliciesResources = new LocalPoliciesResources(
+ store, brokerConfig.getMetadataStoreOperationTimeoutSeconds());
+ Optional<LocalPolicies> localPolicies =
+ localPoliciesResources.getLocalPolicies(topicName.getNamespaceObject());
+ BookKeeperClientContext bookKeeperClientContext = createBookKeeperClientContext(
Review Comment:
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.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java:
##########
@@ -641,11 +645,48 @@ private CompletableFuture<Long> addEntry(LedgerHandle ledgerHandle, SchemaEntry
}
@NonNull
- private CompletableFuture<LedgerHandle> createLedger(String schemaId) {
- Map<String, byte[]> metadata = LedgerMetadataUtils.buildMetadataForSchema(schemaId);
+ @VisibleForTesting
+ CompletableFuture<LedgerHandle> createLedger(String schemaId) {
+ return CompletableFuture.completedFuture(schemaId)
+ .thenCompose(id -> {
+ Optional<TopicName> topicName = getOwnerTopicName(id);
+ if (topicName.isEmpty()) {
+ return createLedger(id, bookKeeper, LedgerMetadataUtils.buildMetadataForSchema(id));
+ }
+ return pulsar.getBookKeeperClientContext(topicName.get())
+ .thenCompose(clientContext -> createLedger(id, clientContext));
Review Comment:
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.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java:
##########
@@ -1659,6 +1663,40 @@ public BookKeeper getBookKeeperClient() {
}
}
+ /**
+ * Resolve the placement policy for a topic and borrow the corresponding shared BookKeeper client.
+ *
+ * @param topicName the topic that owns the ledger
+ * @return a future that completes with the BookKeeper client and matching placement metadata
+ */
+ public CompletableFuture<BookKeeperClientContext> getBookKeeperClientContext(TopicName topicName) {
+ return CompletableFuture.completedFuture(topicName)
+ .thenCompose(name -> {
+ Objects.requireNonNull(name, "topicName");
+ return getPulsarResources().getLocalPolicies()
+ .getLocalPoliciesAsync(name.getNamespaceObject());
+ }).thenCompose(localPolicies -> {
+ EnsemblePlacementPolicyConfig placementPolicyConfig =
+ BookKeeperPlacementPolicyConfigResolver.resolve(config, topicName, localPolicies)
+ .orElse(null);
+ ManagedLedgerStorageClass defaultStorageClass =
+ getManagedLedgerStorage().getDefaultStorageClass();
+ if (!(defaultStorageClass instanceof BookkeeperManagedLedgerStorageClass bkStorageClass)) {
Review Comment:
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.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java:
##########
@@ -1659,6 +1663,40 @@ public BookKeeper getBookKeeperClient() {
}
}
+ /**
+ * Resolve the placement policy for a topic and borrow the corresponding shared BookKeeper client.
+ *
+ * @param topicName the topic that owns the ledger
+ * @return a future that completes with the BookKeeper client and matching placement metadata
+ */
+ public CompletableFuture<BookKeeperClientContext> getBookKeeperClientContext(TopicName topicName) {
+ return CompletableFuture.completedFuture(topicName)
+ .thenCompose(name -> {
+ Objects.requireNonNull(name, "topicName");
+ return getPulsarResources().getLocalPolicies()
+ .getLocalPoliciesAsync(name.getNamespaceObject());
+ }).thenCompose(localPolicies -> {
+ EnsemblePlacementPolicyConfig placementPolicyConfig =
+ BookKeeperPlacementPolicyConfigResolver.resolve(config, topicName, localPolicies)
+ .orElse(null);
+ ManagedLedgerStorageClass defaultStorageClass =
+ getManagedLedgerStorage().getDefaultStorageClass();
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Prevent BookkeeperSchemaStorage, BookkeeperBucketSnapshotStorage, and compactors from bypassing namespace bookie affinity [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26705:
URL: https://github.com/apache/pulsar/pull/26705#discussion_r4133565125
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java:
##########
@@ -641,11 +645,48 @@ private CompletableFuture<Long> addEntry(LedgerHandle ledgerHandle, SchemaEntry
}
@NonNull
- private CompletableFuture<LedgerHandle> createLedger(String schemaId) {
- Map<String, byte[]> metadata = LedgerMetadataUtils.buildMetadataForSchema(schemaId);
+ @VisibleForTesting
+ CompletableFuture<LedgerHandle> createLedger(String schemaId) {
+ return CompletableFuture.completedFuture(schemaId)
+ .thenCompose(id -> {
+ Optional<TopicName> topicName = getOwnerTopicName(id);
+ if (topicName.isEmpty()) {
+ return createLedger(id, bookKeeper, LedgerMetadataUtils.buildMetadataForSchema(id));
+ }
+ return pulsar.getBookKeeperClientContext(topicName.get())
+ .thenCompose(clientContext -> createLedger(id, clientContext));
Review Comment:
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.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java:
##########
@@ -1659,6 +1663,40 @@ public BookKeeper getBookKeeperClient() {
}
}
+ /**
+ * Resolve the placement policy for a topic and borrow the corresponding shared BookKeeper client.
+ *
+ * @param topicName the topic that owns the ledger
+ * @return a future that completes with the BookKeeper client and matching placement metadata
+ */
+ public CompletableFuture<BookKeeperClientContext> getBookKeeperClientContext(TopicName topicName) {
+ return CompletableFuture.completedFuture(topicName)
+ .thenCompose(name -> {
+ Objects.requireNonNull(name, "topicName");
+ return getPulsarResources().getLocalPolicies()
+ .getLocalPoliciesAsync(name.getNamespaceObject());
+ }).thenCompose(localPolicies -> {
+ EnsemblePlacementPolicyConfig placementPolicyConfig =
+ BookKeeperPlacementPolicyConfigResolver.resolve(config, topicName, localPolicies)
+ .orElse(null);
+ ManagedLedgerStorageClass defaultStorageClass =
+ getManagedLedgerStorage().getDefaultStorageClass();
Review Comment:
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.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java:
##########
@@ -1659,6 +1663,40 @@ public BookKeeper getBookKeeperClient() {
}
}
+ /**
+ * Resolve the placement policy for a topic and borrow the corresponding shared BookKeeper client.
+ *
+ * @param topicName the topic that owns the ledger
+ * @return a future that completes with the BookKeeper client and matching placement metadata
+ */
+ public CompletableFuture<BookKeeperClientContext> getBookKeeperClientContext(TopicName topicName) {
+ return CompletableFuture.completedFuture(topicName)
+ .thenCompose(name -> {
+ Objects.requireNonNull(name, "topicName");
+ return getPulsarResources().getLocalPolicies()
+ .getLocalPoliciesAsync(name.getNamespaceObject());
+ }).thenCompose(localPolicies -> {
+ EnsemblePlacementPolicyConfig placementPolicyConfig =
+ BookKeeperPlacementPolicyConfigResolver.resolve(config, topicName, localPolicies)
+ .orElse(null);
+ ManagedLedgerStorageClass defaultStorageClass =
+ getManagedLedgerStorage().getDefaultStorageClass();
+ if (!(defaultStorageClass instanceof BookkeeperManagedLedgerStorageClass bkStorageClass)) {
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Prevent BookkeeperSchemaStorage, BookkeeperBucketSnapshotStorage, and compactors from bypassing namespace bookie affinity [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26705:
URL: https://github.com/apache/pulsar/pull/26705#discussion_r4133552247
##########
pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorTool.java:
##########
@@ -199,8 +252,15 @@ public static void main(String[] args) throws Exception {
EventLoopGroup eventLoopGroup = EventLoopUtil.newEventLoopGroup(1, false,
new DefaultThreadFactory("compactor-io"));
+ LocalPoliciesResources localPoliciesResources = new LocalPoliciesResources(
+ store, brokerConfig.getMetadataStoreOperationTimeoutSeconds());
+ Optional<LocalPolicies> localPolicies =
+ localPoliciesResources.getLocalPolicies(topicName.getNamespaceObject());
+ BookKeeperClientContext bookKeeperClientContext = createBookKeeperClientContext(
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Prevent BookkeeperSchemaStorage, BookkeeperBucketSnapshotStorage, and compactors from bypassing namespace bookie affinity [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26705:
URL: https://github.com/apache/pulsar/pull/26705#discussion_r4259570358
##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorPlacementPolicyTest.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.compaction;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.RETURNS_SELF;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.api.CreateBuilder;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
+import org.apache.pulsar.bookie.rackawareness.IsolatedBookieEnsemblePlacementPolicy;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.storage.BookKeeperClientContext;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.EnsemblePlacementPolicyConfig;
+import org.mockito.ArgumentCaptor;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class CompactorPlacementPolicyTest {
+
+ private static final String TOPIC = "persistent://tenant/namespace/topic";
+
+ @DataProvider(name = "compactorTypes")
+ public static Object[][] compactorTypes() {
+ return new Object[][] {
+ {"publishing"},
+ {"event-time"},
+ {"strategic"}
+ };
+ }
+
+ @Test(dataProvider = "compactorTypes")
+ public void testCreateLedgerUsesPlacementClientAndMatchingMetadata(String compactorType) throws Exception {
Review Comment:
**[QUALITY] No test checks that auxiliary ledgers really land inside the affinity group**
The new tests for compaction, schema, bucket and `CompactorTool` mock `BookKeeper` and `CreateBuilder`. They prove that the policy client and a decodable policy were passed. They would still pass if the client was never initialized with the policy, or if the policy stopped reading the metadata key.
`PulsarServiceBookKeeperClientContextTest` also always stubs the local policies as empty, so a namespace `bookieAffinityGroup` is only covered at resolver level (`BookKeeperPlacementPolicyConfigResolverTest`).
Could you extend `BrokerBookieIsolationTest` (or add a sibling) so that, in a namespace with disjoint bookie groups, it registers a schema, runs a compaction and creates a delayed bucket, then asserts the ledger ensembles from `LedgerMetadata`? That is the guarantee this PR is about, and it would also catch the [`#26704`](https://github.com/apache/pulsar/blob/6498854d74802e69c061a53c7e21f5bb8a26f27b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorPlacementPolicyTest.java#L26704)/#26701 interactions.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/storage/BookkeeperManagedLedgerStorageClass.java:
##########
@@ -33,6 +35,28 @@ public interface BookkeeperManagedLedgerStorageClass extends ManagedLedgerStorag
*/
BookKeeper getBookKeeperClient();
+ /**
+ * Return the BookKeeper client for the specified ensemble placement policy.
+ *
+ * <p>The returned client is owned by the storage class and must not be closed by the caller.
+ *
+ * @param ensemblePlacementPolicyConfig the ensemble placement policy configuration
+ * @return a future that completes with the BookKeeper client
+ */
+ default CompletableFuture<BookKeeper> getBookKeeperClient(
Review Comment:
**[QUALITY] Strict mode now fails these operations for storage classes without the new override**
Good that the default stays source compatible. One thing to document: with `strictBookieAffinityEnabled=true` the resolver returns a policy for every topic, so schema, bucket and compaction ledger creation all call this method. A third-party `BookkeeperManagedLedgerStorageClass` that does not override it will now fail those operations with `UnsupportedOperationException`, where it used to work (without isolation). In this repo only `ManagedLedgerClientFactory` implements it, and it overrides it, so nothing here breaks.
Failing closed fits the goal. Could you mention the new method and this behavior in the Javadoc and the PR's release note?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java:
##########
@@ -641,11 +648,48 @@ private CompletableFuture<Long> addEntry(LedgerHandle ledgerHandle, SchemaEntry
}
@NonNull
- private CompletableFuture<LedgerHandle> createLedger(String schemaId) {
- Map<String, byte[]> metadata = LedgerMetadataUtils.buildMetadataForSchema(schemaId);
+ @VisibleForTesting
+ CompletableFuture<LedgerHandle> createLedger(String schemaId) {
+ return CompletableFuture.completedFuture(schemaId)
+ .thenCompose(id -> {
+ Optional<TopicName> topicName = getOwnerTopicName(id);
+ if (topicName.isEmpty()) {
+ return createLedger(id, bookKeeper, LedgerMetadataUtils.buildMetadataForSchema(id));
+ }
+ return pulsar.getBookKeeperClientContext(topicName.get(), () -> bookKeeper)
+ .thenCompose(clientContext -> createLedger(id, clientContext));
+ });
+ }
+
+ private static Optional<TopicName> getOwnerTopicName(String schemaId) {
+ String[] parts = schemaId.split("/", -1);
+ if (parts.length != 3) {
+ return Optional.empty();
+ }
+ try {
+ String localName = Codec.decode(parts[2]);
+ TopicDomain domain = localName.indexOf('/') >= 0 ? TopicDomain.topic : TopicDomain.persistent;
+ TopicName topicName = TopicName.get(domain.value(), parts[0], parts[1], localName);
+ return schemaId.equals(topicName.getSchemaName()) ? Optional.of(topicName) : Optional.empty();
Review Comment:
**[QUALITY] Round-trip check can silently skip affinity for a valid schema key**
The `schemaId.equals(topicName.getSchemaName())` check rejects keys that `TopicName` produced itself. For a topic such as `x-partition-0-partition-0`, the schema key is `t/ns/x-partition-0`. Rebuilding a `TopicName` from that key and taking its schema name gives `t/ns/x`, so the check fails. The ledger then silently uses the default client with no placement metadata, even in a strict or affinity namespace. This is a rare name, but it is the bypass the PR removes.
Only the namespace is needed for the lookup. Taking `tenant/namespace` from the first two segments would avoid depending on the round trip. Either way, a debug log on the fallback would make it discoverable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org