feat(kinesisfirehose): support S3 file extension format (#33776)

### Issue # (if applicable)

Closes #32154.

### Reason for this change

The file extension can be overridden by specifying `FileExtension`.
For details, see **S3 file extension format (optional)** in [Configure destination settings for Amazon S3](https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-s3)

### Description of changes

Added `fileExtension` prop to `S3BucketProps`.

### Describe any new or updated permissions being added

N/A

### Description of how you validated changes

Added unit tests and updated an integ test to include `fileExtension`.

### 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*

Author: Tietew

packages/@aws-cdk-testing/framework-integ/test/aws-kinesisfirehose/test/integ.s3-bucket.lit.js.snapshot/asset.530055f7515b3f0a47900f5df37e729ba40ca977b2d07b952bdefa2b8f883f42.bundle/index.js renamed to packages/@aws-cdk-testing/framework-integ/test/aws-kinesisfirehose/test/integ.s3-bucket.lit.js.snapshot/asset.97484721f29e34bf38d7a459804dd2d2a8dea6f8c27d7531e215bf4274fbc895.bundle/index.js

@@ -590,6 +590,7 @@
       }
      },
      "ErrorOutputPrefix": "errorPrefix",
+     "FileExtension": ".log.gz",
      "Prefix": "regularPrefix",
      "ProcessingConfiguration": {
       "Enabled": true,

packages/@aws-cdk-testing/framework-integ/test/aws-kinesisfirehose/test/integ.s3-bucket.lit.js.snapshot/aws-cdk-firehose-delivery-stream-s3-all-properties.assets.json

@@ -51,6 +51,7 @@ const deliveryStream = new firehose.DeliveryStream(stack, 'DeliveryStream', {
     compression: firehose.Compression.GZIP,
     dataOutputPrefix: 'regularPrefix',
     errorOutputPrefix: 'errorPrefix',
+    fileExtension: '.log.gz',
     bufferingInterval: cdk.Duration.seconds(60),
     bufferingSize: cdk.Size.mebibytes(1),
     encryptionKey: key,

packages/@aws-cdk-testing/framework-integ/test/aws-kinesisfirehose/test/integ.s3-bucket.lit.js.snapshot/aws-cdk-firehose-delivery-stream-s3-all-properties.template.json

@@ -112,6 +112,16 @@ const s3Destination = new firehose.S3Bucket(bucket, {
 See: [Custom S3 Prefixes](https://docs.aws.amazon.com/firehose/latest/dev/s3-prefixes.html)
 in the *Amazon Data Firehose Developer Guide*.
 
+To override default file extension appended by Data Format Conversion or S3 compression features, specify `fileExtension`.
+
+```ts
+declare const bucket: s3.Bucket;
+const s3Destination = new firehose.S3Bucket(bucket, {
+  compression: firehose.Compression.GZIP,
+  fileExtension: '.json.gz',
+});
+```
+
 ## Server-side Encryption
 
 Enabling server-side encryption (SSE) requires Amazon Data Firehose to encrypt all data

packages/@aws-cdk-testing/framework-integ/test/aws-kinesisfirehose/test/integ.s3-bucket.lit.js.snapshot/integtestsDefaultTestDeployAssert44C8D370.assets.json

@@ -4,12 +4,22 @@ import { DestinationBindOptions, DestinationConfig, IDestination } from './desti
 import * as iam from '../../aws-iam';
 import * as s3 from '../../aws-s3';
 import { createBackupConfig, createBufferingHints, createEncryptionConfig, createLoggingOptions, createProcessingConfig } from './private/helpers';
-import { UnscopedValidationError } from '../../core';
+import { Token, UnscopedValidationError, ValidationError } from '../../core';
 
 /**
  * Props for defining an S3 destination of an Amazon Data Firehose delivery stream.
  */
 export interface S3BucketProps extends CommonDestinationS3Props, CommonDestinationProps {
+  /**
+   * Specify a file extension.
+   * It will override the default file extension appended by Data Format Conversion or S3 compression features such as `.parquet` or `.gz`.
+   *
+   * File extension must start with a period (`.`) and can contain allowed characters: `0-9a-z!-_.*'()`.
+   *
+   * @see https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-s3
+   * @default - The default file extension appended by Data Format Conversion or S3 compression features
+   */
+  readonly fileExtension?: string;
 }
 
 /**
@@ -36,6 +46,17 @@ export class S3Bucket implements IDestination {
     }) ?? {};
 
     const { backupConfig, dependables: backupDependables } = createBackupConfig(scope, role, this.props.s3Backup) ?? {};
+
+    const fileExtension = this.props.fileExtension;
+    if (fileExtension && !Token.isUnresolved(fileExtension)) {
+      if (!fileExtension.startsWith('.')) {
+        throw new ValidationError("fileExtension must start with '.'", scope);
+      }
+      if (/[^0-9a-z!\-_.*'()]/.test(fileExtension)) {
+        throw new ValidationError("fileExtension can contain allowed characters: 0-9a-z!-_.*'()", scope);
+      }
+    }
+
     return {
       extendedS3DestinationConfiguration: {
         cloudWatchLoggingOptions: loggingOptions,
@@ -49,6 +70,7 @@ export class S3Bucket implements IDestination {
         encryptionConfiguration: createEncryptionConfig(role, this.props.encryptionKey),
         errorOutputPrefix: this.props.errorOutputPrefix,
         prefix: this.props.dataOutputPrefix,
+        fileExtension: this.props.fileExtension,
       },
       dependables: [bucketGrant, ...(loggingDependables ?? []), ...(backupDependables ?? [])],
     };

packages/@aws-cdk-testing/framework-integ/test/aws-kinesisfirehose/test/integ.s3-bucket.lit.js.snapshot/integtestsDefaultTestDeployAssert44C8D370.template.json

@@ -597,4 +597,65 @@ describe('S3 destination', () => {
       });
     });
   });
+
+  describe('file extension', () => {
+    it('sets fileExtension', () => {
+      const destination = new firehose.S3Bucket(bucket, {
+        role: destinationRole,
+        fileExtension: '.json',
+      });
+      new firehose.DeliveryStream(stack, 'DeliveryStream', {
+        destination: destination,
+      });
+
+      Template.fromStack(stack).hasResourceProperties('AWS::KinesisFirehose::DeliveryStream', {
+        ExtendedS3DestinationConfiguration: {
+          FileExtension: '.json',
+        },
+      });
+    });
+
+    it('sets fileExtension from a token', () => {
+      const fileExtension = new cdk.CfnParameter(stack, 'FileExtension');
+      const destination = new firehose.S3Bucket(bucket, {
+        role: destinationRole,
+        fileExtension: fileExtension.valueAsString,
+      });
+      new firehose.DeliveryStream(stack, 'DeliveryStream', {
+        destination: destination,
+      });
+
+      Template.fromStack(stack).hasResourceProperties('AWS::KinesisFirehose::DeliveryStream', {
+        ExtendedS3DestinationConfiguration: {
+          FileExtension: { Ref: 'FileExtension' },
+        },
+      });
+    });
+
+    it('throws when fileExtension does not start with a period', () => {
+      const destination = new firehose.S3Bucket(bucket, {
+        role: destinationRole,
+        fileExtension: 'json',
+      });
+
+      expect(() => {
+        new firehose.DeliveryStream(stack, 'DeliveryStream', {
+          destination: destination,
+        });
+      }).toThrow("fileExtension must start with '.'");
+    });
+
+    it('throws when fileExtension contains unallowed characters', () => {
+      const destination = new firehose.S3Bucket(bucket, {
+        role: destinationRole,
+        fileExtension: '.json?',
+      });
+
+      expect(() => {
+        new firehose.DeliveryStream(stack, 'DeliveryStream', {
+          destination: destination,
+        });
+      }).toThrow("fileExtension can contain allowed characters: 0-9a-z!-_.*'()");
+    });
+  });
 });