如何在Kafka侧禁止向指定kafka-topic生产消息？

以下方案均不需要修改生产者侧配置，在服务端即可完成操作：

1. 配置Topic级别的ACL权限拒绝写入 #

这是生产环境最稳妥的常用方案，依赖Kafka自带的访问控制能力：

• 首先确认集群已开启ACL授权，broker配置文件server.properties中存在如下配置：
  authorizer.class.name=kafka.security.authorizer.AclAuthorizer
  如果未开启，需先滚动重启所有broker加载配置，开启前建议先配置好管理员及其他必要业务的权限，避免正常业务受影响。
• 执行权限配置命令，拒绝所有用户对目标Topic的写权限：
  kafka-acls.sh --bootstrap-server <broker访问地址:端口> --command-config <管理员认证配置文件路径> --add --deny-principal User:* --operation Write --topic <目标Topic名称>
• 配置完成后可通过如下命令验证权限是否生效：
  kafka-acls.sh --bootstrap-server <broker访问地址:端口> --command-config <管理员认证配置文件路径> --list --topic <目标Topic名称>

该方案的优势是权限粒度可控，后续如果需要给特定生产者开放写入权限，单独调整对应主体的ACL规则即可。

2. 配置Topic生产配额为0 #

如果集群未开启ACL不想额外重启服务，可以使用Kafka的配额限制能力，直接将该Topic的所有生产者的生产速率限制为0：

• 执行配额修改命令：
  kafka-configs.sh --bootstrap-server <broker访问地址:端口> --command-config <管理员配置文件路径> --alter --entity-type clients --entity-default --add-config producer_byte_rate=<目标Topic名称>=0

配置即时生效，所有生产者向该Topic发送消息时会被broker直接限流拒绝，不需要重启任何服务。

3. 标记Topic为只读状态（适用于Kafka 2.8+版本） #

2.8及以上版本的Kafka新增了Topic状态配置，可直接将目标Topic设置为只读状态：

• 执行状态修改命令：
  kafka-configs.sh --bootstrap-server <broker访问地址:端口> --command-config <管理员配置文件路径> --alter --entity-type topics --entity-name <目标Topic名称> --add-config state=readonly

配置生效后所有生产请求会返回POLICY_VIOLATION错误，排查完成后删除该配置即可恢复写入。

• ACL权限方案恢复：删除之前配置的拒绝规则即可
  kafka-acls.sh --bootstrap-server <broker访问地址:端口> --command-config <管理员认证配置文件路径> --remove --deny-principal User:* --operation Write --topic <目标Topic名称>
• 配额方案恢复：删除对应配额配置即可
  kafka-configs.sh --bootstrap-server <broker访问地址:端口> --command-config <管理员配置文件路径> --alter --entity-type clients --entity-default --delete-config producer_byte_rate=<目标Topic名称>
• 只读状态方案恢复：删除topic的state配置即可
  kafka-configs.sh --bootstrap-server <broker访问地址:端口> --command-config <管理员配置文件路径> --alter --entity-type topics --entity-name <目标Topic名称> --delete-config state

内容的提问来源于stack exchange，提问作者Stav Alfi