Repository: pingcap/tiflow
Pull request #5122 (2022-04-06 09:25:20)

sink/kafka(ticdc): support PLAIN and GSSAPI auth
<!--
Thank you for contributing to TiDB-CDC! Please read MD's [CONTRIBUTING](https://github.com/pingcap/tidb-cdc/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: ref https://github.com/pingcap/tiflow/issues/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 <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - [x] Unit test
 - [x] Manual test (add detailed scripts or steps below)
  - https://github.com/pingcap/tiflow/pull/5122#issuecomment-1091382426
  - https://github.com/pingcap/tiflow/pull/5122#issuecomment-1092413151

Code changes

 - Has exported function/method change
 - Has exported variable/fields change

Side effects

None

Related changes

- [ ] Needs update docs

### Release note <!-- bugfixes or new feature need a release note -->

```release-note
None
```


ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/5122/owners) 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](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/5122/owners) 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](https://prow.tidb.io/command-help?repo=pingcap%2Ftiflow).

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
codecov-commenter: # [Codecov](https://codecov.io/gh/pingcap/tiflow/pull/5122?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
> Merging [#5122](https://codecov.io/gh/pingcap/tiflow/pull/5122?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (76448fb) into [master](https://codecov.io/gh/pingcap/tiflow/commit/687e2486c89a2a7f20bdb1db5ca37e8a777537f4?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (687e248) will **decrease** coverage by `0.3222%`.
> The diff coverage is `56.8295%`.


| Flag | Coverage Δ | |
|---|---|---|
| cdc | `60.6716% <51.7446%> (+0.0729%)` | :arrow_up: |
| dm | `51.9175% <84.3373%> (-0.5521%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#carryforward-flags-in-the-pull-request-comment) to find out more.

```diff
@@               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     
```

hi-rustin: /run-all-tests
hi-rustin: /run-kafka-integration-test
hi-rustin: GSSAPI:
1. start kafka playgroud with https://github.com/hi-rustin/kafka-docker-playground/tree/master/environment/kerberos
2. cp kafka-sarama.key form broker container into `configs`
3. create changefeed:
```sh
./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"
```
4. created
```log
[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":"v3@v3.5.2/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"}
```
hi-rustin: @3AceShowHand @amyangfei ping~
hi-rustin: /run-all-tests
hi-rustin: /merge

Patch:
diff --git a/cdc/sink/producer/kafka/config.go b/cdc/sink/producer/kafka/config.go
index 0d7dc06e0..85fc6e4e4 100644
--- a/cdc/sink/producer/kafka/config.go
+++ b/cdc/sink/producer/kafka/config.go
@@ -43,7 +43,7 @@ type Config struct {
 	Compression     string
 	ClientID        string
 	Credential      *security.Credential
-	SaslScram       *security.SaslScram
+	SASL            *security.SASL
 	// control whether to create topic
 	AutoCreate bool
 
@@ -62,7 +62,7 @@ func NewConfig() *Config {
 		ReplicationFactor: 1,
 		Compression:       "none",
 		Credential:        &security.Credential{},
-		SaslScram:         &security.SaslScram{},
+		SASL:              &security.SASL{},
 		AutoCreate:        true,
 		DialTimeout:       10 * time.Second,
 		WriteTimeout:      10 * time.Second,
@@ -97,22 +97,6 @@ func (c *Config) setPartitionNum(realPartitionCount int32) error {
 	return nil
 }
 
-// AutoCreateTopicConfig is used to create topic configuration.
-type AutoCreateTopicConfig struct {
-	AutoCreate        bool
-	PartitionNum      int32
-	ReplicationFactor int16
-}
-
-// DeriveTopicConfig derive a `topicConfig` from the `Config`
-func (c *Config) DeriveTopicConfig() *AutoCreateTopicConfig {
-	return &AutoCreateTopicConfig{
-		AutoCreate:        c.AutoCreate,
-		PartitionNum:      c.PartitionNum,
-		ReplicationFactor: c.ReplicationFactor,
-	}
-}
-
 // Apply the sinkURI to update Config
 func (c *Config) Apply(sinkURI *url.URL) error {
 	c.BrokerEndpoints = strings.Split(sinkURI.Host, ",")
@@ -174,21 +158,6 @@ func (c *Config) Apply(sinkURI *url.URL) error {
 		c.Credential.KeyPath = s
 	}
 
-	s = params.Get("sasl-user")
-	if s != "" {
-		c.SaslScram.SaslUser = s
-	}
-
-	s = params.Get("sasl-password")
-	if s != "" {
-		c.SaslScram.SaslPassword = s
-	}
-
-	s = params.Get("sasl-mechanism")
-	if s != "" {
-		c.SaslScram.SaslMechanism = s
-	}
-
 	s = params.Get("auto-create-topic")
 	if s != "" {
 		autoCreate, err := strconv.ParseBool(s)
@@ -225,9 +194,101 @@ func (c *Config) Apply(sinkURI *url.URL) error {
 		c.ReadTimeout = a
 	}
 
+	err := c.applySASL(params)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *Config) applySASL(params url.Values) error {
+	s := params.Get("sasl-user")
+	if s != "" {
+		c.SASL.SASLUser = s
+	}
+
+	s = params.Get("sasl-password")
+	if s != "" {
+		c.SASL.SASLPassword = s
+	}
+
+	s = params.Get("sasl-mechanism")
+	if s != "" {
+		mechanism, err := security.SASLMechanismFromString(s)
+		if err != nil {
+			return cerror.WrapError(cerror.ErrKafkaInvalidConfig, err)
+		}
+		c.SASL.SASLMechanism = mechanism
+	}
+
+	s = params.Get("sasl-gssapi-auth-type")
+	if s != "" {
+		authType, err := security.AuthTypeFromString(s)
+		if err != nil {
+			return cerror.WrapError(cerror.ErrKafkaInvalidConfig, err)
+		}
+		c.SASL.GSSAPI.AuthType = authType
+	}
+
+	s = params.Get("sasl-gssapi-keytab-path")
+	if s != "" {
+		c.SASL.GSSAPI.KeyTabPath = s
+	}
+
+	s = params.Get("sasl-gssapi-kerberos-config-path")
+	if s != "" {
+		c.SASL.GSSAPI.KerberosConfigPath = s
+	}
+
+	s = params.Get("sasl-gssapi-service-name")
+	if s != "" {
+		c.SASL.GSSAPI.ServiceName = s
+	}
+
+	s = params.Get("sasl-gssapi-user")
+	if s != "" {
+		c.SASL.GSSAPI.Username = s
+	}
+
+	s = params.Get("sasl-gssapi-password")
+	if s != "" {
+		c.SASL.GSSAPI.Password = s
+	}
+
+	s = params.Get("sasl-gssapi-realm")
+	if s != "" {
+		c.SASL.GSSAPI.Realm = s
+	}
+
+	s = params.Get("sasl-gssapi-disable-pafxfast")
+	if s != "" {
+		disablePAFXFAST, err := strconv.ParseBool(s)
+		if err != nil {
+			return err
+		}
+		c.SASL.GSSAPI.DisablePAFXFAST = disablePAFXFAST
+	}
+
 	return nil
 }
 
+// AutoCreateTopicConfig is used to create topic configuration.
+type AutoCreateTopicConfig struct {
+	AutoCreate        bool
+	PartitionNum      int32
+	ReplicationFactor int16
+}
+
+// DeriveTopicConfig derive a `topicConfig` from the `Config`
+func (c *Config) DeriveTopicConfig() *AutoCreateTopicConfig {
+	return &AutoCreateTopicConfig{
+		AutoCreate:        c.AutoCreate,
+		PartitionNum:      c.PartitionNum,
+		ReplicationFactor: c.ReplicationFactor,
+	}
+}
+
 // NewSaramaConfig return the default config and set the according version and metrics
 func NewSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 	config := sarama.NewConfig()
@@ -312,19 +373,42 @@ func NewSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 			return nil, errors.Trace(err)
 		}
 	}
-	if c.SaslScram != nil && len(c.SaslScram.SaslUser) != 0 {
+
+	completeSaramaSASLConfig(config, c)
+
+	return config, err
+}
+
+func completeSaramaSASLConfig(config *sarama.Config, c *Config) {
+	if c.SASL != nil && c.SASL.SASLMechanism != "" {
 		config.Net.SASL.Enable = true
-		config.Net.SASL.User = c.SaslScram.SaslUser
-		config.Net.SASL.Password = c.SaslScram.SaslPassword
-		config.Net.SASL.Mechanism = sarama.SASLMechanism(c.SaslScram.SaslMechanism)
-		if strings.EqualFold(c.SaslScram.SaslMechanism, "SCRAM-SHA-256") {
-			config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &security.XDGSCRAMClient{HashGeneratorFcn: security.SHA256} }
-		} else if strings.EqualFold(c.SaslScram.SaslMechanism, "SCRAM-SHA-512") {
-			config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &security.XDGSCRAMClient{HashGeneratorFcn: security.SHA512} }
-		} else {
-			return nil, errors.New("Unsupported sasl-mechanism, should be SCRAM-SHA-256 or SCRAM-SHA-512")
+		config.Net.SASL.Mechanism = sarama.SASLMechanism(c.SASL.SASLMechanism)
+		switch c.SASL.SASLMechanism {
+		case sarama.SASLTypeSCRAMSHA256, sarama.SASLTypeSCRAMSHA512, sarama.SASLTypePlaintext:
+			config.Net.SASL.User = c.SASL.SASLUser
+			config.Net.SASL.Password = c.SASL.SASLPassword
+			if strings.EqualFold(string(c.SASL.SASLMechanism), sarama.SASLTypeSCRAMSHA256) {
+				config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+					return &security.XDGSCRAMClient{HashGeneratorFcn: security.SHA256}
+				}
+			} else if strings.EqualFold(string(c.SASL.SASLMechanism), sarama.SASLTypeSCRAMSHA512) {
+				config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+					return &security.XDGSCRAMClient{HashGeneratorFcn: security.SHA512}
+				}
+			}
+		case sarama.SASLTypeGSSAPI:
+			config.Net.SASL.GSSAPI.AuthType = int(c.SASL.GSSAPI.AuthType)
+			config.Net.SASL.GSSAPI.Username = c.SASL.GSSAPI.Username
+			config.Net.SASL.GSSAPI.ServiceName = c.SASL.GSSAPI.ServiceName
+			config.Net.SASL.GSSAPI.KerberosConfigPath = c.SASL.GSSAPI.KerberosConfigPath
+			config.Net.SASL.GSSAPI.Realm = c.SASL.GSSAPI.Realm
+			config.Net.SASL.GSSAPI.DisablePAFXFAST = c.SASL.GSSAPI.DisablePAFXFAST
+			switch c.SASL.GSSAPI.AuthType {
+			case security.UserAuth:
+				config.Net.SASL.GSSAPI.Password = c.SASL.GSSAPI.Password
+			case security.KeyTabAuth:
+				config.Net.SASL.GSSAPI.KeyTabPath = c.SASL.GSSAPI.KeyTabPath
+			}
 		}
 	}
-
-	return config, err
 }
diff --git a/pkg/security/sasl.go b/pkg/security/sasl.go
new file mode 100644
index 000000000..8077589d5
--- /dev/null
+++ b/pkg/security/sasl.go
@@ -0,0 +1,98 @@
+// Copyright 2020 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package security
+
+import (
+	"strings"
+
+	"github.com/Shopify/sarama"
+	"github.com/pingcap/errors"
+)
+
+// SASLMechanism defines SASL mechanism.
+type SASLMechanism string
+
+// The mechanisms we currently support.
+const (
+	// UnknownMechanism means the SASL mechanism is unknown.
+	UnknownMechanism SASLMechanism = ""
+	// PlainMechanism means the SASL mechanism is plain.
+	PlainMechanism SASLMechanism = sarama.SASLTypePlaintext
+	// SCRAM256Mechanism means the SASL mechanism is SCRAM-SHA-256.
+	SCRAM256Mechanism SASLMechanism = sarama.SASLTypeSCRAMSHA256
+	// SCRAM512Mechanism means the SASL mechanism is SCRAM-SHA-512.
+	SCRAM512Mechanism SASLMechanism = sarama.SASLTypeSCRAMSHA512
+	// GSSAPIMechanism means the SASL mechanism is GSSAPI.
+	GSSAPIMechanism SASLMechanism = sarama.SASLTypeGSSAPI
+)
+
+// SASLMechanismFromString converts the string to SASL mechanism.
+func SASLMechanismFromString(s string) (SASLMechanism, error) {
+	switch strings.ToLower(s) {
+	case "plain":
+		return PlainMechanism, nil
+	case "scram-sha-256":
+		return SCRAM256Mechanism, nil
+	case "scram-sha-512":
+		return SCRAM512Mechanism, nil
+	case "gssapi":
+		return GSSAPIMechanism, nil
+	default:
+		return UnknownMechanism, errors.Errorf("unknown %s SASL mechanism", s)
+	}
+}
+
+// SASL holds necessary path parameter to support sasl-scram
+type SASL struct {
+	SASLUser      string        `toml:"sasl-user" json:"sasl-user"`
+	SASLPassword  string        `toml:"sasl-password" json:"sasl-password"`
+	SASLMechanism SASLMechanism `toml:"sasl-mechanism" json:"sasl-mechanism"`
+	GSSAPI        GSSAPI        `toml:"sasl-gssapi" json:"sasl-gssapi"`
+}
+
+// GSSAPIAuthType defines the type of GSSAPI authentication.
+type GSSAPIAuthType int
+
+const (
+	// UnknownAuth means the auth type is unknown.
+	UnknownAuth GSSAPIAuthType = 0
+	// UserAuth means the auth type is user.
+	UserAuth GSSAPIAuthType = sarama.KRB5_USER_AUTH
+	// KeyTabAuth means the auth type is keytab.
+	KeyTabAuth GSSAPIAuthType = sarama.KRB5_KEYTAB_AUTH
+)
+
+// AuthTypeFromString convent the string to GSSAPIAuthType.
+func AuthTypeFromString(s string) (GSSAPIAuthType, error) {
+	switch strings.ToLower(s) {
+	case "user":
+		return UserAuth, nil
+	case "keytab":
+		return KeyTabAuth, nil
+	default:
+		return UnknownAuth, errors.Errorf("unknown %s auth type", s)
+	}
+}
+
+// GSSAPI holds necessary path parameter to support sasl-gssapi.
+type GSSAPI struct {
+	AuthType           GSSAPIAuthType `toml:"sasl-gssapi-auth-type" json:"sasl-gssapi-auth-type"`
+	KeyTabPath         string         `toml:"sasl-gssapi-keytab-path" json:"sasl-gssapi-keytab-path"`
+	KerberosConfigPath string         `toml:"sasl-gssapi-kerberos-config-path" json:"sasl-gssapi-kerberos-config-path"`
+	ServiceName        string         `toml:"sasl-gssapi-service-name" json:"sasl-gssapi-service-name"`
+	Username           string         `toml:"sasl-gssapi-user" json:"sasl-gssapi-user"`
+	Password           string         `toml:"sasl-gssapi-password" json:"sasl-gssapi-password"`
+	Realm              string         `toml:"sasl-gssapi-realm" json:"sasl-gssapi-realm"`
+	DisablePAFXFAST    bool           `toml:"sasl-gssapi-disable-pafxfast" json:"sasl-gssapi-disable-pafxfast"`
+}
diff --git a/pkg/security/sasl_scram.go b/pkg/security/sasl_scram.go
deleted file mode 100644
index f240054b3..000000000
--- a/pkg/security/sasl_scram.go
+++ /dev/null
@@ -1,26 +0,0 @@
-// Copyright 2020 PingCAP, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package security
-
-// SaslScram holds necessary path parameter to support sasl-scram
-type SaslScram struct {
-	SaslUser      string `toml:"sasl-user" json:"sasl-user"`
-	SaslPassword  string `toml:"sasl-password" json:"sasl-password"`
-	SaslMechanism string `toml:"sasl-mechanism" json:"sasl-mechanism"`
-}
-
-// IsSaslScramEnabled checks whether SASL SCRAM is enabled or not.
-func (s *SaslScram) IsSaslScramEnabled() bool {
-	return len(s.SaslUser) != 0
-}


Test patch:
diff --git a/cdc/sink/producer/kafka/config_test.go b/cdc/sink/producer/kafka/config_test.go
index 682dc7df7..fc9ddfcdf 100644
--- a/cdc/sink/producer/kafka/config_test.go
+++ b/cdc/sink/producer/kafka/config_test.go
@@ -73,10 +73,10 @@ func TestNewSaramaConfig(t *testing.T) {
 	saslConfig := NewConfig()
 	saslConfig.Version = "2.6.0"
 	saslConfig.ClientID = "test-sasl-scram"
-	saslConfig.SaslScram = &security.SaslScram{
-		SaslUser:      "user",
-		SaslPassword:  "password",
-		SaslMechanism: sarama.SASLTypeSCRAMSHA256,
+	saslConfig.SASL = &security.SASL{
+		SASLUser:      "user",
+		SASLPassword:  "password",
+		SASLMechanism: sarama.SASLTypeSCRAMSHA256,
 	}
 
 	cfg, err := NewSaramaConfig(ctx, saslConfig)
@@ -421,3 +421,110 @@ func TestConfigurationCombinations(t *testing.T) {
 		_ = adminClient.Close()
 	}
 }
+
+func TestApplySASL(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name      string
+		URI       string
+		exceptErr string
+	}{
+		{
+			name:      "no params",
+			URI:       "kafka://127.0.0.1:9092/abc",
+			exceptErr: "",
+		},
+		{
+			name: "valid PLAIN SASL",
+			URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" +
+				"&sasl-username=user&sasl-password=password&sasl-mechanism=plain",
+			exceptErr: "",
+		},
+		{
+			name: "valid SCRAM SASL",
+			URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" +
+				"&sasl-username=user&sasl-password=password&sasl-mechanism=SCRAM-SHA-512",
+			exceptErr: "",
+		},
+		{
+			name: "valid GSSAPI user auth SASL",
+			URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" +
+				"&sasl-mechanism=GSSAPI&sasl-gssapi-auth-type=USER" +
+				"&sasl-gssapi-kerberos-config-path=/root/config" +
+				"&sasl-gssapi-service-name=a&sasl-gssapi-user=user" +
+				"&sasl-gssapi-password=pwd" +
+				"&sasl-gssapi-realm=realm&sasl-gssapi-disable-pafxfast=false",
+			exceptErr: "",
+		},
+		{
+			name: "valid GSSAPI keytab auth SASL",
+			URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" +
+				"&sasl-mechanism=GSSAPI&sasl-gssapi-auth-type=keytab" +
+				"&sasl-gssapi-kerberos-config-path=/root/config" +
+				"&sasl-gssapi-service-name=a&sasl-gssapi-user=user" +
+				"&sasl-gssapi-keytab-path=/root/keytab" +
+				"&sasl-gssapi-realm=realm&sasl-gssapi-disable-pafxfast=false",
+			exceptErr: "",
+		},
+		{
+			name: "invalid mechanism",
+			URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" +
+				"&sasl-mechanism=a",
+			exceptErr: "unknown a SASL mechanism",
+		},
+		{
+			name: "invalid GSSAPI auth type",
+			URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" +
+				"&sasl-mechanism=gssapi&sasl-gssapi-auth-type=keyta1b",
+			exceptErr: "unknown keyta1b auth type",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			cfg := NewConfig()
+			sinkURI, err := url.Parse(test.URI)
+			require.Nil(t, err)
+			if test.exceptErr == "" {
+				require.Nil(t, cfg.applySASL(sinkURI.Query()))
+			} else {
+				require.Regexp(t, test.exceptErr, cfg.applySASL(sinkURI.Query()).Error())
+			}
+		})
+	}
+}
+
+func TestCompleteSaramaSASLConfig(t *testing.T) {
+	t.Parallel()
+
+	// Test that SASL is turned on correctly.
+	cfg := NewConfig()
+	cfg.SASL = &security.SASL{
+		SASLUser:      "user",
+		SASLPassword:  "password",
+		SASLMechanism: "",
+		GSSAPI:        security.GSSAPI{},
+	}
+	saramaConfig := sarama.NewConfig()
+	completeSaramaSASLConfig(saramaConfig, cfg)
+	require.False(t, saramaConfig.Net.SASL.Enable)
+	cfg.SASL.SASLMechanism = "plain"
+	completeSaramaSASLConfig(saramaConfig, cfg)
+	require.True(t, saramaConfig.Net.SASL.Enable)
+	// Test that the SCRAMClientGeneratorFunc is set up correctly.
+	cfg = NewConfig()
+	cfg.SASL = &security.SASL{
+		SASLUser:      "user",
+		SASLPassword:  "password",
+		SASLMechanism: "plain",
+		GSSAPI:        security.GSSAPI{},
+	}
+	saramaConfig = sarama.NewConfig()
+	completeSaramaSASLConfig(saramaConfig, cfg)
+	require.Nil(t, saramaConfig.Net.SASL.SCRAMClientGeneratorFunc)
+	cfg.SASL.SASLMechanism = "SCRAM-SHA-512"
+	completeSaramaSASLConfig(saramaConfig, cfg)
+	require.NotNil(t, saramaConfig.Net.SASL.SCRAMClientGeneratorFunc)
+}
diff --git a/pkg/security/sasl_test.go b/pkg/security/sasl_test.go
new file mode 100644
index 000000000..71f3c9075
--- /dev/null
+++ b/pkg/security/sasl_test.go
@@ -0,0 +1,144 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package security
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSASLMechanismFromString(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name              string
+		s                 string
+		expectedMechanism string
+		expectErr         string
+	}{
+		{
+			name:              "random mechanism",
+			s:                 "random",
+			expectedMechanism: "",
+			expectErr:         "unknown random SASL mechanism",
+		},
+		{
+			name:              "lower case plain mechanism",
+			s:                 "plain",
+			expectedMechanism: "PLAIN",
+		},
+		{
+			name:              "upper case plain mechanism",
+			s:                 "PLAIN",
+			expectedMechanism: "PLAIN",
+		},
+		{
+			name:              "lower case scram-sha-256 mechanism",
+			s:                 "scram-sha-256",
+			expectedMechanism: "SCRAM-SHA-256",
+		},
+		{
+			name:              "upper case SCRAM-SHA-256 mechanism",
+			s:                 "SCRAM-SHA-256",
+			expectedMechanism: "SCRAM-SHA-256",
+		},
+		{
+			name:              "lower case scram-sha-512 mechanism",
+			s:                 "scram-sha-512",
+			expectedMechanism: "SCRAM-SHA-512",
+		},
+		{
+			name:              "upper case SCRAM-SHA-512 mechanism",
+			s:                 "SCRAM-SHA-512",
+			expectedMechanism: "SCRAM-SHA-512",
+		},
+		{
+			name:              "lower case gssapi mechanism",
+			s:                 "gssapi",
+			expectedMechanism: "GSSAPI",
+		},
+		{
+			name:              "upper case GSSAPI mechanism",
+			s:                 "GSSAPI",
+			expectedMechanism: "GSSAPI",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			mechanism, err := SASLMechanismFromString(test.s)
+			if test.expectErr != "" {
+				require.Error(t, err)
+				require.Regexp(t, test.expectErr, err.Error())
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.expectedMechanism, string(mechanism))
+			}
+		})
+	}
+}
+
+func TestAuthTypeFromString(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name         string
+		s            string
+		expectedType int
+		expectErr    string
+	}{
+		{
+			name:         "unknown",
+			s:            "a",
+			expectedType: 0,
+			expectErr:    "unknown a auth type",
+		},
+		{
+			name:         "lower case user",
+			s:            "user",
+			expectedType: 1,
+		},
+		{
+			name:         "upper case user",
+			s:            "USER",
+			expectedType: 1,
+		},
+		{
+			name:         "lower case keytab",
+			s:            "keytab",
+			expectedType: 2,
+		},
+		{
+			name:         "upper case keytab",
+			s:            "KEYTAB",
+			expectedType: 2,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			authType, err := AuthTypeFromString(test.s)
+			if test.expectErr != "" {
+				require.Error(t, err)
+				require.Regexp(t, test.expectErr, err.Error())
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.expectedType, int(authType))
+			}
+		})
+	}
+}