This is an automated email from the ASF dual-hosted git repository.
wanghailin pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 97a1b00e48 [Improve][Connector-V2] RocketMQ Sink add message tag
config (#7996)
97a1b00e48 is described below
commit 97a1b00e481d23314a9b498556653ab34d8764c7
Author: wengys <[email protected]>
AuthorDate: Wed Nov 13 18:46:59 2024 +0800
[Improve][Connector-V2] RocketMQ Sink add message tag config (#7996)
---
docs/en/connector-v2/sink/RocketMQ.md | 1 +
.../seatunnel/rocketmq/config/ProducerConfig.java | 6 ++
.../serialize/DefaultSeaTunnelRowSerializer.java | 9 ++-
.../seatunnel/rocketmq/sink/ProducerMetadata.java | 2 +
.../seatunnel/rocketmq/sink/RocketMqSink.java | 3 +
.../rocketmq/sink/RocketMqSinkWriter.java | 1 +
.../rocketmq/RocketMqConsumerMessage.java | 30 ++++++++++
.../e2e/connector/rocketmq/RocketMqIT.java | 40 ++++++++++---
...rocketmq-sink_fake_to_rocketmq_message_tag.conf | 70 ++++++++++++++++++++++
9 files changed, 154 insertions(+), 8 deletions(-)
diff --git a/docs/en/connector-v2/sink/RocketMQ.md
b/docs/en/connector-v2/sink/RocketMQ.md
index f0672a3c7f..589aa49699 100644
--- a/docs/en/connector-v2/sink/RocketMQ.md
+++ b/docs/en/connector-v2/sink/RocketMQ.md
@@ -32,6 +32,7 @@ Write Rows to a Apache RocketMQ topic.
| access.key | String | no | | When
ACL_ENABLED is true, access key cannot be empty
|
| secret.key | String | no | | When
ACL_ENABLED is true, secret key cannot be empty
|
| producer.group | String | no | SeaTunnel-producer-Group |
SeaTunnel-producer-Group
|
+| tag | String | no | - |
`RocketMQ` message tag.
|
| partition.key.fields | array | no | - | -
|
| 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.
|
diff --git
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/config/ProducerConfig.java
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/config/ProducerConfig.java
index 134a9908fa..9b9b8d0b07 100644
---
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/config/ProducerConfig.java
+++
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/config/ProducerConfig.java
@@ -32,6 +32,12 @@ public class ProducerConfig extends Config {
.noDefaultValue()
.withDescription("RocketMq topic name. ");
+ public static final Option<String> TAG =
+ Options.key("tag")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("RocketMq message tag.");
+
public static final Option<String> PRODUCER_GROUP =
Options.key("producer.group")
.stringType()
diff --git
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/serialize/DefaultSeaTunnelRowSerializer.java
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/serialize/DefaultSeaTunnelRowSerializer.java
index 558dfb96fc..b4d77fb76b 100644
---
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/serialize/DefaultSeaTunnelRowSerializer.java
+++
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/serialize/DefaultSeaTunnelRowSerializer.java
@@ -37,37 +37,44 @@ import java.util.function.Function;
@Slf4j
public class DefaultSeaTunnelRowSerializer implements
SeaTunnelRowSerializer<byte[], byte[]> {
private final String topic;
+ private final String tag;
private final SerializationSchema keySerialization;
private final SerializationSchema valueSerialization;
public DefaultSeaTunnelRowSerializer(
String topic,
+ String tag,
SeaTunnelRowType seaTunnelRowType,
SchemaFormat format,
String delimiter) {
this(
topic,
+ tag,
element -> null,
createSerializationSchema(seaTunnelRowType, format,
delimiter));
}
public DefaultSeaTunnelRowSerializer(
String topic,
+ String tag,
List<String> keyFieldNames,
SeaTunnelRowType seaTunnelRowType,
SchemaFormat format,
String delimiter) {
this(
topic,
+ tag,
createKeySerializationSchema(keyFieldNames, seaTunnelRowType),
createSerializationSchema(seaTunnelRowType, format,
delimiter));
}
public DefaultSeaTunnelRowSerializer(
String topic,
+ String tag,
SerializationSchema keySerialization,
SerializationSchema valueSerialization) {
this.topic = topic;
+ this.tag = tag;
this.keySerialization = keySerialization;
this.valueSerialization = valueSerialization;
}
@@ -123,6 +130,6 @@ public class DefaultSeaTunnelRowSerializer implements
SeaTunnelRowSerializer<byt
return null;
}
byte[] key = keySerialization.serialize(row);
- return new Message(topic, null, key == null ? null : new String(key),
value);
+ return new Message(topic, tag, key == null ? null : new String(key),
value);
}
}
diff --git
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/ProducerMetadata.java
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/ProducerMetadata.java
index 34add2fd50..85c1433815 100644
---
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/ProducerMetadata.java
+++
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/ProducerMetadata.java
@@ -31,6 +31,8 @@ public class ProducerMetadata implements Serializable {
private RocketMqBaseConfiguration configuration;
/** send topic */
private String topic;
+ /** message tag */
+ private String tag;
/** partition key fields */
private List<String> partitionKeyFields;
diff --git
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/RocketMqSink.java
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/RocketMqSink.java
index 9fda05dece..2bd55150c0 100644
---
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/RocketMqSink.java
+++
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/RocketMqSink.java
@@ -74,6 +74,9 @@ public class RocketMqSink extends
AbstractSimpleSink<SeaTunnelRow, Void> {
}
producerMetadata = new ProducerMetadata();
producerMetadata.setTopic(config.getString(ProducerConfig.TOPIC.key()));
+ if (config.hasPath(ProducerConfig.TAG.key())) {
+
producerMetadata.setTag(config.getString(ProducerConfig.TAG.key()));
+ }
RocketMqBaseConfiguration.Builder baseConfigurationBuilder =
RocketMqBaseConfiguration.newBuilder()
.producer()
diff --git
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/RocketMqSinkWriter.java
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/RocketMqSinkWriter.java
index b8cbaf70da..8e3bd4d1a0 100644
---
a/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/RocketMqSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/sink/RocketMqSinkWriter.java
@@ -78,6 +78,7 @@ public class RocketMqSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void> {
SeaTunnelRowType seaTunnelRowType) {
return new DefaultSeaTunnelRowSerializer(
producerMetadata.getTopic(),
+ producerMetadata.getTag(),
getPartitionKeyFields(seaTunnelRowType),
seaTunnelRowType,
producerMetadata.getFormat(),
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqConsumerMessage.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqConsumerMessage.java
new file mode 100644
index 0000000000..594f193df1
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqConsumerMessage.java
@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.e2e.connector.rocketmq;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+@AllArgsConstructor
+public class RocketMqConsumerMessage {
+ private String value;
+ private String tag;
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
index 01d7c77683..a07818476b 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
@@ -147,6 +147,7 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
DefaultSeaTunnelRowSerializer serializer =
new DefaultSeaTunnelRowSerializer(
"test_topic_source",
+ null,
SEATUNNEL_ROW_TYPE,
DEFAULT_FORMAT,
DEFAULT_FIELD_DELIMITER);
@@ -183,7 +184,7 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
String topicName = "test_topic";
- Map<String, String> data = getRocketMqConsumerData(topicName);
+ Map<String, RocketMqConsumerMessage> data =
getRocketMqConsumerData(topicName);
ObjectMapper objectMapper = new ObjectMapper();
String key = data.keySet().iterator().next();
ObjectNode objectNode = objectMapper.readValue(key, ObjectNode.class);
@@ -199,7 +200,7 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
container.executeJob("/rocketmq-text-sink_fake_to_rocketmq.conf");
Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
String topicName = "test_text_topic";
- Map<String, String> data = getRocketMqConsumerData(topicName);
+ Map<String, RocketMqConsumerMessage> data =
getRocketMqConsumerData(topicName);
Assertions.assertEquals(10, data.size());
}
@@ -209,6 +210,7 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
DefaultSeaTunnelRowSerializer serializer =
new DefaultSeaTunnelRowSerializer(
"test_topic_text",
+ null,
SEATUNNEL_ROW_TYPE,
SchemaFormat.TEXT,
DEFAULT_FIELD_DELIMITER);
@@ -228,6 +230,7 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
DefaultSeaTunnelRowSerializer serializer =
new DefaultSeaTunnelRowSerializer(
"test_topic_text_offset_check",
+ null,
SEATUNNEL_ROW_TYPE,
SchemaFormat.TEXT,
DEFAULT_FIELD_DELIMITER);
@@ -245,6 +248,7 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
DefaultSeaTunnelRowSerializer serializer =
new DefaultSeaTunnelRowSerializer(
"test_topic_json",
+ null,
SEATUNNEL_ROW_TYPE,
DEFAULT_FORMAT,
DEFAULT_FIELD_DELIMITER);
@@ -292,6 +296,7 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
DefaultSeaTunnelRowSerializer serializer =
new DefaultSeaTunnelRowSerializer(
"test_topic_group",
+ null,
SEATUNNEL_ROW_TYPE,
DEFAULT_FORMAT,
DEFAULT_FIELD_DELIMITER);
@@ -299,6 +304,25 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
testRocketMqGroupOffsetsToConsole(container);
}
+ @TestTemplate
+ public void testSinkRocketMqMessageTag(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
+
container.executeJob("/rocketmq-sink_fake_to_rocketmq_message_tag.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+
+ String topicName = "test_topic_message_tag";
+ String tag = "test_tag";
+ Map<String, RocketMqConsumerMessage> data =
getRocketMqConsumerData(topicName);
+ ObjectMapper objectMapper = new ObjectMapper();
+ String key = data.keySet().iterator().next();
+ ObjectNode objectNode = objectMapper.readValue(key, ObjectNode.class);
+ Assertions.assertTrue(objectNode.has("c_map"));
+ Assertions.assertTrue(objectNode.has("c_string"));
+ Assertions.assertEquals(10, data.size());
+ Assertions.assertEquals(tag, data.get(key).getTag());
+ }
+
public void testRocketMqGroupOffsetsToConsole(TestContainer container)
throws IOException, InterruptedException {
Container.ExecResult execResult =
@@ -334,8 +358,8 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
}
}
- private Map<String, String> getRocketMqConsumerData(String topicName) {
- Map<String, String> data = new HashMap<>();
+ private Map<String, RocketMqConsumerMessage>
getRocketMqConsumerData(String topicName) {
+ Map<String, RocketMqConsumerMessage> data = new HashMap<>();
try {
DefaultLitePullConsumer consumer =
RocketMqAdminUtil.initDefaultLitePullConsumer(newConfiguration(), false);
@@ -373,9 +397,11 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
break;
}
for (MessageExt message : messages) {
- data.put(
- message.getKeys(),
- new String(message.getBody(),
StandardCharsets.UTF_8));
+ RocketMqConsumerMessage consumerMessage =
+ new RocketMqConsumerMessage(
+ new String(message.getBody(),
StandardCharsets.UTF_8),
+ message.getTags());
+ data.put(message.getKeys(), consumerMessage);
consumer.getOffsetStore()
.updateConsumeOffsetToBroker(
new MessageQueue(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq-sink_fake_to_rocketmq_message_tag.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq-sink_fake_to_rocketmq_message_tag.conf
new file mode 100644
index 0000000000..0badca6bf7
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/resources/rocketmq-sink_fake_to_rocketmq_message_tag.conf
@@ -0,0 +1,70 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+
+ #spark config
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 1
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ map.size = 10
+ array.size = 10
+ bytes.length = 10
+ string.length = 10
+ schema = {
+ fields {
+ c_map = "map<string, smallint>"
+ c_array = "array<int>"
+ c_string = string
+ c_boolean = boolean
+ c_tinyint = tinyint
+ c_smallint = smallint
+ c_int = int
+ c_bigint = bigint
+ c_float = float
+ c_double = double
+ c_decimal = "decimal(30, 8)"
+ c_null = "null"
+ c_bytes = bytes
+ c_date = date
+ c_timestamp = timestamp
+ }
+ }
+ }
+}
+
+transform {
+}
+
+sink {
+ Rocketmq {
+ name.srv.addr = "rocketmq-e2e:9876"
+ topic = "test_topic_message_tag"
+ partition.key.fields = ["c_map", "c_string"]
+ producer.send.sync = true
+ tag = "test_tag"
+ }
+}
\ No newline at end of file
