EventBridge:Synchronize data from ApsaraMQ for RabbitMQ to ApsaraMQ for Kafka by using the RabbitMQ connector

Configure the data source

1. In the Source step, set Data Provider to Apache Kafka Connect and click Next Step.

2. In the Connector step, configure the following parameters and click Next Step. Update the following fields in the code editor: Sample .properties file:

   Subsection	Parameter	Description
   Kafka Connect Plug-in	Bucket	The OSS bucket to which the RabbitMQ Connector file was uploaded.
   	File	The RabbitMQ Connector file you uploaded to the OSS bucket.
   Message Queue for Apache Kafka Resources	Message Queue for Apache Kafka Parameters	Select Source Connect.
   	Message Queue for Apache Kafka Instance	The ApsaraMQ for Kafka instance you created in the prerequisites.
   	VPC	Auto-populated from your Kafka instance deployment. This value cannot be changed.
   	vSwitch	Auto-populated from your Kafka instance deployment. This value cannot be changed.
   	Security Group	Select a security group.
   Kafka Connect	Parse .properties File in ZIP Package	Select Create .properties File. Then, select the .properties file that contains the source connector configurations from the ZIP package. The path is /etc/source-xxx.properties.

   Parameter	Description
   connector.class	The connector class name. Use the default value.
   tasks.max	The maximum number of tasks.
   rabbitmq.host	The VPC endpoint of the ApsaraMQ for RabbitMQ instance. Find this in the Endpoint Information section of the Instance Details page.
   rabbitmq.username	The username you created in Step 1.
   rabbitmq.password	The password you created in Step 1.
   rabbitmq.virtual.host	The vhost you created in Step 1.
   kafka.topic	The destination ApsaraMQ for Kafka topic. Create this topic before synchronizing data.
   rabbitmq.queue	The queue you created in Step 1.

      connector.class=com.ibm.eventstreams.connect.rabbitmqsource.RabbitMQSourceConnector
      name=rabbitmq-source-connector
      # The VPC endpoint of the ApsaraMQ for RabbitMQ instance.
      rabbitmq.host=<your-rabbitmq-vpc-endpoint>
      # The static password for the ApsaraMQ for RabbitMQ instance.
      rabbitmq.password=<your-rabbitmq-password>
      # The static username for the ApsaraMQ for RabbitMQ instance.
      rabbitmq.username=<your-rabbitmq-username>
      # The vhost of the ApsaraMQ for RabbitMQ instance.
      rabbitmq.virtual.host=<your-rabbitmq-vhost>
      # The destination ApsaraMQ for Kafka topic.
      kafka.topic=<your-kafka-topic>
      # The queue of the ApsaraMQ for RabbitMQ instance.
      rabbitmq.queue=<your-rabbitmq-queue>
      tasks.max=4

Configure workers

1. In the Instance step, configure the following parameters and click Next Step. The following internal resources are created automatically:

   Subsection	Parameter	Description
   Worker Type	Worker Type	Select a worker type.
   	Min. Number of Workers	The minimum number of workers.
   	Max. Number of Workers	The maximum number of workers. The value cannot exceed the tasks.max value.
   	Horizontal Scaling Threshold (%)	The CPU utilization threshold that triggers auto scaling. Required only if the minimum and maximum worker counts differ.
   Worker Configurations	Automatically Create Dependencies for Apache Kafka Connector Worker	Select this option (recommended). The system creates the internal topics and consumer groups required to run Kafka Connect in the selected ApsaraMQ for Kafka instance and fills the resource information in the code editor.

   Resource	Naming format
   Offset Topic	connect-eb-offset-<Task name>
   Config Topic	connect-eb-config-<Task name>
   Status Topic	connect-eb-status-<Task name>
   Kafka Connect Consumer Group	connect-eb-cluster-<Task name>
   Kafka Source Connector Consumer Group	connector-eb-cluster-<task name>-<connector name>

Configure running settings and task properties

1. In the Running Configurations section, set the Log Delivery parameter to Deliver Data to Log Service or Deliver Data to ApsaraMQ for Kafka. In the Role Authorization subsection, select a role from the Role drop-down list, and then click Save.

   Important

   Select a role that has the AliyunSAEFullAccess permission policy attached. Otherwise, the task may fail to run.

2. Configure the retry policy and dead-letter queue for the task. For more information, see Retry policies and dead-letter queues.

When the task status changes to Running, the connector is working as expected.

All tasks fail to run

Error message:

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

Solution: On the Message Inflow Task Details page, click Diagnostics in the Basic Information section to go to the Connector Monitoring page. View the task failure details there.

Kafka Connect unexpectedly exits

Error message:

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.

Solution: The status update of Kafka Connect may be delayed. Refresh the page first. If Kafka Connect still fails, troubleshoot by following these steps:

1. In the Worker Information section of the Message Inflow Task Details page, click the instance name next to SAE Application to go to the Application Details page.

2. On the Basic Information page, click the Instance Deployment Information tab.

3. Click Webshell in the Actions column to log on to the running environment of Kafka Connect.

   • Run vi /home/admin/connector-bootstrap.log to view the startup logs and check for error messages.

   • Run vi /opt/kafka/logs/connect.log to view the running logs and check for ERROR or WARN entries.

   

After troubleshooting the issue, restart the task.

Connector parameter verification fails

Error message:

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`

Solution: Find the parameter with the invalid value from the error message and update it. If the error message does not identify the parameter, log on to the Kafka Connect running environment (see Kafka Connect unexpectedly exits) and run the following command:

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

The response returns the validation result for each connector parameter. If a parameter value is invalid, its errors field is not empty.

"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
}