Docs

Cloud Native Message Engine

SDK reference

Python SDK

This topic describes how to use the Python SDK to connect to the cloud-native messaging engine BMQ and send and receive messages.

Prerequisites

Create a resource instance and obtain the endpoint address. For details, see Manage resource pools.

You must open port 9092 in the security group associated with the instance in advance. For detailed instructions, see Add security group access rules.

(Optional) To authenticate using a SASL username and password, you must first create a user and obtain the password. For detailed instructions, see Create SASL user.

Install dependencies

pip install kafka-python

Configure debug logs

import logging​
import sys​
logger = logging.getLogger('kafka')​
logger.addHandler(logging.StreamHandler(sys.stdout))​
logger.setLevel(logging.DEBUG)​

Send messages

Create and write producer.py to send messages.

PLAINTEXT

When connecting to the BMQ instance using the endpoint address with the PLAINTEXT protocol, authentication is not required.

from kafka import KafkaProducer​
producer = KafkaProducer(​
    bootstrap_servers='your broker list',​
    api_version=(0, 10, 2),​
)​
for _ in range(100):​
    result = producer.send('your topic', b'some_message_bytes').get()​
    print("send message: partition " + str(result.partition) + " offset " + str(result.offset))​

SASL_PLAINTEXT

Authenticate using a SASL username and password. You must obtain the username and password of an authorized user. For instructions, see View SASL user password.

from kafka import KafkaProducer​
producer = KafkaProducer(​
    bootstrap_servers="your broker list",​
    security_protocol="SASL_PLAINTEXT",​
    sasl_mechanism="PLAIN",​
    sasl_plain_username="User name",​
    sasl_plain_password="Password",​
    api_version=(0, 10, 2),​
)​
for _ in range(100):​
    result = producer.send('your topic', b'some_message_bytes').get()​
    print("send message: partition " + str(result.partition) + " offset " + str(result.offset))​

SASL_SSL

Authenticate using a SASL username and password. You must obtain the username and password of an authorized user. For instructions, see View SASL user password.

from kafka import KafkaProducer​
producer = KafkaProducer(​
    bootstrap_servers="your broker list",​
    security_protocol="SASL_SSL",​
    sasl_mechanism="PLAIN",​
    sasl_plain_username="User name",​
    sasl_plain_password="Password",​
    ssl_check_hostname=False,​
    api_version=(0, 10, 2),​
)​
for _ in range(100):​
    result = producer.send('your topic', b'some_message_bytes').get()​
    print("send message: partition " + str(result.partition) + " offset " + str(result.offset))​

consume messages

Create and write consumer.py to receive messages.

PLAINTEXT

When connecting to the BMQ instance using the endpoint address with the PLAINTEXT protocol, authentication is not required.

from kafka import KafkaConsumer​
consumer = KafkaConsumer('your topic', bootstrap_servers="your broker list", group_id="your consumer group",api_version=(0, 10, 2))​
for msg in consumer:​
    print(msg)​

SASL_PLAINTEXT

Authenticate using a SASL username and password. You must obtain the username and password of an authorized user. For instructions, see View SASL user password.

from kafka import KafkaConsumer​
consumer = KafkaConsumer(​
    "your topic",​
    bootstrap_servers="your broker list",​
    group_id="your consumer group",​
    security_protocol="SASL_PLAINTEXT",​
    sasl_mechanism="PLAIN",​
    sasl_plain_username="User name",​
    sasl_plain_password="Password",​
    api_version=(0, 10, 2),​
)​
for msg in consumer:​
    print(msg)​

SASL_SSL

Authenticate using a SASL username and password. You must obtain the username and password of an authorized user. For instructions, see View SASL user password.

from kafka import KafkaConsumer​
consumer = KafkaConsumer(​
    "your topic",​
    bootstrap_servers="your broker list",​
    group_id="your consumer group",​
    security_protocol="SASL_SSL",​
    sasl_mechanism="PLAIN",​
    sasl_plain_username="User name",​
    sasl_plain_password="Password",​
    ssl_check_hostname=False,​
    api_version=(0, 10, 2),​
)​
for msg in consumer:​
    print(msg)​

Last updated: 2026.01.26 15:46:58