[improve][metadata] Add partition key resolver support for Oxia index scans

[void-ptr974]

Motivation

Oxia records can be routed with PartitionKey. Native secondary-index scans return primary paths, and some callers can derive the routing key from each returned path. A single scan-level PartitionKey only covers scans scoped to one routing key; it does not cover scans that may span multiple routing keys.

For example, transaction op records are stored under /txn/op/<txnId>-<seq> with PartitionKey(txnId). A scan by segment can return op records for multiple transaction IDs, so each follow-up primary read can use the transaction ID derived from the returned path.

This PR adds an option for passing that per-result routing hint and uses it in transaction metadata index scans.

Modifications

• Add Option.PartitionKeyResolver to derive a concrete PartitionKey from a primary path discovered during a scan.
• Resolve a concrete PartitionKey before Oxia loads each primary record returned by scanByIndex.
• Pass fixed or path-derived routing hints from transaction metadata index scans.
• Add multi-shard Oxia coverage for partition-key-aware index scans.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added a multi-shard Oxia test for partition-key-aware index scans.
• ./gradlew :pulsar-metadata:compileTestJava :pulsar-broker:compileJava :pulsar-metadata:checkstyleMain :pulsar-metadata:checkstyleTest :pulsar-broker:checkstyleMain -x :managed-ledger:compileTestJava
• ./gradlew :pulsar-metadata:test --tests org.apache.pulsar.metadata.OxiaPartitionKeyTest

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

This adds a new metadata-store Option subtype. It does not change existing option behavior, configuration defaults, wire protocol, or metadata format.

[Denovo1998]

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.

[void-ptr974]

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.

[Denovo1998]

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?

[void-ptr974]

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.

[Denovo1998]

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.

[void-ptr974]

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.