如何将用户名和密码传递给KafkaConsumer？

Question

我找不到如何将用户名和密码传递给KafkaConsumer。我使用SASL明文对Kafka服务器上的客户端进行身份验证。

我试图在上找到此信息https://kafka.apache.org/documentation/#consumerconfigs但结果是否定的。

我发现了一些作者使用参数“用户名”和“密码”的例子。我试图这样做：

props.put("bootstrap.servers", uri.getHost() + ":" + uri.getPort());
if (!groupidName.isBlank())
    props.put("group.id", groupidName);
props.put("enable.auto.commit", "false");
props.put("auto.offset.reset", groupidName.isBlank() ? "latest" : "earliest");
props.put("session.timeout.ms", "30000");
props.put("max.poll.records", "1");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
if (kafkaConnector.getConfig().getUserName() != null && kafkaConnector.getConfig().getPassword() != null)
{
    props.put("security.protocol","SASL_PLAINTEXT");
    props.put("sasl.mechanism","PLAIN");
    props.put("username",kafkaConnector.getConfig().getUserName());
    props.put("password",kafkaConnector.getConfig().getPassword());
}

但我得到了例外"未能构建Kafka消费者"。

如果有人知道如何将用户名和密码传递给消费者配置，请提供帮助。

Answer 1

您必须设置属性sasl的值。贾斯。配置。

我们的工作如下：

props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "PLAIN");
props.put("sasl.jaas.config", PlainLoginModule.class.getName() + " required username=\"" + KAFKA_USER + "\" password=\"" + KAFKA_PASSWORD + "\";")