Skip to main content
Version: Next

RocketMQ

RocketMQ sink connector

Support Apache RocketMQ Version​

  • 4.9.0 or newer

Support These Engines​

Spark
Flink
SeaTunnel Zeta

Key Features​

Description​

Writes SeaTunnel rows to an Apache RocketMQ topic. The sink supports JSON and text message bodies, optional message tags, synchronous sending, partition key fields, and transactional messages when exactly.once = true.

Sink Options​

NameTypeRequiredDefaultDescription
topicStringyes-RocketMQ topic name.
name.srv.addrStringyes-RocketMQ name server address, for example localhost:9876.
acl.enabledBooleannofalseWhether to enable RocketMQ ACL authentication.
access.keyStringno-Access key. Required when acl.enabled is true.
secret.keyStringno-Secret key. Required when acl.enabled is true.
producer.groupStringnoSeaTunnel-Producer-GroupRocketMQ producer group ID.
tagStringno-RocketMQ message tag written with each message.
partition.key.fieldsListno-Field names serialized as the RocketMQ message key. Every listed field must exist in the upstream schema.
formatStringnojsonMessage format. Supported values are json and text.
field.delimiterStringno,Field delimiter used when format = text.
producer.send.syncBooleannofalseWhether to send messages synchronously. When false, messages are sent asynchronously.
exactly.onceBooleannofalseWhether to send transactional messages for exactly-once delivery.
max.message.sizeintno4194304Maximum message body size in bytes.
send.message.timeoutintno3000Send timeout in milliseconds.
common-optionsconfigno-Sink common options. See Sink Common Options.

Option Notes​

partition.key.fields​

partition.key.fields controls the RocketMQ message key. SeaTunnel serializes the configured field values as JSON and writes that value to Message.keys. For non-transactional sends, the same key is also used by RocketMQ's hash queue selector, so rows with the same key are sent to the same queue. If this option is not set, RocketMQ chooses the queue.

For example, if the input schema contains c_int, this configuration uses c_int to build the message key:

partition.key.fields = ["c_int"]

exactly.once​

The sink supports exactly-once writes through RocketMQ transactional messages. This behavior is disabled by default. Set exactly.once = true when the RocketMQ cluster and the job checkpoint settings are ready for transactional writes.

When format = text, SeaTunnel serializes fields in the upstream schema order and joins them with field.delimiter. When format = json, each row is written as a JSON object.

producer.send.sync​

producer.send.sync = true makes the producer wait for RocketMQ to acknowledge each send request. With the default value false, messages are sent asynchronously.

Task Examples​

Write JSON Messages​

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

source {
  FakeSource {
    row.num = 10
    schema = {
      fields {
        c_string = string
        c_int = int
        c_timestamp = timestamp
      }
    }
  }
}

sink {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topic = "test_topic"
    partition.key.fields = ["c_int"]
    producer.send.sync = true
  }
}

Write Text Messages​

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

source {
  FakeSource {
    row.num = 10
    schema = {
      fields {
        id = bigint
        content = string
      }
    }
  }
}

sink {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topic = "test_text_topic"
    format = text
    field.delimiter = ","
    producer.send.sync = true
  }
}

Write Messages With a Tag​

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

source {
  FakeSource {
    row.num = 10
    schema = {
      fields {
        c_string = string
        c_int = int
      }
    }
  }
}

sink {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topic = "test_topic_message_tag"
    tag = "test_tag"
    partition.key.fields = ["c_string"]
    producer.send.sync = true
  }
}

Read and Write RocketMQ​

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topics = "test_topic_source"
    plugin_output = "rocketmq_table"
    format = json
    start.mode = "CONSUME_FROM_FIRST_OFFSET"
    consumer.group = "rocketmq_to_rocketmq_group"
    schema = {
      fields {
        id = bigint
        c_string = string
      }
    }
  }
}

sink {
  Rocketmq {
    plugin_input = "rocketmq_table"
    name.srv.addr = "rocketmq-e2e:9876"
    topic = "test_topic_sink"
    partition.key.fields = ["id"]
    exactly.once = true
  }
}

Changelog​

Change Log
ChangeCommitVersion
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)https://github.com/apache/seatunnel/commit/9212a771402.3.12
[improve] rocketmq options (#9251)https://github.com/apache/seatunnel/commit/4cbe3b91722.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Improve][Connector-V2] RocketMQ Source add message tag config (#8825)https://github.com/apache/seatunnel/commit/5913e8c35f2.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/701f17b5d42.3.10
[Improve][Connector-V2] RocketMQ Sink add message tag config (#7996)https://github.com/apache/seatunnel/commit/97a1b00e482.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Fix][Connector-V2] Fix some throwable error not be caught (#7657)https://github.com/apache/seatunnel/commit/e19d73282e2.3.8
[Feature][Kafka] Support multi-table source read (#5992)https://github.com/apache/seatunnel/commit/60104602d12.3.6
[Fix][connector-rocketmq] commit a correct offset to broker & reduce ThreadInterruptedException log (#6668)https://github.com/apache/seatunnel/commit/b7480e1a892.3.6
[fix][connector-rocketmq]Fix a NPE problem when checkpoint.interval is set too small(#6624) (#6625)https://github.com/apache/seatunnel/commit/6e0c81d4922.3.5
[Test][E2E] Add thread leak check for connector (#5773)https://github.com/apache/seatunnel/commit/1f2f3fc5f02.3.4
[Fix][Connector] Rocketmq source startOffset greater than endOffset error (#6287)https://github.com/apache/seatunnel/commit/cd44b5894e2.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)https://github.com/apache/seatunnel/commit/51c0d709ba2.3.4
[Improve][pom] Formatting pom (#4761)https://github.com/apache/seatunnel/commit/1d6d3815ec2.3.2
[Hotfix][Connector-V2][RocketMQ] Fix rocketmq spark e2e test cases (#4583)https://github.com/apache/seatunnel/commit/e711f6ef4c2.3.2
[Feature][Connector-V2] Add rocketmq source and sink (#4007)https://github.com/apache/seatunnel/commit/e3338975522.3.2