Merge pull request #44540 from tabito-hara/f-aws_lambda_event_source_mapping-add_schema_registry_config

[Enhancement] r/aws_lambda_event_source_mapping: Add `schema_registry_config` to both MSK and self-managed Kafka event sources

Author: ewbankkit

.changelog/44540.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_lambda_event_source_mapping: Add `schema_registry_config` configuration blocks to `amazon_managed_kafka_event_source_config` and `self_managed_kafka_event_source_config` blocks
+```

internal/service/lambda/event_source_mapping.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/YakDriver/regexache"
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/aws/arn"
 	"github.com/aws/aws-sdk-go-v2/service/lambda"
@@ -61,6 +62,7 @@ func resourceEventSourceMapping() *schema.Resource {
 							ForceNew:     true,
 							ValidateFunc: validation.StringLenBetween(1, 200),
 						},
+						"schema_registry_config": kafkaSchemaRegistryConfigSchema(),
 					},
 				},
 			},
@@ -352,6 +354,7 @@ func resourceEventSourceMapping() *schema.Resource {
 							ForceNew:     true,
 							ValidateFunc: validation.StringLenBetween(1, 200),
 						},
+						"schema_registry_config": kafkaSchemaRegistryConfigSchema(),
 					},
 				},
 			},
@@ -417,6 +420,70 @@ func resourceEventSourceMapping() *schema.Resource {
 	}
 }
 
+func kafkaSchemaRegistryConfigSchema() *schema.Schema {
+	return &schema.Schema{
+		Type:     schema.TypeList,
+		Optional: true,
+		MaxItems: 1,
+		ForceNew: true,
+		Elem: &schema.Resource{
+			Schema: map[string]*schema.Schema{
+				"access_config": {
+					Type:     schema.TypeSet,
+					Optional: true,
+					ForceNew: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							names.AttrType: {
+								Type:             schema.TypeString,
+								Optional:         true,
+								ForceNew:         true,
+								ValidateDiagFunc: enum.Validate[awstypes.KafkaSchemaRegistryAuthType](),
+							},
+							names.AttrURI: {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ForceNew:     true,
+								ValidateFunc: verify.ValidARN,
+							},
+						},
+					},
+				},
+				"event_record_format": {
+					Type:             schema.TypeString,
+					Optional:         true,
+					ForceNew:         true,
+					ValidateDiagFunc: enum.Validate[awstypes.SchemaRegistryEventRecordFormat](),
+				},
+				"schema_registry_uri": {
+					Type:     schema.TypeString,
+					Optional: true,
+					ForceNew: true,
+					ValidateFunc: validation.All(
+						validation.StringLenBetween(1, 10000),
+						validation.StringMatch(regexache.MustCompile(`[a-zA-Z0-9-/*:_+=.@-]*`), "must be ARN or URL of the registry"),
+					),
+				},
+				"schema_validation_config": {
+					Type:     schema.TypeSet,
+					Optional: true,
+					ForceNew: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"attribute": {
+								Type:             schema.TypeString,
+								Optional:         true,
+								ForceNew:         true,
+								ValidateDiagFunc: enum.Validate[awstypes.KafkaSchemaValidationAttribute](),
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
 func resourceEventSourceMappingCreate(ctx context.Context, d *schema.ResourceData, meta any) diag.Diagnostics {
 	var diags diag.Diagnostics
 	conn := meta.(*conns.AWSClient).LambdaClient(ctx)
@@ -1134,6 +1201,9 @@ func expandAmazonManagedKafkaEventSourceConfig(tfMap map[string]any) *awstypes.A
 	if v, ok := tfMap["consumer_group_id"].(string); ok && v != "" {
 		apiObject.ConsumerGroupId = aws.String(v)
 	}
+	if v, ok := tfMap["schema_registry_config"].([]any); ok && len(v) > 0 && v[0] != nil {
+		apiObject.SchemaRegistryConfig = expandKafkaSchemaRegistryConfig(v[0].(map[string]any))
+	}
 
 	return apiObject
 }
@@ -1149,6 +1219,10 @@ func flattenAmazonManagedKafkaEventSourceConfig(apiObject *awstypes.AmazonManage
 		tfMap["consumer_group_id"] = aws.ToString(v)
 	}
 
+	if v := apiObject.SchemaRegistryConfig; v != nil {
+		tfMap["schema_registry_config"] = []any{flattenKafkaSchemaRegistryConfig(v)}
+	}
+
 	return tfMap
 }
 
@@ -1163,6 +1237,10 @@ func expandSelfManagedKafkaEventSourceConfig(tfMap map[string]any) *awstypes.Sel
 		apiObject.ConsumerGroupId = aws.String(v)
 	}
 
+	if v, ok := tfMap["schema_registry_config"].([]any); ok && len(v) > 0 && v[0] != nil {
+		apiObject.SchemaRegistryConfig = expandKafkaSchemaRegistryConfig(v[0].(map[string]any))
+	}
+
 	return apiObject
 }
 
@@ -1177,6 +1255,10 @@ func flattenSelfManagedKafkaEventSourceConfig(apiObject *awstypes.SelfManagedKaf
 		tfMap["consumer_group_id"] = aws.ToString(v)
 	}
 
+	if v := apiObject.SchemaRegistryConfig; v != nil {
+		tfMap["schema_registry_config"] = []any{flattenKafkaSchemaRegistryConfig(v)}
+	}
+
 	return tfMap
 }
 
@@ -1442,3 +1524,137 @@ func flattenEventSourceMappingMetricsConfig(apiObject *awstypes.EventSourceMappi
 
 	return tfMap
 }
+
+func expandKafkaSchemaRegistryConfig(tfMap map[string]any) *awstypes.KafkaSchemaRegistryConfig {
+	if tfMap == nil {
+		return nil
+	}
+
+	apiObject := &awstypes.KafkaSchemaRegistryConfig{}
+
+	if v, ok := tfMap["access_config"].(*schema.Set); ok && v != nil && v.Len() > 0 {
+		apiObject.AccessConfigs = expandKafkaSchemaRegistryAccessConfig(v.List())
+	}
+
+	if v, ok := tfMap["event_record_format"].(string); ok && v != "" {
+		apiObject.EventRecordFormat = awstypes.SchemaRegistryEventRecordFormat(v)
+	}
+
+	if v, ok := tfMap["schema_registry_uri"].(string); ok && v != "" {
+		apiObject.SchemaRegistryURI = aws.String(v)
+	}
+
+	if v, ok := tfMap["schema_validation_config"].(*schema.Set); ok && v != nil && v.Len() > 0 {
+		apiObject.SchemaValidationConfigs = expandKafkaSchemaValidationConfig(v.List())
+	}
+
+	return apiObject
+}
+
+func expandKafkaSchemaRegistryAccessConfig(tfList []any) []awstypes.KafkaSchemaRegistryAccessConfig {
+	if len(tfList) == 0 || tfList[0] == nil {
+		return nil
+	}
+
+	var apiObjects []awstypes.KafkaSchemaRegistryAccessConfig
+
+	for _, tfMapRaw := range tfList {
+		tfMap, ok := tfMapRaw.(map[string]any)
+
+		if !ok {
+			continue
+		}
+
+		apiObject := awstypes.KafkaSchemaRegistryAccessConfig{}
+		if v, ok := tfMap[names.AttrType].(string); ok && v != "" {
+			apiObject.Type = awstypes.KafkaSchemaRegistryAuthType(v)
+		}
+		if v, ok := tfMap[names.AttrURI].(string); ok && v != "" {
+			apiObject.URI = aws.String(v)
+		}
+
+		apiObjects = append(apiObjects, apiObject)
+	}
+	return apiObjects
+}
+
+func expandKafkaSchemaValidationConfig(tfList []any) []awstypes.KafkaSchemaValidationConfig {
+	if len(tfList) == 0 || tfList[0] == nil {
+		return nil
+	}
+
+	var apiObjects []awstypes.KafkaSchemaValidationConfig
+
+	for _, tfMapRaw := range tfList {
+		tfMap, ok := tfMapRaw.(map[string]any)
+
+		if !ok {
+			continue
+		}
+
+		apiObject := awstypes.KafkaSchemaValidationConfig{}
+		if v, ok := tfMap["attribute"].(string); ok && v != "" {
+			apiObject.Attribute = awstypes.KafkaSchemaValidationAttribute(v)
+		}
+
+		apiObjects = append(apiObjects, apiObject)
+	}
+	return apiObjects
+}
+
+func flattenKafkaSchemaRegistryConfig(apiObject *awstypes.KafkaSchemaRegistryConfig) map[string]any {
+	if apiObject == nil {
+		return nil
+	}
+
+	tfMap := map[string]any{}
+	if v := apiObject.AccessConfigs; len(v) > 0 {
+		tfMap["access_config"] = flattenKafkaSchemaRegistryAccessConfig(v)
+	}
+	if v := apiObject.EventRecordFormat; v != "" {
+		tfMap["event_record_format"] = v
+	}
+	if v := apiObject.SchemaRegistryURI; v != nil {
+		tfMap["schema_registry_uri"] = aws.ToString(v)
+	}
+	if v := apiObject.SchemaValidationConfigs; len(v) > 0 {
+		tfMap["schema_validation_config"] = flattenSchemaValidationConfig(v)
+	}
+
+	return tfMap
+}
+
+func flattenKafkaSchemaRegistryAccessConfig(apiObjects []awstypes.KafkaSchemaRegistryAccessConfig) []any {
+	if len(apiObjects) == 0 {
+		return nil
+	}
+
+	var tfList []any
+	for _, apiObject := range apiObjects {
+		tfMap := map[string]any{}
+		if v := apiObject.Type; v != "" {
+			tfMap[names.AttrType] = v
+		}
+		if v := apiObject.URI; v != nil {
+			tfMap[names.AttrURI] = aws.ToString(v)
+		}
+		tfList = append(tfList, tfMap)
+	}
+	return tfList
+}
+
+func flattenSchemaValidationConfig(apiObjects []awstypes.KafkaSchemaValidationConfig) []any {
+	if len(apiObjects) == 0 {
+		return nil
+	}
+
+	var tfList []any
+	for _, apiObject := range apiObjects {
+		tfMap := map[string]any{}
+		if v := apiObject.Attribute; v != "" {
+			tfMap["attribute"] = v
+		}
+		tfList = append(tfList, tfMap)
+	}
+	return tfList
+}