[FEATURE]: Configure `CryptoKeyReader` in `FlinkPulsarSource` and `FlinkPulsarSink`

[objecttrouve]

Is your feature request related to a problem? Please describe.

I want to configure a CryptoKeyReader with both a FlinkPulsarSource and a FlinkPulsarSink. I found that this isn't possible just by making settings in the Properties parameter (see also #11562 in the apache/pulsar project).

Describe the solution you'd like

Possibility to pass an instance of CryptoKeyReader as an argument to the construction of FlinkPulsarSource and FlinkPulsarSink.

Describe alternatives you've considered

Specify a class name in the properties and do a PR in the pulsar project that instantiates it. However, in most cases you'll want to configure the CryptoKeyReader and a no-arg instance would be useless. So it's not a good idea.

Additional context

Also asked about it on Slack.

[luigidalmare]

i have the same issue. is there any plan to configure cryptokeyreader?

[objecttrouve]

@luigidalmare, there's a PR about to be merged.

[syhily]

@objecttrouve Sorry for the late response. I still want to drop the Builder class, I would do a little code refactor and get this could be merged on release-1.9, release-1.11, release-1.2 and master branch.