[fix][broker] Fix lookup permit leak when namespace policy reads fail

[void-ptr974]

Motivation

Partition metadata lookup starts isAllowAutoTopicCreationAsync inside a callback without composing its future into the request chain. If the namespace policy read fails asynchronously, the broker sends no error response and never releases the acquired lookup permit. Repeated failures consume the permits shared by broker lookup and partition metadata lookup, leaving clients waiting for timeouts and subsequent requests rejected.

Modifications

• Compose authorization, automatic topic creation policy checks, and metadata queries into one future chain.
• Map failures and send responses within the stage that owns them, using handle while preserving the existing error mappings and response send paths.
• Recycle TopicExistsInfo in the lookup callback's finally block. Release the lookup permit once in the terminal callback and log unexpected callback failures without attempting another response.
• Add regression coverage for policy failures, delayed stage completion, lookup results and recycling, error mappings across both lookup paths, and cleanup-failure logging.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Policy failure tests cover immediately failed and delayed futures, both values of the client's metadata auto-creation flag, and three consecutive requests. They assert a single error response and restoration of the initial permit count.
• Delayed authorization, policy, and metadata lookup tests verify stage ordering and that the permit remains held until the request completes.
• Lookup result and error-mapping tests cover both lookup paths, partitioned/non-partitioned/missing topics, and wrapped failures. Cleanup tests verify recycling, permit restoration, no duplicate response, and one logged cleanup failure.
• Local validation passed 46 scoped test cases (26 in ServerCnxTest and 20 in GetPartitionMetadataTest) and quickCheck, with retries disabled.
• The four policy-failure regression cases fail on the pre-fix implementation due to missing error responses and pass with this fix.

./gradlew :pulsar-broker:test \
  --tests 'org.apache.pulsar.broker.service.ServerCnxTest.testPartitionMetadata*' \
  --tests 'org.apache.pulsar.broker.service.ServerCnxTest.handlePartitionMetadataRequestWithServiceNotReady' \
  --tests 'org.apache.pulsar.broker.admin.GetPartitionMetadataTest.testAutoCreatePartitionedTopic' \
  --tests 'org.apache.pulsar.broker.admin.GetPartitionMetadataTest.testAutoCreateNonPartitionedTopic' \
  --tests 'org.apache.pulsar.broker.admin.GetPartitionMetadataTest.testNamespaceNotExist' \
  --tests 'org.apache.pulsar.broker.admin.GetPartitionMetadataTest.testTenantNotExist' \
  --tests 'org.apache.pulsar.broker.admin.GetPartitionMetadataTest.testGetMetadataIfNotAllowedCreate' \
  --tests 'org.apache.pulsar.broker.admin.GetPartitionMetadataTest.testGetMetadataIfNotAllowedCreateOfNonPersistentTopic' \
  quickCheck -PtestRetryCount=0 -PtestMaxParallelForks=1 -PtestFailFast=false

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• 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

[Denovo1998]

This approach works with the current chain, but I'm concerned that error mapping now depends on inspecting the authorizationFuture and autoCreationFuture to infer which stage failed. This makes the terminal handler tightly coupled to the chain's exact structure. Could we explicitly carry the stage or result—or map each stage’s failure before reaching the terminal handler—instead? Doing so would make it safer to extend later.

[void-ptr974]

Addressed in ace4156. Each stage now maps its own failures in handle, and all branches are composed into the request chain. The terminal callback releases the permit and logs unexpected callback failures. This removes the intermediate future-state checks while preserving the existing error mappings and response send paths.

[Denovo1998]

Can we capture or log a recycle() failure here? The permit is still released correctly. However, the future returned by whenComplete is ignored, and if recycle() throws an exception, it ends up in an unobserved dependent future, effectively disappearing. Since the new test explicitly exercises this scenario, it would be useful to make the cleanup failure visible without altering the response or release behavior.

try {
    if (topicExistsInfo != null) {
        topicExistsInfo.recycle();
    }
} catch (Throwable cleanupError) {
    log.error()
        .attr("topic", topicName)
        .exception(cleanupError)
        .log("Failed to recycle partition metadata lookup result");
} finally {
    lookupSemaphore.release();
}

[void-ptr974]

Addressed in ace4156. TopicExistsInfo is now recycled in the lookup callback’s finally block. A cleanup exception propagates through the composed chain to the terminal callback, which releases the permit and logs the exception without attempting another response. The existing cleanup-failure test now also verifies that the exception is logged once.

[void-ptr974]

Updated in ace4156 to address both review comments.

Error and response handling now stays within each stage, lookup results are recycled locally, and permit release remains centralized.

Added tests for delayed authorization/policy/lookup completion, result recycling, and error mappings across both lookup paths. Local validation passed 46 scoped test cases and quickCheck, with retries disabled. The four policy-failure regression cases fail on the pre-fix implementation due to missing error responses and pass with the fix.

[Denovo1998]

LGTM!