Fixes #24964

Motivation

BinaryProtoLookupService#getTopicsUnderNamespace currently does not deduplicate concurrent requests with identical parameters (namespace, mode, topicsPattern, topicsHash). For each call, it creates a new CompletableFuture and immediately invokes the helper to obtain a connection and send a request, which can cause request amplification when multiple components query the same namespace simultaneously.

Current behavior in code

• BinaryProtoLookupService#getTopicsUnderNamespace allocates a new future and invokes the private helper unconditionally (no shared in-flight future).
• Sibling methods, in the same class, already deduplicate:
  • BinaryProtoLookupService#getBroker uses lookupInProgress.computeIfAbsent(...) with removal in whenComplete(...).
  • BinaryProtoLookupService#getPartitionedTopicMetadata uses partitionedMetadataInProgress.computeIfAbsent(...) with removal in whenComplete(...).

This inconsistency means BinaryProtoLookupService#getTopicsUnderNamespace can issue duplicate network requests for identical inputs, increasing broker load, while BinaryProtoLookupService#getBroker and BinaryProtoLookupService#getPartitionedTopicMetadata already deduplicate.

Real impact scenarios in the current codebase

• Pattern consumers recheck together.
  PatternMultiTopicsConsumerImpl calls client.getLookup().getTopicsUnderNamespace(...) during periodic rechecks. When multiple pattern consumers recheck at the same time, they make concurrent calls with identical (namespace, mode, topicsPattern, topicsHash). Without deduplication, each call sends a separate request to the broker

• Several pattern consumers start at once.
  During initial subscribe, PulsarClientImpl#patternTopicSubscribeAsync also calls BinaryProtoLookupService#getTopicsUnderNamespace. Creating several pattern consumers around the same time leads to multiple concurrent calls with identical parameters (namespace, mode, topicsPattern, topicsHash); without deduplication, each call becomes a separate broker request.

Effect

• Request amplification: multiple identical lookups to the broker.
• Increased broker load: unnecessary processing of duplicate requests.
• Client overhead: duplicate I/O and response handling.
• Inconsistency: behavior diverges from BinaryProtoLookupService#getBroker and BinaryProtoLookupService#getPartitionedTopicMetadata, which already coalesce identical in-flight requests.

What this PR changes
This PR adds deduplication to BinaryProtoLookupService#getTopicsUnderNamespace. Concurrent calls with the same (namespace, mode, topicsPattern, topicsHash) now share one CompletableFuture and one network request, reducing duplicate broker traffic and client work. The implementation follows the same pattern already used by BinaryProtoLookupService#getBroker and BinaryProtoLookupService#getPartitionedTopicMetadata (coalesce via computeIfAbsent, remove on whenComplete), keeping retry/backoff logic unchanged.

Modifications

• Added deduplication map

  private final ConcurrentHashMap<TopicsUnderNamespaceKey, CompletableFuture<GetTopicsResult>>
            topicsUnderNamespaceInProgress = new ConcurrentHashMap<>();
  

• Implemented composite key class
  Added TopicsUnderNamespaceKey (static inner class) to uniquely identify requests by (namespace, mode, topicsPattern, topicsHash)

• Refactored BinaryProtoLookupService#getTopicsUnderNamespace method:
  BinaryProtoLookupService#getTopicsUnderNamespace now:

  • Uses ConcurrentHashMap.computeIfAbsent(...) to atomically coalesce identical in-flight requests and return the same CompletableFuture.
  • Registers whenComplete(...) on the shared future to remove the map entry after completion (success or failure), matching the pattern used by BinaryProtoLookupService#getBroker and BinaryProtoLookupService#getPartitionedTopicMetadata.

• Added unit tests in BinaryProtoLookupServiceTest

Verifying this change

• Make sure that the change passes the CI checks.

Additional validation:

• BinaryProtoLookupServiceTest#testGetTopicsUnderNamespaceDeduplication
  Verifies that two in-flight calls with the same (namespace, mode, topicsPattern, topicsHash) return the same CompletableFuture and only one call is made to the connection pool; after the shared future completes, a new call with the same parameters returns a new CompletableFuture and makes one more call to the connection pool.

• BinaryProtoLookupServiceTest#testGetTopicsUnderNamespaceDeduplicationDifferentHash
  Verifies that calls with different topicsHash values are not combined: each returns a different CompletableFuture and makes a separate call to the connection pool. Cleanup is per key; completing one future does not affect the other in-flight call.

(Tests are deterministic: no sleeps or timing assumptions; they do not execute the network request path)

Personal CI Results

Tested in Personal CI fork: vinkal-chudgar#4

Status: All checks have passed (50 successful checks, 2 skipped)

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

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: vinkal-chudgar#4