sink/kafka(ticdc): support PLAIN and GSSAPI auth

[Rustin170506]

What problem does this PR solve?

Issue Number: ref #4423

What is changed and how it works?

• Support PLAIN auth
• Support GSSAPI auth

I refer to the check logic here at https://github.com/Shopify/sarama/blob/947343309601b4eb3c2fa3e7d15d701b503dd491/config.go#L563 to fill in these values. I didn't add a duplicate check because we create the sink directly when we create the changefeed, so it will be checked.

Check List

Tests

• Unit test
• Manual test (add detailed scripts or steps below)
• sink/kafka(ticdc): support PLAIN and GSSAPI auth #5122 (comment)
• sink/kafka(ticdc): support PLAIN and GSSAPI auth #5122 (comment)

Code changes

• Has exported function/method change
• Has exported variable/fields change

Side effects

None

Related changes

• Needs update docs

Release note

None

[ti-chi-bot]

[REVIEW NOTIFICATION]

This pull request has been approved by:

• 3AceShowHand
• amyangfei

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.

[codecov-commenter]

Codecov Report

Merging #5122 (76448fb) into master (687e248) will decrease coverage by 0.3222%.
The diff coverage is 56.8295%.

Flag	Coverage Δ
cdc	60.6716% <51.7446%> (+0.0729%)	⬆️
dm	51.9175% <84.3373%> (-0.5521%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

@@               Coverage Diff                @@
##             master      #5122        +/-   ##
================================================
- Coverage   56.1240%   55.8017%   -0.3223%     
================================================
  Files           522        526         +4     
  Lines         65325      67919      +2594     
================================================
+ Hits          36663      37900      +1237     
- Misses        25094      26301      +1207     
- Partials       3568       3718       +150     

[Rustin170506]

/run-all-tests

[Rustin170506]

/run-kafka-integration-test

[Rustin170506]

PLAIN:

1. start kafka playgroud with https://github.com/vdesabou/kafka-docker-playground/tree/master/environment/sasl-plain
2. start tidb cluser
3. start ticdc server
4. create changefeed

➜  bin git:(rustin-patch-auth) ./cdc cli changefeed create   --sink-uri="kafka://localhost:29092/ticdc-test?protocol=open-protocol&sasl-mechanism=plain&sasl-user=client&sasl-password=client-secret" -c "test"
[2022/04/07 17:03:57.496 +08:00] [WARN] [kafka.go:447] ["broker's `message.max.bytes` less than the `max-message-bytes`,use broker's `message.max.bytes` to initialize the Kafka producer"] [message.max.bytes=1048588] [max-message-bytes=10485760]
[2022/04/07 17:03:57.496 +08:00] [WARN] [kafka.go:457] ["partition-num is not set, use the default partition count"] [topic=ticdc-test] [partitions=3]
Create changefeed successfully!
ID: test
Info: {"sink-uri":"kafka://localhost:29092/ticdc-test?protocol=open-protocol\u0026sasl-mechanism=plain\u0026sasl-user=client\u0026sasl-password=client-secret","opts":{},"create-time":"2022-04-07T17:03:57.430325+08:00","start-ts":432359928600199169,"target-ts":0,"admin-job-type":0,"sort-engine":"unified","sort-dir":"","config":{"case-sensitive":true,"enable-old-value":true,"force-replicate":false,"check-gc-safe-point":true,"filter":{"rules":["*.*"],"ignore-txn-start-ts":null},"mounter":{"worker-num":16},"sink":{"dispatchers":null,"protocol":"open-protocol","column-selectors":null},"cyclic-replication":{"enable":false,"replica-id":0,"filter-replica-ids":null,"id-buckets":0,"sync-ddl":false},"scheduler":{"type":"table-number","polling-time":-1},"consistent":{"level":"none","max-log-size":64,"flush-interval":1000,"storage":""}},"state":"normal","error":null,"sync-point-enabled":false,"sync-point-interval":600000000000,"creator-version":"v6.0.0-master"}

[Rustin170506]

GSSAPI:

1. build ticdc image named qa/ticdc-hack
2. start kafka playgroud with https://github.com/hi-rustin/kafka-docker-playground/tree/master/environment/kerberos
3. cp kafka-sarama.key and krb5.conf form broker container into configs
4. create changefeed:

./cdc cli changefeed create --pd http://upstream-pd:2379 --log-level debug  --sink-uri="kafka://broker.kerberos-demo.local:9092/connect-offsets?protocol=open-protocol&sasl-mechani
sm=gssapi&sasl-gssapi-auth-type=keytab&sasl-gssapi-keytab-path=/configs/kafka-sarama.key&sasl-gssapi-kerberos-config-path=/configs/krb5.conf&sasl-gssapi-service-name=kafka&sasl-gssapi
-user=sarama&sasl-gssapi-realm=TEST.CONFLUENT.IO&sasl-gssapi-disable-pafxfast=false&replication-factor=1&partition-num=1" -c "test"

5. created

[2022/04/08 03:41:46.096 +00:00] [INFO] [kafka.go:239] ["async client closed"] [duration=595.285µs] [changefeed=] [role=cdc-client]
[2022/04/08 03:41:46.097 +00:00] [DEBUG] [async_producer.go:1111] ["Producer shutting down."] [name=sarama]
[2022/04/08 03:41:46.098 +00:00] [INFO] [kafka.go:249] ["sync client closed"] [duration=957.626µs] [changefeed=] [role=cdc-client]
[2022/04/08 03:41:46.097 +00:00] [DEBUG] [sarama.go:122] ["Closed connection to broker broker.kerberos-demo.local:9092"] [name=sarama]
[2022/04/08 03:41:46.099 +00:00] [DEBUG] [sarama.go:125] ["Closing Client"] [name=sarama]
[2022/04/08 03:41:46.100 +00:00] [INFO] [kafka.go:260] ["kafka cluster admin closed"] [duration=1.741042ms] [changefeed=] [role=cdc-client]
{"level":"debug","ts":"2022-04-08T03:41:46.104Z","caller":"[email protected]/retry_interceptor.go:53","msg":"retrying of unary invoker","target":"etcd-endpoints://0xc0008e8700/upstream-pd:2379","attempt":0}
[2022/04/08 03:41:46.101 +00:00] [DEBUG] [sarama.go:122] ["Closed connection to broker broker2.kerberos-demo.local:9092"] [name=sarama]
[2022/04/08 03:41:46.105 +00:00] [DEBUG] [sarama.go:122] ["Closed connection to broker broker.kerberos-demo.local:9092"] [name=sarama]
[2022/04/08 03:41:46.105 +00:00] [DEBUG] [sarama.go:122] ["Closed connection to broker broker.kerberos-demo.local:9092"] [name=sarama]
Create changefeed successfully!
ID: test
Info: {"sink-uri":"kafka://broker.kerberos-demo.local:9092/connect-offsets?protocol=open-protocol\u0026sasl-mechanism=gssapi\u0026sasl-gssapi-auth-type=keytab\u0026sasl-gssapi-keytab-path=/configs/kafka-sarama.key\u0026sasl-gssapi-kerberos-config-path=/configs/krb5.conf\u0026sasl-gssapi-service-name=kafka\u0026sasl-gssapi-user=sarama\u0026sasl-gssapi-realm=TEST.CONFLUENT.IO\u0026sasl-gssapi-disable-pafxfast=false\u0026replication-factor=1\u0026partition-num=1","opts":{},"create-time":"2022-04-08T03:41:44.782689125Z","start-ts":432377509774884865,"target-ts":0,"admin-job-type":0,"sort-engine":"unified","sort-dir":"","config":{"case-sensitive":true,"enable-old-value":true,"force-replicate":false,"check-gc-safe-point":true,"filter":{"rules":["*.*"],"ignore-txn-start-ts":null},"mounter":{"worker-num":16},"sink":{"dispatchers":null,"protocol":"open-protocol","column-selectors":null},"cyclic-replication":{"enable":false,"replica-id":0,"filter-replica-ids":null,"id-buckets":0,"sync-ddl":false},"scheduler":{"type":"table-number","polling-time":-1},"consistent":{"level":"none","max-log-size":64,"flush-interval":1000,"storage":""}},"state":"normal","error":null,"sync-point-enabled":false,"sync-point-interval":600000000000,"creator-version":"v6.0.0-master-dirty"}

[Rustin170506]

@3AceShowHand @amyangfei ping~

[amyangfei]

Please remember to update docs repo about the SASL mechanism changes, including

• old version TiCDC doesn't support PLAIN mechanism
• PLAIN and GSSAPI auth are added since version (maybe v6.1)?

[Rustin170506]

/run-all-tests

[Rustin170506]

/merge

[ti-chi-bot]

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

Commit hash: 73435f4