Adding authorizer to hosts.yml breaks installation

[fpunzohig]

Hello all. I am trying to complete a poc in which we are implementing a secure Kafka cluster. So far I have SSL and SASL Plain working without issue. The last step is to add ACLs to our "sensitive data" topics so only certain users can see them. I have attached our working hosts.yml file at the bottom for reference. This creates a working installation. However with this config, issuing the kafka-acls command results in the following error (below).

$sudo kafka-acls --command-config /etc/kafka/client.properties --bootstrap-server :9092 --add --allow-principal User: user1 --operation read --operation write --topic SensitiveTopic Adding ACLs for resource ResourcePattern(resourceType=TOPIC, name=SensitiveTopic, patternType=LITERAL): (principal=User:, host=, operation=READ, permissionType=ALLOW) (principal=User:, host=, operation=WRITE, permissionType=ALLOW)

Error while executing ACL command: org.apache.kafka.common.errors.SecurityDisabledException: No Authorizer is configured on the broker. java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SecurityDisabledException: No Authorizer is configured on the broker. 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.SecurityDisabledException: No Authorizer is configured on the broker.

To try and fix this, I added the following custom broker properties to the hosts.yml. You can see where I added it in the Working hosts.yml below (it's commented out).

kafka_broker_custom_properties:
  authorizer.class.name: kafka.security.authorizer.AclAuthorizer
  super.users: User:admin;User:schema_registry    

<<<

However, when I add these custom properties, the ansible install fails with the following error in the ansible play and all of the client commands fail with socket timeout errors :

TASK [confluent.kafka_broker : Get Topics with UnderReplicatedPartitions] ****** Wednesday 10 November 2021 20:56:44 +0000 (0:00:01.206) 0:03:52.054 **** FAILED - RETRYING: Get Topics with UnderReplicatedPartitions (15 retries left). FAILED - RETRYING: Get Topics with UnderReplicatedPartitions (15 retries left). FAILED - RETRYING: Get Topics with UnderReplicatedPartitions (15 retries left). <<<

The server.log output is the following

[2021-11-10 15:58:15,117] ERROR Error checking or creating metrics topic (io.confluent.metrics.reporter.ConfluentMetricsReporter) org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1636577895112, tries=1, nextAllowedTryMs=1636577895213) timed out at 1636577895113 after 1 attempt(s) Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics <<<

Question: can someone help me with the configuration to add to hosts.yml which will allow me to utilize the kafka-acls command to add acls? This is for a 1 way tls with sasl plain setup?

Working hosts.yml

---

#https://docs.confluent.io/ansible/current/overview.html #https://www.confluent.io/blog/confluent-platform-installation-with-cp-ansible/?_ga=2.40712673.2134511676.1635690336-1807251802.1605349031&_gac=1.245501616.1635417349.EAIaIQobChMIjPWqovTs8wIVmoCGCh3LTgSREAAYASAAEgI8MPD_BwE all: vars: ansible_connection: ssh ansible_user: ansible_become: true ansible_ssh_private_key_file:

sasl_protocol: plain
sasl_plain_users:
  admin:
    principal: <admin user>
    password: <admin secret>
  schema_registry:
    principal: <schema user>
    password: <schema secret>
  kafka_connect:
    principal: <connect user>
    password: <connect secret>
  ksql:
    principal: <ksqldb user>
    password: <ksqldb secret>
  kafka_rest:
    principal: <rest user>
    password: <rest secret>
  control_center:
    principal: <control user>
    password: <control secret>
  kafka_connect_replicator:
    principal: <replicator user>
    password: <replicator secret>
  client:
    principal: <client user>
    password: <client secret>
  user1:
    principal: <user1 user>
    password: <user1 secret>
ssl_enabled: true
zookeeper_custom_properties:
    4lw.commands.whitelist: stat, ruok, conf, isro

kafka_broker_custom_properties:

authorizer.class.name: kafka.security.authorizer.AclAuthorizer

super.users: User:admin;User:schema_registry

zookeeper: hosts: : : :

kafka_broker: hosts: : : :

schema_registry: hosts: : kafka_rest: hosts: : ksql: hosts: : kafka_connect: hosts: : control_center: hosts: : <<<