[FLINK-35477],[FLINK-37299] Solve the problem that when not starting from the Checkpoint (CK), the cursor is always reset to the set initial cursor position.

[thomasg19930417]

Purpose of the change

For example: Add dynamic sink topic support for Pulsar connector.

Brief change log

• Change the internal design of ProducerRegister.
• Expose topic metadata query in PulsarSinkContext.
• Change the internal metadata cache in MetadataListener.

Verifying this change

Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added unit tests
• Added integration tests for end-to-end deployment
• Manually verified by running the Pulsar connector on a local Flink cluster.

Significant changes

(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)

• [ ] Dependencies have been added or upgraded
• [ ] Public API has been changed (Public API is any class annotated with @Public(Evolving))
• [ ] Serializers have been changed
• [ ] New feature has been introduced
  • If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)

[boring-cyborg[bot]]

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

[thomasg19930417]

In Pull Request #103, only the isResetSubscriptionCursor parameter was simply enabled. If this parameter is set to false, then there will be no code related to setting the initial cursor. All subscriptions will start from the default earliest or latest position, instead of the explicitly set earliest or latest position.

[vbabenkoru]

@ferenc-csaky @thomasg19930417 How is this different from .setStartCursor(startCursor) set on the PulsarSource?