Repository: gruntwork-io/cloud-nuke
Pull request #574 (2023-08-29 23:23:47)

Add Amazon Managed Streaming for Apache Kafka
<!-- Prepend '[WIP]' to the title if this PR is still a work-in-progress. Remove it when it is ready for review! -->

## Description

Fixes https://github.com/gruntwork-io/cloud-nuke/issues/444
Continuation of this PR - https://github.com/gruntwork-io/cloud-nuke/pull/489

<!-- Description of the changes introduced by this PR. -->

## TODOs

Read the [Gruntwork contribution guidelines](https://gruntwork.notion.site/Gruntwork-Coding-Methodology-02fdcd6e4b004e818553684760bf691e).

- [ ] Update the docs.
- [ ] Run the relevant tests successfully, including pre-commit checks.
- [ ] Ensure any 3rd party code adheres with our [license policy](https://www.notion.so/gruntwork/Gruntwork-licenses-and-open-source-usage-policy-f7dece1f780341c7b69c1763f22b1378) or delete this line if its not applicable.
- [ ] Include release notes. If this PR is backward incompatible, include a migration guide.
- [ ] Attention Grunts - if this PR adds support for a new resource, ensure the `nuke_sandbox` and `nuke_phxdevops` jobs in `.circleci/config.yml` have been updated with appropriate exclusions (either directly in the job or via the `.circleci/nuke_config.yml` file) to prevent nuking IAM roles, groups, resources, etc that are important for the test accounts.


## Release Notes (draft)

<!-- One-line description of the PR that can be included in the final release notes. -->
Added / Removed / Updated [X].

### Migration Guide

<!-- Important: If you made any backward incompatible changes, then you must write a migration guide! -->




Patch:
diff --git a/README.md b/README.md
index ea88077..634c090 100644
--- a/README.md
+++ b/README.md
@@ -490,6 +490,7 @@ of the file that are supported are listed here.
 | lc                          | LaunchConfiguration          | ✅ (Launch Configuration Name)         | ✅ (Created Time)                    | ❌    |
 | lt                          | LaunchTemplate               | ✅ (Launch Template Name)              | ✅ (Created Time)                    | ❌    |
 | macie-member                | MacieMember                  | ❌                                     | ✅ (Creation Time)                   | ❌    |
+| msk-cluster                 | MskCluster                   | ✅ (Cluster Name)                      | ✅ (Creation Time)                   | ❌    |
 | nat-gateway                 | NatGateway                   | ✅ (EC2 Name Tag)                      | ✅ (Creation Time)                   | ❌    |
 | oidcprovider                | OIDCProvider                 | ✅ (Provider URL)                      | ✅ (Creation Time)                   | ❌    |
 | opensearchdomain            | OpenSearchDomain             | ✅ (Domain Name)                       | ✅ (First Seen Tag Time)             | ❌    |
diff --git a/aws/resources/msk_cluster.go b/aws/resources/msk_cluster.go
new file mode 100644
index 0000000..838965b
--- /dev/null
+++ b/aws/resources/msk_cluster.go
@@ -0,0 +1,64 @@
+package resources
+
+import (
+	"github.com/aws/aws-sdk-go/service/kafka"
+	"github.com/gruntwork-io/cloud-nuke/config"
+	"github.com/gruntwork-io/cloud-nuke/logging"
+	"github.com/gruntwork-io/cloud-nuke/report"
+)
+
+func (m MSKCluster) getAll(configObj config.Config) ([]*string, error) {
+	var clusterIDs []*string
+
+	err := m.Client.ListClustersV2Pages(&kafka.ListClustersV2Input{}, func(page *kafka.ListClustersV2Output, lastPage bool) bool {
+		for _, cluster := range page.ClusterInfoList {
+			if m.shouldInclude(cluster, configObj) {
+				clusterIDs = append(clusterIDs, cluster.ClusterArn)
+			}
+		}
+		return !lastPage
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return clusterIDs, nil
+}
+
+func (m MSKCluster) shouldInclude(cluster *kafka.Cluster, configObj config.Config) bool {
+	if *cluster.State == kafka.ClusterStateDeleting {
+		return false
+	}
+
+	// if cluster is still creating, skip it as it will only throw an error when attempting to delete it
+	// BadRequestException: You can't delete cluster in CREATING state.
+	if *cluster.State == kafka.ClusterStateCreating {
+		return false
+	}
+
+	return configObj.MSKCluster.ShouldInclude(config.ResourceValue{
+		Name: cluster.ClusterName,
+		Time: cluster.CreationTime,
+	})
+}
+
+func (m MSKCluster) nukeAll(identifiers []string) error {
+	for _, clusterArn := range identifiers {
+		_, err := m.Client.DeleteCluster(&kafka.DeleteClusterInput{
+			ClusterArn: &clusterArn,
+		})
+		if err != nil {
+			logging.Logger.Errorf("[Failed] %s", err)
+		}
+
+		// Record status of this resource
+		e := report.Entry{
+			Identifier:   clusterArn,
+			ResourceType: "MSKCluster",
+			Error:        err,
+		}
+		report.Record(e)
+	}
+
+	return nil
+}
diff --git a/aws/resources/msk_cluster_types.go b/aws/resources/msk_cluster_types.go
new file mode 100644
index 0000000..4cb024e
--- /dev/null
+++ b/aws/resources/msk_cluster_types.go
@@ -0,0 +1,57 @@
+package resources
+
+import (
+	awsgo "github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/kafka"
+	"github.com/aws/aws-sdk-go/service/kafka/kafkaiface"
+	"github.com/gruntwork-io/cloud-nuke/config"
+	"github.com/gruntwork-io/go-commons/errors"
+)
+
+// MSKCluster - represents all AWS Managed Streaming for Kafka clusters that should be deleted.
+type MSKCluster struct {
+	Client      kafkaiface.KafkaAPI
+	Region      string
+	ClusterArns []string
+}
+
+func (msk MSKCluster) Init(session *session.Session) {
+	msk.Client = kafka.New(session)
+}
+
+// ResourceName - the simple name of the aws resource
+func (msk MSKCluster) ResourceName() string {
+	return "msk-cluster"
+}
+
+// ResourceIdentifiers - The instance ids of the AWS Managed Streaming for Kafka clusters
+func (msk MSKCluster) ResourceIdentifiers() []string {
+	return msk.ClusterArns
+}
+
+func (msk MSKCluster) MaxBatchSize() int {
+	// Tentative batch size to ensure AWS doesn't throttle. Note that nat gateway does not support bulk delete, so
+	// we will be deleting this many in parallel using go routines. We conservatively pick 10 here, both to limit
+	// overloading the runtime and to avoid AWS throttling with many API calls.
+	return 10
+}
+
+func (msk MSKCluster) GetAndSetIdentifiers(configObj config.Config) ([]string, error) {
+	identifiers, err := msk.getAll(configObj)
+	if err != nil {
+		return nil, err
+	}
+
+	msk.ClusterArns = awsgo.StringValueSlice(identifiers)
+	return msk.ClusterArns, nil
+}
+
+// Nuke - nuke 'em all!!!
+func (msk MSKCluster) Nuke(_ *session.Session, identifiers []string) error {
+	if err := msk.nukeAll(identifiers); err != nil {
+		return errors.WithStackTrace(err)
+	}
+
+	return nil
+}
diff --git a/config/config.go b/config/config.go
index c66186c..757b34a 100644
--- a/config/config.go
+++ b/config/config.go
@@ -61,6 +61,7 @@ type Config struct {
 	LaunchConfiguration          ResourceType               `yaml:"LaunchConfiguration"`
 	LaunchTemplate               ResourceType               `yaml:"LaunchTemplate"`
 	MacieMember                  ResourceType               `yaml:"MacieMember"`
+	MSKCluster                   ResourceType               `yaml:"MSKCluster"`
 	NatGateway                   ResourceType               `yaml:"NatGateway"`
 	OIDCProvider                 ResourceType               `yaml:"OIDCProvider"`
 	OpenSearchDomain             ResourceType               `yaml:"OpenSearchDomain"`
diff --git a/go.mod b/go.mod
index 8f499fd..fbe10be 100644
--- a/go.mod
+++ b/go.mod
@@ -3,39 +3,23 @@ module github.com/gruntwork-io/cloud-nuke
 go 1.18
 
 require (
-	github.com/aws/aws-sdk-go v1.44.154
-	github.com/aws/aws-sdk-go-v2/config v1.17.1
-	github.com/aws/aws-sdk-go-v2/service/efs v1.17.10
-	github.com/aws/aws-sdk-go-v2/service/kinesis v1.17.8
-	github.com/aws/aws-sdk-go-v2/service/sns v1.17.13
+	github.com/aws/aws-sdk-go v1.44.304
+	github.com/aws/aws-sdk-go-v2 v1.18.1
 	github.com/charmbracelet/lipgloss v0.6.0
-	github.com/golang/mock v1.6.0
 	github.com/gruntwork-io/go-commons v0.17.0
 	github.com/gruntwork-io/gruntwork-cli v0.7.0
 	github.com/gruntwork-io/terratest v0.41.0
 	github.com/hashicorp/go-multierror v1.1.1
-	github.com/pquerna/otp v1.3.0
 	github.com/pterm/pterm v0.12.45
 	github.com/sirupsen/logrus v1.8.1
 	github.com/stretchr/testify v1.8.1
 	github.com/urfave/cli/v2 v2.10.3
-	golang.org/x/crypto v0.1.0
 	gopkg.in/yaml.v2 v2.4.0
 )
 
 require (
 	atomicgo.dev/cursor v0.1.1 // indirect
 	atomicgo.dev/keyboard v0.2.8 // indirect
-	github.com/aws/aws-sdk-go-v2 v1.17.7 // indirect
-	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.10 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.12.14 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.12 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.31 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.25 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.19 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.12 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.11.17 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.16.13 // indirect
 	github.com/aws/smithy-go v1.13.5 // indirect
 	github.com/boombuler/barcode v1.0.1-0.20190219062509-6c824513bacc // indirect
 	github.com/containerd/console v1.0.3 // indirect
@@ -55,11 +39,13 @@ require (
 	github.com/muesli/reflow v0.3.0 // indirect
 	github.com/muesli/termenv v0.11.1-0.20220204035834-5ac8409525e0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/pquerna/otp v1.3.0 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/urfave/cli v1.22.12 // indirect
 	github.com/xo/terminfo v0.0.0-20210125001918-ca9a967f8778 // indirect
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
+	golang.org/x/crypto v0.1.0 // indirect
 	golang.org/x/exp v0.0.0-20221106115401-f9659909a136 // indirect
 	golang.org/x/net v0.7.0 // indirect
 	golang.org/x/sys v0.5.0 // indirect
diff --git a/go.sum b/go.sum
index 3c93851..36f87a3 100644
--- a/go.sum
+++ b/go.sum
@@ -13,40 +13,10 @@ github.com/MarvinJWendt/testza v0.3.0/go.mod h1:eFcL4I0idjtIx8P9C6KkAuLgATNKpX4/
 github.com/MarvinJWendt/testza v0.4.2 h1:Vbw9GkSB5erJI2BPnBL9SVGV9myE+XmUSFahBGUhW2Q=
 github.com/MarvinJWendt/testza v0.4.2/go.mod h1:mSdhXiKH8sg/gQehJ63bINcCKp7RtYewEjXsvsVUPbE=
 github.com/atomicgo/cursor v0.0.1/go.mod h1:cBON2QmmrysudxNBFthvMtN32r3jxVRIvzkUiF/RuIk=
-github.com/aws/aws-sdk-go v1.44.154 h1:2TaEC8JIM/t7Fu+FBmdOBK5jFUAVL07tbpfJsLuVo3Q=
-github.com/aws/aws-sdk-go v1.44.154/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=
-github.com/aws/aws-sdk-go-v2 v1.16.11/go.mod h1:WTACcleLz6VZTp7fak4EO5b9Q4foxbn+8PIz3PmyKlo=
-github.com/aws/aws-sdk-go-v2 v1.17.7 h1:CLSjnhJSTSogvqUGhIC6LqFKATMRexcxLZ0i/Nzk9Eg=
-github.com/aws/aws-sdk-go-v2 v1.17.7/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw=
-github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.10 h1:dK82zF6kkPeCo8J1e+tGx4JdvDIQzj7ygIoLg8WMuGs=
-github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.10/go.mod h1:VeTZetY5KRJLuD/7fkQXMU6Mw7H5m/KP2J5Iy9osMno=
-github.com/aws/aws-sdk-go-v2/config v1.17.1 h1:BWxTjokU/69BZ4DnLrZco6OvBDii6ToEdfBL/y5I1nA=
-github.com/aws/aws-sdk-go-v2/config v1.17.1/go.mod h1:uOxDHjBemNTF2Zos+fgG0NNfE86wn1OAHDTGxjMEYi0=
-github.com/aws/aws-sdk-go-v2/credentials v1.12.14 h1:AtVG/amkjbDBfnPr/tuW2IG18HGNznP6L12Dx0rLz+Q=
-github.com/aws/aws-sdk-go-v2/credentials v1.12.14/go.mod h1:opAndTyq+YN7IpVG57z2CeNuXSQMqTYxGGlYH0m0RMY=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.12 h1:wgJBHO58Pc1V1QAnzdVM3JK3WbE/6eUF0JxCZ+/izz0=
-github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.12/go.mod h1:aZ4vZnyUuxedC7eD4JyEHpGnCz+O2sHQEx3VvAwklSE=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.18/go.mod h1:348MLhzV1GSlZSMusdwQpXKbhD7X2gbI/TxwAPKkYZQ=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.31 h1:sJLYcS+eZn5EeNINGHSCRAwUJMFVqklwkH36Vbyai7M=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.31/go.mod h1:QT0BqUvX1Bh2ABdTGnjqEjvjzrCfIniM9Sc8zn9Yndo=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.12/go.mod h1:ckaCVTEdGAxO6KwTGzgskxR1xM+iJW4lxMyDFVda2Fc=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.25 h1:1mnRASEKnkqsntcxHaysxwgVoUUp5dkiB+l3llKnqyg=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.25/go.mod h1:zBHOPwhBc3FlQjQJE/D3IfPWiWaQmT06Vq9aNukDo0k=
-github.com/aws/aws-sdk-go-v2/internal/ini v1.3.19 h1:g5qq9sgtEzt2szMaDqQO6fqKe026T6dHTFJp5NsPzkQ=
-github.com/aws/aws-sdk-go-v2/internal/ini v1.3.19/go.mod h1:cVHo8KTuHjShb9V8/VjH3S/8+xPu16qx8fdGwmotJhE=
-github.com/aws/aws-sdk-go-v2/service/efs v1.17.10 h1:3dvqOTkKIdO61/yeXD9m6diukbeSo2TJmjp/HSHOFXI=
-github.com/aws/aws-sdk-go-v2/service/efs v1.17.10/go.mod h1:f1PVX6O2FYtO5ujTK6ewawqGUDYIH7keEWnbYStZVtM=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.12 h1:7iPTTX4SAI2U2VOogD7/gmHlsgnYSgoNHt7MSQXtG2M=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.12/go.mod h1:1TODGhheLWjpQWSuhYuAUWYTCKwEjx2iblIFKDHjeTc=
-github.com/aws/aws-sdk-go-v2/service/kinesis v1.17.8 h1:9Kk24woetm1Tm4cAZNoJStJW1VQAeh92lLD9XZ4176g=
-github.com/aws/aws-sdk-go-v2/service/kinesis v1.17.8/go.mod h1:bXLOKN0GA128n13XAfBHlpO3hOkmmtCjZrp2aFtLjzQ=
-github.com/aws/aws-sdk-go-v2/service/sns v1.17.13 h1:sa8NDFztt68pihEfE31LhX+nJ1wDBJHcFh3T6crluDo=
-github.com/aws/aws-sdk-go-v2/service/sns v1.17.13/go.mod h1:yE3hE9v3YRRI9Rsl38kYJ4fyZ6vKSljaZ+28W5xzqgM=
-github.com/aws/aws-sdk-go-v2/service/sso v1.11.17 h1:pXxu9u2z1UqSbjO9YA8kmFJBhFc1EVTDaf7A+S+Ivq8=
-github.com/aws/aws-sdk-go-v2/service/sso v1.11.17/go.mod h1:mS5xqLZc/6kc06IpXn5vRxdLaED+jEuaSRv5BxtnsiY=
-github.com/aws/aws-sdk-go-v2/service/sts v1.16.13 h1:dl8T0PJlN92rvEGOEUiD0+YPYdPEaCZK0TqHukvSfII=
-github.com/aws/aws-sdk-go-v2/service/sts v1.16.13/go.mod h1:Ru3QVMLygVs/07UQ3YDur1AQZZp2tUNje8wfloFttC0=
-github.com/aws/smithy-go v1.12.1/go.mod h1:Tg+OJXh4MB2R/uN61Ko2f6hTZwB/ZYGOtib8J3gBHzA=
+github.com/aws/aws-sdk-go v1.44.304 h1:crcJBVeewWcVAXDQChzJWZYGFq9i1TYLycAtQ6Xpi4c=
+github.com/aws/aws-sdk-go v1.44.304/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=
+github.com/aws/aws-sdk-go-v2 v1.18.1 h1:+tefE750oAb7ZQGzla6bLkOwfcQCEtC5y2RqoqCeqKo=
+github.com/aws/aws-sdk-go-v2 v1.18.1/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw=
 github.com/aws/smithy-go v1.13.5 h1:hgz0X/DX0dGqTYpGALqXJoRKRj5oQ7150i5FdTePzO8=
 github.com/aws/smithy-go v1.13.5/go.mod h1:Tg+OJXh4MB2R/uN61Ko2f6hTZwB/ZYGOtib8J3gBHzA=
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
@@ -68,8 +38,6 @@ github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxI
 github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og=
 github.com/go-sql-driver/mysql v1.5.0 h1:ozyZYNQW3x3HtqT1jira07DN2PArx2v7/mN66gGcHOs=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
-github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
-github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
 github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
@@ -170,37 +138,29 @@ github.com/xo/terminfo v0.0.0-20210125001918-ca9a967f8778 h1:QldyIu/L63oPpyvQmHg
 github.com/xo/terminfo v0.0.0-20210125001918-ca9a967f8778/go.mod h1:2MuV+tbUrU1zIOPMxZ5EncGwgmMJsa+9ucAQZXxsObs=
 github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 h1:bAn7/zixMGCfxrRTfdpNzjtPYqr8smhKouy9mxVdGPU=
 github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673/go.mod h1:N3UwUGtsrSj3ccvlPHLoLsHnpR27oXr4ZE984MbSER8=
-github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU=
 golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
 golang.org/x/exp v0.0.0-20221106115401-f9659909a136 h1:Fq7F/w7MAa1KJ5bt2aJ62ihqp9HDcRuyILskkpIAurw=
 golang.org/x/exp v0.0.0-20221106115401-f9659909a136/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
-golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
-golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211013075003-97ac67df715c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -225,11 +185,8 @@ golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=


Test patch:
diff --git a/aws/resources/msk_cluster_test.go b/aws/resources/msk_cluster_test.go
new file mode 100644
index 0000000..090e8ff
--- /dev/null
+++ b/aws/resources/msk_cluster_test.go
@@ -0,0 +1,232 @@
+package resources
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/kafka"
+	"github.com/aws/aws-sdk-go/service/kafka/kafkaiface"
+	"github.com/gruntwork-io/cloud-nuke/config"
+)
+
+type mockMSKClient struct {
+	kafkaiface.KafkaAPI
+	listClustersV2PagesFn func(input *kafka.ListClustersV2Input, callback func(*kafka.ListClustersV2Output, bool) bool) error
+	deleteClusterFn       func(input *kafka.DeleteClusterInput) (*kafka.DeleteClusterOutput, error)
+}
+
+func (m mockMSKClient) ListClustersV2Pages(input *kafka.ListClustersV2Input, callback func(*kafka.ListClustersV2Output, bool) bool) error {
+	return m.listClustersV2PagesFn(input, callback)
+}
+
+func (m mockMSKClient) DeleteCluster(input *kafka.DeleteClusterInput) (*kafka.DeleteClusterOutput, error) {
+	return nil, nil
+}
+
+func TestListMSKClustersSingle(t *testing.T) {
+	mockMskClient := mockMSKClient{
+		listClustersV2PagesFn: func(input *kafka.ListClustersV2Input, callback func(*kafka.ListClustersV2Output, bool) bool) error {
+			callback(&kafka.ListClustersV2Output{
+				ClusterInfoList: []*kafka.Cluster{
+					{
+						ClusterArn:   aws.String("arn:aws:kafka:us-east-1:123456789012:cluster/test-cluster-1/1a2b3c4d-5e6f-7g8h-9i0j-1k2l3m4n5o6p"),
+						ClusterName:  aws.String("test-cluster-1"),
+						CreationTime: aws.Time(time.Now()),
+						State:        aws.String(kafka.ClusterStateActive),
+					},
+				},
+			}, true)
+			return nil
+		},
+	}
+
+	msk := MSKCluster{
+		Client: &mockMskClient,
+	}
+
+	clusterIDs, err := msk.getAll(config.Config{})
+	if err != nil {
+		t.Fatalf("Unable to list MSK Clusters: %v", err)
+	}
+
+	if len(clusterIDs) != 1 {
+		t.Fatalf("Expected 1 cluster, got %d", len(clusterIDs))
+	}
+
+	if *clusterIDs[0] != "arn:aws:kafka:us-east-1:123456789012:cluster/test-cluster-1/1a2b3c4d-5e6f-7g8h-9i0j-1k2l3m4n5o6p" {
+		t.Fatalf("Unexpected cluster ID: %s", *clusterIDs[0])
+	}
+}
+
+func TestListMSKClustersMultiple(t *testing.T) {
+	mockMskClient := mockMSKClient{
+		listClustersV2PagesFn: func(input *kafka.ListClustersV2Input, callback func(*kafka.ListClustersV2Output, bool) bool) error {
+			callback(&kafka.ListClustersV2Output{
+				ClusterInfoList: []*kafka.Cluster{
+					{
+						ClusterArn:   aws.String("arn:aws:kafka:us-east-1:123456789012:cluster/test-cluster-1/1a2b3c4d-5e6f-7g8h-9i0j-1k2l3m4n5o6p"),
+						ClusterName:  aws.String("test-cluster-1"),
+						CreationTime: aws.Time(time.Now()),
+						State:        aws.String(kafka.ClusterStateActive),
+					}, {
+						ClusterArn:   aws.String("arn:aws:kafka:us-east-1:123456789012:cluster/test-cluster-2/1a2b3c4d-5e6f-7g8h-9i0j-1k2l3m4n5o6p"),
+						ClusterName:  aws.String("test-cluster-2"),
+						CreationTime: aws.Time(time.Now()),
+						State:        aws.String(kafka.ClusterStateActive),
+					}, {
+						ClusterArn:   aws.String("arn:aws:kafka:us-east-1:123456789012:cluster/test-cluster-3/1a2b3c4d-5e6f-7g8h-9i0j-1k2l3m4n5o6p"),
+						ClusterName:  aws.String("test-cluster-3"),
+						CreationTime: aws.Time(time.Now()),
+						State:        aws.String(kafka.ClusterStateActive),
+					},
+				},
+			}, true)
+			return nil
+		},
+	}
+
+	msk := MSKCluster{
+		Client: &mockMskClient,
+	}
+
+	clusterIDs, err := msk.getAll(config.Config{})
+	if err != nil {
+		t.Fatalf("Unable to list MSK Clusters: %v", err)
+	}
+
+	if len(clusterIDs) != 3 {
+		t.Fatalf("Expected 3 clusters, got %d", len(clusterIDs))
+	}
+
+	for i := range clusterIDs {
+		prefix := fmt.Sprintf("arn:aws:kafka:us-east-1:123456789012:cluster/test-cluster-%d", i+1)
+		if !strings.HasPrefix(*clusterIDs[i], prefix) {
+			t.Fatalf("Unexpected cluster ID: %s", *clusterIDs[i])
+		}
+	}
+}
+
+func TestGetAllMSKError(t *testing.T) {
+	mockMskClient := mockMSKClient{
+		listClustersV2PagesFn: func(input *kafka.ListClustersV2Input, callback func(*kafka.ListClustersV2Output, bool) bool) error {
+			return fmt.Errorf("Error listing MSK Clusters")
+		},
+	}
+
+	msk := MSKCluster{
+		Client: &mockMskClient,
+	}
+
+	_, err := msk.getAll(config.Config{})
+	if err == nil {
+		t.Fatalf("Expected error listing MSK Clusters")
+	}
+}
+
+func TestShouldIncludeMSKCluster(t *testing.T) {
+	clusterName := "test-cluster"
+	creationTime := time.Now()
+
+	tests := map[string]struct {
+		cluster   kafka.Cluster
+		configObj config.Config
+		expected  bool
+	}{
+		"cluster is in deleting state": {
+			cluster: kafka.Cluster{
+				ClusterName:  &clusterName,
+				State:        aws.String(kafka.ClusterStateDeleting),
+				CreationTime: &creationTime,
+			},
+			configObj: config.Config{},
+			expected:  false,
+		},
+		"cluster is in creating state": {
+			cluster: kafka.Cluster{
+				ClusterName:  &clusterName,
+				State:        aws.String(kafka.ClusterStateCreating),
+				CreationTime: &creationTime,
+			},
+			configObj: config.Config{},
+			expected:  false,
+		},
+		"cluster is in active state": {
+			cluster: kafka.Cluster{
+				ClusterName:  &clusterName,
+				State:        aws.String(kafka.ClusterStateActive),
+				CreationTime: &creationTime,
+			},
+			configObj: config.Config{},
+			expected:  true,
+		},
+		"cluster excluded by name": {
+			cluster: kafka.Cluster{
+				ClusterName:  &clusterName,
+				State:        aws.String(kafka.ClusterStateActive),
+				CreationTime: &creationTime,
+			},
+			configObj: config.Config{
+				MSKCluster: config.ResourceType{
+					ExcludeRule: config.FilterRule{
+						NamesRegExp: []config.Expression{
+							{
+								RE: *regexp.MustCompile("test-cluster"),
+							},
+						},
+					},
+				},
+			},
+			expected: false,
+		},
+		"cluster included by name": {
+			cluster: kafka.Cluster{
+				ClusterName:  &clusterName,
+				State:        aws.String(kafka.ClusterStateActive),
+				CreationTime: &creationTime,
+			},
+			configObj: config.Config{
+				MSKCluster: config.ResourceType{
+					IncludeRule: config.FilterRule{
+						NamesRegExp: []config.Expression{
+							{
+								RE: *regexp.MustCompile("test-cluster"),
+							},
+						},
+					},
+				},
+			},
+			expected: true,
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			msk := MSKCluster{}
+			actual := msk.shouldInclude(&tc.cluster, tc.configObj)
+			if actual != tc.expected {
+				t.Fatalf("Expected %v, got %v", tc.expected, actual)
+			}
+		})
+	}
+}
+
+func TestNukeMSKCluster(t *testing.T) {
+	mockMskClient := mockMSKClient{
+		deleteClusterFn: func(input *kafka.DeleteClusterInput) (*kafka.DeleteClusterOutput, error) {
+			return nil, nil
+		},
+	}
+
+	msk := MSKCluster{
+		Client: &mockMskClient,
+	}
+
+	err := msk.Nuke(nil, []string{})
+	if err != nil {
+		t.Fatalf("Unable to nuke MSK Clusters: %v", err)
+	}
+}
diff --git a/config/config_test.go b/config/config_test.go
index 0813f09..3a77281 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -74,6 +74,7 @@ func emptyConfig() *Config {
 		ResourceType{FilterRule{}, FilterRule{}},
 		ResourceType{FilterRule{}, FilterRule{}},
 		ResourceType{FilterRule{}, FilterRule{}},
+		ResourceType{FilterRule{}, FilterRule{}},
 	}
 }