From a92d412d605f37b8385e50f7a83265396bea0bef Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Thu, 27 Jun 2024 11:43:53 +0900
Subject: [PATCH 01/20] Add expandSecretsManagerConfiguration

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 internal/service/firehose/delivery_stream.go | 26 ++++++++++++++++++++
 1 file changed, 26 insertions(+)

diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index 02c293f7c25..defe1bb1001 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -2308,6 +2308,32 @@ func expandProcessorParameter(processorParameter map[string]interface{}) types.P
 	return parameter
 }
 
+func expandSecretsManagerConfiguration(tfMap map[string]interface{}) *types.SecretsManagerConfiguration {
+	config := tfMap["secrets_manager_configuration"].([]interface{})
+	if len(config) == 0 || config[0] == nil {
+		// It is possible to just pass nil here, but this seems to be the
+		// canonical form that AWS uses, and is less likely to produce diffs.
+		return &types.SecretsManagerConfiguration{
+			Enabled:    aws.Bool(false),
+		}
+	}
+
+  secretsManagerConfiguration := config[0].(map[string]interface{})
+  configuration := &types.SecretsManagerConfiguration{
+    Enabled:    aws.Bool(secretsManagerConfiguration[names.AttrEnabled].(bool)),
+  }
+
+  if v, ok := secretsManagerConfiguration["secret_arn"]; ok {
+    configuration.SecretARN = aws.String(v.(string))
+  }
+
+  if v, ok := secretsManagerConfiguration[names.AttrRoleARN]; ok {
+    configuration.RoleARN = aws.String(v.(string))
+  }
+
+  return configuration
+}
+
 func expandEncryptionConfiguration(s3 map[string]interface{}) *types.EncryptionConfiguration {
 	if key, ok := s3[names.AttrKMSKeyARN]; ok && len(key.(string)) > 0 {
 		return &types.EncryptionConfiguration{

From 53c2149fa3bc70c8e0e7206afd63dd621386fbf8 Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Thu, 27 Jun 2024 11:54:47 +0900
Subject: [PATCH 02/20] Add flattenSecretsManagerConfiguration

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 internal/service/firehose/delivery_stream.go | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index defe1bb1001..b906587fed0 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -3802,6 +3802,23 @@ func flattenProcessingConfiguration(pc *types.ProcessingConfiguration, destinati
 	return processingConfiguration
 }
 
+func flattenSecretsManagerConfiguration(smc *types.SecretsManagerConfiguration) []interface{} {
+	if smc == nil {
+		return []interface{}{}
+	}
+
+	secretsManagerConfiguration := map[string]interface{}{
+		names.AttrEnabled: aws.ToBool(smc.Enabled),
+	}
+	if aws.ToBool(smc.Enabled) {
+		secretsManagerConfiguration["secret_arn"] = aws.ToString(smc.SecretARN)
+		if smc.RoleARN != nil {
+			secretsManagerConfiguration[names.AttrRoleARN] = aws.ToString(smc.RoleARN)
+		}
+	}
+	return []interface{}{secretsManagerConfiguration}
+}
+
 func flattenDynamicPartitioningConfiguration(dpc *types.DynamicPartitioningConfiguration) []map[string]interface{} {
 	if dpc == nil {
 		return []map[string]interface{}{}

From 1fd3273f59e1da4ea6e5d572104b3c2aa5bab508 Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Thu, 27 Jun 2024 12:26:40 +0900
Subject: [PATCH 03/20] Implement SecretsManagerConfiguration to Firehose
 Snowflake Delivery Stream

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 internal/service/firehose/delivery_stream.go | 150 ++++++++++++-------
 1 file changed, 100 insertions(+), 50 deletions(-)

diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index b906587fed0..61a356a23c1 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -192,6 +192,33 @@ func resourceDeliveryStream() *schema.Resource {
 					},
 				}
 			}
+			secretsManagerConfigurationSchema := func() *schema.Schema {
+				return &schema.Schema{
+					Type:     schema.TypeList,
+					MaxItems: 1,
+					Optional: true,
+					Computed: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							names.AttrEnabled: {
+								Type:     schema.TypeBool,
+								Optional: true,
+								Default:  false,
+							},
+							"secret_arn": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ValidateFunc: verify.ValidARN,
+							},
+							names.AttrRoleARN: {
+								Type:     schema.TypeString,
+								Optional: true,
+								ValidateFunc: verify.ValidARN,
+							},
+						},
+					},
+				}
+			}
 			requestConfigurationSchema := func() *schema.Schema {
 				return &schema.Schema{
 					Type:     schema.TypeList,
@@ -1007,10 +1034,10 @@ func resourceDeliveryStream() *schema.Resource {
 								ValidateFunc: validation.StringLenBetween(1, 255),
 							},
 							"key_passphrase": {
-								Type:         schema.TypeString,
-								Optional:     true,
-								Sensitive:    true,
-								ValidateFunc: validation.StringLenBetween(7, 255),
+								Type:          schema.TypeString,
+								Optional:      true,
+								Sensitive:     true,
+								ValidateFunc:  validation.StringLenBetween(7, 255),
 							},
 							"metadata_column_name": {
 								Type:         schema.TypeString,
@@ -1018,9 +1045,9 @@ func resourceDeliveryStream() *schema.Resource {
 								ValidateFunc: validation.StringLenBetween(1, 255),
 							},
 							names.AttrPrivateKey: {
-								Type:      schema.TypeString,
-								Required:  true,
-								Sensitive: true,
+								Type:          schema.TypeString,
+								Optional:      true,
+								Sensitive:     true,
 							},
 							"processing_configuration": processingConfigurationSchema(),
 							"retry_duration": {
@@ -1046,6 +1073,7 @@ func resourceDeliveryStream() *schema.Resource {
 								Required:     true,
 								ValidateFunc: validation.StringLenBetween(1, 255),
 							},
+							"secrets_manager_configuration": secretsManagerConfigurationSchema(),
 							"snowflake_role_configuration": {
 								Type:     schema.TypeList,
 								Optional: true,
@@ -1085,9 +1113,9 @@ func resourceDeliveryStream() *schema.Resource {
 								ValidateFunc: validation.StringLenBetween(1, 255),
 							},
 							"user": {
-								Type:         schema.TypeString,
-								Required:     true,
-								ValidateFunc: validation.StringLenBetween(1, 255),
+								Type:          schema.TypeString,
+								Optional:      true,
+								ValidateFunc:  validation.StringLenBetween(1, 255),
 							},
 						},
 					},
@@ -2314,24 +2342,24 @@ func expandSecretsManagerConfiguration(tfMap map[string]interface{}) *types.Secr
 		// It is possible to just pass nil here, but this seems to be the
 		// canonical form that AWS uses, and is less likely to produce diffs.
 		return &types.SecretsManagerConfiguration{
-			Enabled:    aws.Bool(false),
+			Enabled: aws.Bool(false),
 		}
 	}
 
-  secretsManagerConfiguration := config[0].(map[string]interface{})
-  configuration := &types.SecretsManagerConfiguration{
-    Enabled:    aws.Bool(secretsManagerConfiguration[names.AttrEnabled].(bool)),
-  }
+	secretsManagerConfiguration := config[0].(map[string]interface{})
+	configuration := &types.SecretsManagerConfiguration{
+		Enabled: aws.Bool(secretsManagerConfiguration[names.AttrEnabled].(bool)),
+	}
 
-  if v, ok := secretsManagerConfiguration["secret_arn"]; ok {
-    configuration.SecretARN = aws.String(v.(string))
-  }
+	if v, ok := secretsManagerConfiguration["secret_arn"]; ok {
+		configuration.SecretARN = aws.String(v.(string))
+	}
 
-  if v, ok := secretsManagerConfiguration[names.AttrRoleARN]; ok {
-    configuration.RoleARN = aws.String(v.(string))
-  }
+	if v, ok := secretsManagerConfiguration[names.AttrRoleARN]; ok {
+		configuration.RoleARN = aws.String(v.(string))
+	}
 
-  return configuration
+	return configuration
 }
 
 func expandEncryptionConfiguration(s3 map[string]interface{}) *types.EncryptionConfiguration {
@@ -2675,15 +2703,14 @@ func expandAmazonOpenSearchServerlessDestinationUpdate(oss map[string]interface{
 func expandSnowflakeDestinationConfiguration(tfMap map[string]interface{}) *types.SnowflakeDestinationConfiguration {
 	roleARN := tfMap[names.AttrRoleARN].(string)
 	apiObject := &types.SnowflakeDestinationConfiguration{
-		AccountUrl:      aws.String(tfMap["account_url"].(string)),
-		Database:        aws.String(tfMap[names.AttrDatabase].(string)),
-		PrivateKey:      aws.String(tfMap[names.AttrPrivateKey].(string)),
-		RetryOptions:    expandSnowflakeRetryOptions(tfMap),
-		RoleARN:         aws.String(roleARN),
-		S3Configuration: expandS3DestinationConfiguration(tfMap["s3_configuration"].([]interface{})),
-		Schema:          aws.String(tfMap[names.AttrSchema].(string)),
-		Table:           aws.String(tfMap["table"].(string)),
-		User:            aws.String(tfMap["user"].(string)),
+		AccountUrl:                aws.String(tfMap["account_url"].(string)),
+		Database:                  aws.String(tfMap[names.AttrDatabase].(string)),
+		RetryOptions:              expandSnowflakeRetryOptions(tfMap),
+		RoleARN:                   aws.String(roleARN),
+		S3Configuration:           expandS3DestinationConfiguration(tfMap["s3_configuration"].([]interface{})),
+		Schema:                    aws.String(tfMap[names.AttrSchema].(string)),
+		SnowflakeVpcConfiguration: expandSnowflakeVPCConfiguration(tfMap),
+		Table:                     aws.String(tfMap["table"].(string)),
 	}
 
 	if _, ok := tfMap["cloudwatch_logging_options"]; ok {
@@ -2698,6 +2725,10 @@ func expandSnowflakeDestinationConfiguration(tfMap map[string]interface{}) *type
 		apiObject.DataLoadingOption = types.SnowflakeDataLoadingOption(v.(string))
 	}
 
+	if v, ok := tfMap[names.AttrPrivateKey]; ok && v.(string) != "" {
+		apiObject.PrivateKey = aws.String(v.(string))
+	}
+
 	if v, ok := tfMap["key_passphrase"]; ok && v.(string) != "" {
 		apiObject.KeyPassphrase = aws.String(v.(string))
 	}
@@ -2714,6 +2745,10 @@ func expandSnowflakeDestinationConfiguration(tfMap map[string]interface{}) *type
 		apiObject.S3BackupMode = types.SnowflakeS3BackupMode(v.(string))
 	}
 
+	if _, ok := tfMap["secrets_manager_configuration"]; ok {
+		apiObject.SecretsManagerConfiguration = expandSecretsManagerConfiguration(tfMap)
+	}
+
 	if _, ok := tfMap["snowflake_role_configuration"]; ok {
 		apiObject.SnowflakeRoleConfiguration = expandSnowflakeRoleConfiguration(tfMap)
 	}
@@ -2722,6 +2757,10 @@ func expandSnowflakeDestinationConfiguration(tfMap map[string]interface{}) *type
 		apiObject.SnowflakeVpcConfiguration = expandSnowflakeVPCConfiguration(tfMap)
 	}
 
+	if v, ok := tfMap["user"]; ok && v.(string) != "" {
+		apiObject.User = aws.String(v.(string))
+	}
+
 	return apiObject
 }
 
@@ -2730,13 +2769,11 @@ func expandSnowflakeDestinationUpdate(tfMap map[string]interface{}) *types.Snowf
 	apiObject := &types.SnowflakeDestinationUpdate{
 		AccountUrl:   aws.String(tfMap["account_url"].(string)),
 		Database:     aws.String(tfMap[names.AttrDatabase].(string)),
-		PrivateKey:   aws.String(tfMap[names.AttrPrivateKey].(string)),
 		RetryOptions: expandSnowflakeRetryOptions(tfMap),
 		RoleARN:      aws.String(roleARN),
 		S3Update:     expandS3DestinationUpdate(tfMap["s3_configuration"].([]interface{})),
 		Schema:       aws.String(tfMap[names.AttrSchema].(string)),
 		Table:        aws.String(tfMap["table"].(string)),
-		User:         aws.String(tfMap["user"].(string)),
 	}
 
 	if _, ok := tfMap["cloudwatch_logging_options"]; ok {
@@ -2751,6 +2788,10 @@ func expandSnowflakeDestinationUpdate(tfMap map[string]interface{}) *types.Snowf
 		apiObject.DataLoadingOption = types.SnowflakeDataLoadingOption(v.(string))
 	}
 
+	if v, ok := tfMap[names.AttrPrivateKey]; ok && v.(string) != "" {
+		apiObject.PrivateKey = aws.String(v.(string))
+	}
+
 	if v, ok := tfMap["key_passphrase"]; ok && v.(string) != "" {
 		apiObject.KeyPassphrase = aws.String(v.(string))
 	}
@@ -2767,10 +2808,18 @@ func expandSnowflakeDestinationUpdate(tfMap map[string]interface{}) *types.Snowf
 		apiObject.S3BackupMode = types.SnowflakeS3BackupMode(v.(string))
 	}
 
+	if _, ok := tfMap["secrets_manager_configuration"]; ok {
+		apiObject.SecretsManagerConfiguration = expandSecretsManagerConfiguration(tfMap)
+	}
+
 	if _, ok := tfMap["snowflake_role_configuration"]; ok {
 		apiObject.SnowflakeRoleConfiguration = expandSnowflakeRoleConfiguration(tfMap)
 	}
 
+	if v, ok := tfMap["user"]; ok && v.(string) != "" {
+		apiObject.User = aws.String(v.(string))
+	}
+
 	return apiObject
 }
 
@@ -3436,23 +3485,24 @@ func flattenSnowflakeDestinationDescription(apiObject *types.SnowflakeDestinatio
 
 	roleARN := aws.ToString(apiObject.RoleARN)
 	tfMap := map[string]interface{}{
-		"account_url":                  aws.ToString(apiObject.AccountUrl),
-		"cloudwatch_logging_options":   flattenCloudWatchLoggingOptions(apiObject.CloudWatchLoggingOptions),
-		"content_column_name":          aws.ToString(apiObject.ContentColumnName),
-		"data_loading_option":          apiObject.DataLoadingOption,
-		names.AttrDatabase:             aws.ToString(apiObject.Database),
-		"key_passphrase":               configuredKeyPassphrase,
-		"metadata_column_name":         aws.ToString(apiObject.MetaDataColumnName),
-		names.AttrPrivateKey:           configuredPrivateKey,
-		"processing_configuration":     flattenProcessingConfiguration(apiObject.ProcessingConfiguration, destinationTypeSnowflake, roleARN),
-		names.AttrRoleARN:              roleARN,
-		"s3_backup_mode":               apiObject.S3BackupMode,
-		"s3_configuration":             flattenS3DestinationDescription(apiObject.S3DestinationDescription),
-		names.AttrSchema:               aws.ToString(apiObject.Schema),
-		"snowflake_role_configuration": flattenSnowflakeRoleConfiguration(apiObject.SnowflakeRoleConfiguration),
-		"snowflake_vpc_configuration":  flattenSnowflakeVPCConfiguration(apiObject.SnowflakeVpcConfiguration),
-		"table":                        aws.ToString(apiObject.Table),
-		"user":                         aws.ToString(apiObject.User),
+		"account_url":                   aws.ToString(apiObject.AccountUrl),
+		"cloudwatch_logging_options":    flattenCloudWatchLoggingOptions(apiObject.CloudWatchLoggingOptions),
+		"content_column_name":           aws.ToString(apiObject.ContentColumnName),
+		"data_loading_option":           apiObject.DataLoadingOption,
+		names.AttrDatabase:              aws.ToString(apiObject.Database),
+		"key_passphrase":                configuredKeyPassphrase,
+		"metadata_column_name":          aws.ToString(apiObject.MetaDataColumnName),
+		names.AttrPrivateKey:            configuredPrivateKey,
+		"processing_configuration":      flattenProcessingConfiguration(apiObject.ProcessingConfiguration, destinationTypeSnowflake, roleARN),
+		names.AttrRoleARN:               roleARN,
+		"s3_backup_mode":                apiObject.S3BackupMode,
+		"s3_configuration":              flattenS3DestinationDescription(apiObject.S3DestinationDescription),
+		names.AttrSchema:                aws.ToString(apiObject.Schema),
+		"secrets_manager_configuration": flattenSecretsManagerConfiguration(apiObject.SecretsManagerConfiguration),
+		"snowflake_role_configuration":  flattenSnowflakeRoleConfiguration(apiObject.SnowflakeRoleConfiguration),
+		"snowflake_vpc_configuration":   flattenSnowflakeVPCConfiguration(apiObject.SnowflakeVpcConfiguration),
+		"table":                         aws.ToString(apiObject.Table),
+		"user":                          aws.ToString(apiObject.User),
 	}
 
 	if apiObject.RetryOptions != nil {

From 1c5c9d9dfdfab41b1da039347318f8ee00a0f9d2 Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Thu, 27 Jun 2024 18:34:14 +0900
Subject: [PATCH 04/20] Add a test for SecretsManagerConfiguration for
 Snowflake

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 .../service/firehose/delivery_stream_test.go  | 166 ++++++++++++++++++
 1 file changed, 166 insertions(+)

diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index 37ea1e4c3e3..6c55dc558cf 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -1216,6 +1216,71 @@ func TestAccFirehoseDeliveryStream_snowflakeUpdates(t *testing.T) {
 					resource.TestCheckResourceAttrSet(resourceName, "version_id"),
 				),
 			},
+			{
+				Config: testAccDeliveryStreamConfig_snowflakeUpdateSecretsManager(rName, key),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckDeliveryStreamExists(ctx, resourceName, &stream),
+					resource.TestCheckResourceAttrSet(resourceName, names.AttrARN),
+					resource.TestCheckResourceAttr(resourceName, names.AttrDestination, "snowflake"),
+					resource.TestCheckResourceAttrSet(resourceName, "destination_id"),
+					resource.TestCheckResourceAttr(resourceName, "elasticsearch_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, "extended_s3_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, "kinesis_source_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, names.AttrName, rName),
+					resource.TestCheckResourceAttr(resourceName, "opensearch_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, "opensearchserverless_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, "redshift_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, "server_side_encryption.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "server_side_encryption.0.enabled", acctest.CtFalse),
+					resource.TestCheckResourceAttr(resourceName, "server_side_encryption.0.key_arn", ""),
+					resource.TestCheckResourceAttr(resourceName, "server_side_encryption.0.key_type", "AWS_OWNED_CMK"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.account_url", fmt.Sprintf("https://%s.snowflakecomputing.com", rName)),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.cloudwatch_logging_options.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.cloudwatch_logging_options.0.enabled", acctest.CtFalse),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.cloudwatch_logging_options.0.log_group_name", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.cloudwatch_logging_options.0.log_stream_name", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.content_column_name", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.data_loading_option", "JSON_MAPPING"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.database", "test-db"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.key_passphrase", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.metadata_column_name", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.private_key", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.processing_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.processing_configuration.0.enabled", acctest.CtFalse),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.processing_configuration.0.processors.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.retry_duration", "60"),
+					resource.TestCheckResourceAttrSet(resourceName, "snowflake_configuration.0.role_arn"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_backup_mode", "FailedDataOnly"),
+					resource.TestCheckResourceAttrSet(resourceName, "snowflake_configuration.0.s3_configuration.0.bucket_arn"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.buffering_interval", "400"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.buffering_size", acctest.Ct10),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.cloudwatch_logging_options.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.cloudwatch_logging_options.0.enabled", acctest.CtFalse),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.cloudwatch_logging_options.0.log_group_name", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.cloudwatch_logging_options.0.log_stream_name", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.compression_format", "GZIP"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.error_output_prefix", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.kms_key_arn", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.s3_configuration.0.prefix", ""),
+					resource.TestCheckResourceAttrSet(resourceName, "snowflake_configuration.0.s3_configuration.0.role_arn"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.schema", "test-schema"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.secrets_manager_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.secrets_manager_configuration.0.enabled", acctest.CtFalse),
+					resource.TestCheckResourceAttrSet(resourceName, "snowflake_configuration.0.secrets_manager_configuration.0.role_arn"),
+					resource.TestCheckResourceAttrSet(resourceName, "snowflake_configuration.0.secrets_manager_configuration.0.secret_arn"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.snowflake_role_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.snowflake_role_configuration.0.enabled", acctest.CtFalse),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.snowflake_role_configuration.0.snowflake_role", ""),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.snowflake_vpc_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.table", "test-table"),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.user", ""),
+					resource.TestCheckResourceAttr(resourceName, "splunk_configuration.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(resourceName, acctest.CtTagsPercent, acctest.Ct0),
+					resource.TestCheckResourceAttrSet(resourceName, "version_id"),
+				),
+			},
 		},
 	})
 }
@@ -2802,6 +2867,75 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 `, rName)
 }
 
+func testAccDeliveryStreamConfig_baseSecretsManager(rName, privateKey string) string {
+	return fmt.Sprintf(`
+data "aws_caller_identity" "current" {}
+
+resource "aws_iam_role" "iam_for_secretsmanager" {
+  name = "%[1]s-secretsmanager"
+
+  assume_role_policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Sid": "",
+      "Effect": "Allow",
+      "Principal": {
+        "Service": "firehose.amazonaws.com"
+      },
+      "Action": "sts:AssumeRole",
+      "Condition": {
+        "StringEquals": {
+          "sts:ExternalId": "${data.aws_caller_identity.current.account_id}"
+        }
+      }
+    }
+  ]
+}
+EOF
+}
+
+resource "aws_secretsmanager_secret" "test" {
+  name = %[1]q
+}
+
+resource "aws_secretsmanager_secret_version" "test" {
+  secret_id     = aws_secretsmanager_secret.test.id
+  secret_string = jsonencode({
+    user        = "%[1]s"
+    private_key = "%[2]s"
+  })
+}
+
+
+resource "aws_iam_role_policy" "iam_policy_for_secretsmanager" {
+  name = "%[1]s-secretsmanager"
+  role = aws_iam_role.iam_for_secretsmanager.id
+
+  policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Sid": "",
+      "Effect": "Allow",
+      "Action": [
+        "secret:GetSecretValue",
+        "secret:ListSecretVersionIds"
+      ],
+      "Resource": [
+        "${aws_secretmanager_secret.test.arn}"
+      ]
+    }
+  ]
+}
+EOF
+}
+`, rName, acctest.TLSPEMRemoveRSAPrivateKeyEncapsulationBoundaries(acctest.TLSPEMRemoveNewlines(privateKey)))
+}
+
+
 func testAccDeliveryStreamConfig_tags1(rName, tagKey1, tagValue1 string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_kinesis_firehose_delivery_stream" "test" {
@@ -3793,6 +3927,38 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 `, rName, acctest.TLSPEMRemoveRSAPrivateKeyEncapsulationBoundaries(acctest.TLSPEMRemoveNewlines(privateKey))))
 }
 
+func testAccDeliveryStreamConfig_snowflakeUpdateSecretsManager(rName, privateKey string) string {
+	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), testAccDeliveryStreamConfig_baseSecretsManager(rName, privateKey), fmt.Sprintf(`
+resource "aws_kinesis_firehose_delivery_stream" "test" {
+  depends_on  = [aws_iam_role_policy.firehose, aws_iam_role_policy.iam_for_secrets_manager]
+  name        = %[1]q
+  destination = "snowflake"
+
+  snowflake_configuration {
+    account_url = "https://%[1]s.snowflakecomputing.com"
+    database    = "test-db"
+    role_arn    = aws_iam_role.firehose.arn
+    schema      = "test-schema"
+    table       = "test-table"
+
+    secrets_manager_configuration {
+      enabled    = true
+      secret_arn = aws_secretsmanager_secret.test1.arn
+      role_arn   = aws_iam_role.iam_for_secrets_manager.arn
+    }
+
+    s3_configuration {
+      role_arn           = aws_iam_role.firehose.arn
+      bucket_arn         = aws_s3_bucket.bucket.arn
+      buffering_size     = 10
+      buffering_interval = 400
+      compression_format = "GZIP"
+    }
+  }
+}
+`, rName))
+}
+
 func testAccDeliveryStreamConfig_splunkBasic(rName string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_kinesis_firehose_delivery_stream" "test" {

From 3417ceaec4e7f9a3377962aa75c0625cf1be9747 Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Thu, 27 Jun 2024 18:45:56 +0900
Subject: [PATCH 05/20] make fmt

---
 internal/service/firehose/delivery_stream.go  | 28 +++++++++----------
 .../service/firehose/delivery_stream_test.go  |  1 -
 2 files changed, 14 insertions(+), 15 deletions(-)

diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index 61a356a23c1..92008c95729 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -206,13 +206,13 @@ func resourceDeliveryStream() *schema.Resource {
 								Default:  false,
 							},
 							"secret_arn": {
-								Type:     schema.TypeString,
-								Optional: true,
+								Type:         schema.TypeString,
+								Optional:     true,
 								ValidateFunc: verify.ValidARN,
 							},
 							names.AttrRoleARN: {
-								Type:     schema.TypeString,
-								Optional: true,
+								Type:         schema.TypeString,
+								Optional:     true,
 								ValidateFunc: verify.ValidARN,
 							},
 						},
@@ -1034,10 +1034,10 @@ func resourceDeliveryStream() *schema.Resource {
 								ValidateFunc: validation.StringLenBetween(1, 255),
 							},
 							"key_passphrase": {
-								Type:          schema.TypeString,
-								Optional:      true,
-								Sensitive:     true,
-								ValidateFunc:  validation.StringLenBetween(7, 255),
+								Type:         schema.TypeString,
+								Optional:     true,
+								Sensitive:    true,
+								ValidateFunc: validation.StringLenBetween(7, 255),
 							},
 							"metadata_column_name": {
 								Type:         schema.TypeString,
@@ -1045,9 +1045,9 @@ func resourceDeliveryStream() *schema.Resource {
 								ValidateFunc: validation.StringLenBetween(1, 255),
 							},
 							names.AttrPrivateKey: {
-								Type:          schema.TypeString,
-								Optional:      true,
-								Sensitive:     true,
+								Type:      schema.TypeString,
+								Optional:  true,
+								Sensitive: true,
 							},
 							"processing_configuration": processingConfigurationSchema(),
 							"retry_duration": {
@@ -1113,9 +1113,9 @@ func resourceDeliveryStream() *schema.Resource {
 								ValidateFunc: validation.StringLenBetween(1, 255),
 							},
 							"user": {
-								Type:          schema.TypeString,
-								Optional:      true,
-								ValidateFunc:  validation.StringLenBetween(1, 255),
+								Type:         schema.TypeString,
+								Optional:     true,
+								ValidateFunc: validation.StringLenBetween(1, 255),
 							},
 						},
 					},
diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index 6c55dc558cf..87570fd75c9 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -2935,7 +2935,6 @@ EOF
 `, rName, acctest.TLSPEMRemoveRSAPrivateKeyEncapsulationBoundaries(acctest.TLSPEMRemoveNewlines(privateKey)))
 }
 
-
 func testAccDeliveryStreamConfig_tags1(rName, tagKey1, tagValue1 string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_kinesis_firehose_delivery_stream" "test" {

From 92abfc1bb320a8ff7e0b93cd6bbe8f2d1c48e4f1 Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Thu, 27 Jun 2024 18:59:00 +0900
Subject: [PATCH 06/20] Add changelog

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 .changelog/38151.txt | 3 +++
 1 file changed, 3 insertions(+)
 create mode 100644 .changelog/38151.txt

diff --git a/.changelog/38151.txt b/.changelog/38151.txt
new file mode 100644
index 00000000000..5f3acb89eae
--- /dev/null
+++ b/.changelog/38151.txt
@@ -0,0 +1,3 @@
+```release-note:feature
+resource/aws_kinesis_firehose_delivery_stream: Add `secrets_manager_configuariton` attribute inside `snowflake_configuration`.
+```

From 7247c2cf6d9e30d8d1540632d2a7ff91bc08593f Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Thu, 27 Jun 2024 19:50:30 +0900
Subject: [PATCH 07/20] Fix test

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 internal/service/firehose/delivery_stream.go      |  4 ++--
 internal/service/firehose/delivery_stream_test.go | 14 +++++++-------
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index 92008c95729..9f4e9f65daf 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -2351,11 +2351,11 @@ func expandSecretsManagerConfiguration(tfMap map[string]interface{}) *types.Secr
 		Enabled: aws.Bool(secretsManagerConfiguration[names.AttrEnabled].(bool)),
 	}
 
-	if v, ok := secretsManagerConfiguration["secret_arn"]; ok {
+	if v, ok := secretsManagerConfiguration["secret_arn"]; ok && len(v.(string)) > 0 {
 		configuration.SecretARN = aws.String(v.(string))
 	}
 
-	if v, ok := secretsManagerConfiguration[names.AttrRoleARN]; ok {
+	if v, ok := secretsManagerConfiguration[names.AttrRoleARN]; ok && len(v.(string)) > 0 {
 		configuration.RoleARN = aws.String(v.(string))
 	}
 
diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index 87570fd75c9..ee6734bb4d4 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -1267,7 +1267,7 @@ func TestAccFirehoseDeliveryStream_snowflakeUpdates(t *testing.T) {
 					resource.TestCheckResourceAttrSet(resourceName, "snowflake_configuration.0.s3_configuration.0.role_arn"),
 					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.schema", "test-schema"),
 					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.secrets_manager_configuration.#", acctest.Ct1),
-					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.secrets_manager_configuration.0.enabled", acctest.CtFalse),
+					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.secrets_manager_configuration.0.enabled", acctest.CtTrue),
 					resource.TestCheckResourceAttrSet(resourceName, "snowflake_configuration.0.secrets_manager_configuration.0.role_arn"),
 					resource.TestCheckResourceAttrSet(resourceName, "snowflake_configuration.0.secrets_manager_configuration.0.secret_arn"),
 					resource.TestCheckResourceAttr(resourceName, "snowflake_configuration.0.snowflake_role_configuration.#", acctest.Ct1),
@@ -2869,7 +2869,7 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 
 func testAccDeliveryStreamConfig_baseSecretsManager(rName, privateKey string) string {
 	return fmt.Sprintf(`
-data "aws_caller_identity" "current" {}
+data "aws_caller_identity" "current_for_secretsmanager" {}
 
 resource "aws_iam_role" "iam_for_secretsmanager" {
   name = "%[1]s-secretsmanager"
@@ -2887,7 +2887,7 @@ resource "aws_iam_role" "iam_for_secretsmanager" {
       "Action": "sts:AssumeRole",
       "Condition": {
         "StringEquals": {
-          "sts:ExternalId": "${data.aws_caller_identity.current.account_id}"
+          "sts:ExternalId": "${data.aws_caller_identity.current_for_secretsmanager.account_id}"
         }
       }
     }
@@ -2925,7 +2925,7 @@ resource "aws_iam_role_policy" "iam_policy_for_secretsmanager" {
         "secret:ListSecretVersionIds"
       ],
       "Resource": [
-        "${aws_secretmanager_secret.test.arn}"
+        "${aws_secretsmanager_secret.test.arn}"
       ]
     }
   ]
@@ -3929,7 +3929,7 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 func testAccDeliveryStreamConfig_snowflakeUpdateSecretsManager(rName, privateKey string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), testAccDeliveryStreamConfig_baseSecretsManager(rName, privateKey), fmt.Sprintf(`
 resource "aws_kinesis_firehose_delivery_stream" "test" {
-  depends_on  = [aws_iam_role_policy.firehose, aws_iam_role_policy.iam_for_secrets_manager]
+  depends_on  = [aws_iam_role_policy.firehose, aws_iam_role_policy.iam_policy_for_secretsmanager]
   name        = %[1]q
   destination = "snowflake"
 
@@ -3942,8 +3942,8 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 
     secrets_manager_configuration {
       enabled    = true
-      secret_arn = aws_secretsmanager_secret.test1.arn
-      role_arn   = aws_iam_role.iam_for_secrets_manager.arn
+      secret_arn = aws_secretsmanager_secret.test.arn
+      role_arn   = aws_iam_role.iam_for_secretsmanager.arn
     }
 
     s3_configuration {

From cefe73a46ff6d62cfa9a0c7dde663f34a2dae254 Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Thu, 27 Jun 2024 20:34:10 +0900
Subject: [PATCH 08/20] Fix "Error: updating Kinesis Firehose Delivery Stream
 (tf-acc-test-19787509517258576): operation error Firehose: UpdateDestination,
 https response error StatusCode: 400, RequestID:
 e37d3184-7f21-2c18-bd8a-7dcdd054ed55, api error ValidationException: 1
 validation error detected: Value at
 'snowflakeDestinationUpdate.snowflakeRoleConfiguration.snowflakeRole' failed
 to satisfy constraint: Member must have length greater than or equal to 1" on
 deleting the configuration

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 internal/service/firehose/delivery_stream.go | 24 +++++++++++++-------
 1 file changed, 16 insertions(+), 8 deletions(-)

diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index 9f4e9f65daf..149b71c1070 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -3120,16 +3120,22 @@ func expandSnowflakeRetryOptions(tfMap map[string]interface{}) *types.SnowflakeR
 }
 
 func expandSnowflakeRoleConfiguration(tfMap map[string]interface{}) *types.SnowflakeRoleConfiguration {
-	tfList := tfMap["snowflake_role_configuration"].([]interface{})
-	if len(tfList) == 0 {
-		return nil
+	config := tfMap["snowflake_role_configuration"].([]interface{})
+	if len(config) == 0 || config[0] == nil {
+		// It is possible to just pass nil here, but this seems to be the
+		// canonical form that AWS uses, and is less likely to produce diffs.
+		return &types.SnowflakeRoleConfiguration{
+			Enabled: aws.Bool(false),
+		}
 	}
 
-	tfMap = tfList[0].(map[string]interface{})
-
+	snowflakeRoleConfiguration := config[0].(map[string]interface{})
 	apiObject := &types.SnowflakeRoleConfiguration{
-		Enabled:       aws.Bool(tfMap[names.AttrEnabled].(bool)),
-		SnowflakeRole: aws.String(tfMap["snowflake_role"].(string)),
+		Enabled: aws.Bool(snowflakeRoleConfiguration[names.AttrEnabled].(bool)),
+	}
+
+	if v, ok := snowflakeRoleConfiguration["snowflake_role"]; ok && len(v.(string)) > 0 {
+		apiObject.SnowflakeRole = aws.String(v.(string))
 	}
 
 	return apiObject
@@ -3961,7 +3967,9 @@ func flattenSnowflakeRoleConfiguration(apiObject *types.SnowflakeRoleConfigurati
 
 	m := map[string]interface{}{
 		names.AttrEnabled: aws.ToBool(apiObject.Enabled),
-		"snowflake_role":  aws.ToString(apiObject.SnowflakeRole),
+	}
+	if aws.ToBool(apiObject.Enabled) {
+		m["snowflake_role"] = aws.ToString(apiObject.SnowflakeRole)
 	}
 
 	return []map[string]interface{}{m}

From 6bbdb389e6f293b825681fb3f0d6f6f211967317 Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Fri, 28 Jun 2024 06:50:28 +0900
Subject: [PATCH 09/20] Fix terrafmt

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 internal/service/firehose/delivery_stream_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index ee6734bb4d4..078bed65cea 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -2901,7 +2901,7 @@ resource "aws_secretsmanager_secret" "test" {
 }
 
 resource "aws_secretsmanager_secret_version" "test" {
-  secret_id     = aws_secretsmanager_secret.test.id
+  secret_id = aws_secretsmanager_secret.test.id
   secret_string = jsonencode({
     user        = "%[1]s"
     private_key = "%[2]s"

From 424832bb51eb79a3f45e0c3bef8b15f30b6b5991 Mon Sep 17 00:00:00 2001
From: Takahiro Nakayama <civitaspo@gmail.com>
Date: Fri, 28 Jun 2024 20:55:54 +0900
Subject: [PATCH 10/20] Add documents for kinesis_firehose_delivery_stream
 secrets_manager_configuration

Signed-off-by: Takahiro Nakayama <civitaspo@gmail.com>
---
 ...kinesis_firehose_delivery_stream.html.markdown | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
index 7f242b2d883..a2ce3655157 100644
--- a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
+++ b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
@@ -793,9 +793,9 @@ The `http_endpoint_configuration` configuration block supports the following arg
 The `snowflake_configuration` configuration block supports the following arguments:
 
 * `account_url` - (Required) The URL of the Snowflake account. Format: https://[account_identifier].snowflakecomputing.com.
-* `private_key` - (Required) The private key for authentication.
-* `key_passphrase` - (Required) The passphrase for the private key.
-* `user` - (Required) The user for authentication.
+* `private_key` - (Optional) The private key for authentication. This value is required if `secrets_manager_configuration` is not provided.
+* `key_passphrase` - (Optional) The passphrase for the private key.
+* `user` - (Optional) The user for authentication. This value is required if `secrets_manager_configuration` is not provided.
 * `database` - (Required) The Snowflake database name.
 * `schema` - (Required) The Snowflake schema name.
 * `table` - (Required) The Snowflake table name.
@@ -813,6 +813,7 @@ The `snowflake_configuration` configuration block supports the following argumen
 * `retry_duration` - (Optional) After an initial failure to deliver to Snowflake, the total amount of time, in seconds between 0 to 7200, during which Firehose re-attempts delivery (including the first attempt).  After this time has elapsed, the failed documents are written to Amazon S3.  The default value is 60s.  There will be no retry if the value is 0.
 * `s3_backup_mode` - (Optional) The S3 backup mode.
 * `s3_configuration` - (Required) The S3 configuration. See [`s3_configuration` block](#s3_configuration-block) below for details.
+* `secrets_manager_configuration` - (Optional) The SecretsManager configuration. See [`secrets_manager_configuration` block](#secrets_manager_configuration-block) below for details. This value is required if `user` and `private_key` is not provided.
 
 ### `cloudwatch_logging_options` block
 
@@ -927,6 +928,14 @@ The `s3_configuration` configuration block supports the following arguments:
   be used.
 * `cloudwatch_logging_options` - (Optional) The CloudWatch Logging Options for the delivery stream. See [`cloudwatch_logging_options` block](#cloudwatch_logging_options-block) below for details.
 
+### `secrets_manager_configuration` block
+
+The `secrets_manager_configuration` configuration block supports the following arguments:
+
+* `enabled` - (Optional) Enables or disables the Secrets Manager configuration. Defaults to `false`.
+* `secret_arn` - (Optional) The ARN of the Secrets Manager secret. This value is required if `enabled` is true.
+* `role_arn` - (Optional) The ARN of the role the stream assumes.
+
 ### `input_format_configuration` block
 
 The `input_format_configuration` configuration block supports the following arguments:

From 9e3af47d61a6fbb674dc1e1740fe144a039125b6 Mon Sep 17 00:00:00 2001
From: nikhil <nplusg1993@gmail.com>
Date: Wed, 3 Jul 2024 21:00:09 +0100
Subject: [PATCH 11/20] aws_kinesis_firehose_delivery_stream

---
 internal/service/firehose/delivery_stream.go  | 93 +++++++++++++++++--
 .../service/firehose/delivery_stream_test.go  | 61 ++++++++++++
 2 files changed, 144 insertions(+), 10 deletions(-)

diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index 02c293f7c25..95c493fc54e 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -290,6 +290,32 @@ func resourceDeliveryStream() *schema.Resource {
 					Elem:     s3ConfigurationElem(),
 				}
 			}
+			secretsManagerConfigurationSchema := func() *schema.Schema {
+				return &schema.Schema{
+					Type:     schema.TypeList,
+					MaxItems: 1,
+					Optional: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							names.AttrEnabled: {
+								Type:     schema.TypeBool,
+								Required: true,
+								ForceNew: true,
+							},
+							names.AttrRoleARN: {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ValidateFunc: verify.ValidARN,
+							},
+							"secret_arn": {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ValidateFunc: verify.ValidARN,
+							},
+						},
+					},
+				}
+			}
 
 			return map[string]*schema.Schema{
 				names.AttrARN: {
@@ -780,7 +806,8 @@ func resourceDeliveryStream() *schema.Resource {
 								Default:          types.HttpEndpointS3BackupModeFailedDataOnly,
 								ValidateDiagFunc: enum.Validate[types.HttpEndpointS3BackupMode](),
 							},
-							"s3_configuration": s3ConfigurationSchema(),
+							"s3_configuration":             s3ConfigurationSchema(),
+							"secret_manager_configuration": secretsManagerConfigurationSchema(),
 							names.AttrURL: {
 								Type:     schema.TypeString,
 								Required: true,
@@ -2851,6 +2878,10 @@ func expandHTTPEndpointDestinationConfiguration(httpEndpoint map[string]interfac
 		configuration.S3BackupMode = types.HttpEndpointS3BackupMode(s3BackupMode.(string))
 	}
 
+	if _, ok := httpEndpoint["secret_manager_configuration"]; ok {
+		configuration.SecretsManagerConfiguration = expandSecretsManagerConfiguration(httpEndpoint)
+	}
+
 	return configuration
 }
 
@@ -2890,6 +2921,10 @@ func expandHTTPEndpointDestinationUpdate(httpEndpoint map[string]interface{}) *t
 		configuration.S3BackupMode = types.HttpEndpointS3BackupMode(s3BackupMode.(string))
 	}
 
+	if _, ok := httpEndpoint["secret_manager_configuration"]; ok {
+		configuration.SecretsManagerConfiguration = expandSecretsManagerConfiguration(httpEndpoint)
+	}
+
 	return configuration
 }
 
@@ -3075,6 +3110,28 @@ func expandSnowflakeVPCConfiguration(tfMap map[string]interface{}) *types.Snowfl
 	return apiObject
 }
 
+func expandSecretsManagerConfiguration(sc map[string]interface{}) *types.SecretsManagerConfiguration {
+	config := sc["secret_manager_configuration"].([]interface{})
+	if len(config) == 0 {
+		return nil
+	}
+
+	SecretsManagerConfig := config[0].(map[string]interface{})
+	SecretsManagerOptions := &types.SecretsManagerConfiguration{
+		Enabled: aws.Bool(SecretsManagerConfig[names.AttrEnabled].(bool)),
+	}
+
+	if v, ok := SecretsManagerConfig[names.AttrRoleARN]; ok {
+		SecretsManagerOptions.RoleARN = aws.String(v.(string))
+	}
+
+	if v, ok := SecretsManagerConfig["secret_arn"]; ok {
+		SecretsManagerOptions.SecretARN = aws.String(v.(string))
+	}
+
+	return SecretsManagerOptions
+}
+
 func expandSplunkRetryOptions(splunk map[string]interface{}) *types.SplunkRetryOptions {
 	retryOptions := &types.SplunkRetryOptions{}
 
@@ -3812,15 +3869,16 @@ func flattenHTTPEndpointDestinationDescription(description *types.HttpEndpointDe
 		return []map[string]interface{}{}
 	}
 	m := map[string]interface{}{
-		names.AttrAccessKey:          configuredAccessKey,
-		names.AttrURL:                aws.ToString(description.EndpointConfiguration.Url),
-		names.AttrName:               aws.ToString(description.EndpointConfiguration.Name),
-		names.AttrRoleARN:            aws.ToString(description.RoleARN),
-		"s3_backup_mode":             description.S3BackupMode,
-		"s3_configuration":           flattenS3DestinationDescription(description.S3DestinationDescription),
-		"request_configuration":      flattenHTTPEndpointRequestConfiguration(description.RequestConfiguration),
-		"cloudwatch_logging_options": flattenCloudWatchLoggingOptions(description.CloudWatchLoggingOptions),
-		"processing_configuration":   flattenProcessingConfiguration(description.ProcessingConfiguration, destinationTypeHTTPEndpoint, aws.ToString(description.RoleARN)),
+		names.AttrAccessKey:            configuredAccessKey,
+		names.AttrURL:                  aws.ToString(description.EndpointConfiguration.Url),
+		names.AttrName:                 aws.ToString(description.EndpointConfiguration.Name),
+		names.AttrRoleARN:              aws.ToString(description.RoleARN),
+		"s3_backup_mode":               description.S3BackupMode,
+		"s3_configuration":             flattenS3DestinationDescription(description.S3DestinationDescription),
+		"request_configuration":        flattenHTTPEndpointRequestConfiguration(description.RequestConfiguration),
+		"cloudwatch_logging_options":   flattenCloudWatchLoggingOptions(description.CloudWatchLoggingOptions),
+		"processing_configuration":     flattenProcessingConfiguration(description.ProcessingConfiguration, destinationTypeHTTPEndpoint, aws.ToString(description.RoleARN)),
+		"secret_manager_configuration": flattenSecretsManagerConfiguration(description.SecretsManagerConfiguration),
 	}
 
 	if description.RetryOptions != nil {
@@ -3861,6 +3919,21 @@ func flattenDocumentIDOptions(apiObject *types.DocumentIdOptions) map[string]int
 	return tfMap
 }
 
+func flattenSecretsManagerConfiguration(sc *types.SecretsManagerConfiguration) []interface{} {
+	if sc == nil {
+		return []interface{}{}
+	}
+
+	secretsManagerOptions := map[string]interface{}{
+		names.AttrEnabled: aws.ToBool(sc.Enabled),
+	}
+	if aws.ToBool(sc.Enabled) {
+		secretsManagerOptions[names.AttrRoleARN] = aws.ToString(sc.RoleARN)
+		secretsManagerOptions["secret_arn"] = aws.ToString(sc.SecretARN)
+	}
+	return []interface{}{secretsManagerOptions}
+}
+
 func flattenSnowflakeRoleConfiguration(apiObject *types.SnowflakeRoleConfiguration) []map[string]interface{} {
 	if apiObject == nil {
 		return []map[string]interface{}{}
diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index 37ea1e4c3e3..034a7d2ccec 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -1469,6 +1469,36 @@ func TestAccFirehoseDeliveryStream_HTTPEndpoint_retryDuration(t *testing.T) {
 	})
 }
 
+func TestAccFirehoseDeliveryStream_HTTPEndpoint_SecretsManagerConfiguration(t *testing.T) {
+	ctx := acctest.Context(t)
+	var stream types.DeliveryStreamDescription
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_kinesis_firehose_delivery_stream.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.FirehoseServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckDeliveryStreamDestroy_ExtendedS3(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDeliveryStreamConfig_httpEndpointSecretsManager(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDeliveryStreamExists(ctx, resourceName, &stream),
+					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.0.enabled", acctest.CtTrue),
+					resource.TestCheckResourceAttrPair(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.0.secret_arn", "aws_secretsmanager_secret.test", names.AttrARN),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccFirehoseDeliveryStream_elasticSearchUpdates(t *testing.T) {
 	ctx := acctest.Context(t)
 	var stream types.DeliveryStreamDescription
@@ -4010,6 +4040,37 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 `, rName))
 }
 
+func testAccDeliveryStreamConfig_httpEndpointSecretsManager(rName string) string {
+	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
+resource "aws_secretsmanager_secret" "test" {
+  name = %[1]q
+}
+
+resource "aws_kinesis_firehose_delivery_stream" "test" {
+  depends_on  = [aws_iam_role_policy.firehose]
+  name        = %[1]q
+  destination = "http_endpoint"
+
+  http_endpoint_configuration {
+    url      = "https://input-test.com:443"
+    name     = "HTTP_test"
+    role_arn = aws_iam_role.firehose.arn
+
+    s3_configuration {
+      role_arn   = aws_iam_role.firehose.arn
+      bucket_arn = aws_s3_bucket.bucket.arn
+    }
+
+    secret_manager_configuration {
+      enabled    = true
+      role_arn   = aws_iam_role.firehose.arn
+      secret_arn = aws_secretsmanager_secret.test.arn
+    }
+  }
+}
+`, rName))
+}
+
 func testAccDeliveryStreamConfig_baseElasticsearch(rName string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_elasticsearch_domain" "test_cluster" {

From cb57ee72788376f869dff56fa0132703b8fb1442 Mon Sep 17 00:00:00 2001
From: nikhil <nplusg1993@gmail.com>
Date: Wed, 3 Jul 2024 21:08:42 +0100
Subject: [PATCH 12/20] aws_kinesis_firehose_delivery_stream

---
 .changelog/38245.txt                                     | 3 +++
 .../r/kinesis_firehose_delivery_stream.html.markdown     | 9 +++++++++
 2 files changed, 12 insertions(+)
 create mode 100644 .changelog/38245.txt

diff --git a/.changelog/38245.txt b/.changelog/38245.txt
new file mode 100644
index 00000000000..e9ebd19b928
--- /dev/null
+++ b/.changelog/38245.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+resource/aws_kinesis_firehose_delivery_stream: Add `secret_manager_configuration` attribute in `http_endpoint_configuration`
+```
\ No newline at end of file
diff --git a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
index 7f242b2d883..878509bc3cf 100644
--- a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
+++ b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
@@ -781,6 +781,7 @@ The `http_endpoint_configuration` configuration block supports the following arg
 * `role_arn` - (Required) Kinesis Data Firehose uses this IAM role for all the permissions that the delivery stream needs. The pattern needs to be `arn:.*`.
 * `s3_configuration` - (Required) The S3 Configuration. See [`s3_configuration` block](#s3_configuration-block) below for details.
 * `s3_backup_mode` - (Optional) Defines how documents should be delivered to Amazon S3.  Valid values are `FailedDataOnly` and `AllData`.  Default value is `FailedDataOnly`.
+* `secret_manager_configuration` - (Optional) The Secret Manager Configuration. See [`secret_manager_configuration` block](#secret_manager_configuration-block) below for details.
 * `buffering_size` - (Optional) Buffer incoming data to the specified size, in MBs, before delivering it to the destination. The default value is 5.
 * `buffering_interval` - (Optional) Buffer incoming data for the specified period of time, in seconds, before delivering it to the destination. The default value is 300 (5 minutes).
 * `cloudwatch_logging_options` - (Optional) The CloudWatch Logging Options for the delivery stream. See [`cloudwatch_logging_options` block](#cloudwatch_logging_options-block) below for details.
@@ -927,6 +928,14 @@ The `s3_configuration` configuration block supports the following arguments:
   be used.
 * `cloudwatch_logging_options` - (Optional) The CloudWatch Logging Options for the delivery stream. See [`cloudwatch_logging_options` block](#cloudwatch_logging_options-block) below for details.
 
+### `secret_manager_configuration` block
+
+The `secret_manager_configuration` configuration block supports the following arguments:
+
+* `enabled` - (Required) Enables or disables secrets manager feature.
+* `role_arn` - (Optional) The role that Firehose assumes when calling the Secrets Manager API operation.
+* `secret_arn` - (Optional) The ARN of the secret that stores your credentials.
+
 ### `input_format_configuration` block
 
 The `input_format_configuration` configuration block supports the following arguments:

From 25d354f7561b654c0d78dc5a8f3d14a27fe3c0ba Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 18 Jul 2024 08:14:13 -0400
Subject: [PATCH 13/20] Revert "aws_kinesis_firehose_delivery_stream"

This reverts commit cb57ee72788376f869dff56fa0132703b8fb1442.
---
 .changelog/38245.txt                                     | 3 ---
 .../r/kinesis_firehose_delivery_stream.html.markdown     | 9 ---------
 2 files changed, 12 deletions(-)
 delete mode 100644 .changelog/38245.txt

diff --git a/.changelog/38245.txt b/.changelog/38245.txt
deleted file mode 100644
index e9ebd19b928..00000000000
--- a/.changelog/38245.txt
+++ /dev/null
@@ -1,3 +0,0 @@
-```release-note:bug
-resource/aws_kinesis_firehose_delivery_stream: Add `secret_manager_configuration` attribute in `http_endpoint_configuration`
-```
\ No newline at end of file
diff --git a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
index 878509bc3cf..7f242b2d883 100644
--- a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
+++ b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
@@ -781,7 +781,6 @@ The `http_endpoint_configuration` configuration block supports the following arg
 * `role_arn` - (Required) Kinesis Data Firehose uses this IAM role for all the permissions that the delivery stream needs. The pattern needs to be `arn:.*`.
 * `s3_configuration` - (Required) The S3 Configuration. See [`s3_configuration` block](#s3_configuration-block) below for details.
 * `s3_backup_mode` - (Optional) Defines how documents should be delivered to Amazon S3.  Valid values are `FailedDataOnly` and `AllData`.  Default value is `FailedDataOnly`.
-* `secret_manager_configuration` - (Optional) The Secret Manager Configuration. See [`secret_manager_configuration` block](#secret_manager_configuration-block) below for details.
 * `buffering_size` - (Optional) Buffer incoming data to the specified size, in MBs, before delivering it to the destination. The default value is 5.
 * `buffering_interval` - (Optional) Buffer incoming data for the specified period of time, in seconds, before delivering it to the destination. The default value is 300 (5 minutes).
 * `cloudwatch_logging_options` - (Optional) The CloudWatch Logging Options for the delivery stream. See [`cloudwatch_logging_options` block](#cloudwatch_logging_options-block) below for details.
@@ -928,14 +927,6 @@ The `s3_configuration` configuration block supports the following arguments:
   be used.
 * `cloudwatch_logging_options` - (Optional) The CloudWatch Logging Options for the delivery stream. See [`cloudwatch_logging_options` block](#cloudwatch_logging_options-block) below for details.
 
-### `secret_manager_configuration` block
-
-The `secret_manager_configuration` configuration block supports the following arguments:
-
-* `enabled` - (Required) Enables or disables secrets manager feature.
-* `role_arn` - (Optional) The role that Firehose assumes when calling the Secrets Manager API operation.
-* `secret_arn` - (Optional) The ARN of the secret that stores your credentials.
-
 ### `input_format_configuration` block
 
 The `input_format_configuration` configuration block supports the following arguments:

From 5f34035bb02b5d26c91034f31053d414c67655f3 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 18 Jul 2024 08:14:19 -0400
Subject: [PATCH 14/20] Revert "aws_kinesis_firehose_delivery_stream"

This reverts commit 9e3af47d61a6fbb674dc1e1740fe144a039125b6.
---
 internal/service/firehose/delivery_stream.go  | 93 ++-----------------
 .../service/firehose/delivery_stream_test.go  | 61 ------------
 2 files changed, 10 insertions(+), 144 deletions(-)

diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index 95c493fc54e..02c293f7c25 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -290,32 +290,6 @@ func resourceDeliveryStream() *schema.Resource {
 					Elem:     s3ConfigurationElem(),
 				}
 			}
-			secretsManagerConfigurationSchema := func() *schema.Schema {
-				return &schema.Schema{
-					Type:     schema.TypeList,
-					MaxItems: 1,
-					Optional: true,
-					Elem: &schema.Resource{
-						Schema: map[string]*schema.Schema{
-							names.AttrEnabled: {
-								Type:     schema.TypeBool,
-								Required: true,
-								ForceNew: true,
-							},
-							names.AttrRoleARN: {
-								Type:         schema.TypeString,
-								Optional:     true,
-								ValidateFunc: verify.ValidARN,
-							},
-							"secret_arn": {
-								Type:         schema.TypeString,
-								Optional:     true,
-								ValidateFunc: verify.ValidARN,
-							},
-						},
-					},
-				}
-			}
 
 			return map[string]*schema.Schema{
 				names.AttrARN: {
@@ -806,8 +780,7 @@ func resourceDeliveryStream() *schema.Resource {
 								Default:          types.HttpEndpointS3BackupModeFailedDataOnly,
 								ValidateDiagFunc: enum.Validate[types.HttpEndpointS3BackupMode](),
 							},
-							"s3_configuration":             s3ConfigurationSchema(),
-							"secret_manager_configuration": secretsManagerConfigurationSchema(),
+							"s3_configuration": s3ConfigurationSchema(),
 							names.AttrURL: {
 								Type:     schema.TypeString,
 								Required: true,
@@ -2878,10 +2851,6 @@ func expandHTTPEndpointDestinationConfiguration(httpEndpoint map[string]interfac
 		configuration.S3BackupMode = types.HttpEndpointS3BackupMode(s3BackupMode.(string))
 	}
 
-	if _, ok := httpEndpoint["secret_manager_configuration"]; ok {
-		configuration.SecretsManagerConfiguration = expandSecretsManagerConfiguration(httpEndpoint)
-	}
-
 	return configuration
 }
 
@@ -2921,10 +2890,6 @@ func expandHTTPEndpointDestinationUpdate(httpEndpoint map[string]interface{}) *t
 		configuration.S3BackupMode = types.HttpEndpointS3BackupMode(s3BackupMode.(string))
 	}
 
-	if _, ok := httpEndpoint["secret_manager_configuration"]; ok {
-		configuration.SecretsManagerConfiguration = expandSecretsManagerConfiguration(httpEndpoint)
-	}
-
 	return configuration
 }
 
@@ -3110,28 +3075,6 @@ func expandSnowflakeVPCConfiguration(tfMap map[string]interface{}) *types.Snowfl
 	return apiObject
 }
 
-func expandSecretsManagerConfiguration(sc map[string]interface{}) *types.SecretsManagerConfiguration {
-	config := sc["secret_manager_configuration"].([]interface{})
-	if len(config) == 0 {
-		return nil
-	}
-
-	SecretsManagerConfig := config[0].(map[string]interface{})
-	SecretsManagerOptions := &types.SecretsManagerConfiguration{
-		Enabled: aws.Bool(SecretsManagerConfig[names.AttrEnabled].(bool)),
-	}
-
-	if v, ok := SecretsManagerConfig[names.AttrRoleARN]; ok {
-		SecretsManagerOptions.RoleARN = aws.String(v.(string))
-	}
-
-	if v, ok := SecretsManagerConfig["secret_arn"]; ok {
-		SecretsManagerOptions.SecretARN = aws.String(v.(string))
-	}
-
-	return SecretsManagerOptions
-}
-
 func expandSplunkRetryOptions(splunk map[string]interface{}) *types.SplunkRetryOptions {
 	retryOptions := &types.SplunkRetryOptions{}
 
@@ -3869,16 +3812,15 @@ func flattenHTTPEndpointDestinationDescription(description *types.HttpEndpointDe
 		return []map[string]interface{}{}
 	}
 	m := map[string]interface{}{
-		names.AttrAccessKey:            configuredAccessKey,
-		names.AttrURL:                  aws.ToString(description.EndpointConfiguration.Url),
-		names.AttrName:                 aws.ToString(description.EndpointConfiguration.Name),
-		names.AttrRoleARN:              aws.ToString(description.RoleARN),
-		"s3_backup_mode":               description.S3BackupMode,
-		"s3_configuration":             flattenS3DestinationDescription(description.S3DestinationDescription),
-		"request_configuration":        flattenHTTPEndpointRequestConfiguration(description.RequestConfiguration),
-		"cloudwatch_logging_options":   flattenCloudWatchLoggingOptions(description.CloudWatchLoggingOptions),
-		"processing_configuration":     flattenProcessingConfiguration(description.ProcessingConfiguration, destinationTypeHTTPEndpoint, aws.ToString(description.RoleARN)),
-		"secret_manager_configuration": flattenSecretsManagerConfiguration(description.SecretsManagerConfiguration),
+		names.AttrAccessKey:          configuredAccessKey,
+		names.AttrURL:                aws.ToString(description.EndpointConfiguration.Url),
+		names.AttrName:               aws.ToString(description.EndpointConfiguration.Name),
+		names.AttrRoleARN:            aws.ToString(description.RoleARN),
+		"s3_backup_mode":             description.S3BackupMode,
+		"s3_configuration":           flattenS3DestinationDescription(description.S3DestinationDescription),
+		"request_configuration":      flattenHTTPEndpointRequestConfiguration(description.RequestConfiguration),
+		"cloudwatch_logging_options": flattenCloudWatchLoggingOptions(description.CloudWatchLoggingOptions),
+		"processing_configuration":   flattenProcessingConfiguration(description.ProcessingConfiguration, destinationTypeHTTPEndpoint, aws.ToString(description.RoleARN)),
 	}
 
 	if description.RetryOptions != nil {
@@ -3919,21 +3861,6 @@ func flattenDocumentIDOptions(apiObject *types.DocumentIdOptions) map[string]int
 	return tfMap
 }
 
-func flattenSecretsManagerConfiguration(sc *types.SecretsManagerConfiguration) []interface{} {
-	if sc == nil {
-		return []interface{}{}
-	}
-
-	secretsManagerOptions := map[string]interface{}{
-		names.AttrEnabled: aws.ToBool(sc.Enabled),
-	}
-	if aws.ToBool(sc.Enabled) {
-		secretsManagerOptions[names.AttrRoleARN] = aws.ToString(sc.RoleARN)
-		secretsManagerOptions["secret_arn"] = aws.ToString(sc.SecretARN)
-	}
-	return []interface{}{secretsManagerOptions}
-}
-
 func flattenSnowflakeRoleConfiguration(apiObject *types.SnowflakeRoleConfiguration) []map[string]interface{} {
 	if apiObject == nil {
 		return []map[string]interface{}{}
diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index 034a7d2ccec..37ea1e4c3e3 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -1469,36 +1469,6 @@ func TestAccFirehoseDeliveryStream_HTTPEndpoint_retryDuration(t *testing.T) {
 	})
 }
 
-func TestAccFirehoseDeliveryStream_HTTPEndpoint_SecretsManagerConfiguration(t *testing.T) {
-	ctx := acctest.Context(t)
-	var stream types.DeliveryStreamDescription
-	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
-	resourceName := "aws_kinesis_firehose_delivery_stream.test"
-
-	resource.ParallelTest(t, resource.TestCase{
-		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
-		ErrorCheck:               acctest.ErrorCheck(t, names.FirehoseServiceID),
-		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
-		CheckDestroy:             testAccCheckDeliveryStreamDestroy_ExtendedS3(ctx),
-		Steps: []resource.TestStep{
-			{
-				Config: testAccDeliveryStreamConfig_httpEndpointSecretsManager(rName),
-				Check: resource.ComposeTestCheckFunc(
-					testAccCheckDeliveryStreamExists(ctx, resourceName, &stream),
-					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.#", acctest.Ct1),
-					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.0.enabled", acctest.CtTrue),
-					resource.TestCheckResourceAttrPair(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.0.secret_arn", "aws_secretsmanager_secret.test", names.AttrARN),
-				),
-			},
-			{
-				ResourceName:      resourceName,
-				ImportState:       true,
-				ImportStateVerify: true,
-			},
-		},
-	})
-}
-
 func TestAccFirehoseDeliveryStream_elasticSearchUpdates(t *testing.T) {
 	ctx := acctest.Context(t)
 	var stream types.DeliveryStreamDescription
@@ -4040,37 +4010,6 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 `, rName))
 }
 
-func testAccDeliveryStreamConfig_httpEndpointSecretsManager(rName string) string {
-	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
-resource "aws_secretsmanager_secret" "test" {
-  name = %[1]q
-}
-
-resource "aws_kinesis_firehose_delivery_stream" "test" {
-  depends_on  = [aws_iam_role_policy.firehose]
-  name        = %[1]q
-  destination = "http_endpoint"
-
-  http_endpoint_configuration {
-    url      = "https://input-test.com:443"
-    name     = "HTTP_test"
-    role_arn = aws_iam_role.firehose.arn
-
-    s3_configuration {
-      role_arn   = aws_iam_role.firehose.arn
-      bucket_arn = aws_s3_bucket.bucket.arn
-    }
-
-    secret_manager_configuration {
-      enabled    = true
-      role_arn   = aws_iam_role.firehose.arn
-      secret_arn = aws_secretsmanager_secret.test.arn
-    }
-  }
-}
-`, rName))
-}
-
 func testAccDeliveryStreamConfig_baseElasticsearch(rName string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_elasticsearch_domain" "test_cluster" {

From c437a847c9228bcbb1dc308b3912463a6048df3a Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 18 Jul 2024 08:39:01 -0400
Subject: [PATCH 15/20] Changes from #38245.

---
 .changelog/38245.txt                          |   3 +
 internal/service/firehose/delivery_stream.go  | 208 +++++++++---------
 .../service/firehose/delivery_stream_test.go  |  62 ++++++
 ...sis_firehose_delivery_stream.html.markdown |   5 +-
 4 files changed, 175 insertions(+), 103 deletions(-)
 create mode 100644 .changelog/38245.txt

diff --git a/.changelog/38245.txt b/.changelog/38245.txt
new file mode 100644
index 00000000000..c2a344ed459
--- /dev/null
+++ b/.changelog/38245.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_kinesis_firehose_delivery_stream: Add `secrets_manager_configuration` to `http_endpoint_configuration`
+```
\ No newline at end of file
diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index 149b71c1070..f6e158c33ab 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -192,33 +192,6 @@ func resourceDeliveryStream() *schema.Resource {
 					},
 				}
 			}
-			secretsManagerConfigurationSchema := func() *schema.Schema {
-				return &schema.Schema{
-					Type:     schema.TypeList,
-					MaxItems: 1,
-					Optional: true,
-					Computed: true,
-					Elem: &schema.Resource{
-						Schema: map[string]*schema.Schema{
-							names.AttrEnabled: {
-								Type:     schema.TypeBool,
-								Optional: true,
-								Default:  false,
-							},
-							"secret_arn": {
-								Type:         schema.TypeString,
-								Optional:     true,
-								ValidateFunc: verify.ValidARN,
-							},
-							names.AttrRoleARN: {
-								Type:         schema.TypeString,
-								Optional:     true,
-								ValidateFunc: verify.ValidARN,
-							},
-						},
-					},
-				}
-			}
 			requestConfigurationSchema := func() *schema.Schema {
 				return &schema.Schema{
 					Type:     schema.TypeList,
@@ -317,6 +290,34 @@ func resourceDeliveryStream() *schema.Resource {
 					Elem:     s3ConfigurationElem(),
 				}
 			}
+			secretsManagerConfigurationSchema := func() *schema.Schema {
+				return &schema.Schema{
+					Type:     schema.TypeList,
+					MaxItems: 1,
+					Optional: true,
+					Computed: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							names.AttrEnabled: {
+								Type:     schema.TypeBool,
+								Optional: true,
+								Computed: true,
+								ForceNew: true,
+							},
+							"secret_arn": {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ValidateFunc: verify.ValidARN,
+							},
+							names.AttrRoleARN: {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ValidateFunc: verify.ValidARN,
+							},
+						},
+					},
+				}
+			}
 
 			return map[string]*schema.Schema{
 				names.AttrARN: {
@@ -807,7 +808,8 @@ func resourceDeliveryStream() *schema.Resource {
 								Default:          types.HttpEndpointS3BackupModeFailedDataOnly,
 								ValidateDiagFunc: enum.Validate[types.HttpEndpointS3BackupMode](),
 							},
-							"s3_configuration": s3ConfigurationSchema(),
+							"s3_configuration":              s3ConfigurationSchema(),
+							"secrets_manager_configuration": secretsManagerConfigurationSchema(),
 							names.AttrURL: {
 								Type:     schema.TypeString,
 								Required: true,
@@ -2338,12 +2340,9 @@ func expandProcessorParameter(processorParameter map[string]interface{}) types.P
 
 func expandSecretsManagerConfiguration(tfMap map[string]interface{}) *types.SecretsManagerConfiguration {
 	config := tfMap["secrets_manager_configuration"].([]interface{})
+
 	if len(config) == 0 || config[0] == nil {
-		// It is possible to just pass nil here, but this seems to be the
-		// canonical form that AWS uses, and is less likely to produce diffs.
-		return &types.SecretsManagerConfiguration{
-			Enabled: aws.Bool(false),
-		}
+		return nil
 	}
 
 	secretsManagerConfiguration := config[0].(map[string]interface{})
@@ -2891,81 +2890,86 @@ func expandSplunkDestinationUpdate(splunk map[string]interface{}) *types.SplunkD
 	return configuration
 }
 
-func expandHTTPEndpointDestinationConfiguration(httpEndpoint map[string]interface{}) *types.HttpEndpointDestinationConfiguration {
-	roleARN := httpEndpoint[names.AttrRoleARN].(string)
-	configuration := &types.HttpEndpointDestinationConfiguration{
-		RetryOptions:    expandHTTPEndpointRetryOptions(httpEndpoint),
-		RoleARN:         aws.String(roleARN),
-		S3Configuration: expandS3DestinationConfiguration(httpEndpoint["s3_configuration"].([]interface{})),
+func expandHTTPEndpointDestinationConfiguration(tfMap map[string]interface{}) *types.HttpEndpointDestinationConfiguration {
+	roleARN := tfMap[names.AttrRoleARN].(string)
+	apiObject := &types.HttpEndpointDestinationConfiguration{
+		EndpointConfiguration: expandHTTPEndpointConfiguration(tfMap),
+		RetryOptions:          expandHTTPEndpointRetryOptions(tfMap),
+		RoleARN:               aws.String(roleARN),
+		S3Configuration:       expandS3DestinationConfiguration(tfMap["s3_configuration"].([]interface{})),
 	}
 
-	configuration.EndpointConfiguration = expandHTTPEndpointConfiguration(httpEndpoint)
-
 	bufferingHints := &types.HttpEndpointBufferingHints{}
-
-	if bufferingInterval, ok := httpEndpoint["buffering_interval"].(int); ok {
-		bufferingHints.IntervalInSeconds = aws.Int32(int32(bufferingInterval))
+	if v, ok := tfMap["buffering_interval"].(int); ok {
+		bufferingHints.IntervalInSeconds = aws.Int32(int32(v))
 	}
-	if bufferingSize, ok := httpEndpoint["buffering_size"].(int); ok {
-		bufferingHints.SizeInMBs = aws.Int32(int32(bufferingSize))
+	if v, ok := tfMap["buffering_size"].(int); ok {
+		bufferingHints.SizeInMBs = aws.Int32(int32(v))
 	}
-	configuration.BufferingHints = bufferingHints
+	apiObject.BufferingHints = bufferingHints
 
-	if _, ok := httpEndpoint["processing_configuration"]; ok {
-		configuration.ProcessingConfiguration = expandProcessingConfiguration(httpEndpoint, destinationTypeHTTPEndpoint, roleARN)
+	if _, ok := tfMap["cloudwatch_logging_options"]; ok {
+		apiObject.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(tfMap)
 	}
 
-	if _, ok := httpEndpoint["request_configuration"]; ok {
-		configuration.RequestConfiguration = expandHTTPEndpointRequestConfiguration(httpEndpoint)
+	if _, ok := tfMap["processing_configuration"]; ok {
+		apiObject.ProcessingConfiguration = expandProcessingConfiguration(tfMap, destinationTypeHTTPEndpoint, roleARN)
 	}
 
-	if _, ok := httpEndpoint["cloudwatch_logging_options"]; ok {
-		configuration.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(httpEndpoint)
+	if _, ok := tfMap["request_configuration"]; ok {
+		apiObject.RequestConfiguration = expandHTTPEndpointRequestConfiguration(tfMap)
 	}
-	if s3BackupMode, ok := httpEndpoint["s3_backup_mode"]; ok {
-		configuration.S3BackupMode = types.HttpEndpointS3BackupMode(s3BackupMode.(string))
+
+	if v, ok := tfMap["s3_backup_mode"]; ok {
+		apiObject.S3BackupMode = types.HttpEndpointS3BackupMode(v.(string))
 	}
 
-	return configuration
+	if _, ok := tfMap["secrets_manager_configuration"]; ok {
+		apiObject.SecretsManagerConfiguration = expandSecretsManagerConfiguration(tfMap)
+	}
+
+	return apiObject
 }
 
-func expandHTTPEndpointDestinationUpdate(httpEndpoint map[string]interface{}) *types.HttpEndpointDestinationUpdate {
-	roleARN := httpEndpoint[names.AttrRoleARN].(string)
-	configuration := &types.HttpEndpointDestinationUpdate{
-		RetryOptions: expandHTTPEndpointRetryOptions(httpEndpoint),
-		RoleARN:      aws.String(roleARN),
-		S3Update:     expandS3DestinationUpdate(httpEndpoint["s3_configuration"].([]interface{})),
+func expandHTTPEndpointDestinationUpdate(tfMap map[string]interface{}) *types.HttpEndpointDestinationUpdate {
+	roleARN := tfMap[names.AttrRoleARN].(string)
+	apiObject := &types.HttpEndpointDestinationUpdate{
+		EndpointConfiguration: expandHTTPEndpointConfiguration(tfMap),
+		RetryOptions:          expandHTTPEndpointRetryOptions(tfMap),
+		RoleARN:               aws.String(roleARN),
+		S3Update:              expandS3DestinationUpdate(tfMap["s3_configuration"].([]interface{})),
 	}
 
-	configuration.EndpointConfiguration = expandHTTPEndpointConfiguration(httpEndpoint)
-
 	bufferingHints := &types.HttpEndpointBufferingHints{}
-
-	if bufferingInterval, ok := httpEndpoint["buffering_interval"].(int); ok {
-		bufferingHints.IntervalInSeconds = aws.Int32(int32(bufferingInterval))
+	if v, ok := tfMap["buffering_interval"].(int); ok {
+		bufferingHints.IntervalInSeconds = aws.Int32(int32(v))
 	}
-	if bufferingSize, ok := httpEndpoint["buffering_size"].(int); ok {
-		bufferingHints.SizeInMBs = aws.Int32(int32(bufferingSize))
+	if v, ok := tfMap["buffering_size"].(int); ok {
+		bufferingHints.SizeInMBs = aws.Int32(int32(v))
 	}
-	configuration.BufferingHints = bufferingHints
+	apiObject.BufferingHints = bufferingHints
 
-	if _, ok := httpEndpoint["processing_configuration"]; ok {
-		configuration.ProcessingConfiguration = expandProcessingConfiguration(httpEndpoint, destinationTypeHTTPEndpoint, roleARN)
+	if _, ok := tfMap["cloudwatch_logging_options"]; ok {
+		apiObject.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(tfMap)
 	}
 
-	if _, ok := httpEndpoint["request_configuration"]; ok {
-		configuration.RequestConfiguration = expandHTTPEndpointRequestConfiguration(httpEndpoint)
+	if _, ok := tfMap["processing_configuration"]; ok {
+		apiObject.ProcessingConfiguration = expandProcessingConfiguration(tfMap, destinationTypeHTTPEndpoint, roleARN)
 	}
 
-	if _, ok := httpEndpoint["cloudwatch_logging_options"]; ok {
-		configuration.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(httpEndpoint)
+	if _, ok := tfMap["request_configuration"]; ok {
+		apiObject.RequestConfiguration = expandHTTPEndpointRequestConfiguration(tfMap)
 	}
 
-	if s3BackupMode, ok := httpEndpoint["s3_backup_mode"]; ok {
-		configuration.S3BackupMode = types.HttpEndpointS3BackupMode(s3BackupMode.(string))
+	if v, ok := tfMap["s3_backup_mode"]; ok {
+		apiObject.S3BackupMode = types.HttpEndpointS3BackupMode(v.(string))
 	}
 
-	return configuration
+	if _, ok := tfMap["secrets_manager_configuration"]; ok {
+		apiObject.SecretsManagerConfiguration = expandSecretsManagerConfiguration(tfMap)
+	}
+
+	return apiObject
 }
 
 func expandHTTPEndpointCommonAttributes(ca []interface{}) []types.HttpEndpointCommonAttribute {
@@ -3484,9 +3488,9 @@ func flattenRedshiftDestinationDescription(description *types.RedshiftDestinatio
 	return []map[string]interface{}{m}
 }
 
-func flattenSnowflakeDestinationDescription(apiObject *types.SnowflakeDestinationDescription, configuredKeyPassphrase, configuredPrivateKey string) []map[string]interface{} {
+func flattenSnowflakeDestinationDescription(apiObject *types.SnowflakeDestinationDescription, configuredKeyPassphrase, configuredPrivateKey string) []interface{} {
 	if apiObject == nil {
-		return []map[string]interface{}{}
+		return []interface{}{}
 	}
 
 	roleARN := aws.ToString(apiObject.RoleARN)
@@ -3515,7 +3519,7 @@ func flattenSnowflakeDestinationDescription(apiObject *types.SnowflakeDestinatio
 		tfMap["retry_duration"] = int(aws.ToInt32(apiObject.RetryOptions.DurationInSeconds))
 	}
 
-	return []map[string]interface{}{tfMap}
+	return []interface{}{tfMap}
 }
 
 func flattenSplunkDestinationDescription(description *types.SplunkDestinationDescription) []map[string]interface{} {
@@ -3906,32 +3910,34 @@ func flattenKinesisStreamSourceDescription(desc *types.KinesisStreamSourceDescri
 	return []interface{}{mDesc}
 }
 
-func flattenHTTPEndpointDestinationDescription(description *types.HttpEndpointDestinationDescription, configuredAccessKey string) []map[string]interface{} {
-	if description == nil {
-		return []map[string]interface{}{}
+func flattenHTTPEndpointDestinationDescription(apiObject *types.HttpEndpointDestinationDescription, configuredAccessKey string) []interface{} {
+	if apiObject == nil {
+		return []interface{}{}
 	}
-	m := map[string]interface{}{
-		names.AttrAccessKey:          configuredAccessKey,
-		names.AttrURL:                aws.ToString(description.EndpointConfiguration.Url),
-		names.AttrName:               aws.ToString(description.EndpointConfiguration.Name),
-		names.AttrRoleARN:            aws.ToString(description.RoleARN),
-		"s3_backup_mode":             description.S3BackupMode,
-		"s3_configuration":           flattenS3DestinationDescription(description.S3DestinationDescription),
-		"request_configuration":      flattenHTTPEndpointRequestConfiguration(description.RequestConfiguration),
-		"cloudwatch_logging_options": flattenCloudWatchLoggingOptions(description.CloudWatchLoggingOptions),
-		"processing_configuration":   flattenProcessingConfiguration(description.ProcessingConfiguration, destinationTypeHTTPEndpoint, aws.ToString(description.RoleARN)),
+
+	tfMap := map[string]interface{}{
+		names.AttrAccessKey:             configuredAccessKey,
+		"cloudwatch_logging_options":    flattenCloudWatchLoggingOptions(apiObject.CloudWatchLoggingOptions),
+		names.AttrName:                  aws.ToString(apiObject.EndpointConfiguration.Name),
+		"processing_configuration":      flattenProcessingConfiguration(apiObject.ProcessingConfiguration, destinationTypeHTTPEndpoint, aws.ToString(apiObject.RoleARN)),
+		"request_configuration":         flattenHTTPEndpointRequestConfiguration(apiObject.RequestConfiguration),
+		names.AttrRoleARN:               aws.ToString(apiObject.RoleARN),
+		"s3_backup_mode":                apiObject.S3BackupMode,
+		"s3_configuration":              flattenS3DestinationDescription(apiObject.S3DestinationDescription),
+		"secrets_manager_configuration": flattenSecretsManagerConfiguration(apiObject.SecretsManagerConfiguration),
+		names.AttrURL:                   aws.ToString(apiObject.EndpointConfiguration.Url),
 	}
 
-	if description.RetryOptions != nil {
-		m["retry_duration"] = int(aws.ToInt32(description.RetryOptions.DurationInSeconds))
+	if apiObject.BufferingHints != nil {
+		tfMap["buffering_interval"] = int(aws.ToInt32(apiObject.BufferingHints.IntervalInSeconds))
+		tfMap["buffering_size"] = int(aws.ToInt32(apiObject.BufferingHints.SizeInMBs))
 	}
 
-	if description.BufferingHints != nil {
-		m["buffering_interval"] = int(aws.ToInt32(description.BufferingHints.IntervalInSeconds))
-		m["buffering_size"] = int(aws.ToInt32(description.BufferingHints.SizeInMBs))
+	if apiObject.RetryOptions != nil {
+		tfMap["retry_duration"] = int(aws.ToInt32(apiObject.RetryOptions.DurationInSeconds))
 	}
 
-	return []map[string]interface{}{m}
+	return []interface{}{tfMap}
 }
 
 func expandDocumentIDOptions(tfMap map[string]interface{}) *types.DocumentIdOptions {
diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index 078bed65cea..81aeb07f862 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -1534,6 +1534,36 @@ func TestAccFirehoseDeliveryStream_HTTPEndpoint_retryDuration(t *testing.T) {
 	})
 }
 
+func TestAccFirehoseDeliveryStream_HTTPEndpoint_SecretsManagerConfiguration(t *testing.T) {
+	ctx := acctest.Context(t)
+	var stream types.DeliveryStreamDescription
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_kinesis_firehose_delivery_stream.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.FirehoseServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckDeliveryStreamDestroy_ExtendedS3(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDeliveryStreamConfig_httpEndpointSecretsManager(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDeliveryStreamExists(ctx, resourceName, &stream),
+					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.0.enabled", acctest.CtTrue),
+					resource.TestCheckResourceAttrPair(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.0.secret_arn", "aws_secretsmanager_secret.test", names.AttrARN),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccFirehoseDeliveryStream_elasticSearchUpdates(t *testing.T) {
 	ctx := acctest.Context(t)
 	var stream types.DeliveryStreamDescription
@@ -4175,6 +4205,38 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 `, rName))
 }
 
+func testAccDeliveryStreamConfig_httpEndpointSecretsManager(rName string) string {
+	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
+resource "aws_secretsmanager_secret" "test" {
+  name = %[1]q
+}
+
+resource "aws_kinesis_firehose_delivery_stream" "test" {
+  depends_on  = [aws_iam_role_policy.firehose]
+
+  name        = %[1]q
+  destination = "http_endpoint"
+
+  http_endpoint_configuration {
+    url      = "https://input-test.com:443"
+    name     = "HTTP_test"
+    role_arn = aws_iam_role.firehose.arn
+
+    s3_configuration {
+      role_arn   = aws_iam_role.firehose.arn
+      bucket_arn = aws_s3_bucket.bucket.arn
+    }
+
+    secret_manager_configuration {
+      enabled    = true
+      role_arn   = aws_iam_role.firehose.arn
+      secret_arn = aws_secretsmanager_secret.test.arn
+    }
+  }
+}
+`, rName))
+}
+
 func testAccDeliveryStreamConfig_baseElasticsearch(rName string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_elasticsearch_domain" "test_cluster" {
diff --git a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
index a2ce3655157..75efc6c4475 100644
--- a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
+++ b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
@@ -787,6 +787,7 @@ The `http_endpoint_configuration` configuration block supports the following arg
 * `processing_configuration` - (Optional) The data processing configuration.  See [`processing_configuration` block](#processing_configuration-block) below for details.
 * `request_configuration` - (Optional) The request configuration.  See [`request_configuration` block](#request_configuration-block) below for details.
 * `retry_duration` - (Optional) Total amount of seconds Firehose spends on retries. This duration starts after the initial attempt fails, It does not include the time periods during which Firehose waits for acknowledgment from the specified destination after each attempt. Valid values between `0` and `7200`. Default is `300`.
+* `secrets_manager_configuration` - (Optional) The Secret Manager Configuration. See [`secrets_manager_configuration` block](#secrets_manager_configuration-block) below for details.
 
 ### `snowflake_configuration` block
 
@@ -813,7 +814,7 @@ The `snowflake_configuration` configuration block supports the following argumen
 * `retry_duration` - (Optional) After an initial failure to deliver to Snowflake, the total amount of time, in seconds between 0 to 7200, during which Firehose re-attempts delivery (including the first attempt).  After this time has elapsed, the failed documents are written to Amazon S3.  The default value is 60s.  There will be no retry if the value is 0.
 * `s3_backup_mode` - (Optional) The S3 backup mode.
 * `s3_configuration` - (Required) The S3 configuration. See [`s3_configuration` block](#s3_configuration-block) below for details.
-* `secrets_manager_configuration` - (Optional) The SecretsManager configuration. See [`secrets_manager_configuration` block](#secrets_manager_configuration-block) below for details. This value is required if `user` and `private_key` is not provided.
+* `secrets_manager_configuration` - (Optional) The Secrets Manager configuration. See [`secrets_manager_configuration` block](#secrets_manager_configuration-block) below for details. This value is required if `user` and `private_key` are not provided.
 
 ### `cloudwatch_logging_options` block
 
@@ -932,7 +933,7 @@ The `s3_configuration` configuration block supports the following arguments:
 
 The `secrets_manager_configuration` configuration block supports the following arguments:
 
-* `enabled` - (Optional) Enables or disables the Secrets Manager configuration. Defaults to `false`.
+* `enabled` - (Optional) Enables or disables the Secrets Manager configuration.
 * `secret_arn` - (Optional) The ARN of the Secrets Manager secret. This value is required if `enabled` is true.
 * `role_arn` - (Optional) The ARN of the role the stream assumes.
 

From 81a7700709baca9d77669591728cb49602f59b24 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 18 Jul 2024 08:39:11 -0400
Subject: [PATCH 16/20] Tweak CHANGELOG entry.

---
 .changelog/38151.txt | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.changelog/38151.txt b/.changelog/38151.txt
index 5f3acb89eae..dc5c115c020 100644
--- a/.changelog/38151.txt
+++ b/.changelog/38151.txt
@@ -1,3 +1,3 @@
 ```release-note:feature
-resource/aws_kinesis_firehose_delivery_stream: Add `secrets_manager_configuariton` attribute inside `snowflake_configuration`.
+resource/aws_kinesis_firehose_delivery_stream: Add `secrets_manager_configuration` to `snowflake_configuration`
 ```

From cba13916d1224800ef364a2dfd768796f56b5d0b Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 18 Jul 2024 10:19:12 -0400
Subject: [PATCH 17/20] 'secret_manager_configuration' ->
 'secrets_manager_configuration'.

---
 internal/service/firehose/delivery_stream_test.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index 81aeb07f862..bc7a55853f8 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -1550,9 +1550,9 @@ func TestAccFirehoseDeliveryStream_HTTPEndpoint_SecretsManagerConfiguration(t *t
 				Config: testAccDeliveryStreamConfig_httpEndpointSecretsManager(rName),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckDeliveryStreamExists(ctx, resourceName, &stream),
-					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.#", acctest.Ct1),
-					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.0.enabled", acctest.CtTrue),
-					resource.TestCheckResourceAttrPair(resourceName, "http_endpoint_configuration.0.secret_manager_configuration.0.secret_arn", "aws_secretsmanager_secret.test", names.AttrARN),
+					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secrets_manager_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "http_endpoint_configuration.0.secrets_manager_configuration.0.enabled", acctest.CtTrue),
+					resource.TestCheckResourceAttrPair(resourceName, "http_endpoint_configuration.0.secrets_manager_configuration.0.secret_arn", "aws_secretsmanager_secret.test", names.AttrARN),
 				),
 			},
 			{
@@ -4227,7 +4227,7 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
       bucket_arn = aws_s3_bucket.bucket.arn
     }
 
-    secret_manager_configuration {
+    secrets_manager_configuration {
       enabled    = true
       role_arn   = aws_iam_role.firehose.arn
       secret_arn = aws_secretsmanager_secret.test.arn

From 796fcf0bb01d35ce63858079bc258e92259e03ac Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 18 Jul 2024 10:44:20 -0400
Subject: [PATCH 18/20] r/aws_kinesis_firehose_delivery_stream: Add
 `secrets_manager_configuration` to `redshift_configuration`.

---
 .changelog/38151.txt                          |   2 +-
 internal/service/firehose/delivery_stream.go  | 374 ++++++++++--------
 .../service/firehose/delivery_stream_test.go  |  60 +++
 ...sis_firehose_delivery_stream.html.markdown |   5 +-
 4 files changed, 264 insertions(+), 177 deletions(-)

diff --git a/.changelog/38151.txt b/.changelog/38151.txt
index dc5c115c020..a85d090316b 100644
--- a/.changelog/38151.txt
+++ b/.changelog/38151.txt
@@ -1,3 +1,3 @@
 ```release-note:feature
-resource/aws_kinesis_firehose_delivery_stream: Add `secrets_manager_configuration` to `snowflake_configuration`
+resource/aws_kinesis_firehose_delivery_stream: Add `secrets_manager_configuration` to `redshift_configuration` and `snowflake_configuration`
 ```
diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index f6e158c33ab..beb2b822271 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -1008,120 +1008,6 @@ func resourceDeliveryStream() *schema.Resource {
 						},
 					},
 				},
-				"snowflake_configuration": {
-					Type:     schema.TypeList,
-					Optional: true,
-					MaxItems: 1,
-					Elem: &schema.Resource{
-						Schema: map[string]*schema.Schema{
-							"account_url": {
-								Type:     schema.TypeString,
-								Required: true,
-							},
-							"cloudwatch_logging_options": cloudWatchLoggingOptionsSchema(),
-							"content_column_name": {
-								Type:         schema.TypeString,
-								Optional:     true,
-								ValidateFunc: validation.StringLenBetween(1, 255),
-							},
-							"data_loading_option": {
-								Type:             schema.TypeString,
-								Optional:         true,
-								Default:          types.SnowflakeDataLoadingOptionJsonMapping,
-								ValidateDiagFunc: enum.Validate[types.SnowflakeDataLoadingOption](),
-							},
-							names.AttrDatabase: {
-								Type:         schema.TypeString,
-								Required:     true,
-								ValidateFunc: validation.StringLenBetween(1, 255),
-							},
-							"key_passphrase": {
-								Type:         schema.TypeString,
-								Optional:     true,
-								Sensitive:    true,
-								ValidateFunc: validation.StringLenBetween(7, 255),
-							},
-							"metadata_column_name": {
-								Type:         schema.TypeString,
-								Optional:     true,
-								ValidateFunc: validation.StringLenBetween(1, 255),
-							},
-							names.AttrPrivateKey: {
-								Type:      schema.TypeString,
-								Optional:  true,
-								Sensitive: true,
-							},
-							"processing_configuration": processingConfigurationSchema(),
-							"retry_duration": {
-								Type:         schema.TypeInt,
-								Optional:     true,
-								Default:      60,
-								ValidateFunc: validation.IntBetween(0, 7200),
-							},
-							names.AttrRoleARN: {
-								Type:         schema.TypeString,
-								Required:     true,
-								ValidateFunc: verify.ValidARN,
-							},
-							"s3_backup_mode": {
-								Type:             schema.TypeString,
-								Optional:         true,
-								Default:          types.SnowflakeS3BackupModeFailedDataOnly,
-								ValidateDiagFunc: enum.Validate[types.SnowflakeS3BackupMode](),
-							},
-							"s3_configuration": s3ConfigurationSchema(),
-							names.AttrSchema: {
-								Type:         schema.TypeString,
-								Required:     true,
-								ValidateFunc: validation.StringLenBetween(1, 255),
-							},
-							"secrets_manager_configuration": secretsManagerConfigurationSchema(),
-							"snowflake_role_configuration": {
-								Type:     schema.TypeList,
-								Optional: true,
-								MaxItems: 1,
-								Elem: &schema.Resource{
-									Schema: map[string]*schema.Schema{
-										names.AttrEnabled: {
-											Type:     schema.TypeBool,
-											Optional: true,
-											Default:  false,
-										},
-										"snowflake_role": {
-											Type:         schema.TypeString,
-											Optional:     true,
-											ValidateFunc: validation.StringLenBetween(1, 255),
-										},
-									},
-								},
-								DiffSuppressFunc: verify.SuppressMissingOptionalConfigurationBlock,
-							},
-							"snowflake_vpc_configuration": {
-								Type:     schema.TypeList,
-								Optional: true,
-								MaxItems: 1,
-								Elem: &schema.Resource{
-									Schema: map[string]*schema.Schema{
-										"private_link_vpce_id": {
-											Type:     schema.TypeString,
-											Required: true,
-										},
-									},
-								},
-							},
-							"table": {
-								Type:         schema.TypeString,
-								Required:     true,
-								ValidateFunc: validation.StringLenBetween(1, 255),
-							},
-							"user": {
-								Type:         schema.TypeString,
-								Optional:     true,
-								ValidateFunc: validation.StringLenBetween(1, 255),
-							},
-						},
-					},
-				},
 				"opensearchserverless_configuration": {
 					Type:     schema.TypeList,
 					Optional: true,
@@ -1229,7 +1115,7 @@ func resourceDeliveryStream() *schema.Resource {
 							},
 							names.AttrPassword: {
 								Type:      schema.TypeString,
-								Required:  true,
+								Optional:  true,
 								Sensitive: true,
 							},
 							"processing_configuration": processingConfigurationSchema(),
@@ -1251,10 +1137,11 @@ func resourceDeliveryStream() *schema.Resource {
 								Default:          types.RedshiftS3BackupModeDisabled,
 								ValidateDiagFunc: enum.Validate[types.RedshiftS3BackupMode](),
 							},
-							"s3_configuration": s3ConfigurationSchema(),
+							"s3_configuration":              s3ConfigurationSchema(),
+							"secrets_manager_configuration": secretsManagerConfigurationSchema(),
 							names.AttrUsername: {
 								Type:     schema.TypeString,
-								Required: true,
+								Optional: true,
 							},
 						},
 					},
@@ -1288,6 +1175,120 @@ func resourceDeliveryStream() *schema.Resource {
 						},
 					},
 				},
+				"snowflake_configuration": {
+					Type:     schema.TypeList,
+					Optional: true,
+					MaxItems: 1,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"account_url": {
+								Type:     schema.TypeString,
+								Required: true,
+							},
+							"cloudwatch_logging_options": cloudWatchLoggingOptionsSchema(),
+							"content_column_name": {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ValidateFunc: validation.StringLenBetween(1, 255),
+							},
+							"data_loading_option": {
+								Type:             schema.TypeString,
+								Optional:         true,
+								Default:          types.SnowflakeDataLoadingOptionJsonMapping,
+								ValidateDiagFunc: enum.Validate[types.SnowflakeDataLoadingOption](),
+							},
+							names.AttrDatabase: {
+								Type:         schema.TypeString,
+								Required:     true,
+								ValidateFunc: validation.StringLenBetween(1, 255),
+							},
+							"key_passphrase": {
+								Type:         schema.TypeString,
+								Optional:     true,
+								Sensitive:    true,
+								ValidateFunc: validation.StringLenBetween(7, 255),
+							},
+							"metadata_column_name": {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ValidateFunc: validation.StringLenBetween(1, 255),
+							},
+							names.AttrPrivateKey: {
+								Type:      schema.TypeString,
+								Optional:  true,
+								Sensitive: true,
+							},
+							"processing_configuration": processingConfigurationSchema(),
+							"retry_duration": {
+								Type:         schema.TypeInt,
+								Optional:     true,
+								Default:      60,
+								ValidateFunc: validation.IntBetween(0, 7200),
+							},
+							names.AttrRoleARN: {
+								Type:         schema.TypeString,
+								Required:     true,
+								ValidateFunc: verify.ValidARN,
+							},
+							"s3_backup_mode": {
+								Type:             schema.TypeString,
+								Optional:         true,
+								Default:          types.SnowflakeS3BackupModeFailedDataOnly,
+								ValidateDiagFunc: enum.Validate[types.SnowflakeS3BackupMode](),
+							},
+							"s3_configuration": s3ConfigurationSchema(),
+							names.AttrSchema: {
+								Type:         schema.TypeString,
+								Required:     true,
+								ValidateFunc: validation.StringLenBetween(1, 255),
+							},
+							"secrets_manager_configuration": secretsManagerConfigurationSchema(),
+							"snowflake_role_configuration": {
+								Type:     schema.TypeList,
+								Optional: true,
+								MaxItems: 1,
+								Elem: &schema.Resource{
+									Schema: map[string]*schema.Schema{
+										names.AttrEnabled: {
+											Type:     schema.TypeBool,
+											Optional: true,
+											Default:  false,
+										},
+										"snowflake_role": {
+											Type:         schema.TypeString,
+											Optional:     true,
+											ValidateFunc: validation.StringLenBetween(1, 255),
+										},
+									},
+								},
+								DiffSuppressFunc: verify.SuppressMissingOptionalConfigurationBlock,
+							},
+							"snowflake_vpc_configuration": {
+								Type:     schema.TypeList,
+								Optional: true,
+								MaxItems: 1,
+								Elem: &schema.Resource{
+									Schema: map[string]*schema.Schema{
+										"private_link_vpce_id": {
+											Type:     schema.TypeString,
+											Required: true,
+										},
+									},
+								},
+							},
+							"table": {
+								Type:         schema.TypeString,
+								Required:     true,
+								ValidateFunc: validation.StringLenBetween(1, 255),
+							},
+							"user": {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ValidateFunc: validation.StringLenBetween(1, 255),
+							},
+						},
+					},
+				},
 				"splunk_configuration": {
 					Type:     schema.TypeList,
 					Optional: true,
@@ -2420,68 +2421,92 @@ func expandPrefix(s3 map[string]interface{}) *string {
 	return nil
 }
 
-func expandRedshiftDestinationConfiguration(redshift map[string]interface{}) *types.RedshiftDestinationConfiguration {
-	roleARN := redshift[names.AttrRoleARN].(string)
-	configuration := &types.RedshiftDestinationConfiguration{
-		ClusterJDBCURL:  aws.String(redshift["cluster_jdbcurl"].(string)),
-		RetryOptions:    expandRedshiftRetryOptions(redshift),
-		Password:        aws.String(redshift[names.AttrPassword].(string)),
-		Username:        aws.String(redshift[names.AttrUsername].(string)),
+func expandRedshiftDestinationConfiguration(tfMap map[string]interface{}) *types.RedshiftDestinationConfiguration {
+	roleARN := tfMap[names.AttrRoleARN].(string)
+	apiObject := &types.RedshiftDestinationConfiguration{
+		ClusterJDBCURL:  aws.String(tfMap["cluster_jdbcurl"].(string)),
+		CopyCommand:     expandCopyCommand(tfMap),
+		RetryOptions:    expandRedshiftRetryOptions(tfMap),
 		RoleARN:         aws.String(roleARN),
-		CopyCommand:     expandCopyCommand(redshift),
-		S3Configuration: expandS3DestinationConfiguration(redshift["s3_configuration"].([]interface{})),
+		S3Configuration: expandS3DestinationConfiguration(tfMap["s3_configuration"].([]interface{})),
 	}
 
-	if _, ok := redshift["cloudwatch_logging_options"]; ok {
-		configuration.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(redshift)
+	if _, ok := tfMap["cloudwatch_logging_options"]; ok {
+		apiObject.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(tfMap)
 	}
-	if _, ok := redshift["processing_configuration"]; ok {
-		configuration.ProcessingConfiguration = expandProcessingConfiguration(redshift, destinationTypeRedshift, roleARN)
+
+	if v, ok := tfMap[names.AttrPassword]; ok && v.(string) != "" {
+		apiObject.Password = aws.String(v.(string))
 	}
-	if s3BackupMode, ok := redshift["s3_backup_mode"]; ok {
-		configuration.S3BackupMode = types.RedshiftS3BackupMode(s3BackupMode.(string))
-		configuration.S3BackupConfiguration = expandS3DestinationConfigurationBackup(redshift)
+
+	if _, ok := tfMap["processing_configuration"]; ok {
+		apiObject.ProcessingConfiguration = expandProcessingConfiguration(tfMap, destinationTypeRedshift, roleARN)
 	}
 
-	return configuration
+	if v, ok := tfMap["s3_backup_mode"]; ok {
+		apiObject.S3BackupMode = types.RedshiftS3BackupMode(v.(string))
+		apiObject.S3BackupConfiguration = expandS3DestinationConfigurationBackup(tfMap)
+	}
+
+	if _, ok := tfMap["secrets_manager_configuration"]; ok {
+		apiObject.SecretsManagerConfiguration = expandSecretsManagerConfiguration(tfMap)
+	}
+
+	if v, ok := tfMap[names.AttrUsername]; ok && v.(string) != "" {
+		apiObject.Username = aws.String(v.(string))
+	}
+
+	return apiObject
 }
 
-func expandRedshiftDestinationUpdate(redshift map[string]interface{}) *types.RedshiftDestinationUpdate {
-	roleARN := redshift[names.AttrRoleARN].(string)
-	configuration := &types.RedshiftDestinationUpdate{
-		ClusterJDBCURL: aws.String(redshift["cluster_jdbcurl"].(string)),
-		RetryOptions:   expandRedshiftRetryOptions(redshift),
-		Password:       aws.String(redshift[names.AttrPassword].(string)),
-		Username:       aws.String(redshift[names.AttrUsername].(string)),
+func expandRedshiftDestinationUpdate(tfMap map[string]interface{}) *types.RedshiftDestinationUpdate {
+	roleARN := tfMap[names.AttrRoleARN].(string)
+	apiObject := &types.RedshiftDestinationUpdate{
+		ClusterJDBCURL: aws.String(tfMap["cluster_jdbcurl"].(string)),
+		CopyCommand:    expandCopyCommand(tfMap),
+		RetryOptions:   expandRedshiftRetryOptions(tfMap),
 		RoleARN:        aws.String(roleARN),
-		CopyCommand:    expandCopyCommand(redshift),
 	}
 
-	s3Config := expandS3DestinationUpdate(redshift["s3_configuration"].([]interface{}))
+	s3Config := expandS3DestinationUpdate(tfMap["s3_configuration"].([]interface{}))
 	// Redshift does not currently support ErrorOutputPrefix,
 	// which is set to the empty string within "updateS3Config",
 	// thus we must remove it here to avoid an InvalidArgumentException.
 	s3Config.ErrorOutputPrefix = nil
-	configuration.S3Update = s3Config
+	apiObject.S3Update = s3Config
 
-	if _, ok := redshift["cloudwatch_logging_options"]; ok {
-		configuration.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(redshift)
+	if _, ok := tfMap["cloudwatch_logging_options"]; ok {
+		apiObject.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(tfMap)
 	}
-	if _, ok := redshift["processing_configuration"]; ok {
-		configuration.ProcessingConfiguration = expandProcessingConfiguration(redshift, destinationTypeRedshift, roleARN)
+
+	if v, ok := tfMap[names.AttrPassword]; ok && v.(string) != "" {
+		apiObject.Password = aws.String(v.(string))
 	}
-	if s3BackupMode, ok := redshift["s3_backup_mode"]; ok {
-		configuration.S3BackupMode = types.RedshiftS3BackupMode(s3BackupMode.(string))
-		configuration.S3BackupUpdate = expandS3DestinationUpdateBackup(redshift)
-		if configuration.S3BackupUpdate != nil {
+
+	if _, ok := tfMap["processing_configuration"]; ok {
+		apiObject.ProcessingConfiguration = expandProcessingConfiguration(tfMap, destinationTypeRedshift, roleARN)
+	}
+
+	if s3BackupMode, ok := tfMap["s3_backup_mode"]; ok {
+		apiObject.S3BackupMode = types.RedshiftS3BackupMode(s3BackupMode.(string))
+		apiObject.S3BackupUpdate = expandS3DestinationUpdateBackup(tfMap)
+		if apiObject.S3BackupUpdate != nil {
 			// Redshift does not currently support ErrorOutputPrefix,
 			// which is set to the empty string within "updateS3BackupConfig",
 			// thus we must remove it here to avoid an InvalidArgumentException.
-			configuration.S3BackupUpdate.ErrorOutputPrefix = nil
+			apiObject.S3BackupUpdate.ErrorOutputPrefix = nil
 		}
 	}
 
-	return configuration
+	if _, ok := tfMap["secrets_manager_configuration"]; ok {
+		apiObject.SecretsManagerConfiguration = expandSecretsManagerConfiguration(tfMap)
+	}
+
+	if v, ok := tfMap[names.AttrUsername]; ok && v.(string) != "" {
+		apiObject.Username = aws.String(v.(string))
+	}
+
+	return apiObject
 }
 
 func expandElasticsearchDestinationConfiguration(es map[string]interface{}) *types.ElasticsearchDestinationConfiguration {
@@ -3458,34 +3483,35 @@ func flattenExtendedS3DestinationDescription(description *types.ExtendedS3Destin
 	return []map[string]interface{}{m}
 }
 
-func flattenRedshiftDestinationDescription(description *types.RedshiftDestinationDescription, configuredPassword string) []map[string]interface{} {
-	if description == nil {
-		return []map[string]interface{}{}
+func flattenRedshiftDestinationDescription(apiObject *types.RedshiftDestinationDescription, configuredPassword string) []interface{} {
+	if apiObject == nil {
+		return []interface{}{}
 	}
 
-	m := map[string]interface{}{
-		"cloudwatch_logging_options": flattenCloudWatchLoggingOptions(description.CloudWatchLoggingOptions),
-		"cluster_jdbcurl":            aws.ToString(description.ClusterJDBCURL),
-		names.AttrPassword:           configuredPassword,
-		"processing_configuration":   flattenProcessingConfiguration(description.ProcessingConfiguration, destinationTypeRedshift, aws.ToString(description.RoleARN)),
-		names.AttrRoleARN:            aws.ToString(description.RoleARN),
-		"s3_backup_configuration":    flattenS3DestinationDescription(description.S3BackupDescription),
-		"s3_backup_mode":             description.S3BackupMode,
-		"s3_configuration":           flattenS3DestinationDescription(description.S3DestinationDescription),
-		names.AttrUsername:           aws.ToString(description.Username),
+	tfMap := map[string]interface{}{
+		"cloudwatch_logging_options":    flattenCloudWatchLoggingOptions(apiObject.CloudWatchLoggingOptions),
+		"cluster_jdbcurl":               aws.ToString(apiObject.ClusterJDBCURL),
+		names.AttrPassword:              configuredPassword,
+		"processing_configuration":      flattenProcessingConfiguration(apiObject.ProcessingConfiguration, destinationTypeRedshift, aws.ToString(apiObject.RoleARN)),
+		names.AttrRoleARN:               aws.ToString(apiObject.RoleARN),
+		"s3_backup_configuration":       flattenS3DestinationDescription(apiObject.S3BackupDescription),
+		"s3_backup_mode":                apiObject.S3BackupMode,
+		"s3_configuration":              flattenS3DestinationDescription(apiObject.S3DestinationDescription),
+		"secrets_manager_configuration": flattenSecretsManagerConfiguration(apiObject.SecretsManagerConfiguration),
+		names.AttrUsername:              aws.ToString(apiObject.Username),
 	}
 
-	if description.CopyCommand != nil {
-		m["copy_options"] = aws.ToString(description.CopyCommand.CopyOptions)
-		m["data_table_columns"] = aws.ToString(description.CopyCommand.DataTableColumns)
-		m["data_table_name"] = aws.ToString(description.CopyCommand.DataTableName)
+	if apiObject.CopyCommand != nil {
+		tfMap["copy_options"] = aws.ToString(apiObject.CopyCommand.CopyOptions)
+		tfMap["data_table_columns"] = aws.ToString(apiObject.CopyCommand.DataTableColumns)
+		tfMap["data_table_name"] = aws.ToString(apiObject.CopyCommand.DataTableName)
 	}
 
-	if description.RetryOptions != nil {
-		m["retry_duration"] = int(aws.ToInt32(description.RetryOptions.DurationInSeconds))
+	if apiObject.RetryOptions != nil {
+		tfMap["retry_duration"] = aws.ToInt32(apiObject.RetryOptions.DurationInSeconds)
 	}
 
-	return []map[string]interface{}{m}
+	return []interface{}{tfMap}
 }
 
 func flattenSnowflakeDestinationDescription(apiObject *types.SnowflakeDestinationDescription, configuredKeyPassphrase, configuredPrivateKey string) []interface{} {
diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index bc7a55853f8..9dd0420da63 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -1062,6 +1062,36 @@ func TestAccFirehoseDeliveryStream_redshiftUpdates(t *testing.T) {
 	})
 }
 
+func TestAccFirehoseDeliveryStream_Redshift_SecretsManagerConfiguration(t *testing.T) {
+	ctx := acctest.Context(t)
+	var stream types.DeliveryStreamDescription
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_kinesis_firehose_delivery_stream.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.FirehoseServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckDeliveryStreamDestroy(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDeliveryStreamConfig_redshiftSecretsManager(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDeliveryStreamExists(ctx, resourceName, &stream),
+					resource.TestCheckResourceAttr(resourceName, "redshift_configuration.0.secrets_manager_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "redshift_configuration.0.secrets_manager_configuration.0.enabled", acctest.CtTrue),
+					resource.TestCheckResourceAttrPair(resourceName, "redshift_configuration.0.secrets_manager_configuration.0.secret_arn", "aws_secretsmanager_secret.test", names.AttrARN),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccFirehoseDeliveryStream_snowflakeUpdates(t *testing.T) {
 	ctx := acctest.Context(t)
 	var stream types.DeliveryStreamDescription
@@ -3871,6 +3901,36 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 `, rName))
 }
 
+func testAccDeliveryStreamConfig_redshiftSecretsManager(rName string) string {
+	return acctest.ConfigCompose(testAccDeliveryStreamConfig_baseRedshift(rName), fmt.Sprintf(`
+resource "aws_secretsmanager_secret" "test" {
+  name = %[1]q
+}
+
+resource "aws_kinesis_firehose_delivery_stream" "test" {
+  name        = %[1]q
+  destination = "redshift"
+
+  redshift_configuration {
+    role_arn        = aws_iam_role.firehose.arn
+    cluster_jdbcurl = "jdbc:redshift://${aws_redshift_cluster.test.endpoint}/${aws_redshift_cluster.test.database_name}"
+    data_table_name = "test-table"
+
+    s3_configuration {
+      role_arn   = aws_iam_role.firehose.arn
+      bucket_arn = aws_s3_bucket.bucket.arn
+    }
+
+    secrets_manager_configuration {
+      enabled    = true
+      role_arn   = aws_iam_role.firehose.arn
+      secret_arn = aws_secretsmanager_secret.test.arn
+    }
+  }
+}
+`, rName))
+}
+
 func testAccDeliveryStreamConfig_snowflakeBasic(rName, privateKey string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_kinesis_firehose_delivery_stream" "test" {
diff --git a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
index 75efc6c4475..334e17d485f 100644
--- a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
+++ b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
@@ -687,13 +687,14 @@ The `extended_s3_configuration` configuration block supports the same fields fro
 The `redshift_configuration` configuration block supports the following arguments:
 
 * `cluster_jdbcurl` - (Required) The jdbcurl of the redshift cluster.
-* `username` - (Required) The username that the firehose delivery stream will assume. It is strongly recommended that the username and password provided is used exclusively for Amazon Kinesis Firehose purposes, and that the permissions for the account are restricted for Amazon Redshift INSERT permissions.
-* `password` - (Required) The password for the username above.
+* `username` - (Optional) The username that the firehose delivery stream will assume. It is strongly recommended that the username and password provided is used exclusively for Amazon Kinesis Firehose purposes, and that the permissions for the account are restricted for Amazon Redshift INSERT permissions. This value is required if `secrets_manager_configuration` is not provided.
+* `password` - (Optional) The password for the username above. This value is required if `secrets_manager_configuration` is not provided.
 * `retry_duration` - (Optional) The length of time during which Firehose retries delivery after a failure, starting from the initial request and including the first attempt. The default value is 3600 seconds (60 minutes). Firehose does not retry if the value of DurationInSeconds is 0 (zero) or if the first delivery attempt takes longer than the current value.
 * `role_arn` - (Required) The arn of the role the stream assumes.
 * `s3_configuration` - (Required) The S3 Configuration. See [s3_configuration](#s3_configuration-block) below for details.
 * `s3_backup_mode` - (Optional) The Amazon S3 backup mode.  Valid values are `Disabled` and `Enabled`.  Default value is `Disabled`.
 * `s3_backup_configuration` - (Optional) The configuration for backup in Amazon S3. Required if `s3_backup_mode` is `Enabled`. Supports the same fields as `s3_configuration` object.
+`secrets_manager_configuration` - (Optional) The Secrets Manager configuration. See [`secrets_manager_configuration` block](#secrets_manager_configuration-block) below for details. This value is required if `username` and `password` are not provided.
 * `data_table_name` - (Required) The name of the table in the redshift cluster that the s3 bucket will copy to.
 * `copy_options` - (Optional) Copy options for copying the data from the s3 intermediate bucket into redshift, for example to change the default delimiter. For valid values, see the [AWS documentation](http://docs.aws.amazon.com/firehose/latest/APIReference/API_CopyCommand.html)
 * `data_table_columns` - (Optional) The data table columns that will be targeted by the copy command.

From 9adb82d928c8ff4488cca2cff12cabfdcc5e4e73 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 18 Jul 2024 11:11:11 -0400
Subject: [PATCH 19/20] r/aws_kinesis_firehose_delivery_stream: Add
 `secrets_manager_configuration` to `splunk_configuration`.

---
 .changelog/38151.txt                          |   2 +-
 internal/service/firehose/delivery_stream.go  | 137 ++++++++++--------
 .../service/firehose/delivery_stream_test.go  |  59 ++++++++
 ...sis_firehose_delivery_stream.html.markdown |   3 +-
 4 files changed, 139 insertions(+), 62 deletions(-)

diff --git a/.changelog/38151.txt b/.changelog/38151.txt
index a85d090316b..b1e40930c85 100644
--- a/.changelog/38151.txt
+++ b/.changelog/38151.txt
@@ -1,3 +1,3 @@
 ```release-note:feature
-resource/aws_kinesis_firehose_delivery_stream: Add `secrets_manager_configuration` to `redshift_configuration` and `snowflake_configuration`
+resource/aws_kinesis_firehose_delivery_stream: Add `secrets_manager_configuration` to `redshift_configuration`, `snowflake_configuration`, and `splunk_configuration`
 ```
diff --git a/internal/service/firehose/delivery_stream.go b/internal/service/firehose/delivery_stream.go
index beb2b822271..dcc056e5fad 100644
--- a/internal/service/firehose/delivery_stream.go
+++ b/internal/service/firehose/delivery_stream.go
@@ -1326,7 +1326,7 @@ func resourceDeliveryStream() *schema.Resource {
 							},
 							"hec_token": {
 								Type:     schema.TypeString,
-								Required: true,
+								Optional: true,
 							},
 							"processing_configuration": processingConfigurationSchema(),
 							"retry_duration": {
@@ -1341,7 +1341,8 @@ func resourceDeliveryStream() *schema.Resource {
 								Default:          types.SplunkS3BackupModeFailedEventsOnly,
 								ValidateDiagFunc: enum.Validate[types.SplunkS3BackupMode](),
 							},
-							"s3_configuration": s3ConfigurationSchema(),
+							"s3_configuration":              s3ConfigurationSchema(),
+							"secrets_manager_configuration": secretsManagerConfigurationSchema(),
 						},
 					},
 				},
@@ -2487,8 +2488,8 @@ func expandRedshiftDestinationUpdate(tfMap map[string]interface{}) *types.Redshi
 		apiObject.ProcessingConfiguration = expandProcessingConfiguration(tfMap, destinationTypeRedshift, roleARN)
 	}
 
-	if s3BackupMode, ok := tfMap["s3_backup_mode"]; ok {
-		apiObject.S3BackupMode = types.RedshiftS3BackupMode(s3BackupMode.(string))
+	if v, ok := tfMap["s3_backup_mode"]; ok {
+		apiObject.S3BackupMode = types.RedshiftS3BackupMode(v.(string))
 		apiObject.S3BackupUpdate = expandS3DestinationUpdateBackup(tfMap)
 		if apiObject.S3BackupUpdate != nil {
 			// Redshift does not currently support ErrorOutputPrefix,
@@ -2847,72 +2848,86 @@ func expandSnowflakeDestinationUpdate(tfMap map[string]interface{}) *types.Snowf
 	return apiObject
 }
 
-func expandSplunkDestinationConfiguration(splunk map[string]interface{}) *types.SplunkDestinationConfiguration {
-	configuration := &types.SplunkDestinationConfiguration{
-		HECToken:                          aws.String(splunk["hec_token"].(string)),
-		HECEndpointType:                   types.HECEndpointType(splunk["hec_endpoint_type"].(string)),
-		HECEndpoint:                       aws.String(splunk["hec_endpoint"].(string)),
-		HECAcknowledgmentTimeoutInSeconds: aws.Int32(int32(splunk["hec_acknowledgment_timeout"].(int))),
-		RetryOptions:                      expandSplunkRetryOptions(splunk),
-		S3Configuration:                   expandS3DestinationConfiguration(splunk["s3_configuration"].([]interface{})),
+func expandSplunkDestinationConfiguration(tfMap map[string]interface{}) *types.SplunkDestinationConfiguration {
+	apiObject := &types.SplunkDestinationConfiguration{
+		HECAcknowledgmentTimeoutInSeconds: aws.Int32(int32(tfMap["hec_acknowledgment_timeout"].(int))),
+		HECEndpoint:                       aws.String(tfMap["hec_endpoint"].(string)),
+		HECEndpointType:                   types.HECEndpointType(tfMap["hec_endpoint_type"].(string)),
+		RetryOptions:                      expandSplunkRetryOptions(tfMap),
+		S3Configuration:                   expandS3DestinationConfiguration(tfMap["s3_configuration"].([]interface{})),
 	}
 
 	bufferingHints := &types.SplunkBufferingHints{}
-
-	if bufferingInterval, ok := splunk["buffering_interval"].(int); ok {
+	if bufferingInterval, ok := tfMap["buffering_interval"].(int); ok {
 		bufferingHints.IntervalInSeconds = aws.Int32(int32(bufferingInterval))
 	}
-	if bufferingSize, ok := splunk["buffering_size"].(int); ok {
+	if bufferingSize, ok := tfMap["buffering_size"].(int); ok {
 		bufferingHints.SizeInMBs = aws.Int32(int32(bufferingSize))
 	}
-	configuration.BufferingHints = bufferingHints
+	apiObject.BufferingHints = bufferingHints
 
-	if _, ok := splunk["processing_configuration"]; ok {
-		configuration.ProcessingConfiguration = expandProcessingConfiguration(splunk, destinationTypeSplunk, "")
+	if _, ok := tfMap["cloudwatch_logging_options"]; ok {
+		apiObject.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(tfMap)
 	}
 
-	if _, ok := splunk["cloudwatch_logging_options"]; ok {
-		configuration.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(splunk)
+	if v, ok := tfMap["hec_token"]; ok && v.(string) != "" {
+		apiObject.HECToken = aws.String(v.(string))
 	}
-	if s3BackupMode, ok := splunk["s3_backup_mode"]; ok {
-		configuration.S3BackupMode = types.SplunkS3BackupMode(s3BackupMode.(string))
+
+	if _, ok := tfMap["processing_configuration"]; ok {
+		apiObject.ProcessingConfiguration = expandProcessingConfiguration(tfMap, destinationTypeSplunk, "")
 	}
 
-	return configuration
+	if v, ok := tfMap["s3_backup_mode"]; ok {
+		apiObject.S3BackupMode = types.SplunkS3BackupMode(v.(string))
+	}
+
+	if _, ok := tfMap["secrets_manager_configuration"]; ok {
+		apiObject.SecretsManagerConfiguration = expandSecretsManagerConfiguration(tfMap)
+	}
+
+	return apiObject
 }
 
-func expandSplunkDestinationUpdate(splunk map[string]interface{}) *types.SplunkDestinationUpdate {
-	configuration := &types.SplunkDestinationUpdate{
-		HECToken:                          aws.String(splunk["hec_token"].(string)),
-		HECEndpointType:                   types.HECEndpointType(splunk["hec_endpoint_type"].(string)),
-		HECEndpoint:                       aws.String(splunk["hec_endpoint"].(string)),
-		HECAcknowledgmentTimeoutInSeconds: aws.Int32(int32(splunk["hec_acknowledgment_timeout"].(int))),
-		RetryOptions:                      expandSplunkRetryOptions(splunk),
-		S3Update:                          expandS3DestinationUpdate(splunk["s3_configuration"].([]interface{})),
+func expandSplunkDestinationUpdate(tfMap map[string]interface{}) *types.SplunkDestinationUpdate {
+	apiObject := &types.SplunkDestinationUpdate{
+		HECAcknowledgmentTimeoutInSeconds: aws.Int32(int32(tfMap["hec_acknowledgment_timeout"].(int))),
+		HECEndpoint:                       aws.String(tfMap["hec_endpoint"].(string)),
+		HECEndpointType:                   types.HECEndpointType(tfMap["hec_endpoint_type"].(string)),
+		RetryOptions:                      expandSplunkRetryOptions(tfMap),
+		S3Update:                          expandS3DestinationUpdate(tfMap["s3_configuration"].([]interface{})),
 	}
 
 	bufferingHints := &types.SplunkBufferingHints{}
-
-	if bufferingInterval, ok := splunk["buffering_interval"].(int); ok {
+	if bufferingInterval, ok := tfMap["buffering_interval"].(int); ok {
 		bufferingHints.IntervalInSeconds = aws.Int32(int32(bufferingInterval))
 	}
-	if bufferingSize, ok := splunk["buffering_size"].(int); ok {
+	if bufferingSize, ok := tfMap["buffering_size"].(int); ok {
 		bufferingHints.SizeInMBs = aws.Int32(int32(bufferingSize))
 	}
-	configuration.BufferingHints = bufferingHints
+	apiObject.BufferingHints = bufferingHints
+
+	if _, ok := tfMap["cloudwatch_logging_options"]; ok {
+		apiObject.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(tfMap)
+	}
 
-	if _, ok := splunk["processing_configuration"]; ok {
-		configuration.ProcessingConfiguration = expandProcessingConfiguration(splunk, destinationTypeSplunk, "")
+	if v, ok := tfMap["hec_token"]; ok && v.(string) != "" {
+		apiObject.HECToken = aws.String(v.(string))
 	}
 
-	if _, ok := splunk["cloudwatch_logging_options"]; ok {
-		configuration.CloudWatchLoggingOptions = expandCloudWatchLoggingOptions(splunk)
+	if _, ok := tfMap["processing_configuration"]; ok {
+		apiObject.ProcessingConfiguration = expandProcessingConfiguration(tfMap, destinationTypeSplunk, "")
 	}
-	if s3BackupMode, ok := splunk["s3_backup_mode"]; ok {
-		configuration.S3BackupMode = types.SplunkS3BackupMode(s3BackupMode.(string))
+
+	if v, ok := tfMap["s3_backup_mode"]; ok {
+		apiObject.S3BackupMode = types.SplunkS3BackupMode(v.(string))
 	}
 
-	return configuration
+	if _, ok := tfMap["secrets_manager_configuration"]; ok {
+		apiObject.SecretsManagerConfiguration = expandSecretsManagerConfiguration(tfMap)
+	}
+
+	return apiObject
 }
 
 func expandHTTPEndpointDestinationConfiguration(tfMap map[string]interface{}) *types.HttpEndpointDestinationConfiguration {
@@ -3548,31 +3563,33 @@ func flattenSnowflakeDestinationDescription(apiObject *types.SnowflakeDestinatio
 	return []interface{}{tfMap}
 }
 
-func flattenSplunkDestinationDescription(description *types.SplunkDestinationDescription) []map[string]interface{} {
-	if description == nil {
-		return []map[string]interface{}{}
+func flattenSplunkDestinationDescription(apiObject *types.SplunkDestinationDescription) []interface{} {
+	if apiObject == nil {
+		return []interface{}{}
 	}
-	m := map[string]interface{}{
-		"cloudwatch_logging_options": flattenCloudWatchLoggingOptions(description.CloudWatchLoggingOptions),
-		"hec_acknowledgment_timeout": int(aws.ToInt32(description.HECAcknowledgmentTimeoutInSeconds)),
-		"hec_endpoint_type":          description.HECEndpointType,
-		"hec_endpoint":               aws.ToString(description.HECEndpoint),
-		"hec_token":                  aws.ToString(description.HECToken),
-		"processing_configuration":   flattenProcessingConfiguration(description.ProcessingConfiguration, destinationTypeSplunk, ""),
-		"s3_backup_mode":             description.S3BackupMode,
-		"s3_configuration":           flattenS3DestinationDescription(description.S3DestinationDescription),
+
+	tfMap := map[string]interface{}{
+		"cloudwatch_logging_options":    flattenCloudWatchLoggingOptions(apiObject.CloudWatchLoggingOptions),
+		"hec_acknowledgment_timeout":    int(aws.ToInt32(apiObject.HECAcknowledgmentTimeoutInSeconds)),
+		"hec_endpoint":                  aws.ToString(apiObject.HECEndpoint),
+		"hec_endpoint_type":             apiObject.HECEndpointType,
+		"hec_token":                     aws.ToString(apiObject.HECToken),
+		"processing_configuration":      flattenProcessingConfiguration(apiObject.ProcessingConfiguration, destinationTypeSplunk, ""),
+		"s3_backup_mode":                apiObject.S3BackupMode,
+		"s3_configuration":              flattenS3DestinationDescription(apiObject.S3DestinationDescription),
+		"secrets_manager_configuration": flattenSecretsManagerConfiguration(apiObject.SecretsManagerConfiguration),
 	}
 
-	if description.BufferingHints != nil {
-		m["buffering_interval"] = int(aws.ToInt32(description.BufferingHints.IntervalInSeconds))
-		m["buffering_size"] = int(aws.ToInt32(description.BufferingHints.SizeInMBs))
+	if apiObject.BufferingHints != nil {
+		tfMap["buffering_interval"] = int(aws.ToInt32(apiObject.BufferingHints.IntervalInSeconds))
+		tfMap["buffering_size"] = int(aws.ToInt32(apiObject.BufferingHints.SizeInMBs))
 	}
 
-	if description.RetryOptions != nil {
-		m["retry_duration"] = int(aws.ToInt32(description.RetryOptions.DurationInSeconds))
+	if apiObject.RetryOptions != nil {
+		tfMap["retry_duration"] = int(aws.ToInt32(apiObject.RetryOptions.DurationInSeconds))
 	}
 
-	return []map[string]interface{}{m}
+	return []interface{}{tfMap}
 }
 
 func flattenS3DestinationDescription(description *types.S3DestinationDescription) []map[string]interface{} {
diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index 9dd0420da63..b0683e1bba4 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -1424,6 +1424,36 @@ func TestAccFirehoseDeliveryStream_Splunk_ErrorOutputPrefix(t *testing.T) {
 	})
 }
 
+func TestAccFirehoseDeliveryStream_Splunk_SecretsManagerConfiguration(t *testing.T) {
+	ctx := acctest.Context(t)
+	var stream types.DeliveryStreamDescription
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_kinesis_firehose_delivery_stream.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.FirehoseServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckDeliveryStreamDestroy_ExtendedS3(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDeliveryStreamConfig_splunkSecretsManager(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDeliveryStreamExists(ctx, resourceName, &stream),
+					resource.TestCheckResourceAttr(resourceName, "splunk_configuration.0.secrets_manager_configuration.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "splunk_configuration.0.secrets_manager_configuration.0.enabled", acctest.CtTrue),
+					resource.TestCheckResourceAttrPair(resourceName, "splunk_configuration.0.secrets_manager_configuration.0.secret_arn", "aws_secretsmanager_secret.test", names.AttrARN),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccFirehoseDeliveryStream_httpEndpoint(t *testing.T) {
 	ctx := acctest.Context(t)
 	var stream types.DeliveryStreamDescription
@@ -4127,6 +4157,35 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 `, rName))
 }
 
+func testAccDeliveryStreamConfig_splunkSecretsManager(rName string) string {
+	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
+resource "aws_secretsmanager_secret" "test" {
+  name = %[1]q
+}
+
+resource "aws_kinesis_firehose_delivery_stream" "test" {
+  depends_on  = [aws_iam_role_policy.firehose]
+  name        = %[1]q
+  destination = "splunk"
+
+  splunk_configuration {
+    hec_endpoint = "https://input-test.com:443"
+
+    s3_configuration {
+      role_arn   = aws_iam_role.firehose.arn
+      bucket_arn = aws_s3_bucket.bucket.arn
+    }
+
+    secrets_manager_configuration {
+      enabled    = true
+      role_arn   = aws_iam_role.firehose.arn
+      secret_arn = aws_secretsmanager_secret.test.arn
+    }
+  }
+}
+`, rName))
+}
+
 func testAccDeliveryStreamConfig_splunkErrorOutputPrefix(rName, errorOutputPrefix string) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_kinesis_firehose_delivery_stream" "test" {
diff --git a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
index 334e17d485f..5f9b587f4cb 100644
--- a/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
+++ b/website/docs/r/kinesis_firehose_delivery_stream.html.markdown
@@ -765,9 +765,10 @@ The `splunk_configuration` configuration block supports the following arguments:
 * `hec_acknowledgment_timeout` - (Optional) The amount of time, in seconds between 180 and 600, that Kinesis Firehose waits to receive an acknowledgment from Splunk after it sends it data.
 * `hec_endpoint` - (Required) The HTTP Event Collector (HEC) endpoint to which Kinesis Firehose sends your data.
 * `hec_endpoint_type` - (Optional) The HEC endpoint type. Valid values are `Raw` or `Event`. The default value is `Raw`.
-* `hec_token` - (Required) The GUID that you obtain from your Splunk cluster when you create a new HEC endpoint.
+* `hec_token` - (Optional) The GUID that you obtain from your Splunk cluster when you create a new HEC endpoint. This value is required if `secrets_manager_configuration` is not provided.
 * `s3_configuration` - (Required) The S3 Configuration. See [`s3_configuration` block](#s3_configuration-block) below for details.
 * `s3_backup_mode` - (Optional) Defines how documents should be delivered to Amazon S3.  Valid values are `FailedEventsOnly` and `AllEvents`.  Default value is `FailedEventsOnly`.
+`secrets_manager_configuration` - (Optional) The Secrets Manager configuration. See [`secrets_manager_configuration` block](#secrets_manager_configuration-block) below for details. This value is required if `hec_token` is not provided.
 * `retry_duration` - (Optional) After an initial failure to deliver to Splunk, the total amount of time, in seconds between 0 to 7200, during which Firehose re-attempts delivery (including the first attempt).  After this time has elapsed, the failed documents are written to Amazon S3.  The default value is 300s.  There will be no retry if the value is 0.
 * `cloudwatch_logging_options` - (Optional) The CloudWatch Logging Options for the delivery stream. See [`cloudwatch_logging_options` block](#cloudwatch_logging_options-block) below for details.
 * `processing_configuration` - (Optional) The data processing configuration.  See [`processing_configuration` block](#processing_configuration-block) below for details.

From 85e2fbc90684f8ac5d38973ae0d7ec6b048aab4d Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 18 Jul 2024 11:31:15 -0400
Subject: [PATCH 20/20] Fix terrafmt error.

---
 internal/service/firehose/delivery_stream_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/internal/service/firehose/delivery_stream_test.go b/internal/service/firehose/delivery_stream_test.go
index b0683e1bba4..fdf8fe160e9 100644
--- a/internal/service/firehose/delivery_stream_test.go
+++ b/internal/service/firehose/delivery_stream_test.go
@@ -4331,7 +4331,7 @@ resource "aws_secretsmanager_secret" "test" {
 }
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
-  depends_on  = [aws_iam_role_policy.firehose]
+  depends_on = [aws_iam_role_policy.firehose]
 
   name        = %[1]q
   destination = "http_endpoint"