[fix][broker] Avoid blocking metadata read on the IO thread when redirecting migrated producers/consumers - #26051

Motivation

AbstractTopic.getMigratedClusterUrl() resolved the migrated-cluster URL by blocking on getMigratedClusterUrlAsync().get(operationTimeoutSeconds). Its clean, non-throwing signature hid that it blocks, and several callers run on threads where blocking is harmful:

• ServerCnx subscribe-success path, inside .thenAcceptAsync(..., ctx.executor()) (the connection's Netty IO thread), via Consumer.checkAndApplyTopicMigration();
• the producer/consumer migration-redirect exceptionallyAsync(..., ctx.executor()) handlers in ServerCnx;
• AbstractBaseDispatcher.checkAndApplyReachedEndOfTopicOrTopicMigration() on the dispatch path;
• PersistentTopic / NonPersistentTopic.checkClusterMigration() and the topic write-failed path.

A slow or contended metadata read could therefore stall a Netty IO thread for up to the metadata operation timeout (default 30s). Blocking with get() on one pool while the future is completed on another is also a deadlock hazard if the completing pool is saturated.

Modifications

Route every caller through the already-existing async getMigratedClusterUrlAsync() and delete the blocking getMigratedClusterUrl() overloads:

• Consumer.checkAndApplyTopicMigration() → checkAndApplyTopicMigrationAsync() returning CompletableFuture<Boolean> (reuses topicMigrated(...) for the send + disconnect).
• ServerCnx: subscribe-success uses thenCompose so the migration check stays async on ctx.executor(); the three producer/consumer migration-redirect handlers resolve the URL asynchronously and then redirect-or-fail in small helpers that preserve each branch's exact cleanup and logging.
• AbstractBaseDispatcher, PersistentTopic, NonPersistentTopic: resolve the URL asynchronously. checkClusterMigration resolves it once and reuses it for each consumer, dropping a redundant per-consumer blocking re-resolve.

A failed metadata resolution is no longer swallowed into "not migrated"; it surfaces instead. No public/admin API or wire-protocol change — getMigratedClusterUrlAsync() already existed and already hops completion to pulsar.getExecutor() to avoid the metadata-thread deadlock.

Verifying this change

This change is already covered by existing tests: ClusterMigrationTest — producer & consumer redirect, persistent & non-persistent topics, replication-backlog, resource-created, and namespace-migration cases (18 executions, all passing locally).

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