fix reviews

Author: sannya-singal

README.md

@@ -2,7 +2,7 @@
 
 ## Introduction
 
-This scenario demonstrates how to use Database Migration Service (DMS) to create change data capture (CDC) tasks using the Cloud Development Kit in Python. It is a self-contained setup that will create a VPC to host the DMS replication instance, a database, a Kinesis stream with non-default settings, and a replication task.
+This scenario demonstrates how to use Database Migration Service (DMS) to create change data capture (CDC) tasks using the Cloud Development Kit in Python. It is a self-contained setup that will create a VPC to host the DMS replication instance, a database, a Kinesis stream, and a replication task.
 
 ![dms-postgres-to-kinesis](./dms-postgres-to-kinesis.jpg)
 
@@ -41,6 +41,9 @@ To deploy the infrastructure, you can run the following command:
 make deploy
 ```
 
+> NOTE: By default we create kinesis target endpoint with default settings.
+> In order to create the target with non-default values set the environment to `KINESIS_TARGET=non-default`. 
+
 After successful deployment, you will see the following output:
 
 ```bash

dms_sample/stack.py

@@ -15,6 +15,7 @@
 USERNAME = os.getenv("USERNAME", "")
 USER_PWD = os.getenv("USERPWD", "")
 DB_NAME = os.getenv("DB_NAME", "")
+KINESIS_TARGET = os.getenv("KINESIS_TARGET", "default")
 SCHEMA_NAME = "public"
 
 
@@ -227,23 +228,36 @@ def create_postgres_access_role(stack: Stack, postgres_secret: secretsmanager.Cf
 
 
 def create_kinesis_target_endpoint(stack: Stack, target_stream: kinesis.Stream, dms_assume_role: iam.Role) -> dms.CfnEndpoint:
+    if str.lower(KINESIS_TARGET) == "non-default":
+        return dms.CfnEndpoint(
+            stack,
+            "target",
+            endpoint_type="target",
+            engine_name="kinesis",
+            kinesis_settings=dms.CfnEndpoint.KinesisSettingsProperty(
+                stream_arn=target_stream.stream_arn,
+                message_format="json",
+                service_access_role_arn=dms_assume_role.role_arn,
+                include_control_details=True,
+                include_null_and_empty=True,
+                include_partition_value=True,
+                include_table_alter_operations=True,
+                include_transaction_details=False,
+                partition_include_schema_table=True,
+            ),
+        )
+
     return dms.CfnEndpoint(
-        stack,
-        "target",
-        endpoint_type="target",
-        engine_name="kinesis",
-        kinesis_settings=dms.CfnEndpoint.KinesisSettingsProperty(
-            stream_arn=target_stream.stream_arn,
-            message_format="json",
-            service_access_role_arn=dms_assume_role.role_arn,
-            include_control_details=True,
-            include_null_and_empty=True,
-            include_partition_value=True,
-            include_table_alter_operations=True,
-            include_transaction_details=False,
-            partition_include_schema_table=True,
-        ),
-    )
+            stack,
+            "target",
+            endpoint_type="target",
+            engine_name="kinesis",
+            kinesis_settings=dms.CfnEndpoint.KinesisSettingsProperty(
+                stream_arn=target_stream.stream_arn,
+                message_format="json",
+                service_access_role_arn=dms_assume_role.role_arn,
+            ),
+        )       
 
 
 def create_replication_instance(

run.py

@@ -14,6 +14,7 @@
 STACK_NAME = os.getenv("STACK_NAME", "DMSPostgresKinesis")
 
 ENDPOINT_URL = os.getenv("ENDPOINT_URL")
+KINESIS_TARGET = os.getenv("KINESIS_TARGET", "default")
 
 cfn = client("cloudformation", endpoint_url=ENDPOINT_URL)
 dms = client("dms", endpoint_url=ENDPOINT_URL)
@@ -240,7 +241,10 @@ def execute_cdc(cfn_output: CfnOutput):
     threshold_timestamp = int(time.time())
     sleep(1)
     run_queries_on_postgres(credentials, q.ALTER_TABLES)
-    wait_for_kinesis(stream, 3, threshold_timestamp)
+    if str.lower(KINESIS_TARGET) == "non-default":
+        wait_for_kinesis(stream, 3, threshold_timestamp)
+    else:
+        wait_for_kinesis(stream, 0, threshold_timestamp)
     print("\n****End of ALTER tables events****\n")
 
     print("\n****Table Statistics****\n")