支持云审计的操作列表_分布式消息服务Kafka版_用户指南_审计_华为云

(1)

用户指南

文档版本 01

发布日期 2022-02-22

(2)

非经本公司书面许可，任何单位和个人不得擅自摘抄、复制本文档内容的部分或全部，并不得以任何形式传播。

商标声明

和其他华为商标均为华为技术有限公司的商标。

本文档提及的其他所有商标或注册商标，由各自的所有人拥有。

注意

您购买的产品、服务或特性等应受华为公司商业合同和条款的约束，本文档中描述的全部或部分产品、服务或特性可能不在您的购买或使用范围之内。除非合同另有约定，华为公司对本文档内容不做任何明示或暗示的声明或保证。

由于产品版本升级或其他原因，本文档内容会不定期进行更新。除非另有约定，本文档仅作为使用指导，本文档中的所有陈述、信息和建议不构成任何明示或暗示的担保。

(3)

1 ^权限管理

1.1 创建用户并授权使用 DMS for Kafka

如果您需要对您所拥有的DMS for Kafka服务进行精细的权限管理，您可以使用统一身份认证服务（Identity and Access Management，简称IAM），通过IAM，您可以：

● 根据企业的业务组织，在您的华为云帐号中，给企业中不同职能部门的员工创建 IAM用户，让员工拥有唯一安全凭证，并使用DMS for Kafka资源。

● 根据企业用户的职能，设置不同的访问权限，以达到用户之间的权限隔离。

● 将DMS for Kafka资源委托给更专业、高效的其他华为云帐号或者云服务，这些帐号或者云服务可以根据权限进行代运维。

如果华为云帐号已经能满足您的要求，不需要创建独立的IAM用户，您可以跳过本章节，不影响您使用DMS for Kafka服务的其它功能。

本章节为您介绍对用户授权的方法，操作流程如图1-1所示。

前提条件

给用户组授权之前，请您了解用户组可以添加的DMS for Kafka系统策略，并结合实际需求进行选择，DMS for Kafka支持的系统策略及策略间的对比，请参见：DMS for

Kafka系统策略。若您需要对除DMS for Kafka之外的其它服务授权，IAM支持服务的

所有策略请参见权限策略。

(6)

示例流程

图1-1 给用户授权 DMS for Kafka 权限流程

1. 创建用户组并授权

在IAM控制台创建用户组，并授予DMS for Kafka的只读权限“DMS ReadOnlyAccess”。

2. 创建用户并加入用户组

在IAM控制台创建用户，并将其加入1中创建的用户组。

3. 用户登录并验证权限

新创建的用户登录控制台，验证DMS for Kafka的只读权限。

1.2 DMS for Kafka 自定义策略

如果系统预置的DMS for Kafka权限，不满足您的授权要求，可以创建自定义策略。自定义策略中可以添加的授权项（Action）请参考细粒度策略支持的授权项。

目前华为云支持以下两种方式创建自定义策略：

● 可视化视图创建自定义策略：无需了解策略语法，按可视化视图导航栏选择云服务、操作、资源、条件等策略内容，可自动生成策略。

● JSON视图创建自定义策略：可以在选择策略模板后，根据具体需求编辑策略内容；也可以直接在编辑框内编写JSON格式的策略内容。

具体创建步骤请参见：创建自定义策略。本章为您介绍常用的DMS for Kafka自定义策略样例。

(7)

说明

● DMS for Kafka的权限与策略基于分布式消息服务DMS，因此在IAM服务中为DMS for Kafka 分配用户与权限时，请选择并使用“DMS”的权限与策略。

● 由于缓存的存在，对用户、用户组以及企业项目授予OBS相关的细粒度策略后，大概需要等待5分钟细粒度策略才能生效。

DMS 自定义策略样例

● 示例1：授权用户删除实例和重启实例

{ "Version": "1.1", "Statement": [ {

"Effect": "Allow", "Action": [ "

dms:instance:delete dms:instance:modifyStatus "

] } ] }

● 示例2：拒绝用户删除实例

拒绝策略需要同时配合其他策略使用，否则没有实际作用。用户被授予的策略中，一个授权项的作用如果同时存在Allow和Deny，则遵循Deny优先。

如果您给用户授予DMS FullAccess的系统策略，但不希望用户拥有DMS

FullAccess中定义的删除实例权限，您可以创建一条拒绝删除实例的自定义策略，

然后同时将DMS FullAccess和拒绝策略授予用户，根据Deny优先原则，则用户可以对DMS for Kafka执行除了删除实例外的所有操作。拒绝策略示例如下：

{ "Version": "1.1", "Statement": [ {

"Effect": "Deny", "Action": [

"dms:instance:delete"

] } ] }

(8)

2 ^{准备实例依赖资源}

概述

在创建Kafka实例前，您需要提前准备相关依赖资源，包括虚拟私有云（Virtual Private Cloud，以下简称VPC）、子网和安全组，并配置安全组策略。每个Kafka实例都部署在某个VPC中，并绑定具体的子网和安全组，通过这样的方式为Kafka提供一个隔离的、用户自主配置和管理的虚拟网络环境以及安全保护策略，提升实例的安全性。

准备依赖资源

Kafka实例所需资源的具体要求和创建指导如表2-1所示。

表2-1 Kafka 实例依赖资源

准备资源要求创建指导

VPC和子网不同的Kafka实例可以重复使用相同的VPC和子网，也可以使用不同的VPC和子网，请根据实际需要进行配置。在创建VPC和子网时应注意如下要求：

● 创建的VPC与使用的Kafka服务应在相同的区域。

● 创建VPC和子网时，如无特殊需求，配置参数使用默认配置即可。

● 建议子网不要开启IPv6功能，

Kafka实例使用开启IPv6的子网后，无法开启公网访问。

创建VPC和子网的操作指导请参考创建虚拟私有云和子网，若需要在已有VPC上创建和使用新的子网，请参考为虚拟私有云创建新的子网。

(9)

准备资源要求创建指导安全组不同的Kafka实例可以重复使用相

同的安全组，也可以使用不同的安全组，请根据实际需要进行配置。

在创建安全组时应注意如下要求：

● 创建安全组时，“模板”选择

“自定义”。

● 创建安全组后，请保留系统默认添加的入方向“允许安全组内的弹性云服务器彼此通信”规则和出方向“放通全部流量”规则。

● 使用Kafka服务必须添加表2-2所示安全组规则，其他规则请根据实际需要添加。

创建安全组的操作指导请参考创建安全组，为安全组添加规则的操作指导请参考添加安全组规则。

表2-2 安全组规则

方向协议端口源地址说明

入方向 TCP 9094 0.0.0.0/0 通过公网访问Kafka（关闭SSL 加密）。

入方向 TCP 9092 0.0.0.0/0 通过VPC内网访问Kafka（关闭 SSL加密）。

入方向 TCP 9095 0.0.0.0/0 通过公网访问Kafka（开启SSL 加密）。

入方向 TCP 9093 0.0.0.0/0 通过VPC内网访问Kafka（开启 SSL加密）。

入方向 TCP 9999 0.0.0.0/0 访问Kafka Manager。

(10)

3 ^购买实例

操作场景

目前，DMS提供Kafka实例的服务，Kafka实例采用物理隔离的方式部署，租户独占 Kafka实例。支持用户自定义规格和自定义特性，您可以根据业务需要定制相应计算能力和存储空间的Kafka实例。

购买前须知

● Kafka实例运行于虚拟私有云，购买实例前，需保证有可用的虚拟私有云，并且已配置好安全组与子网。

● 实例不同规格的节点数和底层资源相关。由于不同Region，底层资源不一样，实例创建后，实例节点数也会不一样。

操作步骤

步骤1 登录管理控制台。

步骤2 在管理控制台左上角单击，选择区域。

说明

此处请选择与您的应用服务相同的区域。

步骤3 在管理控制台左上角单击，选择“应用中间件 > 分布式消息服务Kafka版”，进入分布式消息服务Kafka专享版页面。

步骤4 单击页面右上方的“购买Kafka实例”。

每个项目默认最多可以创建100个Kafka实例，如果您想创建更多实例，请联系客服申请增加配额。

步骤5 选择“计费模式”、“区域”、“项目”和“可用区”。

步骤6 设置“实例名称”和“企业项目”。

步骤7 设置实例信息。

1. 版本：Kafka的版本号，当前支持1.1.0和2.3.0，根据实际情况选择，推荐使用 2.3.0。

(11)

2. CPU架构：当前仅支持“x86计算”，保持默认值即可。

3. 当前控制台存在2种规格，具体如下：

– 在除华东-上海一、华东-上海二、华北-北京四和华南-广州以外的区域，在

“规格”中，请根据业务流量选择带宽。

选择基准带宽时，可以查看此带宽对应的底层ECS数量和规格、分区上限、以及建议的消费组数量。

分区上限指Kafka实例可创建的最大分区总数。当所有Topic的总分区数大于此值时，创建Topic失败。

– 在华东-上海一、华东-上海二、华北-北京四和华南-广州区域，在“规格”

中，请根据业务需求选择相应的代理规格。在“代理个数”中，选择代理个数。

单个代理最大分区数*代理个数=实例分区数上限。当所有Topic的总分区数大于实例分区数上限时，创建Topic失败。

4. 在“存储空间”区域，您根据实际需要选择存储Kafka数据的总磁盘大小。

存储空间包含所有副本存储空间总和，建议根据业务消息体积以及副本数量选择存储空间大小。假设业务存储数据保留天数内磁盘大小为100GB，则磁盘容量最少为100GB*副本数 + 预留磁盘大小100GB。

创建实例时会进行磁盘格式化，磁盘格式化会导致实际可用磁盘为总磁盘的 93%~95%。

– 基准带宽为100MB/s时，存储空间取值范围：600GB ~ 90000GB。

说明

– 高IO + 100MB/s带宽：假设平均每条消息1KB，则高吞吐场景可达10万条/秒，同步复制场景可达6万条/秒。

– 高IO + 300MB/s带宽 : 假设平均每条消息1KB，则高吞吐场景下可达30万条/秒，同步复制场景可达15万条/秒。

– 超高IO + 100MB/s带宽：假设平均每条消息1KB，则高吞吐场景可达10万条/秒，同步复制场景可达8万条/秒。

5. 在“容量阈值策略”区域，设置磁盘使用达到容量阈值后的消息处理策略，容量阈值默认为95%。

– 自动删除：可以正常生产和消费消息，但是会删除最早的10%的消息，以保证磁盘容量充足。该场景优先保障业务不中断，数据存在丢失的风险。

– 生产受限：无法继续生产消息，但可以继续消费消息。该场景适用于对数据不能丢的业务场景，但是会导致生产业务失败。

6. 在“虚拟私有云”下拉列表，选择已经创建好的虚拟私有云和子网。

虚拟私有云可以为您的Kafka实例构建隔离的、能自主配置和管理的虚拟网络环境。

(12)

说明

– 建议选择未开启IPv6的子网，Kafka实例使用开启IPv6的子网后，无法开启公网访问。

– 虚拟私有云和子网在Kafka实例创建完成后，不支持修改。

7. 在“安全组”下拉列表，可以选择已经创建好的安全组。

安全组是一组对弹性云服务器的访问规则的集合，为同一个VPC内具有相同安全保护需求并相互信任的弹性云服务器提供访问策略。

您可以单击右侧的“管理安全组”，跳转到网络控制台的“安全组”页面，查看或创建安全组。

图3-1 购买 Kafka 实例

步骤8 设置登录Kafka Manager的用户名和密码。

Kafka Manager是开源的kafka集群管理工具，实例创建成功后，实例详情页面会展示 Kafka Manager登录地址，您可登录Kafka Manager页面，查看Kafka集群的监控、代理等信息。

步骤9 设置实例购买时长。

当选择了“包年/包月”付费模式时，页面才显示“购买时长”参数，您需要根据业务需要选择。

步骤10 单击“更多配置”，设置更多相关信息。

1. 设置“公网访问”。

“公网访问”默认为关闭状态，您可以选择是否开启。

开启公网访问后，还需要为每个代理设置对应的IPv4弹性IP地址。

说明

公网访问支持的区域如下：华北-北京一、华北-北京四、华北-乌兰察布一、华东-上海一、

华东-上海二和华南-广州。

2. 选择是否开启转储。

通过开启转储将数据转储至OBS桶，实现数据的备份。

(13)

说明

– 消息包含消息属性（由属性名及属性值组成，以K-V的形式体现）时，属性名key不支持转储，属性值value支持转储。

– 开启转储后，暂不支持关闭转储。

3. 设置“Kafka SASL_SSL”。

客户端连接Kafka实例时，是否开启SSL认证。开启Kafka SASL_SSL，则数据加密传输，安全性更高。

创建实例后，Kafka SASL_SSL开关不支持修改，请慎重选择。如果创建后需要修改，需要重新创建实例。

开启Kafka SASL_SSL后，您需要设置连接Kafka实例的用户名和密码。

4. 设置“Kafka自动创建Topic”。

选择开启“Kafka自动创建Topic”，表示生产或消费一个未创建的Topic时，会自动创建一个包含3个分区和3个副本的Topic。

5. 设置“标签”。

标签用于标识云资源，当您拥有相同类型的许多云资源时，可以使用标签按各种维度（例如用途、所有者或环境）对云资源进行分类。

– 如果您已经预定义了标签，在“标签键”和“标签值”中选择已经定义的标签键值对。另外，您可以单击“查看预定义标签”，跳转到标签管理服务页面，查看已经预定义的标签，或者创建新的标签。

– 您也可以直接在“标签键”和“标签值”中设置标签。

当前每个Kafka实例最多支持设置20个不同标签，标签的命名规格，请参考管理实例标签章节。

说明

部分区域不支持标签功能，请以控制台为准。

6. 设置实例的描述信息。

步骤11 填写完上述信息后，单击“立即购买”，进入规格确认页面。

步骤12 确认实例信息无误且阅读并同意《华为云用户协议》后，如果“计费模式”选择“包年/包月”，单击“去支付”。根据界面提示信息，支付费用。如果“计费模式”选择

“按需付费”，单击“提交”。

步骤13 在实例列表页面，查看Kafka实例是否创建成功。

创建实例大约需要3到15分钟，此时实例的“状态”为“创建中”。

● 当实例的“状态”变为“运行中”时，说明实例创建成功。

● 如果创建实例失败，在信息栏的“创建失败任务”中查看创建失败的实例。请参考删除实例，删除创建失败的实例，然后重新创建。如果重新创建仍然失败，请联系客服。

说明

创建失败的实例，不会占用其他资源。

----结束

(14)

4 ^{连接 Kafka}

4.1 连接未开启 SASL 的 Kafka 实例

目前，华为云提供Kafka实例的服务，Kafka实例采用物理隔离的方式部署，租户独占 Kafka实例。创建Kafka实例之后，使用开源Kafka客户端向Kafka实例生产消息和消费消息。

本章节介绍如何使用开源的Kafka客户端访问未开启SASL的Kafka实例的方法。

多语言客户端的使用请参考Kafka官网：https://cwiki.apache.org/confluence/

display/KAFKA/Clients

说明

● 本章节主要描述使用命令行模式连接Kafka实例，如果是在业务代码中连接Kafka实例，请参考《Kafka实例开发指南》。

● 2020年7月以及之后购买的实例，Kafka服务器允许客户端单IP连接的个数为1000个，在此之前购买的实例，Kafka服务器允许客户端单IP连接的个数为200个，如果超过了，会出现连接失败问题。您可以通过修改配置参数来修改单IP的连接数。

前提条件

● 已配置正确的安全组。

访问未开启SASL的Kafka实例时，支持VPC内访问和公网访问两种方式，实例需要配置正确的安全组规则，具体安全组配置要求，请参考表2-2。

● 已获取连接Kafka实例的地址。

– 如果是VPC内访问，实例端口为9092，实例连接地址获取如下图。

图4-1 获取 VPC 内访问 Kafka 实例的连接地址（实例未开启 SASL）

– 如果是公网访问，实例端口为9094，实例连接地址获取如下图。

图4-2 获取公网访问 Kafka 实例的连接地址（实例未开启 SASL）

(15)

● Kafka实例已创建Topic，否则请提前创建Topic。

● 已下载Kafka命令行工具1.1.0版本或者Kafka命令行工具2.3.0版本，确保Kafka实例版本与命令行工具版本相同。

● 已创建弹性云服务器，弹性云服务器的VPC、子网、安全组与Kafka实例的VPC、

子网、安全组保持一致。在弹性云服务器中安装Java Development Kit 1.8.111 或以上版本，并完成环境变量配置。

命令行模式连接实例

以下操作命令以Linux系统为例进行说明：

步骤1 解压Kafka命令行工具。

进入文件压缩包所在目录，然后执行以下命令解压文件。

tar -zxf

[kafka_tar]

其中，[kafka_tar]表示命令行工具的压缩包名称。

例如：

tar -zxf kafka_2.11-2.3.0.tgz

步骤2 进入Kafka命令行工具的“/bin”目录下。

注意，Windows系统下需要进入“/bin/windows”目录下。

步骤3 执行如下命令进行生产消息。

./kafka-console-producer.sh --broker-list ${连接地址} --topic ${Topic名称}

参数说明如下：

● 连接地址：从前提条件中获取的连接地址，如果是公网访问，请使用“公网连接地址”，如果是VPC内访问，请使用“内网连接地址”，请根据实际情况选择。

● Topic名称：Kafka实例下创建的Topic名称。

本文以公网连接为例，获取的Kafka实例公网连接地址为

“10.3.196.45:9094,10.78.42.127:9094,10.4.49.103:9094”。执行完命令后输入内容，按“Enter”发送消息到Kafka实例，输入的每一行内容都将作为一条消息发送到 Kafka实例。

[root@ecs-kafka bin]# ./kafka-console-producer.sh --broker-list 10.3.196.45:9094,10.78.42.127:9094,10.4.49.103:9094 --topic topic-demo

>Hello

>DMS

>Kafka!

>^C[root@ecs-kafka bin]#

如需停止生产使用Ctrl+C命令退出。

步骤4 执行如下命令消费消息。

./kafka-console-consumer.sh --bootstrap-server ${连接地址} --topic ${Topic名称} --group ${消费组名称} -- from-beginning

● 连接地址：从前提条件中获取的连接地址，如果是公网访问，请使用“公网连接地址”，如果是VPC内访问，请使用“内网连接地址”，请根据实际情况选择。

(16)

● 消费组名称：根据您的业务需求，设定消费组名称。

示例如下：

[root@ecs-kafka bin]# ./kafka-console-consumer.sh --bootstrap-server

10.3.196.45:9094,10.78.42.127:9094,10.4.49.103:9094 --topic topic-demo --group order-test --from-beginning Kafka!

DMSHello

^CProcessed a total of 3 messages [root@ecs-kafka bin]#

如需停止消费使用Ctrl+C命令退出。

----结束

4.2 连接已开启 SASL 的 Kafka 实例

创建实例时开启SASL_SSL访问，则数据加密传输，安全性更高。

由于安全问题，2021年3月20日前以及当天创建的实例，支持的加密套件为 TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256，

TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256和

TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256。2021年3月20日后创建的实例，

支持的加密套件为TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256。

本节介绍如何使用开源的Kafka客户端访问开启SASL的Kafka实例的方法。

说明

● 2020年7月以及之后购买的实例，Kafka服务器允许客户端单IP连接的个数为1000个，在此之前购买的实例，Kafka服务器允许客户端单IP连接的个数为200个，如果超过了，会出现连接失败问题。您可以通过修改配置参数来修改单IP的连接数。

● 使用SASL方式连接Kafka实例时，为了客户端能够快速解析实例的Broker，建议在客户端所在主机的“/etc/hosts”文件中配置host和IP的映射关系。

其中，IP地址必须为实例连接地址（Broker地址），host为每个实例主机的名称（您可以自定义主机的名称，但不能重复）。

例如：

10.154.48.120 server01 10.154.48.121 server02 10.154.48.122 server03

前提条件

● 已配置正确的安全组。

访问开启SASL的Kafka实例时，支持VPC内访问和公网访问两种方式，实例需要配置正确的安全组规则，具体安全组配置要求，请参考表2-2。

● 已获取连接Kafka实例的地址。

– 如果是VPC内访问，实例端口为9093，实例连接地址获取如下图。

图4-3 获取 VPC 内访问 Kafka 实例的连接地址（实例已开启 SASL）

– 如果是公网访问，实例端口为9095，实例连接地址获取如下图。

(17)

图4-4 获取公网访问 Kafka 实例的连接地址（实例已开启 SASL）

● Kafka实例已创建Topic，否则请提前创建Topic。

● 已下载client.jks证书。如果没有，在控制台单击Kafka实例名称，进入实例详情页面，在“连接信息 > SSL证书”所在行，单击“下载”。下载压缩包后解压，获取压缩包中的客户端证书文件：client.jks。

● 已下载Kafka命令行工具1.1.0版本或者Kafka命令行工具2.3.0版本，确保Kafka实例版本与命令行工具版本相同。

● 已创建弹性云服务器，弹性云服务器的VPC、子网、安全组与Kafka实例的VPC、

子网、安全组保持一致。在弹性云服务器中安装Java Development Kit 1.8.111 或以上版本，并完成环境变量配置。

命令行模式连接实例

以下操作命令以Linux系统为例进行说明。

步骤1 解压Kafka命令行工具。

进入文件压缩包所在目录，然后执行以下命令解压文件。

tar -zxf

[kafka_tar]

其中，[kafka_tar]表示命令行工具的压缩包名称。

例如：

tar -zxf kafka_2.11-2.3.0.tgz 步骤2 修改Kafka命令行工具配置文件。

在Kafka命令行工具的“/config”目录中找到“consumer.properties”和

“producer.properties”文件，并分别在文件中增加如下内容。

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="**********" \

password="**********";

sasl.mechanism=PLAIN security.protocol=SASL_SSL

ssl.truststore.location=/opt/kafka_2.11-2.3.0/config/client.jks ssl.truststore.password=dms@kafka

ssl.endpoint.identification.algorithm=

参数说明：

● username和password为创建Kafka实例过程中开启SASL_SSL时填入的用户名和密码，或者创建SASL_SSL用户时设置的用户名和密码。

● ssl.truststore.location配置为client.jks证书的存放路径。注意，Windows系统下证书路径中也必须使用“/”，不能使用Windows系统中拷贝路径时的“\”，否则客户端获取证书失败。

● ssl.truststore.password为服务器证书密码，不可更改，需要保持为 dms@kafka。

● ssl.endpoint.identification.algorithm为证书域名校验开关，为空则表示关闭。这里需要保持关闭状态，必须设置为空。

(18)

步骤3 进入Kafka命令行工具的“/bin”目录下。

注意，Windows系统下需要进入“/bin/windows”目录下。

步骤4 执行如下命令进行生产消息。

./kafka-console-producer.sh --broker-list ${连接地址} --topic ${Topic名称} --producer.config ../config/

producer.properties

● 连接地址：从前提条件获取的连接地址，如果是公网访问，请使用“公网连接地址”，如果是VPC内访问，请使用“内网连接地址”，请根据实际情况选择。

本文以公网访问为例，Kafka实例连接地址为“10.xxx.xxx.202:9095,10.xxx.xxx.

197:9095,10.xxx.xxx.68:9095”。

执行完命令后，输入需要生产的消息内容，按“Enter”发送消息到Kafka实例，输入的每一行内容都将作为一条消息发送到Kafka实例。

[root@ecs-kafka bin]#./kafka-console-producer.sh --broker-list 10.xxx.xxx.202:9095,10.xxx.xxx.

197:9095,10.xxx.xxx.68:9095 --topic topic-demo --producer.config ../config/producer.properties

>hello

>DMS

>Kafka!

>^C[root@ecs-kafka bin]#

如需停止生产使用Ctrl+C命令退出。

步骤5 执行如下命令消费消息。

./kafka-console-consumer.sh --bootstrap-server ${连接地址} --topic ${Topic名称} --group ${消费组名称} -- from-beginning --consumer.config ../config/consumer.properties

● 连接地址：从前提条件获取的连接地址，如果是公网访问，请使用“公网连接地址”，如果是VPC内访问，请使用“内网连接地址”，请根据实际情况选择。

● 消费组名称：根据您的业务需求，设定消费组名称。

示例如下：

[root@ecs-kafka bin]# ./kafka-console-consumer.sh --bootstrap-server 10.xxx.xxx.202:9095,10.xxx.xxx.

197:9095,10.xxx.xxx.68:9095 --topic topic-demo --group order-test --from-beginning --consumer.config ../

config/consumer.properties hello

Kafka!

DMShello

^CProcessed a total of 4 messages [root@ecs-kafka bin]#

如需停止消费使用Ctrl+C命令退出。

----结束

4.3 连接和查看 Kafka Manager

Kafka Manager是开源的Kafka集群管理工具，需要通过浏览器才能访问Kafka Manager的地址。在Kafka Manager页面，您可以查看Kafka集群的监控、代理等信息。

(19)

前提条件

已正确配置安全组。

登录 Kafka Manager

步骤1 （可选）创建一台与Kafka实例相同VPC和相同安全组的Windows服务器，详细步骤请参考《购买并登录Windows弹性云服务器》。

如果是已经开启了公网访问，该步骤为可选，在本地浏览器中即可访问，不需要单独的Windows弹性云服务器。

步骤2 在实例详情信息页面，获取Kafka Manager地址。

● 未开启公网访问时，Kafka Manager地址为“Manager内网访问地址”。

图4-5 Kafka Manager 内网访问地址

● 已开启公网访问时，Kafka Manager地址为“Manager公网访问地址”。

图4-6 Kafka Manager 公网访问地址

步骤3 在浏览器中输入Kafka Manager的地址，进入Kafka Manager登录页面。

如果是开启了公网访问，在本地浏览器输入Kafka Manager地址访问；如果没有开启公网访问，需要登录步骤1的弹性云服务器，然后在浏览器输入Kafka Manager地址访问。

步骤4 输入创建实例时设置的Kafka Manager用户名和密码，即可管理Kafka集群。

----结束

查看 Kafka Manager

在进入Kafka Manager集群管理页面后，您可以查看Kafka集群的监控、代理等信息。

● 集群信息页

单击Clusters中的集群列表，即可进入集群信息页。如图4-7所示。

– 图中①区域表示功能导航栏

▪

Cluster: 集群，统计集群列表和集群详情。

▪

Brokers: 代理，统计当前集群中各代理的状态信息。

▪

Topic: 队列，统计当前集群中的kafka队列。

▪

Preferred Replica Election: 强制进行一次队列leader的最优选举（不 建议用户操作）。

▪

Reassign Partitions: 进行分区副本的重分配（不建议用户操作）。

(20)

▪

Consumers: 统计集群中的消费组状态。

– 图中②区域表示集群信息统计，包含集群的Topic数和集群的代理数。

图4-7 集群信息页

● 集群所有代理统计页

单击功能导航栏中的Brokers，即可进入代理统计页。如图4-8所示。

– 图中①区域代理列表，包含总的字节流入和字节流出。

– 图中②集群监控信息。

图4-8 所有代理统计页

● 具体代理统计页

单击id列表中具体的Broker，即可查看对应代理的统计信息。如图4-9所示。

– 图中①区域表示对应代理总的统计信息，包括队列数、分区数、分区leader 数、消息速率占比、写入字节占比以及流出字节占比。

– 图中②区域表示代理监控信息。

(21)

图4-9 具体 Broker 信息

● 查看实例的Topic

在导航栏选择Topic，并在下拉列表中选择List。页面如图4-10所示，展示了队列列表以及分区数等。

须知

列表中以“__”开头的队列为内部队列，严禁操作，否则可能导致业务问题。

图4-10 查看实例的 Topic

● 队列详情页

单击具体的Topic名称，进入如图4-11所示页面。

(22)

– 图中①区域表示队列基本信息，包括副本数(Replication)，分区数(Number of Partitions)，消息数(Sum of partition offsets)等。

– 图中②区域表示代理与队列分区的对应关系。

– 图中③区域表示该队列的消费组列表。单击消费组名称可进入该消费组的详情页。

– 图中④区域表示队列的配置信息。详情参考kafka队列官方配置文档(https://

kafka.apache.org/documentation/#topicconfigs)。

– 图中⑤区域表示队列监控数据统计。

– 图中⑥区域表示队列分区信息，包括分区消息数(Latest Offset)，分区 leader(Leader)，副本列表(Replicas)，同步副本列表(In Sync Replicas)。

图4-11 队列详情页

● 查看消费组列表

导航栏中单击Consumers页签，即可查看当前集群中的消费组列表。

说明

只显示14天内有消费记录的消费组。

图4-12 集群的消费组列表

(23)

● 查看消费组详情页

单击消费组名称可进入消费组详情页面，展示消费组消费的所有队列列表以及每个队列的可消费数(Total Lag)。

图4-13 消费组详情页面

● 查看消费组队列详情页

单击队列名称，即可进入详情页面，查看消费组消费在队列中每个分区的消费状态。包括分区编号(Partition)，分区消息数(LogSize)，分区消费进度(Consumer Offset)，分区剩余可消费数(Lag)，最近消费该分区的消费者(Consumer Instance Owner)。

图4-14 消费组队列详情页面

4.4 跨 VPC 访问 Kafka 实例

操作场景

Kafka客户端和Kafka实例在同一区域的不同VPC中，由于VPC之间逻辑隔离，Kafka客户端和Kafka实例不能直接通信。您可以通过以下任意一个方式实现跨VPC访问：

● 创建VPC对等连接，将两个VPC的网络打通，实现跨VPC访问。具体步骤请参考

VPC对等连接说明。

● 创建一个云连接实例，然后在创建的云连接实例中加载需要互通的VPC，实现跨 VPC访问。具体步骤请参考同区域同帐号VPC互通。

(24)

● 利用VPC终端节点在不同VPC间建立跨VPC的连接通道，实现Kafka客户端通过内网访问Kafka实例。

本章节主要介绍通过VPC终端节点实现跨VPC访问的方法。

VPC终端节点由“终端节点服务”和“终端节点”两种资源实例组成。

● 终端节点服务：将Kafka实例配置为VPC终端节点支持的服务，可以被终端节点连接和访问。

● 终端节点：用于在VPC和终端节点服务之间建立便捷、安全、私密的连接通道。

图4-15 跨 VPC 访问 Kafka 实例原理图

(25)

操作流程

图4-16 跨 VPC 访问 Kafka 实例流程图

创建终端节点服务

说明

步骤4 单击Kafka实例名称，进入实例详情页面。

步骤5 获取Kafka实例的地址和对应的Port ID。

(26)

图4-17 获取 Kafka 实例的地址和对应的 Port ID

步骤6 查看Kafka实例所在的VPC名称。

图4-18 查看 Kafka 实例所在的 VPC 名称

步骤7 单击VPC名称，进入VPC基本信息页签，获取VPC ID。

图4-19 获取 VPC ID

步骤8 调用VPC终端节点的API创建终端节点服务，API详情请参考创建终端节点服务。

curl -i -k -H 'Accept:application/json' -H 'Content-Type:application/json;charset=utf8' -X POST -H "X-Auth- Token:$token" -d '{"port_id":"38axxxeac","vpc_id":"706xxx888","ports":[{"protocol":"TCP","client_port":

9011,"server_port":9011 }],"approval_enabled":false,"service_type":"interface","server_type":"VM"}' https://

{endpoint}/v1/{project_id}/vpc-endpoint-services

(27)

● token：Token是系统颁发给IAM用户的访问令牌，承载用户的身份、权限等信息，获取方法请参考获取用户Token。

● port_id：输入步骤5中获取的其中一个Port ID。

● vpc_id：输入步骤7中获取的VPC ID。

● endpoint：获取VPC终端节点的终端节点，区域必须与Kafka实例保持一致。

● project_id：获取VPC终端节点的项目ID，区域必须与Kafka实例保持一致。

记录响应信息中的“service_name”参数值，此参数表示终端节点服务的名称。

步骤9 参考步骤8，为步骤5中其他Port ID创建终端节点服务，并记录终端节点服务的名称。

----结束

（可选）添加白名单

Kafka客户端和Kafka实例属于不同帐号时，将Kafka客户端所在帐号的ID添加到终端节点服务的白名单中，完成跨帐号终端节点的访问授权，具体操作步骤请参考添加白名单。

购买终端节点

步骤1 在管理控制台左上角单击，选择“网络 > VPC终端节点”，进入终端节点页面。

步骤2 单击“购买终端节点”，进入“购买终端节点”页面。

步骤3 设置如下参数。

● 区域：与Kafka实例保持一致。

● 服务类别：选择“按名称查找服务”。

● 服务名称：输入步骤8中记录的终端节点服务名称，单击“验证”。显示“已找到服务”后，继续后续操作。

● 虚拟私有云：选择Kafka客户端所属的VPC。

● 子网：选择Kafka客户端所属的子网。

● 节点IP：选择“自动分配”。

其他参数保持默认，如果想要了解更多的参数信息，请参考购买终端节点。

(28)

图4-20 终端节点参数设置

步骤4 单击“立即购买”，进入规格确认页面。

步骤5 确认无误后，提交请求。

步骤6 购买成功后，返回终端节点页面，查看终端节点状态是否为“已接受”，“已接受”

表示终端节点已成功连接至终端节点服务。

图4-21 查看终端节点状态

步骤7 单击终端节点ID，在“基本信息”页签，查看并记录节点IP。

您可以使用节点IP访问终端节点服务，进行跨VPC资源通信。

图4-22 查看节点 IP

步骤8 参考步骤1~步骤7，为步骤9中创建的终端节点服务购买终端节点，查看并记录节点 IP。

----结束

(29)

修改 advertised.listeners IP

步骤2 单击Kafka实例名称，进入实例详情页面。

步骤3 在“基本信息”页签，修改“跨VPC访问”的advertised.listeners IP，

advertised.listeners IP为步骤7和步骤8中记录的节点IP。

须知

节点IP必须与Port ID一一对应，否则会导致网络不通。

图4-23 修改 advertised.listeners IP

----结束

验证接口连通性

参考连接未开启SASL的Kafka实例或者连接已开启SASL的Kafka实例，测试是否可以生产和消费消息。

测试接口连通性时，注意以下两点：

● 连接Kafka实例的地址为“advertised.listeners IP:9011”，以图4-23为例，连接 Kafka实例的地址为

“192.168.0.71:9011,192.168.0.11:9011,192.168.0.21:9011”。

● 在Kafka实例安全组的入方向规则中放通9011端口。

4.5 制作并替换证书

Kafka客户端连接开启SASL的Kafka实例时，您可以使用分布式消息服务Kafka版提供的证书，也可以使用自己制作的证书。如果使用自己制作的证书，请参考本章节将分布式消息服务Kafka版提供的证书替换为自己的证书。

制作并替换证书为白名单特性，需要联系后台管理人员开通后方可使用。

说明

替换证书将重启实例，请谨慎操作。

前提条件

● 已准备一台Linux系统的服务器。

(30)

● 实例已开启Kafka SASL_SSL。

步骤一：制作证书

步骤1 登录Linux系统的服务器，执行以下命令，生成server.keystore.jks证书的孵化器。

keytool -genkey -keystore server.keystore.jks -alias localhost -validity 3650 -keyalg RSA

按照提示信息输入keystore密码。密码需要满足如下要求：

● 8~32个字符。

● 至少包含以下字符中的3种：大写字母、小写字母、数字、特殊字符`~!@#$

%^&*()-_=+\|[{}]:'",<.>/?和空格，并且不能以-开头。

● 非弱密码（如果不能确定哪些是弱密码，请在替换证书对话框中输入密码，系统会自动判定是否为弱密码）。

按照提示信息输入证书拥有者的信息，如姓名、单位、所在城市等。

步骤2 执行以下命令，创建CA的认证证书。

openssl req -new -x509 -keyout ca-key -out ca-cert -days 3650

按照提示信息输入PEM密码。

按照提示信息输入证书拥有者的信息。

步骤3 执行以下命令，通过CA创建一个服务器的Truststore证书，有了Truststore证书才可以进行证书有效性的检测。

keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert

按照提示信息输入服务器证书的Truststore密码。密码需要满足如下要求：

● 8~32个字符。

%^&*()-_=+\|[{}]:'",<.>/?和空格，并且不能以-开头。

● 非弱密码（如果不能确定哪些是弱密码，请在替换证书对话框中输入密码，系统会自动判定是否为弱密码）。

出现如下提示信息时，输入“y”。

Trust this certificate？

步骤4 执行以下命令，通过CA创建一个客户端的Truststore证书。

keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert

按照提示信息输入客户端证书的Truststore密码。此密码用于客户端连接Kafka实例时，配置文件中“ssl.truststore.password”参数的值。

步骤5 执行以下命令，为服务器证书签名。

1. 导出服务器证书“server.cert-file”。

keytool -keystore server.keystore.jks -alias localhost -certreq -file server.cert-file

按照提示信息输入步骤1中设置的keystore密码。

2. 通过CA给服务器证书进行签名处理。

openssl x509 -req -CA ca-cert -CAkey ca-key -in server.cert-file -out server.cert-signed -days 3650 - CAcreateserial

按照提示信息输入步骤2中设置的PEM密码。

(31)

3. 将CA证书导入到服务器keystore。

keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert

4. 将已签名的服务器证书导入到服务器keystore。

keytool -keystore server.keystore.jks -alias localhost -import -file server.cert-signed

步骤6 导出证书“server.keystore.jks”和“server.truststore.jks”到本地PC机。

图4-24 证书目录

----结束

步骤二：替换证书

说明

步骤4 单击待替换证书的实例名称，进入实例详情页面。

步骤5 在“高级配置 > SSL证书”后，单击，弹出“替换SSL证书”对话框。

图4-25 高级配置

步骤6 参考表4-1，设置替换SSL证书的参数。

(32)

图4-26 替换 SSL 证书

表4-1 替换 SSL 证书参数说明

参数名称说明

Key密码输入制作证书中设置的keystore密码 Keystore密码输入制作证书中设置的keystore密码 Keystore文件导入“server.keystore.jks”证书 Truststore密码输入服务端证书的Truststore密码 Truststore文件导入“server.truststore.jks”证书

步骤7 单击“确定”，弹出“替换SSL证书”对话框。

(33)

图4-27 确认是否替换证书

步骤8 单击“确认”，完成证书的替换。

在“后台任务管理”页签，查看证书是否替换成功。

图4-28 查看证书是否替换成功

说明

证书替换成功后，在实例详情页单击，下载的证书为分布式消息服务Kafka版提供的证书，并非您自己制作的证书。

----结束

步骤三：修改客户端配置文件

替换证书后，需要在客户端的“consumer.properties”和“producer.properties”文件中，分别修改“ssl.truststore.location”和“ssl.truststore.password”参数。

security.protocol=SASL_SSL

ssl.truststore.location=/opt/kafka_2.11-2.3.0/config/client.truststore.jks ssl.truststore.password=dms@kafka

● ssl.truststore.location配置为client.truststore.jks证书的存放路径。

● ssl.truststore.password为客户端证书的Truststore密码。

(34)

4.6 配置 SSL 双向认证

操作场景

SSL双向认证是指在通信认证时，将同时验证客户端和服务端的证书，以确保通信中涉及的双方都是受信任的。

开启SSL双向认证，主要用于对安全性要求较高的场景。

SSL双向认证为白名单特性，需要联系后台管理人员开通后方可使用。

说明

开启/关闭SSL双向认证将重启实例，请谨慎操作。

前提条件

● 已准备一台Linux系统的服务器。

● 实例已开启Kafka SASL_SSL。

配置 SSL 双向认证流程

图4-29 配置 SSL 双向认证流程图

(35)

步骤一：制作客户端验证服务端的证书

步骤1 登录Linux系统的服务器，执行以下命令，生成server.keystore.jks证书的孵化器。

keytool -genkey -keystore server.keystore.jks -alias localhost -validity 3650 -keyalg RSA

● 8~32个字符。

%^&*()-_=+\|[{}]:'",<.>/?和空格，并且不能以-开头。

● 非弱密码（如果不能确定哪些是弱密码，请在开启SSL双向认证对话框中输入密码，系统会自动判定是否为弱密码）。

步骤3 执行以下命令，从步骤1中生成的server.keystore.jks中导出证书，并命名为

“server.crt”。

keytool -keystore server.keystore.jks -alias localhost -certreq -file server.crt

按照提示信息输入keystore密码。

步骤4 执行以下命令，使用CA的私钥对server.crt进行签名，并为签名后的证书命名为

“server-signed.crt”。

openssl x509 -req -CA ca-cert -CAkey ca-key -in server.crt -out server-signed.crt -days 3650 -CAcreateserial

按照提示信息输入步骤2中的PEM密码。

步骤5 执行以下命令，将CA证书和“server-signed.crt”导入秘钥仓库。

keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert

keytool -keystore server.keystore.jks -alias localhost -import -file server-signed.crt

步骤6 执行以下命令，使客户端信任服务端证书。

keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert

按照提示信息输入client.truststore.jks的密码。密码需要满足如下要求：

● 8~32个字符。

%^&*()-_=+\|[{}]:'",<.>/?和空格，并且不能以-开头。

步骤7 导出“client.truststore.jks”和“server.keystore.jks”证书到本地PC机。

----结束

(36)

步骤二：制作服务端验证客户端的证书

步骤1 登录Linux系统的服务器，执行以下命令，生成client.keystore.jks证书的孵化器。

keytool -genkey -keystore client.keystore.jks -alias localhost -validity 3650 -keyalg RSA

● 8~32个字符。

%^&*()-_=+\|[{}]:'",<.>/?和空格，并且不能以-开头。

步骤3 执行以下命令，从步骤1中生成的client.keystore.jks中导出证书，并命名为

“client.crt”。

keytool -keystore client.keystore.jks -alias localhost -certreq -file client.crt

步骤4 执行以下命令，使用CA的私钥对client.crt进行签名，并为签名后的证书命名为

“client-signed.crt”。

openssl x509 -req -CA ca-cert -CAkey ca-key -in client.crt -out client-signed.crt -days 3650 -CAcreateserial

按照提示信息输入步骤2中的PEM密码。

步骤5 执行以下命令，将CA证书和“client-signed.crt”导入秘钥仓库。

keytool -keystore client.keystore.jks -alias CARoot -import -file ca-cert

keytool -keystore client.keystore.jks -alias localhost -import -file client-signed.crt

步骤6 执行以下命令，使服务端信任客户端证书。

keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert

按照提示信息输入server.truststore.jks的密码。

密码需要满足如下要求：

● 8~32个字符。

%^&*()-_=+\|[{}]:'",<.>/?和空格，并且不能以-开头。

步骤7 导出“server.truststore.jks”和“client.keystore.jks”证书到本地PC机。

----结束

(37)

步骤三：开启 SSL 双向认证

说明

步骤4 单击待开启SSL双向认证的实例名称，进入实例详情页面。

步骤5 在“高级配置 > SSL双向认证”后，单击，开启SSL双向认证。

步骤6 在弹出的“SSL双向认证”对话框中，参考表4-2，设置开启SSL双向认证的参数。

图4-30 开启 SSL 双向认证

表4-2 开启 SSL 双向认证参数说明

参数名称说明

Key密码输入server.keystore.jks的密码 Keystore密码输入server.keystore.jks的密码 Keystore文件导入“server.keystore.jks”证书 Truststore密码输入server.truststore.jks的密码 Truststore文件导入“server.truststore.jks”证书

(38)

须知

开启SSL双向认证将重启实例，请谨慎操作。

步骤7 单击“确定”，完成开启SSL双向认证。

----结束

步骤四：修改客户端配置文件

开启SSL双向认证后，需要在客户端的“consumer.properties”和

“producer.properties”文件中，分别修改服务端证书配置，并增加客户端证书配置。

security.protocol=SSL

#增加以下的客户端证书配置

ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks ssl.keystore.password=test123

ssl.key.password=test123

● security.protocol配置证书协议类型，开启SSL双向认证时，必须设置为SSL。

● ssl.trustsore.location配置为client.truststore.jks证书的存放路径。

● ssl.truststore.password为client.truststore.jks的密码。

● ssl.keystore.location配置为client.keystore.jks证书的存放路径。

● ssl.keystore.password配置为client.keystore.jks的密码。

● ssl.key.password配置为client.keystore.jks的密码。

关闭 SSL 双向认证

说明

步骤4 单击待关闭SSL双向认证的实例名称，进入实例详情页面。

步骤5 在“高级配置 > SSL双向认证”后，单击，关闭SSL双向认证。

须知

关闭SSL双向认证将重启实例，请谨慎操作。

(39)

步骤6 关闭SSL双向认证后，需要在客户端的“consumer.properties”和

“producer.properties”文件中，分别修改服务端证书协议类型，并删除客户端证书配置。

security.protocol=SASL_SSL

#删除以下的客户端证书配置

ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks ssl.keystore.password=test123

ssl.key.password=test123

security.protocol配置证书协议类型，关闭SSL双向认证时，必须设置为SASL_SSL。

“ssl.truststore.location”、“ssl.truststore.password”和

“ssl.endpoint.identification.algorithm”无需修改。

----结束

(40)

5 ^实例管理

5.1 变更实例规格

操作场景

Kafka实例创建成功后，您可以根据业务需要，扩容基准带宽、存储空间和代理个数。

老实例支持扩容存储空间和基准带宽，不支持同时扩容存储空间和基准带宽，如果需要，可以分别进行扩容。

新实例支持扩容存储空间和代理个数，不支持同时扩容存储空间和代理个数，如果需要，可以分别进行扩容。

怎样判断老实例和新实例？

● 老实例：在实例列表中，“规格”显示为带宽（如100MB/s）的实例。

● 新实例：在实例列表中，“规格”显示为ECS规格和代理个数（如 c6.2u4g.cluster*4 broker）的实例。

图5-1 实例列表

当前只支持扩容，不支持缩容。

扩容存储空间有次数限制，只能扩容20次。

扩容基准带宽/代理个数，可以扩大总分区数。扩容基准带宽/代理个数时，系统会根据当前磁盘大小进行相应比例的存储空间扩容。

操作步骤

(41)

说明

步骤4 在实例所在行，单击“更多 > 变更规格”，进入“分布式消息服务Kafka版变更规格”

页面。

步骤5 根据实际情况选择扩容存储空间、代理个数或者基准带宽。

老实例支持扩容存储空间和基准带宽，不支持同时扩容存储空间和基准带宽，如果需要，可以分别进行扩容。

新实例支持扩容存储空间和代理个数，不支持同时扩容存储空间和代理个数，如果需要，可以分别进行扩容。

老实例扩容步骤如下：

● 扩容基准带宽。

在“规格”中，选择扩容后的带宽，单击“下一步”。确认扩容信息无误后，单击“提交”。

在实例列表页面的“规格”中查看扩容后的带宽大小。

说明

– 扩容带宽是通过扩容代理实现的，原来代理不受影响，业务也不受影响。

– 新创建的Topic才有可能创建在新代理上，原来Topic还分布在原有代理上，如果需要使用新扩容的代理，可以通过Kafka Manager迁移Topic或者新建Topic来实现。

– 已开启公网访问，且设置了弹性IP地址的实例，在扩容基准带宽时，需要为新扩容的代理设置弹性IP地址。

● 扩容存储空间。

在“存储空间”中，选择扩容后的存储空间大小，单击“下一步”。确认扩容信息无误后，单击“提交”。

在实例列表页面的“可用存储空间”中查看扩容后的存储空间大小。

说明

– 扩容存储空间不会影响业务。

– 可用存储空间=实际存储空间-用于存储日志和ZK的数据的存储空间-格式化磁盘的损耗。

例如，实际扩容存储空间到700GB，用于存储日志和ZK的数据的存储空间为100GB，

格式化磁盘损耗7GB，那么扩容后的可用存储空间为593GB。

新实例扩容步骤如下：

● 扩容存储空间。

在“变更方式”中，选择“存储扩容”，在“单个代理存储空间”中，选择扩容后的单个代理的存储空间大小，单击“下一步”。确认扩容信息无误后，单击

“提交”。

在实例列表页面的“可用存储空间”中查看扩容后的总存储空间大小（即扩容后的单个代理的存储空间*代理个数）。

(42)

说明

– 扩容存储空间不会影响业务。

– 可用存储空间=实际存储空间-用于存储日志和ZK的数据的存储空间-格式化磁盘的损耗。

例如，实际扩容存储空间到700GB，用于存储日志和ZK的数据的存储空间为100GB，

格式化磁盘损耗7GB，那么扩容后的可用存储空间为593GB。

● 扩容代理个数。

在“变更方式”中，选择“代理扩容”，在“代理个数”中，选择扩容后的代理个数，单击“下一步”。确认扩容信息无误后，单击“提交”。

在实例列表页面的“规格”中查看扩容后的代理个数。

说明

– 扩容代理个数不会影响原来的代理，业务也不受影响。

– 新创建的Topic才有可能创建在新代理上，原来Topic还分布在原有代理上，如果需要使用新扩容的代理，可以通过Kafka Manager迁移Topic或者新建Topic来实现。

– 已开启公网访问，且设置了弹性IP地址的实例，在扩容代理个数时，需要为新扩容的代理设置弹性IP地址。

----结束

5.2 查看实例

操作场景

本节介绍如何在分布式消息服务Kafka版管理控制台查看Kafka实例的详细信息。例如，连接Kafka实例时，需要获取连接IP和端口。

操作步骤

说明

步骤4 Kafka实例支持通过筛选来查询对应的Kafka实例。当前支持的筛选条件为项目、状态、名称、ID和连接地址。Kafka实例状态请参见表5-1。

表5-1 Kafka 实例状态说明

状态说明

创建中创建Kafka实例后，在Kafka实例状态进入运行中之前的状态。

(43)

状态说明

运行中 Kafka实例正常运行状态。

在这个状态的实例可以运行您的业务。

故障 Kafka实例处于故障状态。

启动中 Kafka实例从已冻结到运行中的中间状态。

重启中 Kafka实例正在进行重启操作。

变更中 Kafka实例正在进行规格变更、开启转储、开启公网访问等操作。

变更失败 Kafka实例处于规格变更、开启转储、开启公网访问操作失败的状态。

已冻结 Kafka实例处于已冻结状态。

冻结中 Kafka实例从运行中到已冻结的中间状态。

创建失败 Kafka实例处于创建失败的状态。

步骤5 单击Kafka实例的名称，进入该Kafka实例的基本信息页面，查看Kafka实例的详细信息。

表5-2为连接实例的相关参数，其他参数，请查看页面显示。

表5-2 连接参数说明 信息类

型

参数说明

连接信息

内网连接地址未开启公网访问时，连接实例的地址。

显示的连接地址个数为实例代理个数。

Manager内网访问地址

未开启公网访问时，连接Kafka Manager的地址。

公网访问是否开启公网访问开关。

公网连接地址开启公网访问后，连接实例的地址。

仅当开启了公网访问才显示该参数。

Manager公网访问地址

开启公网访问后，连接Kafka Manager的地址。

仅当开启了公网访问才显示该参数。

----结束

(44)

5.3 重启实例

操作场景

分布式消息服务Kafka版管理控制台支持重启运行中的Kafka实例，且可实现批量重启 Kafka实例。

须知

在Kafka实例重启过程中，客户端的生产与消费消息等请求会被拒绝。

前提条件

只有当Kafka实例处于“运行中”或“故障”状态，才能执行重启操作。

操作步骤

说明

步骤4 通过以下任意一种方法，重启Kafka实例。

● 勾选Kafka实例名称左侧的方框，可选一个或多个，单击信息栏左上侧的“重启”。

● 在待重启Kafka实例所在行，单击“重启”。

● 单击Kafka实例名称，进入实例详情页面。单击右上角的“重启”。

步骤5 单击“确定”，完成重启Kafka实例。

重启Kafka实例大约需要3到15分钟。Kafka实例重启成功后，实例状态切换为“运行中”。

说明

重启Kafka实例只会重启实例进程，不会重启实例所在虚拟机。

----结束

(45)

5.4 删除实例

操作场景

分布式消息服务Kafka版管理控制台支持删除Kafka实例，且可实现批量删除Kafka实例、一键式删除创建失败的Kafka实例、单个删除创建失败的实例记录。

须知

Kafka实例删除后，实例中原有的数据将被删除，且没有备份，请谨慎操作。

前提条件

● Kafka实例状态为运行中、故障、已冻结的按需付费实例才能执行删除操作。

● 包年/包月类型的Kafka实例，不支持进行删除和批量删除操作。若不再使用，可单击“操作”栏下的“更多 > 退订”进行退订。

删除 Kafka 实例

说明

步骤4 通过以下任意一种方法，删除Kafka实例。

● 勾选Kafka实例名称左侧的方框，可选一个或多个，单击信息栏左上侧的“更多 >

删除”。

● 在待删除Kafka实例所在行，单击“更多 > 删除”。

● 单击Kafka实例名称，进入实例详情页面。单击右上角的“更多 > 删除”。

说明

Kafka实例状态为创建中、启动中、变更中、变更失败、重启中时不允许执行删除操作。

步骤5 单击“确定”，完成删除Kafka实例。

删除Kafka实例大约需要1到60秒。

----结束

删除创建失败的 Kafka 实例

(46)

说明

若当前存在创建失败的Kafka实例，界面信息栏会显示“创建失败任务”及失败数量信息。

说明

创建失败的实例，不会占用其他资源。

步骤4 单击“创建失败任务”后的图标或者数量。

弹出“创建失败任务”对话框。

步骤5 在“创建失败任务”界面删除创建失败的Kafka实例。

● 单击“清理失败任务”，一键式删除所有创建失败的Kafka实例。

● 单击需要删除的Kafka实例右侧的“删除任务”，依次删除创建失败的Kafka实例。

----结束

5.5 修改实例信息

创建Kafka实例成功后，您可以根据自己的业务情况对Kafka实例的部分参数进行调整，包括实例名称、描述、维护时间窗、安全组和容量阈值策略等。

操作步骤

说明

步骤4 单击Kafka实例的名称，进入实例详情页面。

步骤5 以下参数支持修改。

● 实例名称

● 维护时间窗

● 企业项目（修改企业项目，不会重启实例）

● 转储（转储开启后，暂不支持动态关闭）

支持云审计的操作列表_分布式消息服务Kafka版_用户指南_审计_华为云

用户指南

目 录

1 权限管理...1

2 准备实例依赖资源... 4

3 购买实例...6

4 连接 Kafka...10

5 实例管理...36

6 Topic 管理...53

7 消息管理...60

8 用户管理...62

9 修改配置参数...65

10 调整资源配额...69

11 监控... 71

12 审计... 82

1 权限管理

1.1 创建用户并授权使用 DMS for Kafka

前提条件

Kafka系统策略。若您需要对除DMS for Kafka之外的其它服务授权，IAM支持服务的

示例流程

1.2 DMS for Kafka 自定义策略

DMS 自定义策略样例

2 准备实例依赖资源

概述

准备依赖资源

3 购买实例

操作场景

购买前须知

操作步骤

4 连接 Kafka

4.1 连接未开启 SASL 的 Kafka 实例

display/KAFKA/Clients

前提条件

命令行模式连接实例

[kafka_tar]

4.2 连接已开启 SASL 的 Kafka 实例

前提条件

命令行模式连接实例

[kafka_tar]

4.3 连接和查看 Kafka Manager

前提条件

登录 Kafka Manager

查看 Kafka Manager

▪

▪

▪

▪

▪

▪

kafka.apache.org/documentation/#topicconfigs)。

4.4 跨 VPC 访问 Kafka 实例

操作场景

VPC对等连接说明。

操作流程

创建终端节点服务

（可选）添加白名单

购买终端节点

修改 advertised.listeners IP

验证接口连通性

4.5 制作并替换证书

前提条件

步骤一：制作证书

步骤二：替换证书

步骤三：修改客户端配置文件

4.6 配置 SSL 双向认证

操作场景

前提条件

配置 SSL 双向认证流程

步骤一：制作客户端验证服务端的证书

步骤二：制作服务端验证客户端的证书

步骤三：开启 SSL 双向认证

步骤四：修改客户端配置文件

关闭 SSL 双向认证

5 实例管理

5.1 变更实例规格

操作场景

操作步骤

5.2 查看实例

操作场景

操作步骤

目录

1 ^权限管理

2 ^{准备实例依赖资源}

3 ^购买实例

4 ^{连接 Kafka}

5 ^实例管理