Version: Next

RocketMQ

RocketMQ source connector

Support Apache RocketMQ Version

4.9.0 or newer

Support These Engines

Spark
Flink
SeaTunnel Zeta

Key Features

Description

Reads messages from Apache RocketMQ topics. The connector can read one or more topics with one schema, or use tables_configs to read multiple topics with different schemas.

Source Options

Name	Type	Required	Default	Description
name.srv.addr	String	yes	-	RocketMQ name server address, for example `localhost:9876`.
topics	String	no	-	Topic name list separated by commas, for example `"topic_a,topic_b"`. Configure only one of `topics`, `tables_configs`, and `table_list`.
tables_configs	List	no	-	Multi-table read configuration. Each item must contain `topics` and can contain `format`, `schema`, `tags`, `start.mode`, `start.mode.timestamp`, `start.mode.offsets`, and `ignore_parse_errors`.
table_list	List	no	-	Deprecated. Use `tables_configs` instead.
tags	String	no	-	Tag list separated by commas. Only messages whose RocketMQ tag exactly matches one configured value are consumed.
acl.enabled	Boolean	no	false	Whether to enable RocketMQ ACL authentication.
access.key	String	no	-	Access key. Required when `acl.enabled` is `true`.
secret.key	String	no	-	Secret key. Required when `acl.enabled` is `true`.
batch.size	int	no	100	Maximum number of messages pulled each time.
consumer.group	String	no	SeaTunnel-Consumer-Group	RocketMQ consumer group ID.
commit.on.checkpoint	Boolean	no	true	Whether to commit offsets when SeaTunnel checkpoints are completed.
schema	config	no	-	Message schema. See Schema Feature. If omitted, the connector reads message bodies as text.
format	String	no	json	Message format. Supported values are `json` and `text`.
field.delimiter	String	no	`,`	Field delimiter used when `format = text`.
start.mode	String	no	CONSUME_FROM_GROUP_OFFSETS	Startup position. Supported values: `CONSUME_FROM_LAST_OFFSET`, `CONSUME_FROM_FIRST_OFFSET`, `CONSUME_FROM_GROUP_OFFSETS`, `CONSUME_FROM_TIMESTAMP`, `CONSUME_FROM_SPECIFIC_OFFSETS`.
start.mode.offsets	Map	no	-	Required when `start.mode = CONSUME_FROM_SPECIFIC_OFFSETS`. The key format is `topic-queueId`, for example `test_topic-0`.
start.mode.timestamp	Long	no	-	Required when `start.mode = CONSUME_FROM_TIMESTAMP`. Use a millisecond timestamp.
partition.discovery.interval.millis	long	no	-1	Topic and partition discovery interval in milliseconds. `-1` disables dynamic discovery.
ignore_parse_errors	Boolean	no	false	Whether to skip JSON messages that cannot be parsed.
consumer.poll.timeout.millis	long	no	5000	Pull timeout in milliseconds.
common-options	config	no	-	Source common options. See Source Common Options.

Option Notes

Startup Position

start.mode controls where the source starts reading:

CONSUME_FROM_GROUP_OFFSETS: start from committed offsets of the consumer group.
CONSUME_FROM_FIRST_OFFSET: start from the earliest available offset.
CONSUME_FROM_LAST_OFFSET: start from the latest available offset.
CONSUME_FROM_TIMESTAMP: start from the first offset at or after start.mode.timestamp.
CONSUME_FROM_SPECIFIC_OFFSETS: start from the offsets in start.mode.offsets.

When start.mode = CONSUME_FROM_TIMESTAMP, start.mode.timestamp must be a non-negative millisecond timestamp and cannot be later than the current time of the running job.

start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
start.mode.offsets = {
  test_topic-0 = 50
}

start.mode = "CONSUME_FROM_TIMESTAMP"
start.mode.timestamp = 1667179890315

Message Format

When format = json, define schema so that SeaTunnel can parse the JSON message body into typed fields. ignore_parse_errors = true can be used to skip invalid JSON messages instead of failing the job.

When format = text, SeaTunnel splits the message body by field.delimiter and maps the values to fields in schema order. If schema is omitted, the message body is read as a single text value.

Multi-Table Read

Use tables_configs when different topics have different schemas. Each item must contain topics and can define its own schema, format, tags, and startup position. If schema.table is not set, the output table name defaults to the topic name.

topics, tables_configs, and the deprecated table_list are mutually exclusive. In tables_configs, options that are not set on an item inherit the top-level defaults, so each item only needs to override the topic-specific schema, tags, or startup position.

When a tables_configs item uses start.mode = CONSUME_FROM_TIMESTAMP, it must also set start.mode.timestamp. When it uses start.mode = CONSUME_FROM_SPECIFIC_OFFSETS, it must also set a non-empty start.mode.offsets map.

Task Examples

Read JSON Messages

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topics = "test_topic_json"
    plugin_output = "rocketmq_table"
    format = json
    schema = {
      fields {
        id = bigint
        c_string = string
        c_int = int
        c_timestamp = timestamp
      }
    }
  }
}

sink {
  Console {
    plugin_input = "rocketmq_table"
  }
}

Read Text Messages With Tags

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topics = "test_topic_text"
    plugin_output = "rocketmq_table"
    format = text
    field.delimiter = ","
    tags = "tag_a,tag_b"
    schema = {
      fields {
        id = bigint
        content = string
      }
    }
  }
}

sink {
  Console {
    plugin_input = "rocketmq_table"
  }
}

Read From Specific Offsets

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topics = "test_topic_source"
    plugin_output = "rocketmq_table"
    format = json
    start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
    start.mode.offsets = {
      test_topic_source-0 = 50
    }
    schema = {
      fields {
        id = bigint
      }
    }
  }
}

sink {
  Console {
    plugin_input = "rocketmq_table"
  }
}

Read Multiple Topics With Different Schemas

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    start.mode = "CONSUME_FROM_LAST_OFFSET"
    tables_configs = [
      {
        topics = "test_topic_multi_a"
        start.mode = "CONSUME_FROM_FIRST_OFFSET"
        format = json
        schema = {
          fields {
            id = bigint
            c_string = string
          }
        }
      },
      {
        topics = "test_topic_multi_b"
        start.mode = "CONSUME_FROM_FIRST_OFFSET"
        tags = "tag_b"
        format = json
        schema = {
          table = "rocketmq_multi_custom"
          fields {
            id = bigint
            description = string
          }
        }
      }
    ]
  }
}

sink {
  Console {}
}

Read From a Timestamp

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topics = "test_topic_source"
    plugin_output = "rocketmq_table"
    format = json
    start.mode = "CONSUME_FROM_TIMESTAMP"
    start.mode.timestamp = 1667179890315
    schema = {
      fields {
        id = bigint
      }
    }
  }
}

sink {
  Console {
    plugin_input = "rocketmq_table"
  }
}

Changelog

Change Log

Change	Commit	Version
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)	https://github.com/apache/seatunnel/commit/9212a77140	2.3.12
[improve] rocketmq options (#9251)	https://github.com/apache/seatunnel/commit/4cbe3b9172	2.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[Improve][Connector-V2] RocketMQ Source add message tag config (#8825)	https://github.com/apache/seatunnel/commit/5913e8c35f	2.3.10
[Improve][Connector-V2] Add optional flag for rocketmq connector to skip parse errors instead of failing (#8737)	https://github.com/apache/seatunnel/commit/701f17b5d4	2.3.10
[Improve][Connector-V2] RocketMQ Sink add message tag config (#7996)	https://github.com/apache/seatunnel/commit/97a1b00e48	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Fix][Connector-V2] Fix some throwable error not be caught (#7657)	https://github.com/apache/seatunnel/commit/e19d73282e	2.3.8
[Feature][Kafka] Support multi-table source read (#5992)	https://github.com/apache/seatunnel/commit/60104602d1	2.3.6
[Fix][connector-rocketmq] commit a correct offset to broker & reduce ThreadInterruptedException log (#6668)	https://github.com/apache/seatunnel/commit/b7480e1a89	2.3.6
[fix][connector-rocketmq]Fix a NPE problem when checkpoint.interval is set too small(#6624) (#6625)	https://github.com/apache/seatunnel/commit/6e0c81d492	2.3.5
[Test][E2E] Add thread leak check for connector (#5773)	https://github.com/apache/seatunnel/commit/1f2f3fc5f0	2.3.4
[Fix][Connector] Rocketmq source startOffset greater than endOffset error (#6287)	https://github.com/apache/seatunnel/commit/cd44b5894e	2.3.4
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
[Improve] Remove use `SeaTunnelSink::getConsumedType` method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)	https://github.com/apache/seatunnel/commit/51c0d709ba	2.3.4
[Improve][pom] Formatting pom (#4761)	https://github.com/apache/seatunnel/commit/1d6d3815ec	2.3.2
[Hotfix][Connector-V2][RocketMQ] Fix rocketmq spark e2e test cases (#4583)	https://github.com/apache/seatunnel/commit/e711f6ef4c	2.3.2
[Feature][Connector-V2] Add rocketmq source and sink (#4007)	https://github.com/apache/seatunnel/commit/e333897552	2.3.2

RocketMQ

Support Apache RocketMQ Version​

Support These Engines​

Key Features​

Description​

Source Options​

Option Notes​

Startup Position​

Message Format​

Tags​

Multi-Table Read​

Task Examples​

Read JSON Messages​

Read Text Messages With Tags​

Read From Specific Offsets​

Read Multiple Topics With Different Schemas​

Read From a Timestamp​

Changelog​