MSK - failed to create ACL and Topic using SASL/SCRAM about terraform-provider-kafka

[mbuotidem]

It's possible that your sasl user doesn't have the privileges yet. Try running

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=xxxxxxx:2181,xxxxxxx:2181,xxxxxxx:2181 --add --allow-principal User:xxxxxxx --operation All --cluster '*'

Once it works, remember to scope down the permissions. You probably only need the Alter, Describe and Create operations.

[qq304635576]

@rssaini01 pls set up as below:

provider "msk" {
bootstrap_servers = var.msk_kafka_brokers
tls_enabled = true
skip_tls_verify = true
sasl_username = local.raw_data.username
sasl_password = local.raw_data.password
sasl_mechanism = "scram-sha512"
}

[thennati]

Getting an error when trying to create ACL or Topic to MSK cluster using SASL/SCRAM auth and also public access is on.. so I'm using public endpoint of brokers

provider "kafka" {
  bootstrap_servers = ["<msk-cluster-sasl-public-brokers>"]
  sasl_username  = "xxxxxxxx"
  sasl_password  = "xxxxxxxxx"
  sasl_mechanism = "scram-sha512"
  tls_enabled = true
}

resource "kafka_acl" "acl_1" {
  resource_name                = "test_topic_terraform_acl"
  resource_type                = "Cluster"
  acl_principal                = "User:CN=kafka-dev"
  acl_host                     = "*"
  acl_operation                = "Any"
  acl_permission_type          = "Allow"
}

resource "kafka_topic" "topic_1" {
  name               = "test_topic_terraform"
  replication_factor = 1
  partitions         = 3
}

getting the below error for ACL

Error: kafka server: failed to create one or more ACL rules: kafka server: The client is not authorized to send this request type

this one for topic

Error: kafka server: The client is not authorized to access this topic
instead of resource_name = "test_topic_terraform_acl" try using resource_name= "kafka-cluster"

[rssaini01]

But When I'm creating ACLs then getting org.apache.kafka.common.errors.ClusterAuthorizationException error. I have configured EC2 instance to configure MSK using SASL/SCRAM.

Error while executing ACL command: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(processor=3, connectionId=10.100.101.9:9096-10.100.101.89:44912-0, session=Session(User:kafka-dev,/10.100.101.89), listenerName=ListenerName(CLIENT_SASL_SCRAM), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=62 lim=62 cap=62], envelope=None) is not authorized.
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(processor=3, connectionId=10.100.101.9:9096-10.100.101.89:44912-0, session=Session(User:kafka-dev,/10.100.101.89), listenerName=ListenerName(CLIENT_SASL_SCRAM), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=62 lim=62 cap=62], envelope=None) is not authorized.
        at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
        at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
        at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
        at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
        at kafka.admin.AclCommand$AdminClientService.$anonfun$addAcls$3(AclCommand.scala:112)
        at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:563)
        at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:561)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:919)
        at scala.collection.IterableOps$WithFilter.foreach(Iterable.scala:889)
        at kafka.admin.AclCommand$AdminClientService.$anonfun$addAcls$1(AclCommand.scala:109)
        at kafka.admin.AclCommand$AdminClientService.addAcls(AclCommand.scala:108)
        at kafka.admin.AclCommand$.main(AclCommand.scala:70)
        at kafka.admin.AclCommand.main(AclCommand.scala)
Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(processor=3, connectionId=10.100.101.9:9096-10.100.101.89:44912-0, session=Session(User:kafka-dev,/10.100.101.89), listenerName=ListenerName(CLIENT_SASL_SCRAM), securityProtocol=SASL_SSL, buffer=java.nio.HeapByteBuffer[pos=62 lim=62 cap=62], envelope=None) is not authorized.

[qq304635576]

@rssaini01 I can create & destroy ACL via SASL_SSL, as shown below:
Terraform will perform the following actions:

kafka_acl.brokertopic will be destroyed

(because kafka_acl.brokertopic is not in configuration)

• resource "kafka_acl" "brokertopic" {
  • acl_host = "*" -> null
  • acl_operation = "All" -> null
  • acl_permission_type = "Allow" -> null
  • acl_principal = "User:broker" -> null
  • id = "User:broker|*|All|Allow|Topic|TEST_|Prefixed" -> null
  • resource_name = "TEST_" -> null
  • resource_pattern_type_filter = "Prefixed" -> null
  • resource_type = "Topic" -> null
    }

Plan: 0 to add, 0 to change, 1 to destroy.

Do you want to perform these actions?
Terraform will perform the actions described above.
Only 'yes' will be accepted to approve.

Enter a value: yes

kafka_acl.brokertopic: Destroying... [id=User:broker|*|All|Allow|Topic|TEST_|Prefixed]
kafka_acl.brokertopic: Destruction complete after 3s
Releasing state lock. This may take a few moments...

Apply complete! Resources: 0 added, 0 changed, 1 destroyed.