MSK connectors should have logging enabled

How to fix it

• Replaces the resource: Logging is fixed when a connector is created: create a new connector with log delivery from the same plugin and configuration, confirm it is running, then delete this one.
• Affects another check: Encryption in transit is also fixed at creation, so give the new connector TLS encryption as well, which MSK.3 asks for.

In the console

Managed Streaming for Apache Kafka → MSK Connect → Connectors → Create connector

Log delivery

Deliver to Amazon CloudWatch Logs, Amazon S3 or Amazon Data Firehose; any one passes

Log group

An existing CloudWatch Logs group, created first, when CloudWatch Logs is the destination

With the AWS CLI

aws kafkaconnect create-connector --connector-name <new-connector-name> --kafka-connect-version <kafka-connect-version> --capacity <capacity-json> --connector-configuration <connector-configuration-json> --kafka-cluster <kafka-cluster-json> --kafka-cluster-client-authentication authenticationType=<authentication-type> --kafka-cluster-encryption-in-transit encryptionType=TLS --log-delivery '{"workerLogDelivery":{"cloudWatchLogs":{"enabled":true,"logGroup":"<log-group-name>"}}}' --plugins <plugins-json> --service-execution-role-arn <service-execution-role-arn> --region <region>

With Terraform

• Interrupts service: In Terraform log_delivery forces a new connector, so adding it to this connector's block deletes the connector before its replacement runs.

resource "aws_mskconnect_connector" "this" {
  name                       = var.new_connector_name
  kafkaconnect_version       = var.kafkaconnect_version
  connector_configuration    = var.connector_configuration
  service_execution_role_arn = var.service_execution_role_arn

  capacity {
    provisioned_capacity {
      worker_count = var.worker_count
    }
  }

  kafka_cluster {
    apache_kafka_cluster {
      bootstrap_servers = var.bootstrap_servers_tls

      vpc {
        security_groups = var.security_groups
        subnets         = var.subnets
      }
    }
  }

  kafka_cluster_client_authentication {
    authentication_type = var.authentication_type
  }

  kafka_cluster_encryption_in_transit {
    encryption_type = "TLS"
  }

  log_delivery {
    worker_log_delivery {
      cloudwatch_logs {
        enabled   = true
        log_group = var.log_group_name
      }
    }
  }

  plugin {
    custom_plugin {
      arn      = var.custom_plugin_arn
      revision = var.custom_plugin_revision
    }
  }
}