This is an automated email from the ASF dual-hosted git repository.
chenguangsheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new a8d715ee5 [ISSUE #3920] Add Sink Connector RocketMQ Module
new 435c8a279 Merge pull request #3921 from
xwm1992/add-sink-connector-rocketmq
a8d715ee5 is described below
commit a8d715ee598d9db94f7945bd93d151d260750711
Author: xwm1992 <[email protected]>
AuthorDate: Sat May 13 21:59:17 2023 +0800
[ISSUE #3920] Add Sink Connector RocketMQ Module
---
.../sink-connector-rocketmq/build.gradle | 41 ++++++++
.../sink-connector-rocketmq/gradle.properties | 17 ++++
.../connector/rocketmq/EventMeshTestUtils.java | 113 +++++++++++++++++++++
.../connector/rocketmq/RocketMQSinkWorker.java | 104 +++++++++++++++++++
.../sink/connector/rocketmq/UtilsConstants.java | 52 ++++++++++
.../rocketmq/config/RocketMQSinkConfig.java | 63 ++++++++++++
.../rocketmq/connector/RocketMQSinkConnector.java | 95 +++++++++++++++++
.../connector/rocketmq/RocketMQSourceWorker.java | 4 +-
settings.gradle | 1 +
9 files changed, 488 insertions(+), 2 deletions(-)
diff --git a/eventmesh-connectors/sink-connector-rocketmq/build.gradle
b/eventmesh-connectors/sink-connector-rocketmq/build.gradle
new file mode 100644
index 000000000..180094cf2
--- /dev/null
+++ b/eventmesh-connectors/sink-connector-rocketmq/build.gradle
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+List rocketmq = [
+ "org.apache.rocketmq:rocketmq-client:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-broker:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-common:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-store:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-namesrv:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-tools:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-remoting:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-logging:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-test:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-srvutil:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-filter:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-acl:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-srvutil:$rocketmq_version",
+
+]
+
+dependencies {
+ implementation project(":eventmesh-connectors:eventmesh-connector-api")
+ implementation project(":eventmesh-sdk-java")
+ implementation rocketmq
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+}
diff --git a/eventmesh-connectors/sink-connector-rocketmq/gradle.properties
b/eventmesh-connectors/sink-connector-rocketmq/gradle.properties
new file mode 100644
index 000000000..89a6e4612
--- /dev/null
+++ b/eventmesh-connectors/sink-connector-rocketmq/gradle.properties
@@ -0,0 +1,17 @@
+#
+# 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.
+#
+rocketmq_version=4.9.5
\ No newline at end of file
diff --git
a/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/EventMeshTestUtils.java
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/EventMeshTestUtils.java
new file mode 100644
index 000000000..44c76803c
--- /dev/null
+++
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/EventMeshTestUtils.java
@@ -0,0 +1,113 @@
+/*
+ * 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.eventmesh.sink.connector.rocketmq;
+
+import static
org.apache.eventmesh.common.protocol.tcp.Command.RESPONSE_TO_SERVER;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
+import org.apache.eventmesh.client.tcp.common.MessageUtils;
+import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
+import org.apache.eventmesh.common.protocol.tcp.Header;
+import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.JsonUtils;
+
+public class EventMeshTestUtils {
+
+ private static final int SEQ_LENGTH = 10;
+
+ private static final String ASYNC_MSG_BODY = "testAsyncMessage";
+
+ private static final String DEFAULT_TTL_MS = "30000";
+
+ // generate pub-client
+ public static UserAgent generateClient1() {
+ final UserAgent agent = UserAgent.builder()
+ .env(UtilsConstants.ENV)
+ .host(UtilsConstants.HOST)
+ .password(generateRandomString(UtilsConstants.PASSWORD_LENGTH))
+ .username(UtilsConstants.USER_NAME)
+ .group(UtilsConstants.GROUP)
+ .path(UtilsConstants.PATH)
+ .port(UtilsConstants.PORT_1)
+ .subsystem(UtilsConstants.SUB_SYSTEM_1)
+ .pid(UtilsConstants.PID_1)
+ .version(UtilsConstants.VERSION)
+ .idc(UtilsConstants.IDC)
+ .build();
+ return MessageUtils.generatePubClient(agent);
+ }
+
+ // generate sub-client
+ public static UserAgent generateClient2() {
+ final UserAgent agent = UserAgent.builder()
+ .env(UtilsConstants.ENV)
+ .host(UtilsConstants.HOST)
+ .password(generateRandomString(UtilsConstants.PASSWORD_LENGTH))
+ .username(UtilsConstants.USER_NAME)
+ .group(UtilsConstants.GROUP)
+ .path(UtilsConstants.PATH)
+ .port(UtilsConstants.PORT_2)
+ .subsystem(UtilsConstants.SUB_SYSTEM_2)
+ .pid(UtilsConstants.PID_2)
+ .version(UtilsConstants.VERSION)
+ .idc(UtilsConstants.IDC)
+ .build();
+ return MessageUtils.generateSubClient(agent);
+ }
+
+ public static Package rrResponse(final EventMeshMessage request) {
+ final Package msg = new Package();
+ msg.setHeader(new Header(RESPONSE_TO_SERVER, 0, null,
generateRandomString(SEQ_LENGTH)));
+ msg.setBody(request);
+ return msg;
+ }
+
+ private static String generateRandomString(final int length) {
+ final StringBuilder builder = new StringBuilder(length);
+ for (int i = 0; i < length; i++) {
+ builder.append((char) ThreadLocalRandom.current().nextInt(48, 57));
+ }
+ return builder.toString();
+ }
+
+ public static CloudEvent generateCloudEventV1(String destination, String
message) {
+ final Map<String, String> content = new HashMap<>();
+ content.put(UtilsConstants.CONTENT, message);
+
+ return CloudEventBuilder.v1()
+ .withId(UUID.randomUUID().toString())
+ .withSubject(destination)
+ .withSource(URI.create("/"))
+ .withDataContentType("application/cloudevents+json")
+ .withType(EventMeshCommon.CLOUD_EVENTS_PROTOCOL_NAME)
+
.withData(Objects.requireNonNull(JsonUtils.toJSONString(content)).getBytes(StandardCharsets.UTF_8))
+ .withExtension(UtilsConstants.TTL, DEFAULT_TTL_MS)
+ .build();
+ }
+
+}
diff --git
a/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/RocketMQSinkWorker.java
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/RocketMQSinkWorker.java
new file mode 100644
index 000000000..2dc333e92
--- /dev/null
+++
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/RocketMQSinkWorker.java
@@ -0,0 +1,104 @@
+/*
+ * 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.eventmesh.sink.connector.rocketmq;
+
+import io.cloudevents.CloudEvent;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+
+import org.apache.eventmesh.client.tcp.EventMeshTCPClient;
+import org.apache.eventmesh.client.tcp.EventMeshTCPClientFactory;
+import org.apache.eventmesh.client.tcp.common.ReceiveMsgHook;
+import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.connector.api.data.ConnectRecord;
+import org.apache.eventmesh.sink.connector.rocketmq.config.RocketMQSinkConfig;
+import
org.apache.eventmesh.sink.connector.rocketmq.connector.RocketMQSinkConnector;
+
+public class RocketMQSinkWorker {
+
+ public static final String SINK_GROUP = "DEFAULT-SINK-GROUP";
+
+ public static final String SINK_CONNECT_NAMESRVADDR = "127.0.0.1:9877";
+
+ public static final String SINK_TOPIC = "TopicTest";
+
+ public static RocketMQSinkConnector rocketMQSinkConnector;
+
+ public static void main(String[] args) throws Exception {
+
+ UserAgent userAgent = EventMeshTestUtils.generateClient1();
+ EventMeshTCPClientConfig eventMeshTcpClientConfig =
EventMeshTCPClientConfig.builder()
+ .host("127.0.0.1")
+ .port(10002)
+ .userAgent(userAgent)
+ .build();
+
+ final EventMeshTCPClient<CloudEvent> client =
+
EventMeshTCPClientFactory.createEventMeshTCPClient(eventMeshTcpClientConfig,
CloudEvent.class);
+
+ client.init();
+
+ client.subscribe(SINK_TOPIC, SubscriptionMode.CLUSTERING,
+ SubscriptionType.ASYNC);
+ client.registerSubBusiHandler(new MessageHandler());
+
+ client.listen();
+
+ rocketMQSinkConnector = new RocketMQSinkConnector();
+
+ RocketMQSinkConfig rocketMQSinkConfig = new RocketMQSinkConfig();
+
+ rocketMQSinkConfig.setSinkNameserver(SINK_CONNECT_NAMESRVADDR);
+ rocketMQSinkConfig.setSinkTopic(SINK_TOPIC);
+ rocketMQSinkConfig.setSinkGroup(SINK_GROUP);
+
+ rocketMQSinkConnector.init(rocketMQSinkConfig);
+
+ rocketMQSinkConnector.start();
+
+ }
+
+ public static class MessageHandler implements ReceiveMsgHook<CloudEvent> {
+
+ @Override
+ public Optional<CloudEvent> handle(CloudEvent event) {
+ byte[] body = Objects.requireNonNull(event.getData()).toBytes();
+ //todo: recordPartition & recordOffset
+ ConnectRecord connectRecord = new ConnectRecord(null, null,
System.currentTimeMillis(), body);
+ for(String extensionName : event.getExtensionNames()) {
+ connectRecord.addExtension(extensionName,
Objects.requireNonNull(event.getExtension(extensionName)).toString());
+ }
+ connectRecord.addExtension("id", event.getId());
+ connectRecord.addExtension("topic", event.getSubject());
+ connectRecord.addExtension("source", event.getSource().toString());
+ connectRecord.addExtension("type", event.getType());
+ List<ConnectRecord> connectRecords = new ArrayList<>();
+ connectRecords.add(connectRecord);
+ rocketMQSinkConnector.put(connectRecords);
+ return Optional.empty();
+ }
+ }
+
+}
diff --git
a/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/UtilsConstants.java
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/UtilsConstants.java
new file mode 100644
index 000000000..afd2ce616
--- /dev/null
+++
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/UtilsConstants.java
@@ -0,0 +1,52 @@
+/*
+ * 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.eventmesh.sink.connector.rocketmq;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class UtilsConstants {
+
+ public static final String ENV = "test";
+ public static final String HOST = "localhost";
+ public static final Integer PASSWORD_LENGTH = 8;
+ public static final String USER_NAME = "PU4283";
+ public static final String GROUP = "EventmeshTestGroup";
+ public static final String PATH = "/data/app/umg_proxy";
+ public static final Integer PORT_1 = 8362;
+ public static final Integer PORT_2 = 9362;
+ public static final String SUB_SYSTEM_1 = "5023";
+ public static final String SUB_SYSTEM_2 = "5017";
+ public static final Integer PID_1 = 32_893;
+ public static final Integer PID_2 = 42_893;
+ public static final String VERSION = "2.0.11";
+ public static final String IDC = "FT";
+ /**
+ * PROPERTY KEY NAME .
+ */
+ public static final String MSG_TYPE = "msgtype";
+ public static final String TTL = "ttl";
+ public static final String KEYS = "keys";
+ public static final String REPLY_TO = "replyto";
+ public static final String PROPERTY_MESSAGE_REPLY_TO =
"propertymessagereplyto";
+ public static final String CONTENT = "content";
+
+
+}
diff --git
a/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/config/RocketMQSinkConfig.java
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/config/RocketMQSinkConfig.java
new file mode 100644
index 000000000..b2b36d3c6
--- /dev/null
+++
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/config/RocketMQSinkConfig.java
@@ -0,0 +1,63 @@
+/*
+ * 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.eventmesh.sink.connector.rocketmq.config;
+
+import org.apache.eventmesh.connector.api.config.SinkConfig;
+
+public class RocketMQSinkConfig extends SinkConfig {
+
+ String connectorName;
+
+ String sinkNameserver;
+
+ String sinkTopic;
+
+ String sinkGroup;
+
+ public String getConnectorName() {
+ return connectorName;
+ }
+
+ public void setConnectorName(String connectorName) {
+ this.connectorName = connectorName;
+ }
+
+ public String getSinkNameserver() {
+ return sinkNameserver;
+ }
+
+ public void setSinkNameserver(String sinkNameserver) {
+ this.sinkNameserver = sinkNameserver;
+ }
+
+ public String getSinkTopic() {
+ return sinkTopic;
+ }
+
+ public void setSinkTopic(String sinkTopic) {
+ this.sinkTopic = sinkTopic;
+ }
+
+ public String getSinkGroup() {
+ return sinkGroup;
+ }
+
+ public void setSinkGroup(String sinkGroup) {
+ this.sinkGroup = sinkGroup;
+ }
+}
diff --git
a/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/connector/RocketMQSinkConnector.java
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/connector/RocketMQSinkConnector.java
new file mode 100644
index 000000000..eba62155e
--- /dev/null
+++
b/eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/connector/RocketMQSinkConnector.java
@@ -0,0 +1,95 @@
+/*
+ * 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.eventmesh.sink.connector.rocketmq.connector;
+
+import java.util.List;
+import org.apache.eventmesh.connector.api.config.Config;
+import org.apache.eventmesh.connector.api.data.ConnectRecord;
+import org.apache.eventmesh.connector.api.sink.Sink;
+
+import org.apache.eventmesh.sink.connector.rocketmq.config.RocketMQSinkConfig;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageAccessor;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RocketMQSinkConnector implements Sink {
+
+ private RocketMQSinkConfig sinkConfig;
+
+ private final DefaultMQProducer producer = new DefaultMQProducer();
+
+ @Override
+ public Class<? extends Config> configClass() {
+ return RocketMQSinkConfig.class;
+ }
+
+ @Override
+ public void init(Config config) throws Exception {
+ // init config for rocketmq source connector
+ this.sinkConfig = (RocketMQSinkConfig) config;
+ producer.setProducerGroup(sinkConfig.getSinkGroup());
+ producer.setNamesrvAddr(sinkConfig.getSinkNameserver());
+ }
+
+ @Override
+ public void start() throws Exception {
+ producer.start();
+ }
+
+ @Override
+ public void commit(ConnectRecord record) {
+
+ }
+
+ @Override
+ public String name() {
+ return this.sinkConfig.getConnectorName();
+ }
+
+ @Override
+ public void stop() {
+ producer.shutdown();
+ }
+
+ @Override
+ public void put(List<ConnectRecord> sinkRecords) {
+ for (ConnectRecord connectRecord : sinkRecords) {
+ Message message = convertRecordToMessage(connectRecord);
+ try {
+ SendResult sendResult = producer.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public static Message convertRecordToMessage(ConnectRecord connectRecord) {
+ Message message = new Message();
+ message.setTopic(connectRecord.getExtension("topic"));
+ message.setBody((byte[]) connectRecord.getData());
+ for (String key : connectRecord.getExtensions().keySet()) {
+ MessageAccessor.putProperty(message, key,
connectRecord.getExtension(key));
+ }
+ return message;
+ }
+}
diff --git
a/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/RocketMQSourceWorker.java
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/RocketMQSourceWorker.java
index 41a758ecd..5acf10d86 100644
---
a/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/RocketMQSourceWorker.java
+++
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/RocketMQSourceWorker.java
@@ -29,7 +29,7 @@ import
org.apache.eventmesh.source.connector.rocketmq.connector.RocketMQSourceCo
public class RocketMQSourceWorker {
- public static final String SOURCE_CONSUMER_GROUP =
"DEFAULT-CONSUMER-GROUP";
+ public static final String SOURCE_GROUP = "DEFAULT-CONSUMER-GROUP";
public static final String SOURCE_CONNECT_NAMESRVADDR = "127.0.0.1:9877";
public static final String SOURCE_TOPIC = "TopicTest";
@@ -56,7 +56,7 @@ public class RocketMQSourceWorker {
rocketMQSourceConfig.setSourceNameserver(SOURCE_CONNECT_NAMESRVADDR);
rocketMQSourceConfig.setSourceTopic(SOURCE_TOPIC);
- rocketMQSourceConfig.setSourceGroup(SOURCE_CONSUMER_GROUP);
+ rocketMQSourceConfig.setSourceGroup(SOURCE_GROUP);
rocketMQSourceConnector.init(rocketMQSourceConfig);
diff --git a/settings.gradle b/settings.gradle
index 722fa0ab7..e1e742ee3 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -25,6 +25,7 @@ include 'eventmesh-examples'
include 'eventmesh-spi'
include 'eventmesh-connectors:eventmesh-connector-api'
include 'eventmesh-connectors:source-connector-rocketmq'
+include 'eventmesh-connectors:sink-connector-rocketmq'
include 'eventmesh-storage-plugin:eventmesh-storage-api'
include 'eventmesh-storage-plugin:eventmesh-storage-standalone'
include 'eventmesh-storage-plugin:eventmesh-storage-kafka'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
