++ skip kafka acls on ugmk broker without authorizer

infrastructure.yaml

@@ -56,6 +56,10 @@ environments:
     buckets: []
 
     kafka:
+      # Broker has no authorizer configured (authorizer.class.name unset) —
+      # any ACL write is rejected with "Security features are disabled".
+      # Topics/users still work; skip ACLs until the broker enables one.
+      create_acls: false
       kafka_cluster_refs:
         main:
           bootstrap_servers: kafka-kafka-contour.kafka.svc.cluster.local:9092

live/kafka-topics/terragrunt.hcl

@@ -49,4 +49,5 @@ inputs = {
   users              = try(local.env_config.kafka.users, [])
   kafka_cluster_refs = try(local.env_config.kafka.kafka_cluster_refs, {})
   kafka_policy       = try(local.env_config.kafka.kafka_policy, {})
+  create_acls        = try(local.env_config.kafka.create_acls, true)
 }

modules/kafka-topics-incluster/main.tf

@@ -178,7 +178,7 @@ resource "kafka_user_scram_credential" "this" {
 }
 
 resource "kafka_acl" "this" {
-  for_each = var.create_users ? local.acl_entries : {}
+  for_each = var.create_users && var.create_acls ? local.acl_entries : {}
 
   resource_name                = each.value.resource_name
   resource_type                = each.value.resource_type

modules/kafka-topics-incluster/variables.tf

@@ -76,6 +76,12 @@ variable "create_users" {
   default     = true
 }
 
+variable "create_acls" {
+  description = "Create ACL entries. Some brokers run without an authorizer configured (authorizer.class.name unset) and reject any ACL write with \"Security features are disabled\" — set false on those contours."
+  type        = bool
+  default     = true
+}
+
 variable "default_user_roles" {
   description = "Kafka roles granted to generated owner users on their topics."
   type        = list(string)