Merge pull request #45471 from data-storyteller/f-kinesisanalyticsv2-support-application-encryption

[Enhancement] r/aws_kinesisanalyticsv2_application: add application_encryption_configuration block

Author: ewbankkit

.changelog/45356.txt

@@ -0,0 +1,7 @@
+```release-note:enhancement
+resource/aws_kinesisanalyticsv2_application: Add `application_configuration.application_encryption_configuration` argument
+```
+
+```release-note:enhancement
+resource/aws_kinesisanalyticsv2_application: Support `FLINK-1_20` as a valid value for `runtime_environment`
+```

internal/service/kinesisanalyticsv2/application.go

@@ -120,6 +120,26 @@ func resourceApplication() *schema.Resource {
 									},
 								},
 							},
+							"application_encryption_configuration": {
+								Type:     schema.TypeList,
+								Optional: true,
+								Computed: true,
+								MaxItems: 1,
+								Elem: &schema.Resource{
+									Schema: map[string]*schema.Schema{
+										names.AttrKeyID: {
+											Type:         schema.TypeString,
+											Optional:     true,
+											ValidateFunc: verify.ValidARN,
+										},
+										"key_type": {
+											Type:             schema.TypeString,
+											Required:         true,
+											ValidateDiagFunc: enum.Validate[awstypes.KeyType](),
+										},
+									},
+								},
+							},
 							"application_snapshot_configuration": {
 								Type:     schema.TypeList,
 								Optional: true,
@@ -970,6 +990,12 @@ func resourceApplicationUpdate(ctx context.Context, d *schema.ResourceData, meta
 				updateApplication = true
 			}
 
+			if d.HasChange("application_configuration.0.application_encryption_configuration") {
+				applicationConfigurationUpdate.ApplicationEncryptionConfigurationUpdate = expandApplicationEncryptionConfigurationUpdate(d.Get("application_configuration.0.application_encryption_configuration").([]any))
+
+				updateApplication = true
+			}
+
 			if d.HasChange("application_configuration.0.application_snapshot_configuration") {
 				applicationConfigurationUpdate.ApplicationSnapshotConfigurationUpdate = expandApplicationSnapshotConfigurationUpdate(d.Get("application_configuration.0.application_snapshot_configuration").([]any))
 
@@ -1823,6 +1849,10 @@ func expandApplicationConfiguration(vApplicationConfiguration []any) *awstypes.A
 		applicationConfiguration.ApplicationCodeConfiguration = applicationCodeConfiguration
 	}
 
+	if vApplicationEncryptionConfiguration, ok := mApplicationConfiguration["application_encryption_configuration"].([]any); ok && len(vApplicationEncryptionConfiguration) > 0 && vApplicationEncryptionConfiguration[0] != nil {
+		applicationConfiguration.ApplicationEncryptionConfiguration = expandApplicationEncryptionConfiguration(vApplicationEncryptionConfiguration)
+	}
+
 	if vApplicationSnapshotConfiguration, ok := mApplicationConfiguration["application_snapshot_configuration"].([]any); ok && len(vApplicationSnapshotConfiguration) > 0 && vApplicationSnapshotConfiguration[0] != nil {
 		applicationSnapshotConfiguration := &awstypes.ApplicationSnapshotConfiguration{}
 
@@ -2710,6 +2740,10 @@ func flattenApplicationConfigurationDescription(applicationConfigurationDescript
 		mApplicationConfiguration["application_code_configuration"] = []any{mApplicationCodeConfiguration}
 	}
 
+	if applicationEncryptionConfigurationDescription := applicationConfigurationDescription.ApplicationEncryptionConfigurationDescription; applicationEncryptionConfigurationDescription != nil {
+		mApplicationConfiguration["application_encryption_configuration"] = flattenApplicationEncryptionConfigurationDescription(applicationEncryptionConfigurationDescription)
+	}
+
 	if applicationSnapshotConfigurationDescription := applicationConfigurationDescription.ApplicationSnapshotConfigurationDescription; applicationSnapshotConfigurationDescription != nil {
 		mApplicationSnapshotConfiguration := map[string]any{
 			"snapshots_enabled": aws.ToBool(applicationSnapshotConfigurationDescription.SnapshotsEnabled),
@@ -3099,3 +3133,57 @@ func expandStopApplicationInput(d *schema.ResourceData) *kinesisanalyticsv2.Stop
 
 	return apiObject
 }
+
+func expandApplicationEncryptionConfiguration(vApplicationEncryptionConfiguration []any) *awstypes.ApplicationEncryptionConfiguration {
+	if len(vApplicationEncryptionConfiguration) == 0 || vApplicationEncryptionConfiguration[0] == nil {
+		return nil
+	}
+
+	mApplicationEncryptionConfiguration := vApplicationEncryptionConfiguration[0].(map[string]any)
+	applicationEncryptionConfiguration := &awstypes.ApplicationEncryptionConfiguration{}
+
+	if vKeyType, ok := mApplicationEncryptionConfiguration["key_type"].(string); ok && vKeyType != "" {
+		applicationEncryptionConfiguration.KeyType = awstypes.KeyType(vKeyType)
+	}
+
+	if vKeyID, ok := mApplicationEncryptionConfiguration[names.AttrKeyID].(string); ok && vKeyID != "" {
+		applicationEncryptionConfiguration.KeyId = aws.String(vKeyID)
+	}
+
+	return applicationEncryptionConfiguration
+}
+
+func expandApplicationEncryptionConfigurationUpdate(vApplicationEncryptionConfiguration []any) *awstypes.ApplicationEncryptionConfigurationUpdate {
+	if len(vApplicationEncryptionConfiguration) == 0 || vApplicationEncryptionConfiguration[0] == nil {
+		return nil
+	}
+
+	mApplicationEncryptionConfiguration := vApplicationEncryptionConfiguration[0].(map[string]any)
+	applicationEncryptionConfiguration := &awstypes.ApplicationEncryptionConfigurationUpdate{}
+
+	if vKeyType, ok := mApplicationEncryptionConfiguration["key_type"].(string); ok && vKeyType != "" {
+		applicationEncryptionConfiguration.KeyTypeUpdate = awstypes.KeyType(vKeyType)
+	}
+
+	if vKeyID, ok := mApplicationEncryptionConfiguration[names.AttrKeyID].(string); ok && vKeyID != "" {
+		applicationEncryptionConfiguration.KeyIdUpdate = aws.String(vKeyID)
+	}
+
+	return applicationEncryptionConfiguration
+}
+
+func flattenApplicationEncryptionConfigurationDescription(applicationEncryptionConfiguration *awstypes.ApplicationEncryptionConfigurationDescription) []any {
+	if applicationEncryptionConfiguration == nil {
+		return []any{}
+	}
+
+	mApplicationEncryptionConfiguration := map[string]any{
+		"key_type": applicationEncryptionConfiguration.KeyType,
+	}
+
+	if applicationEncryptionConfiguration.KeyId != nil {
+		mApplicationEncryptionConfiguration[names.AttrKeyID] = aws.ToString(applicationEncryptionConfiguration.KeyId)
+	}
+
+	return []any{mApplicationEncryptionConfiguration}
+}

internal/service/kinesisanalyticsv2/application_test.go

@@ -4215,6 +4215,68 @@ func TestAccKinesisAnalyticsV2Application_RunConfiguration_Update(t *testing.T)
 	})
 }
 
+func TestAccKinesisAnalyticsV2Application_ApplicationEncryptionConfiguration_update(t *testing.T) {
+	ctx := acctest.Context(t)
+	var v awstypes.ApplicationDetail
+	resourceName := "aws_kinesisanalyticsv2_application.test"
+	iamRoleResourceName := "aws_iam_role.test.0"
+	kmsKey1ResourceName := "aws_kms_key.test.0"
+	kmsKey2ResourceName := "aws_kms_key.test.1"
+	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{
+			{
+				Config: testAccApplicationConfig_encryptionConfiguration(rName, "CUSTOMER_MANAGED_KEY", 0),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckApplicationExists(ctx, resourceName, &v),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_encryption_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_encryption_configuration.0.key_type", "CUSTOMER_MANAGED_KEY"),
+					resource.TestCheckResourceAttrPair(resourceName, "application_configuration.0.application_encryption_configuration.0.key_id", kmsKey1ResourceName, names.AttrARN),
+					acctest.CheckResourceAttrRegionalARN(ctx, resourceName, names.AttrARN, "kinesisanalytics", fmt.Sprintf("application/%s", rName)),
+					resource.TestCheckResourceAttrPair(resourceName, "service_execution_role", iamRoleResourceName, names.AttrARN),
+					resource.TestCheckResourceAttr(resourceName, names.AttrStatus, "READY"),
+					resource.TestCheckResourceAttr(resourceName, "version_id", "1"),
+				),
+			},
+			{
+				Config: testAccApplicationConfig_encryptionConfiguration(rName, "CUSTOMER_MANAGED_KEY", 1),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckApplicationExists(ctx, resourceName, &v),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_encryption_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_encryption_configuration.0.key_type", "CUSTOMER_MANAGED_KEY"),
+					resource.TestCheckResourceAttrPair(resourceName, "application_configuration.0.application_encryption_configuration.0.key_id", kmsKey2ResourceName, names.AttrARN),
+					resource.TestCheckResourceAttr(resourceName, names.AttrStatus, "READY"),
+					resource.TestCheckResourceAttr(resourceName, "version_id", "2"),
+				),
+			},
+			{
+				Config: testAccApplicationConfig_encryptionConfiguration(rName, "AWS_OWNED_KEY", 0),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckApplicationExists(ctx, resourceName, &v),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_encryption_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_encryption_configuration.0.key_type", "AWS_OWNED_KEY"),
+					resource.TestCheckResourceAttr(resourceName, "application_configuration.0.application_encryption_configuration.0.key_id", ""),
+					resource.TestCheckResourceAttr(resourceName, names.AttrStatus, "READY"),
+					resource.TestCheckResourceAttr(resourceName, "version_id", "3"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func testAccCheckApplicationDestroy(ctx context.Context) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		conn := acctest.Provider.Meta().(*conns.AWSClient).KinesisAnalyticsV2Client(ctx)
@@ -6157,3 +6219,87 @@ resource "aws_kinesisanalyticsv2_application" "test" {
 }
 `, rName))
 }
+
+func testAccApplicationConfig_encryptionConfiguration(rName, keyType string, keyIndex int) string {
+	return acctest.ConfigCompose(
+		testAccApplicationConfig_baseServiceExecutionIAMRole(rName),
+		testAccApplicationConfig_baseFlinkApplication(rName),
+		testAccApplicationConfig_baseKMSKeys(rName, 2),
+		fmt.Sprintf(`
+resource "aws_kinesisanalyticsv2_application" "test" {
+  name                   = %[1]q
+  runtime_environment    = "FLINK-1_20"
+  service_execution_role = aws_iam_role.test[0].arn
+
+  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
+        }
+      }
+
+      code_content_type = "ZIPFILE"
+    }
+
+    application_encryption_configuration {
+      key_type = %[2]q
+      %[3]s
+    }
+  }
+}
+`, rName, keyType, func() string {
+			if keyType == "CUSTOMER_MANAGED_KEY" {
+				return fmt.Sprintf(`key_id = aws_kms_key.test[%d].arn`, keyIndex)
+			}
+			return ""
+		}()))
+}
+
+func testAccApplicationConfig_baseKMSKeys(rName string, count int) string {
+	return fmt.Sprintf(`
+resource "aws_kms_key" "test" {
+  count = %[2]d
+
+  description             = "%[1]s-${count.index}"
+  deletion_window_in_days = 7
+}
+
+data "aws_caller_identity" "current" {}
+
+resource "aws_kms_key_policy" "test" {
+  count = %[2]d
+
+  key_id = aws_kms_key.test[count.index].id
+  policy = jsonencode({
+    Id = %[1]q
+    Statement = [
+      {
+        Action = "kms:*"
+        Effect = "Allow"
+        Principal = {
+          AWS = data.aws_caller_identity.current.arn
+        }
+
+        Resource = "*"
+        Sid      = "Enable IAM User Permissions"
+      },
+      {
+        Action = [
+          "kms:*",
+        ]
+        Effect = "Allow"
+        Principal = {
+          Service = ["kinesisanalytics.amazonaws.com", "infrastructure.kinesisanalytics.amazonaws.com"]
+        }
+
+        Resource = "*"
+        Sid      = "Enable IAM MSF Permissions"
+      },
+    ]
+    Version = "2012-10-17"
+  })
+}
+`, rName, count)
+}

website/docs/r/kinesisanalyticsv2_application.html.markdown

@@ -257,7 +257,7 @@ This resource supports the following arguments:
 
 * `region` - (Optional) Region where this resource will be [managed](https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints). Defaults to the Region set in the [provider configuration](https://registry.terraform.io/providers/hashicorp/aws/latest/docs#aws-configuration-reference).
 * `name` - (Required) The name of the application.
-* `runtime_environment` - (Required) The runtime environment for the application. Valid values: `SQL-1_0`, `FLINK-1_6`, `FLINK-1_8`, `FLINK-1_11`, `FLINK-1_13`, `FLINK-1_15`, `FLINK-1_18`, `FLINK-1_19`.
+* `runtime_environment` - (Required) The runtime environment for the application. Valid values: `SQL-1_0`, `FLINK-1_6`, `FLINK-1_8`, `FLINK-1_11`, `FLINK-1_13`, `FLINK-1_15`, `FLINK-1_18`, `FLINK-1_19`, `FLINK-1_20`.
 * `service_execution_role` - (Required) The ARN of the [IAM role](/docs/providers/aws/r/iam_role.html) used by the application to access Kinesis data streams, Kinesis Data Firehose delivery streams, Amazon S3 objects, and other external resources.
 * `application_configuration` - (Optional) The application's configuration
 * `application_mode` - (Optional) The application's mode. Valid values are `STREAMING`, `INTERACTIVE`.
@@ -270,6 +270,7 @@ This resource supports the following arguments:
 The `application_configuration` object supports the following:
 
 * `application_code_configuration` - (Required) The code location and type parameters for the application.
+* `application_encryption_configuration` - (Optional) The encryption configuration for the application. This can be used to encrypt data at rest in the application.
 * `application_snapshot_configuration` - (Optional) Describes whether snapshots are enabled for a Flink-based application.
 * `environment_properties` - (Optional) Describes execution properties for a Flink-based application.
 * `flink_application_configuration` - (Optional) The configuration of a Flink-based application.
@@ -282,6 +283,11 @@ The `application_code_configuration` object supports the following:
 * `code_content_type` - (Required) Specifies whether the code content is in text or zip format. Valid values: `PLAINTEXT`, `ZIPFILE`.
 * `code_content` - (Optional) The location and type of the application code.
 
+The `application_encryption_configuration` object supports the following:
+
+* `key_type` - (Required) The type of encryption key to use. Valid values: `CUSTOMER_MANAGED_KEY`, `AWS_OWNED_KEY`.
+* `key_id` - (Optional) The ARN of the KMS key to use for encryption. Required when `key_type` is set to `CUSTOMER_MANAGED_KEY`. The KMS key must be in the same region as the application.
+
 The `code_content` object supports the following:
 
 * `s3_content_location` - (Optional) Information about the Amazon S3 bucket containing the application code.