本节介绍如何访问 RadonDB Kafka。

前提条件

  • 已获取 KubeSphere 企业版平台登录账号和密码，且已获取平台操作管理权限。

  • RadonDB Kafka 实例状态为运行中。

  • 已安装用于访问 Kafka 的客户端。

下载 CA 证书和用户证书

  1. 以 platform-admin 角色登录 KubeSphere 企业版 Web 控制台并进入数据库管理平台。

  2. 在左侧导航栏选择 Kafka。

  3. 在 Kafka 实例列表中，点击一个实例名称打开其详情页面。

  4. 在页面左侧的传输密码区域，下载 CA 证书并保存密码。

  5. 在页面右侧点击 Kafka 用户页签，将鼠标悬浮在加密密钥列的 CA 证书上，下载用户证书并保存密码。

修改 Kafka 配置文件

RadonDB Kafka 支持三种认证类型：TLS、SCRAM-SHA-512 和无需认证。其使用的认证类型可在详情页的参数管理页签中查看。不同的认证模式下，配置文件的修改方式不同。

在修改配置文件之前，请先下载并解压 Kafka 客户端二进制包。

TLS 认证类型

备注

当认证类型为 TLS 时，参数管理页签的 TLS 参数不能设置为 false。

在 Kafka 二进制包中的 config/producer.properties 和 config/consumer.properties 文件末尾添加下面的内容。

#在文件末尾添加以下内容。

security.protocol=ssl
ssl.truststore.location=<ca.p12>
ssl.truststore.password=<truststore_password>
ssl.truststore.type=pkcs12

ssl.keystore.location=<user.p12>
ssl.keystore.password=<keystore_password>
ssl.keystore.type=pkcs12
ssl.endpoint.identification.algorithm=

参数说明如下：

  • <ca.p12>：CA 证书的保存路径。

  • <truststore_password>：CA 证书的密码。

  • <user.p12>：用户证书的保存路径。

  • <keystore_password>：用户证书的密码。

SCRAM-SHA-512 认证类型

备注

当认证类型设置为 SCRAM-SHA-512，参数管理页签的 TLS 的值可以设置为 true 和 false。

  1. 当 TLS 的值设置为 true 时，在 Kafka 二进制包中的 config/producer.properties 和 config/consumer.properties 文件末尾添加以下内容。

    #在文件末尾添加以下内容。
security.protocol=sasl_ssl
sasl.mechanism=SCRAM-SHA-512
ssl.truststore.location=<ca.p12>
ssl.truststore.password=<truststore_password>
ssl.truststore.type=pkcs12
ssl.endpoint.identification.algorithm=

    参数说明如下：

    • <ca.p12>：CA 证书的保存路径。

    • <truststore_password>：CA 证书的密码。

  2. 当 TLS 的值设置为 false 时，在 Kafka 二进制包中的 config/producer.properties 和 config/consumer.properties 文件末尾添加以下内容。

    #在文件末尾添加以下内容。

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
ssl.endpoint.identification.algorithm=

    创建 config/kafka_client_jaas.conf 文件，并添加以下内容。

    > cat config/kafka_client_jaas.conf
KafkaClient {
  org.apache.kafka.common.security.scram.ScramLoginModule required
  username="username"
  password="password";
};

    将 username 和 password 修改为实际的 Kafka 用户名和密码。

  3. 执行以下命令修改 Kafka 二进制包中的 bin/kafka-console-producer.sh 和 bin/kafka-console-consumer.sh 文件。

    #修改 bin/kafka-console-producer.sh 文件。

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M -Djava.security.auth.login.config=/opt/kafka/config/kafka_client_jaas.conf"
fi
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
    if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M -Djava.security.auth.login.config=/opt/kafka/config/kafka_client_jaas.conf"
fi

exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"
    备注

    请将 config/kafka_client_jaas.conf 的地址根据实际情况修改为绝对地址。

无需认证模式

当认证类型设置为无需认证，TLS 的值可以设置为 true 和 false。

  1. 当 TLS 的值为 true 时，在 Kafka 二进制包中的 config/producer.properties 和 config/consumer.properties 文件后增加以下内容：

    #在文件末尾添加以下内容。
security.protocol=SSL
ssl.truststore.location=<ca.p12>
ssl.truststore.password=<truststore_password>
ssl.truststore.type=pkcs12
ssl.endpoint.identification.algorithm=

    参数说明如下：

    • <ca.p12>：CA 证书的保存路径。

    • <truststore_password>：CA 证书的密码。

  2. 当 TLS 的值为 false 时，您无需做任何修改。

使用 Kafka 客户端生产和消费消息

执行以下命令使用 Kafka 客户端生产和消费消息。

生产消息

# bin/kafka-console-producer.sh --bootstrap-server 192.168.100.2:31522 --topic demo-topic --producer.config config/producer.properties
>hello
>radondb
>kafka

消费消息

# bin/kafka-console-consumer.sh --bootstrap-server 192.168.100.2:31522 --topic demo-topic --consumer.config config/consumer.properties
hello
radondb
kafka

其中，bootstrap-server，即命令中的 192.168.100.2:31522，需要修改为 Kafka 实例详情页左侧的访问地址，如下图所示。topic，即命令中的 demo-topic，需要修改为实际使用的主题名称。

kafka address
备注

上图框中第一个读写地址是使用 Kubernetes 节点网络访问 Kafka 服务时使用的地址，如在 Kubernetes 节点上访问 Kafka，或 Kubernetes 节点同网段主机上访问 Kafka 服务。第二个读写地址是使用 Kubernetes 集群中 pod 访问 kafka 服务时使用的地址。