sink(ticdc): Optimize the performance when getting the metadata of Kafka topics (#9060)

[ti-chi-bot]

This is an automated cherry-pick of #9060

What problem does this PR solve?

Issue Number: close #8959 close #8957

What is changed and how it works?

Get the metadata by needs instead of getting all topic's metadata.

Fixed a bug in the original implementation, the semantics of the original GetTopicMeta v2 implementation and v1 were not the same. Now it's the same, add a continue and it's fine.

Check List

Tests

• Unit test
• Manual test

Questions

Will it cause performance regression or break compatibility?

No

Do you need to update user documentation, design documentation or monitoring documentation?

No

Release note

Optimize the performance when TiCDC gets the metadata of Kafka topics

[ti-chi-bot]

[REVIEW NOTIFICATION]

This pull request has not been approved.

To complete the pull request process, please ask the reviewers in the list to review by filling /cc @reviewer in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the list by filling /assign @committer in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found here.

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.

[Rustin170506]

/run-all-tests

/test all

[Rustin170506]

/run-all-tests

[Rustin170506]

• Close topic manager

[Rustin170506]

/run-all-tests

[Rustin170506]

Tested locally:
exist:

[2023/06/02 16:02:11.879 +08:00] [INFO] [kafka.go:412] ["topic already exists"] [topic=ticdc-test1]
[2023/06/02 16:02:11.891 +08:00] [INFO] [kafka.go:598] ["Kafka config item found"] [topicName=ticdc-test1] [configName=min.insync.replicas] [configValue=1]
[2023/06/02 16:02:11.894 +08:00] [INFO] [kafka.go:598] ["Kafka config item found"] [topicName=ticdc-test1] [configName=max.message.bytes] [configValue=3145728]
[2023/06/02 16:02:13.035 +08:00] [INFO] [kafka_manager.go:142] ["store topic partition number"] [namespace=] [changefeed=] [topic=ticdc-test1] [partitionNumber=3]

not exist:

[2023/06/02 16:04:27.110 +08:00] [INFO] [kafka_manager.go:269] ["Kafka admin client create the topic success"] [namespace=] [changefeed=] [topic=ticdc-test2] [partitionNumber=3] [replicationFactor=1] [duration=95.302833ms]

[Rustin170506]

/merge

[ti-chi-bot]

This pull request has been accepted and is ready to merge.

Commit hash: 1854e38

[Rustin170506]

/run-all-tests

[Rustin170506]

/retest

[Rustin170506]

/run-integration-test

[Rustin170506]

/run-all-tests

[ti-chi-bot]

In response to a cherrypick label: new pull request created to branch release-6.1: #9145.