r/aws_appflow_flow: suppress diff when task_type value of Map_all is used

.changelog/35993.txt

@@ -0,0 +1,7 @@
+```release-note:bug
+resource/aws_appflow_flow: Fix perpetual diff when `task.task_type` is set to `Map_all`
+```
+
+```release-note:enhancement
+resource/aws_appflow_flow: Allow `task.source_fields` to be a `null` value
+```

internal/service/appflow/flow.go

@@ -6,6 +6,7 @@ package appflow
 import (
 	"context"
 	"log"
+	"slices"
 	"time"
 
 	"github.com/YakDriver/regexache"
@@ -1143,11 +1144,24 @@ func resourceFlow() *schema.Resource {
 						},
 						"source_fields": {
 							Type:     schema.TypeList,
-							Required: true,
+							Optional: true,
+							Computed: true,
 							Elem: &schema.Schema{
 								Type:         schema.TypeString,
 								ValidateFunc: validation.StringLenBetween(0, 2048),
 							},
+							DiffSuppressFunc: func(k, oldValue, newValue string, d *schema.ResourceData) bool {
+								if v, ok := d.Get("task").(*schema.Set); ok && v.Len() == 1 {
+									if tl, ok := v.List()[0].(map[string]interface{}); ok && len(tl) > 0 {
+										if sf, ok := tl["source_fields"].([]interface{}); ok && len(sf) == 1 {
+											if sf[0] == "" {
+												return oldValue == "0" && newValue == "1"
+											}
+										}
+									}
+								}
+								return false
+							},
 						},
 						"task_properties": {
 							Type:     schema.TypeMap,
@@ -1341,8 +1355,9 @@ func resourceFlowUpdate(ctx context.Context, d *schema.ResourceData, meta interf
 			TriggerConfig:             expandTriggerConfig(d.Get("trigger_config").([]interface{})[0].(map[string]interface{})),
 		}
 
-		if d.HasChange(names.AttrDescription) {
-			input.Description = aws.String(d.Get(names.AttrDescription).(string))
+		// always send description when updating a task
+		if v, ok := d.GetOk(names.AttrDescription); ok {
+			input.Description = aws.String(v.(string))
 		}
 
 		_, err := conn.UpdateFlow(ctx, input)
@@ -2440,6 +2455,8 @@ func expandTask(tfMap map[string]interface{}) *types.Task {
 
 	if v, ok := tfMap["source_fields"].([]interface{}); ok && len(v) > 0 {
 		a.SourceFields = flex.ExpandStringValueList(v)
+	} else {
+		a.SourceFields = []string{} // send an empty object if source_fields is empty (required by API)
 	}
 
 	if v, ok := tfMap["task_properties"].(map[string]interface{}); ok && len(v) > 0 {
@@ -3466,6 +3483,15 @@ func flattenTasks(tasks []types.Task) []interface{} {
 
 	var l []interface{}
 
+	t := slices.IndexFunc(tasks, func(t types.Task) bool {
+		return t.TaskType == types.TaskTypeMapAll
+	})
+
+	if t != -1 {
+		l = append(l, flattenTask(tasks[t]))
+		return l
+	}
+
 	for _, task := range tasks {
 		l = append(l, flattenTask(task))
 	}

internal/service/appflow/flow_test.go

@@ -266,6 +266,35 @@ func TestAccAppFlowFlow_taskUpdate(t *testing.T) {
 	})
 }
 
+func TestAccAppFlowFlow_task_mapAll(t *testing.T) {
+	ctx := acctest.Context(t)
+	var flowOutput types.FlowDefinition
+	rSourceName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	rDestinationName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	rFlowName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_appflow_flow.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.AppFlowServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckFlowDestroy(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccFlowConfig_task_mapAll(rSourceName, rDestinationName, rFlowName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckFlowExists(ctx, resourceName, &flowOutput),
+					resource.TestCheckResourceAttr(resourceName, "task.#", "1"),
+				),
+			},
+			{
+				Config:   testAccFlowConfig_task_mapAll(rSourceName, rDestinationName, rFlowName),
+				PlanOnly: true,
+			},
+		},
+	})
+}
+
 func TestAccAppFlowFlow_tags(t *testing.T) {
 	ctx := acctest.Context(t)
 	var flowOutput types.FlowDefinition
@@ -741,6 +770,59 @@ resource "aws_appflow_flow" "test" {
 	)
 }
 
+func testAccFlowConfig_task_mapAll(rSourceName, rDestinationName, rFlowName string) string {
+	return acctest.ConfigCompose(
+		testAccFlowConfig_base(rSourceName, rDestinationName),
+		fmt.Sprintf(`
+resource "aws_appflow_flow" "test" {
+  name = %[1]q
+
+  source_flow_config {
+    connector_type = "S3"
+    source_connector_properties {
+      s3 {
+        bucket_name   = aws_s3_bucket_policy.test_source.bucket
+        bucket_prefix = "flow"
+      }
+    }
+  }
+
+  destination_flow_config {
+    connector_type = "S3"
+    destination_connector_properties {
+      s3 {
+        bucket_name = aws_s3_bucket_policy.test_destination.bucket
+
+        s3_output_format_config {
+          prefix_config {
+            prefix_type = "PATH"
+          }
+        }
+      }
+    }
+  }
+
+  task {
+    task_type = "Map_all"
+
+    connector_operator {
+      s3 = "NO_OP"
+    }
+
+    task_properties = {
+      "DESTINATION_DATA_TYPE" = "id"
+      "SOURCE_DATA_TYPE"      = "id"
+    }
+  }
+
+  trigger_config {
+    trigger_type = "OnDemand"
+  }
+}
+`, rFlowName),
+	)
+}
+
 func testAccFlowConfig_tags1(rSourceName, rDestinationName, rFlowName string, tagKey1 string, tagValue1 string) string {
 	return acctest.ConfigCompose(
 		testAccFlowConfig_base(rSourceName, rDestinationName),