RocketMQ source connector

Support Apache RocketMQ Version

  • 4.9.0 (Or a newer version, for reference)

Support These Engines

Spark
Flink
SeaTunnel Zeta

Key Features

Description

Source connector for Apache RocketMQ.

Source Options

Name Type Required Default Description
topics String yes - RocketMQ topic name. If there are multiple topics, use , to split, for example: "tpc1,tpc2".
name.srv.addr String yes - RocketMQ name server cluster address.
acl.enabled Boolean no false If true, access control is enabled, and access key and secret key need to be configured.
access.key String no
secret.key String no When ACL_ENABLED is true, secret key cannot be empty.
batch.size int no 100 RocketMQ consumer pull batch size
consumer.group String no SeaTunnel-Consumer-Group RocketMQ consumer group id, used to distinguish different consumer groups.
commit.on.checkpoint Boolean no true If true the consumer’s offset will be periodically committed in the background.
schema no - The structure of the data, including field names and field types.
format String no json Data format. The default format is json. Optional text format. The default field separator is “,”.If you customize the delimiter, add the “field.delimiter” option.
field.delimiter String no , Customize the field delimiter for data format
start.mode String no CONSUME_FROM_GROUP_OFFSETS The initial consumption pattern of consumers,there are several types: [CONSUME_FROM_LAST_OFFSET],[CONSUME_FROM_FIRST_OFFSET],[CONSUME_FROM_GROUP_OFFSETS],[CONSUME_FROM_TIMESTAMP],[CONSUME_FROM_SPECIFIC_OFFSETS]
start.mode.offsets no
start.mode.timestamp Long no The time required for consumption mode to be “CONSUME_FROM_TIMESTAMP”.
partition.discovery.interval.millis long no -1 The interval for dynamically discovering topics and partitions.
common-options config no - Source plugin common parameters, please refer to Source Common Options for details.

start.mode.offsets

The offset required for consumption mode to be “CONSUME_FROM_SPECIFIC_OFFSETS”.

for example:

  1. start.mode.offsets = {
  2.   topic1-0 = 70
  3.   topic1-1 = 10
  4.   topic1-2 = 10
  5. }

Task Example

Simple:

Consumer reads Rocketmq data and prints it to the console type

  1. env {
  2.   parallelism = 1
  3.   job.mode = "BATCH"
  4. }
  6. source {
  7.   Rocketmq {
  8.     name.srv.addr = "rocketmq-e2e:9876"
  9.     topics = "test_topic_json"
  10.     result_table_name = "rocketmq_table"
  11.     schema = {
  12.       fields {
  13.         id = bigint
  14.         c_map = "map<string, smallint>"
  15.         c_array = "array<tinyint>"
  16.         c_string = string
  17.         c_boolean = boolean
  18.         c_tinyint = tinyint
  19.         c_smallint = smallint
  20.         c_int = int
  21.         c_bigint = bigint
  22.         c_float = float
  23.         c_double = double
  24.         c_decimal = "decimal(2, 1)"
  25.         c_bytes = bytes
  26.         c_date = date
  27.         c_timestamp = timestamp
  28.       }
  29.     }
  30.   }
  31. }
  33. transform {
  34.   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
  35.   # please go to https://seatunnel.apache.org/docs/category/transform
  36. }
  38. sink {
  39.   Console {
  40.   }
  41. }

Specified format consumption Simple:

When I consume the topic data in json format parsing and pulling the number of bars each time is 400, the consumption starts from the original location

  1. env {
  2.   parallelism = 1
  3.   job.mode = "BATCH"
  4. }
  6. source {
  7.   Rocketmq {
  8.     name.srv.addr = "localhost:9876"
  9.     topics = "test_topic"
  10.     result_table_name = "rocketmq_table"
  11.     start.mode = "CONSUME_FROM_FIRST_OFFSET"
  12.     batch.size = "400"
  13.     consumer.group = "test_topic_group"
  14.     format = "json"
  15.     format = json
  16.     schema = {
  17.       fields {
  18.         c_map = "map<string, string>"
  19.         c_array = "array<int>"
  20.         c_string = string
  21.         c_boolean = boolean
  22.         c_tinyint = tinyint
  23.         c_smallint = smallint
  24.         c_int = int
  25.         c_bigint = bigint
  26.         c_float = float
  27.         c_double = double
  28.         c_decimal = "decimal(30, 8)"
  29.         c_bytes = bytes
  30.         c_date = date
  31.         c_timestamp = timestamp
  32.       }
  33.     }
  34.   }
  35. }
  37. transform {
  38.   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
  39.   # please go to https://seatunnel.apache.org/docs/category/transform
  40. }
  41. sink {
  42.   Console {
  43.   }
  44. }

Specified timestamp Simple:

This is to specify a time to consume, and I dynamically sense the existence of a new partition every 1000 milliseconds to pull the consumption

  1. env {
  2.   parallelism = 1
  3.   spark.app.name = "SeaTunnel"
  4.   spark.executor.instances = 2
  5.   spark.executor.cores = 1
  6.   spark.executor.memory = "1g"
  7.   spark.master = local
  8.   job.mode = "BATCH"
  9. }
  11. source {
  12.   Rocketmq {
  13.     name.srv.addr = "localhost:9876"
  14.     topics = "test_topic"
  15.     partition.discovery.interval.millis = "1000"
  16.     start.mode.timestamp="1694508382000"
  17.     consumer.group="test_topic_group"
  18.     format="json"
  19.     format = json
  20.     schema = {
  21.       fields {
  22.         c_map = "map<string, string>"
  23.         c_array = "array<int>"
  24.         c_string = string
  25.         c_boolean = boolean
  26.         c_tinyint = tinyint
  27.         c_smallint = smallint
  28.         c_int = int
  29.         c_bigint = bigint
  30.         c_float = float
  31.         c_double = double
  32.         c_decimal = "decimal(30, 8)"
  33.         c_bytes = bytes
  34.         c_date = date
  35.         c_timestamp = timestamp
  36.       }
  37.     }
  38.   }
  39. }
  41. transform {
  42.   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
  43.   # please go to https://seatunnel.apache.org/docs/category/transform
  44. }
  46. sink {
  47.   Console {
  48.   }
  49. }