Added ability to add annotations to Connector Configs

[srkukarni]

(If this PR fixes a github issue, please add Fixes #<xyz>.)

Fixes #

(or if this PR is one task of a github issue, please add Master Issue: #<xyz> to link to the master issue.)

Master Issue: #

Motivation

This pr adds the ability for pulsar io sources and sinks to define annotations in their configs and have them checked at submission time. This means that creation/update of connectors with improper config can be rejected outright at request time rather than at runtime which is the current case. This will lead to faster troubleshooting when developing and running connectors. We make this check gated by a worker config parameter.

Modifications

Describe the modifications you've done.

Verifying this change

• Make sure that the change passes the CI checks.

(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 integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API: (yes / no)
• The schema: (yes / no / don't know)
• The default values of configurations: (yes / no)
• The wire protocol: (yes / no)
• The rest endpoints: (yes / no)
• The admin cli options: (yes / no)
• Anything that affects deployment: (yes / no / don't know)

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

[jerrypeng]

Please use ObjectMapperFactory.getThreadLocal

[srkukarni]

Changed

[jerrypeng]

You can achieve the same thing by using the following method

mapper.convertValue(map, MyPojo.class);

[srkukarni]

Changed

[jerrypeng]

@srkukarni thinking more about it, I am wondering for sources and sinks whether the following interface is appropriate now that we formalized the concept of a connector config.

    void open(final Map<String, Object> config, SourceContext sourceContext) throws Exception;

Since a connector will have a config class attached to it should we still have "Map<String, Object> config" as an argument or should we somehow just pass in the config POJO?

Though not sure that is possible given we only allow developers to specify in the connector config class in "pulsar-io.yaml". There is not way to use Java generics to automatically specify the type of the config argument.

[srkukarni]

@jerrypeng wrt your comment on the correct interface, yes in hindsight it does seem as if we should have had an explicit config class in the open. Almost all of the connectors in the repo seem to be following this.
Having said that I also believe that the current approach gives more flexibility wrt how connector writers evolve their config in the future. I'm not sure if it's really worthwhile for us to change the api now for the elegance that the explicit config class provides. Its something that connector writers can live with without too much difficultly.