You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by zh...@apache.org on 2022/11/24 09:05:34 UTC
[rocketmq-connect] branch master updated: add mqtt sink connector (#360)
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-connect.git
The following commit(s) were added to refs/heads/master by this push:
new abda21dd add mqtt sink connector (#360)
abda21dd is described below
commit abda21dd797e0384a1be216f56aae77e04016568
Author: AetherZhuang <52...@users.noreply.github.com>
AuthorDate: Thu Nov 24 17:05:29 2022 +0800
add mqtt sink connector (#360)
* add mqtt sink connector
* add mqtt sink connector use schema
* add mqtt sink connector 3
Co-authored-by: zhuangxingwang <zh...@cmss.chinamobile.com>
---
connectors/rocketmq-connect-mqtt/pom.xml | 15 ---
.../connect/mqtt/config/SinkConnectorConfig.java | 90 +++++++++++++
.../connect/mqtt/connector/MqttSinkConnector.java | 57 ++++++++
.../connect/mqtt/connector/MqttSinkTask.java | 92 +++++++++++++
.../apache/rocketmq/connect/mqtt/sink/Updater.java | 143 +++++++++++++++++++++
.../test/java/connector/MqttSinkConnectorTest.java | 51 ++++++++
.../src/test/java/connector/MqttSinkTaskTest.java | 127 ++++++++++++++++++
.../src/test/java/sink/UpdaterTest.java | 101 +++++++++++++++
8 files changed, 661 insertions(+), 15 deletions(-)
diff --git a/connectors/rocketmq-connect-mqtt/pom.xml b/connectors/rocketmq-connect-mqtt/pom.xml
index fe9eb4e3..69003ebe 100644
--- a/connectors/rocketmq-connect-mqtt/pom.xml
+++ b/connectors/rocketmq-connect-mqtt/pom.xml
@@ -181,11 +181,6 @@
<artifactId>openmessaging-api</artifactId>
<version>0.3.1-alpha</version>
</dependency>
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-openmessaging</artifactId>
- <version>4.3.2</version>
- </dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
@@ -201,16 +196,6 @@
<artifactId>rocketmq-remoting</artifactId>
<version>${rocketmq.version}</version>
</dependency>
- <dependency>
- <groupId>com.alibaba</groupId>
- <artifactId>fastjson</artifactId>
- <version>1.2.83</version>
- </dependency>
- <dependency>
- <groupId>io.javalin</groupId>
- <artifactId>javalin</artifactId>
- <version>1.3.0</version>
- </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
diff --git a/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/config/SinkConnectorConfig.java b/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/config/SinkConnectorConfig.java
new file mode 100644
index 00000000..9c12b724
--- /dev/null
+++ b/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/config/SinkConnectorConfig.java
@@ -0,0 +1,90 @@
+/*
+ * 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.rocketmq.connect.mqtt.config;
+
+import io.openmessaging.KeyValue;
+import java.lang.reflect.Method;
+import java.util.HashSet;
+import java.util.Set;
+
+public class SinkConnectorConfig extends ConnectorConfig {
+
+ @SuppressWarnings("serial")
+ public static final Set<String> REQUEST_CONFIG = new HashSet<String>() {
+ {
+ add(MQTT_SINK_TOPIC);
+ }
+ };
+ public static final String MESSAGE = "message";
+ public static final String MQTT_SINK_TOPIC = "sinkTopic";
+ protected String sinkTopic;
+
+ public void load(KeyValue props) {
+
+ properties2Object(props, this);
+ }
+
+ private void properties2Object(final KeyValue p, final Object object) {
+
+ Method[] methods = object.getClass().getMethods();
+ for (Method method : methods) {
+ String mn = method.getName();
+ if (mn.startsWith("set")) {
+ try {
+ String tmp = mn.substring(4);
+ String first = mn.substring(3, 4);
+
+ String key = first.toLowerCase() + tmp;
+ String property = p.getString(key);
+ if (property != null) {
+ Class<?>[] pt = method.getParameterTypes();
+ if (pt != null && pt.length > 0) {
+ String cn = pt[0].getSimpleName();
+ Object arg;
+ if (cn.equals("int") || cn.equals("Integer")) {
+ arg = Integer.parseInt(property);
+ } else if (cn.equals("long") || cn.equals("Long")) {
+ arg = Long.parseLong(property);
+ } else if (cn.equals("double") || cn.equals("Double")) {
+ arg = Double.parseDouble(property);
+ } else if (cn.equals("boolean") || cn.equals("Boolean")) {
+ arg = Boolean.parseBoolean(property);
+ } else if (cn.equals("float") || cn.equals("Float")) {
+ arg = Float.parseFloat(property);
+ } else if (cn.equals("String")) {
+ arg = property;
+ } else {
+ continue;
+ }
+ method.invoke(object, arg);
+ }
+ }
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+
+ public String getSinkTopic() {
+ return sinkTopic;
+ }
+
+ public void setSinkTopic(String sinkTopic) {
+ this.sinkTopic = sinkTopic;
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/connector/MqttSinkConnector.java b/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/connector/MqttSinkConnector.java
new file mode 100644
index 00000000..4067b8e7
--- /dev/null
+++ b/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/connector/MqttSinkConnector.java
@@ -0,0 +1,57 @@
+/*
+ * 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.rocketmq.connect.mqtt.connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.sink.SinkConnector;
+import io.openmessaging.connector.api.errors.ConnectException;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.connect.mqtt.config.SinkConnectorConfig;
+
+public class MqttSinkConnector extends SinkConnector {
+
+ private KeyValue config;
+
+ @Override public void start(KeyValue config) {
+ for (String requestKey : SinkConnectorConfig.REQUEST_CONFIG) {
+ if (!config.containsKey(requestKey)) {
+ throw new ConnectException("Request config key: " + requestKey);
+ }
+ }
+ this.config = config;
+ }
+
+ @Override
+ public void stop() {
+ this.config = null;
+ }
+
+ @Override public List<KeyValue> taskConfigs(int maxTasks) {
+ List<KeyValue> config = new ArrayList<>();
+ config.add(this.config);
+ return config;
+ }
+
+ @Override
+ public Class<? extends Task> taskClass() {
+ return MqttSinkTask.class;
+ }
+
+}
diff --git a/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/connector/MqttSinkTask.java b/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/connector/MqttSinkTask.java
new file mode 100644
index 00000000..12016a41
--- /dev/null
+++ b/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/connector/MqttSinkTask.java
@@ -0,0 +1,92 @@
+/*
+ * 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.rocketmq.connect.mqtt.connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.sink.SinkTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.errors.ConnectException;
+import org.apache.rocketmq.connect.mqtt.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.mqtt.util.ConfigUtil;
+import org.apache.rocketmq.connect.mqtt.sink.Updater;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ *
+ */
+public class MqttSinkTask extends SinkTask {
+ private static final Logger log = LoggerFactory.getLogger(MqttSinkTask.class);
+
+ private SinkConnectorConfig sinkConnectConfig;
+ private Updater updater;
+
+ public MqttSinkTask() {
+ this.sinkConnectConfig = new SinkConnectorConfig();
+ }
+
+ @Override
+ public void put(List<ConnectRecord> sinkDataEntries) throws ConnectException {
+ try {
+ log.info("MQTT Sink Task trying to put()");
+ for (ConnectRecord record : sinkDataEntries) {
+ log.info("MQTT Sink Task trying to call updater.push()");
+ Boolean isSuccess = updater.push(record);
+ if (!isSuccess) {
+ log.error("mqtt sink push data error, record:{}", record);
+ }
+ log.debug("mqtt pushed data : " + record);
+ }
+ } catch (Exception e) {
+ log.error("put sinkDataEntries error, {}", e);
+ }
+ }
+
+ /**
+ * @param props
+ */
+ @Override
+ public void start(KeyValue props) {
+ try {
+ ConfigUtil.load(props, this.sinkConnectConfig);
+ log.info("init data source success");
+ } catch (Exception e) {
+ log.error("Cannot start MQTT Sink Task because of configuration error{}", e);
+ }
+ try {
+ updater = new Updater(sinkConnectConfig);
+ updater.start();
+ } catch (Throwable e) {
+ log.error("fail to start updater{}", e);
+ }
+
+ }
+
+ @Override
+ public void stop() {
+ try {
+ updater.stop();
+ log.info("mqtt sink task connection is closed.");
+ } catch (Throwable e) {
+ log.warn("sink task stop error while closing connection to mqtt", e);
+ }
+ }
+
+}
diff --git a/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/sink/Updater.java b/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/sink/Updater.java
new file mode 100644
index 00000000..44a691a8
--- /dev/null
+++ b/connectors/rocketmq-connect-mqtt/src/main/java/org/apache/rocketmq/connect/mqtt/sink/Updater.java
@@ -0,0 +1,143 @@
+/*
+ * 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.rocketmq.connect.mqtt.sink;
+
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.FieldType;
+import io.openmessaging.connector.api.data.Struct;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.mqtt.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.mqtt.util.MqttConnectionUtil;
+import org.apache.rocketmq.connect.mqtt.util.Utils;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Updater {
+
+ private final Logger log = LoggerFactory.getLogger(Updater.class);
+ private SinkConnectorConfig sinkConnectConfig;
+ private MqttClient mqttClient;
+ private volatile boolean started = false;
+
+ public Updater(SinkConnectorConfig sinkConnectConfig) {
+ this.sinkConnectConfig = sinkConnectConfig;
+
+ }
+
+ private MqttMessage sinkDataEntry2MqttMessage(ConnectRecord record) {
+ try {
+ Object object = record.getData();
+ log.info("Updater.sinkDataEntry2MqttMessage() get called ");
+ final Struct struct = (Struct) object;
+ final Object[] values = struct.getValues();
+ final List<Field> fields = struct.getSchema().getFields();
+ for (int i = 0; i < fields.size(); i++) {
+ final String name = fields.get(i).getName();
+ Object value = values[i];
+ if (name.equals(SinkConnectorConfig.MESSAGE) && value != null) {
+ byte[] recordBytes = ((String) value).getBytes(StandardCharsets.UTF_8);
+ MqttMessage mqttMessage = new MqttMessage(recordBytes);
+ return mqttMessage;
+ }
+ }
+ } catch (Exception e) {
+ log.error("convert record to mqttMessage error", e);
+ }
+ return null;
+ }
+
+ public void start() throws Exception {
+ MemoryPersistence memoryPersistence = new MemoryPersistence();
+ String brokerUrl = sinkConnectConfig.getMqttBrokerUrl();
+ String sendClientId = Utils.createClientId("send");
+ MqttConnectOptions mqttConnectOptions = MqttConnectionUtil.buildMqttConnectOptions(sendClientId, sinkConnectConfig);
+ mqttClient = new MqttClient(brokerUrl, sendClientId, memoryPersistence);
+ mqttClient.setCallback(new MqttCallbackExtended() {
+ @Override
+ public void connectComplete(boolean reconnect, String serverURI) {
+ log.info("{} connect success to {}", sendClientId, serverURI);
+ }
+
+ @Override
+ public void connectionLost(Throwable throwable) {
+ log.error("connection lost {}", throwable.getMessage());
+ }
+
+ @Override
+ public void messageArrived(String topic, MqttMessage mqttMessage) {
+ }
+
+ @Override
+ public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+ }
+ });
+ try {
+ mqttClient.connect(mqttConnectOptions);
+ started = true;
+ } catch (Exception e) {
+ log.error("MQTT Client of task {} start failed.", e);
+ }
+ log.info("MQTT sink task started");
+ }
+
+ public boolean push(ConnectRecord record) {
+ log.info("Updater Trying to push data");
+ Boolean isSuccess = true;
+ if (record == null) {
+ log.warn("Updater push sinkDataRecord null.");
+ return true;
+ }
+ try {
+ MqttMessage message = sinkDataEntry2MqttMessage(record);
+ if (message != null) {
+ mqttClient.publish(sinkConnectConfig.getSinkTopic(), message);
+ }
+ } catch (Exception e) {
+ log.error("Updater commit occur error", e);
+ isSuccess = false;
+ }
+
+ return isSuccess;
+ }
+
+ public void stop() {
+ if (started) {
+ if (this.mqttClient != null) {
+ try {
+ this.mqttClient.disconnect();
+ log.info("MQTT sink updater stopped.");
+ } catch (MqttException e) {
+ log.error("MQTT Client disConnect {} failed.", this.mqttClient, e);
+ }
+ }
+ started = false;
+ }
+ }
+
+}
diff --git a/connectors/rocketmq-connect-mqtt/src/test/java/connector/MqttSinkConnectorTest.java b/connectors/rocketmq-connect-mqtt/src/test/java/connector/MqttSinkConnectorTest.java
new file mode 100644
index 00000000..d876d0ee
--- /dev/null
+++ b/connectors/rocketmq-connect-mqtt/src/test/java/connector/MqttSinkConnectorTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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 connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.internal.DefaultKeyValue;
+import org.apache.rocketmq.connect.mqtt.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.mqtt.config.SourceConnectorConfig;
+import org.apache.rocketmq.connect.mqtt.connector.MqttSinkConnector;
+import org.apache.rocketmq.connect.mqtt.connector.MqttSinkTask;
+import org.apache.rocketmq.connect.mqtt.connector.MqttSourceConnector;
+import org.apache.rocketmq.connect.mqtt.connector.MqttSourceTask;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class MqttSinkConnectorTest {
+
+ MqttSinkConnector connector = new MqttSinkConnector();
+
+ @Test
+ public void taskClassTest() {
+ assertEquals(connector.taskClass(), MqttSinkTask.class);
+ }
+
+ @Test
+ public void taskConfigsTest() {
+ assertEquals(connector.taskConfigs(2).get(0), null);
+ KeyValue keyValue = new DefaultKeyValue();
+ for (String requestKey : SinkConnectorConfig.REQUEST_CONFIG) {
+ keyValue.put(requestKey, requestKey);
+ }
+ connector.start(keyValue);
+ assertEquals(connector.taskConfigs(2).get(0), keyValue);
+ }
+}
diff --git a/connectors/rocketmq-connect-mqtt/src/test/java/connector/MqttSinkTaskTest.java b/connectors/rocketmq-connect-mqtt/src/test/java/connector/MqttSinkTaskTest.java
new file mode 100644
index 00000000..42b774d8
--- /dev/null
+++ b/connectors/rocketmq-connect-mqtt/src/test/java/connector/MqttSinkTaskTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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 connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.apache.rocketmq.connect.mqtt.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.mqtt.config.SourceConnectorConfig;
+import org.apache.rocketmq.connect.mqtt.connector.MqttSinkTask;
+import org.apache.rocketmq.connect.mqtt.connector.MqttSourceTask;
+import org.apache.rocketmq.connect.mqtt.sink.Updater;
+import org.apache.rocketmq.connect.mqtt.source.Replicator;
+import org.apache.rocketmq.connect.mqtt.util.HmacSHA1Util;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class MqttSinkTaskTest {
+
+ @Before
+ public void before() {
+ KeyValue kv = new DefaultKeyValue();
+ kv.put("mqttBrokerUrl", "tcp://100.76.11.96:1883");
+ kv.put("mqttAccessKey", "rocketmq");
+ kv.put("mqttSecretKey", "12345678");
+ kv.put("sinkTopic", "topic_consumer");
+ MqttSinkTask task = new MqttSinkTask();
+ task.start(kv);
+ }
+
+ @Test
+ public void putTest() throws Exception {
+ MqttSinkTask task = new MqttSinkTask();
+ MqttMessage message = new MqttMessage();
+ message.setPayload("hello rocketmq".getBytes(StandardCharsets.UTF_8));
+ Schema schema = SchemaBuilder.struct().name("mqtt").build();
+ final List<io.openmessaging.connector.api.data.Field> fields = buildFields();
+ schema.setFields(fields);
+ final ConnectRecord connectRecord = new ConnectRecord(buildRecordPartition(),
+ buildRecordOffset(message),
+ System.currentTimeMillis(),
+ schema,
+ buildPayLoad(message));
+ ArrayList<ConnectRecord> list = new ArrayList<>();
+ list.add(connectRecord);
+
+ Updater updaterObject = Mockito.mock(Updater.class);
+
+ Field updater = MqttSinkTask.class.getDeclaredField("updater");
+ updater.setAccessible(true);
+ updater.set(task, updaterObject);
+
+ Field config = MqttSinkTask.class.getDeclaredField("sinkConnectConfig");
+ config.setAccessible(true);
+ config.set(task, new SinkConnectorConfig());
+ task.put(list);
+ Assert.assertEquals(list.size(), 1);
+
+ }
+
+ private RecordOffset buildRecordOffset(MqttMessage message) {
+ Map<String, Long> offsetMap = new HashMap<>();
+ offsetMap.put(SourceConnectorConfig.POSITION, 0l);
+ RecordOffset recordOffset = new RecordOffset(offsetMap);
+ return recordOffset;
+ }
+
+ private RecordPartition buildRecordPartition() {
+ Map<String, String> partitionMap = new HashMap<>();
+ partitionMap.put("partition", "defaultPartition");
+ RecordPartition recordPartition = new RecordPartition(partitionMap);
+ return recordPartition;
+ }
+
+ private List<io.openmessaging.connector.api.data.Field> buildFields() {
+ final Schema structSchema = SchemaBuilder.struct().build();
+ List<io.openmessaging.connector.api.data.Field> fields = new ArrayList<>();
+ fields.add(new io.openmessaging.connector.api.data.Field(0, SourceConnectorConfig.MESSAGE, structSchema));
+ return fields;
+ }
+
+ private String buildPayLoad(MqttMessage message) {
+ return new String(message.getPayload());
+
+ }
+
+}
diff --git a/connectors/rocketmq-connect-mqtt/src/test/java/sink/UpdaterTest.java b/connectors/rocketmq-connect-mqtt/src/test/java/sink/UpdaterTest.java
new file mode 100644
index 00000000..2bcf7e3a
--- /dev/null
+++ b/connectors/rocketmq-connect-mqtt/src/test/java/sink/UpdaterTest.java
@@ -0,0 +1,101 @@
+package sink;/*
+ * 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.
+ */
+
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.mqtt.config.SinkConnectorConfig;
+import org.apache.rocketmq.connect.mqtt.config.SourceConnectorConfig;
+import org.apache.rocketmq.connect.mqtt.connector.MqttSourceTask;
+import org.apache.rocketmq.connect.mqtt.sink.Updater;
+import org.apache.rocketmq.connect.mqtt.source.Replicator;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class UpdaterTest {
+
+ Updater updater;
+ SinkConnectorConfig config;
+
+ @Before
+ public void before() throws Exception {
+ config = new SinkConnectorConfig();
+ config.setMqttBrokerUrl("tcp://100.76.11.96:1883");
+ config.setMqttAccessKey("rocketmq");
+ config.setMqttSecretKey("12345678");
+ config.setSinkTopic("topic_consumer");
+ updater = new Updater(config);
+ updater.start();
+ }
+
+ @Test
+ public void stop() {
+ updater.stop();
+ }
+
+ @Test
+ public void pushTest() {
+ MqttMessage message = new MqttMessage();
+ message.setPayload("hello rocketmq".getBytes(StandardCharsets.UTF_8));
+ Schema schema = SchemaBuilder.struct().name("mqtt").build();
+ final List<Field> fields = buildFields();
+ schema.setFields(fields);
+ final ConnectRecord connectRecord = new ConnectRecord(buildRecordPartition(),
+ buildRecordOffset(message),
+ System.currentTimeMillis(),
+ schema,
+ buildPayLoad(message));
+ updater.push(connectRecord);
+ }
+
+ private RecordOffset buildRecordOffset(MqttMessage message) {
+ Map<String, Long> offsetMap = new HashMap<>();
+ offsetMap.put(SourceConnectorConfig.POSITION, 0l);
+ RecordOffset recordOffset = new RecordOffset(offsetMap);
+ return recordOffset;
+ }
+
+ private RecordPartition buildRecordPartition() {
+ Map<String, String> partitionMap = new HashMap<>();
+ partitionMap.put("partition", "defaultPartition");
+ RecordPartition recordPartition = new RecordPartition(partitionMap);
+ return recordPartition;
+ }
+
+ private List<Field> buildFields() {
+ final Schema structSchema = SchemaBuilder.struct().build();
+ List<Field> fields = new ArrayList<>();
+ fields.add(new Field(0, SourceConnectorConfig.MESSAGE, structSchema));
+ return fields;
+ }
+
+ private String buildPayLoad(MqttMessage message) {
+ return new String(message.getPayload());
+
+ }
+
+}