b-aws_kinesisanalyticsv2_application-43003 :: Passing ACC test

Author: cjlludwig

internal/service/kinesisanalyticsv2/application_test.go

@@ -1272,6 +1272,51 @@ func TestAccKinesisAnalyticsV2Application_FlinkApplication_restoreFromSnapshot(t
 	})
 }
 
+func TestAccKinesisAnalyticsV2Application_FlinkApplication_snapshotToggleRestoreTypeBug(t *testing.T) {
+	ctx := acctest.Context(t)
+	var v awstypes.ApplicationDetail
+	resourceName := "aws_kinesisanalyticsv2_application.test"
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t); testAccPreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.KinesisAnalyticsV2ServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckApplicationDestroy(ctx),
+		Steps: []resource.TestStep{
+			// Step 1: Create application with snapshots enabled and SKIP_RESTORE_FROM_SNAPSHOT
+			{
+				Config: testAccApplicationConfig_snapshotToggleBug(rName, true),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckApplicationExists(ctx, resourceName, &v),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_snapshot_configuration.0.snapshots_enabled", acctest.CtTrue),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.run_configuration.0.application_restore_configuration.0.application_restore_type", "SKIP_RESTORE_FROM_SNAPSHOT"),
+				),
+			},
+			// Step 2: Disable snapshots but maintain SKIP_RESTORE_FROM_SNAPSHOT
+			{
+				Config: testAccApplicationConfig_snapshotToggleBug(rName, false),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckApplicationExists(ctx, resourceName, &v),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_snapshot_configuration.0.snapshots_enabled", acctest.CtFalse),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.run_configuration.0.application_restore_configuration.0.application_restore_type", "SKIP_RESTORE_FROM_SNAPSHOT"),
+				),
+			},
+			// Step 3: Re-enable snapshots - application_restore_type should remain SKIP_RESTORE_FROM_SNAPSHOT
+			{
+				Config: testAccApplicationConfig_snapshotToggleBug(rName, true),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckApplicationExists(ctx, resourceName, &v),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_snapshot_configuration.0.snapshots_enabled", acctest.CtTrue),
+					// This is the critical check - the restore type should remain SKIP_RESTORE_FROM_SNAPSHOT
+					// but the bug causes it to default to restore from latest snapshot
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.run_configuration.0.application_restore_configuration.0.application_restore_type", "SKIP_RESTORE_FROM_SNAPSHOT"),
+				),
+			},
+		},
+	})
+}
+
 func TestAccKinesisAnalyticsV2Application_FlinkApplicationStartApplication_onCreate(t *testing.T) {
 	ctx := acctest.Context(t)
 	var v awstypes.ApplicationDetail
@@ -5750,3 +5795,83 @@ resource "aws_kinesisanalyticsv2_application" "test" {
 }
 `, rName))
 }
+
+func testAccApplicationConfig_snapshotToggleBug(rName string, snapshotsEnabled bool) string {
+	return acctest.ConfigCompose(
+		testAccApplicationConfig_baseServiceExecutionIAMRole(rName),
+		testAccApplicationConfig_baseFlinkApplication(rName),
+		fmt.Sprintf(`
+resource "aws_kinesisanalyticsv2_application" "test" {
+  name                   = %[1]q
+  runtime_environment    = "FLINK-1_20"
+  service_execution_role = aws_iam_role.test[0].arn
+  start_application      = true
+
+  application_configuration {
+    application_code_configuration {
+      code_content {
+        s3_content_location {
+          bucket_arn     = aws_s3_bucket.test.arn
+          file_key       = aws_s3_object.test[0].key
+          object_version = aws_s3_object.test[0].version_id
+        }
+      }
+
+      code_content_type = "ZIPFILE"
+    }
+
+    application_snapshot_configuration {
+      snapshots_enabled = %[2]t
+    }
+
+    environment_properties {
+      property_group {
+        property_group_id = "InputStream0"
+
+        property_map = {
+          "aws.region"  = data.aws_region.current.region
+          "stream.name" = aws_kinesis_stream.input.name
+        }
+      }
+
+      property_group {
+        property_group_id = "OutputStream0"
+
+        property_map = {
+          "aws.region"  = data.aws_region.current.region
+          "stream.name" = aws_kinesis_stream.output.name
+        }
+      }
+    }
+
+    flink_application_configuration {
+      checkpoint_configuration {
+        configuration_type = "DEFAULT"
+      }
+
+      monitoring_configuration {
+        configuration_type = "CUSTOM"
+        log_level          = "DEBUG"
+        metrics_level      = "TASK"
+      }
+
+      parallelism_configuration {
+        auto_scaling_enabled = true
+        configuration_type   = "CUSTOM"
+        parallelism          = 1
+        parallelism_per_kpu  = 1
+      }
+    }
+
+    run_configuration {
+      application_restore_configuration {
+        application_restore_type = "SKIP_RESTORE_FROM_SNAPSHOT"
+      }
+      flink_run_configuration {
+        allow_non_restored_state = false
+      }
+    }
+  }
+}
+`, rName, snapshotsEnabled))
+}