feat(dynamodb): add precision timestamp for kinesis stream (#31863)

### Issue # (if applicable)

Closes #31761 

### Reason for this change

Missing feature for kinesis data streams

### Description of changes

Added feature and unit tests

### Description of how you validated changes

Unit tests and integ tests included

### Checklist
- [X] My code adheres to the [CONTRIBUTING
GUIDE](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md) and
[DESIGN
GUIDELINES](https://github.com/aws/aws-cdk/blob/main/docs/DESIGN_GUIDELINES.md)

----

*By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache-2.0 license*

---------

Co-authored-by: Lee Hannigan <[email protected]>

Author: LeeroyHannigan

packages/@aws-cdk-testing/framework-integ/test/aws-dynamodb/test/integ.dynamodb.kinesis-stream.js.snapshot/aws-cdk-dynamodb-kinesis-stream.assets.json

@@ -37,6 +37,7 @@
      }
     ],
     "KinesisStreamSpecification": {
+     "ApproximateCreationDateTimePrecision": "MILLISECOND",
      "StreamArn": {
       "Fn::GetAtt": [
        "Stream790BDEE4",

packages/@aws-cdk-testing/framework-integ/test/aws-dynamodb/test/integ.dynamodb.kinesis-stream.js.snapshot/aws-cdk-dynamodb-kinesis-stream.template.json

@@ -11,6 +11,7 @@ new dynamodb.Table(stack, 'Table', {
   partitionKey: { name: 'hashKey', type: dynamodb.AttributeType.STRING },
   removalPolicy: cdk.RemovalPolicy.DESTROY,
   kinesisStream: stream,
+  kinesisPrecisionTimestamp: dynamodb.ApproximateCreationDateTimePrecision.MILLISECOND,
 });
 
 app.synth();

packages/@aws-cdk-testing/framework-integ/test/aws-dynamodb/test/integ.dynamodb.kinesis-stream.js.snapshot/cdk.out

@@ -222,6 +222,8 @@ const sortKey = schema.sortKey;
 
 A Kinesis Data Stream can be configured on the DynamoDB table to capture item-level changes.
 
+You can optionally configure the `kinesisPrecisionTimestamp` parameter to specify the precision level of the approximate creation date and time. The allowed values are `MICROSECOND` and `MILLISECOND`. If this parameter is not specified, the default precision is set to `MICROSECOND`.
+
 ```ts
 import * as kinesis from 'aws-cdk-lib/aws-kinesis';
 

packages/@aws-cdk-testing/framework-integ/test/aws-dynamodb/test/integ.dynamodb.kinesis-stream.js.snapshot/integ.json

@@ -208,6 +208,23 @@ export interface ImportSourceSpecification {
   readonly keyPrefix?: string;
 }
 
+/**
+ * The precision associated with the DynamoDB write timestamps that will be replicated to Kinesis.
+ * The default setting for record timestamp precision is microseconds. You can change this setting at any time.
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-dynamodb-table-kinesisstreamspecification.html#aws-properties-dynamodb-table-kinesisstreamspecification-properties
+ */
+export enum ApproximateCreationDateTimePrecision {
+  /**
+   * Millisecond precision
+   */
+  MILLISECOND = 'MILLISECOND',
+
+  /**
+   * Microsecond precision
+   */
+  MICROSECOND = 'MICROSECOND',
+}
+
 /**
  * Properties of a DynamoDB Table
  *
@@ -423,6 +440,13 @@ export interface TableProps extends TableOptions {
    * @default - no Kinesis Data Stream
    */
   readonly kinesisStream?: kinesis.IStream;
+
+  /**
+   * Kinesis Data Stream approximate creation timestamp prescision
+   *
+   * @default ApproximateCreationDateTimePrecision.MICROSECOND
+   */
+  readonly kinesisPrecisionTimestamp?: ApproximateCreationDateTimePrecision;
 }
 
 /**
@@ -1172,6 +1196,13 @@ export class Table extends TableBase {
     }
     this.validateProvisioning(props);
 
+    const kinesisStreamSpecification = props.kinesisStream
+      ? {
+        streamArn: props.kinesisStream.streamArn,
+        ...(props.kinesisPrecisionTimestamp && { approximateCreationDateTimePrecision: props.kinesisPrecisionTimestamp }),
+      }
+      : undefined;
+
     this.table = new CfnTable(this, 'Resource', {
       tableName: this.physicalName,
       keySchema: this.keySchema,
@@ -1196,7 +1227,7 @@ export class Table extends TableBase {
       tableClass: props.tableClass,
       timeToLiveSpecification: props.timeToLiveAttribute ? { attributeName: props.timeToLiveAttribute, enabled: true } : undefined,
       contributorInsightsSpecification: props.contributorInsightsEnabled !== undefined ? { enabled: props.contributorInsightsEnabled } : undefined,
-      kinesisStreamSpecification: props.kinesisStream ? { streamArn: props.kinesisStream.streamArn } : undefined,
+      kinesisStreamSpecification: kinesisStreamSpecification,
       deletionProtectionEnabled: props.deletionProtection,
       importSourceSpecification: this.renderImportSourceSpecification(props.importSource),
       resourcePolicy: props.resourcePolicy

packages/@aws-cdk-testing/framework-integ/test/aws-dynamodb/test/integ.dynamodb.kinesis-stream.js.snapshot/manifest.json

@@ -24,6 +24,7 @@ import {
   CfnTable,
   InputCompressionType,
   InputFormat,
+  ApproximateCreationDateTimePrecision,
 } from '../lib';
 import { ReplicaProvider } from '../lib/replica-provider';
 
@@ -3783,3 +3784,34 @@ test('Warm Throughput test provisioned', () => {
   });
 
 });
+
+test('Kinesis Stream - precision timestamp', () => {
+  // GIVEN
+  const app = new App();
+  const stack = new Stack(app, 'Stack');
+
+  const stream = new kinesis.Stream(stack, 'Stream');
+
+  // WHEN
+  const table = new Table(stack, 'Table', {
+    partitionKey: { name: 'id', type: AttributeType.STRING },
+    kinesisStream: stream,
+    kinesisPrecisionTimestamp: ApproximateCreationDateTimePrecision.MILLISECOND,
+  });
+
+  // THEN
+  Template.fromStack(stack).hasResourceProperties('AWS::DynamoDB::Table', {
+    KeySchema: [
+      { AttributeName: 'id', KeyType: 'HASH' },
+    ],
+    AttributeDefinitions: [
+      { AttributeName: 'id', AttributeType: 'S' },
+    ],
+    KinesisStreamSpecification: {
+      StreamArn: {
+        'Fn::GetAtt': ['Stream790BDEE4', 'Arn'],
+      },
+      ApproximateCreationDateTimePrecision: 'MILLISECOND',
+    },
+  });
+});