diff --git a/go.mod b/go.mod
index 3ca7ebe..f192c99 100644
--- a/go.mod
+++ b/go.mod
@@ -25,7 +25,11 @@ require (
 	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.1.2 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
 	golang.org/x/crypto v0.38.0 // indirect
 	golang.org/x/net v0.40.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect
+	golang.org/x/text v0.25.0 // indirect
 )
diff --git a/go.sum b/go.sum
index 2449463..0ec569e 100644
--- a/go.sum
+++ b/go.sum
@@ -55,6 +55,12 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
@@ -88,7 +94,10 @@ golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4=
+golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
diff --git a/kafka/kafka.go b/kafka/kafka.go
index b3c5c89..5baf446 100644
--- a/kafka/kafka.go
+++ b/kafka/kafka.go
@@ -15,6 +15,8 @@
 package kafka
 
 import (
+	"crypto/sha256"
+	"crypto/sha512"
 	"crypto/tls"
 	"fmt"
 	"time"
@@ -23,6 +25,7 @@ import (
 	"git.zabbix.com/ap/plugin-support/log"
 	"git.zabbix.com/ap/plugin-support/tlsconfig"
 	"github.com/IBM/sarama"
+	"github.com/xdg-go/scram"
 )
 
 const (
@@ -60,6 +63,35 @@ type Configuration struct {
 	Timeout        int    `conf:"default=1"`
 	TLSAuth        bool   `conf:"default=false"`
 	EnableTLS      bool   `conf:"optional"`
+	SASLMechanism  string `conf:"optional"`
+}
+
+// XDGSCRAMClient defines SCRAM client
+type XDGSCRAMClient struct {
+	*scram.Client
+	*scram.ClientConversation
+	scram.HashGeneratorFcn
+}
+
+// Begin creates the SCRAM client
+func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) {
+	x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID)
+	if err != nil {
+		return err
+	}
+	x.ClientConversation = x.Client.NewConversation()
+	return nil
+}
+
+// Step is the adapter with the step for the SCRAM client
+func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) {
+	response, err = x.ClientConversation.Step(challenge)
+	return
+}
+
+// Done ends the connection for the SCRAM client
+func (x *XDGSCRAMClient) Done() bool {
+	return x.ClientConversation.Done()
 }
 
 // ProduceItem produces Kafka message to the item topic
@@ -120,6 +152,7 @@ func NewProducer(c *Configuration) (*DefaultProducer, error) {
 		c.Retry,
 		c.TLSAuth,
 		c.EnableTLS,
+		c.SASLMechanism,
 		time.Duration(c.Timeout)*time.Second,
 		time.Duration(c.KeepAlive)*time.Second,
 		tlsConfig,
@@ -186,6 +219,7 @@ func newConfig(
 	retries int,
 	tlsAuth,
 	enableTLS bool,
+	saslMechanism string,
 	timeout,
 	keepAlive time.Duration,
 	tlsConf *tls.Config,
@@ -209,6 +243,18 @@ func newConfig(
 		config.Net.SASL.Enable = true
 		config.Net.SASL.User = username
 		config.Net.SASL.Password = password
+		switch saslMechanism {
+		case "scram-sha-512":
+			config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
+			config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+				return &XDGSCRAMClient{HashGeneratorFcn: sha512.New}
+			}
+		case "scram-sha-256":
+			config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA256
+			config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+				return &XDGSCRAMClient{HashGeneratorFcn: sha256.New}
+			}
+		}
 	}
 
 	if tlsAuth {
diff --git a/kafka/kafka_test.go b/kafka/kafka_test.go
index 931ce4a..259ab0a 100644
--- a/kafka/kafka_test.go
+++ b/kafka/kafka_test.go
@@ -25,14 +25,15 @@ func Test_newConfig(t *testing.T) {
 	t.Parallel()
 
 	type args struct {
-		username  string
-		password  string
-		retries   int
-		tlsAuth   bool
-		enableTLS bool
-		timeout   time.Duration
-		keepAlive time.Duration
-		tlsConf   *tls.Config
+		username      string
+		password      string
+		retries       int
+		tlsAuth       bool
+		enableTLS     bool
+		saslMechanism string
+		timeout       time.Duration
+		keepAlive     time.Duration
+		tlsConf       *tls.Config
 	}
 
 	tests := []struct {
@@ -60,6 +61,7 @@ func Test_newConfig(t *testing.T) {
 				2,
 				false,
 				false,
+				"",
 				3,
 				30,
 				nil,
@@ -86,6 +88,7 @@ func Test_newConfig(t *testing.T) {
 				2,
 				false,
 				false,
+				"",
 				3,
 				30,
 				nil,
@@ -112,6 +115,7 @@ func Test_newConfig(t *testing.T) {
 				2,
 				true,
 				false,
+				"",
 				3,
 				30,
 				&tls.Config{ServerName: "127.0.0.1"}, //nolint:gosec // struct fields are not used
@@ -138,6 +142,7 @@ func Test_newConfig(t *testing.T) {
 				2,
 				true,
 				true,
+				"",
 				3,
 				30,
 				&tls.Config{ServerName: "127.0.0.1"}, //nolint:gosec // struct fields are not used
@@ -169,6 +174,7 @@ func Test_newConfig(t *testing.T) {
 				tt.args.retries,
 				tt.args.tlsAuth,
 				tt.args.enableTLS,
+				tt.args.saslMechanism,
 				tt.args.timeout,
 				tt.args.keepAlive,
 				tt.args.tlsConf,
diff --git a/kafka_connector.conf b/kafka_connector.conf
index 72029d4..e2fd591 100644
--- a/kafka_connector.conf
+++ b/kafka_connector.conf
@@ -168,6 +168,16 @@ Kafka.URL=localhost
 # Default:
 # Kafka.Password=
 
+### Option: Kafka.SASLMechanism
+#	SASL mechanism for SASL authorization.
+#	If set to empty string, SASL_PLAIN is used for SASL authorization
+#	scram-sha-256
+#	scram-sha-512
+#
+# Mandatory: no
+# Default:
+# Kafka.SASLMechanism=
+
 ### Option: Kafka.EnableTLS
 #	Enables kafka TLS authorization.
 #
