feat: Add missing functionality for AppFlow Custom Connector

.changelog/26766.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_appflow_connector_profile: Add support for `connector_type` CustomConnector. Add `cluster_identifier`, `database_name`, and `data_api_role_arn` attributes for `redshift` `connection_profile_properties`
+```

internal/service/appflow/connector_profile.go

@@ -33,6 +33,7 @@ func ResourceConnectorProfile() *schema.Resource {
 			"name": {
 				Type:     schema.TypeString,
 				Required: true,
+				ForceNew: true,
 				ValidateFunc: validation.All(
 					validation.StringLenBetween(1, 256),
 					validation.StringMatch(regexp.MustCompile(`[\w/!@#+=.-]+`), "must match [\\w/!@#+=.-]+"),
@@ -1093,6 +1094,19 @@ func ResourceConnectorProfile() *schema.Resource {
 													Optional:     true,
 													ValidateFunc: validation.StringLenBetween(0, 512),
 												},
+												"cluster_identifier": {
+													Type:     schema.TypeString,
+													Optional: true,
+												},
+												"data_api_role_arn": {
+													Type:         schema.TypeString,
+													Optional:     true,
+													ValidateFunc: verify.ValidARN,
+												},
+												"database_name": {
+													Type:     schema.TypeString,
+													Optional: true,
+												},
 												"database_url": {
 													Type:         schema.TypeString,
 													Optional:     true,
@@ -1594,7 +1608,7 @@ func expandCustomConnectorProfileCredentials(m map[string]interface{}) *appflow.
 		credentials.Custom = expandCustomAuthCredentials(v[0].(map[string]interface{}))
 	}
 
-	if v, ok := m["oauth2_credentials"].([]interface{}); ok && len(v) > 0 {
+	if v, ok := m["oauth2"].([]interface{}); ok && len(v) > 0 {
 		credentials.Oauth2 = expandOAuth2Credentials(v[0].(map[string]interface{}))
 	}
 
@@ -1911,6 +1925,10 @@ func expandConnectorProfileProperties(m map[string]interface{}) *appflow.Connect
 		cpc.Amplitude = v[0].(*appflow.AmplitudeConnectorProfileProperties)
 	}
 
+	if v, ok := m["custom_connector"].([]interface{}); ok && len(v) > 0 {
+		cpc.CustomConnector = expandCustomConnectorProfileProperties(v[0].(map[string]interface{}))
+	}
+
 	if v, ok := m["datadog"].([]interface{}); ok && len(v) > 0 && v[0] != nil {
 		cpc.Datadog = expandDatadogConnectorProfileProperties(v[0].(map[string]interface{}))
 	}
@@ -2012,8 +2030,11 @@ func expandMarketoConnectorProfileProperties(m map[string]interface{}) *appflow.
 
 func expandRedshiftConnectorProfileProperties(m map[string]interface{}) *appflow.RedshiftConnectorProfileProperties {
 	properties := appflow.RedshiftConnectorProfileProperties{
-		BucketName: aws.String(m["bucket_name"].(string)),
-		RoleArn:    aws.String(m["role_arn"].(string)),
+		BucketName:        aws.String(m["bucket_name"].(string)),
+		ClusterIdentifier: aws.String(m["cluster_identifier"].(string)),
+		RoleArn:           aws.String(m["role_arn"].(string)),
+		DataApiRoleArn:    aws.String(m["data_api_role_arn"].(string)),
+		DatabaseName:      aws.String(m["database_name"].(string)),
 	}
 
 	if v, ok := m["bucket_prefix"].(string); ok && v != "" {
@@ -2049,6 +2070,20 @@ func expandSalesforceConnectorProfileProperties(m map[string]interface{}) *appfl
 	return &properties
 }
 
+func expandCustomConnectorProfileProperties(m map[string]interface{}) *appflow.CustomConnectorProfileProperties {
+	properties := appflow.CustomConnectorProfileProperties{}
+
+	if v, ok := m["oauth2_properties"].([]interface{}); ok && len(v) > 0 {
+		properties.OAuth2Properties = expandOAuth2Properties(v[0].(map[string]interface{}))
+	}
+
+	if v, ok := m["profile_properties"].(map[string]interface{}); ok && len(v) > 0 {
+		properties.ProfileProperties = flex.ExpandStringMap(v)
+	}
+
+	return &properties
+}
+
 func expandSAPODataConnectorProfileProperties(m map[string]interface{}) *appflow.SAPODataConnectorProfileProperties {
 	properties := appflow.SAPODataConnectorProfileProperties{
 		ApplicationHostUrl:     aws.String(m["application_host_url"].(string)),
@@ -2132,6 +2167,19 @@ func expandOAuthProperties(m map[string]interface{}) *appflow.OAuthProperties {
 	return &properties
 }
 
+func expandOAuth2Properties(m map[string]interface{}) *appflow.OAuth2Properties {
+	properties := appflow.OAuth2Properties{
+		OAuth2GrantType: aws.String(m["oauth2_grant_type"].(string)),
+		TokenUrl:        aws.String(m["token_url"].(string)),
+	}
+
+	if v, ok := m["token_url_custom_properties"].(map[string]interface{}); ok && len(v) > 0 {
+		properties.TokenUrlCustomProperties = flex.ExpandStringMap(v)
+	}
+
+	return &properties
+}
+
 func flattenConnectorProfileConfig(cpp *appflow.ConnectorProfileProperties, cpc []interface{}) []interface{} {
 	m := make(map[string]interface{})
 
@@ -2148,6 +2196,9 @@ func flattenConnectorProfileProperties(cpp *appflow.ConnectorProfileProperties)
 	if cpp.Amplitude != nil {
 		result["amplitude"] = []interface{}{m}
 	}
+	if cpp.CustomConnector != nil {
+		result["custom_connector"] = flattenCustomConnectorProfileProperties(cpp.CustomConnector)
+	}
 	if cpp.Datadog != nil {
 		m["instance_url"] = aws.StringValue(cpp.Datadog.InstanceUrl)
 		result["datadog"] = []interface{}{m}
@@ -2222,6 +2273,23 @@ func flattenRedshiftConnectorProfileProperties(properties *appflow.RedshiftConne
 	}
 
 	m["role_arn"] = aws.StringValue(properties.RoleArn)
+	m["cluster_identifier"] = aws.StringValue(properties.ClusterIdentifier)
+	m["data_api_role_arn"] = aws.StringValue(properties.DataApiRoleArn)
+	m["database_name"] = aws.StringValue(properties.DatabaseName)
+
+	return []interface{}{m}
+}
+
+func flattenCustomConnectorProfileProperties(properties *appflow.CustomConnectorProfileProperties) []interface{} {
+	m := make(map[string]interface{})
+
+	if properties.OAuth2Properties != nil {
+		m["oauth2_properties"] = flattenOAuth2Properties(properties.OAuth2Properties)
+	}
+
+	if properties.ProfileProperties != nil {
+		m["profile_properties"] = aws.StringValueMap(properties.ProfileProperties)
+	}
 
 	return []interface{}{m}
 }
@@ -2294,3 +2362,13 @@ func flattenOAuthProperties(properties *appflow.OAuthProperties) []interface{} {
 
 	return []interface{}{m}
 }
+
+func flattenOAuth2Properties(properties *appflow.OAuth2Properties) []interface{} {
+	m := make(map[string]interface{})
+
+	m["oauth2_grant_type"] = aws.StringValue(properties.OAuth2GrantType)
+	m["token_url"] = aws.StringValue(properties.TokenUrl)
+	m["token_url_custom_properties"] = aws.StringValueMap(properties.TokenUrlCustomProperties)
+
+	return []interface{}{m}
+}

internal/service/appflow/connector_profile_test.go

@@ -220,7 +220,7 @@ resource "aws_redshift_subnet_group" "test" {
 }
 
 data "aws_iam_policy" "test" {
-  name = "AmazonRedshiftAllCommandsFullAccess"
+  name = "AmazonRedshiftFullAccess"
 }
 
 resource "aws_iam_role" "test" {
@@ -236,7 +236,7 @@ resource "aws_iam_role" "test" {
         Effect = "Allow"
         Sid    = ""
         Principal = {
-          Service = "ec2.amazonaws.com"
+          Service = "appflow.amazonaws.com"
         }
       },
     ]
@@ -301,9 +301,12 @@ resource "aws_appflow_connector_profile" "test" {
 
     connector_profile_properties {
       redshift {
-        bucket_name  = %[1]q
-        database_url = "jdbc:redshift://${aws_redshift_cluster.test.endpoint}/dev"
-        role_arn     = aws_iam_role.test.arn
+        bucket_name        = %[1]q
+        cluster_identifier = aws_redshift_cluster.test.cluster_identifier
+        database_name      = "dev"
+        database_url       = "jdbc:redshift://${aws_redshift_cluster.test.endpoint}/dev"
+        data_api_role_arn  = aws_iam_role.test.arn
+        role_arn           = aws_iam_role.test.arn
       }
     }
   }
@@ -340,10 +343,13 @@ resource "aws_appflow_connector_profile" "test" {
 
     connector_profile_properties {
       redshift {
-        bucket_name   = %[1]q
-        bucket_prefix = %[4]q
-        database_url  = "jdbc:redshift://${aws_redshift_cluster.test.endpoint}/dev"
-        role_arn      = aws_iam_role.test.arn
+        bucket_name        = %[1]q
+        bucket_prefix      = %[4]q
+        cluster_identifier = aws_redshift_cluster.test.cluster_identifier
+        database_name      = "dev"
+        database_url       = "jdbc:redshift://${aws_redshift_cluster.test.endpoint}/dev"
+        data_api_role_arn  = aws_iam_role.test.arn
+        role_arn           = aws_iam_role.test.arn
       }
     }
   }