diff --git a/kafka/config.go b/kafka/config.go
index 306d2eb7..ce66b164 100644
--- a/kafka/config.go
+++ b/kafka/config.go
@@ -19,26 +19,28 @@ import (
 )
 
 type Config struct {
-	BootstrapServers        *[]string
-	Timeout                 int
-	CACert                  string
-	ClientCert              string
-	ClientCertKey           string
-	ClientCertKeyPassphrase string
-	KafkaVersion            string
-	TLSEnabled              bool
-	SkipTLSVerify           bool
-	SASLUsername            string
-	SASLPassword            string
-	SASLMechanism           string
-	SASLAWSRegion           string
-	SASLAWSRoleArn          string
-	SASLAWSProfile          string
-	SASLAWSAccessKey        string
-	SASLAWSSecretKey        string
-	SASLAWSToken            string
-	SASLAWSCredsDebug       bool
-	SASLTokenUrl            string
+	BootstrapServers            *[]string
+	Timeout                     int
+	CACert                      string
+	ClientCert                  string
+	ClientCertKey               string
+	ClientCertKeyPassphrase     string
+	KafkaVersion                string
+	TLSEnabled                  bool
+	SkipTLSVerify               bool
+	SASLUsername                string
+	SASLPassword                string
+	SASLMechanism               string
+	SASLAWSRegion               string
+	SASLAWSRoleArn              string
+	SASLAWSWebIdentityToken     string
+	SASLAWSWebIdentityTokenFile string
+	SASLAWSProfile              string
+	SASLAWSAccessKey            string
+	SASLAWSSecretKey            string
+	SASLAWSToken                string
+	SASLAWSCredsDebug           bool
+	SASLTokenUrl                string
 }
 
 type OAuth2Config interface {
@@ -84,8 +86,19 @@ func (o *oauthbearerTokenProvider) Token() (*sarama.AccessToken, error) {
 func (c *Config) Token() (*sarama.AccessToken, error) {
 	signer.AwsDebugCreds = c.SASLAWSCredsDebug
 	var token string
+	var webIdentityTokenBuffer []byte
 	var err error
-	if c.SASLAWSRoleArn != "" {
+	if c.SASLAWSRoleArn != "" && (c.SASLAWSWebIdentityToken != "" || c.SASLAWSWebIdentityTokenFile != "") {
+		log.Printf("[INFO] Generating auth token with a web identity role '%s' in '%s'", c.SASLAWSRoleArn, c.SASLAWSRegion)
+		if c.SASLAWSWebIdentityTokenFile != "" {
+			webIdentityTokenBuffer, err = os.ReadFile(c.SASLAWSWebIdentityTokenFile)
+			if err != nil {
+				return nil, err
+			}
+			c.SASLAWSWebIdentityToken = string(webIdentityTokenBuffer)
+		}
+		token, _, err = signer.GenerateAuthTokenFromWebIdentity(context.TODO(), c.SASLAWSRegion, c.SASLAWSRoleArn, c.SASLAWSWebIdentityToken, "terraform-kafka-provider")
+	} else if c.SASLAWSRoleArn != "" {
 		log.Printf("[INFO] Generating auth token with a role '%s' in '%s'", c.SASLAWSRoleArn, c.SASLAWSRegion)
 		token, _, err = signer.GenerateAuthTokenFromRole(context.TODO(), c.SASLAWSRegion, c.SASLAWSRoleArn, "terraform-kafka-provider")
 	} else if c.SASLAWSProfile != "" {
@@ -305,6 +318,8 @@ func (config *Config) copyWithMaskedSensitiveValues() Config {
 		config.SASLMechanism,
 		config.SASLAWSRegion,
 		config.SASLAWSRoleArn,
+		"*****",
+		config.SASLAWSWebIdentityTokenFile,
 		config.SASLAWSProfile,
 		config.SASLAWSAccessKey,
 		"*****",
diff --git a/kafka/provider.go b/kafka/provider.go
index 11236278..9126fad7 100644
--- a/kafka/provider.go
+++ b/kafka/provider.go
@@ -79,6 +79,18 @@ func Provider() *schema.Provider {
 				DefaultFunc: schema.EnvDefaultFunc("AWS_ROLE_ARN", nil),
 				Description: "Arn of an AWS IAM role to assume",
 			},
+			"sasl_aws_web_identity_token": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				Default:     "",
+				Description: "Arn of an AWS IAM role to assume",
+			},
+			"sasl_aws_web_identity_token_file": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("AWS_WEB_IDENTITY_TOKEN_FILE", nil),
+				Description: "Arn of an AWS IAM role to assume",
+			},
 			"sasl_aws_profile": {
 				Type:        schema.TypeString,
 				Optional:    true,
@@ -179,26 +191,28 @@ func providerConfigure(d *schema.ResourceData) (interface{}, error) {
 	}
 
 	config := &Config{
-		BootstrapServers:        brokers,
-		CACert:                  d.Get("ca_cert").(string),
-		ClientCert:              d.Get("client_cert").(string),
-		ClientCertKey:           d.Get("client_key").(string),
-		ClientCertKeyPassphrase: d.Get("client_key_passphrase").(string),
-		KafkaVersion:            d.Get("kafka_version").(string),
-		SkipTLSVerify:           d.Get("skip_tls_verify").(bool),
-		SASLAWSRegion:           d.Get("sasl_aws_region").(string),
-		SASLUsername:            d.Get("sasl_username").(string),
-		SASLPassword:            d.Get("sasl_password").(string),
-		SASLTokenUrl:            d.Get("sasl_token_url").(string),
-		SASLAWSRoleArn:          d.Get("sasl_aws_role_arn").(string),
-		SASLAWSProfile:          d.Get("sasl_aws_profile").(string),
-		SASLAWSAccessKey:        d.Get("sasl_aws_access_key").(string),
-		SASLAWSSecretKey:        d.Get("sasl_aws_secret_key").(string),
-		SASLAWSToken:            d.Get("sasl_aws_token").(string),
-		SASLAWSCredsDebug:       d.Get("sasl_aws_creds_debug").(bool),
-		SASLMechanism:           saslMechanism,
-		TLSEnabled:              d.Get("tls_enabled").(bool),
-		Timeout:                 d.Get("timeout").(int),
+		BootstrapServers:            brokers,
+		CACert:                      d.Get("ca_cert").(string),
+		ClientCert:                  d.Get("client_cert").(string),
+		ClientCertKey:               d.Get("client_key").(string),
+		ClientCertKeyPassphrase:     d.Get("client_key_passphrase").(string),
+		KafkaVersion:                d.Get("kafka_version").(string),
+		SkipTLSVerify:               d.Get("skip_tls_verify").(bool),
+		SASLAWSRegion:               d.Get("sasl_aws_region").(string),
+		SASLUsername:                d.Get("sasl_username").(string),
+		SASLPassword:                d.Get("sasl_password").(string),
+		SASLTokenUrl:                d.Get("sasl_token_url").(string),
+		SASLAWSRoleArn:              d.Get("sasl_aws_role_arn").(string),
+		SASLAWSWebIdentityToken:     d.Get("sasl_aws_web_identity_token").(string),
+		SASLAWSWebIdentityTokenFile: d.Get("sasl_aws_web_identity_token_file").(string),
+		SASLAWSProfile:              d.Get("sasl_aws_profile").(string),
+		SASLAWSAccessKey:            d.Get("sasl_aws_access_key").(string),
+		SASLAWSSecretKey:            d.Get("sasl_aws_secret_key").(string),
+		SASLAWSToken:                d.Get("sasl_aws_token").(string),
+		SASLAWSCredsDebug:           d.Get("sasl_aws_creds_debug").(bool),
+		SASLMechanism:               saslMechanism,
+		TLSEnabled:                  d.Get("tls_enabled").(bool),
+		Timeout:                     d.Get("timeout").(int),
 	}
 
 	if config.CACert == "" {