Fix ConsumerBuilderImpl#subscribeAsync blocks calling thread.

[Technoboy-]

Fixes #14413

Motivation

When Retry topics are enabled, ConsumerBuilderImpl performs a backwards-compatibility check to look for DLQ or Retry topics that were created on previous version of Pulsar using a different naming scheme.

It does this by calling getPartitionedTopicMetadata and then using .get() to block while waiting for the results

if (client.getPartitionedTopicMetadata(oldRetryLetterTopic)
        .get(client.conf.getOperationTimeoutMs(), TimeUnit.MILLISECONDS).partitions > 0) {
    retryLetterTopic = oldRetryLetterTopic;
}
if (client.getPartitionedTopicMetadata(oldDeadLetterTopic)
        .get(client.conf.getOperationTimeoutMs(), TimeUnit.MILLISECONDS).partitions > 0) {
    deadLetterTopic = oldDeadLetterTopic;
}

This was implemented in #10129
A partial fix to add a Timeout was implemented in #11597

However this fix does still block the calling thread during the lookup.

This can be an issue for code that attempt to call subscribeAsync on a non-blocking Pool (such as when using Netty). The signature of subscribeAsync implies that it's non-blocking. (And it seems somewhat pointless to have subscribeAsync if it blocks anyway)

Note that this is an undocumented breaking change in behavior. Up until 2.9, it was safe to call this from a non-blocking pool.

In our case, we were running a custom SLF4J log exporter on the same thread pool, which resulted in a deadlock when the number of concurrent subscribeAsync calls exceeded the pool size. (This is probably an extreme example and a poor decision on our part, but perhaps a good example of why unexpected blocking can be dangerous)

Note that blocking call is still made even if the retry and DLQ names are explicitly specified in DeadLetterPolicy. In that scenario the check should not be needed. Aside from blocking the calling Thread, this also results in unneeded lookup requests.

Modifications

• Make retry and DLQ metadata async.

Documentation

• no-need-doc

[Jason918]

Overall LGTM.
I wonder if it's possible that we can add some kind of client version check to avoid this compatibility check?

[jimmydivvy]

I wonder if it's possible that we can add some kind of client version check to avoid this compatibility check?

Perhaps a flag on DeadLetterPolicy that checks for legacy topics. Enabled by default for compatibility, but can be disabled to avoid the performance overhead?

DeadLetterPolicy.builder().checkForLegacyTopics(false)

[Technoboy-]

I wonder if it's possible that we can add some kind of client version check to avoid this compatibility check?

Perhaps a flag on DeadLetterPolicy that checks for legacy topics. Enabled by default for compatibility, but can be disabled to avoid the performance overhead?

DeadLetterPolicy.builder().checkForLegacyTopics(false)

Yes, we can open a new issue for discussion.

[Technoboy-]

@Jason918 @codelipenghui @gaoran10 @mattisonchao
Could you please help review again?

[github-actions]

@Technoboy- Please provide a correct documentation label for your PR.
Instructions see Pulsar Documentation Label Guide.