生产者代码示例 java package com.justin.kafka.service.gw.saslssl; 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 "testtopic3"; public final static String BROKERADDR "192.168.0.11:8098,192.168.0.9:8098,192.168.0.10:8098"; 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); 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",""); 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(); } }

天翼云计划于2025年12月10日 00:00（北京时间）下线二类节点分布式消息服务RabbitMQ的部分插件，具体下线插件如下： rabbitmqamqp10 rabbitmqmqtt rabbitmqsharding rabbitmqstomp rabbitmqtracing rabbitmqwebmqtt rabbitmqwebstomp 下线范围 下线范围为天翼云二类节点：北京2、成都3、福州、广州4、贵州、杭州、昆明、南昌、内蒙3、青岛、上海4、深圳、石家庄、苏州、太原、芜湖、武汉2、西安2、华北、长沙2、兰州、重庆、郑州、乌鲁木齐。 下线原因 开启上述插件后，容易引起业务过载风险、影响业务系统稳定。 rabbitmqwebstomp插件通过临时队列提供WebSocket能力，同时其连接和释放依赖于心跳机制，过多的临时队列以及过多连接数会导致内存高水位，从而导致节点状态异常。 下线影响 未开启待下线插件的实例，插件列表中将不再显示这些插件。 已开启待下线插件的实例，仍然可以使用。如果您在控制台关闭这些插件后，也将不再显示。 如有任何问题可通过天翼云官网(www.ctyun.cn)提交工单或服务热线（4008109889）与我们联系。 消减措施 已开启插件的RabbitMQ实例，建议参考设置RabbitMQ告警规则，配置相应的告警，提前预知并消减风险。

分布式消息服务RocketMQ版的事务消息支持在业务逻辑与发送消息之间提供事务保证，通过两阶段的方式提供对事务消息的支持，事务消息交互流程如图1所示。 图1 事务消息交互流程 事务消息生产者首先发送半消息，然后执行本地事务。如果执行成功，则发送事务提交，否则发送事务回滚。服务端在一段时间后如果一直收不到提交或回滚，则发起回查，生产者在收到回查后重新发送事务提交或回滚。消息只有在提交之后才投递给消费者，消费者对回滚的消息不可见。 收发事务消息前，请参考收集连接信息收集RocketMQ所需的连接信息。 准备环境 1. 在命令行输入python，检查是否已安装Python。得到如下回显，说明Python已安装。 PS C:> python Python 3.9.9 (tags/v3.9.9:ccb0e6a, Nov 15 2021, 18:08:50) [MSC v.1929 64 bit (AMD64)] on win32 Type "help", "copyright", "credits" or "license" for more information. 如果未安装Python，请使用以下命令安装： pip install rocketmqclientpython 2. 安装librocketmq库和rocketmqclientpython。 说明 建议下载rocketmqclientcpp2.2.0，获取librocketmq库。 3. 将librocketmq.so添加到系统动态库搜索路径。 1. 查找librocketmq.so的路径。 find / name librocketmq.so 2. 将librocketmq.so添加到系统动态库搜索路径。 ln s /查找到的librocketmq.so路径/librocketmq.so /usr/lib sudo ldconfig 以下示例代码中的参数说明如下，请参考收集连接信息获取参数值。 GROUP：表示消费组名称。 ENDPOINT：表示实例连接地址和端口。 TOPIC：表示Topic名称。

场景描述 RocketMQ的Topic是消息的逻辑分类单位，用于将消息进行分组和管理。创建Topic的场景可以根据具体业务需求来确定，以下是一些常见的场景描述： 消息发布与订阅：当需要实现消息发布与订阅模式时，可以创建一个主题来管理相关的消息。发布者可以将消息发送到该主题，而订阅者可以订阅该主题以接收感兴趣的消息。 事件驱动架构：在事件驱动的架构中，不同的模块之间通过事件进行通信。每个事件可以对应一个主题，模块可以将事件发送到相应的主题，其他模块可以订阅该主题以接收事件通知。 日志收集与分析：当需要收集大量的日志数据并进行分析时，可以创建一个主题来管理日志消息。日志产生者可以将日志消息发送到该主题，而日志消费者可以订阅该主题以进行实时分析或存储。 异步处理：当需要将某些操作异步处理时，可以创建一个主题来管理相关的异步消息。操作发起者可以将异步消息发送到该主题，而异步处理器可以订阅该主题以进行后续的异步处理。 分布式事务消息：在分布式系统中，当需要实现分布式事务消息时，可以创建一个主题来管理相关的事务消息。事务发起者可以将事务消息发送到该主题，而事务消费者可以订阅该主题以进行事务的处理和确认。 总的来说，创建主题的场景可以根据具体的业务需求来确定。主题可以帮助将消息进行逻辑分类和管理，实现不同的消息传递模式和业务场景 新建Topic 1、 天翼云官网点击控制中心，选择产品分布式消息服务RocketMQ。 2、 登录分布式消息服务RocketMQ控制台，点击右上角地域选择对应资源池。 3、 进入实例列表，点击【管理】按钮进入管理菜单。 4、 进入Topic管理菜单，点击【创建Topic】按钮 5、 在弹出的创建Topic页面，填写如下字段信息 1）默认展示当前集群名称，不可修改。 2）选择Topic所在的Broker，按照实例创建时候选择的主备节点对数列出每个broker，可复选。 3）填写Topic名称，名字限制2到64个字符，超过限制会导致创建Topic失败，用户创建Topic只能包含大小写字母数字以及和符号。 4）按照实际需求填写Topic备注。 5）填写每个Broker分区数，分区数必须大于0，小于等于8，创建严格顺序队列时，设置分区数为1，且只能选择一个broker。 6）选择生产模式，RocketMQ是一个开源的分布式消息中间件，它支持两种消息生产模式：有序和无序。 有序消息生产模式（Ordered Message）是指按照特定规则将消息发送到相同的Message Queue中，并且确保消息在消费者端按照相同的顺序进行消费。这种模式适用于那些需要严格按照消息顺序进行处理的场景，比如订单处理、流程审批等。 无序消息生产模式（Unordered Message）是指消息发送到不同的Message Queue中，每个Queue都是独立的。消费者可以并行地从多个Queue中消费消息，而无需关心消息的顺序。这种模式适用于那些不需要严格按照消息顺序处理的场景，比如日志收集、异步通知等。 需要注意的是，无论是有序还是无序消息生产模式，RocketMQ都提供了高可靠性的消息传输和存储，并支持水平扩展和高吞吐量的特性。根据具体的业务需求，选择适合的消息生产模式能够更好地满足应用的要求。 7）选择Topic的读写权限，支持读写、只读、只写3类权限。 6、 完成Topic信息填写后，保存确认即可新增主题。 7、 若希望批量创建Topic，可点击【导入Topic】按钮 导入Topic 注意：输入的Topic名不要带空格等特殊字符。 通过上传csv文件,批量创建主题。格式：点击【下载模板】按钮下载。 Topic模板 批量上传Topic的模板，必须使用模板，才能够上传成功，模板格式如下： TopicName QueueNums Perm MessageType Topic1 4 6 NORMAL Topic2 1 6 FIFO Topic3 4 6 DELAY Topic4 4 6 TRANSACTION

本文介绍如何在事件总线EventBridge管理控制台中添加分布式消息服务RocketMQ类型的自定义事件源。 前提条件 事件总线EventBridge 开通事件总线EventBridge并委托授权。 创建自定义总线。 分布式消息服务RocketMQ 开通分布式消息服务RocketMQ。 创建实例，并创建对应主题与消费组。 操作步骤 1. 登录事件总线管理控制台。 2. 在左侧导航栏，单击事件总线。 3. 在事件总线页面，单击目标总线名称。 4. 在左侧导航栏，单击事件源。 5. 在事件源 页面，单击添加事件源。 6. 在添加自定义事件源 面板，输入名称 和描述 ，事件提供方 选择分布式消息服务RocketMQ，选择实例、主题等配置然后单击确认，如图1所示。 图1 创建事件源时事件提供方选择分布式消息服务RocketMQ 参数说明 参数 说明 示例 名称 事件源名。 source1 实例名称 选择前提条件中已创建的分布式消息服务RocketMQ版实例。 xxx Topic 选择当前实例中的Topic。 topic1 Group 快速创建：自动创建以GIDEVENTBRIDGExxx命名的Group ID。 使用已有：选择当前实例中已创建的Group，请不要与已有业务的Group混用，以免影响已有的消息收发。 group1 消费位点 开始消费的位置。仅支持从最早位点进行消费。 最早位点：从最初位点开始消费。 最早位点

本章节主要介绍如何创建KafkaSSL类型跨源认证。 操作场景 通过在DLI控制台创建的KafkaSSL类型的跨源认证，将Kafka的认证信息存储到DLI，无需在SQL作业中配置帐号密码，安全访问Kafka实例。 说明 MRS Kafka开启Kerberos认证，未开启SSL认证时，创建Kerberos类型的认证。建表时通过krbauthname关联跨源认证。 MRS Kafka开启Kerberos认证，同时开启了SSL认证时，需要同时创建Kerberos和KafkaSSL类型的认证。建表时分别通过krbauthname和sslauthname关联跨源认证。 MRS Kafka未开启Kerberos认证，仅开启了SASL认证时（例如使用帐号密码认证PlainLoginModule场景），无需使用跨源认证。 MRS Kafka未开启Kerberos认证，仅开启了SSL认证时，创建KafkaSSL类型的认证。建表时通过sslauthname关联跨源认证。 MRS Kafka未开启Kerberos认证，开启了SASL认证和SSL认证时，创建KafkaSSL类型的认证。建表时通过sslauthname关联跨源认证。 操作步骤 1. 下载认证凭证。 DMS Kafka a.登录DMS Kafka控制台，单击实例名称进入详情页面。 b.在连接信息中，找到SSL证书，单击“下载”。 解压下载的kafkacerts压缩包，获取client.jks和phyca.crt文件。 MRS Kafka a.登录MRS Manager界面。 b.选择“系统 > 权限 > 用户”。 c.单击“更多 > 下载认证凭据”，保存后解压得到Truststore文件。 2. 上传认证凭证到OBS桶。 3. 创建跨源认证。 a.登录DLI管理控制台。 b.选择“跨源管理 > 跨源认证”。 c.单击“创建”。 填写Kafka认证信息，详细参数说明请参考下表。 参数说明 参数 参数说明 类型 选择KafkaSSL。 认证信息名称 所创建的跨源认证信息名称。 名称只能包含数字、英文字母和下划线，但不能是纯数字，且不能以下划线开头。 输入长度不能超过128个字符。 Truststore路径 上传SSL Truststore文件的OBS路径。 MRS Kafka请填写Truststore.jks文件的OBS路径。 DMS Kafka请填写client.jks文件的OBS路径。 Truststore密码 truststore密码。 Keystore路径 上传SSL KEYSTORE(密钥和证书)文件的OBS路径。 Keystore密码 keystore(密钥和证书)密码。 Key密码 keystore中的私钥密码。 4. 访问开启SASLSSL认证的Kafka。 跨源认证创建成功后，在创建访问数据源时只需关联跨源认证即可安全访问数据源。

本章节主要介绍创建Flink SQL作业 。 Flink SQL是一种由用户根据自己的逻辑需求编写作业的方式。用SQL表达业务逻辑，可以简便快捷的实现业务。目前Flink SQL作业支持两种Flink SQL语句编辑方式：SQL编辑器和可视化编辑器。本章节主要介绍使用SQL编辑器编写Flink SQL作业的方式。 前提条件 创建Flink SQL作业时，需要事先准备数据源以及数据输出通道，具体内容请参见准备Flink作业数据。 创建Flink SQL作业，访问其他外部数据源时，如访问OpenTSDB、HBase、Kafka、DWS、RDS、CSS、CloudTable、DCS Redis、DDS Mongo等，需要先创建跨源连接，打通作业运行队列到外部数据源之间的网络。 1. 当前Flink作业支持访问的外部数据源详情请参考增强型跨源连接概述。 2. 创建跨源连接操作请参见增强型跨源连接。创建完跨源连接后，可以通过“队列管理”页面，单击“操作”列“更多”中的“测试地址连通性”，验证队列到外部数据源之间的网络连通是否正常。详细操作可以参考测试地址连通性。 创建FLink SQL作业 1. 在DLI管理控制台的左侧导航栏中，单击“作业管理”>“Flink作业”，进入“Flink作业”页面。 2. 在“Flink作业”页面右上角单击“创建作业”，弹出“创建作业”对话框。 3. 配置作业信息。 详见下表：作业配置信息 参数 参数说明 类型 选择“Flink SQL”：用户通过编辑SQL语句来启动作业。 名称 作业名称，只能由字母、中文、数字、中划线和下划线组成，并且长度为1～57字节。 说明 作业名称必须是唯一的。 描述 作业的相关描述，长度为0～512字节。 模板名称 用户可以选择样例模板或自定义的作业模板。关于模板的详细信息，请参见Flink模板管理。 4. 单击“确定”，进入作业“编辑”页面。 5. 编辑SQL作业。 在SQL语句编辑区域，输入详细的SQL语句。相关SQL语句请参考《数据湖探索SQL语法参考》。 6. 单击“语义校验”，确保语义校验成功。 只有语义校验成功后，才可以执行“调试”或“启动”作业的操作。 如果校验成功，提示“SQL语义校验成功”。 如果校验失败，会在错误的SQL语句前面显示红色的“X”记号，鼠标移动到“X”号上可查看详细错误，请根据错误提示修改SQL语句。 7. 设置作业运行参数 参见下表：作业运行参数说明 参数 参数说明 所属队列 默认选择“共享队列”，用户也可以选择自定义的独享队列。 说明 当子用户在创建作业时，子用户只能选择已经被分配的队列。 当所选择队列的剩余容量不能满足作业需求时，系统会自动扩容。当队列空闲时，系统也会自动缩容。 UDF Jar 选择“独享队列”时可配置该参数。 用户自定义UDF文件，在选择UDF Jar之前需要将对应的jar包上传至OBS桶中，并在“数据管理>程序包管理”中创建程序包，具体操作请参考创建程序包。 用户可以在SQL中调用插入Jar包中的自定义函数。 CU数量 CU数量为DLI的计算单元数量和管理单元数量总和，1CU1核4G。 当前配置的CU数量为运行作业时所需的CU数，不能超过其绑定队列的CU数量。 管理单元 管理单元CU数量。 并行数 并行数是指同时运行Flink SQL作业的最大任务数。适度增加并行数会提高作业整体算力，但也须考虑线程增多带来的切换开销。 说明 最大并行数不能大于计算单元（CU数量管理单元）的4倍。 该界面并行数优先级低于代码中并行数设置。 TaskManager配置 用于设置TaskManager资源参数。 勾选后需配置下列参数： “单TM所占CU数”：每个TaskManager占用的资源数量。 “单TM Slot”：每个TaskManager包含的Slot数量。 OBS桶 选择OBS桶用于保存用户作业日志信息、checkpoint等信息。如果选择的OBS桶是未授权状态，需要单击“OBS授权”。 说明 如果同时勾选了“开启Checkpoint”和“保存作业日志”，OBS授权一次即可。 保存作业日志 设置是否将作业运行时的日志信息保存到OBS。日志信息的保存路径为：“桶名/jobs/logs/作业id开头的目录”。在作业列表中，单击对应的作业名称，在“运行日志”页签，可以单击页面提供的OBS链接跳转至对应的路径下。 注意 该参数建议勾选，否则作业运行完成后不会生成运行日志，后续如果作业运行异常则无法获取运行日志进行定位。 勾选后需配置下列参数： “OBS桶”：选择OBS桶用于保存用户作业日志信息。如果选择的OBS桶是未授权状态，需要单击“OBS授权”。 说明 如果同时勾选了“开启Checkpoint”和“保存作业日志”，OBS授权一次即可。 作业异常告警 设置是否将作业异常告警信息，如作业出现运行异常或者欠费情况，以SMN的方式通知用户。 勾选后需配置下列参数： “SMN主题”： 选择一个自定义的SMN主题。如何自定义SMN主题，请参见《消息通知服务用户指南》中“创建主题”章节。 开启Checkpoint 设置是否开启作业快照，开启后可基于Checkpoint（一致性检查点）恢复作业。 勾选后需配置下列参数： “Checkpoint间隔”：Checkpoint的时间间隔，单位为秒，输入范围1~999999，默认值为30s。 “Checkpoint模式”：支持如下两种模式： − At least once：事件至少被处理一次。 − Exactly once：事件仅被处理一次。 “OBS桶”：选择OBS桶用于保存用户Checkpoint。如果选择的OBS桶是未授权状态，需要单击“OBS授权”。 Checkpoint保存路径为：“桶名/jobs/checkpoint/作业id开头的目录”。 说明 如果同时勾选了“开启Checkpoint”和“保存作业日志”，OBS授权一次即可。 异常自动重启 设置是否启动异常自动重启功能，当作业异常时将自动重启并恢复作业。 勾选后需配置下列参数： “异常重试最大次数”：配置异常重试最大次数。单位为“次/小时”。 − 无限：无限次重试。 − 有限：自定义重试次数。 “从Checkpoint恢复”：需要同时勾选“开启Checkpoint”才可配置该参数。 空闲状态保留时长 用于清除GroupBy或Window经过最大保留时间后仍未更新的中间状态，默认设置为1小时。 脏数据策略 选择处理脏数据的策略。支持如下三种策略：“忽略”，“抛出异常”和“保存”。 说明 “保存”是指将脏数据保存到OBS桶中。 脏数据转储地址 “脏数据策略”选择“保存”时，配置该参数。单击地址框选择保存脏数据的OBS路径。 8. （可选）根据需要调试参数。作业调试功能只用于验证SQL逻辑，不会有数据写入操作。 具体操作请参见 调试Flink作业。 9. （可选）根据需要设置自定义配置。自定义配置为用户自定义。 10. 单击“保存”，保存作业和相关参数。 11. 单击“启动”，进入“启动Flink作业”页面，确认作业规格后，单击“立即启动”，启动作业。 启动作业后，系统将自动跳转到Flink作业管理页面，新创建的作业将显示在作业列表中，在“状态”列中可以查看作业状态。作业提交成功后，状态将由“提交中”变为“运行中”。运行完成后显示“已完成”。 如果作业状态为“提交失败”或“运行异常”，表示作业提交或运行失败。用户可以在作业列表中的“状态”列中，将鼠标移动到状态图标上查看错误信息，单击可以复制错误信息。根据错误信息解决故障后，重新提交。 说明 其他功能按钮说明如下： 另存为：将新建作业另存为一个新作业。 调试：对作业进行调试。具体操作请参见 格式化：对SQL语句进行格式化。 设为模板：将新创建的作业设置为作业模板。 主题设置：设置页面主题，可以设置字体大小，自动换行和页面风格。

本节介绍如何使用的公共接入点接入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(); } }

本文介绍Kafka如何配置安全组 客户端只能部署在与Kafka实例处于相同虚拟私有云（VPC）和相同子网的弹性云主机（CTECS）上。 除了CTECS、Kafka实例必须处于相同VPC之外，还需要他们的安全组分别配置了正确的规则，客户端才能访问Kafka实例。 如果CTECS、Kafka实例配置相同的安全组，安全组创建后，默认包含组内网络访问不受限制的规则。 如果CTECS、Kafka实例配置了不同安全组，可参考如下配置方式： 说明 假设CTECS、Kafka实例分别配置了安全组：sgsCTECS、sgsKafka。 以Kafka访问端口9098为例，其它实例请以实际情况为准。 以下规则，远端可使用安全组，也可以使用具体的IP地址。 配置CTECS所在安全组。CTECS所在安全组需要增加如下出方向规则，以保证客户端能正常访问Kafka实例。如果出方向规则不受限，则不用添加。 配置Kafka实例所在安全组。Kafka实例所在安全组需要增加如下入方向规则，以保证能被客户端访问。

client.sources staticlogsource client.channels staticlogchannel client.sinks kafkasink LOGTOHDFSONLINE1 client.sources.staticlogsource.type spooldir client.sources.staticlogsource.spoolDir 监控目录 client.sources.staticlogsource.fileSuffix .COMPLETED client.sources.staticlogsource.ignorePattern ^$ client.sources.staticlogsource.trackerDir 传输过程中元数据存储路径 client.sources.staticlogsource.maxBlobLength 16384 client.sources.staticlogsource.batchSize 51200 client.sources.staticlogsource.inputCharset UTF8 client.sources.staticlogsource.deserializer LINE client.sources.staticlogsource.selector.type replicating client.sources.staticlogsource.fileHeaderKey file client.sources.staticlogsource.fileHeader false client.sources.staticlogsource.basenameHeader true client.sources.staticlogsource.basenameHeaderKey basename client.sources.staticlogsource.deletePolicy never client.channels.staticlogchannel.type file client.channels.staticlogchannel.dataDirs 数据缓存路径，设置多个路径可提升性能，中间用逗号分开 client.channels.staticlogchannel.checkpointDir 检查点存放路径 client.channels.staticlogchannel.maxFileSize 2146435071 client.channels.staticlogchannel.capacity 1000000 client.channels.staticlogchannel.transactionCapacity 612000 client.channels.staticlogchannel.minimumRequiredSpace 524288000 client.sinks.kafkasink.type org.apache.flume.sink.kafka.KafkaSink client.sinks.kafkasink.kafka.topic 数据写入的topic ，如flumetest client.sinks.kafkasink.kafka.bootstrap.servers XXX . XXX . XXX . XXX :kafka 端口号 , XXX . XXX . XXX . XXX :kafka 端口号 , XXX . XXX . XXX . XXX :kafka端口号 client.sinks.kafkasink.flumeBatchSize 1000 client.sinks.kafkasink.kafka.producer.type sync client.sinks.kafkasink.kafka.security.protocol SASLPLAINTEXT client.sinks.kafkasink.kafka.kerberos.domain.name Kafka Domain名称，安全集群必填，如hadoop.xxx.1com client.sinks.kafkasink.requiredAcks 0 client.sources.staticlogsource.channels staticlogchannel client.sinks.kafkasink.channel staticlogchannel 说明 client.sinks.kafkasink.kafka.topic：数据写入的topic。若kafka中该topic不存在，默认情况下会自动创建该topic。 client.sinks.kafkasink.kafka.bootstrap.servers：Kafkabrokers列表，多个用英文逗号分隔。默认情况下，安全集群端口21007，普通集群对应端口9092。 client.sinks.kafkasink.kafka.security.protocol：安全集群为SASLPLAINTEXT，普通集群为PLAINTEXT。 client.sinks.kafkasink.kafka.kerberos.domain.name： 普通集群无需配置此参数。安全集群对应此参数的值为Kafka集群中“kerberos.domain.name”对应的值。 具体可到Broker实例所在节点上查看${BIGDATAHOME}/MRSCurrent/1 X Broker/etc/server.properties。 其中X为随机生成的数字，请根据实际情况修改。同时文件需要以Flume客户端安装用户身份保存，例如root用户。 具体可到Broker实例所在节点上查看“${BIGDATAHOME}/FusionInsightCurrent/1XBroker/etc/server.properties”。 9. 参数配置并保存后，Flume客户端将自动加载“properties.properties”中配置的内容。当spoolDir生成新的日志文件，文件内容将发送到Kafka生产者，并支持Kafka消费者消费。

接口功能介绍 重发死信队列 接口约束 无 URI POST /v3/dlq/resend 路径参数 无 Query参数 无 请求参数 请求头header参数 参数 是否必填 参数类型 说明 示例 下级对象 regionId 是 String 资源池编码 200000002368 请求体body参数 参数 是否必填 参数类型 说明 示例 下级对象 prodInstId 是 String 实例ID groupId 是 String 订阅组名称 msgId 是 String 消息ID 响应参数 参数 参数类型 说明 示例 下级对象 statusCode String 接口系统层面状态码。成功：800，失败：900 message String 描述状态 returnObj Object 返回对象。此参数所包含的参数请见“响应示例'里面的注释 error String 错误码，只有非成功才有这个字段，方便快速定位问题 201 枚举参数 无 请求示例 请求url 请求头header 无 请求体body { "prodInstId":"mqtest", "groupId": "T1", "msgId": "xxx" } 响应示例 响应成功示例 { "returnObj":{ }, "message":"success", "statusCode":800 } 响应失败示例 { "returnObj": {}, "message": "...", "error":"201", "statusCode": "900" }

本文主要介绍准备实例依赖资源。 概述 在创建Kafka实例前，您需要提前准备相关依赖资源，包括虚拟私有云（Virtual Private Cloud，以下简称VPC）、子网和安全组，并配置安全组策略。每个Kafka实例都部署在某个VPC中，并绑定具体的子网和安全组，通过这样的方式为Kafka提供一个隔离的、用户自主配置和管理的虚拟网络环境以及安全保护策略，提升实例的安全性。 如果需要通过公网访问Kafka实例，请提前准备弹性IP地址。 准备依赖资源 Kafka实例所需资源的具体要求和创建指导如下表所示。 表 Kafka实例依赖资源 准备资源 要求 创建指导 ::: VPC和子网 不同的Kafka实例可以重复使用相同的VPC和子网，也可以使用不同的VPC和子网，请根据实际需要进行配置。 在创建VPC和子网时应注意如下要求： 创建的VPC与Kafka实例在相同的区域。 子网开启IPv6后，Kafka实例支持IPv6功能。Kafka实例开启IPv6后，客户端可以使用IPv6地址连接实例。 创建VPC和子网的操作指导请参考《虚拟私有云 用户指南》的创建虚拟私有云和子网。 安全组 不同的Kafka实例可以重复使用相同的安全组，也可以使用不同的安全组，请根据实际需要进行配置。 在创建安全组时应注意如下要求： 创建安全组时，“模板”选择“自定义”。 使用Kafka实例前，添加下表中的安全组规则，其他规则请根据实际需要添加。 说明 创建安全组后，系统默认添加入方向“允许安全组内的弹性云主机彼此通信”规则和出方向“放通全部流量”规则，此时使用内网通过同一个VPC访问Kafka实例，无需添加下表中的规则。 创建安全组的操作指导请参考《虚拟私有云 用户指南》的创建安全组，为安全组添加规则的操作指导请参考《虚拟私有云 用户指南》的添加安全组规则。 弹性IP地址 在创建弹性IP地址时，应注意如下要求： 创建的弹性IP地址与Kafka实例在相同的区域。 弹性IP地址的数量必须与Kafka实例的代理个数相同。 Kafka控制台无法识别开启IPv6转换功能的弹性IP地址。 创建弹性IP地址的操作指导请参考《弹性IP 用户指南》的“申请和绑定弹性IP”。 表 安全组规则 方向 协议 类型 端口 源地址 说明 入方向 TCP IPv4 9094 0.0.0.0/0 通过公网访问Kafka实例（明文接入）。 入方向 TCP IPv4 9092 0.0.0.0/0 使用内网通过同一个VPC访问Kafka实例（明文接入）。 使用对等连接跨VPC访问Kafka实例（明文接入）。 入方向 TCP IPv6 9192 ::/0 通过IPv6地址访问Kafka实例（关闭SSL加密）（内网和公网都适用）。 入方向 TCP IPv4 9095 0.0.0.0/0 通过公网访问Kafka实例（密文接入）。 入方向 TCP IPv4 9093 0.0.0.0/0 使用内网通过同一个VPC访问Kafka实例（密文接入）。 使用对等连接跨VPC访问Kafka实例（密文接入）。 入方向 TCP IPv6 9193 ::/0 通过IPv6地址访问Kafka实例（开启SSL加密）（内网和公网都适用）。 入方向 TCP IPv4 9011 198.19.128.0/17 通过VPC终端节点实现跨VPC访问Kafka实例（密文接入和明文接入都适用）。 入方向 TCP IPv4 9011 0.0.0.0/0 使用DNAT访问Kafka实例（密文接入和明文接入都适用）。

天翼云AOne 安卓版本应用权限列表 权限名称 权限说明 具体场景或目的 INTERNET 允许应用访问网络 用于优化服务过程中的网络质量，wifi/4g网络切换状态下的网络服务保障。 ACCESSNETWORKSTATE 检测网络连接状态 用于优化服务过程中的网络质量，wifi/4g网络切换状态下的网络服务保障。 ACCESSWIFISTATE 获取WIFI连接状态 用于优化服务过程中的网络质量，wifi/4g网络切换状态下的网络服务保障。 REQUESTINSTALLPACKAGES 允许进行应用安装 用于更新应用，检测用户是否具备使用产品功能的条件，优化体验。 REQUESTIGNOREBATTERYOPTIMIZATIONS 忽略电池优化 允许VPN服务能够在后台运行服务（国内手机系统可能无法生效，需要用户到设置中手动开启，不同机型出来的权限蒙版不一样，有一些机型因为不需要用户确认没有弹出蒙版）。 MODIFYAUDIOSETTINGS 修改音频设置 会议服务：设置音频模式。 RECORDAUDIO 录制音频 AI应用中心：以语音的方式与AI进行交互、语音输入，收集语音信息（语音转文本信息） 音视频会议：发言、传输音视频流。 CAMERA 拍摄照片和视频 AI应用中心：拍摄图片，收集图片信息 消息服务：使用摄像头拍照发送图片或视频 音视频会议：开启视频、设置虚拟背景 邮件服务：使用摄像头拍摄照片 消息服务：使用摄像头拍摄照片 个人信息：拍照上传头像。 WAKELOCK 唤醒设备 保持设备唤醒状态。 WRITEEXTERNALSTORAGE 写入外部存储 日程服务：在日程中添加文件 消息服务：发送照片、文件 邮件服务：上传附件、保存邮件附件至本地、从本地选择文件插入邮件 APP更新：在设备上存储数据 消息服务：下载文件，保存图片。 READEXTERNALSTORAGE 读取外部存储 日程服务：在日程中添加文件 消息服务：发送照片、文件 邮件服务：上传附件、保存邮件附件至本地、从本地选择文件插入邮件 消息服务：发送图片、发送文件。 MANAGEEXTERNALSTORAGE 管理所有外部存储 文件管理增强（android10以上版本申请，权限等同于WRITEEXTERNALSTORAGE 和READEXTERNALSTORAGE）。 READMEDIAIMAGES 读取媒体图片 消息服务：从相册选择图片发送 AI应用中心：从相册上传照片 邮件服务：从相册选择图片插入邮件。 USEBIOMETRICUSEFINGERPRINT 生物识别/指纹 使用生物识别/指纹进行登录。 BLUETOOTHBLUETOOTHADMIN BLUETOOTHCONNECT 蓝牙权限 会议中使用蓝牙设备。

本文以自建Elasticsearch 7.10.2版本迁移至天翼云Elasticsearch实例为例子，介绍如何使用天翼云Logstash实现迁移Elasticsearch实例的数据。 Logstash 是一个开源的数据处理管道工具，广泛用于数据收集、处理和传输。它通常作为“ELK Stack”（Elasticsearch、Logstash、Kibana、Beats）的一个核心组件，用于处理结构化和非结构化数据。 天翼云Logstash可以实现将源Elasticsearch实例（如天翼云、自建或第三方Elasticsearch实例）中的数据迁移至天翼云Elasticsearch实例。在升级实例版本、实例架构调整、或跨区域的实例数据迁移时，可以选择使用天翼云Logstash迁移源Elasticsearch实例数据。 Logstash的方式迁移数据支持跨大版本，且迁移方式灵活，下表是支持的集群版本： 源目标 Elasticsearch7.10.2 OpenSearch2.19.1 Elasticsearch6.x √ √ Elasticsearch7.x小于7.10.2 √ √ Elasticsearch7.x大于7.10.2 √ √ Elasticsearch8.x √ √ 本文以自建Elasticsearch 7.10.2版本迁移至天翼云Elasticsearch实例为例子。 前提条件 已经创建天翼云Elasticsearch实例。 已经在创建的Elasticsearch实例中加装了Logstash实例。 加装Logstash能够通过内网或公网访问需要迁移的源Elasticsearch实例。 Logstash工作模型 Logstash工作模型核心部分为三部分：输入（Input）、过滤器（Filter）、输出（Output），按照配置管道文件的顺序对数据进行提取、处理转换、输出。 1.输入（Input）：Logstash支持多种数据输入源，如文件、数据库、消息队列以及Elasticsearch等。在我们的场景中，源Elasticsearch实例就是输入数据源。Logstash会批量提取源Elasticsearch实例中的数据。 2.过滤器（Filter）：过滤器是可选的，用于对输入数据进行实时处理和转换。它提供了一些强大的插件，可以对数据进行解析、变换、裁剪或其他操作。在我们的场景中，可以选择是否使用过滤器来处理迁移中的数据，例如删除源数据中不需要迁移的字段等操作。 3.输出（Output）：Logstash的输出插件负责将处理后的数据写入到目标位置，这可以是文件、数据库、消息队列，或者像本例中的天翼云Elasticsearch实例。

本节介绍分布式消息服务Kafka客户端参数配置 根据实际业务场景，客户端参数配置适当的值，客户端参数列表及说明如下所示： 表1.生产者客户端 参数 说明 retries 消息发送失败时的重试次数。 retry.backoff.ms 消息发送失败时的重试间隔，建议设置为1000。单位：毫秒。 acks 发送消息的持久化机制。为了提升发送性能， 建议设置为acks1。acks0：无需服务端的Response，性能较高、丢数据风险较大。acks1：服务端主节点写成功即返回Response，性能中等、丢数据风险中等、主节点宕机可能导致数据丢失。acksall：服务端主节点写成功且备节点同步成功才返回Response，性能较差、数据较为安全、主节点和备节点都宕机才会导致数据丢失。 batch.size 发往每个分区的消息缓存量。达到设置的数值时，就会触发一次网络请求，然后Producer客户端把消息批量发往服务器。如果batch.size设置过小，有可能影响发送性能和稳定性。建议保持默认值16384。单位：字节。 linger.ms 每条消息在缓存中的最长时间。若超过这个时间，Producer客户端就会忽略batch.size的限制，立即把消息发往服务器。建议根据业务场景， 将linger.ms设置在100~1000之间。单位：毫秒。 partitioner.class 设置分区策略。建议采用粘性分区策略，可提升发送性能。发送客户端2.4及以上版本，默认采用粘性分区策略模式。 buffer.memory 发送的内存池大小。如果内存池设置过小，则有可能导致申请内存耗时过长，从而影响发送性能，甚至导致发送超时。建议buffer.memory ≧ batch.size 分区数 2。单位：字节。 表2.消费者客户端参数 参数 说明 fetch.min.bytes 消费者从服务端获取数据的最小字节数。设置该参数时请尽量评估消息发送端的消息量，若设置过大可能会导致消费端延迟增大，过小可能会导致消费端频繁拉取消息。单位：字节。 fetch.max.wait.ms 服务端等待的最大时间。单位：毫秒。如果使用Local存储引擎，配置了fetch.min.bytes参数，服务器会等待足够的数据才会返回。超过此时间即使没有足够数据也会返回。如果是云存储引擎，一旦有新数据发送进来， 服务器就会结束等待，不需要等待fetch.min.bytes的值。 max.partion.fetch.bytes 每个分区返回的最大字节数。单位：字节。 session.timeout.ms 消费端发送心跳的时间间隔，如果在心跳时间间隔内没有发送心跳，则服务端会认为消费者死亡，从而触发Rebalance，Rebalance期间客户端将会停止消费数据等待Rebalance完成。建议将此参数设置为3000060000。单位：毫秒。默认有效值：6000300000。 max.poll.records 每次Poll获取的最大消息数量，若此值设置过大则需要尽快处理业务逻辑，避免处理过慢影响下一次Poll数据，从而导致在session.timeout.ms时间内没有发送心跳引起Rebalance。建议该值小于 的值。重要 在Java Client 0.10.1及其以上版本有单独的线程发送心跳，小于此版本或者其他语言的客户端都需考虑处理数据时间和发送心跳的间隔，防止频繁Rebalance影响正常消费。 max.poll.interval.ms 最大的Poll间隔时间，仅在Java Client 0.10.1及其以上版本需配置该参数。如果在间隔时间内消费者没有发送Poll请求，即使在session.timeout.ms参数设置的时间内发送了心跳。服务端也会认为消费者死亡，从而触发Rebalance。因此需注意此值需要合理设置，建议该值大于 的值。通常使用默认值即可。单位：毫秒。默认值：300000。 enable.auto.commit 是否采用自动提交位点机制。true：默认采用自动提交机制。false：不采用自动提交机制。默认值：true。 auto.commit.interval.ms 自动提交位点时间间隔。默认值为1000，单位：毫秒。 auto.offset.reset 消费位点重置策略。latest：从最大位点开始消费。earliest：从最小位点开始消费。none：不做任何操作，即不重置。说明建议设置成latest，而不要设置成earliest，避免因位点非法时从头开始消费，从而造成大量重复。如果是您自己管理位点，可以设置为none。

检查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。

支持跨源认证的数据源与作业类型 不同类型的作业支持跨源认证的数据源与认证方式不同。 Spark SQL支持跨源认证的数据源与约束限制请参考下表“Spark SQL支持跨源认证的数据源”。 Flink SQL 支持跨源认证的数据源与约束限制请参考下表“Flink SQL支持跨源认证的数据源”。 Spark SQL支持跨源认证的数据源 跨源认证类型 数据源 约束与限制 CSS CSS CSS集群版本选择“6.5.4”或“6.5.4”以上版本。 CSS集群已开启“安全模式”。 Password DWS、RDS、DDS、Redis Flink SQL支持跨源认证的数据源 表类型 跨源认证类型 数据源 约束与限制 源表 Kerberos Kafka MRS Kafka开启Kerberos认证。 源表 KafkaSSL Kafka DMS Kafka开启SASLSSL认证。 MRS Kafka开启SASL认证。 MRS Kafka开启SSL认证。 结果表 Kerberos HBase MRS安全集群已开启Kerberos认证。 结果表 Kerberos Kafka MRS Kafka开启Kerberos认证。 结果表 KafkaSSL Kafka DMS Kafka开启SASLSSL认证。 MRS Kafka开启SASL认证。 MRS Kafka开启SSL认证。 结果表 Password DWS、RDS、CSS 维表 Password RDS、Redis

计费项 分布式消息服务MQTT的计费项由实例费用组成，具体费用详情见下文实例规格单价。 计费项 含义 适用的计费模式 实例费用 用户选择的选择的实例规格计费，具体请参见产品规格。 按需/包周期 说明 分布式消息服务MQTT如下资源池支持X86计算CPU架构类型的计算增强型主机，提供集群和单机两种规格实例，集群版可选39代理数量。 目前在 华东1、长沙42、华南2、武汉41、西南1、华北2、呼和浩特3、太原4 资源池开放订购。 上述资源池订购和续订可享受1年83折，2年7折，3年5折优惠。 价格计算公式 分布式消息服务MQTT实例费用单价如下表所示，计费公式为： 实例费用实例规格单价 代理数量，单机版代理数量为1。 实例规格单价 Intel计算增强型 规格名称 实例单价（单个节点） 规格名称 按需标准价格(元/小时) 包月标准价格(元/月) mqtt.2u4g.cluster 0.98 441 mqtt.4u8g.cluster 2.24 1008 mqtt.8u16g.cluster 4.86 2187 mqtt.12u24g.cluster 7.38 3321 mqtt.16u32g.cluster 9 4050

检查Kafka配置的直接内存大小 在FusionInsight Manager首页，选择“集群 > 待操作集群的名称 > 服务 > Kafka > 配置 > 全部配置 > Broker（角色）> 环境变量”。将“KAFKAHEAPOPTS”参数中配置的“Xmx”值参考如下说明调大。 说明 建议“KAFKAHEAPOPTS”参数中“Xmx”和“Xms”值保持一致。 建议根据步骤2查看“Kafka直接内存使用率”，调整“KAFKAHEAPOPTS”的值为“Kafka使用的直接内存大小”的两倍（可根据实际业务场景进行修改）。 3. 保存配置，并重启Kafka服务。 4. 观察界面告警是否清除。 是，处理完毕。 否，执行步骤7。 收集故障信息 在FusionInsight Manager界面，选择“运维 > 日志 > 下载”。 5. 在“服务”中勾选待操作集群的“Kafka”。 6. 单击右上角的设置日志收集的“开始时间”和“结束时间”分别为告警产生时间的前后10分钟，单击“下载”。 7. 请联系运维人员，并发送已收集的故障日志信息。 告警清除 此告警修复后，系统会自动清除此告警，无需手工清除。 参考信息 无。

本章节主要介绍使用Kafka的操作指导。 操作场景 用户可以在集群客户端完成Topic的创建、查询、删除等基本操作。 前提条件 已安装客户端，例如安装目录为“/opt/hadoopclient”，以下操作的客户端目录只是举例，请根据实际安装目录修改。 使用Kafka客户端 1. 安装客户端，具体请参考安装客户端章节。 2. 登录安装客户端的节点。 3. 执行以下命令，切换到客户端目录，例如： plaintext cd /opt/hadoopclient/Kafka/kafka/bin 4. 执行以下命令，配置环境变量。 plaintext source /opt/hadoopclient/bigdataenv 5. 集群默认为安全模式，将实际的keytab文件路径替换以下命令中的/path/example.keytab，执行命令进行用户认证。 plaintext kinit kt /path/example.keytab klist kt /path/example.keytab sed n 4p awk '{print $NF}' 6. 创建一个Topic。 plaintext sh kafkatopics.sh create topic 主题名称partitions 主题占用的分区数 replicationfactor 主题的备份个数 zookeeper ZooKeeper角色实例所在节点IP地址:clientPort/kafka 7. 执行以下命令，查询集群中的Topic信息。 plaintext sh kafkatopics.sh list zookeeper ZooKeeper角色实例所在节点IP地址:clientPort/kafka 8. 删除创建的Topic。 plaintext sh kafkatopics.sh delete topic 主题名称 zookeeper ZooKeeper角色实例所在节点IP地址:clientPort/kafka 输入 "y"，回车。

本章节主要介绍ALM38002 Kafka堆内存使用率超过阈值的告警。 告警解释 系统每60秒周期性检测Kafka服务堆内存使用状态，当连续10次检测到Kafka实例堆内存使用率超出阈值（最大内存的95%）时产生该告警。 平滑次数为1，堆内存使用率小于或等于阈值时，告警恢复；平滑次数大于1，堆内存使用率小于或等于阈值的90%时，告警恢复。 告警属性 告警ID 告警级别 是否自动清除 38002 重要 是 告警参数 参数名称 参数含义 来源 产生告警的集群名称。 服务名 产生告警的服务名称。 角色名 产生告警的角色名称。 主机名 产生告警的主机名。 Trigger Condition 系统当前指标取值满足自定义的告警设置条件。 对系统的影响 Kafka可用内存不足，可能会造成内存溢出导致服务崩溃。 可能原因 该节点Kafka实例堆内存使用率过大，或配置的堆内存大小不合理，导致使用率超过阈值。 处理步骤 检查Kafka实例堆内存使用率 在FusionInsight Manager首页，选择“运维 > 告警 > 告警 > Kafka堆内存使用率超过阈值 > 定位信息”。查看告警上报的实例的主机名。 1. 在FusionInsight Manager首页，选择“集群 > 待操作集群的名称 > 服务 > Kafka > 实例”，选择上报告警实例主机名对应的角色。单击图表区域右上角的下拉菜单，选择“定制 > 进程 > Kafka堆内存使用率”，单击“确定”。 2. 查看Kafka使用的堆内存是否已达到Kafka设定的最大堆内存的95%。 是，执行步骤4。 否，执行步骤6。

本文主要介绍DMS for Kafka请求条件。 您可以在创建自定义策略时，通过添加“请求条件”（Condition元素）来控制策略何时生效。请求条件包括条件键和运算符，条件键表示策略语句的 Condition元素，分为全局级条件键和服务级条件键。全局级条件键（前缀为g:）适用于所有操作，服务级条件键（前缀为服务缩写，如dms:）仅适用于对应服务的操作。运算符与条件键一起使用，构成完整的条件判断语句。 DMS for Kafka通过IAM预置了一组条件键，例如，您可以先使用dms:ssl条件键检查Kafka实例是否开启SASL，然后再允许执行操作。下表显示了适用于DMS for Kafka服务特定的条件键。 表DMS for Kafka请求条件 DMS for Kafka条件键 运算符 描述 dms:publicIP BoolIsNullOrEmptyBoolIfExists 是否开启公网 dms:ssl BoolIsNullOrEmptyBoolIfExists 是否开启SASL

Apache Kafka Apache Kafka连接适用于用户在本地数据中心或ECS上自建的第三方Kafka，其中本地数据中心的Kafka需通过专线连接。 连接Apache Hadoop上的Kafka数据源时，相关参数详见下表：Apache Kafka连接参数 参数名 说明 取值样例 名称 连接的名称，根据连接的数据源类型，用户可自定义便于记忆、区分的连接名。 kafkalink Kafka broker Kafka broker的IP地址和端口。 192.168.1.1:9092 单击“显示高级属性”，然后单击“添加”，您可以添加客户端的配置属性。所添加的每个属性需配置属性名称和值。对于不再需要的属性，可单击属性后的“删除”按钮进行删除。

本章节介绍普通消息的收发方法和示例代码。其中，普通消息发送方式分为同步发送、异步发送、单向发送。 同步发送：同步发送是指消息发送方发出一条消息后，会在收到服务端同步响应之后才发下一条消息的通讯方式。 异步发送：异步发送是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。 单向发送：发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发。 收发消息前，请参考收集连接信息收集RocketMQ所需的连接信息。 准备环境 开源的Java客户端支持连接分布式消息服务RocketMQ版，推荐使用的客户端版本为4.9.7。 通过以下任意一种方式引入依赖： 1. 使用Maven方式引入依赖。 org.apache.rocketmq rocketmqclient 4.9.7 org.apache.rocketmq rocketmqacl 4.9.7 2. 点击下载依赖JAR包：rocketmqall4.9.7binrelease.zip 同步发送 同步发送是最常用的方式，是指消息发送方发出一条消息后，会在收到服务端同步响应之后才发下一条消息的通讯方式，可靠的同步传输被广泛应用于各种场景，如重要的通知消息、短消息通知等。 参考如下示例代码 import org.apache.rocketmq.acl.common.AclClientRPCHook; import org.apache.rocketmq.acl.common.SessionCredentials; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.RPCHook; import org.apache.rocketmq.remoting.common.RemotingHelper; public class ProducerNormalExample { private static RPCHook getAclRPCHook() { return new AclClientRPCHook(new SessionCredentials( "accessKey", // 分布式消息服务RocketMQ控制台用户管理菜单中创建的用户ID "accessSecret" // 分布式消息服务RocketMQ控制台用户管理菜单中创建的密钥 )); } public static void main(String[] args) throws Exception { DefaultMQProducer producer new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook()); // 填入控制台获取NAMESRV接入点地址 producer.setNamesrvAddr("XXX:xxx"); ; // 如果需要开启SSL，请增加此行代码 producer.start(); for (int i 0; i < 128; i++) try { { Message msg new Message("YOUR TOPIC", "TagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULTCHARSET)); SendResult sendResult producer.send(msg); System.out.println(sendResult); } } catch (Exception e) { e.printStackTrace(); } producer.shutdown(); } }

云产品 日志类型 备注 文本日志 文本日志 容器标准输出 容器文件日志 容器标准输出 容器文件日志 容器实例日志 数据库慢日志、错误日志 当前白名单试用中，如有需求请提交工单反馈。 数据库慢日志、错误日志 当前白名单试用中，如有需求请提交工单反馈。 数据库慢日志、错误日志 当前白名单试用中，如有需求请提交工单反馈。 数据库慢日志、错误日志 当前白名单试用中，如有需求请提交工单反馈。 数据库慢日志、错误日志 当前白名单试用中，如有需求请提交工单反馈。 访问日志 命令审计日志 <分布式消息服务Kafka> 重平衡日志 控制面（control plane）、数据面（sidecar）日志以及应用服务网格网关日志 函数调用日志 云服务操作日志 访问日志 会话日志 Elasticsearch、OpenSearch、Logstash实例日志，包括运行日志（含错误日志）、GC日志、慢索引日志、慢查询日志、Logstash运行日志 集群组件日志 微服务引擎云原生网关 网关访问日志 虚拟私有云VPC 流日志 当前白名单试用中，如有需求请提交工单反馈。 内网DNS DNS解析日志 SDWAN 流日志 当前白名单试用中，如有需求请提交工单反馈。 云安全中心 云堡垒机操作日志； 云防火墙威胁日志； 服务器安全卫士漏洞信息、弱口令、告警日志； 数据库审计日志； Web应用防火墙告警日志； 云等保专区V1.0日志 服务器安全卫士 告警日志 网页防篡改 告警日志

本文介绍如何应用事件总线EventBridge的事件流功能实现分布式消息服务RocketMQ的消息路由。 前提条件 开通事件总线EventBridge并授权。 开通分布式消息服务RocketMQ并创建最少两个主题。 背景信息 事件流作为更轻量、实时端到端的流式事件通道，提供轻量级的流式数据的过滤和转换的能力，在不同的数据仓库之间、数据处理程序之间、数据分析和处理系统之间进行数据同步。源端分布式消息服务RocketMQ生产的消息可以通过事件流这个通道被路由到目标端的分布式消息服务RocketMQ。 步骤一：创建事件流 1. 登录事件总线EventBridge控制台。 2. 在左侧导航栏，单击事件流。 3. 在事件流页面，单击创建事件流。 4. 在创建事件流面板，设置任务名称和描述，配置以下参数，然后单击保存。 a.在Source（源）配置向导，选择数据提供方为分布式消息服务RocketMQ，设置以下参数，然后单击下一步。 参数 说明 示例 实例名称 前提条件中已创建的分布式消息服务RocketMQ版实例。 xxx Topic 当前实例中的Topic。 topic1 Group 消费组名。 快速创建：自动创建以GIDEVENTBRIDGExxx命名的Group ID。 使用已有：选择当前实例中已创建的Group，请不要与已有业务的Group混用，以免影响已有的消息收发。 group1 消费位点 开始消费的位置。 最新位点：从最新位点开始消费。 最新位点 Tag 用于过滤消息的Tag值，非必填。 tag1 b.在Filtering（过滤）配置向导，设置事件过滤规则，单击下一步。 c.在Sink（目标）配置向导，选择服务类型为分布式消息服务RocketMQ，配置以下参数，单击保存，如图1所示。 参数 说明 示例 实例 选择分布式消息服务RocketMQ实例。 instancexxx Topic 选择RocketMQ实例的Topic。 topic1 消息体 选择消息体（Body）的内容，更多参考“事件内容转换”。 完整事件 自定义属性 选择自定义属性（Properties）的内容，更多参考“事件内容转换”。 空 索引 选择索引（Keys）的内容，更多参考“事件内容转换”。 空 标签 选择标签（Tags）的内容，更多参考“事件内容转换”。 空 图1 创建事件流时选择服务类型为分布式消息服务RocketMQ的事件目标 5. 创建事件流后，会有30秒~60秒的延迟时间，您可以在事件流页面的状态栏查看启动进度。

本文介绍创建Kafka自定义策略 如果系统预置的Kafka权限，不满足您的授权要求，可以创建自定义策略。自定义策略中可以添加的授权项请参考主子账号和IAM权限管理。 目前天翼云支持可视化视图和JSON视图两种方式设置策略内容，详细介绍请查看统一身份认证用户指南权限管理自定义策略 创建自定义策略 可视化视图：通过可视化视图创建自定义策略，无需了解JSON语法，按可视化视图导航栏选择云服务、操作、资源、条件等策略内容，可自动生成策略。 JSON视图：通过JSON视图创建自定义策略，可以在选择策略模板后，根据具体需求编辑策略内容；也可以直接在编辑框内编写JSON格式的策略内容。 Kafka自定义策略样例 以下策略样例表示：仅允许IAM用户使用Kafka主题管理、实例列表功能 { "Version": "1.1", "Statement": [ { "Effect": "Allow" "Action": [ "KAFKA:inst:kafkamanagetopic", "KAFKA:inst:kafkaclusterlist" ] } ] }

场景描述 在某些情况下，您可能需要将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协议接入。

实例概览展示分布式消息服务RabbitMQ实例详情信息,包括基本信息、统计信息、节点、端口等,并提供导出服务、元数据导入服务等功能。 场景描述 实例概览展示分布式消息服务RabbitMQ实例详情信息，包括基本信息、统计信息、节点、端口等，并提供导出服务、元数据导入服务等功能。 基本信息：展示实例订购选择的参数信息，支持安全组修改和绑定公网弹性IP。 统计信息：统计实例连接数、信道数、交换器数、队列数、消费者数等数据。 节点：列表展示实例节点信息。 端口：列表展示实例端口信息。 导出服务：下载ssl文件和broker元数据。 元数据导入服务：支持本地元数据文件Json格式导入。 操作步骤 基本信息 （1）登录管理控制台。 （2）进入RabbitMQ管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“实例概览”后点击“基本信息”。 统计信息 （1）登录管理控制台。 （2）进入RabbitMQ管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“实例概览”后点击“统计信息”。 节点 （1）登录管理控制台。 （2）进入RabbitMQ管理控制台。 （3）在实例列表页在操作列，目标实例行点击“管理”。 （4）点击“实例概览”后点击“节点”。

Hue连接hiveserver，不释放session，报错over max user connections如何处理？ 适用版本：MRS 3.1.0及之前的MRS 3.x版本。 1.修改两个Hue节点的以下文件： /opt/Bigdata/FusionInsightPorter8. /install/FusionInsightHue /hue/apps/beeswax/src/beeswax/models.py 2.修改文件中的396和404行的值 q self.filter(owneruser, applicationapplication).exclude(guid'').exclude(secret'') 改为 q self.filter(owneruser, applicationapplication).exclude(guidNone).exclude(secretNone) 如何重置Kafka数据？ 删除Kafka topic信息即重置Kafka数据，具体命令请参考： 删除topic： kafkatopics.sh delete zookeeper ZooKeeper集群业务IP:2181/kafka topic topicname 查询所有topic： kafkatopics.sh zookeeper ZooKeeper集群业务IP:2181/kafka list 执行删除命令后topic数据为空则此topic会立刻被删除，如果有数据则会标记删除，后续Kafka会自行进行实际删除。 Kafka目前支持的访问协议类型有哪些？ 当前支持4种协议类型的访问：PLAINTEXT、SSL、SASLPLAINTEXT、SASLSSL。 消费Kafka Topic时报错“Not Authorized to access group XXX”如何处理？ 该问题是由于集群的Ranger鉴权和集群自带的ACL鉴权冲突导致。Kafka集群使用自带的ACL进行权限访问控制，且集群的Kafka服务也开启Ranger鉴权控制时，该组件所有鉴权将由Ranger统一管理，原鉴权插件设置的权限将会失效，导致ACL权限授权未生效。可通过关闭Kafka的Ranger鉴权并重启Kafka服务来处理该问题。操作步骤如下： 1.登录FusionInsight Manager页面，选择“集群 > Kafka”。 2.在服务“概览”页面右上角单击“更多”，选择“停用Ranger鉴权”。在弹出的对话框中输入密码，单击“确定”，操作成功后单击“完成”。 3.在服务“概览”页面右上角单击“更多”，选择“重启服务”，重启Kafka服务。

DLI数据源约束 使用CDM服务迁移数据到DLI时，当前用户需拥有OBS的读取权限。 Oracle数据源约束 不支持Oracle实时增量数据同步。 分布式缓存服务（DCS ）和Redis数据源约束 1. 由于分布式缓存服务（DCS）限制了获取所有Key的命令，CDM无法支持DCS作为源端，但可以作为迁移目的端，第三方云的Redis服务也无法支持作为源端。如果是用户在本地数据中心或ECS上自行搭建的Redis支持作为源端或目的端。 2. 仅支持Hash和String两种数据格式。 文档数据库服务（DDS ）和MongoDB数据源约束 从MongoDB、DDS迁移数据时，CDM会读取集合的首行数据作为字段列表样例，如果首行数据未包含该集合的所有字段，用户需要自己手工添加字段。 云搜索服务和Elasticsearch数据源约束 1. CDM支持自动创建索引和类型，索引和类型名称只能全部小写，不能有大写。 2. 索引下的字段类型创建后不能修改，只能创建新字段。如果一定要修改字段类型，需要创建新索引或到Kibana上用Elasticsearch命令删除当前索引重新创建（数据也会删除）。 3. CDM自动创建的索引，字段类型为date时，要求数据格式为“yyyyMMdd HH:mm:ss.SSS Z”，即“20180808 08:08:08.888 +08:00”。迁移数据到云搜索服务时如果date字段的原始数据不满足格式要求，可以通过CDM的表达式转换功能转换为上述格式。 Kafka数据源约束 消息体中的数据是一条类似CSV格式的记录，可以支持多种分隔符。不支持二进制格式或其他格式的消息内容解析。

本文将为您介绍分布式消息服务RocketMQ入门的基本流程，主要包括控制台创建RocketMQ专享版实例、使用弹性云服务器连接实例的操作，帮助您快速上手RocketMQ。 操作流程 图1 RocketMQ使用流程 环境准备 RocketMQ实例运行于虚拟私有云中，在创建实例前需要确保有可用的虚拟私有云。 创建RocketMQ实例 在创建实例时，您可以根据需求选择需要的实例规格和数量，并开启SSL访问。开启SSL后，数据加密传输，安全性更高。 创建Topic 在实例创建成功后，您需要创建Topic，用于发送与接收消息。 连接RocketMQ实例 使用客户端连接实例，并通过命令行生产消费消息。 配置告警 配置RocketMQ实例监控告警策略，监控实际业务运行状态。