Skip to content

Commit

Permalink
Add rocketmq source and sink
Browse files Browse the repository at this point in the history
  • Loading branch information
@sunxiaojian
sunxiaojian committed Feb 6, 2023
1 parent 6644194 commit f7f08cb
Show file tree
Hide file tree
Showing 46 changed files with 4,065 additions and 0 deletions.
80 changes: 80 additions & 0 deletions docs/en/connector-v2/sink/RocketMQ.md
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
# RocketMQ

> RocketMQ sink connector
## Description

Write Rows to a Apache RocketMQ topic.

## Key features

- [x] [exactly-once](../../concept/connector-v2-features.md)

By default, we will use 2pc to guarantee the message is sent to RocketMQ exactly once.

## Options

| name | type | required | default value |
|----------------------|---------|----------|--------------------------|
| topic | string | yes | - |
| name.srv.addr | string | yes | - |
| acl.enabled | Boolean | no | false |
| access.key | String | no | |
| secret.key | String | no | |
| producer.group | String | no | SeaTunnel-producer-Group |
| semantic | string | no | NON |
| partition.key.fields | array | no | - |
| format | String | no | json |
| field.delimiter | String | no | , |
| common-options | config | no | - |

### topic [string]

`RocketMQ topic` name.

### name.srv.addr [string]

`RocketMQ` name server cluster address.

### semantic [string]

Semantics that can be chosen EXACTLY_ONCE/AT_LEAST_ONCE/NON, default NON.

### partition.key.fields [array]

Configure which fields are used as the key of the RocketMQ message.

For example, if you want to use value of fields from upstream data as key, you can assign field names to this property.

Upstream data is the following:

| name | age | data |
|------|-----|---------------|
| Jack | 16 | data-example1 |
| Mary | 23 | data-example2 |

If name is set as the key, then the hash value of the name column will determine which partition the message is sent to.

### format

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

Customize the field delimiter for data format.

### common options [config]

Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details.

## Examples

```hocon
sink {
Rocketmq {
name.srv.addr = "localhost:9876"
topic = "test-topic-003"
partition.key.fields = ["name"]
}
}
```
141 changes: 141 additions & 0 deletions docs/en/connector-v2/source/RocketMQ.md
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
# RocketMQ

> RocketMQ source connector
## Description

Source connector for Apache RocketMQ.

## Key features

- [x] [batch](../../concept/connector-v2-features.md)
- [x] [stream](../../concept/connector-v2-features.md)
- [x] [exactly-once](../../concept/connector-v2-features.md)
- [ ] [column projection](../../concept/connector-v2-features.md)
- [x] [parallelism](../../concept/connector-v2-features.md)
- [ ] [support user-defined split](../../concept/connector-v2-features.md)

## Options

| name | type | required | default value |
|-------------------------------------|---------|----------|-----------------------------|
| topics | String | yes | - |
| name.srv.addr | String | yes | - |
| acl.enabled | Boolean | no | false |
| access.key | String | no | |
| secret.key | String | no | |
| batch.size | int | no | 100 |
| consumer.group | String | no | SeaTunnel-Consumer-Group |
| commit.on.checkpoint | Boolean | no | true |
| schema | | no | - |
| format | String | no | json |
| field.delimiter | String | no | , |
| start.mode | String | no | CONSUME_FROM_GROUP_OFFSETS |
| start.mode.offsets | | no | |
| start.mode.timestamp | Long | no | |
| partition.discovery.interval.millis | long | no | -1 |
| common-options | config | no | - |

### topics [string]

`RocketMQ topic` name. If there are multiple `topics`, use `,` to split, for example: `"tpc1,tpc2"`.

### name.srv.addr [string]

`RocketMQ` name server cluster address.

### consumer.group [string]

`RocketMQ consumer group id`, used to distinguish different consumer groups.

### acl.enabled [boolean]

If true, access control is enabled, and access key and secret key need to be configured.

### access.key [string]

When ACL_ENABLED is true, access key cannot be empty.

### secret.key [string]

When ACL_ENABLED is true, secret key cannot be empty.

### batch.size [int]

`RocketMQ` consumer pull batch size

### commit.on.checkpoint [boolean]

If true the consumer's offset will be periodically committed in the background.

## partition.discovery.interval.millis [long]

The interval for dynamically discovering topics and partitions.

### schema

The structure of the data, including field names and field types.

## format

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

Customize the field delimiter for data format.

## start.mode

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.timestamp

The time required for consumption mode to be "CONSUME_FROM_TIMESTAMP".

## start.mode.offsets

The offset required for consumption mode to be "CONSUME_FROM_SPECIFIC_OFFSETS".

for example:

```hocon
start.mode.offsets = {
topic1-0 = 70
topic1-1 = 10
topic1-2 = 10
}
```

### common-options [config]

Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details.


## Example

### Simple

```hocon
source {
Rocketmq {
name.srv.addr = "localhost:9876"
topics = "test-topic-002"
consumer.group = "consumer-group"
parallelism = 2
batch.size = 20
schema = {
fields {
age = int
name = string
}
}
start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
start.mode.offsets = {
test-topic-002-0 = 20
}
}
}
```
2 changes: 2 additions & 0 deletions plugin-mapping.properties
View file
Original file line number Diff line number Diff line change
Expand Up @@ -170,4 +170,6 @@ seatunnel.source.TDengine = connector-tdengine
seatunnel.sink.TDengine = connector-tdengine
seatunnel.source.Persistiq = connector-http-persistiq
seatunnel.sink.SelectDBCloud = connector-selectdb-cloud
seatunnel.source.Rocketmq = connector-rocketmq
seatunnel.sink.Rocketmq = connector-rocketmq

62 changes: 62 additions & 0 deletions seatunnel-connectors-v2/connector-rocketmq/pom.xml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>seatunnel-connectors-v2</artifactId>
<groupId>org.apache.seatunnel</groupId>
<version>${revision}</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>connector-rocketmq</artifactId>

<properties>
<rocketmq.version>4.9.4</rocketmq.version>
</properties>
<dependencies>

<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>connector-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>seatunnel-format-json</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>seatunnel-format-text</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>${rocketmq.version}</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-tools</artifactId>
<version>${rocketmq.version}</version>
</dependency>
</dependencies>
</project>
Loading

0 comments on commit f7f08cb

Please sign in to comment.