Terraform provider for managing Apache Kafka Topics + ACLs
terraform-provider-kafka
A [Terraform][1] plugin for managing [Apache Kafka][2].
Contents
* Developing *kafkatopic
* kafkaacl
* kafkaquota
Installation
terraform-provider-kafka is available on the terraform registry. To install, add the below into your main.tf and execute terraform init
terraform {
required_providers {
kafka = {
source = "Mongey/kafka"
}
}
}
provider "kafka" { bootstrap_servers = ["localhost:9092"] ca_cert = file("../secrets/ca.crt") client_cert = file("../secrets/terraform-cert.pem") client_key = file("../secrets/terraform.pem") tls_enabled = true }
Otherwise, install by downloading and extracting the latest release to your [terraform plugin directory][third-party-plugins] (typically ~/.terraform.d/plugins/)
Developing
- [Install go][install-go]
- Clone repository to:
$GOPATH/src/github.com/Mongey/terraform-provider-kafka
bash
mkdir -p $GOPATH/src/github.com/Mongey/terraform-provider-kafka; cd $GOPATH/src/github.com/Mongey/
git clone https://github.com/Mongey/terraform-provider-kafka.git
cd terraform-provider-kafka- Build the provider
make build - Run the tests
make test - Start a TLS enabled kafka-cluster
docker-compose up - Run the acceptance tests
make testacc
Provider Configuration
Example
Example provider with TLS client authentication.
provider "kafka" { bootstrap_servers = ["localhost:9092"] ca_cert = file("../secrets/ca.crt") client_cert = file("../secrets/terraform-cert.pem") client_key = file("../secrets/terraform.pem") tls_enabled = true }Example provider with aws-iam(Assume role) client authentication.
provider "kafka" { bootstrap_servers = ["localhost:9098"] tls_enabled = true sasl_mechanism = "aws-iam" saslawsregion = "us-east-1" saslawsrole_arn = "arn:aws:iam::account:role/role-name" }Example provider with aws-iam(Aws Profile) client authentication.
provider "kafka" { bootstrap_servers = ["localhost:9098"] tls_enabled = true sasl_mechanism = "aws-iam" saslawsregion = "us-east-1" saslawsprofile = "dev" }Example provider with aws-iam(Aws Profile in non-default awssharedconfig_file path) client authentication.
provider "kafka" { bootstrap_servers = ["localhost:9098"] tls_enabled = true sasl_mechanism = "aws-iam" saslawsregion = "us-east-1" saslawsprofile = "dev" saslawssharedconfigfiles = ["/path/to/custom/aws/config"] }Example provider with aws-iam(Static Creds) client authentication using explicit credentials.
provider "vault" { authloginjwt { role = "jwt-role-name" } }
data "vaultawsaccess_credentials" "creds" { backend = "aws" type = "sts" role = "sts-role-name" }
provider "kafka" { bootstrap_servers = ["localhost:9098"] tls_enabled = true sasl_mechanism = "aws-iam" saslawsregion = "us-east-1" saslawsaccesskey = data.vaultawsaccesscredentials.creds.access_key saslawssecretkey = data.vaultawsaccesscredentials.creds.secret_key saslawstoken = data.vaultawsaccesscredentials.creds.securitytoken }
Example provider with aws-iam(Static Creds) client authentication. You have to export AWSACCESSKEYID, AWSSECRETACCESSKEY, AWSSESSIONTOKEN(Optional if you are using temp creds)
provider "kafka" { bootstrap_servers = ["localhost:9098"] tls_enabled = true sasl_mechanism = "aws-iam" saslawsregion = "us-east-1" }Example provider with aws-iam(Container Creds) client authentication. You have to export AWSCONTAINERAUTHORIZATIONTOKENFILE and AWSCONTAINERCREDENTIALSFULLURI
provider "kafka" { bootstrap_servers = ["localhost:9098"] tls_enabled = true sasl_mechanism = "aws-iam" saslawsregion = "us-east-1" }Compatibility with Redpanda
provider "kafka" {
bootstrap_servers = ["localhost:9092"]
kafka_version = "2.1.0"
}Due to Redpanda not implementing some Metadata APIs, we need to force the Kafka version to use when creating the provider.
| Property | Description | Default | | ------------------- | --------------------------------------------------------------------------------------------------------------------- | ---------- | | bootstrap_servers | A list of host:port addresses that will be used to discover the full set of alive brokers | Required | | ca_cert | The CA certificate or path to a CA certificate file in PEM format to validate the server's certificate. | "" | | clientcert | The client certificate or path to a file containing the client certificate in PEM format. Use for Client authentication to Kafka.
If you have Intermediate CA certificate(s) append them to clientcert.| "" | | client_key | The private key or path to a file containing the private key that the client certificate was issued for. | "" | | clientkeypassphrase | The passphrase for the private key that the certificate was issued for. | "" | | kafka_version | The version of Kafka protocol to use in $MAJOR.$MINOR.$PATCH format. Some features may not be available on older versions. | "" | | tls_enabled | Enable communication with the Kafka Cluster over TLS. | true | | skiptlsverify | Skip TLS verification. | false | | sasl_username | Username for SASL authentication. | "" | | sasl_password | Password for SASL authentication. | "" | | sasl_mechanism | Mechanism for SASL authentication. Allowed values are plain, aws-iam, scram-sha256, scram-sha512 or oauthbearer | plain | | saslawsregion | AWS region for IAM authentication. | "" | | saslawscontainerauthorizationtoken_file | Path to a file containing the AWS pod identity authorization token. | "" | | saslawscontainercredentialsfull_uri | URI to retrieve AWS credentials from. | "" | | saslawsrole_arn | Arn of AWS IAM role to assume for IAM authentication. | "" | | saslawsprofile | AWS profile to use for IAM authentication. | "" | | saslawssharedconfigfiles | List of paths to AWS shared config files | "" | | saslawsaccess_key | AWS access key. | "" | | saslawssecret_key | AWS secret key. | "" | | saslawstoken | AWS session token. | "" | | saslawscreds_debug | Enable debug logging for AWS authentication. | false | | sasltokenurl | The url to retrieve oauth2 tokens from, when using sasl mechanism oauthbearer | "" | | sasloauthscopes | OAuth scopes to request when using the oauthbearer mechanism | [] |
Resources
kafka_topic
A resource for managing Kafka topics. Increases partition count without destroying the topic.
Example
provider "kafka" {
bootstrap_servers = ["localhost:9092"]
}
resource "kafka_topic" "logs" { name = "systemd_logs" replication_factor = 2 partitions = 100
config = { "segment.ms" = "20000" "cleanup.policy" = "compact" } }
Properties
| Property | Description | | -------------------- | ---------------------------------------------- | | name | The name of the topic | | partitions | The number of partitions the topic should have | | replication_factor | The number of replicas the topic should have | | config | A map of string [K/V attributes][topic-config] |
Importing Existing Topics
You can import topics with the followingterraform import kafkatopic.logs systemdlogskafka_acl
A resource for managing Kafka ACLs.
Example
provider "kafka" {
bootstrap_servers = ["localhost:9092"]
ca_cert = file("../secrets/ca.crt")
client_cert = file("../secrets/terraform-cert.pem")
client_key = file("../secrets/terraform.pem")
}
resource "kafka_acl" "test" { resource_name = "syslog" resource_type = "Topic" acl_principal = "User:Alice" acl_host = "*" acl_operation = "Write" aclpermissiontype = "Deny" }
Properties
| Property | Description | Valid values | | ------------------------------ | ------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------- | | acl_principal | Principal that is being allowed or denied | * | | aclhost | Host from which principal listed in aclprincipal will have access | * | | acl_operation | Operation that is being allowed or denied | Unknown, Any, All, Read, Write, Create, Delete, Alter, Describe, ClusterAction, DescribeConfigs, AlterConfigs, IdempotentWrite | | aclpermissiontype | Type of permission | Unknown, Any, Allow, Deny | | resource_name | The name of the resource | * | | resource_type | The type of resource | Unknown, Any, Topic, Group, Cluster, TransactionalID,DelegationToken | | resourcepatterntype_filter | | Prefixed, Any, Match, Literal |
Importing Existing ACLs
For import, use as a parameter the items separated by| character. Quote it to avoid shell expansion.
# Fields in shell notation are
${aclprincipal}|${aclhost}|${acloperation}|${aclpermissiontype}|${resourcetype}|${resourcename}|${resourcepatterntypefilter}
terraform import kafka_acl.admin 'User:12345|*|Describe|Allow|Topic|experimental-topic|Prefixed'kafka_quota
A resource for managing Kafka Quotas.
Example
provider "kafka" {
bootstrap_servers = ["localhost:9092"]
ca_cert = file("../secrets/ca.crt")
client_cert = file("../secrets/terraform-cert.pem")
client_key = file("../secrets/terraform.pem")
}
resource "kafka_quota" "test" { entity_name = "client1" entity_type = "client-id" config = { "consumerbyterate" = "4000000" "producerbyterate" = "3500000" } }
resource "kafkaquota" "defaultuser_quota" { entity_type = "user" config = { "consumerbyterate" = "2000000" "producerbyterate" = "1500000" } }
Properties
| Property | Description | | -------------------- | --------------------------------------------------------------------------------------------------- | | entityname | The name of the entity (if entityname is not provided, it will create entity-default Kafka quota) | | entity_type | The entity type (client-id, user, ip) | | config | A map of string attributes for the entity |
kafkauserscram_credential
A resource for managing Kafka SCRAM user credentials.
Example
provider "kafka" {
bootstrap_servers = ["localhost:9092"]
ca_cert = file("../secrets/ca.crt")
client_cert = file("../secrets/terraform-cert.pem")
client_key = file("../secrets/terraform.pem")
}
Legacy usage with 'password' (deprecated)
resource "kafkauserscram_credential" "test" {
username = "user1"
scram_mechanism = "SCRAM-SHA-256"
scram_iterations = "8192"
password = "password"
}
Recommended usage with write-only password (Terraform 1.11+). Password isn't stored in tfstate anymore
resource "kafkauserscram_credential" "secure" {
username = "user2"
scram_mechanism = "SCRAM-SHA-256"
scram_iterations = "8192"
password_wo = "secure-password"
passwordwoversion = "1"
}You can fill passwordwoversion with your secret engine metadata. For example, Hashicorp Vault returns it in the [data source][secret-version].
Importing Existing SCRAM user credentials
For import, use as a parameter the items separated by| character. Quote it to avoid shell expansion.
# Fields in shell notation are
${username}|${scram_mechanism}|${password} (legacy format)
or
${username}|${scram_mechanism} (for write-only passwords)
terraform import kafkauserscram_credential.test 'user1|SCRAM-SHA-256|password'
or for write-only passwords (passwordwo and passwordwo_version must be set manually after import)
terraform import kafkauserscram_credential.test 'user1|SCRAM-SHA-256'Properties
| Property | Description | | -------------------- | ---------------------------------------------- | | username | The username | | scram_mechanism | The SCRAM mechanism (SCRAM-SHA-256 or SCRAM-SHA-512) | | scram_iterations | The number of SCRAM iterations (must be >= 4096). Default: 4096 | | password | The password for the user (deprecated, use password_wo instead) | | password_wo | The write-only password for the user (recommended, requires Terraform 1.11+) | | passwordwoversion | Version identifier for the write-only password to track changes |
Note: Either password or passwordwo must be specified, but not both. The passwordwo field is recommended for better security as it's write-only and never returned by the API.
Common Issues and Troubleshooting
Provider Crashes
If you encounter "Empty Summary" errors or nil pointer dereferences, common causes include:- Empty
bootstrap_serverslist - ensure you always provide valid broker addresses - Insufficient IAM permissions when using AWS MSK - see the AWS MSK Integration Guide
- Attempting to modify immutable properties on MSK Serverless
AWS MSK Authentication
For IAM authentication issues:- Ensure you're using the correct port (9098 for IAM, 9096 for SASL/SCRAM)
- For EKS/ECS, set
saslawsrole_arn = ""to use pod/task credentials - Check your IAM policy includes necessary
kafka-cluster:*permissions
Dynamic Configuration
The provider requiresbootstrap_servers at initialization time. For dynamic environments:
- Consider using separate Terraform workspaces/states
- Use
countorfor_eachon resources instead of conditional provider configuration
Documentation
- Quick Start Guide - Get started quickly with common scenarios
- Authentication Guide - Detailed authentication configuration
- AWS MSK Integration - Complete MSK setup guide
- Troubleshooting Guide - Common issues and solutions
Requirements
- [>= Kafka 1.0.0][3]
