Huawei Consume

Huawei Consume

Certified

Consume messages from a Huawei DMS for Kafka topic

Polls the configured topic until maxRecords or maxDuration is reached (at least one is required), or until the topic is fully drained — whichever comes first. The task stops early when all assigned partitions have been read to their current end offset, so maxRecords acts as an upper bound rather than a target the task will block waiting for. Messages are written to Kestra internal storage as ION at uri. Offsets are committed after all records have been written, ensuring at-least-once delivery semantics.

yaml
type: io.kestra.plugin.huawei.dms.kafka.Consume
yaml
id: dms_kafka_consume
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.huawei.dms.kafka.Consume
    bootstrapServers: "dms-instance-id.kafka.eu-west-101.myhuaweicloud.com:9093"
    saslMechanism: PLAIN
    username: "{{ secret('DMS_KAFKA_USERNAME') }}"
    password: "{{ secret('DMS_KAFKA_PASSWORD') }}"
    topic: my-topic
    groupId: kestra-consumer-group
    maxRecords: 100
    valueSerdeType: JSON
Properties

Kafka bootstrap servers

Comma-separated list of host: port pairs that the Kafka client uses for the initial cluster connection. For DMS for Kafka, copy this value from the instance detail page in the console.

Consumer group ID

Identifies the consumer group used for offset tracking. Multiple tasks with the same group ID share the topic partitions.

Kafka topic to consume from

Access Key (AK) used to authenticate with Huawei Cloud

Huawei Cloud access key used together with secretAccessKey to sign API requests. Required for AK/SK-based authentication; not required when providing a pre-obtained securityToken. Sensitive — always provide via {{ secret('NAME') }}.

Huawei Cloud Account Domain ID

Identifies the Huawei Cloud account (domain). Required when authenticating against global services such as IAM, or when requesting a domain-scoped IAM token.

DefaultSTRING
Possible Values
STRINGJSONBINARY

Key serializer/deserializer type

STRING (default), JSON, or BINARY.

Stop after this duration has elapsed

ISO-8601 duration, e.g. PT30S. At least one of maxRecords or maxDuration must be set.

Stop after consuming this many records

Upper bound on the number of records to consume. The task may return fewer records if the topic is drained before this limit is reached. At least one of maxRecords or maxDuration must be set.

SASL password

Required when saslMechanism is PLAIN or SCRAM_SHA_512. Sensitive — always provide via {{ secret('NAME') }}.

Reference (ref) of the pluginDefaults to apply to this task.

DefaultPT5S

How long each poll waits for new records

ISO-8601 duration, e.g. PT5S (default). Shorter values reduce latency; longer values reduce CPU load.

Huawei Cloud Project ID

Identifies the region-scoped project against which most regional services authenticate. Mutually exclusive with domainId for global services such as IAM.

Huawei Cloud region

Region identifier such as eu-west-101, ap-southeast-1, or cn-north-4.

DefaultPLAIN
Possible Values
PLAINSCRAM_SHA_512NONE

SASL mechanism used for authentication

PLAIN — username/password (default, used by most DMS for Kafka instances). SCRAM_SHA_512 — stronger challenge-response, supported on newer instances. NONE — no SASL; for VPC-internal clusters with no auth enabled.

Secret Key (SK) used to authenticate with Huawei Cloud

Huawei Cloud secret key paired with accessKeyId. Required for AK/SK-based authentication. Sensitive — always provide via {{ secret('NAME') }}.

Pre-obtained Huawei Cloud IAM token used as bearer credential for downstream API calls

When set, downstream Huawei tasks send this value in the X-Auth-Token header instead of signing requests with AK/SK. Sensitive.

Defaultfalse

Enable TLS for the Kafka connection

Set to true to use SASL_SSL instead of SASL_PLAINTEXT. DMS for Kafka instances accessed over the public internet require TLS.

Inline IAM credential exchange

When set, the connection layer calls the Huawei IAM STS API once per task execution and uses the returned temporary AK/SK + security token instead of the static accessKeyId and secretAccessKey properties.

Configure once via pluginDefaults to apply transparently to every task in a namespace without per-task credential wiring: 

pluginDefaults: 
  - type: io.kestra.plugin.huawei.obs
    values: 
      region: eu-west-101
      temporaryCredentials: 
        authMethod: PASSWORD
        username: my-iam-user
        password: "{{ secret('HUAWEI_IAM_PASSWORD') }}"
        domainName: my-account-domain
        durationSeconds: 3600

**Long-running tasks: ** the exchange runs once at execution start. For RealtimeTrigger or long-running Consume tasks that outlive durationSeconds, credentials will expire mid-run. Use long-lived AK/SK properties or refresh externally in that case.

Definitions
authMethodstring
DefaultPASSWORD
Possible Values
PASSWORDTOKEN

Authentication method

Controls which credentials are used to obtain the session token before exchanging for temporary STS credentials.

  • PASSWORD (default): provide username, password, and domainName.
  • TOKEN: provide an existing iamToken (X-Auth-Token).
domainNamestring

Account domain name (PASSWORD method only)

The Huawei Cloud account name (domain name) that owns the IAM user. Required when authMethod is PASSWORD. Visible in the Huawei Cloud console under My Credentials → Domain Name.

durationSecondsintegerstring
Default900

Lifetime of the temporary credentials in seconds

How long the returned temporary AK/SK/security-token should remain valid. Huawei Cloud accepts values between 900 (15 minutes) and 86400 (24 hours). Defaults to 900 seconds.

endpointSuffixstring
Defaultmyhuaweicloud.com

Huawei Cloud IAM endpoint suffix

Domain suffix used to build the IAM endpoint URL when no explicit endpoint override is set. Defaults to myhuaweicloud.com. Set to myhuaweicloud.eu for the European sovereign cloud (region eu-west-101 / EU-Dublin).

iamTokenstring

IAM token to exchange (TOKEN method only)

An existing Huawei Cloud X-Auth-Token to exchange for temporary STS credentials. Required when authMethod is TOKEN. Sensitive — always provide via {{ secret('NAME') }}.

passwordstring

IAM password (PASSWORD method only)

Password for the IAM user identified by username. Required when authMethod is PASSWORD. Sensitive — always provide via {{ secret('NAME') }}.

projectNamestring

Project name for project-scoped tokens (PASSWORD method only)

Overrides the project name used for scope=PROJECT token requests. Defaults to the task's region value when omitted, which is correct for most regions.

scopestring
DefaultPROJECT
Possible Values
PROJECTDOMAIN

Token scope (PASSWORD method only)

Scope of the session token obtained during password authentication.

  • PROJECT (default): token is scoped to the project matching projectName (or the task's region when projectName is omitted). Use for most downstream tasks.
  • DOMAIN: token is scoped to the domain.
usernamestring

IAM username (PASSWORD method only)

Huawei Cloud IAM username. Required when authMethod is PASSWORD.

SASL username

Required when saslMechanism is PLAIN or SCRAM_SHA_512.

DefaultSTRING
Possible Values
STRINGJSONBINARY

Value serializer/deserializer type

STRING (default), JSON, or BINARY.

Number of records consumed from the DMS Kafka topic

Formaturi

URI of the ION file in Kestra internal storage containing the consumed messages

Unitrecords

Number of records consumed from the DMS Kafka topic.

More Plugins in this Category

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

OPA (Open Policy Agent)
OPA (Open Policy Agent)
Enterprise

Tasks to manage Open Policy Agent policies and decisions, including upload, evaluation, and partial evaluation.

Infrastructure

5 tasks

3 blueprints