void-ptr974 commented on code in PR #26148:
URL: https://github.com/apache/pulsar/pull/26148#discussion_r3566299726
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java:
##########
@@ -277,8 +277,9 @@ protected CompletableFuture<Void> storeScanByIndex(
// (it's the scan-and-filter compat-path predicate).
CompletableFuture<Void> chain =
CompletableFuture.completedFuture(null);
for (String key : primaryKeys) {
+ Set<Option> getOpts =
OptionsHelper.withResolvedPartitionKey(opts, key);
chain = chain
- .thenCompose(__ -> storeGet(key, opts))
+ .thenCompose(__ -> storeGet(key, getOpts))
Review Comment:
Fixed. The partition-key resolution now happens inside the sequential
`thenCompose`, so the resolver for the next primary key is not evaluated until
the preceding `storeGet` has completed.
Added a regression test that holds the first read pending and makes the
second resolver fail. It verifies that the scan remains pending before the
first read completes, then observes `onNext` followed by `onError`, with no
read for the second key.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/OptionsHelper.java:
##########
@@ -55,6 +56,32 @@ public static String partitionKey(Set<Option> opts) {
return null;
}
+ /**
+ * Return {@code opts} with a concrete {@link Option.PartitionKey} added
when a
+ * {@link Option.PartitionKeyResolver} can derive one from {@code path}.
Existing fixed
+ * partition keys are preserved.
+ */
+ public static Set<Option> withResolvedPartitionKey(Set<Option> opts,
String path) {
+ if (opts == null || opts.isEmpty()) {
+ return Set.of();
+ }
+ if (partitionKey(opts) != null) {
+ return opts;
+ }
+ for (Option o : opts) {
+ if (o instanceof Option.PartitionKeyResolver resolver) {
+ String partitionKey = resolver.resolver().apply(path);
Review Comment:
Fixed. `PartitionKeyResolver` now requires a non-null key for every scanned
primary path. Returning `null` fails the scan with an `IllegalStateException`
instead of falling back to an unpartitioned read, preventing a silent
cross-shard miss.
Added a test that verifies the error callback and confirms that no primary
read is issued when the resolver cannot provide a key.
##########
pulsar-metadata/src/test/java/org/apache/pulsar/metadata/OxiaPartitionKeyTest.java:
##########
@@ -114,4 +118,45 @@ public void getChildrenWithPartitionKey() throws Exception
{
assertTrue(children.containsAll(List.of("a", "b")),
"expected children a and b, got: " + children);
}
+
+ @Test
+ public void scanByIndexUsesPartitionKeyResolverForPrimaryReads() throws
Exception {
+ @Cleanup
+ MetadataStore store = newStore();
+
+ String parent = "/partition-key-index-scan-" + System.nanoTime();
+ String indexName = "idx:partition-key-resolver-" + System.nanoTime();
+ String indexKey = "match";
+ RoutedKey routedKey =
createIndexedKeyOnlyVisibleWithPartitionKey(store, parent, indexName, indexKey);
Review Comment:
Updated. The integration test now retains two cross-shard indexed records
with distinct partition keys and asserts that both primary paths are returned
by the scan. This exercises per-result resolution rather than a single
scan-level routing key.
The focused unit and multi-shard Oxia integration tests pass.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
