使用RabbitMQ Connector同步RabbitMQ数据

前提条件

• 开通事件总线EventBridge并授权。

• 创建RabbitMQ资源。

• 开通对象存储OSS服务并创建存储空间（Bucket）。更多信息，请参见控制台创建存储空间。

• 开通Serverless应用引擎服务。更多信息，请参见准备工作。

• 创建专有网络和交换机。

• 购买并部署云消息队列 Kafka 版实例。更多信息，请参见购买和部署实例。

步骤一：创建RabbitMQ资源

1. 登录云消息队列 RabbitMQ 版控制台，创建RabbitMQ实例。操作步骤，请参见创建实例。

2. 单击已创建的实例，在实例详情页面创建以下资源。

   1. 在左侧导航栏，单击静态用户名密码，然后单击创建用户名密码。更多信息，请参见创建用户名密码。

      创建完成后保存用户名和密码。

   2. 在左侧导航栏，单击Vhost列表，然后单击创建Vhost。更多信息，请参见创建Exchange。

   3. 在左侧导航栏，单击Queue列表，在当前 Vhost右侧的切换下拉列表中，选择已创建的Vhost，然后单击创建Queue。更多信息，请参见创建Queue。

步骤二：创建Connector

1. 下载RabbitMQ Connector文件，上传至提前创建好的OSS Bucket。更多信息，请参见控制台上传文件。

2. 登录云消息队列 Kafka 版控制台，在概览页面的资源分布区域，选择地域。

3. 在左侧导航栏，选择Connector生态集成 > 任务列表。

4. 在任务列表页面，单击创建任务列表。

5. 在创建任务面板。设置任务名称，配置以下配置项。

   • 任务创建

     1. 在Source（源）配置向导，选择数据提供方为Apache Kafka Connect，单击下一步。

     2. 在连接器配置配置向导，设置以下配置项，然后单击下一步。

        配置项	参数	说明
        Kafka Connect插件	Bucket存储桶	选择OSS Bucket。
        	文件	选择上传的.ZIP文件。
        Kafka资源信息	Kafka 参数配置	选择Source Connect。
        	Kafka实例	选择前提条件中创建的实例。
        	专有网络VPC	默认选择部署Kafka实例时选择的VPC ID且不可更改。
        	交换机	默认选择部署Kafka实例时选择的vSwitch ID且不可更改。
        	安全组	选择安全组。
        Kafka Connect配置信息	解析当前ZIP包下的properties文件	选择新建properties文件。选择.ZIP文件中包含的SourceConnector对应的.properties文件。路径为/etc/source-xxx.properties。在输入框中更新下列字段的取值。 connector.class：运行的Connector的包名称，无需修改。 tasks.max：Task的最大数量。 rabbitmq.host：填写RabbitMQ实例VPC接入点地址。可在RabbitMQ实例详情页面的接入点信息区域查看。 rabbitmq.username：填写步骤一：创建RabbitMQ资源中创建的RabbitMQ实例静态用户名。 rabbitmq.password：填写步骤一：创建RabbitMQ资源中创建的RabbitMQ实例静态用户名密码。 rabbitmq.virtual.host：填写步骤一：创建RabbitMQ资源中创建的Vhost。 kafka.topic：目标Kafka Topic，请在投递数据前，提前创建好目标Topic。 rabbitmq.queue：填写步骤一：创建RabbitMQ资源中创建的Queue。 示例代码如下： connector.class=com.ibm.eventstreams.connect.rabbitmqsource.RabbitMQSourceConnector name=rabbitmq-source-connector # RabbitMQ实例VPC接入点信息。 rabbitmq.host=xxx # RabbitMQ实例静态用户名密码。 rabbitmq.password=xxx # RabbitMQ实例静态用户名。 rabbitmq.username=xxx # RabbitMQ实例Vhost。 rabbitmq.virtual.host=xxx # 目标Kafka Topic。 kafka.topic=xxx # RabbitMQ实例队列。 rabbitmq.queue=xxx tasks.max=4

     3. 在实例配置配置向导，设置以下参数，然后单击下一步。

        配置项	参数	说明
        Worker规格	Worker规格	选择合适的Worker规格。
        	最小Worker数	设置最小Worker数量。
        	最大Worker数	设置最大Worker数量。此数值不得超过Task的最大数量。
        	横向扩缩容阈值 %	当利用率大于或小于设置的CPU和Memory数值时，触发自动扩容或缩容。仅当最小Worker数和最大Worker数值不相等时，需要配置此参数。
        Kafka Connect Worker 配置	自动创建Kafka Connect Worker依赖资源	建议勾选此项，此时会在选择的Kafka实例中自动创建Kafka Connect运行所需的一些Internal Topic以及ConsumerGroup，并将这些必填配置自动填入配置框中，包括以下配置项： Offset Topic：用于存储源数据偏移量，命名规则为connect-eb-offset-<任务名称>。 Config Topic：用于存储Connectors以及Tasks的配置信息，命名规则为connect-eb-config-<任务名称>。 Status Topic：用于存储Connectors以及Tasks状态信息，命名规则为connect-eb-status-<任务名称>。 Kafka Connect Consumer Group：Kafka Connect Worker用于消费Internal Topics的消费组，命名规则为connect-eb-cluster-<任务名称>。 Kafka Source Connector Consumer Group：只针对Sink Connector有效，用于消费源Topic中的数据，命名规则为connector-eb-cluster-<任务名称>-<connector名称>。

     4. 在运行配置区域，将日志投递方式设置为投递至SLS或者投递至Kafka，在角色授权卡片设置Connect依赖的角色配置，然后单击保存。

        重要

        建议配置的角色包含AliyunSAEFullAccess权限，否则可能会导致任务运行失败。

   • 任务属性

     设置此任务的重试策略及死信队列。更多信息，请参见重试和死信。

   等待任务状态变为运行中，此时Connector已经在正常工作中。

步骤三：测试Connector

1. 登录云消息队列 RabbitMQ 版控制台，然后在左侧导航栏选择实例列表。

2. 在实例列表页面的顶部菜单栏选择地域，然后在实例列表中，单击目标实例名称。

3. 在左侧导航栏，单击Queue列表，然后单击目标Queue右侧操作列的详情。

4. 在Queue详情页面，单击被绑定信息页签的添加被绑定。

5. 在添加被绑定面板，选择源Exchange为amq.direct，单击确定。

6. 在被绑定信息页签，单击amq.direct Exchange右侧操作列的发送消息，向Kafka的目标Topic发送消息。更多信息，请参见发送消息。

7. 登录云消息队列 Kafka 版控制台，在实例列表页面，单击目标实例。

8. 在目标实例页面，单击目标Topic，然后单击消息查询，查看插入的消息数据。

All tasks under connector mongo-source failed, please check the error trace of the task.

Kafka connect exited! Please check the error log /opt/kafka/logs/connect.log on sae application to find out the reason why kafka connect exited and update the event streaming with valid arguments to solve it.

Start or update connector xxx failed. Error code=400. Error message=Connector configuration is invalid and contains the following 1 error(s):
Value must be one of never, initial_only, when_needed, initial, schema_only, schema_only_recovery
You can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`

curl -i -X PUT -H "Accept:application/json" -H  "Content-Type:application/json" -d @$CONNECTOR_PROPERTIES_MAPPING http://localhost:8083/connector-plugins/io.confluent.connect.jdbc.JdbcSinkConnector/config/validate

"value":{
    "name":"snapshot.mode",
    "value":null,
    "recommended_values":[
        "never",
        "initial_only",
        "when_needed",
        "initial",
        "schema_only",
        "schema_only_recovery"
    ],
    "errors":[
        "Value must be one of never, initial_only, when_needed, initial, schema_only, schema_only_recovery"
    ],
    "visible":true
}