[server][vpj][controller] Add support to ingest from separate RT topic in A/A SIT when feature is enabled

[sixpluszero]

[server][vpj][controller] Add support to ingest from separate RT topic in A/A SIT when feature is enabled

This PR continue the work from #1262 and #1172 to add server side ingestion functionality for separate realtime topic.

1. The scope of this PR only covers A/A SIT, and does not support in L/F SIT. This can be left as future work if we identify it is useful in other use cases other than incremental push.
2. This PR adjusts an assumption in previous PR: We will only write incremental push to separate RT topic when specified. This is to make sure we will have a smooth transition period, as there can be mixed separated RT config specification for current / backup version, and VPJ directly writing incremental push to the separate RT topic will result in data loss in one version. For this, we add a new config in requestTopic API in VPJ and user can override it in VPJ config.
3. For subscription, we will require adding new config in server kafka cluster mapping. This PR added some utils methods to check / resolve special URL/region naming and topic resolution to make sure we have correct subscription and offset tracking. For TopicSwitch I decided not to touch it at all, and we will only complement the separate RT offset calculation when needed.
4. For consumer assignment, this PR assigned separate RT into 2nd level of priority pool. For AAWC leader strategy, it is assigned to the default pool; For current version prioritization strategy, this PR adds a current_sep_rt pool, it is assigned to that pool; For non_current version, it is on non AAWC leader pool. Personally I think it is fine as beginning, we might want to adjust to allocate separate pool if there is performance issue.
5. For metrics: There is no HB metrics as leader does not emit HB into it (should we?? maybe we should as we might still need stuck pipeline detection, but without alert). For lag/rate metrics, AASIT stats implementation natively offer the metrics, and we will mute separate RT region metric registration if the SIT does not enable separate topic.
6. Did minor code clean up in AASIT as I think there are some code duplication, but tried to keep it as minimal as the PR itself is large.
7. Adjusted a watermark filter implementation in RmdUtils used in CDC (Although it is somewhat weird and open for comment to make sure I understand the original intention)

How was this PR tested?

Added new integration test; Added some unit tests for test coverage; Adjusted some old integration test / unit test behaviors.

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.

[haoxu07]

Thanks for preparing this! I am still in the middle of reading the SIT changes, just left a few comments.

[haoxu07]

Seems this test-only PubSubTopicRepository is not necessary, PubSubTopicImpl could handle equal check by comparing String name

[sixpluszero]

Hmm, actually this is used in main code as well. PubSubTopicImpl constructor is package private by design so it is not accessible from SIT class.