feat(msk): throw ValidationError instead of untyped errors (#34214)

### Issue # (if applicable)

`aws-msk-alpha` for #32569

### Description of changes
ValidationErrors everywhere


<!--
What code changes did you make? 
Have you made any important design decisions?
What AWS use cases does this change enable? To enable the use cases,
which AWS service features are utilized?
-->

### Describe any new or updated permissions being added
n/a
<!-- What new or updated IAM permissions are needed to support the
changes being introduced ? -->


### Description of how you validated changes
Existing tests. Exemptions granted as this is basically a refactor of
existing code.

<!--Have you added any unit tests and/or integration tests?-->

### 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: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

Author: mazyu36

packages/@aws-cdk/aws-msk-alpha/.eslintrc.js

@@ -4,5 +4,12 @@ baseConfig.parserOptions.project = __dirname + '/tsconfig.json';
 baseConfig.rules['import/no-extraneous-dependencies'] = ['error', { devDependencies: true, peerDependencies: true } ];
 baseConfig.rules['import/order'] = 'off';
 baseConfig.rules['@aws-cdk/invalid-cfn-imports'] = 'off';
+baseConfig.rules['@cdklabs/no-throw-default-error'] = ['error'];
+baseConfig.overrides.push({
+  files: ["./test/**"],
+  rules: {
+    "@cdklabs/no-throw-default-error": "off",
+  },
+});
 
 module.exports = baseConfig;

packages/@aws-cdk/aws-msk-alpha/lib/cluster.ts

@@ -46,7 +46,7 @@ export abstract class ClusterBase extends core.Resource implements ICluster {
   /** Manages connections for the cluster */
   public get connections(): ec2.Connections {
     if (!this._connections) {
-      throw new Error('An imported Cluster cannot manage its security groups');
+      throw new core.ValidationError('An imported Cluster cannot manage its security groups', this);
     }
     return this._connections;
   }
@@ -477,26 +477,24 @@ export class Cluster extends ClusterBase {
     });
 
     if (subnetSelection.subnets.length < 2) {
-      throw Error(`Cluster requires at least 2 subnets, got ${subnetSelection.subnets.length}`);
+      throw new core.ValidationError(`Cluster requires at least 2 subnets, got ${subnetSelection.subnets.length}`, this);
     }
 
     if (props.encryptionInTransit?.clientBroker === ClientBrokerEncryption.PLAINTEXT && props.clientAuthentication) {
-      throw Error('To enable client authentication, you must enabled TLS-encrypted traffic between clients and brokers.');
+      throw new core.ValidationError('To enable client authentication, you must enabled TLS-encrypted traffic between clients and brokers.', this);
     } else if (
       props.encryptionInTransit?.clientBroker ===
         ClientBrokerEncryption.TLS_PLAINTEXT &&
       (props.clientAuthentication?.saslProps?.scram ||
         props.clientAuthentication?.saslProps?.iam)
     ) {
-      throw Error(
-        'To enable SASL/SCRAM or IAM authentication, you must only allow TLS-encrypted traffic between clients and brokers.',
-      );
+      throw new core.ValidationError('To enable SASL/SCRAM or IAM authentication, you must only allow TLS-encrypted traffic between clients and brokers.', this);
     }
 
     const volumeSize = props.ebsStorageInfo?.volumeSize ?? 1000;
     // Minimum: 1 GiB, maximum: 16384 GiB
     if (volumeSize < 1 || volumeSize > 16384) {
-      throw Error('EBS volume size should be in the range 1-16384');
+      throw new core.ValidationError('EBS volume size should be in the range 1-16384', this);
     }
 
     const instanceType = props.instanceType
@@ -507,12 +505,12 @@ export class Cluster extends ClusterBase {
 
     if (props.storageMode && props.storageMode === StorageMode.TIERED) {
       if (!props.kafkaVersion.isTieredStorageCompatible()) {
-        throw Error(`To deploy a tiered cluster you must select a compatible Kafka version, got ${props.kafkaVersion.version}`);
+        throw new core.ValidationError(`To deploy a tiered cluster you must select a compatible Kafka version, got ${props.kafkaVersion.version}`, this);
       }
       if (instanceType === this.mskInstanceType(
         ec2.InstanceType.of(ec2.InstanceClass.T3, ec2.InstanceSize.SMALL),
       )) {
-        throw Error('Tiered storage doesn\'t support broker type t3.small');
+        throw new core.ValidationError('Tiered storage doesn\'t support broker type t3.small', this);
       }
     }
 
@@ -883,9 +881,7 @@ export class Cluster extends ClusterBase {
         installLatestAwsSdk: false,
       });
     } else {
-      throw Error(
-        'Cannot create users if an authentication KMS key has not been created/provided.',
-      );
+      throw new core.ValidationError('Cannot create users if an authentication KMS key has not been created/provided.', this);
     }
   }
 }

packages/@aws-cdk/aws-msk-alpha/lib/serverless-cluster.ts

@@ -1,5 +1,5 @@
 import * as ec2 from 'aws-cdk-lib/aws-ec2';
-import { Fn, Lazy, Names } from 'aws-cdk-lib';
+import { Fn, Lazy, Names, ValidationError } from 'aws-cdk-lib';
 import * as constructs from 'constructs';
 import { ClusterBase, ICluster } from '.';
 import { CfnServerlessCluster } from 'aws-cdk-lib/aws-msk';
@@ -87,7 +87,7 @@ export class ServerlessCluster extends ClusterBase {
     addConstructMetadata(this, props);
 
     if (props.vpcConfigs.length < 1 || props.vpcConfigs.length > 5) {
-      throw Error(`\`vpcConfigs\` must contain between 1 and 5 configurations, got ${props.vpcConfigs.length} configurations.`);
+      throw new ValidationError(`\`vpcConfigs\` must contain between 1 and 5 configurations, got ${props.vpcConfigs.length} configurations.`, this);
     }
 
     const vpcConfigs = props.vpcConfigs.map((vpcConfig, index) => this._renderVpcConfig(vpcConfig, index));
@@ -127,16 +127,14 @@ export class ServerlessCluster extends ClusterBase {
     const subnetSelection = vpcConfig.vpc.selectSubnets(vpcConfig.vpcSubnets);
 
     if (subnetSelection.subnets.length < 2) {
-      throw Error(
-        `Cluster requires at least 2 subnets, got ${subnetSelection.subnets.length} subnet.`,
-      );
+      throw new ValidationError(`Cluster requires at least 2 subnets, got ${subnetSelection.subnets.length} subnet.`, this);
     }
 
     let securityGroups: ec2.ISecurityGroup[] = [];
 
     if (vpcConfig.securityGroups) {
       if (vpcConfig.securityGroups.length < 1 || vpcConfig.securityGroups.length > 5) {
-        throw Error(`\`securityGroups\` must contain between 1 and 5 elements, got ${vpcConfig.securityGroups.length} elements.`);
+        throw new ValidationError(`\`securityGroups\` must contain between 1 and 5 elements, got ${vpcConfig.securityGroups.length} elements.`, this);
       }
       securityGroups = vpcConfig.securityGroups;
     } else {