本文介绍如何在消息队列Kafka版控制台查看Connector任务配置信息、查看Connector日志、启暂停Connector、修改Connector配置、测试Connector、查看消费详情和删除Connector。

前提条件

您已创建以下任意一种Connector:
说明 仅FC Sink Connector和MySQL Source Connector支持修改Connector配置。

注意事项

  • 仅FC Sink Connector支持修改Connector配置。
  • DLA Sink Connector的相关操作,请参见创建DLA Sink Connector
  • 暂停Connector时,Connector需处于运行中状态;启用Connector时,Connector需处于暂停状态。

查看Connector任务配置

您可以查看已创建的Connector的基础信息、源实例配置、目标实例配置等信息。

  1. 登录消息队列Kafka版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 在左侧导航栏,单击Connector 任务列表
  4. Connector 任务列表页面,从选择实例的下拉列表选择Connector所属的实例,在实例下的Connector列表中找到目标Connector,单击操作列的详情
    在Connector详情页面,显示Connector的基础信息源信息目标信息

查看Connector日志

如果您的Connector出现异常,您可以在消息队列Kafka版控制台查看Connector的日志以进行异常排查。

  1. 登录消息队列Kafka版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 在左侧导航栏,单击Connector 任务列表
  4. Connector 任务列表页面,从选择实例的下拉列表选择Connector所属的实例,在实例下的Connector列表中找到目标Connector,在其操作列,选择更多 > 查看日志
    您也可以在目标Connector名称处单击或在其操作列,单击详情,然后在Connector详情页面右上角单击查看日志
    说明 操作查看日志位置和任务状态有关,实际请以界面显示为准。
    在查看Connector日志页面,查看Connector的日志。

修改Connector配置

成功创建FC Sink Connector或MySQL Source Connector后,您可以在消息队列Kafka版控制台更新其配置。

  1. 登录消息队列Kafka版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 在左侧导航栏,单击Connector 任务列表
  4. Connector 任务列表页面,从选择实例的下拉列表选择Connector所属的实例,在实例下的Connector列表中找到目标Connector,在其操作列,选择更多 > 修改配置
    您也可以在目标Connector名称处单击或在其操作列,单击详情,然后在Connector详情页面右上角单击修改配置
  5. 修改配置面板,按需修改配置参数,然后单击确定
    • 修改FC Sink Connector配置
      参数 描述
      消费线程并发数 数据源Topic的消费线程并发数。默认值为6。取值说明如下:
      • 1
      • 2
      • 3
      • 6
      • 12
      失败处理 消息发送失败后,是否继续订阅出现错误的Topic的分区。取值说明如下:
      • 继续订阅:继续订阅出现错误的Topic的分区,并打印错误日志。
      • 停止订阅:停止订阅出现错误的Topic的分区,并打印错误日志。
      说明
      • 如何查看日志,请参见Connector相关操作
      • 如何根据错误码查找解决方案,请参见错误码列表
      • 如需恢复对出现错误的Topic的分区的订阅,您需要提交工单联系消息队列Kafka版技术人员。
      发送模式 消息的发送模式。取值如下:
      • 同步:发送端和接收端必须同时处于运行状态,随时做好相互通信的准备。
      • 异步: 发送端在发送消息之后可以不用等待服务端的响应, 可以接着处理其他任务。
      发送批大小 批量发送消息的条数。

      默认值为20,取值范围为[1,5000]。

      重试次数 消息发送失败后的重试次数。默认为2。取值范围为1~3。部分导致消息发送失败的错误不支持重试。关于错误码的详细信息,请参见错误码列表。错误码与是否支持重试的对应关系如下:
      • 4xx:除429支持重试外,其余错误码不支持重试。
      • 5xx:支持重试。
      说明 Connector调用InvokeFunction向函数计算发送消息。
    • 修改MySQL Source Connector
      参数 描述
      数据库表 需要同步的阿里云数据库RDS MySQL版实例数据库表的名称,多个表名以英文逗号(,)分隔。

      数据库表和目标Topic是一对一的关系。

      说明 如果是在原任务基础上增加或修改表,需提前在消息队列Kafka版控制台创建与之对应的Topic,即Topic名称为Topic前缀_数据表名,否则会在运行时出错。创建Topic的具体操作,请参见步骤一:创建Topic
    修改完成后,在Connector 任务列表页面,找到目标Connector。在其操作列,单击详情。在Connector 详情页面,查看到更新后的Connector配置。

启停Connector

如果您需要暂时中止某个运行中的Connector或恢复某个暂停的Connector,您可以在消息队列Kafka版控制台暂停或启用该Connector。

  1. 登录消息队列Kafka版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 在左侧导航栏,单击Connector 任务列表
  4. Connector 任务列表页面,从选择实例的下拉列表选择Connector所属的实例,在实例下的Connector列表中找到目标Connector,根据需要暂停或启用该Connector。
    • 暂停Connector:在其操作列,选择更多 > 暂停,然后在提示对话框,确认暂停Connector,单击确认
    • 启用Connector:在其操作列,单击启用,然后在提示对话框,确认启动Connector,单击确认

    您也可以在目标Connector名称处单击或在其操作列,单击详情,然后在Connector详情页面右上角单击暂停启用

    Connector 任务列表页面,找到暂停或启用的Connector,状态已变更为已暂停运行中
    说明 如果是MySQL Source Connector任务,暂停时间比较久时,会出现启动失败的情况。您可以在目标任务的操作,单击任务配置,在DataWorks页面,目标同步任务所在行,单击启动。在启动对话框,勾选重置位点,并配置启动时间点位,然后单击确定

测试Connector

如果您需要测试某个Connector,您可以在消息队列Kafka版控制台向Connector发送测试消息。

  1. 登录消息队列Kafka版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 在左侧导航栏,单击Connector 任务列表
  4. Connector 任务列表页面,从选择实例的下拉列表选择Connector所属的实例,在实例下的Connector列表中找到目标Connector,在其右侧操作列,单击测试
    您也可以在目标Connector名称处单击或在其操作列,单击详情,然后在Connector详情页面右上角单击测试
    说明 操作测试位置和任务状态有关,实际请以界面显示为准。
  5. 发送消息面板,发送测试消息。
    1. 消息 Key文本框中输入消息的Key值,例如demo。
    2. 消息内容文本框输入测试的消息内容,例如 {"key": "test"}。
    3. 设置发送到指定分区,选择是否指定分区。
      • 单击,在分区 ID文本框中输入分区的ID,例如0。如果您需查询分区的ID,请参见查看分区状态
      • 单击,不指定分区。

查看消费进度

Connector任务创建完成并开始运行后,您可以查看Connector数据同步任务使用的Group在数据源Topic中的消费情况。其中,Group的名称为connect-{任务名称}。

  1. 登录消息队列Kafka版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 在左侧导航栏,单击Connector 任务列表
  4. Connector 任务列表页面,从选择实例的下拉列表选择Connector所属的实例,在实例下的Connector列表中找到目标Connector,在其操作列,单击消费进度
    您也可以在目标Connector任务名称处单击或在其操作列单击详情,然后在Connector详情页面右上角单击消费进度
    Group消费详情面板,您可以查看Connector任务使用的Group在数据源Topic中的消费详情。

删除Connector

消息队列Kafka版限制了每个实例的Connector数量。如果您不再需要某个Connector,您可以在消息队列Kafka版控制台删除该Connector。

注意
  • 如果MySQL Source Connector任务处于运行状态,在消息队列Kafka版控制台将无法直接删除,您需登录DataWorks控制台停止并下线Connector任务,然后提交工单清理消息队列Kafka版Connector任务的元信息。其他FC Sink Connector、MaxCompute Sink Connector、OSS Sink Connector以及Elasticsearch Sink Connector任务,均可在消息队列Kafka版控制台直接删除。
  • 当删除Connector时,系统会同时删除该Connector依赖的5个Topic和2个Group,无论这些资源当时是自动创建的还是手动创建的。
  1. 登录消息队列Kafka版控制台
  2. 概览页面的资源分布区域,选择地域。
  3. 在左侧导航栏,单击Connector 任务列表
  4. Connector 任务列表页面,从选择实例的下拉列表选择Connector所属的实例,在实例下的Connector列表中找到目标Connector,在其操作列,选择更多 > 删除
    您也可以在目标Connector任务名称处单击或在其操作列单击详情,然后在Connector详情页面右上角单击删除
  5. 提示对话框,单击确认,删除Connector。