Denovo1998 commented on code in PR #26148:
URL: https://github.com/apache/pulsar/pull/26148#discussion_r3565830383
##########
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:
This helper may try multiple candidate records, but every unsuccessful
candidate is deleted and it returns immediately after finding the first
cross-shard record. Therefore only one indexed record exists when scanByIndex
runs. The motivating case is a single scan containing primary records with
different partition keys. Could the test retain at least two cross-shard
records with distinct partition keys and assert that both are returned? This
would catch an implementation that accidentally resolves or caches one
partition key per scan rather than per result.
##########
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:
The PartitionKeyResolver explicitly allows returning null, but the current
behavior then performs a lookup without a partition key. If the indexed record
was originally stored with a custom partition key, this lookup can access a
different shard and return an empty result—silently dropping the indexed data
while the scan reports success. Could we either cause the scan to fail when
routing cannot be resolved or, alternatively, clearly document and test the
precise conditions under which the unpartitioned fallback is safe?
##########
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:
The resolver is evaluated eagerly during the chain assembly. Because the
first thenCompose is attached to an already-completed future, an earlier
storeGet may already be in flight when a later resolver throws. The outer stage
then calls onError, but the detached get can still complete and invoke onNext
afterward, which violates scanByIndex’s terminal-callback contract. Resolve the
partition key inside the sequential thenCompose stage and add a
resolver-failure ordering test.
--
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]
