fix: Allow run_configurations to be set on Kinesis AnalyticsV2

internal/service/kinesisanalyticsv2/application.go

@@ -1380,6 +1380,20 @@ func resourceApplicationUpdate(d *schema.ResourceData, meta interface{}) error {
 				}
 			}
 
+			if d.HasChange("application_configuration.0.run_configuration") {
+				application, err := FindApplicationDetailByName(conn, applicationName)
+
+				if err != nil {
+					return fmt.Errorf("error reading Kinesis Analytics v2 Application (%s): %w", applicationName, err)
+				}
+
+				if actual, expected := aws.StringValue(application.ApplicationStatus), kinesisanalyticsv2.ApplicationStatusRunning; actual == expected {
+					input.RunConfigurationUpdate = expandRunConfigurationUpdate(d.Get("application_configuration.0.run_configuration").([]interface{}))
+
+					updateApplication = true
+				}
+			}
+
 			input.ApplicationConfigurationUpdate = applicationConfigurationUpdate
 		}
 
@@ -2483,6 +2497,46 @@ func expandVPCConfigurationUpdate(vVpcConfiguration []interface{}) *kinesisanaly
 	return vpcConfigurationUpdate
 }
 
+func expandRunConfigurationUpdate(vRunConfigurationUpdate []interface{}) *kinesisanalyticsv2.RunConfigurationUpdate {
+	if len(vRunConfigurationUpdate) == 0 || vRunConfigurationUpdate[0] == nil {
+		return nil
+	}
+
+	runConfigurationUpdate := &kinesisanalyticsv2.RunConfigurationUpdate{}
+
+	mRunConfiguration := vRunConfigurationUpdate[0].(map[string]interface{})
+
+	if vApplicationRestoreConfiguration, ok := mRunConfiguration["application_restore_configuration"].([]interface{}); ok && len(vApplicationRestoreConfiguration) > 0 && vApplicationRestoreConfiguration[0] != nil {
+		applicationRestoreConfiguration := &kinesisanalyticsv2.ApplicationRestoreConfiguration{}
+
+		mApplicationRestoreConfiguration := vApplicationRestoreConfiguration[0].(map[string]interface{})
+
+		if vApplicationRestoreType, ok := mApplicationRestoreConfiguration["application_restore_type"].(string); ok && vApplicationRestoreType != "" {
+			applicationRestoreConfiguration.ApplicationRestoreType = aws.String(vApplicationRestoreType)
+		}
+
+		if vSnapshotName, ok := mApplicationRestoreConfiguration["snapshot_name"].(string); ok && vSnapshotName != "" {
+			applicationRestoreConfiguration.SnapshotName = aws.String(vSnapshotName)
+		}
+
+		runConfigurationUpdate.ApplicationRestoreConfiguration = applicationRestoreConfiguration
+	}
+
+	if vFlinkRunConfiguration, ok := mRunConfiguration["flink_run_configuration"].([]interface{}); ok && len(vFlinkRunConfiguration) > 0 && vFlinkRunConfiguration[0] != nil {
+		flinkRunConfiguration := &kinesisanalyticsv2.FlinkRunConfiguration{}
+
+		mFlinkRunConfiguration := vFlinkRunConfiguration[0].(map[string]interface{})
+
+		if vAllowNonRestoredState, ok := mFlinkRunConfiguration["allow_non_restored_state"].(bool); ok {
+			flinkRunConfiguration.AllowNonRestoredState = aws.Bool(vAllowNonRestoredState)
+		}
+
+		runConfigurationUpdate.FlinkRunConfiguration = flinkRunConfiguration
+	}
+
+	return runConfigurationUpdate
+}
+
 func flattenApplicationConfigurationDescription(applicationConfigurationDescription *kinesisanalyticsv2.ApplicationConfigurationDescription) []interface{} {
 	if applicationConfigurationDescription == nil {
 		return []interface{}{}

internal/service/kinesisanalyticsv2/application_snapshot_test.go

@@ -141,5 +141,5 @@ func testAccCheckApplicationSnapshotExists(n string, v *kinesisanalyticsv2.Snaps
 }
 
 func testAccApplicationSnapshotConfig(rName string) string {
-	return testAccApplicationConfigStartSnapshotableFlinkApplication(rName, "SKIP_RESTORE_FROM_SNAPSHOT", "")
+	return testAccApplicationConfigStartSnapshotableFlinkApplication(rName, "SKIP_RESTORE_FROM_SNAPSHOT", "", false)
 }