[controller][server] TopicManager refactoring and performance improvements

[sushantmane]

TopicManager Refactoring and Performance Improvements

Currently, we have three components that work together to fetch topic metadata,
such as endOffset and the timestamp of the last produced message. These
components are TopicManager, PartitionOffsetFetcher, and
CachedPubSubMetadataGetter. We create one TopicManager instance per broker
address/region, which is shared by all the components in the system.
PartitionOffsetFetcher is a part of the topic manager. There is one instance of
CachedPubSubMetadataGetter per StoreIngestionTask, which makes use of
TopicManager and PartitionOffsetFetcher to retrieve topic metadata.

PartitionOffsetFetcher uses pubsub admin and consumer clients to fetch offsets on
behalf of components like CachedPubSubMetadataGetter and other components. Each
PartitionOffsetFetcher has just one instance of a consumer, which is protected
with a consumer mutex since consumers are not thread-safe. This means that
PartitionOffsetFetcher can handle only one request at a time. The consumer mutex
is coarse-grained, which means it holds the lock even when making calls with the
admin client, which is unnecessary since admin clients are thread-safe.

Also, when the mutex is held for topics that do not exist or have metadata fetching
issues, we perform retries while holding the mutex. This results in the mutex being
held for a relatively longer time when there is an issue with one topic, directly
causing delays for other requests waiting for the lock. We have often observed high
contention on this mutex, leading to several issues.

The CachedPubSubMetadataGetter fetches offsets asynchronously using
TopicManager/PartitionOffsetFetcher, and caches them in a local cache. These
asynchronous requests are initiated when an entry is either not in the cache or
when it is stale. Currently, these asynchronous requests are served by the ForkJoin
common thread pool.

When there is an issue like a non-existent topic, we see high contention on the
consumer mutex, as discussed earlier. This results in ForkJoin threads being
blocked on the consumer lock for longer periods. When all the ForkJoin common
thread pools get blocked, other components in the system that rely on them are
starved, which affects their functionality.

To enhance the resiliency of TopicManager in the face of failures, this PR
introduces the following high-level changes:

1. Admin operations during metadata fetch are now performed without holding any lock.
   This prevents bad topics from acquiring the consumer in the first place and reduces
   the chances of contention when getting the consumer.

2. A configurable fetcher consumer pool is introduced for each topic manager. This
   allows us to process requests with other consumers even if one is busy.

3. We have discontinued the use of the ForkJoin common thread pool for submitting
   async fetch offset requests. Instead, dedicated fetcher thread pools are used for this purpose.

Miscellaneous

• This commit introduced the following configs:
  1. pubsub.topic.manager.metadata.fetcher.consumer.pool.size - A config to specify number of PubSub consumer clients to use per topic manager (default: 2)
  2. pubsub.topic.manager.metadata.fetcher.thread.pool.size - Controls the size of per TM thread-pool used for get offset async requests (default: 2).
• The TopicManagerRepository.SSLPropertiesSupplier has been extracted into PubSubPropertiesSupplier.
• The CachedPubSubMetadataGetter and PartitionOffsetFetcher have been merged to form TopicManagerMetadataFetcher.
• The method getOffsetLagFor in AggKafkaConsumerService has been renamed to getConsumerLagBasedOnMetrics.
• The method getPartitionOffsetLag in ActiveActiveStoreIngestionTask has been renamed to getPartitionOffsetLagBasedOnMetrics.
• TopicManagerContext has been created/extracted from TopicManagerRepository.builder() to pass the required dependencies for TopicManagerRepository.

How was this PR tested?

CI, frankenwar testing

Does this PR introduce any user-facing changes?

• No. You can skip the rest of this section.
• Yes. Make sure to explain your proposed changes and call out the behavior change.

[FelixGV]

A few comments, all minor... looks great so far. I still have about half to review.

[lluwm]

Nice work and it makes a lot of sense to me to have a centralized place/class to coordinate the work of getting offset. Thank you @sushantmane!

[FelixGV]

Finished first pass... a few more comments. Nothing dramatic, LG overall.

[FelixGV]

Just one minor comment... probably good to go very soon! Thanks for continuing to iterate on this big change!

[gaojieliu]

Looks great overall, and left some minor comments.

[FelixGV]

Thank you Sushant! Let's take this monster PR across the finish line 🚀🚢 !!!!!

[sushantmane]

Thanks a lot @FelixGV, @lluwm, and @gaojieliu for the review, discussions, and help in improving this PR. 🙏 🚀