你的意思是这样的：

topic = "test"
    sasl_mechanism = "PLAIN"
    username = "admin"
    password = "pwd$"
    security_protocol = "SASL_PLAINTEXT"

    #context = ssl.create_default_context()
    #context.options &= ssl.OP_NO_TLSv1
    #context.options &= ssl.OP_NO_TLSv1_1

    consumer = KafkaConsumer(topic, bootstrap_servers='kafka1:9092',
                              #api_version=(0, 10),
                              security_protocol=security_protocol,
                              #ssl_context=context,
                              #ssl_check_hostname=True,
                              #ssl_cafile='../keys/CARoot.pem',
                              sasl_mechanism = sasl_mechanism,
                              sasl_plain_username = username,
                              sasl_plain_password = password)
                              #ssl_certfile='../keys/certificate.pem',
                              #ssl_keyfile='../keys/key.pem')#,api_version = (0, 10))

是的，每个主题可以有用户/通行证。参见官方文件授权和acls。
您可以使用ssl或sasl（kafka的sasl支持）启用安全性：
sasl/gssapi（kerberos）-从版本0.9.0.0开始
sasl/plain-从版本0.10.0.0开始
sasl/scram-sha-256和sasl/scram-sha-512-从版本0.10.2.0开始
从文档中，添加ACL的示例：
假设要添加acl“principals”user:bob and user:alice are 允许对ip 198.51.100.0和ip 198.51.100.1中的测试主题执行读写操作。您可以通过使用以下选项执行cli来实现这一点：1

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Bob --allow-principal User:Alice --allow-host 198.51.100.0 --allow-host 198.51.100.1 --operation Read --operation Write --topic Test-topic

在这篇博文中你也可以找到一些信息
我不确定您使用的是什么库，但这应该只是将适当的属性传递给生产者/客户的问题；kafka python支持：
支持sasl/kerberos
支持基于acl的kafka

如果要使用用户名+密码进行身份验证，则需要在集群上使用普通机制启用sasl身份验证。有关完整说明，请参阅kafka网站上的authentication using sasl部分。
请注意，您可能还希望启用ssl（sasl\u ssl），否则，sasl plain将以明文形式传输凭据。
一些python客户端支持sasl平原，例如：
KafkaPython：https://github.com/dpkp/kafka-python
融合的KafkaPython：https://github.com/confluentinc/confluent-kafka-python
关于授权，使用默认授权人， kafka.security.auth.SimpleAclAuthorizer ，您可以将生产者限制为只能产生主题。Kafka的网站上的授权和acls部分同样完整地记录了这一切。
例如，对于saslplain，默认情况下，主体名称是用于连接的用户名。使用以下命令可以限制用户 Alice 只能生成到名为 testtopic :

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Alice --producer --topic testtopic