# AWS MSK Kafka Cluster Terraform module Terraform module which creates AWS MSK (Managed Streaming for Kafka) resources. [![SWUbanner](https://raw.githubusercontent.com/vshymanskyy/StandWithUkraine/main/banner2-direct.svg)](https://github.com/vshymanskyy/StandWithUkraine/blob/main/docs/README.md) ## Usage See [`examples`](https://github.com/terraform-aws-modules/terraform-aws-msk-kafka-cluster/tree/master/examples) directory for working examples to reference: ```hcl module "msk_kafka_cluster" { source = "terraform-aws-modules/msk-kafka-cluster/aws" name = local.name kafka_version = "3.5.1" number_of_broker_nodes = 3 enhanced_monitoring = "PER_TOPIC_PER_PARTITION" broker_node_client_subnets = ["subnet-12345678", "subnet-024681012", "subnet-87654321"] broker_node_storage_info = { ebs_storage_info = { volume_size = 100 } } broker_node_instance_type = "kafka.t3.small" broker_node_security_groups = ["sg-12345678"] encryption_in_transit_client_broker = "TLS" encryption_in_transit_in_cluster = true configuration_name = "example-configuration" configuration_description = "Example configuration" configuration_server_properties = { "auto.create.topics.enable" = true "delete.topic.enable" = true } jmx_exporter_enabled = true node_exporter_enabled = true cloudwatch_logs_enabled = true s3_logs_enabled = true s3_logs_bucket = "aws-msk-kafka-cluster-logs" s3_logs_prefix = local.name scaling_max_capacity = 512 scaling_target_value = 80 client_authentication = { sasl = { scram = true } } create_scram_secret_association = true scram_secret_association_secret_arn_list = [ aws_secretsmanager_secret.one.arn, aws_secretsmanager_secret.two.arn, ] # AWS Glue Registry schema_registries = { team_a = { name = "team_a" description = "Schema registry for Team A" } team_b = { name = "team_b" description = "Schema registry for Team B" } } # AWS Glue Schemas schemas = { team_a_tweets = { schema_registry_name = "team_a" schema_name = "tweets" description = "Schema that contains all the tweets" compatibility = "FORWARD" schema_definition = "{\"type\": \"record\", \"name\": \"r1\", \"fields\": [ {\"name\": \"f1\", \"type\": \"int\"}, {\"name\": \"f2\", \"type\": \"string\"} ]}" tags = { Team = "Team A" } } team_b_records = { schema_registry_name = "team_b" schema_name = "records" description = "Schema that contains all the records" compatibility = "FORWARD" team_b_records = { schema_registry_name = "team_b" schema_name = "records" description = "Schema that contains all the records" compatibility = "FORWARD" schema_definition = jsonencode({ type = "record" name = "r1" fields = [ { name = "f1" type = "int" }, { name = "f2" type = "string" }, { name = "f3" type = "boolean" } ] }) tags = { Team = "Team B" } } } } tags = { Terraform = "true" Environment = "dev" } } ``` ## Examples Examples codified under the [`examples`](https://github.com/terraform-aws-modules/terraform-aws-msk-kafka-cluster/tree/master/examples) are intended to give users references for how to use the module(s) as well as testing/validating changes to the source code of the module. If contributing to the project, please be sure to make any appropriate updates to the relevant examples to allow maintainers to test your changes and to keep the examples up to date for users. Thank you! - [Basic](https://github.com/terraform-aws-modules/terraform-aws-msk-kafka-cluster/tree/master/examples/basic) - [Complete](https://github.com/terraform-aws-modules/terraform-aws-msk-kafka-cluster/tree/master/examples/complete) - [Connect](https://github.com/terraform-aws-modules/terraform-aws-msk-kafka-cluster/tree/master/examples/connect) - [serverless](https://github.com/terraform-aws-modules/terraform-aws-msk-kafka-cluster/tree/master/examples/serverless) ## Requirements | Name | Version | | ---- | ------- | | [terraform](#requirement\_terraform) | >= 1.5.7 | | [aws](#requirement\_aws) | >= 6.40 | | [random](#requirement\_random) | >= 3.6 | ## Providers | Name | Version | | ---- | ------- | | [aws](#provider\_aws) | >= 6.40 | | [random](#provider\_random) | >= 3.6 | ## Modules No modules. ## Resources | Name | Type | | ---- | ---- | | [aws_appautoscaling_policy.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/appautoscaling_policy) | resource | | [aws_appautoscaling_target.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/appautoscaling_target) | resource | | [aws_cloudwatch_log_group.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/cloudwatch_log_group) | resource | | [aws_glue_registry.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/glue_registry) | resource | | [aws_glue_schema.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/glue_schema) | resource | | [aws_msk_cluster.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_cluster) | resource | | [aws_msk_cluster_policy.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_cluster_policy) | resource | | [aws_msk_configuration.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_configuration) | resource | | [aws_msk_scram_secret_association.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_scram_secret_association) | resource | | [aws_msk_topic.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_topic) | resource | | [aws_msk_vpc_connection.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_vpc_connection) | resource | | [aws_mskconnect_custom_plugin.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/mskconnect_custom_plugin) | resource | | [aws_mskconnect_worker_configuration.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/mskconnect_worker_configuration) | resource | | [random_id.this](https://registry.terraform.io/providers/hashicorp/random/latest/docs/resources/id) | resource | | [aws_iam_policy_document.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/data-sources/iam_policy_document) | data source | ## Inputs | Name | Description | Type | Default | Required | | ---- | ----------- | ---- | ------- | :------: | | [broker\_node\_az\_distribution](#input\_broker\_node\_az\_distribution) | The distribution of broker nodes across availability zones ([documentation](https://docs.aws.amazon.com/msk/1.0/apireference/clusters.html#clusters-model-brokerazdistribution)). Currently the only valid value is `DEFAULT` | `string` | `null` | no | | [broker\_node\_client\_subnets](#input\_broker\_node\_client\_subnets) | A list of subnets to connect to in client VPC ([documentation](https://docs.aws.amazon.com/msk/1.0/apireference/clusters.html#clusters-prop-brokernodegroupinfo-clientsubnets)) | `list(string)` | `[]` | no | | [broker\_node\_connectivity\_info](#input\_broker\_node\_connectivity\_info) | Information about the cluster access configuration |

object({
    vpc_connectivity = optional(object({
      client_authentication = optional(object({
        sasl = optional(object({
          iam   = optional(bool)
          scram = optional(bool)
        }))
        tls = optional(bool)
      }))
    }))
    public_access = optional(object({
      type = optional(string)
    }))
    network_type = optional(string)
  })

| `null` | no | | [broker\_node\_instance\_type](#input\_broker\_node\_instance\_type) | Specify the instance type to use for the kafka brokers. e.g. kafka.m5.large. ([Pricing info](https://aws.amazon.com/msk/pricing/)) | `string` | `null` | no | | [broker\_node\_security\_groups](#input\_broker\_node\_security\_groups) | A list of the security groups to associate with the elastic network interfaces to control who can communicate with the cluster | `list(string)` | `[]` | no | | [broker\_node\_storage\_info](#input\_broker\_node\_storage\_info) | A block that contains information about storage volumes attached to MSK broker nodes |

object({
    ebs_storage_info = optional(object({
      provisioned_throughput = optional(object({
        enabled           = optional(bool)
        volume_throughput = optional(number)
      }))
      volume_size = optional(number, 64)
    }))
  })

object({
    sasl = optional(object({
      iam   = optional(bool)
      scram = optional(bool)
    }))
    tls = optional(object({
      certificate_authority_arns = optional(list(string))
    }))
    unauthenticated = optional(bool)
  })

map(object({
    sid           = optional(string)
    actions       = optional(list(string))
    not_actions   = optional(list(string))
    effect        = optional(string)
    resources     = optional(list(string))
    not_resources = optional(list(string))
    principals = optional(list(object({
      type        = string
      identifiers = list(string)
    })))
    not_principals = optional(list(object({
      type        = string
      identifiers = list(string)
    })))
    condition = optional(list(object({
      test     = string
      values   = list(string)
      variable = string
    })))
  }))

| `null` | no | | [cluster\_source\_policy\_documents](#input\_cluster\_source\_policy\_documents) | Source policy documents for cluster policy | `list(string)` | `null` | no | | [configuration\_arn](#input\_configuration\_arn) | ARN of an externally created configuration to use | `string` | `null` | no | | [configuration\_description](#input\_configuration\_description) | Description of the configuration | `string` | `null` | no | | [configuration\_name](#input\_configuration\_name) | Name of the configuration | `string` | `null` | no | | [configuration\_revision](#input\_configuration\_revision) | Revision of the externally created configuration to use | `number` | `null` | no | | [configuration\_server\_properties](#input\_configuration\_server\_properties) | Contents of the server.properties file. Supported properties are documented in the [MSK Developer Guide](https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-properties.html) | `map(string)` | `{}` | no | | [connect\_custom\_plugins](#input\_connect\_custom\_plugins) | Map of custom plugin configuration details (map of maps) |

map(object({
    name              = string
    description       = optional(string)
    content_type      = string
    s3_bucket_arn     = string
    s3_file_key       = string
    s3_object_version = optional(string)
    timeouts = optional(object({
      create = optional(string)
      delete = optional(string)
    }))
  }))

| `{}` | no | | [connect\_worker\_config\_description](#input\_connect\_worker\_config\_description) | A summary description of the worker configuration | `string` | `null` | no | | [connect\_worker\_config\_name](#input\_connect\_worker\_config\_name) | The name of the worker configuration | `string` | `null` | no | | [connect\_worker\_config\_properties\_file\_content](#input\_connect\_worker\_config\_properties\_file\_content) | Contents of connect-distributed.properties file. The value can be either base64 encoded or in raw format | `string` | `null` | no | | [create](#input\_create) | Determines whether cluster resources will be created | `bool` | `true` | no | | [create\_cloudwatch\_log\_group](#input\_create\_cloudwatch\_log\_group) | Determines whether to create a CloudWatch log group | `bool` | `true` | no | | [create\_cluster\_policy](#input\_create\_cluster\_policy) | Determines whether to create an MSK cluster policy | `bool` | `false` | no | | [create\_configuration](#input\_create\_configuration) | Determines whether to create a configuration | `bool` | `true` | no | | [create\_connect\_worker\_configuration](#input\_create\_connect\_worker\_configuration) | Determines whether to create connect worker configuration | `bool` | `false` | no | | [create\_schema\_registry](#input\_create\_schema\_registry) | Determines whether to create a Glue schema registry for managing Avro schemas for the cluster | `bool` | `true` | no | | [create\_scram\_secret\_association](#input\_create\_scram\_secret\_association) | Determines whether to create SASL/SCRAM secret association | `bool` | `false` | no | | [enable\_storage\_autoscaling](#input\_enable\_storage\_autoscaling) | Determines whether autoscaling is enabled for storage | `bool` | `true` | no | | [encryption\_at\_rest\_kms\_key\_arn](#input\_encryption\_at\_rest\_kms\_key\_arn) | You may specify a KMS key short ID or ARN (it will always output an ARN) to use for encrypting your data at rest. If no key is specified, an AWS managed KMS ('aws/msk' managed service) key will be used for encrypting the data at rest | `string` | `null` | no | | [encryption\_in\_transit\_client\_broker](#input\_encryption\_in\_transit\_client\_broker) | Encryption setting for data in transit between clients and brokers. Valid values: `TLS`, `TLS_PLAINTEXT`, and `PLAINTEXT`. Default value is `TLS` | `string` | `null` | no | | [encryption\_in\_transit\_in\_cluster](#input\_encryption\_in\_transit\_in\_cluster) | Whether data communication among broker nodes is encrypted. Default value: `true` | `bool` | `null` | no | | [enhanced\_monitoring](#input\_enhanced\_monitoring) | Specify the desired enhanced MSK CloudWatch monitoring level. See [Monitoring Amazon MSK with Amazon CloudWatch](https://docs.aws.amazon.com/msk/latest/developerguide/monitoring.html) | `string` | `null` | no | | [firehose\_delivery\_stream](#input\_firehose\_delivery\_stream) | Name of the Kinesis Data Firehose delivery stream to deliver logs to | `string` | `null` | no | | [firehose\_logs\_enabled](#input\_firehose\_logs\_enabled) | Indicates whether you want to enable or disable streaming broker logs to Kinesis Data Firehose | `bool` | `false` | no | | [jmx\_exporter\_enabled](#input\_jmx\_exporter\_enabled) | Indicates whether you want to enable or disable the JMX Exporter | `bool` | `false` | no | | [kafka\_version](#input\_kafka\_version) | Specify the desired Kafka software version | `string` | `null` | no | | [name](#input\_name) | Name of the MSK cluster | `string` | `"msk"` | no | | [node\_exporter\_enabled](#input\_node\_exporter\_enabled) | Indicates whether you want to enable or disable the Node Exporter | `bool` | `false` | no | | [number\_of\_broker\_nodes](#input\_number\_of\_broker\_nodes) | The desired total number of broker nodes in the kafka cluster. It must be a multiple of the number of specified client subnets | `number` | `null` | no | | [rebalancing](#input\_rebalancing) | Configuration block for intelligent rebalancing |

object({
    status = string
  })

map(object({
    name        = string
    description = optional(string)
    tags        = optional(map(string), {})
  }))

| `{}` | no | | [schemas](#input\_schemas) | A map schemas to be created within the schema registry |

map(object({
    schema_name          = string
    schema_registry_name = string
    description          = optional(string)
    data_format          = optional(string, "AVRO")
    compatibility        = string
    schema_definition    = string
    tags                 = optional(map(string), {})
  }))

| `{}` | no | | [scram\_secret\_association\_secret\_arn\_list](#input\_scram\_secret\_association\_secret\_arn\_list) | List of AWS Secrets Manager secret ARNs to associate with SCRAM | `list(string)` | `[]` | no | | [storage\_mode](#input\_storage\_mode) | Controls storage mode for supported storage tiers. Valid values are: `LOCAL` or `TIERED` | `string` | `null` | no | | [tags](#input\_tags) | A map of tags to assign to the resources created | `map(string)` | `{}` | no | | [timeouts](#input\_timeouts) | Create, update, and delete timeout configurations for the cluster |

object({
    create = optional(string)
    update = optional(string)
    delete = optional(string)
  })

map(object({
    configs            = optional(string)
    name               = optional(string)
    partition_count    = number
    replication_factor = number
  }))

| `{}` | no | | [vpc\_connections](#input\_vpc\_connections) | Map of VPC Connections to create |

map(object({
    authentication  = string
    client_subnets  = list(string)
    security_groups = list(string)
    vpc_id          = string
    tags            = optional(map(string), {})
  }))

| `{}` | no | ## Outputs | Name | Description | | ---- | ----------- | | [appautoscaling\_policy\_arn](#output\_appautoscaling\_policy\_arn) | The ARN assigned by AWS to the scaling policy | | [appautoscaling\_policy\_name](#output\_appautoscaling\_policy\_name) | The scaling policy's name | | [appautoscaling\_policy\_policy\_type](#output\_appautoscaling\_policy\_policy\_type) | The scaling policy's type | | [arn](#output\_arn) | Amazon Resource Name (ARN) of the MSK cluster | | [bootstrap\_brokers](#output\_bootstrap\_brokers) | Comma separated list of one or more hostname:port pairs of Kafka brokers suitable to bootstrap connectivity to the Kafka cluster | | [bootstrap\_brokers\_plaintext](#output\_bootstrap\_brokers\_plaintext) | Comma separated list of one or more hostname:port pairs of Kafka brokers suitable to bootstrap connectivity to the Kafka cluster. Contains a value if `encryption_in_transit_client_broker` is set to `PLAINTEXT` or `TLS_PLAINTEXT` | | [bootstrap\_brokers\_public](#output\_bootstrap\_brokers\_public) | Comma separated list of one or more hostname:port pairs of Kafka brokers suitable to bootstrap connectivity to the Kafka cluster | | [bootstrap\_brokers\_public\_sasl\_iam](#output\_bootstrap\_brokers\_public\_sasl\_iam) | One or more DNS names (or IP addresses) and SASL IAM port pairs | | [bootstrap\_brokers\_public\_sasl\_scram](#output\_bootstrap\_brokers\_public\_sasl\_scram) | One or more DNS names (or IP addresses) and SASL SCRAM port pairs | | [bootstrap\_brokers\_public\_tls](#output\_bootstrap\_brokers\_public\_tls) | One or more DNS names (or IP addresses) and TLS port pairs | | [bootstrap\_brokers\_sasl\_iam](#output\_bootstrap\_brokers\_sasl\_iam) | One or more DNS names (or IP addresses) and SASL IAM port pairs. This attribute will have a value if `encryption_in_transit_client_broker` is set to `TLS_PLAINTEXT` or `TLS` and `client_authentication_sasl_iam` is set to `true` | | [bootstrap\_brokers\_sasl\_scram](#output\_bootstrap\_brokers\_sasl\_scram) | One or more DNS names (or IP addresses) and SASL SCRAM port pairs. This attribute will have a value if `encryption_in_transit_client_broker` is set to `TLS_PLAINTEXT` or `TLS` and `client_authentication_sasl_scram` is set to `true` | | [bootstrap\_brokers\_tls](#output\_bootstrap\_brokers\_tls) | One or more DNS names (or IP addresses) and TLS port pairs. This attribute will have a value if `encryption_in_transit_client_broker` is set to `TLS_PLAINTEXT` or `TLS` | | [bootstrap\_brokers\_vpc\_connectivity](#output\_bootstrap\_brokers\_vpc\_connectivity) | Comma separated list of one or more hostname:port pairs of Kafka brokers suitable to bootstrap connectivity to the Kafka cluster | | [bootstrap\_brokers\_vpc\_connectivity\_sasl\_iam](#output\_bootstrap\_brokers\_vpc\_connectivity\_sasl\_iam) | One or more DNS names (or IP addresses) and SASL IAM port pairs for VPC connectivity | | [bootstrap\_brokers\_vpc\_connectivity\_sasl\_scram](#output\_bootstrap\_brokers\_vpc\_connectivity\_sasl\_scram) | One or more DNS names (or IP addresses) and SASL SCRAM port pairs for VPC connectivity | | [bootstrap\_brokers\_vpc\_connectivity\_tls](#output\_bootstrap\_brokers\_vpc\_connectivity\_tls) | One or more DNS names (or IP addresses) and TLS port pairs for VPC connectivity | | [cluster\_name](#output\_cluster\_name) | Name of the MSK cluster | | [cluster\_uuid](#output\_cluster\_uuid) | UUID of the MSK cluster, for use in IAM policies | | [configuration\_arn](#output\_configuration\_arn) | Amazon Resource Name (ARN) of the configuration | | [configuration\_latest\_revision](#output\_configuration\_latest\_revision) | Latest revision of the configuration | | [connect\_custom\_plugins](#output\_connect\_custom\_plugins) | A map of output attributes for the connect custom plugins created | | [connect\_worker\_configuration\_arn](#output\_connect\_worker\_configuration\_arn) | The Amazon Resource Name (ARN) of the worker configuration | | [connect\_worker\_configuration\_latest\_revision](#output\_connect\_worker\_configuration\_latest\_revision) | An ID of the latest successfully created revision of the worker configuration | | [current\_version](#output\_current\_version) | Current version of the MSK Cluster used for updates, e.g. `K13V1IB3VIYZZH` | | [log\_group\_arn](#output\_log\_group\_arn) | The Amazon Resource Name (ARN) specifying the log group | | [schema\_registries](#output\_schema\_registries) | A map of output attributes for the schema registries created | | [schemas](#output\_schemas) | A map of output attributes for the schemas created | | [scram\_secret\_association\_id](#output\_scram\_secret\_association\_id) | Amazon Resource Name (ARN) of the MSK cluster | | [topics](#output\_topics) | A map of output attributes for the MSK topics created | | [vpc\_connections](#output\_vpc\_connections) | A map of output attributes for the VPC connections created | | [zookeeper\_connect\_string](#output\_zookeeper\_connect\_string) | A comma separated list of one or more hostname:port pairs to use to connect to the Apache Zookeeper cluster. The returned values are sorted alphabetically | | [zookeeper\_connect\_string\_tls](#output\_zookeeper\_connect\_string\_tls) | A comma separated list of one or more hostname:port pairs to use to connect to the Apache Zookeeper cluster via TLS. The returned values are sorted alphabetically | ## License Apache-2.0 Licensed. See [LICENSE](https://github.com/terraform-aws-modules/terraform-aws-msk-kafka-cluster/blob/master/LICENSE).

AWS MSK Kafka Cluster Terraform module

Terraform Module Source