How to Route Kafka Messages to MongoDB DML with Alibaba Cloud Function Compute

Solution Overview

In big‑data ETL pipelines, Kafka streams are often forwarded to downstream services. This solution extends the pattern by using the Kafka message key to decide which DML operation (insert, update, delete) should be applied to MongoDB. When a message arrives, Alibaba Cloud Function Compute (FC) is automatically invoked, parses the key, and executes the corresponding MongoDB command.

Key Components

Virtual Private Cloud (VPC) – isolated network where Kafka, FC and MongoDB are deployed.

Alibaba Cloud Message Queue – Kafka Edition – source of the message stream.

Function Compute (FC) – event‑driven, fully managed compute service that runs the DML logic.

ApsaraDB for MongoDB – MongoDB‑compatible database service.

Architecture

The resources are placed in a single region (example: Beijing) and a single VPC. A VPC switch in a specific zone (G‑zone) connects the Kafka broker, FC instance and MongoDB cluster. FC accesses MongoDB through the VPC security group.

Processing Flow

Kafka receives a message whose key encodes the desired operation, e.g. INSERT, UPDATE, DELETE.

FC is subscribed to the Kafka topic. The platform automatically triggers the function for each new record.

The function extracts the key, maps it to a MongoDB command, and executes the command against the target collection using the connection string stored in an environment variable.

FC logs the processing result; on failure the built‑in retry policy and callback mechanism can re‑invoke the function or invoke a compensation routine.

Prerequisites

Alibaba Cloud account with real‑name verification.

Paid resources with sufficient balance (≥ 100 CNY) to cover VPC, Kafka, FC and MongoDB usage.

Deployment Steps

Use the Cloud Architect Design Tools (CADT) template to provision VPC, subnet, security group, Kafka instance, FC service and MongoDB cluster in one click.

Configure MongoDB:

Add the VPC CIDR to the whitelist.

Record the connection URI (e.g. mongodb://username:password@host:port/db).

Create the target database and collection.

Configure Function Compute:

Set environment variables such as MONGODB_URI, DB_NAME, COLLECTION.

Define the function handler (e.g. index.handler) that parses the Kafka record.

Attach any required layers (e.g. MongoDB driver).

Upload the function code that implements the key‑to‑DML mapping.

Validate the pipeline:

Publish test messages to the Kafka topic with different keys.

Query MongoDB to confirm that insert, update and delete operations were performed as expected.

When testing is complete, release the resources or delete the stack via CADT.

Function Code Sketch

import os
import json
from pymongo import MongoClient

client = MongoClient(os.getenv('MONGODB_URI'))
db = client[os.getenv('DB_NAME')]
coll = db[os.getenv('COLLECTION')]

def handler(event, context):
    for record in event['records']:
        key = record['key'].decode('utf-8')
        payload = json.loads(record['value'].decode('utf-8'))

        if key == 'INSERT':
            coll.insert_one(payload)
        elif key == 'UPDATE':
            # payload should contain 'filter' and 'update' fields
            coll.update_one(payload['filter'], {'$set': payload['update']})
        elif key == 'DELETE':
            coll.delete_one(payload['filter'])
        else:
            raise ValueError(f'Unsupported key: {key}')
    return 'OK'

Limitations

The current implementation only supports Alibaba Cloud Kafka as the source; other Kafka providers are not covered.