Using Terraform to deploy Flink App or Statement¶

• Get the last Terraform cli version, from the installation guide..

  terraform version
  

• See Confluent Terraform sample project for an end-to-end tutorial.
• The Terraform Confluent provider formal documentation, with the deployment examples.

There are two approaches to manage Confluent Platform Kafka cluster in the same Terraform workspace:

1. Manage multiple Kafka clusters
2. Manage a single cluster

Terraform plugin supports Flink compute pool creation and Flink statement deployments on Confluent Cloud.

• This repository includes in deployment/cc-terraform Terraform definitions to create:

  • Confluent cloud environment
  • A Service Account as environement admin
  • Kafka Cluster with API key
  • A schema registry with API key
  • Flink service accounts: flink-app (with ClusterAdmin role, DeveloperRead and DeveloperWrite on schema registry), flink-developer-sa (FlinkDeveloper), Flink API key owned by Flink Developer SA
  • Flink compute pools: a default and one for data generations
  • Flink statements for clicks, customers, products data generation

  terraform validate
  

Pre-requisites¶

• If not done already, create Confluent Cloud API key with secret for your user or for a service account (See confluent.cloud/settings/api-keys). Your user needs OrganizationAdmin role to create those keys. For production do not use, user keys but prefer service account keys.

• If not already done create a service account for terraform runner. Assign the OrganizationAdmin role to this service account by following this guide.

• To get the visibility of the existing keys use the command:

  confluent api-key list | grep <cc_userid>
  

• Export as environment variables:

  export CONFLUENT_CLOUD_API_KEY=
  export CONFLUENT_CLOUD_API_SECRET=
  

• Recall that terraform.tfstate and env variables should not be committed to git.

Infrastructure¶

For any flink project, we need to define or import the following resources:

• Confluent cloud environment
• A Service Account as environement admin
• Kafka Cluster with API key
• A schema registry with API key
• Kafka Cluster
• Schema Registry
• Flink service accounts: flink-app (with ClusterAdmin role, DeveloperRead and DeveloperWrite on schema registry), flink-developer-sa (FlinkDeveloper), Flink API key owned by Flink Developer SA
• Flink compute pools: a default and one for data generations
• Flink statements for clicks, customers, products data generation

Defining Terraform Provider and Variables¶

1. Create a main.tf with the confluent cloud provider. (see Confluent sample here), or the main.tf in this repo

2. Init:

   terraform init
   

3. Add variables.tf: we need at least: confluent_cloud_api_key, confluent_cloud_api_secret, cloud_provider, cloud_region and prefix. Each variable definition has the same structure:

   variable "prefix" {
       description = "Prefix for resource names to avoid conflicts"
       type        = string
       default     = "j9r"
   }
   

4. Define specific variable values using: terraform.tfvars or use default values.

Iterative configuration¶

The approach is to add resources by increment to have a clear understanding of all dependencies. The iteration includes at least the following steps:

1. Add resources and variables
2. Decide to import existing resources or create new one
3. Run CLI:

   terraform validate
   terraform plan
   terraform apply
   

Environment¶

• Define the Confluent environment in env.tf, and use a naming convention as: "${var.prefix}-env".

  resource "confluent_environment" "env" {
  display_name = "${var.prefix}-env"
  
  stream_governance {
      package = "ADVANCED"
  }
  }
  

• For an existing environment, use the import statement:

  import {
      to = confluent_environment.env
      id = "env-nknqp3"
  }
  

• Add service account and role binding as Environment admin

  resource "confluent_service_account" "env-manager" {
      display_name = "${var.prefix}-env-manager"
      description  = "Service account to manage ${var.prefix}-env environment"
      depends_on = [
          confluent_environment.env
      ]
  }
  

  • or import existing service account id

    import {
        to = confluent_service_account.env-manager
        id = "sa-7vvrvw"
    }
    

Kafka¶

See the Product documentation to create kafka cluster with Terraform. The basic cluster sample project describes the needed steps, but it is recommended to use standard kafka cluster with RBAC access control. As an alternate this is the standard cluster definitions.

1. Create a Kafka cluster resources:

   resource "confluent_kafka_cluster" "standard" {
       display_name = "${var.prefix}-kafka"
       availability = "SINGLE_ZONE"
       cloud        = var.cloud_provider
       region       = var.cloud_region
       standard {}
   
       environment {
           id = confluent_environment.env.id
       }
         lifecycle {
               prevent_destroy = true
           }
   }
   

   or import one exisitng

   terraform import confluent_kafka_cluster.standard env-nknqp3/lkc-3mnm0m
   

This repository includes a demo with IaC definition in the deployment cc-terraform folder which defines the following components:

• A Service account to manage the environment: env_manager with the role of EnvironmentAdmin and API keys
• A Kafka cluster in a single AZ (for demo), with service accounts for app-manager, producer and consumer apps

Schema Registry¶

• A schema registry with API keys to access the registry at runtime

Compute pool¶

Flink Compute pool can also be configured with Terraform, and see this example in the Terraform Confluent quickstart repository.

The flink.tf in deployment cc-terraform: defines the following components:

• A flink pool, with 2 service accounts, one for flink app management and one for developing flink statements.

Deploy the configuration¶

• Use the classical tf commands:

  terraform init
  terraform plan
  terraform apply --auto-approve
  

• If there is a 401 error on accessing Confluent, it is a problem of api_key within the environment variables.

• The output of this configuration needs to be used by other deployment like the Flink statement ones. It can be retrieved at any time with:

  terraform output
  

Deploy a Flink statement¶

Flink statement can be deployed using Terraform: DDL is the easier part. Automation of DML statement is more complex: the first deployment on a fresh environment is simple, the challenges come when doing canary deployment on existing running statements for logic upgrade. For that see the statement life cycle chapter in the cookbook chapter. and potentially when the number of statements are important and need to take into account running statement, it is recommended to consider adopting shift_left util

See the Terraform definition of the Flink Statement resource.. One important constraint is that all 7 flink_api_key, flink_api_secret, flink_rest_endpoint, organization_id, environment_id, flink_compute_pool_id, flink_principal_id attributes should be set or not set in the provider block at the same time.

resource "confluent_flink_statement" "ddl-dim_tenant-v1" {
properties = {
    "sql.current-catalog"  = var.current_catalog
    "sql.current-database" = var.current_database
}

statement  = file("${path.module}/sql-scripts/ddl.dim_tenant.sql")
statement_name = "dml-dim-tenant_v1"

stopped = false  # change when updating the statement
}

statement can be a reference to a file, or a string including the statement content.

resource "confluent_flink_statement" "create-table" {
statement  = "CREATE TABLE ${local.table_name}(ts TIMESTAMP_LTZ(3), random_value INT);"