diff --git a/clients/pkg/promtail/scrapeconfig/scrapeconfig.go b/clients/pkg/promtail/scrapeconfig/scrapeconfig.go
index 2fc2bec148a59..007c9c94f1b2d 100644
--- a/clients/pkg/promtail/scrapeconfig/scrapeconfig.go
+++ b/clients/pkg/promtail/scrapeconfig/scrapeconfig.go
@@ -5,6 +5,9 @@ import (
 	"reflect"
 	"time"
 
+	"github.com/Shopify/sarama"
+	"github.com/grafana/dskit/flagext"
+
 	promconfig "github.com/prometheus/common/config"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/discovery"
@@ -244,6 +247,52 @@ type KafkaTargetConfig struct {
 
 	// Rebalancing strategy to use. (e.g sticky, roundrobin or range)
 	Assignor string `yaml:"assignor"`
+
+	// Authentication strategy with Kafka brokers
+	Authentication KafkaAuthentication `yaml:"authentication"`
+}
+
+// KafkaAuthenticationType specifies method to authenticate with Kafka brokers
+type KafkaAuthenticationType string
+
+const (
+	// KafkaAuthenticationTypeNone represents using no authentication
+	KafkaAuthenticationTypeNone = "none"
+	// KafkaAuthenticationTypeSSL represents using SSL/TLS to authenticate
+	KafkaAuthenticationTypeSSL = "ssl"
+	// KafkaAuthenticationTypeSASL represents using SASL to authenticate
+	KafkaAuthenticationTypeSASL = "sasl"
+)
+
+// KafkaAuthentication describe the configuration for authentication with Kafka brokers
+type KafkaAuthentication struct {
+	// Type is authentication type
+	// Possible values: none, sasl and ssl (defaults to none).
+	Type KafkaAuthenticationType `yaml:"type"`
+
+	// TLSConfig is used for TLS encryption and authentication with Kafka brokers
+	TLSConfig promconfig.TLSConfig `yaml:"tls_config,omitempty"`
+
+	// SASLConfig is used for SASL authentication with Kafka brokers
+	SASLConfig KafkaSASLConfig `yaml:"sasl_config,omitempty"`
+}
+
+// KafkaSASLConfig describe the SASL configuration for authentication with Kafka brokers
+type KafkaSASLConfig struct {
+	// SASL mechanism. Supports PLAIN, SCRAM-SHA-256 and SCRAM-SHA-512
+	Mechanism sarama.SASLMechanism `yaml:"mechanism"`
+
+	// SASL Username
+	User string `yaml:"user"`
+
+	// SASL Password for the User
+	Password flagext.Secret `yaml:"password"`
+
+	// UseTLS sets whether TLS is used with SASL
+	UseTLS bool `yaml:"use_tls"`
+
+	// TLSConfig is used for SASL over TLS. It is used only when UseTLS is true
+	TLSConfig promconfig.TLSConfig `yaml:",inline"`
 }
 
 // GcplogTargetConfig describes a scrape config to pull logs from any pubsub topic.
diff --git a/clients/pkg/promtail/targets/kafka/authentication.go b/clients/pkg/promtail/targets/kafka/authentication.go
new file mode 100644
index 0000000000000..a58d55589c629
--- /dev/null
+++ b/clients/pkg/promtail/targets/kafka/authentication.go
@@ -0,0 +1,69 @@
+package kafka
+
+import (
+	"crypto/sha256"
+	"crypto/sha512"
+	"crypto/tls"
+	"crypto/x509"
+	"os"
+
+	promconfig "github.com/prometheus/common/config"
+	"github.com/xdg-go/scram"
+)
+
+func createTLSConfig(cfg promconfig.TLSConfig) (*tls.Config, error) {
+	tc := &tls.Config{
+		InsecureSkipVerify: cfg.InsecureSkipVerify,
+		ServerName:         cfg.ServerName,
+	}
+	// load ca cert
+	if len(cfg.CAFile) > 0 {
+		caCert, err := os.ReadFile(cfg.CAFile)
+		if err != nil {
+			return nil, err
+		}
+		caCertPool := x509.NewCertPool()
+		caCertPool.AppendCertsFromPEM(caCert)
+		tc.RootCAs = caCertPool
+	}
+	// load client cert
+	if len(cfg.CertFile) > 0 && len(cfg.KeyFile) > 0 {
+		cert, err := tls.LoadX509KeyPair(cfg.CertFile, cfg.KeyFile)
+		if err != nil {
+			return nil, err
+		}
+		tc.Certificates = []tls.Certificate{cert}
+	}
+	return tc, nil
+}
+
+// copied from https://github.com/Shopify/sarama/blob/44627b731c60bb90efe25573e7ef2b3f8df3fa23/examples/sasl_scram_client/scram_client.go
+var (
+	SHA256 scram.HashGeneratorFcn = sha256.New
+	SHA512 scram.HashGeneratorFcn = sha512.New
+)
+
+// XDGSCRAMClient implements sarama.SCRAMClient
+type XDGSCRAMClient struct {
+	*scram.Client
+	*scram.ClientConversation
+	scram.HashGeneratorFcn
+}
+
+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
+}
+
+func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) {
+	response, err = x.ClientConversation.Step(challenge)
+	return
+}
+
+func (x *XDGSCRAMClient) Done() bool {
+	return x.ClientConversation.Done()
+}
diff --git a/clients/pkg/promtail/targets/kafka/target_syncer.go b/clients/pkg/promtail/targets/kafka/target_syncer.go
index 94c24b6eb2724..55c3f217035af 100644
--- a/clients/pkg/promtail/targets/kafka/target_syncer.go
+++ b/clients/pkg/promtail/targets/kafka/target_syncer.go
@@ -73,6 +73,10 @@ func NewSyncer(
 	default:
 		return nil, fmt.Errorf("unrecognized consumer group partition assignor: %s", cfg.KafkaConfig.Assignor)
 	}
+	config, err = withAuthentication(*config, cfg.KafkaConfig.Authentication)
+	if err != nil {
+		return nil, fmt.Errorf("error setting up kafka authentication: %w", err)
+	}
 	client, err := sarama.NewClient(cfg.KafkaConfig.Brokers, config)
 	if err != nil {
 		return nil, fmt.Errorf("error creating kafka client: %w", err)
@@ -113,6 +117,74 @@ func NewSyncer(
 	return t, nil
 }
 
+func withAuthentication(cfg sarama.Config, authCfg scrapeconfig.KafkaAuthentication) (*sarama.Config, error) {
+	if len(authCfg.Type) == 0 || authCfg.Type == scrapeconfig.KafkaAuthenticationTypeNone {
+		return &cfg, nil
+	}
+
+	switch authCfg.Type {
+	case scrapeconfig.KafkaAuthenticationTypeSSL:
+		return withSSLAuthentication(cfg, authCfg)
+	case scrapeconfig.KafkaAuthenticationTypeSASL:
+		return withSASLAuthentication(cfg, authCfg)
+	default:
+		return nil, fmt.Errorf("unsupported authentication type %s", authCfg.Type)
+	}
+}
+
+func withSSLAuthentication(cfg sarama.Config, authCfg scrapeconfig.KafkaAuthentication) (*sarama.Config, error) {
+	cfg.Net.TLS.Enable = true
+	tc, err := createTLSConfig(authCfg.TLSConfig)
+	if err != nil {
+		return nil, err
+	}
+	cfg.Net.TLS.Config = tc
+	return &cfg, nil
+}
+
+func withSASLAuthentication(cfg sarama.Config, authCfg scrapeconfig.KafkaAuthentication) (*sarama.Config, error) {
+	cfg.Net.SASL.Enable = true
+	cfg.Net.SASL.User = authCfg.SASLConfig.User
+	cfg.Net.SASL.Password = authCfg.SASLConfig.Password.Value
+	cfg.Net.SASL.Mechanism = authCfg.SASLConfig.Mechanism
+	if cfg.Net.SASL.Mechanism == "" {
+		cfg.Net.SASL.Mechanism = sarama.SASLTypePlaintext
+	}
+
+	supportedMechanism := []string{
+		sarama.SASLTypeSCRAMSHA512,
+		sarama.SASLTypeSCRAMSHA256,
+		sarama.SASLTypePlaintext,
+	}
+	if !util.StringSliceContains(supportedMechanism, string(authCfg.SASLConfig.Mechanism)) {
+		return nil, fmt.Errorf("error unsupported sasl mechanism: %s", authCfg.SASLConfig.Mechanism)
+	}
+
+	if cfg.Net.SASL.Mechanism == sarama.SASLTypeSCRAMSHA512 {
+		cfg.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+			return &XDGSCRAMClient{
+				HashGeneratorFcn: SHA512,
+			}
+		}
+	}
+	if cfg.Net.SASL.Mechanism == sarama.SASLTypeSCRAMSHA256 {
+		cfg.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+			return &XDGSCRAMClient{
+				HashGeneratorFcn: SHA256,
+			}
+		}
+	}
+	if authCfg.SASLConfig.UseTLS {
+		tc, err := createTLSConfig(authCfg.SASLConfig.TLSConfig)
+		if err != nil {
+			return nil, err
+		}
+		cfg.Net.TLS.Config = tc
+		cfg.Net.TLS.Enable = true
+	}
+	return &cfg, nil
+}
+
 func (ts *TargetSyncer) loop() {
 	topicChanged := make(chan []string)
 	ts.wg.Add(2)
diff --git a/clients/pkg/promtail/targets/kafka/target_syncer_test.go b/clients/pkg/promtail/targets/kafka/target_syncer_test.go
index 8099476eaa164..83f6dfb1a9133 100644
--- a/clients/pkg/promtail/targets/kafka/target_syncer_test.go
+++ b/clients/pkg/promtail/targets/kafka/target_syncer_test.go
@@ -6,6 +6,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/grafana/dskit/flagext"
+	"github.com/prometheus/common/config"
+
 	"github.com/Shopify/sarama"
 	"github.com/go-kit/log"
 	"github.com/grafana/loki/clients/pkg/promtail/client/fake"
@@ -208,3 +211,144 @@ func Test_validateConfig(t *testing.T) {
 		})
 	}
 }
+
+func Test_withAuthentication(t *testing.T) {
+	var (
+		tlsConf = config.TLSConfig{
+			CAFile:             "testdata/example.com.ca.pem",
+			CertFile:           "testdata/example.com.pem",
+			KeyFile:            "testdata/example.com-key.pem",
+			ServerName:         "example.com",
+			InsecureSkipVerify: true,
+		}
+		expectedTLSConf, _ = createTLSConfig(config.TLSConfig{
+			CAFile:             "testdata/example.com.ca.pem",
+			CertFile:           "testdata/example.com.pem",
+			KeyFile:            "testdata/example.com-key.pem",
+			ServerName:         "example.com",
+			InsecureSkipVerify: true,
+		})
+		cfg = sarama.NewConfig()
+	)
+
+	// no authentication
+	noAuthCfg, err := withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeNone,
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, false, noAuthCfg.Net.TLS.Enable)
+	assert.Equal(t, false, noAuthCfg.Net.SASL.Enable)
+	assert.NoError(t, noAuthCfg.Validate())
+
+	// specify unsupported auth type
+	illegalAuthTypeCfg, err := withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: "illegal",
+	})
+	assert.NotNil(t, err)
+	assert.Nil(t, illegalAuthTypeCfg)
+
+	// mTLS authentication
+	mTLSCfg, err := withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type:      scrapeconfig.KafkaAuthenticationTypeSSL,
+		TLSConfig: tlsConf,
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, true, mTLSCfg.Net.TLS.Enable)
+	assert.NotNil(t, mTLSCfg.Net.TLS.Config)
+	assert.Equal(t, "example.com", mTLSCfg.Net.TLS.Config.ServerName)
+	assert.Equal(t, true, mTLSCfg.Net.TLS.Config.InsecureSkipVerify)
+	assert.Equal(t, expectedTLSConf.Certificates, mTLSCfg.Net.TLS.Config.Certificates)
+	assert.NotNil(t, mTLSCfg.Net.TLS.Config.RootCAs)
+	assert.NoError(t, mTLSCfg.Validate())
+
+	// mTLS authentication expect ignore sasl
+	mTLSCfg, err = withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type:      scrapeconfig.KafkaAuthenticationTypeSSL,
+		TLSConfig: tlsConf,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypeSCRAMSHA256,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+			UseTLS: false,
+		},
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, false, mTLSCfg.Net.SASL.Enable)
+
+	// SASL/PLAIN
+	saslCfg, err := withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeSASL,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypePlaintext,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+		},
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, false, saslCfg.Net.TLS.Enable)
+	assert.Equal(t, true, saslCfg.Net.SASL.Enable)
+	assert.Equal(t, "user", saslCfg.Net.SASL.User)
+	assert.Equal(t, "pass", saslCfg.Net.SASL.Password)
+	assert.Equal(t, sarama.SASLTypePlaintext, string(saslCfg.Net.SASL.Mechanism))
+	assert.NoError(t, saslCfg.Validate())
+
+	// SASL/SCRAM
+	saslCfg, err = withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeSASL,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypeSCRAMSHA512,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+		},
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, false, saslCfg.Net.TLS.Enable)
+	assert.Equal(t, true, saslCfg.Net.SASL.Enable)
+	assert.Equal(t, "user", saslCfg.Net.SASL.User)
+	assert.Equal(t, "pass", saslCfg.Net.SASL.Password)
+	assert.Equal(t, sarama.SASLTypeSCRAMSHA512, string(saslCfg.Net.SASL.Mechanism))
+	assert.NoError(t, saslCfg.Validate())
+
+	// SASL unsupported mechanism
+	_, err = withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeSASL,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypeGSSAPI,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+		},
+	})
+	assert.Error(t, err)
+	assert.Equal(t, err.Error(), "error unsupported sasl mechanism: GSSAPI")
+
+	// SASL over TLS
+	saslCfg, err = withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeSASL,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypeSCRAMSHA512,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+			UseTLS:    true,
+			TLSConfig: tlsConf,
+		},
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, true, saslCfg.Net.TLS.Enable)
+	assert.Equal(t, true, saslCfg.Net.SASL.Enable)
+	assert.NotNil(t, saslCfg.Net.TLS.Config)
+	assert.Equal(t, "example.com", saslCfg.Net.TLS.Config.ServerName)
+	assert.Equal(t, true, saslCfg.Net.TLS.Config.InsecureSkipVerify)
+	assert.Equal(t, expectedTLSConf.Certificates, saslCfg.Net.TLS.Config.Certificates)
+	assert.NotNil(t, saslCfg.Net.TLS.Config.RootCAs)
+	assert.NoError(t, saslCfg.Validate())
+}
diff --git a/clients/pkg/promtail/targets/kafka/testdata/example.com-key.pem b/clients/pkg/promtail/targets/kafka/testdata/example.com-key.pem
new file mode 100644
index 0000000000000..cef0fa6cc5467
--- /dev/null
+++ b/clients/pkg/promtail/targets/kafka/testdata/example.com-key.pem
@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEogIBAAKCAQEAp7T84Nzd5y54a0qEFNeCXIDoz4LdfQqB2MloSl6aicKFRX7Y
+rTPDgwul9trau0dtDi7goT7GoyesRRJBsAqmUEaX9YjgRbWCiiKr47SxQJRn0Hom
+sNgCfIgCJ+Ej8no87JaxM7gklqoXrWK2KeG+NoC7jO2hrlBomkTkjgjU466kYpXM
+8z2+p3Jt3JHNJYTnOl36s/Iaw0ugmEPOJpLrsdnt2Irs5J2dmbMbeDJ5+yaE2c8B
+YSzAnLh5GQZO+5G7Ez0SXs1VJ2DALXybLuLD9pEZpAt40TK9jNwU5dfTYZ6yEeyn
+jw6qqJMcxVEYRru7x4ArOP7Uvs0qlOLDxJdD+QIDAQABAoIBAFWbj9KBLE0Mba/n
+E9FHyWXK8Aytcr6XlHzDIxeDf0N/JmS5QYX4fH7yfT+rrCgZZ9PrngLxdphmcgu4
+LAvfA9LKlltiCYnMA9zbof7Uh/69Qtkq4YE4YtyK2P7ecGkgeOUUb1RFVXgLT5bU
+YsSyVVShFhv0WaoPpbXKIRlX7MRj188laiYQnhLJ8gZeutk9A3wkaPsRp2s1ME9R
+oyUDs9OAEJOXB7wEqGLmKXqxSqzTDSoC/7wbjQut8G63/ri+EYwp5ukRP886cbep
+hVUDt8bzoratw+IFsGt+fBe9MGNbAMvva7x0yb/7gdEf505+c1h3Q19CLbPHFIJv
+hEcbhAECgYEAwkZp26SQANi7zDVoGqS7PXIWs2TL2g9Y2vO5g0MTvbno/0OPo3+e
+GOtFW4rK2k6SoK3RtdOdmobJcxKf2G4W7ncdM6X1Rgxfgqyt58qfXSL/TlS3wfH2
+vDFQDyLEofZ64vSMEKinSkv5kY/FCCtw8q8hCgiqmpwY51PFlI/52bkCgYEA3P2h
+LmRYGImkw8eiKm4eAFYXdVlxzsytCF3s2cCI3GavoGyMwxayYUkEKRA7bcup1pK+
+RdIRaaVkNcFICrX4N9zaEqSkBEQkQ+ZQN0hW9HM3XKPR7VPQEusRCbV5hMNgI3tY
+irAZXgeOzS61Rl+o3Ta3SQ2jDcCoK31xzE1s3EECgYAYYb/tWfTctlazZUyAc4Yw
+Sv5AW3keD+kF6aqxp5x1pjxwtOj1CxIrbHOS7pNQ3KWYVthH6pwQBbSIpaC8B+0G
+1poqnjxvIyRlgQh+W7aueLL0ALvjMlvV+JZkn+dvsEBx9WESwifksi5LL3D5+oG9
+Y29RFA9dQhP6DFByubMQuQKBgAyIogykik6R9/NWrj7j0fXI7DmuogLNnv67fQR4
+pAqEFG/v2Cf0cJeN8Zt2nThD9dUCq6IAIRax17YoyTI6UeKxNvkZt2e6iagENwZ7
+ptrkcf5iGDTyrPl1tZisX0EFZ717cHElPbsUiKfgf02HfWdWhByzlkzgYWleCwdA
+WO1BAoGAMJA/dYgbEachKXl9nZ9HvqAtc30ioYG0eq5Zx9iYZWqhb1vqGJXS2M2t
+Onmw/+2pnvPIM1uP66G5VG0rcVL7jdJ0pibCgw+L9IHpDwCAuVGn1Cuzcty+NiAy
+qLkQAgPOMjwL9zJYrn6LAOVchnIbpuoaGNG14BdZWeh25kozk/k=
+-----END RSA PRIVATE KEY-----
diff --git a/clients/pkg/promtail/targets/kafka/testdata/example.com.ca.pem b/clients/pkg/promtail/targets/kafka/testdata/example.com.ca.pem
new file mode 100644
index 0000000000000..2880d2c27de9e
--- /dev/null
+++ b/clients/pkg/promtail/targets/kafka/testdata/example.com.ca.pem
@@ -0,0 +1,25 @@
+-----BEGIN CERTIFICATE-----
+MIIEMjCCAxqgAwIBAgIUDpfGq/7zYQRROCP65MxUy+LYewkwDQYJKoZIhvcNAQEL
+BQAwgbAxGDAWBgNVBAYTD0V4YW1wbGUgQ291bnRyeTEWMBQGA1UECBMNRXhhbXBs
+ZSBTdGF0ZTEZMBcGA1UEBxMQRXhhbXBsZSBMb2NhbGl0eTEdMBsGA1UEChMURXhh
+bXBsZSBPcmdhbml6YXRpb24xIjAgBgNVBAsTGUV4YW1wbGUgT3JnYW5pemF0aW9u
+IFVuaXQxHjAcBgNVBAMTFVNhbXBsZSBTZWxmLVNpZ25lZCBDQTAeFw0yMDEwMjMx
+NTUxMDBaFw0yNTEwMjIxNTUxMDBaMIGwMRgwFgYDVQQGEw9FeGFtcGxlIENvdW50
+cnkxFjAUBgNVBAgTDUV4YW1wbGUgU3RhdGUxGTAXBgNVBAcTEEV4YW1wbGUgTG9j
+YWxpdHkxHTAbBgNVBAoTFEV4YW1wbGUgT3JnYW5pemF0aW9uMSIwIAYDVQQLExlF
+eGFtcGxlIE9yZ2FuaXphdGlvbiBVbml0MR4wHAYDVQQDExVTYW1wbGUgU2VsZi1T
+aWduZWQgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCXiANIcYCd
+AaniIG59ZEIhgT2tx4PpsjPc07sx4B8cPAhYVPlpom2lSfI9lj14i37vEBFEsuwD
+Dk7xz6MGIAsLbmw8eEAbUyMaKeCfUZDVwWWW/OQo/riCc1Vw85nKAUQ7YA4TyvpO
+ORf5UAhLx0/ZhJsoAOgEOGa+S7NZjNCwADvhOzE0j5oDCblv3+EeiB8zAYzAG1xm
+onkVsS09mfmiE2V5mA6zAn60E9Ssfx4hJbxFNAvTlv+im37uumipKEGr/gRCcnFp
+QcgamxjCbmD+XNjJ35u0/r/mXHeghvRNl+2ARl3XngKclFeNwhgm7DWJsx66bbGy
+Hv2YQb9xxTNbAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNVHRMBAf8EBTAD
+AQH/MB0GA1UdDgQWBBSuamXDGCosE/xvsAWBVbSshGQ65DANBgkqhkiG9w0BAQsF
+AAOCAQEAd21HOCRsQENKxOsMbsYyla1lyWPjTTnn+c4IbgfcKA7lkf8ESFa7ChVq
+Q2z8mTAxblvnGy1bHaeFw1vy0hIYnV1rizsb+3nBN5oBZQtG6Rmc9iL5MhIaHprB
+bHqx/9zuCwH2jzSMcIYGUbfJcC5+W67P/zpX5rKkCqiyu81Unw5GAmcawaU6600b
+Dtx1YEWWLjwnBXXQp+4udHHCChsq5SFwJuWZ13+KNIrrD0QqcXn4lGtEWLD7/Anl
+BFGcQJpHq5x17Of1pxOqEogXTk44+cMKNDRr05mH2xc6nZqvZlUkdusl34XLC9Lo
+kIim0bc2p5dHjfAeBS7HSkXuwYzeQA==
+-----END CERTIFICATE-----
diff --git a/clients/pkg/promtail/targets/kafka/testdata/example.com.pem b/clients/pkg/promtail/targets/kafka/testdata/example.com.pem
new file mode 100644
index 0000000000000..6fee105b19de4
--- /dev/null
+++ b/clients/pkg/promtail/targets/kafka/testdata/example.com.pem
@@ -0,0 +1,27 @@
+-----BEGIN CERTIFICATE-----
+MIIEfzCCA2egAwIBAgIUJr0Q3zi9e99mQtYS9nK2jniXnZkwDQYJKoZIhvcNAQEL
+BQAwgbAxGDAWBgNVBAYTD0V4YW1wbGUgQ291bnRyeTEWMBQGA1UECBMNRXhhbXBs
+ZSBTdGF0ZTEZMBcGA1UEBxMQRXhhbXBsZSBMb2NhbGl0eTEdMBsGA1UEChMURXhh
+bXBsZSBPcmdhbml6YXRpb24xIjAgBgNVBAsTGUV4YW1wbGUgT3JnYW5pemF0aW9u
+IFVuaXQxHjAcBgNVBAMTFVNhbXBsZSBTZWxmLVNpZ25lZCBDQTAeFw0yMDEwMjMx
+NTUyMDBaFw0yMTEwMjMxNTUyMDBaMIGmMRgwFgYDVQQGEw9FeGFtcGxlIENvdW50
+cnkxFjAUBgNVBAgTDUV4YW1wbGUgU3RhdGUxGTAXBgNVBAcTEEV4YW1wbGUgTG9j
+YWxpdHkxHTAbBgNVBAoTFEV4YW1wbGUgT3JnYW5pemF0aW9uMSIwIAYDVQQLExlF
+eGFtcGxlIE9yZ2FuaXphdGlvbiBVbml0MRQwEgYDVQQDEwtleGFtcGxlLmNvbTCC
+ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAKe0/ODc3ecueGtKhBTXglyA
+6M+C3X0KgdjJaEpemonChUV+2K0zw4MLpfba2rtHbQ4u4KE+xqMnrEUSQbAKplBG
+l/WI4EW1gooiq+O0sUCUZ9B6JrDYAnyIAifhI/J6POyWsTO4JJaqF61itinhvjaA
+u4ztoa5QaJpE5I4I1OOupGKVzPM9vqdybdyRzSWE5zpd+rPyGsNLoJhDziaS67HZ
+7diK7OSdnZmzG3gyefsmhNnPAWEswJy4eRkGTvuRuxM9El7NVSdgwC18my7iw/aR
+GaQLeNEyvYzcFOXX02GeshHsp48OqqiTHMVRGEa7u8eAKzj+1L7NKpTiw8SXQ/kC
+AwEAAaOBmDCBlTAOBgNVHQ8BAf8EBAMCBaAwHQYDVR0lBBYwFAYIKwYBBQUHAwEG
+CCsGAQUFBwMCMAwGA1UdEwEB/wQCMAAwHQYDVR0OBBYEFNfuWV6Y129JoLnf12l+
+/OHgSyXEMB8GA1UdIwQYMBaAFK5qZcMYKiwT/G+wBYFVtKyEZDrkMBYGA1UdEQQP
+MA2CC2V4YW1wbGUuY29tMA0GCSqGSIb3DQEBCwUAA4IBAQBoIy1KhsjfHMwO1yAZ
+ECiyoPr7cfqPFSteAEk7CxqgdCW2ZsZJnKHdZYFhBfc7cEGTHss9gANEgOogzgEV
+ZOny0fINK3+GVuDlvm9DEB/r5sXm9zWoS5qvbtZ58lUlgOXDSlQ5gsMxOSFl7Gp5
+GAKxuGIvF8bFejQP8u1f5YBxSPgHhg+vzyHJ2vmOw4r+dxpMG5B/NDpbOxQNpNmw
+GTOETPbQh01Y/D6jI4HGZ90c6C9dP0+Tc1PScfaE3uOqa3GEToahXzLDLCW4sQFS
+PYma10nXk29DfjYtaYYFMmpZnemhZG8E61OXe7fLn5/DLt1a4lZRfoEoiGAaKqQd
+glLB
+-----END CERTIFICATE-----