Function Compute:DTS triggers

How it works

When you create a DTS trigger, Function Compute automatically creates an event stream in EventBridge. Each time the change tracking task captures incremental data, EventBridge pushes one or more change records to your function in a single invocation, based on your batch configuration.

You can view the trigger in the Function Compute console and the auto-created event stream in the EventBridge console.

Limits

• The DTS change tracking task and the Function Compute function must be in the same region.

• If the number of event streams reaches the quota limit, you cannot create additional DTS triggers. For quota details, see Limits.

Prerequisites

Before you begin, ensure that you have:

• Activated EventBridge and granted the required permissions. See Activate EventBridge and grant permissions.

• A Function Compute function. See Create a function.

• A DTS change tracking task. See Manage a change tracking task.

• A consumer group for the change tracking task. See Create a consumer group.

Create a DTS trigger

1. Log on to the Function Compute console. In the left-side navigation pane, click Functions.

2. In the top navigation bar, select a region. On the Functions page, click the function you want to manage.

3. On the function details page, click the Triggers tab, then click Create Trigger.

4. In the Create Trigger panel, configure the parameters described below, then click OK.

Configure test parameters

The DTS trigger passes an event parameter to your function — a JSON array of change records conforming to the CloudEvents specification. To test your function locally, configure a test event that simulates this input.

1. On the Code tab of the function details page, click the icon next to Test Function and select Configure Test Parameters.

2. In the Configure Test Parameters panel, select Create New Test Event or Edit Existing Test Event, enter a name, paste the event JSON, then click OK.

[
  {
    "id": "12f701a43741d404fa9a7be89d9acae0-321****",
    "source": "DTSstreamDemo",
    "specversion": "1.0",
    "type": "dts:ConsumeMessage",
    "datacontenttype": "application/json; charset=utf-8",
    "time": "2022-06-10T07:55:57Z",
    "subject": "acs:dts:cn-hangzhou:12345****:kk123abc60g782/dtsabcdet1ro",
    "data": {
      "id": 321,
      "operationType": "UPDATE",
      "sourceTimestamp": 1654847757,
      "offset": 3218099,
      "topicPartition": {
        "topic": "cn_hangzhou_rm_1234****_test_version2",
        "partition": 0,
        "hash": 0
      },
      "schema": {
        "databaseName": "hangzhou-test-db",
        "tableName": "message_info",
        "databaseInfo": {
          "databaseType": "MySQL",
          "version": "5.7.35-log"
        }
      },
      "beforeImage": { "values": [...], "size": 45 },
      "afterImage":  { "values": [...], "size": 47 }
    }
  }
]

Write and test the function code

After creating the trigger, write your function handler to process the incoming change records.

1. On the Code tab of the function details page, write your handler in the code editor, then click Deploy Code. The following Node.js example parses each change record from the event and logs the operation type, affected table, and changed fields:

   'use strict';
   
   exports.handler = (event, context, callback) => {
     const records = JSON.parse(event.toString());
   
     records.forEach(record => {
       const { operationType, schema, beforeImage, afterImage } = record.data;
   
       console.log(`Operation: ${operationType}`);
       console.log(`Table: ${schema.databaseName}.${schema.tableName}`);
   
       // For UPDATE and DELETE, log the values before the change.
       if (beforeImage) {
         console.log('Before:', JSON.stringify(beforeImage.values));
       }
   
       // For INSERT and UPDATE, log the values after the change.
       if (afterImage) {
         console.log('After:', JSON.stringify(afterImage.values));
       }
     });
   
     callback(null, 'processed');
   };

2. Click Test Function.

What's next

• Configure triggers using Serverless Devs. See Common commands of Serverless Devs.

• Configure triggers using the SDK. See SDKs.

• Modify or delete an existing trigger. See Manage triggers.