本章节主要介绍ALM38000 Kafka服务不可用的告警。 告警解释 系统按照30秒的周期检测Kafka服务是否可用，当Kafka服务不可用，系统产生此告警。 当Kafka服务恢复正常，告警自动清除。 告警属性 告警ID 告警级别 是否自动清除 38000 紧急 是 告警参数 参数名称 参数含义 来源 产生告警的集群名称。 服务名 产生告警的服务名称。 角色名 产生告警的角色名称。 主机名 产生告警的主机名。 对系统的影响 集群无法对外提供Kafka服务，用户无法执行新的Kafka任务。 可能原因 KrbServer服务故障。（非普通模式集群） ZooKeeper服务故障或无响应。 Kafka服务中Broker实例状态异常。 处理步骤 检查KrbServer服务状态（普通模式集群跳过此步骤） 1. 在FusionInsight Manager管理界面，选择“集群 > 待操作集群的名称 > 服务 > KrbServer”。 2. 查看KrbServer服务的运行状态是否为“良好”。 是，执行步骤5。 否，执行步骤3。 3. 参考“ALM25500 KrbServer服务不可用”的处理步骤进行操作。 4. 再次执行步骤2。 检查ZooKeeper服务状态 5. 查看ZooKeeper服务的运行状态是否为“良好”。 是，执行步骤8。 否，执行步骤6。 6. 如果ZooKeeper服务已停止，则启动ZooKeeper服务，否则参考“ALM13000 ZooKeeper服务不可用”的处理步骤进行操作。 7. 再次执行步骤5。

本文主要介绍消费组问题。 Kafka实例是否需要创建消费组、生产者和消费者？ 不需要单独创建消费组、生产者和消费者，在使用时自动生成，实例创建后，直接使用即可。 连接Kafka实例后，生产消息和消费消息，请参考向Kafka实例生产消息和消费消息。 如果消息组中没有在线的消费者（如empty状态），是否14天后会自动被删除？ 消息组中没有在线的消费者（如empty状态），14天后是否会自动被删除与offsets.retention.minutes参数有关： 2020年6月16日前创建的实例，offsets.retention.minutes默认为2147483646分钟，约1491308天，消费组14天后不会被删除。 2020年6月16日以及之后创建的实例，offsets.retention.minutes默认为20160分钟，即14天，消费组14天后会自动被删除。 原因如下： Kafka通过offsets.retention.minutes参数控制消费组中offsets保留时间，在此时间内如果没有提交offset，offsets将会被删除。Kafka判定消息组中没有在线的消费者（如empty状态），且没有offsets时，将会删除此消费组。 客户端删除消费组后，在Kafka Manager中仍可以看到此消费组？ 客户端删除消费组后，此消费组已经被删除了。在Kafka Manager中仍可以看到此消费组，是因为Kafka Manager存在缓存。 通过以下任意一种方法解决此问题： 重启Kafka Manager。 Kafka Manager只显示14天内有消费记录的消费组，如果您不想重启Kafka Manager，可以等待14天后Kafka Manager自动清除此消费组。

操作步骤 步骤 1 登录分布式消息服务Kafka控制台，选择Kafka实例所在的区域。 步骤 2 在Kafka实例名称后，单击，进入云监控该实例的监控指标页面。 步骤 3 在实例监控指标页面中，找到需要创建告警的指标项，鼠标移动到指标区域，然后单击指标右上角的，创建告警规则，跳转到创建告警规则页面。 步骤 4 在告警规则页面，设置告警信息。 创建告警规则操作，请查看《云监控服务用户指南》的“创建告警规则和告警通知”章节。 1. 设置告警名称和告警的描述。 2. 设置告警策略和告警级别。 如下图所示，在进行指标监控时，如果连续3个周期，磁盘容量使用率原始值超过85%，则产生告警，如果未及时处理，则产生告警通知。 图设置告警策略和告警级别 3. 设置“发送通知”开关。当开启时，设置告警生效时间、产生告警时通知的对象以及触发的条件。 4. 单击“立即创建”，等待创建告警规则成功。

分布式消息服务Kafka引擎类型云原生引擎调整为白名单特性，更多了解请查看快速入门创建实例引擎类型说明。 调整时间 2024年7月5日 影响范围 所有区域 调整影响 新用户默认不开放云原生引擎订购开通，如需要该特性，请联系技术支持开通后使用。 已购买云原生引擎实例的用户，原实例仍可正常使用，续费、扩容等操作不受影响。

面向长期稳定客户，我们特别推出了更加优惠的包年订阅选项，旨在通过深度折扣，帮助客户显著降低资源单位成本，优化整体支出。 自2024年11月9日起订购和续订分布式消息服务Kafka产品更长包周期即可享受2年7折，3年5折优惠。 注意 本次包年优惠适用于新资费产品范围，具体支持资源池请参阅

分布式消息服务Kafka主机类型通用型规格调整为白名单特性，更多了解请查看计费项产品规格说明。 调整时间 2024年6月25日 影响范围 所有区域 调整影响 新用户默认不开放主机类型通用型规格订购开通，如需要该特性，请联系技术支持开通后使用。 已购买主机类型通用型规格实例的用户，原实例仍可正常使用，续费、扩容等操作不受影响。

尊敬的天翼云客户，分布式消息服务Kafka自2025年12月27日起，订购和续订2年、3年选项默认不开放，调整为白名单特性。 调整时间 2025年12月27日 影响范围 所有区域 调整影响 新订购和续订的实例默认不开放2年、3年选项，您可以选择1年包年选项，如仍需要23年包周期选项，请联系技术支持开通后使用。 已购买2年、3年且还在服务期间的实例仍可继续正常使用不受影响。

本章节主要介绍翼MapReduce服务的集群管理功能。 集群生命周期管理 翼MR支持集群的生命周期管理包括创建集群和删除集群。 创建集群：支持用户定制集群的类型，组件范围，各类型的节点数、虚拟机规格、可用区、VPC网络、认证信息，翼MR将为用户自动创建一个符合配置的集群，全程无需用户参与；同时支持用户在集群中运行自定义内容；支持快速创建多应用场景集群，比如创建Hadoop分析集群、HBase集群、Kafka集群。大数据平台同时支持部署异构集群，在集群中存在不同规格的虚机，允许在CPU类型，硬盘容量，硬盘类型，内存大小灵活组合。在集群中支持多种虚机规格混合使用。 删除集群：当按需计费的集群不再需要时（包括集群中的数据和配置），用户可以选择删除集群，翼MR会将集群相关的资源全部删除。 创建集群 通过在翼MR服务管理面，客户可以按需创建翼MR集群，通过选择集群所建的区域及使用的云资源规格，一键式创建适合企业业务的翼MR集群。翼MR服务会根据用户选择的集群类型、版本和节点规格，帮助客户自动完成企业级大数据平台的安装部署和参数调优。 翼MR服务为客户提供完全可控的大数据集群，客户在创建时可设置虚拟机的登录方式（密码或者密钥对），所创建的翼MR集群资源完全归客户所用。同时翼MR支持在最小可在两节点4U8G的ECS上部署大数据集群，为客户测试开发提供更多的灵活选择。 翼MR集群类型包括分析集群、流式集群和混合集群。 分析集群：用来做离线数据分析，提供的是Hadoop体系的组件。 流式集群：用来做流处理任务，提供的是流式处理组件。 混合集群：既可以用来做离线数据分析，又可以用来做流处理任务，提供的是Hadoop体系的组件和流式处理组件。 自定义：根据业务需求，可以灵活搭配所需组件（翼MR 3.x及后续版本）。 翼MR集群节点类型包括Master节点、Core节点和Task节点。 Master节点：集群中的管理节点，分布式系统的Master进程和Manager以及数据库均部署在该节点；该类型节点不可扩容。该类型节点的处理能力决定了整个集群的管理上限，MRS服务支持将Master节点规格提高，以支持更大集群的管理。 Core节点：支持存储和计算两种目标的节点，可扩容、缩容。因承载的数据存储，因此在缩容时，为保证数据不丢失，有较多限制，无法进行弹性伸缩。 Task节点：仅用于计算的节点，可扩容、缩容。因只承载计算任务，因此可以进行弹性伸缩。 翼MR创建集群方式支持自定义创建集群和快速创建集群两种。 自定义创建集群：自定义创建可以灵活地选择计费模式、配置项，针对不同的应用场景，可以选择不同规格的弹性云服务器，全方位贴合您的业务诉求。 快速创建集群：用户可以根据应用场景，快速创建对应配置的集群，提高了配置效率，更加方便快捷。当前支持快速创建Hadoop分析集群、HBase集群、Kafka集群。 − Hadoop分析集群：Hadoop分析集群完全使用开源Hadoop生态，采用YARN管理集群资源，提供Hive、Spark离线大规模分布式数据存储和计算，SparkStreaming、Flink流式数据计算，Presto交互式查询，Tez有向无环图的分布式计算框等Hadoop生态圈的组件，进行海量数据分析与查询。 − HBase集群：HBase集群使用Hadoop和HBase组件提供一个稳定可靠，性能卓越、可伸缩、面向列的分布式云存储系统，适用于海量数据存储以及分布式计算的场景，用户可以利用HBase搭建起TB至PB级数据规模的存储系统，对数据轻松进行过滤分析，毫秒级得到响应，快速发现数据价值。 − Kafka集群：Kafka集群使用Kafka和Storm组件提供一个开源高吞吐量，可扩展性的消息系统。广泛用于日志收集、监控数据聚合等场景，实现高效的流式数据采集，实时数据处理存储等。

Hudi Hudi是数据湖的文件组织层，对Parquet格式文件进行管理提供数据湖能力，支持多种计算引擎，提供IUD接口，在 HDFS的数据集上提供了插入更新和增量拉取的流原语。 资源池 ModelArts提供的大规模计算集群，可应用于模型开发、训练和部署。支持公共资源池和专属资源池两种，分别为共享资源池和独享资源池。ModelArts默认提供公共资源池。专属资源池需单独购买，专属使用，不与其他用户共享。 Hue Hue是一组WEB应用，用于和MRS大数据组件进行交互，能够帮助用户浏览HDFS，进行Hive查询，启动MapReduce任务等，它承载了与所有MRS大数据组件交互的应用。 Kafka Kafka是一个分布式的、分区的、多副本的消息发布订阅系统，它提供了类似于JMS的特性，但在设计上完全不同，它具有消息持久化、高吞吐、分布式、多客户端支持、实时等特性，适用于离线和在线的消息消费，如常规的消息收集、网站活性跟踪、聚合统计系统运营数据（监控数据）、日志收集等大量数据的互联网服务的数据收集场景。 Spark Spark是一个开源的，并行数据处理框架，能够帮助用户简单的开发快速，统一的大数据应用，对数据进行离线处理，流式处理，交互式分析等等。

检查Kafka数据规划。 11. 选择上报告警实例主机名对应的角色“Broker”。单击图表区域右上角的下拉菜单，选择“定制”，来自定义监控项。 12. 在弹出的“定制”对话框中，选择“磁盘 > Broker磁盘使用率”，并单击“确定”。 关于Kafka磁盘使用情况信息会被显示。 13. 根据步骤12的显示信息，查看是否只有步骤2中上报告警的磁盘分区。 是，执行步骤14。 否，执行步骤15。 14. 重新进行磁盘规划，挂载新的磁盘，进入当前问题节点“实例配置”页面，重新配置“log.dirs”，增加其他磁盘相应路径，重启当前Kafka实例。 15. 查看Kafka配置的数据保存时间配置，根据业务需求和业务量权衡，考虑是否需要调小数据保存时间。 是，执行步骤16。 否，执行步骤17。 16. 在FusionInsight Manager界面，选择“集群 > 待操作集群的名称 > 服务 > Kafka > 配置 > 全部配置”，在右侧搜索框中填写配置项名称“log.retention.hours”，然后会显示该配置的当前值，此处的值为Topic默认的数据保存时间，可以适当调小该值。 说明 对于单独配置数据保存时间的Topic，修改Kafka服务配置页面上配置的数据保存时间不生效。 如果需要对某个Topic单独配置的话，可以使用Kafka客户端命令行，来单独配置该Topic。 例如： kafkatopics.sh zookeeper “ ZooKeeper 地址: 2181/kafka ”alter topic “ Topic 名称 ” config retention.ms “ 保存时间 ” 17. 查看是否由于某些Topic的Partition配置不合理导致部分磁盘使用率达到上限（例如：数据量非常大的Topic的Partition数目小于配置的磁盘个数，导致各磁盘上数据分配无法均匀，进而部分磁盘达到使用率上限）。 说明 对于单独配置数据保存时间的Topic，修改Kafka服务配置页面上配置的数据保存时间不生效。 如果需要对某个Topic单独配置的话，可以使用Kafka客户端命令行，来单独配置该Topic。 例如： kafkatopics.sh zookeeper “ ZooKeeper 地址: 2181/kafka ”alter topic “ Topic 名称 ” config retention.ms “ 保存时间 ” 18. 通过Kafka客户端对Topic的Partition进行扩展，命令行操作命令如下： kafkatopics.sh zookeeper“ ZooKeeper 地址 :2181/kafka ” alter topic “ Topic 名称 ” partitions “ 新Partition 数目 ” 说明 新Partition数目建议配置为Kafka数据磁盘数量的倍数。 当前步骤修改可能不会很快解决当前告警，需要结合步骤11中的数据保存时间逐渐均衡数据。 19. 考虑是否需要扩容。 说明 建议当前Kafka磁盘使用率超过80%时，则需要扩容。 是，执行步骤20。 否，执行步骤21。 20. 扩展磁盘容量，扩展后检查告警是否消失。 是，操作结束。 否，执行步骤22。 21. 检查告警是否清除。 是，操作结束。 否，执行步骤22。

本文主要介绍 连接和查看Kafka Manager。 Kafka Manager是开源的Kafka集群管理工具，需要通过浏览器才能访问Kafka Manager的地址。在Kafka Manager页面，您可以查看Kafka集群的监控、代理等信息。 前提条件 已配置如下表所示安全组。 表安全组规则 方向 协议 端口 源地址 说明 入方向 TCP 9999 0.0.0.0/0 访问Kafka Manager。 登录Kafka Manager 步骤 1 （可选）创建一台与Kafka实例相同VPC和相同安全组的Windows服务器，详细步骤请参考《弹性云主机用户指南》的“创建弹性主机”章节。 如果是已经开启了公网访问，该步骤为可选，在本地浏览器中即可访问，不需要单独的Windows弹性云主机。 步骤 2 在实例详情信息页面，获取Kafka Manager地址。 未开启公网访问时，Kafka Manager地址为“Manager内网访问地址”。 图 Kafka Manager内网访问地址 已开启公网访问时，Kafka Manager地址为“Manager公网访问地址”。 图 Kafka Manager公网访问地址 步骤 3 在浏览器中输入Kafka Manager的地址，进入Kafka Manager登录页面。 如果是开启了公网访问，在本地浏览器输入Kafka Manager地址访问；如果没有开启公网访问，需要登录步骤1的弹性云主机，然后在浏览器输入Kafka Manager地址访问。 步骤 4 输入创建实例时设置的Kafka Manager用户名和密码，即可管理Kafka集群。

本文主要介绍自动续订规则及操作流程。 为避免由于未及时对资源采取续订操作，资源被到期冻结或超期释放，客户购买包月包年产品后，可设置开通自动续订。开通自动续订后，系统将在资源到期前自动续订，无需客户再手动操作。 适用范围 自动续订仅针对采用包月、包年计费模式的资源。 已到期资源不支持设置/修改自动续订。 自动续订仅适于 付费用户 ，不适用于非付费用户。 目前支持设置自动续订的产品有：弹性云主机、GPU云主机、物理机、云桌面、云硬盘、对象存储经典版、云主机备份、专属云计算独享型、专属云存储独享型、弹性IP、共享带宽、天翼云SDWAN、云间高速、关系数据库MySQL版、关系数据库Postgre SQL版、分布式关系数据库、分布式缓存服务Redis版、分布式缓存服务Memcache、文档数据库服务、云HBASE数据库、分布式消息服务RocketMQ、分布式消息服务RabbitMQ、分布式消息服务Kafka、Web应用防火墙、服务器安全卫士、域名无忧、DDos高防IP、云解析、登录保护、网站安全监测、内容安全。 单次最多支持20个资源实例批量续订。 开通、变更、关闭自动续订 用户在续订管理页可开通自动续订功能，变更自动续约周期，或关闭自动续订。 不关闭自动续订的情况下，只要预付费账户余额充足，或为后付费客户，系统将持续按设定的周期自动续订下去。 预付费用户可在官网自主控制自动续订功能的开通、变更、关闭。后付费用户需要客户经理协助开启自动续订权限后才可以自主管理。

本文主要介绍了自动续订规则及操作流程。 为避免由于未及时对资源采取续订操作，资源被到期冻结或超期释放，客户购买包月包年产品后，可设置开通自动续订。开通自动续订后，系统将在资源到期前自动续订，无需客户再手动操作。 适用范围 自动续订仅针对采用包月、包年计费模式的资源。 已到期资源不支持设置/修改自动续订。 自动续订仅适于 付费用户 ，不适用于非付费用户。 目前支持设置自动续订的产品有：弹性云主机、GPU云主机、物理机、云桌面、云硬盘、对象存储经典版、云主机备份、专属云计算独享型、专属云存储独享型、弹性IP、共享带宽、天翼云SDWAN、云间高速、关系数据库MySQL版、关系数据库Postgre SQL版、分布式关系数据库、分布式缓存服务Redis版、分布式缓存服务Memcache、文档数据库服务、云HBASE数据库、分布式消息服务RocketMQ、分布式消息服务RabbitMQ、分布式消息服务Kafka、Web应用防火墙、服务器安全卫士、域名无忧、DDos高防IP、云解析、登录保护、网站安全监测、内容安全。 单次最多支持20个资源实例批量续订。 开通、变更、关闭自动续订 用户在续订管理页可开通自动续订功能，变更自动续约周期，或关闭自动续订。 不关闭自动续订的情况下，只要预付费账户余额充足，或为后付费客户，系统将持续按设定的周期自动续订下去。 预付费用户可在官网自主控制自动续订功能的开通、变更、关闭。后付费用户需要客户经理协助开启自动续订权限后才可以自主管理。

开发环境 Maven Apache Maven 2.5及以上版本，可至Maven官方网站下载。 JDK Java Development Kit 1.8及以上版本，可至Oracle官方网站下载。 安装后注意配置JAVA的环境变量。 IntelliJ IDEA 获取并安装IntelliJ IDEA，可至IntelliJ IDEA官方网站下载。 操作步骤 1. 下载Demo包kafkajavademo.zip 下载后解压，有如下文件： 表1 KAFKA Demo文件清单 文件名 路径 说明 ::: JavaKafkaConfigurer.java .srcmainjavajavaDemo 读取Kafka配置文件。 KafkaConsumerDemo.java .srcmainjavajavaDemo 消费消息。 KafkaMultiConsumerDemo.java .srcmainjavajavaDemo 批量消费消息。 KafkaProducerDemo.java .srcmainjavajavaDemo 生产消息。 kafka.properties .srcmainresources kafka配置参数 pom.xml . maven配置文件，包含Kafka客户端引用。 2. 打开IntelliJ IDEA，导入Demo。Demo是一个Maven构建的Java工程，因此需要配置JDK环境，以及IDEA的Maven插件。 3. 修改Kafka配置信息。 修改kafka.properties 修改demo文件中的属性变量

本章节主要介绍DataArts Studio的获取Rest Client算子返回值操作。 Rest Client算子可以执行RESTful请求。 本教程主要介绍如何获取Rest Client的返回值，包含以下两个使用场景举例。 通过“响应消息体解析为传递参数定义”获取返回值。 通过EL表达式获取返回值。 通过“响应消息体解析为传递参数定义”获取返回值 如下图所示，第一个Rest Client调用了MRS服务查询集群列表的API，下图为API返回值的JSON消息体。 使用场景：需要获取集群列表中第一个集群的cluster Id，然后作为参数传递给后面的节点使用。 关键配置：在第一个Rest Client的“响应消息体解析为传递参数定义”配置中，配置clusterIdclusters[0].clusterId，后续的Rest Client节点就可以用${clusterId}的方式引用到集群列表中的第一个集群的cluster Id。 Rest Clien作业样例1 JSON消息体 通过EL表达式获取返回值 Rest Client算子可与EL表达式相配合，根据具体的场景选择不同的EL表达式来实现更丰富的用法。您可以参考本教程，根据您的实际业务需要，开发您自己的作业。EL表达式用法可参考表达式概述。 如下图所示，Rest Client调用了MRS服务查询集群列表的API，然后执行Kafka Client发送消息。 使用场景：Kafka Client发送字符串消息，消息内容为集群列表中第一个集群的cluster Id。 关键配置：在Kafka Client中使用如下EL表达式获取Rest API返回消息体中的特定字段：

本章主要介绍翼MapReduce的配置Kafka数据传输加密功能。 操作场景 Kafka客户端和Broker之间的数据传输默认采用明文传输，客户端可能部署在不受信任的网络中，传输的数据可能遭到泄漏和篡改。 操作步骤 默认情况下，组件间的通道是不加密的。用户可以配置如下参数，设置安全通道为加密的。 参数修改入口：在FusionInsight Manager系统中，选择“集群 > 待操作集群的名称 > 服务 > Kafka > 配置”，展开“全部配置”页签。在搜索框中输入参数名称。 说明 配置后应重启对应服务使参数生效。 Kafka服务端的传输加密相关配置参数详见下表。 配置项 描述 默认值 ssl.mode.enable 是否开启SSL对应服务。如果设置为“true”，那么Broker启动过程中会启动SSL的相关服务。 false security.inter.broker.protocol Broker间通信协议。支持PLAINTEXT、SSL、SASLPLAINTEXT、SASLSSL这四种协议类型。 SASLPLAINTEXT “ssl.mode.enable”配置为“true”后，Broker会开启SSL、SASLSSL两种协议的服务，然后服务端或者客户端才能配置相关的SSL协议，进行传输加密通信。

本节主要介绍分布式消息服务Kafka的删除Topic步骤 场景描述 在以下场景中，可以考虑删除Kafka主题： 主题不再使用：当一个主题不再被使用或者不再需要时，可以选择删除该主题。这可能是因为业务需求变化、数据不再有效或者主题被合并到其他主题中等原因。 数据保留策略变更：Kafka中可以设置数据保留策略，即数据在主题中的保留时间或者大小。当需要更改数据保留策略时，可能需要删除旧的主题并创建一个新的主题来应用新的策略。 清理测试数据：在测试环境中，经常需要清理旧的测试数据，以确保环境的可用空间和性能。当测试数据不再需要时，可以删除相应的主题来释放资源。 主题配置错误：在创建主题时，可能会出现配置错误或者误操作导致主题创建失败或者无法正常使用。在这种情况下，可以删除有问题的主题，并重新创建正确的主题配置。 需要注意的是，在删除主题之前，需要确保主题中的数据已经备份或者不再需要。删除主题将永久删除主题中的所有数据，并且无法恢复。因此，在删除主题之前，建议先进行备份或者确认数据不再需要。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“Topic管理”后进入Topic管理页面。 （5）在Topic所在行，点击其右侧的“更多按钮”，再单击“删除”，并选择确定。

Kafka实例完全兼容开源Kafka协议，可以直接使用kafka开源客户端连接。如果使用SASL认证方式，则在开源客户端基础上使用云服务提供的证书文件。 本指南主要介绍实例连接信息的收集，如获取实例连接地址、SASL连接使用的证书、公网访问信息等，然后提供Java、Python和Go等语言的连接示例。 本指南的示例仅展示Kafka的API调用，生产与消费的API集，请参考Kafka官网。

本节介绍分布式消息服务Kafka计费类常见问题 可以购买哪些版本？ 不同资源池可购买的版本、规格不一致，具体请查看产品规格说明 （1）华东1、华北2、西南1、华南2、上海36、青岛20、长沙42、南昌5、武汉41、杭州7、西南2贵州、太原4、郑州5、西安7、呼和浩特3 可以选购Kafka引擎，选择主机类型、节点规格、节点数和存储空间。 （2）芜湖2、上海7、重庆2、乌鲁木齐27、石家庄20、内蒙6、北京5 可以选购高级版和基础版两个版本。 到期后如何续费？ 在集群列表中点击“续费”，进入购买时长页面，购买成功后自动续费。 手动续订：对于包年/包月订购的分布式缓存服务，用户在资源到期前进行续费操作，可以延长原有资源到期时间，避免资源到期后冻结或超过保留期后被系统回收。详细操作请参考费用中心续订管理手动续订。 自动续订：自动续订仅针对采用包月、包年计费模式的资源，详细操作请参考费用中心续订管理自动续订。 产品订购时可选资源池节点不一致？ 已上线资源池节点的剩余容量达到一定比例后，为确保老客户权益，将不再面向新客户开放，产品订购时的可选资源节点范围以实际为准。

本节介绍如何使用 SASLSSL 接入点接入Kafka,文档以Java代码为例。 前提条件 已配置正确的安全组。 已获取连接Kafka实例的地址。 如果Kafka实例未开启自动创建Topic功能，在连接实例前，请先创建Topic。 已创建弹性云服务器，如果使用内网同一个VPC访问实例，请设置弹性云服务器的VPC、子网、安全组与Kafka实例的VPC、子网、安全组一致。 需要用户先在用户管理页面创建用户，然后给对应的topic授予生产消费权限。 接口地址及证书下载，需使用内网同一个VPC访问，实例端口为8098。具体信息从控制台实例详情菜单处获取，如下图所示，点击中间下载箭头即可下载证书。 Maven中引入Kafka客户端 java Kafka实例基于社区版本2.8.2/3.6.2，推荐客户端保持一致。 org.apache.kafka kafkaclients 2.8.2/3.6.2 客户端关键参数 java Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(CommonClientConfigs.SECURITYPROTOCOLCONFIG, "SASLSSL"); props.put("sasl.mechanism", "SCRAMSHA512"); props.put(SaslConfigs.SASLJAASCONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username"testuser" password"Kafka@Test";"); props.put("ssl.truststore.location","/kafka/client.truststore.jks"); props.put("ssl.truststore.password","sJses2tin1@23"); props.put("ssl.endpoint.identification.algorithm","");

本节介绍如何使用的安全接入点接入Kafka的方法，文档以Java代码为例。 前提条件 已配置正确的安全组。 已获取连接Kafka实例的地址。 如果Kafka实例未开启自动创建Topic功能，在连接实例前，请先创建Topic。 已创建弹性云服务器，如果使用内网同一个VPC访问实例，请设置弹性云服务器的VPC、子网、安全组与Kafka实例的VPC、子网、安全组一致。 需要用户先在用户管理页面创建用户，然后给对应的topic授予生产消费权限。 使用内网同一个VPC访问，实例端口为8092，实例连接地址从控制台实例详情菜单处获取，如下图所示。 Maven中引入Kafka客户端 java Kafka实例基于社区版本2.8.2/3.6.2，推荐客户端保持一致。 org.apache.kafka kafkaclients 2.8.2/3.6.2 客户端关键参数 java Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(CommonClientConfigs.SECURITYPROTOCOLCONFIG, "SASLPLAINTEXT"); props.put("sasl.mechanism", "SCRAMSHA512"); props.put(SaslConfigs.SASLJAASCONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username"testuser" password"Kafka@Test";"); 生产者代码示例 java package com.justin.kafka.service.gw.sasl; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; public class Producer { private final KafkaProducer producer; public final static String TOPIC "testtopic2"; public final static String BROKERADDR "192.168.0.11:8092,192.168.0.9:8092,192.168.0.10:8092"; public Producer() { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(CommonClientConfigs.SECURITYPROTOCOLCONFIG, "SASLPLAINTEXT"); props.put("sasl.mechanism", "SCRAMSHA512"); props.put(SaslConfigs.SASLJAASCONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username"testuser" password"Kafka@Test";"); props.put(ProducerConfig.ACKSCONFIG, "all"); props.put("retries",3); producer new KafkaProducer<>(props); } public void produce() { try { for (int i 0; i (TOPIC, data), new Callback() { public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception ! null) { // TODO: 异常处理 exception.printStackTrace(); return; } System.out.println("produce msg completed, partition id " + metadata.partition()); } }); } } catch (Exception e) { // TODO: 异常处理 e.printStackTrace(); } producer.flush(); producer.close(); } public static void main(String[] args) { Producer producer new Producer(); producer.produce(); } }

本章节主要介绍配置DMS Kafka连接。 连接DMS的Kafka队列时，相关参数详见下表：DMS Kafka连接参数 参数名 说明 取值样例 名称 连接的名称，根据连接的数据源类型，用户可自定义便于记忆、区分的连接名。 dmslink 服务类型 选择DMS Kafka版本，目前只有专享版。 专享版 Kafka Broker Kafka专享版实例的地址，格式为host:port。 Kafka SASLSSL 选择是否打开客户端连接Kafka专享版实例时SSL认证的开关。 开启Kafka SASLSSL，则数据加密传输，安全性更高，但性能会下降。 是 用户名 开启Kafka SASLSSL时显示该参数，表示连接DMS Kafka的用户名。 密码 开启Kafka SASLSSL时显示该参数，表示连接DMS Kafka的密码。

事件模式是事件总线EventBridge用来过滤相关事件的模式定义。事件总线EventBridge通过事件模式过滤事件并将事件路由到事件目标，事件模式必须和匹配的事件具有相同的结构。本文介绍事件模式的常用类型。 注意事项 事件模式匹配的注意点如下： 事件必须包含事件模式中列出的所有字段名对应的过滤项。 事件模式是逐个字符精确匹配的 ，需注意大小写，匹配过程中不会对字符串进行任何标准化的操作。 事件模式匹配中，JSON对象的各个属性支持AND 语义，数组中的各元素支持OR语义。 事件模式当前匹配仅支持根据事件的属性进行过滤，暂不支持data数据过滤。 创建事件模式时，某一字段下的匹配字段需符合一定的规则，详见下文。 指定值匹配 您可以指定某个字段的值进行匹配。 例如，以下示例事件模式只匹配source是分布式消息服务Kafka的事件。下表介绍事件从事件源发出经过事件模式过滤的过程。 从事件源接收的事件 事件模式 模式过滤后的事件 plaintext { "id": "f36b49b668914b59b11cbf689xxxx712", "source": "ctyun:kafka", "type": "ctyun:kafka:Topic:Message", "specversion": "1.0", "datacontentype": "application/json; charsetutf8", "subject": "ctyun:kafka:75dcxx1eb5:topic:source1", "time": "20241203T09:48:00.696Z", "data": { "key": "test2", "value": "test1" } } plaintext { "source": [ { "exact" : "ctyun:kafka" } ] } plaintext { "id": "f36b49b668914b59b11cbf689xxxx712", "source": "ctyun:kafka", "type": "ctyun:kafka:Topic:Message", "specversion": "1.0", "datacontentype": "application/json; charsetutf8", "subject": "ctyun:kafka:75dcxx1eb5:topic:source1", "time": "20241203T09:48:00.696Z", "data": { "key": "test2", "value": "test1" } }

本节介绍如何使用的公共接入点接入Kafka的方法，文档以Java代码为例。 前提条件 已配置正确的安全组。 已获取连接Kafka实例的地址。 如果Kafka实例未开启自动创建Topic功能，在连接实例前，请先创建Topic。 已创建弹性云服务器，如果使用内网同一个VPC访问实例，请设置弹性云服务器的VPC、子网、安全组与Kafka实例的VPC、子网、安全组一致。 使用内网同一个VPC访问，实例端口为8090，实例连接地址从控制台实例详情菜单处获取，如下图所示。 Maven中引入Kafka客户端 java Kafka实例基于社区版本2.8.2/3.6.2，推荐客户端保持一致。 org.apache.kafka kafkaclients 2.8.2/3.6.2 客户端关键参数 java Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); 生产者代码示例 java package com.justin.kafka.service.gw.plain; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; public class Producer { private final KafkaProducer producer; public final static String TOPIC "testtopic"; public final static String BROKERADDR "192.168.0.11:8090,192.168.0.9:8090,192.168.0.10:8090"; public Producer() { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAPSERVERSCONFIG, BROKERADDR); props.put(ProducerConfig.VALUESERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.KEYSERIALIZERCLASSCONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.ACKSCONFIG, "all"); props.put("retries",3); producer new KafkaProducer<>(props); } public void produce() { try { for (int i 0; i (TOPIC, data), new Callback() { public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception ! null) { // TODO: 异常处理 exception.printStackTrace(); return; } System.out.println("produce msg completed, partition id " + metadata.partition()); } }); } } catch (Exception e) { // TODO: 异常处理 e.printStackTrace(); } producer.flush(); producer.close(); } public static void main(String[] args) { Producer producer new Producer(); producer.produce(); } }

分布式消息服务RocketMQ是一款低成本、高可靠、高性能的消息中间件产品,兼容开源RocketMQ客户端,提供高效可靠的消息传递服务,解决分布式应用系统之间的消息数据通信难题,用于系统间的解耦,用户只需专注业务,无需部署运维,适用于电商、金融、政企等多样业务场景。

分布式消息服务RocketMQ是一款低成本、高可靠、高性能的消息中间件产品,兼容开源RocketMQ客户端,提供高效可靠的消息传递服务,解决分布式应用系统之间的消息数据通信难题,用于系统间的解耦,用户只需专注业务,无需部署运维,适用于电商、金融、政企等多样业务场景。

软件配置 MRS集群软件配置 参数 参数说明 区域 选择区域。 不同区域的云服务产品之间内网互不相通。请就近选择靠近您业务的区域，可减少网络时延，提高访问速度。 集群名称 集群名称不允许重复。 只能由字母、数字、中划线和下划线组成，并且长度为1～64个字符。 默认名称为mrsxxxx，xxxx为字母和数字的四位随机组合数，系统自动组合。 集群版本 当前版本为MRS 3.1.0。 集群类型 提供几种集群类型： 分析集群：用来做离线数据分析，提供的是Hadoop体系的组件。 流式集群：用来做流处理任务，提供的是流式处理组件。 混合集群：既可以用来做离线数据分析，也可以用来做流处理任务，提供的是Hadoop体系的组件和流式处理组件。建议同时需要做离线数据分析和流处理任务时使用混合集群。 自定义：用户可按照业务需求调整集群服务的部署方式，具体请参见创建自定义拓扑集群。（目前仅MRS 3.x版本支持） 说明 MRS流式集群不支持“作业管理”和“文件管理”功能。如需在集群中安装全部组件，请选择“自定义”类型集群。 组件选择 MRS配套的组件如下： 分析集群组件 Presto：开源、分布式SQL查询引擎。 Hadoop：分布式系统基础架构。 Spark：内存分布式系统框架。（MRS 3.x版本不支持） Spark2x：Spark2x是一个对大规模数据处理的快速和通用引擎,基于开源Spark2.x版本开发。（仅MRS 3.x版本支持） Hive：建立在Hadoop上的数据仓库框架。 HBase：分布式列数据库。 Tez：提供有向无环图的分布式计算框架。 Hue：提供Hadoop UI能力，让用户通过浏览器分析处理Hadoop集群数据。 Loader：基于开源sqoop 1.99.7开发，专为Apache Hadoop和结构化数据库（如关系型数据库）设计的高效传输大量数据的工具。（MRS 3.x版本不支持） Hadoop为必选组件，且Spark与Hive组件需要配套使用。请根据业务选择搭配组件。 Flink：分布式大数据处理引擎，可对有限数据流和无限数据流进行有状态计算。 Oozie：Hadoop作业调度系统。（仅MRS 3.x版本支持） HetuEngine：HetuEngine是一个同异构大数据集的分布式SQL查询引擎。（仅MRS 3.1.x LTS版本支持） Ranger：一个基于Hadoop平台监控和管理数据安全的框架。 Impala：一种处理大量数据的SQL查询引擎。 ClickHouse：ClickHouse是一个用于联机分析(OLAP)的列式数据库管理系统(DBMS)。CPU架构为鲲鹏计算的ClickHouse集群表引擎不支持使用HDFS和Kafka。 Kudu：一种列存储管理器。 流式集群组件 Kafka：提供分布式消息订阅的系统。 Flume：提供分布式、高可用、高可靠的海量日志采集、聚合和传输系统。 ZooKeeper：一个集中的服务，它用于维护配置信息、命名、提供分布式的同步和提供分组服务。（仅MRS 3.x版本支持） Ranger：一个基于Hadoop平台监控和管理数据安全的框架。（仅MRS 3.x版本支持） 元数据 是否使用外部数据源存储元数据。 本地元数据： 元数据存储在集群本地。 数据连接：使用外部数据源元数据，若集群异常或删除时将不影响元数据，适用于存储计算分离的场景。 支持Hive或Ranger组件的集群支持该功能。 组件名 当“元数据”选择“数据连接”时该参数有效。用于表示可以设置外部数据源的组件类型。 Hive Ranger 数据连接类型 当“元数据”选择“数据连接”时该参数有效。用于表示外部数据源的类型。 Hive组件支持的数据连接类型： − RDS服务MySQL数据库− 本地数据库 Ranger组件支持的数据连接类型： − RDS服务MySQL数据库− 本地数据库 数据连接实例 当“数据连接类型”选择“RDS服务MySQL数据库”时，该参数有效。用于表示MRS集群与RDS服务数据库连接的名称，该实例必选先创建才能在此处引用。可单击“创建数据连接”进行创建，具体请参考配置数据连接。

本文带您了解什么是分布式消息服务RocketMQ产品。 分布式消息服务RocketMQ是一款低成本、高可靠、高性能的消息中间件产品，兼容开源RocketMQ客户端，提供高效可靠的消息传递服务，解决分布式应用系统之间的消息数据通信难题，用于系统间的解耦，用户只需专注业务，无需部署运维，适用于电商、金融、政企等多样业务场景。 产品示意图 分布式消息服务RocketMQ发布订阅基本流程如下： 1、Producer连接Nameserver，产生数据放入不同的Topic； 2、对于RocketMQ，一个Topic可以分布在各个Broker上，我们可以把一个Topic分布在一个Broker上的子集定义为一个Topic分片； 3、将Topic分片再切分为若干等分，其中的一份就是一个Queue。每个Topic分片等分的Queue的数量可以不同，由用户在创建Topic时指定； 4、Consumer连接Nameserver，根据Broker分配的Queue来消费数据。 核心组件 分布式消息服务RocketMQ主要由Producer、Broker、Consumer三部分组成，其中Producer负责生产消息，Consumer负责消费消息，Broker负责存储消息。Broker在实际部署过程中对应1台或者多台服务器，每个Broker可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的Broker。Message Queue用于存储消息的物理地址，每个Topic中的消息地址存储于多个Message Queue中。ConsumerGroup由多个Consumer实例构成。更多信息请参见产品架构。

java8 步骤 4 下载开源的Kafka客户端。 对应1.1.0版本实例的下载地址： wget 对应2.3.0版本实例的下载地址： wget 对应2.7版本实例的下载地址： wget 步骤 5 解压Kafka客户端文件。 tar zxf ${kafkatar} “ kafkatar ” 表示客户端的压缩包名称。例如： tar zxf kafka2.122.7.2.tgz 后续步骤 步骤二：创建Kafka实例

支持普通集成的天翼云服务列表 天翼云工作流支持通过OpenAPI集成多个服务，涵盖以下产品分类： 产品分类 产品名称 存储 混合存储网关、云硬盘、弹性文件服务、媒体存储、云备份、云硬盘备份、对象存储、云主机备份、海量文件服务 OceanFS、云容灾、对象存储（经典版）I型、并行文件 网络与CDN 弹性IP、天翼云SDWAN、NAT网关、VPN连接、云间高速、虚拟私有云、弹性负载均衡、共享流量包、智能边缘云、云专线CDA、应用加速、边缘安全加速平台、CDN加速、智能DNS、VPC终端节点、全站加速、科研助手 计算 弹性云主机、天翼云电脑（政企版）、函数计算、弹性高性能计算、镜像服务、弹性伸缩服务、物理机、云骁智算平台、数据加密 数据库 关系数据库SQL Server版、关系数据库PostgreSQL版、分布式关系型数据库、关系数据库MySQL版、数据传输服务、文档数据库服务、数据管理服务、分布式缓存服务Redis版、时序数据库Influx版、分布式融合数据库HTAP、云数据库ClickHouse 监控与管理 云日志服务、应用性能监控 应用服务 API网关、EasyCoding敏捷开发平台、软件开发生产线CodeArts 数据库 关系型数据库MySQL版（CTRDS）、关系型数据库PostgreSQL版、关系型数据库SQL Server版、云数据库GaussDB、分布式关系型数据库、分布式融合数据库HTAP、文档数据库服务、时序数据库Influx版、分布式缓存服务Redis版 人工智能 AI能力开放平台、慧聚一站式智算服务平台 大数据 云搜索服务、大数据管理平台 DataWings、翼MapReduce 容器与中间件 应用服务网格、弹性容器实例、分布式容器云平台、容器镜像服务、分布式消息服务RabbitMQ、分布式消息服务RocketMQ、微服务应用平台MSAP、云容器引擎、分布式消息服务Kafka、微服务引擎注册配置中心、微服务引擎MSE、应用性能监控apm、分布式容器云平台CCSE ONE 安全及管理 云防火墙（原生版）、统一身份认证、密钥管理、服务器安全卫士（原生版）、云监控、Web应用防火墙（原生版）、证书管理服务、DDoS高防（边缘云版）、云堡垒机、Web应用防火墙（边缘云版）、云审计 企业应用 云通信短信 视频 视频直播、智能视图服务、云点播 专属云 专属云（计算独享型） 价格 账务 其他 数据库审计、云日志服务、轻量型云主机、云原生网关、API网关 说明 ：具体支持的服务和API请参考天翼云OpenAPI文档。

场景描述 在某些情况下，您可能需要将Kafka开启公网访问。以下是一些可能的场景描述： 跨地域数据传输：如果您有多个位于不同地理位置的数据中心或云服务提供商，您可能希望在它们之间传输数据。通过将Kafka开启公网访问，您可以轻松地在不同地区之间进行数据传输。 跨组织数据共享：如果您需要与其他组织或合作伙伴共享数据，而且这些组织不在同一个网络环境中，您可以将Kafka开启公网访问，以便安全地共享数据。 远程访问和监控：有时候，您可能需要从外部网络访问和监控Kafka集群。通过配置Kafka开启公网访问，您可以远程连接到Kafka集群，监控其运行状态，并执行必要的管理操作。 请注意，在设置Kafka开启公网访问时，确保采取适当的安全措施，比如使用安全组或者用户权限来限制访问权限，以保护Kafka集群的安全性。 约束与限制 开启公网访问功能只支持绑定IPv4弹性IP地址，不支持绑定IPv6弹性IP地址。 操作步骤 （1）登录管理控制台。 （2）进入Kafka管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“实例详情”后找到如下接入点信息模块。 （5）点击右上角“添加路由策略”按钮，选择“公网接入”方式。 （6）需要为每个节点设置对应的弹性IP地址，点击其右侧的“绑定”按钮，在弹窗中选择某个弹性IP绑定。如果没有足够数量的弹性IP地址，可参考申请弹性IP购买弹性IP地址。 （7）点击刷新按钮查看弹性IP绑定状态，待所有节点都绑定弹性IP后点击“确定”按钮下发开启公网后台任务。 （8）当实例状态重新变为“运行中”后，表示开启公网访问成功。 （9）开启公网访问后，需要设置好对应的安全组规则 才能成功连接Kafka，设置规则参考如下: 安全组参考规则（先判断安全组的出/入方向是否放通了对弹性ip地址的访问，如果没有，可以参考如下规则配置）： 方向 协议 类型 端口 远端(源地址) 说明 入方向 TCP IPv4 Kafka公网接入端口(8094)，或更大范围 Kafka客户端所在的IP地址或地址组 通过公网访问Kafka （10）如果创建了 网络ACL规则 ，则ACL规则也需要正确配置才能成功连接Kafka，设置规则参考如下: 网络ACL参考规则（先判断网络ACL的出/入方向是否放通了客户端地址的访问，如果没有，可以参考如下规则配置）： 方向 协议 IP版本 策略 源地址 源端口范围 目的地址 目的端口范围 描述 入方向 TCP IPv4 允许 Kafka客户端所在的IP地址或地址组 165535 VPC IP地址组 Kafka公网接入端口(8094)，或更大范围 控制入方向ACL规则 出方向 TCP IPv4 允许 VPC IP地址组 Kafka公网接入端口(8094)，或更大范围 Kafka客户端所在的IP地址或地址组 165535 控制出方向ACL规则 （11）连接Kafka可以根据开启公网访问时选择的SASLPLAINTEXT或SASLSSL协议接入。

本章节主要介绍翼MapReduce集群组件使用规则。 Kafka支持四种协议类型的访问，分别为：PLAINTEXT、SSL、SASLPLAINTEXT、SASLSSL。当前，翼MR集群默认采用Kerberos安全验证服务，Kafka协议类型建议使用SASLPLAINTEXT、SASLSSL这两种。