阅读路径

1. 谷歌搜索kafka中文介绍及翻译
2. 学习成熟的kafka的dockerfile及compose
3. 阅读官方文档介绍性章节
4. 阅读官方文档security章节

主要问题

• 重启要全面

  • 更改了配置文件重启测试时，记得先删除kafka的log(数据)目录
  • zookeeper也需要一并重置
  • 不然的话可能会遇到1 partitions have leader brokers without a matching listener, including [test-0]

• 先配置认证

  • kafka调试3A的时候最好先不开启鉴权，先调试认证
  • kafka、hbase等apache家一众的java大数据项目的认证是类似的，涉及SASL和JAAS和SSL
    • kafka配置SSL可以用于传输层的加密，同时可选用于认证
    • JAAS是java语言内部的认证授权框架，实现了(或使用内置的)具体认证细节的模块后，可以用它处理认证过程、管理认证session和鉴权policy等等
    • SASL是规定了一些认证流程和接口&不涉及具体认证细节算法的认证规范
      • SASL是一个规范，自然就有对应的实现，网络服务以及客户端可以调用其实现进行符合SASL框架的认证交互
    • 在kafka中配置SASL的大部分配置是写在JAAS所负责解析的配置文件中的，自然就产生了在kafka中JAAS和SASL的调用关系是什么的问题，谁在谁之上？还是并列调用(JAAS只是读个配置而已)？
      • 在读了认证模块相关代码后，确认是先调用JAAS加载kafka自己编写的认证模块，而这个模块是一个adapter适配器，在其中调用sasl进行认证。即由JAAS调用SASL
      • 吐槽一下，如果用kerberos的话加上GSSAPI会有JAAS-SASL-GSSAPI三层类似的认证抽象层层调用，这就是架构嘛hhhh

• 再配置鉴权

  • 场景中多用户多topic再开启鉴权比较好，不然徒增不少麻烦
  • 不认证但开启鉴权的问题：
    • 一般outside外部访问方向会开启认证
    • inside集群内部访问不开启认证且使用PLAINTEXT明文传输
      • 导致不开启认证 && 开启鉴权时，kafka会为inside方向没有认证的请求默认判定为User:ANONYMOUS用户
    • kafka的所有broker需要走inside方向登录自己以及其他broker进行同步
    • 鉴权所使用的访问控制ACL规则存储在zk中，不手动显式进行添加就不存在ACL规则
    • 最终导致kafka启动后无法登录自己，且日志中显示有User：ANONYMOUS用户的请求并被拒绝
      • 此时读写消息会报错Error while fetching metadata with correlation id 28 : {test=LEADER_NOT_AVAILABLE}
  • 解决的办法至少有下面几种
    • 在properties中设置User:ANONYMOUS是superuser，拥有绝对的权限
      • 或为User:ANONYMOUS配置相应的高权限ACL规则(还没有做具体实验)
      • 在outside方向没有ANONYMOUS用户存在时，我觉得安全性可以接受，尽管superuser和anonymous放一起很怪
    • inside方向传输层使用SSL，并给与证书对应的用户superuser
      • 同理也可以给证书用户很高的访问控制权限
      • 安全性最高，但配置也麻烦
    • properties开启KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND
      • 即不存在ACL策略应用于本资源时，任何人都可以访问，非常不安全

搭建运行

镜像

• 在dockerhub和google搜索kafka容器化相关内容，找到wurstmeister/kafka-docker/
  • 下载量大，star多
  • Dockerfile entrypoint清晰
  • Tag管理简洁规范
  • Env配置注入灵活

编排

• 编排目前使用docker-compose
• 示例中zookeeper没有特殊的配置
• 示例中kafka
  • 配置主要通过环境变量注入
  • 认证相关的jaas配置文件通过volume挂载进etc目录

  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    ports:
      - "172.26.0.2:2181:2181"
    networks:
      ninestates: {}

  kafka:
    image: wurstmeister/kafka:2.13-2.6.0
    ports:
      - "172.26.0.2:9092:9092"
      - "172.26.0.2:9093:9093"
    environment:
      # kafka基础配置
      KAFKA_BROKER_ID: 1
      KAFKA_CREATE_TOPICS: "test:1:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: INSIDE://172.26.0.2:9092,OUTSIDE://172.26.0.2:9093
      KAFKA_LISTENERS: INSIDE://:9092,OUTSIDE://:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:SASL_PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      # 以下是认证鉴权相关配置
      #KAFKA_HEAP_OPTS: "-Xmx1G -Xms1G -Djava.security.auth.login.config=\\/etc\\/jaas\\/kafka_server_jaas.conf"
      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/jaas/kafka_server_jaas.conf"
      KAFKA_SASL_MECHANISM: PLAIN
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_SUPER_USERS: "User:admin;User:ANONYMOUS"
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      #KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
    volumes:
      - "/var/kafka/:/kafka"
      - "./kafka/jaas/:/etc/jaas"
    networks:
      ninestates: {}

配置约定

基础配置

配置解释

• KAFKA_BROKER_ID：每个broker在集群中的唯一ID
• KAFKA_CREATE_TOPICS：启动时自动创建topic，topic名：分区数：副本数
• KAFKA_ZOOKEEPER_CONNECT：kafka集群使用的zk集群地址
• KAFKA_ADVERTISED_LISTENERS、KAFKA_LISTENERS
  • URI的协议名INSIDE和OUTSIDE是别名，kafka并不理解其含义
  • 多个Listener通过指定不同的URI监听不同的socket，为不同的客户端差异服务
  • SECURITY_PROTOCOL_MAP是别名到传输层安全机制的映射，如OUTSIDE使用SASL认证，传输层使用PLAINTEXT明文传输
  • INTER_BROKER_LISTENER_NAME指定哪个别名所属的listener用于集群内部通信

安全配置

配置解释

• KAFKA_HEAP_OPTS 或 KAFKA_OPTS：指定java启动时读取jaas配置文件的路径
• KAFKA_SASL_MECHANISM：SASL框架内使用的插件认证机制
• KAFKA_SASL_ENABLED_MECHANISMS：SASL框架内允许使用插件认证机制
• KAFKA_SUPER_USERS：超级用户，分号分割
• KAFKA_AUTHORIZER_CLASS_NAME：
  • 指定鉴权插件路径
  • 同时也是鉴权开启开关
  • 关闭Authorization的话注释此项即可

JAAS配置文件

• 以下是kafka开启SASL认证时，通过jaas配置文件传递给sasl的配置内容
• username和password是kafka集群内部通讯使用的用户名和密码
  • 但是目前的配置在outside的listener开启认证，所以不涉及内部通讯
• user_username="password"是用于登陆的用户名和密码
  • 可以看到sasl使用plain认证机制时用户名和密码是明文写在配置文件中的

kafka_server_jaas.conf: 

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_alice="alice-secret";
};

• 以下是客户端与服务端对称的jaas配置文件
  • 由于配置内容不多，所以可以写在客户端的properties配置文件中，如下下图
  • 如果非要使用jaas.conf文件的话，java程序启动时同样需要指定-Djava.security.auth.login.config=路径

kafka_client_jaas.conf：

    sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
        username="alice" \
        password="alice-secret";

kafka_client_alice.properties:

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret";
group.id=test-group

生产消费

无认证鉴权

• 消费./kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092 --topic test
• 生产./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test --from-beginning

有认证鉴权

• 消费./kafka-console-consumer.sh --bootstrap-server 172.26.0.2:9093 --topic test --from-beginning --consumer.config /etc/jaas/kafka_client.properties
• 生产./kafka-console-producer.sh --bootstrap-server 172.26.0.2:9093 --topic test --producer.config /etc/jaas/kafka_client.properties

ACL配置

• 进入kafka容器后需要unset KAFKA_OPTS避免去读取用于连接zookeeper的jaas配置
• ./kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=zookeeper:2181 --add --allow-principal User:alice --operation Read --topic test
• ./kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=zookeeper:2181 --add --allow-principal User:alice --operation Read --group test-group
  • 注意consumer读取时涉及对自己所属consumer-group的组相关信息的读取
  • 不加的话会报 org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: console-consumer-5462

Finally

• 第一次记录大数据平台方面的学习，欢迎评论交流