kafka whitelist 如何生效

Kafka 白名单（Whitelist）是一种安全机制，用于限制哪些客户端可以连接到 Kafka 集群。要使 Kafka 白名单生效，请按照以下步骤操作：

1. 配置 Kafka 客户端：在客户端配置文件中，添加一个名为 security.protocol 的属性，并将其值设置为 SASL_PLAINTEXT（如果尚未设置）。这将使客户端使用 SASL 进行身份验证。例如：

security.protocol=SASL_PLAINTEXT

2. 配置 Kafka 服务器：在 Kafka 服务器的配置文件（通常是 server.properties）中，添加以下属性以启用 SASL/PLAIN 身份验证和创建一个名为 client.whitelist 的属性，其中包含允许连接到集群的客户端列表。例如：

listeners=SASL_PLAINTEXT://your.server.address:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
client.whitelist=client1,client2,client3

在这个例子中，client1、client2 和 client3 是允许连接到 Kafka 集群的客户端。您可以使用逗号分隔这些客户端。

3. 重启 Kafka 服务器：保存更改并重启 Kafka 服务器以使更改生效。

4. 配置客户端：在客户端应用程序中，配置 SASL 凭据以进行身份验证。例如，如果您使用的是 Java 客户端，可以使用以下代码创建一个 SaslClient 实例：

Properties props = new Properties();
props.put("sasl.mechanism", "PLAIN");
props.put("sasl.username", "your_username");
props.put("sasl.password", "your_password");
SaslClient saslClient = SaslClient.create(props);

5. 连接到 Kafka：使用配置的 SASL 凭据和 SaslClient 实例连接到 Kafka 集群。例如：

KafkaClient<String, String> kafkaClient = new KafkaClient<>(props);
kafkaClient.connect();

现在，只有 client.whitelist 中列出的客户端才能成功连接到 Kafka 集群。如果尝试使用不在白名单中的客户端连接，将收到一个连接被拒绝的错误。