api(ticdc): parse the kafka config

[Rustin170506]

This is an automated cherry-pick of #8938

What problem does this PR solve?

Issue Number: close #8865

What is changed and how it works?

add oauth support for sarama Kafka sink.

Added:

• SASLOAuthClientID (required)
• SASLOAuthClientSecret (required)
• SASLOAuthTokenURL (required)
• SASLOAuthScopes
• SASLOAuthGrantType (default: client_credentials)

Check List

Tests

• Manual test

Questions

Will it cause performance regression or break compatibility?

No

Do you need to update user documentation, design documentation or monitoring documentation?

No

Release note

None

[Rustin170506]

/test all

[Rustin170506]

KOP test:

1. Create an API on Auth0
2. Find its test application in Applications
   1. Use domain as the issuer URL
   2. Get the client ID and client secret
   3. Get the signing secret from the Token Settings of the API
3. Start the pulsar cluster

version: '3.5'

services:
  standalone:
    container_name: standalone
    hostname: localhost
    image: streamnative/sn-pulsar:2.10.3.7
    command: >
      bash /waitForOAuth.sh
    environment:
      allowAutoTopicCreationType: partitioned
      brokerDeleteInactiveTopicsEnabled: "false"
      PULSAR_PREFIX_saslAllowedMechanisms: OAUTHBEARER
      PULSAR_PREFIX_kopOauth2AuthenticateCallbackHandler: io.streamnative.pulsar.handlers.kop.security.oauth.OauthValidatorCallbackHandler
      PULSAR_PREFIX_kopOauth2ConfigFile: /conf/kop-handler.properties
      PULSAR_PREFIX_messagingProtocols: kafka
      PULSAR_PREFIX_kafkaListeners: SASL_PLAINTEXT://0.0.0.0:9092
      PULSAR_PREFIX_kafkaAdvertisedListeners: SASL_PLAINTEXT://xxxx:19092
      PULSAR_PREFIX_brokerEntryMetadataInterceptors: org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor
      PULSAR_PREFIX_authenticationEnabled: true
      PULSAR_PREFIX_authenticationProviders: org.apache.pulsar.broker.authentication.AuthenticationProviderToken
      PULSAR_PREFIX_brokerClientAuthenticationPlugin: org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2
      PULSAR_PREFIX_brokerClientAuthenticationParameters: "{\"type\":\"client_credentials\",\"privateKey\":\"/conf/client_credentials.json\",\"issuerUrl\":\"https://dev-xxx.us.auth0.com\", \"audience\": \"pulsar\"}"
      PULSAR_PREFIX_tokenSecretKey: "data:;base64,xxxxxx="
      LOGGING_LEVEL: debug
    ports:
      - 6650:6650
      - 8080:8080
      - 19092:9092
    volumes:
      - ./client_credentials.json:/conf/client_credentials.json
      - ./kop-handler.properties:/conf/kop-handler.properties
      - ./waitForOAuth.sh:/waitForOAuth.sh

client_credentials.json:

{
  "type": "client_credentials",
  "client_id": "h2IA1jxxjxFZBgpX6z",
  "client_secret": "-BHtxxxxGW6bJwfihRO9YN2ADUl",
  "issuer_url": "https://dxxx44.us.auth0.com",
  "audience": "pulsar"
}

kop-handler.properties:

oauth.validate.method=token

waitForOAuth.sh:

bin/apply-config-from-env.py conf/standalone.conf && exec bin/pulsar standalone

4. start the tidb cluster: tiup playground v6.5.0 --tiflash 0
5. start the ticdc cluster: ./cdc server --log-level=debug > test.log
6. create a chanegfeed with this config: ./cdc cli changefeed create --sink-uri="kafka://10.2.7.27:19092/testdsada?max-message-bytes=1048576&protocol=open-protocol&kafka-version=2.0.0&replication-factor=1" --config=./changefeed.toml

[sink]
kafka-config.sasl-mechanism = "OAUTHBEARER"
kafka-config.sasl-oauth-client-id = "xxx"
kafka-config.sasl-oauth-client-secret = "xxxx=="
kafka-config.sasl-oauth-token-url = "https://dxxxh/token"
kafka-config.sasl-oauth-grant-type = "client_credentials"
kafka-config.sasl-oauth-audience = "pulsar"

7. write some data:

mysql> use test;
Database changed
mysql> CREATE TABLE example1 (     id INT PRIMARY KEY,     b  INT );
Query OK, 0 rows affected (0.12 sec)

mysql> INSERT INTO example1 (id, b) VALUES (2, 2);
Query OK, 1 row affected (0.01 sec)

8. check the changefeed lag:

 bin git:(rustin-patch-oauth) ✗ ./cdc cli changefeed list
[
  {
    "id": "e3b3b577-2197-4cfc-9b42-e0a1775abe44",
    "namespace": "default",
    "summary": {
      "state": "normal",
      "tso": 441689899176558596,
      "checkpoint": "2023-05-24 15:27:30.643",
      "error": null
    }
  }
]

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: 3AceShowHand, CharlesCheung96

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

• OWNERS [3AceShowHand,CharlesCheung96]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[ti-chi-bot]

[LGTM Timeline notifier]

Timeline:

• 2023-07-26 01:48:40.262068942 +0000 UTC m=+319062.856599936: ☑️ agreed by 3AceShowHand.
• 2023-07-26 02:22:20.255372882 +0000 UTC m=+321082.849903869: ☑️ agreed by CharlesCheung96.

[Rustin170506]

/test all

[Rustin170506]

/test all

[Rustin170506]

Self check