# Beginner - trying to enable ACLS on a 3 broker kraft cluster **URL:** \ **Category:** Architecture and Design\ **Created:** [16 April 2024 15:57 UTC](https://forum.confluent.io/t/beginner-trying-to-enable-acls-on-a-3-broker-kraft-cluster/10521 "2024-04-16T15:57:53Z")\ **Posts on this page:** 4\ **Page:** 1
**Author:** ![shardservant](https://avatars.discourse-cdn.com/v4/letter/s/f0a364/32.png) [@shardservant](https://forum.confluent.io/u/shardservant)\ **Post date:** [16 April 2024 15:57 UTC](https://forum.confluent.io/t/beginner-trying-to-enable-acls-on-a-3-broker-kraft-cluster/10521/1 "2024-04-16T15:57:53Z")
Running kafka\_2.13-3.7.0 on RHES 8.9, I created a three broker Kraft cluster. In each of the server property files for the brokers, I added the following and restarted. authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer allow.everyone.if.no.acl.found=true I now see the serializer failed to start and the broker crashes. Any idea what I am doing wrong? I am under the presumption that I do not need SASL for basic user matching. I just want to only permit based on the incoming uid. No SSL has been configured yet. ---
**Author:** ![mmuehlbeyer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mmuehlbeyer/32/1088_2.png) [@mmuehlbeyer](https://forum.confluent.io/u/mmuehlbeyer)\ **Post date:** [20 April 2024 19:33 UTC](https://forum.confluent.io/t/beginner-trying-to-enable-acls-on-a-3-broker-kraft-cluster/10521/2 "2024-04-20T19:33:53Z")
could you elaborate what you would like to do? authenticate by uid and setting an acl? best, michael ---
**Author:** ![shardservant](https://avatars.discourse-cdn.com/v4/letter/s/f0a364/32.png) [@shardservant](https://forum.confluent.io/u/shardservant)\ **Post date:** [22 April 2024 12:19 UTC](https://forum.confluent.io/t/beginner-trying-to-enable-acls-on-a-3-broker-kraft-cluster/10521/3 "2024-04-22T12:19:52Z")
I set up a three broker kraft cluster all on the same host. I tried enabling the StandardAuthorizer so I could set acl values for specific topics. When I do this, the standard authorizer fails to start and the broker crashes. Below is one of the property files. Thoughts? # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the “License”); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # [Apache License, Version 2.0 | Apache Software Foundations](http://www.apache.org/licenses/LICENSE-2.0) # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an “AS IS” BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # This configuration file is intended for use in KRaft mode, where # Apache ZooKeeper is not present. See config/kraft/README.md for details. # ############################# Server Basics ############################# # The role of this server. Setting this puts us in KRaft mode process.roles=broker,controller # The node id associated with this instance’s roles node.id=1 # The connect string for the controller quorum controller.quorum.voters=1@bosukafkbrkrd01:19092,2@bosukafkbrkrd01:19093,3@bosukafkbrkrd01:19094 ############################# Socket Server Settings ############################# # The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured. # FORMAT: # listeners = listener\_name://host\_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 listeners=SSL://bosukafkbrkrd01:9092,CONTROLLER://:19092 ssl.keystore.location=/opt/kafka/kafka-poc.mfs.com/kafka-poc.mfs.com.jks ssl.keystore.password=KeepMeSecure ssl.key.password=KeepMeSecure ssl.truststore.location=/opt/kafka/kafka-poc.mfs.com/kafka-poc.mfs.com.p12 ssl.truststore.password=KeepMeSecure ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1 ssl.client.auth=required inter.broker.listener.name=SSL # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for “listeners” if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). advertised.listeners=SSL://bosukafkbrkrd01:9092,CONTROLLER://bosukafkbrkrd01:19092 # Listener, host name, and port for the controller to advertise to the brokers. If # this server is a controller, this listener must be configured. controller.listener.names=CONTROLLER # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL\_PLAINTEXT:SASL\_PLAINTEXT,SASL\_SSL:SASL\_SSL # The number of threads that the server uses for receiving requests from the network and sending responses to the network num.network.threads=3 # The number of threads that the server uses for processing requests, which may include disk I/O num.io.threads=8 # The send buffer (SO\_SNDBUF) used by the socket server socket.send.buffer.bytes=102400 # The receive buffer (SO\_RCVBUF) used by the socket server socket.receive.buffer.bytes=102400 # The maximum size of a request that the socket server will accept (protection against OOM) socket.request.max.bytes=104857600 ############################# Log Basics ############################# # A comma separated list of directories under which to store log files log.dirs=/opt/kafka/kraft-combined-logs-1 # The default number of log partitions per topic. More partitions allow greater # parallelism for consumption, but this will also result in more files across # the brokers. num.partitions=1 # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. # This value is recommended to be increased for installations with data dirs located in RAID array. num.recovery.threads.per.data.dir=1 ############################# Internal Topic Settings ############################# # The replication factor for the group metadata internal topics “\_\_consumer\_offsets” and “\_\_transaction\_state” # For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3. offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 ############################# Log Flush Policy ############################# # Messages are immediately written to the filesystem but by default we only fsync() to sync # the OS cache lazily. The following configurations control the flush of data to disk. # There are a few important trade-offs here: # 1. Durability: Unflushed data may be lost if you are not using replication. # 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush. # 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks. # The settings below allow one to configure the flush policy to flush data after a period of time or # every N messages (or both). This can be done globally and overridden on a per-topic basis. # The number of messages to accept before forcing a flush of data to disk #log.flush.interval.messages=10000 # The maximum amount of time a message can sit in a log before we force a flush #log.flush.interval.ms=1000 ############################# Log Retention Policy ############################# # The following configurations control the disposal of log segments. The policy can # be set to delete segments after a period of time, or after a given size has accumulated. # A segment will be deleted whenever _either_ of these criteria are met. Deletion always happens # from the end of the log. # The minimum age of a log file to be eligible for deletion due to age log.retention.hours=168 # A size-based retention policy for logs. Segments are pruned from the log unless the remaining # segments drop below log.retention.bytes. Functions independently of log.retention.hours. #log.retention.bytes=1073741824 # The maximum size of a log segment file. When this size is reached a new log segment will be created. log.segment.bytes=1073741824 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies log.retention.check.interval.ms=300000 # Enable StandardAuthorizer on a KRaft-based Kafka cluster # List of enabled mechanisms, can be more than one #sasl.enabled.mechanisms=PLAIN #sasl.mechanism.controller.protocol=PLAIN #sasl.mechanism.inter.broker.protocol=PLAIN authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer #allow.everyone.if.no.acl.found=true #Super.users=User:kafka; #log4j.logger.kafka.authorizer.logger=DEBUG, authorizerAppender #log4j.additivity.kafka.authorizer.logger=false ---
**Author:** ![mmuehlbeyer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mmuehlbeyer/32/1088_2.png) [@mmuehlbeyer](https://forum.confluent.io/u/mmuehlbeyer)\ **Post date:** [23 April 2024 06:31 UTC](https://forum.confluent.io/t/beginner-trying-to-enable-acls-on-a-3-broker-kraft-cluster/10521/4 "2024-04-23T06:31:42Z")
> [@shardservant](#): > > # Enable StandardAuthorizer on a KRaft-based Kafka cluster would you please be so kind and reformat the config above? 🙂 and did you check this post: > [@Self-hosted Kafka with KRaft, SSL and SASL (scram-sha-256)](https://forum.confluent.io/t/self-hosted-kafka-with-kraft-ssl-and-sasl-scram-sha-256/8863/4): > > I’ve managed to get Kafka v3.6.0 working with SASL\_SSL + PLAINTEXT, with a binary build from Apache website. Should also work with confluent-server. Here’s my server.properties: process.roles=broker,controller node.id=1 controller.quorum.voters=1@kafka-test-01:9093,2@kafka-test-02:9093,3@kafka-test-03:9093 listeners=BROKER://:9092,CONTROLLER://:9093 advertised.listeners=BROKER://:9092 inter.broker.listener.name=BROKER controller.listener.names=CONTROLLER listener.security.protocol.map=BROKER:SA… there is a working config. hth, michael