[pulsar-io] Support authentication on debezium connector

[rdhabalia]

Motivation

Right now, Kafka-connector/Debezium connector doesn't support tls with source configuration which is used to create pulsar consumer for offset topic. adding support of authentication with source-configuration pulsar.auth.plugin, pulsar.auth.plugin.param, pulsar.tls.insecure.connection , pulsar.tls.trust.cert

Sample command:

 ./pulsar-admin source localrun \
--name debezium-postgres-source \
--tenant pulsar \
--namespace us-west/ns1 \
--destination-topic-name persistent://pulsar/us-west/ns1/deb-dest \
--archive /pulsar-io-debezium-mysql.nar \
--processingGuarantees ATLEAST_ONCE \
--broker-service-url pulsar+ssl://my-cluster.com:6651/ \
--client-auth-params tlsCertFile:/client.cert.pem,tlsKeyFile:/client.key.pem \
--client-auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationTls \
--tls-trust-cert-path /ca.cert.pem \
--use-tls \
--source-config '{"database.hostname":"db.com","database.port":"1111","database.user":"u1","database.password":"p1","database.dbname":"db","database.server.name":"s1","table.include.list":"t1","pulsar.service.url":"pulsar+ssl://my-cluster.com:6651","offset.storage.topic":"persistent://pulsar/us-west/ns1/deb-storage-topic","pulsar.auth.plugin":"org.apache.pulsar.client.impl.auth.AuthenticationTls", "pulsar.auth.plugin.param":"tlsCertFile:/client.cert.pem,tlsKeyFile:/client.key.pem", "pulsar.tls.insecure.connection":"true","pulsar.tls.trust.cert":"/ca.cert.pem"}'

[sijie]

@rdhabalia I moved this milestone from 2.7.0 to 2.8.0 because @codelipenghui is cutting a release for 2.7.0.

[sijie]

This doesn't seem to be the right approach. I think the pulsar cluster settings should be inherited from Pulsar Functions Worker settings and passed to the Pulsar IO connector, and then to the Kafka Connect settings.

The change seems to take a different approach to get the cluster settings from functions config. It seems to an anti-pattern on how we manage the pulsar cluster settings for functions and connectors.

[rdhabalia]

I guess you misunderstood the change. There are two kinds of topics here: source-topic and connector-specific topic (debezium-offset.storage.topic) . Now, debezium source level configurations are part of source-config such as database.hostname, password, offset.storage.topic. and this authentication credential is associated with offset.storage.topic so, it should be part of debezium source-config only.

[sijie]

@rdhabalia I realized that when answering the question in the other issue. @jiazhai and I made a mistake when we added the kafka-connect-adapter for debezium. We should try to inherit the pulsar service URL from the function worker. Otherwise, it causes confusion when people attempt to use debezium connector. I'd suggest we fixing the issue instead of introducing more pulsar settings to debezium connector.

[freeznet]

@sijie I agree with your thought that try to inherit the pulsar service URL and auth configs from the function worker. But I dont think we should expose them to Context, which might have some security issues.
I dont have too much experiences about Kafka-connector/Debezium connector, but what I noticed from this PR is to inherit PulsarClient to connector. If i am correct, I would like to suggest to add PulsarClient getClient() into function/connector Context, then we can use the client to PulsarOffsetBackingStore. /cc @rdhabalia

[sijie]

@rdhabalia Can you review @freeznet 's comment?