labelList) {
+ if (labelList == null || labelList.isEmpty()) {
+ return "";
+ }
+ StringBuilder stringBuilder = new StringBuilder();
+ for (String label : labelList) {
+ stringBuilder.append(":").append(label);
+ }
+ return stringBuilder.substring(1);
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/helper/ValueType.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/helper/ValueType.java
new file mode 100644
index 00000000..b4ab5cea
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/helper/ValueType.java
@@ -0,0 +1,132 @@
+/*
+ * 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.neo4j.helper;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
+import com.alibaba.fastjson.JSONObject;
+
+public enum ValueType {
+ /**
+ * transfer gdb element object value to DataX Column data
+ *
+ * int, long -> LongColumn
+ * float, double -> DoubleColumn
+ * bool -> BooleanColumn
+ * string -> StringColumn
+ */
+ INT(Integer.class, "int", ValueTypeHolder::longColumnMapper),
+ INTEGER(Integer.class, "integer", ValueTypeHolder::longColumnMapper),
+ LONG(Long.class, "long", ValueTypeHolder::longColumnMapper),
+ DOUBLE(Double.class, "double", ValueTypeHolder::doubleColumnMapper),
+ FLOAT(Float.class, "float", ValueTypeHolder::doubleColumnMapper),
+ BOOLEAN(Boolean.class, "boolean", ValueTypeHolder::boolColumnMapper),
+ STRING(String.class, "string", ValueTypeHolder::stringColumnMapper),
+ ;
+
+ private Class type = null;
+ private String shortName = null;
+ private Function columnFunc = null;
+
+ ValueType(Class type, String name, Function columnFunc) {
+ this.type = type;
+ this.shortName = name;
+ this.columnFunc = columnFunc;
+
+ ValueTypeHolder.shortName2type.put(shortName, this);
+ }
+
+ public static ValueType fromShortName(String name) {
+ return ValueTypeHolder.shortName2type.get(name);
+ }
+
+ public Object applyObject(Object value) {
+ if (value == null) {
+ return null;
+ }
+ return columnFunc.apply(value);
+ }
+
+ private static class ValueTypeHolder {
+ private static Map shortName2type = new HashMap<>();
+
+ private static Long longColumnMapper(Object o) {
+ long v;
+ if (o instanceof Integer) {
+ v = (int)o;
+ } else if (o instanceof Long) {
+ v = (long)o;
+ } else if (o instanceof String) {
+ v = Long.valueOf((String)o);
+ } else {
+ throw new RuntimeException("Failed to cast " + o.getClass() + " to Long");
+ }
+
+ return v;
+ }
+
+ private static Double doubleColumnMapper(Object o) {
+ double v;
+ if (o instanceof Integer) {
+ v = (double)(int)o;
+ } else if (o instanceof Long) {
+ v = (double)(long)o;
+ } else if (o instanceof Float) {
+ v = (double)(float)o;
+ } else if (o instanceof Double) {
+ v = (double)o;
+ } else if (o instanceof String) {
+ v = Double.valueOf((String)o);
+ } else {
+ throw new RuntimeException("Failed to cast " + o.getClass() + " to Double");
+ }
+
+ return v;
+ }
+
+ private static Boolean boolColumnMapper(Object o) {
+ boolean v;
+ if (o instanceof Integer) {
+ v = ((int)o != 0);
+ } else if (o instanceof Long) {
+ v = ((long)o != 0);
+ } else if (o instanceof Boolean) {
+ v = (boolean)o;
+ } else if (o instanceof String) {
+ v = Boolean.valueOf((String)o);
+ } else {
+ throw new RuntimeException("Failed to cast " + o.getClass() + " to Boolean");
+ }
+
+ return v;
+ }
+
+ private static String stringColumnMapper(Object o) {
+ if (o instanceof String) {
+ return (String)o;
+ } else if (MappingRuleFactory.isPrimitive(o)) {
+ return String.valueOf(o);
+ } else {
+ return JSONObject.toJSONString(o);
+ }
+ }
+ }
+
+}
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkConnector.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkConnector.java
new file mode 100644
index 00000000..ecd9a410
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkConnector.java
@@ -0,0 +1,60 @@
+/*
+ * 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.neo4j.sink;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSinkConfig;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.sink.SinkConnector;
+
+public class Neo4jSinkConnector extends SinkConnector {
+ private KeyValue keyValue;
+
+ @Override
+ public List taskConfigs(int maxTasks) {
+ List configs = new ArrayList<>();
+ for (int i = 0; i < maxTasks; i++) {
+ configs.add(this.keyValue);
+ }
+ return configs;
+ }
+
+ @Override
+ public Class taskClass() {
+ return Neo4jSinkTask.class;
+ }
+
+ @Override
+ public void start(KeyValue keyValue) {
+ for (String requestKey : Neo4jSinkConfig.SINK_REQUEST_CONFIG) {
+ if (!keyValue.containsKey(requestKey)) {
+ throw new RuntimeException("Request config key: " + requestKey);
+ }
+ }
+ this.keyValue = keyValue;
+ }
+
+ @Override
+ public void stop() {
+ this.keyValue = null;
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkTask.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkTask.java
new file mode 100644
index 00000000..f41d7fea
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkTask.java
@@ -0,0 +1,72 @@
+/*
+ * 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.neo4j.sink;
+
+import java.util.List;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSinkConfig;
+import org.apache.rocketmq.connect.neo4j.helper.LabelTypeEnum;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jClient;
+import org.apache.rocketmq.connect.neo4j.sink.write.NodeWriteProcessor;
+import org.apache.rocketmq.connect.neo4j.sink.write.Processor;
+import org.apache.rocketmq.connect.neo4j.sink.write.RelationshipWriteProcessor;
+
+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;
+
+public class Neo4jSinkTask extends SinkTask {
+ private Neo4jSinkConfig config;
+ private Neo4jClient client;
+ private Processor processor;
+
+ @Override
+ public void put(List sinkRecords) throws ConnectException {
+ if (sinkRecords == null || sinkRecords.size() < 1) {
+ return;
+ }
+ for (ConnectRecord connectRecord : sinkRecords) {
+ processor.write(connectRecord);
+ }
+ }
+
+ @Override
+ public void start(KeyValue keyValue) {
+ this.config = new Neo4jSinkConfig();
+ this.config.load(keyValue);
+ LabelTypeEnum labelTypeEnum = LabelTypeEnum.nameOf(config.getLabelType());
+ if (labelTypeEnum == null) {
+ throw new RuntimeException("labelType only support node or relationship");
+ }
+ this.client = new Neo4jClient(this.config);
+ if (!client.ping()) {
+ throw new RuntimeException("Cannot connect to neo4j server!");
+ }
+ if (LabelTypeEnum.node == labelTypeEnum) {
+ processor = new NodeWriteProcessor(config, client);
+ } else {
+ processor = new RelationshipWriteProcessor(config, client);
+ }
+ }
+
+ @Override
+ public void stop() {
+ this.client = null;
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/mapping/WriteMapper.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/mapping/WriteMapper.java
new file mode 100644
index 00000000..eaf6b1d4
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/mapping/WriteMapper.java
@@ -0,0 +1,88 @@
+/*
+ * 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.neo4j.sink.mapping;
+
+import java.util.function.BiConsumer;
+
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jElement;
+import org.apache.rocketmq.connect.neo4j.helper.MappingRule;
+import org.apache.rocketmq.connect.neo4j.helper.ValueType;
+
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.Struct;
+
+public class WriteMapper {
+ public static BiConsumer getMapper(MappingRule rule) {
+ return (record, neo4jElement) -> {
+ final Schema recordSchema = record.getSchema();
+ final Struct struct = (Struct)record.getData();
+
+ rule.getColumns().forEach(columnMappingRule -> {
+ Object value = null;
+ ValueType type = columnMappingRule.getValueType();
+ String name = columnMappingRule.getName();
+ boolean extractFixedValue = false;
+ final String valueExtract = columnMappingRule.getValueExtract();
+ if (valueExtract.startsWith("#{") && valueExtract.endsWith("}")) {
+ final String val = valueExtract.substring(2, valueExtract.length() - 1);
+ final Field field = recordSchema.getField(val);
+ if (field != null) {
+ value = struct.get(field);
+ }
+ } else {
+ value = valueExtract;
+ extractFixedValue = true;
+ }
+ switch (columnMappingRule.getColumnType()) {
+ case primaryKey:
+ neo4jElement.setPrimaryKey(name);
+ neo4jElement.setPrimaryValue(String.valueOf(value));
+ break;
+ case primaryLabel:
+ neo4jElement.setLabel(String.valueOf(value));
+ break;
+ case dstPrimaryKey:
+ neo4jElement.setToPrimaryKey(name);
+ neo4jElement.setToPrimaryValue(String.valueOf(value));
+ break;
+ case srcPrimaryKey:
+ neo4jElement.setFromPrimaryKey(name);
+ neo4jElement.setFromPrimaryValue(String.valueOf(value));
+ break;
+ case srcPrimaryLabel:
+ neo4jElement.setFromLabel(String.valueOf(value));
+ break;
+ case dstPrimaryLabel:
+ neo4jElement.setToLabel(String.valueOf(value));
+ break;
+ case nodeProperty:
+ case relationshipProperty:
+ case nodeJsonProperty:
+ case relationshipJsonProperty:
+ if (value != null) {
+ neo4jElement.getProperties().put(name, type.applyObject(value));
+ }
+ default:
+ break;
+ }
+ });
+ };
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/NodeWriteProcessor.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/NodeWriteProcessor.java
new file mode 100644
index 00000000..13d7fdec
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/NodeWriteProcessor.java
@@ -0,0 +1,72 @@
+/*
+ * 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.neo4j.sink.write;
+
+import java.util.Map;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSinkConfig;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jClient;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jElement;
+
+import io.openmessaging.connector.api.data.*;
+
+public class NodeWriteProcessor extends WriteProcessor {
+ private Neo4jSinkConfig neo4jSinkConfig;
+
+ public NodeWriteProcessor(Neo4jSinkConfig config, Neo4jClient neo4jClient) {
+ super(config, neo4jClient);
+ this.neo4jSinkConfig = config;
+ }
+
+ //MERGE (keanu:Person {name: 'Keanu Reeves'})
+ //ON CREATE
+ // SET keanu.created = timestamp()
+ //ON MATCH
+ // SET keanu.lastSeen = timestamp()
+ //RETURN keanu.name, keanu.created, keanu.lastSeen
+ protected String buildCql(Neo4jElement neo4jElement, Schema schema) {
+ StringBuilder cql = new StringBuilder();
+ cql.append("use ").append(neo4jSinkConfig.getNeo4jDataBase()).append(" ");
+ cql.append("merge (n:").append(neo4jElement.getLabel()).append("{").append(neo4jElement.getPrimaryKey())
+ .append(":").append(neo4jElement.getPrimaryValue()).append("})");
+ if (!neo4jElement.getProperties().isEmpty()) {
+ cql.append(" on create set ").append(setProperty(neo4jElement.getProperties(), schema));
+ cql.append(" on match set ").append(setProperty(neo4jElement.getProperties(), schema));
+ }
+ return cql.toString();
+ }
+
+ protected String setProperty(Map property, Schema recordSchema) {
+ StringBuilder cql = new StringBuilder();
+ for (String key : property.keySet()) {
+ final Object o = property.get(key);
+ final Field field = recordSchema.getField(key);
+ if (field.getSchema().getFieldType() == FieldType.STRING) {
+ cql.append("n.").append(key).append(" = ").append(setStringValue((String)o));
+ } else {
+ cql.append("n.").append(key).append(" = ").append(o);
+ }
+ cql.append(",");
+ }
+ return cql.substring(0, cql.length() - 1);
+ }
+
+ protected String setStringValue(String v) {
+ return "'" + v + "'";
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/Processor.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/Processor.java
new file mode 100644
index 00000000..c56c7104
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/Processor.java
@@ -0,0 +1,23 @@
+/*
+ * 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.neo4j.sink.write;
+
+import io.openmessaging.connector.api.data.ConnectRecord;
+
+public interface Processor {
+ void write(ConnectRecord record);
+}
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/RelationshipWriteProcessor.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/RelationshipWriteProcessor.java
new file mode 100644
index 00000000..5e6e5b31
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/RelationshipWriteProcessor.java
@@ -0,0 +1,76 @@
+/*
+ * 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.neo4j.sink.write;
+
+import java.util.Map;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSinkConfig;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jClient;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jElement;
+
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.FieldType;
+import io.openmessaging.connector.api.data.Schema;
+
+public class RelationshipWriteProcessor extends WriteProcessor {
+ private Neo4jSinkConfig neo4jSinkConfig;
+
+ public RelationshipWriteProcessor(Neo4jSinkConfig config, Neo4jClient neo4jClient) {
+ super(config, neo4jClient);
+ this.neo4jSinkConfig = config;
+ }
+
+ protected String buildCql(Neo4jElement neo4jElement, Schema schema) {
+ StringBuilder cql = new StringBuilder();
+ cql.append("use ").append(neo4jSinkConfig.getNeo4jDataBase()).append(" ");
+ cql.append("merge (source:").append(neo4jElement.getFromLabel()).append("{")
+ .append(neo4jElement.getFromPrimaryKey()).append(":").append(neo4jElement.getFromPrimaryValue())
+ .append("}) ");
+ cql.append("merge (target:").append(neo4jElement.getToLabel()).append("{")
+ .append(neo4jElement.getToPrimaryKey()).append(":").append(neo4jElement.getToPrimaryValue()).append("}) ");
+ cql.append("merge (source)-[:").append(neo4jElement.getLabel()).append("{")
+ .append(addProperty(schema, neo4jElement)).append("}]").append("->(target)");
+ return cql.toString();
+ }
+
+ protected String addProperty(Schema schema, Neo4jElement neo4jElement) {
+ StringBuilder cql = new StringBuilder();
+ cql.append(setValue(neo4jElement.getPrimaryKey(), neo4jElement.getPrimaryValue(), schema)).append(",");
+ final Map properties = neo4jElement.getProperties();
+ if (!properties.isEmpty()) {
+ for (String key : properties.keySet()) {
+ final Object o = properties.get(key);
+ cql.append(setValue(key, o, schema));
+ cql.append(",");
+ }
+ }
+ return cql.substring(0, cql.length() - 1);
+ }
+
+ protected String setValue(String key, Object value, Schema schema) {
+ final Field field = schema.getField(key);
+ if (field == null) {
+ return key + ":" + value;
+ }
+ if (field.getSchema().getFieldType() == FieldType.STRING) {
+ return key + ":'" + value + "'";
+ } else {
+ return key + ":" + value;
+ }
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/WriteProcessor.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/WriteProcessor.java
new file mode 100644
index 00000000..1014e4b3
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/sink/write/WriteProcessor.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.neo4j.sink.write;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSinkConfig;
+import org.apache.rocketmq.connect.neo4j.helper.LabelTypeEnum;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jClient;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jElement;
+import org.apache.rocketmq.connect.neo4j.sink.mapping.WriteMapper;
+import org.apache.rocketmq.connect.neo4j.helper.MappingRule;
+import org.apache.rocketmq.connect.neo4j.helper.MappingRuleFactory;
+
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Schema;
+
+public abstract class WriteProcessor implements Processor {
+ private Neo4jSinkConfig neo4jSinkConfig;
+ private Neo4jClient neo4jClient;
+ private MappingRule mappingRule;
+
+ public WriteProcessor(Neo4jSinkConfig config, Neo4jClient neo4jClient) {
+ this.neo4jSinkConfig = config;
+ this.neo4jClient = neo4jClient;
+ final LabelTypeEnum labelTypeEnum = LabelTypeEnum.nameOf(neo4jSinkConfig.getLabelType());
+ if (labelTypeEnum == null) {
+ throw new RuntimeException("label type only support node or relationship");
+ }
+ mappingRule = MappingRuleFactory.getInstance().create(neo4jSinkConfig, labelTypeEnum);
+ mappingRule.validateConfig();
+ }
+
+ public void write(ConnectRecord record) {
+ final Schema recordSchema = record.getSchema();
+ Neo4jElement neo4jElement = new Neo4jElement();
+ WriteMapper.getMapper(mappingRule).accept(record, neo4jElement);
+ String cql = buildCql(neo4jElement, recordSchema);
+ neo4jClient.insert(cql);
+ }
+
+ protected abstract String buildCql(Neo4jElement neo4jElement, Schema schema);
+
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceConnector.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceConnector.java
new file mode 100644
index 00000000..f0597aa2
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceConnector.java
@@ -0,0 +1,97 @@
+/*
+ * 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.neo4j.source;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jConstants;
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSourceConfig;
+import org.apache.rocketmq.connect.neo4j.helper.LabelTypeEnum;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jClient;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.source.SourceConnector;
+import io.openmessaging.internal.DefaultKeyValue;
+
+public class Neo4jSourceConnector extends SourceConnector {
+ private KeyValue keyValue;
+ private Neo4jClient neo4jClient;
+
+ @Override
+ public List taskConfigs(int maxTasks) {
+ final String type = keyValue.getString(Neo4jConstants.LABEL_TYPE);
+ final String labels = keyValue.getString(Neo4jConstants.LABELS);
+ List labelList = null;
+ if (labels != null && !labels.equals("")) {
+ final String[] split = labels.split(",");
+ labelList = Arrays.asList(split);
+ }
+ if (labelList == null) {
+ if (LabelTypeEnum.node.name().equals(type)) {
+ labelList = neo4jClient.getAllLabels();
+ } else {
+ labelList = neo4jClient.getAllType();
+ }
+ }
+ // 按标签分配任务
+ List configs = new ArrayList<>();
+ for (int i = 0; i < labelList.size(); i++) {
+ final String label = labelList.get(i);
+ KeyValue config = new DefaultKeyValue();
+ for (String key : keyValue.keySet()) {
+ config.put(key, keyValue.getString(key));
+ }
+ config.put(Neo4jConstants.LABEL, label);
+ configs.add(config);
+ }
+ return configs;
+ }
+
+ @Override
+ public Class taskClass() {
+ return Neo4jSourceTask.class;
+ }
+
+ @Override
+ public void start(KeyValue keyValue) {
+ for (String requestKey : Neo4jSourceConfig.REQUEST_CONFIG) {
+ if (!keyValue.containsKey(requestKey)) {
+ throw new RuntimeException("Request config key: " + requestKey);
+ }
+ }
+ Neo4jSourceConfig config = new Neo4jSourceConfig();
+ config.load(keyValue);
+ final LabelTypeEnum labelTypeEnum = LabelTypeEnum.nameOf(config.getLabelType());
+ if (labelTypeEnum == null) {
+ throw new RuntimeException("labelType is only support node or relationship!");
+ }
+ neo4jClient = new Neo4jClient(config);
+ if (!neo4jClient.ping()) {
+ throw new RuntimeException("Cannot connect to neo4j server!");
+ }
+ this.keyValue = keyValue;
+ }
+
+ @Override
+ public void stop() {
+ this.keyValue = null;
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceTask.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceTask.java
new file mode 100644
index 00000000..f5fc329a
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceTask.java
@@ -0,0 +1,140 @@
+/*
+ * 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.neo4j.source;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jConstants;
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSourceConfig;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jClient;
+import org.apache.rocketmq.connect.neo4j.source.query.CqlQueryProcessor;
+import org.apache.rocketmq.connect.neo4j.source.query.QueryRegistrar;
+
+import org.neo4j.driver.Record;
+import org.neo4j.driver.util.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+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.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+
+public class Neo4jSourceTask extends SourceTask {
+ private static final Logger log = LoggerFactory.getLogger(Neo4jSourceTask.class);
+
+ private Neo4jSourceConfig config;
+
+ private Neo4jClient client;
+
+ @Override
+ public List poll() {
+ List res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String cql = buildCql(offset);
+ try {
+ final List recordList = client.query(cql);
+ for (Record record : recordList) {
+ final ConnectRecord connectRecord = neo4jRecord2ConnectRecord(record);
+ res.add(connectRecord);
+ }
+ } catch (Exception e) {
+ log.error(String.format("Fail to poll data from neo4j! offset=%d", offset));
+ }
+ return res;
+ }
+
+ @Override
+ public void start(KeyValue keyValue) {
+ this.config = new Neo4jSourceConfig();
+ this.config.load(keyValue);
+ this.client = new Neo4jClient(this.config);
+ if (!client.ping()) {
+ throw new RuntimeException("Cannot connect to neo4j server!");
+ }
+ QueryRegistrar.register(config);
+ }
+
+ @Override
+ public void stop() {
+ this.client = null;
+ }
+
+ // build cypher query
+ private String buildCql(long offset) {
+ final String queryType = config.getLabelType();
+ final CqlQueryProcessor cqlQueryProcessor = QueryRegistrar.querySqlBuilder(queryType);
+ return cqlQueryProcessor.buildCql(offset);
+ }
+
+ private long readRecordOffset() {
+ if (sourceTaskContext == null) {
+ return 0L;
+ }
+ final RecordOffset positionInfo =
+ this.sourceTaskContext.offsetStorageReader().readOffset(buildRecordPartition(config.getNeo4jDataBase()));
+ if (positionInfo == null) {
+ return 0;
+ }
+ Object offset = positionInfo.getOffset().get(config.getTaskName() + "_" + Neo4jConstants.NEO4J_OFFSET);
+ return offset == null ? 0 : Long.parseLong(offset.toString());
+ }
+
+ private ConnectRecord neo4jRecord2ConnectRecord(Record record) {
+ final String queryType = config.getLabelType();
+ final CqlQueryProcessor cqlQueryProcessor = QueryRegistrar.querySqlBuilder(queryType);
+ final Pair structPair = cqlQueryProcessor.buildStruct(record);
+ final Long id = structPair.key();
+ final Struct struct = structPair.value();
+ final ConnectRecord connectRecord =
+ new ConnectRecord(buildRecordPartition(config.getTaskName()), buildRecordOffset(id),
+ System.currentTimeMillis(), struct.getSchema(), struct);
+
+ connectRecord.setExtensions(this.buildExtensions());
+ return connectRecord;
+ }
+
+ private RecordPartition buildRecordPartition(String partitionValue) {
+ Map partitionMap = new HashMap<>();
+ partitionMap.put(Neo4jConstants.NEO4J_PARTITION, partitionValue);
+ return new RecordPartition(partitionMap);
+ }
+
+ private KeyValue buildExtensions() {
+ KeyValue keyValue = new DefaultKeyValue();
+ String topicName = config.getTopic();
+ if (topicName == null || topicName.equals("")) {
+ String connectorName = this.sourceTaskContext.getConnectorName();
+ topicName = config.getTaskName() + "_" + connectorName;
+ }
+ keyValue.put(Neo4jConstants.NEO4J_TOPIC, topicName);
+ return keyValue;
+ }
+
+ private RecordOffset buildRecordOffset(long offset) {
+ Map offsetMap = new HashMap<>();
+ offsetMap.put(config.getTaskName() + "_" + Neo4jConstants.NEO4J_OFFSET, offset);
+ return new RecordOffset(offsetMap);
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/mapping/ReadMapper.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/mapping/ReadMapper.java
new file mode 100644
index 00000000..f47b0ccc
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/mapping/ReadMapper.java
@@ -0,0 +1,106 @@
+/*
+ * 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.neo4j.source.mapping;
+
+import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import org.apache.rocketmq.connect.neo4j.helper.LabelTypeEnum;
+import org.apache.rocketmq.connect.neo4j.helper.MappingRule;
+import org.apache.rocketmq.connect.neo4j.helper.ValueType;
+
+public class ReadMapper {
+
+ public static BiConsumer, Map> getMapper(MappingRule rule) {
+ return (neo4jElement, record) -> rule.getColumns().forEach(columnMappingRule -> {
+ Object value = null;
+ ValueType type = columnMappingRule.getValueType();
+ String name = columnMappingRule.getName();
+ final String valueExtract = columnMappingRule.getValueExtract();
+
+ switch (columnMappingRule.getColumnType()) {
+ case dstPrimaryKey:
+ final Map targetMap = (Map)neo4jElement.get("");
+ value = extractValue(valueExtract, "", targetMap::get, type);
+ record.put(name, value);
+ break;
+ case srcPrimaryKey:
+ final Map sourceMap = (Map)neo4jElement.get("");
+ value = extractValue(valueExtract, "", sourceMap::get, type);
+ record.put(name, value);
+ break;
+ case primaryKey:
+ final LabelTypeEnum labelTypeEnum = rule.getType();
+ if (LabelTypeEnum.node == labelTypeEnum) {
+ value = extractValue(valueExtract, "", neo4jElement::get, type);
+ } else {
+ value = extractValue(valueExtract, "", neo4jElement::get, type);
+ }
+
+ record.put(name, value);
+ break;
+ case primaryLabel:
+ final LabelTypeEnum exportType = rule.getType();
+ if (LabelTypeEnum.node == exportType) {
+ value = neo4jElement.get("");
+ } else {
+ value = neo4jElement.get("");
+ }
+ record.put(name, value);
+ break;
+ case dstPrimaryLabel:
+ final Map dstTargetMap = (Map)neo4jElement.get("");
+ value = extractValue(valueExtract, "", dstTargetMap::get, type);
+ record.put(name, value);
+ break;
+ case srcPrimaryLabel:
+ final Map srcSourceMap = (Map)neo4jElement.get("");
+ value = extractValue(valueExtract, "", srcSourceMap::get, type);
+ record.put(name, value);
+ break;
+ case nodeProperty:
+ case relationshipProperty:
+ value = extractValue(valueExtract, name, neo4jElement::get, type);
+ record.put(name, value);
+ break;
+ case relationshipJsonProperty:
+ case nodeJsonProperty:
+ value = extractValue(valueExtract, name, neo4jElement::get, type);
+ record.put(name, type.applyObject(value));
+ break;
+ default:
+ break;
+ }
+ });
+ }
+
+ private static Object extractValue(String valueExtract, String defaultExtract, Function function,
+ ValueType valueType) {
+ if (valueExtract == null) {
+ return function.apply(defaultExtract);
+ } else {
+ if (valueExtract.startsWith("#{") && valueExtract.endsWith("}")) {
+ final String val = valueExtract.substring(2, valueExtract.length() - 1);
+ return function.apply(val);
+ } else {
+ return valueType.applyObject(valueExtract);
+ }
+ }
+ }
+
+}
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/CqlQueryProcessor.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/CqlQueryProcessor.java
new file mode 100644
index 00000000..e2105697
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/CqlQueryProcessor.java
@@ -0,0 +1,31 @@
+/*
+ * 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.neo4j.source.query;
+
+import org.neo4j.driver.Record;
+import org.neo4j.driver.util.Pair;
+
+import io.openmessaging.connector.api.data.Struct;
+
+public interface CqlQueryProcessor {
+ String buildCql(long offset);
+
+ Pair buildStruct(Record record);
+
+ String queryType();
+}
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/NodeQueryStrategy.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/NodeQueryStrategy.java
new file mode 100644
index 00000000..207d4f39
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/NodeQueryStrategy.java
@@ -0,0 +1,88 @@
+/*
+ * 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.neo4j.source.query;
+
+import java.util.*;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jConstants;
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSourceConfig;
+import org.apache.rocketmq.connect.neo4j.helper.LabelTypeEnum;
+import org.apache.rocketmq.connect.neo4j.helper.RecordConverter;
+import org.apache.rocketmq.connect.neo4j.source.mapping.ReadMapper;
+import org.apache.rocketmq.connect.neo4j.helper.MappingRule;
+import org.apache.rocketmq.connect.neo4j.helper.MappingRuleFactory;
+import org.neo4j.driver.Record;
+import org.neo4j.driver.internal.InternalPair;
+import org.neo4j.driver.types.Node;
+import org.neo4j.driver.util.Pair;
+
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.Struct;
+
+// match (n:Person) where id(n) > 10000 return n order by id(n) limit 1000;
+// match (n:Person) return n order by id(n) skip 10000 limit 1000;
+// Performance optimization of the above two query statements with the same function
+
+public class NodeQueryStrategy implements CqlQueryProcessor {
+ private final Neo4jSourceConfig neo4jSourceConfig;
+ private final MappingRule mappingRule;
+
+ public NodeQueryStrategy(Neo4jSourceConfig neo4jSourceConfig) {
+ this.neo4jSourceConfig = neo4jSourceConfig;
+ final LabelTypeEnum labelTypeEnum = LabelTypeEnum.nameOf(neo4jSourceConfig.getLabelType());
+ this.mappingRule = MappingRuleFactory.getInstance().create(neo4jSourceConfig, labelTypeEnum);
+ mappingRule.validateConfig();
+ }
+
+ @Override
+ public String buildCql(long offset) {
+ StringBuilder cql = new StringBuilder();
+ cql.append("use ").append(neo4jSourceConfig.getNeo4jDataBase()).append(" ");
+ cql.append("match (n:").append(neo4jSourceConfig.getLabel()).append(")");
+ cql.append("where id(n) >=").append(offset);
+ cql.append(" return n ").append(" limit ").append(Neo4jConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME);
+ return cql.toString();
+ }
+
+ public Pair buildStruct(Record record) {
+ final Map nMap = record.asMap();
+ final Object value = nMap.get("n");
+ if (!(value instanceof Node)) {
+ throw new RuntimeException("cypher query Inconsistent");
+ }
+ Node node = (Node)value;
+ final Iterable labels = node.labels();
+ final Map nodeMap = RecordConverter.asNodeMap(node);
+ Map columnMap = new LinkedHashMap<>();
+ ReadMapper.getMapper(mappingRule).accept(nodeMap, columnMap);
+ final Schema valueSchema = RecordConverter.map2StructSchema(columnMap);
+ StringBuilder labelName = new StringBuilder();
+ final Iterator iterator = labels.iterator();
+ while (iterator.hasNext()) {
+ labelName.append("_").append(iterator.next());
+ }
+ valueSchema.setName(labelName.substring(1));
+ Struct struct = RecordConverter.buildStruct(valueSchema, columnMap);
+ return InternalPair.of(node.id(), struct);
+ }
+
+ @Override
+ public String queryType() {
+ return LabelTypeEnum.node.name();
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/QueryRegistrar.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/QueryRegistrar.java
new file mode 100644
index 00000000..30315a0a
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/QueryRegistrar.java
@@ -0,0 +1,40 @@
+/*
+ * 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.neo4j.source.query;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSourceConfig;
+
+public class QueryRegistrar {
+ private static final Map registrar = new HashMap<>();
+
+ public static void register(Neo4jSourceConfig neo4jSourceConfig) {
+ doRegister(new NodeQueryStrategy(neo4jSourceConfig));
+ doRegister(new RelationshipQueryStrategy(neo4jSourceConfig));
+ }
+
+ static void doRegister(CqlQueryProcessor cqlQueryProcessor) {
+ registrar.put(cqlQueryProcessor.queryType(), cqlQueryProcessor);
+ }
+
+ public static CqlQueryProcessor querySqlBuilder(String queryType) {
+ return registrar.get(queryType);
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/RelationshipQueryStrategy.java b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/RelationshipQueryStrategy.java
new file mode 100644
index 00000000..03f25ac1
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/main/java/org/apache/rocketmq/connect/neo4j/source/query/RelationshipQueryStrategy.java
@@ -0,0 +1,87 @@
+/*
+ * 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.neo4j.source.query;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jConstants;
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSourceConfig;
+import org.apache.rocketmq.connect.neo4j.helper.LabelTypeEnum;
+import org.apache.rocketmq.connect.neo4j.helper.RecordConverter;
+import org.apache.rocketmq.connect.neo4j.source.mapping.ReadMapper;
+import org.apache.rocketmq.connect.neo4j.helper.MappingRule;
+import org.apache.rocketmq.connect.neo4j.helper.MappingRuleFactory;
+import org.neo4j.driver.Record;
+import org.neo4j.driver.internal.InternalPair;
+import org.neo4j.driver.types.Node;
+import org.neo4j.driver.types.Relationship;
+import org.neo4j.driver.util.Pair;
+
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.Struct;
+
+public class RelationshipQueryStrategy implements CqlQueryProcessor {
+ private Neo4jSourceConfig neo4jSourceConfig;
+ private MappingRule mappingRule;
+
+ public RelationshipQueryStrategy(Neo4jSourceConfig config) {
+ this.neo4jSourceConfig = config;
+ final LabelTypeEnum labelTypeEnum = LabelTypeEnum.nameOf(neo4jSourceConfig.getLabelType());
+ this.mappingRule = MappingRuleFactory.getInstance().create(neo4jSourceConfig, labelTypeEnum);
+ mappingRule.validateConfig();
+ }
+
+ @Override
+ public String buildCql(long offset) {
+ StringBuilder cql = new StringBuilder();
+ cql.append("use ").append(neo4jSourceConfig.getNeo4jDataBase()).append(" ");
+ cql.append("match (source)-[r:").append(neo4jSourceConfig.getLabel()).append("]->(target)");
+ cql.append(" where id(r) >=").append(offset);
+ cql.append(" return source , r, target ").append(" limit ")
+ .append(Neo4jConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME);
+ return cql.toString();
+ }
+
+ @Override
+ public Pair buildStruct(Record record) {
+ final Map recordMap = record.asMap();
+ final Object value = recordMap.get("r");
+ if (!(value instanceof Relationship)) {
+ throw new RuntimeException("cypher query Inconsistent");
+ }
+ final Node source = (Node)recordMap.get("source");
+ final Map sourceMap = RecordConverter.asNodeMap(source);
+ final Node target = (Node)recordMap.get("target");
+ final Map targetMap = RecordConverter.asNodeMap(target);
+ Relationship relationship = (Relationship)value;
+ final Map relationshipMap = RecordConverter.asRelationshipMap(relationship);
+ relationshipMap.put("", sourceMap);
+ relationshipMap.put("", targetMap);
+ Map columnMap = new HashMap<>();
+ ReadMapper.getMapper(mappingRule).accept(relationshipMap, columnMap);
+ final Schema valueSchema = RecordConverter.map2StructSchema(columnMap);
+ valueSchema.setName(relationship.type());
+ Struct struct = RecordConverter.buildStruct(valueSchema, columnMap);
+ return InternalPair.of(relationship.id(), struct);
+ }
+
+ @Override
+ public String queryType() {
+ return LabelTypeEnum.relationship.name();
+ }
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/test/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkTaskTest.java b/connectors/rocketmq-connect-neo4j/src/test/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkTaskTest.java
new file mode 100644
index 00000000..08ee88e4
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/test/java/org/apache/rocketmq/connect/neo4j/sink/Neo4jSinkTaskTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.neo4j.sink;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jConstants;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.*;
+import io.openmessaging.internal.DefaultKeyValue;
+
+public class Neo4jSinkTaskTest {
+
+// private static final String host= "localhost";
+// private static final Integer port = 7687;
+// private static final String db = "tmp";
+// private static final String user = "test";
+// private static final String password = "root123456";
+//
+// public static void main(String[] args) {
+// List records = new ArrayList<>();
+// // build schema
+// Schema schema = SchemaBuilder.struct()
+// .name("tableName")
+// .field("label",SchemaBuilder.string().build())
+// .field("recordId", SchemaBuilder.string().build())
+// .build();
+//
+//
+// for (int i = 0; i < 4; i++) {
+// // build record
+// String param0 = "Record";
+// Struct struct= new Struct(schema);
+// struct.put("label", param0);
+// struct.put("recordId",String.valueOf(i));
+// ConnectRecord record = new ConnectRecord(
+// // offset partition
+// // offset partition"
+// new RecordPartition(new ConcurrentHashMap<>()),
+// new RecordOffset(new HashMap<>()),
+// System.currentTimeMillis(),
+// schema,
+// struct
+// );
+// records.add(record);
+//
+// }
+// Neo4jSinkTask task = new Neo4jSinkTask();
+// KeyValue config = new DefaultKeyValue();
+// config.put(Neo4jConstants.NEO4J_HOST, host);
+// config.put(Neo4jConstants.NEO4J_PORT, port);
+// config.put(Neo4jConstants.NEO4J_USER, user);
+// config.put(Neo4jConstants.NEO4J_PASSWORD, password);
+// config.put(Neo4jConstants.NEO4J_DB, db);
+// config.put(Neo4jConstants.LABEL_TYPE, "node");
+// config.put(Neo4jConstants.COLUMN, "[\n" + " {\n" + " \"name\":\"recordId\",\n"
+// + " \"type\":\"long\",\n" + " \"columnType\":\"primaryKey\",\n"
+// + " \"valueExtract\":\"#{recordId}\"\n" + " },\n" + " {\n"
+// + " \"name\":\"Goods\",\n" + " \"type\":\"string\",\n"
+// + " \"columnType\":\"primaryLabel\",\n" + " \"valueExtract\":\"#{label}\"\n"
+// + " }\n" + " ]");
+//
+// task.start(config);
+// task.put(records);
+//
+// }
+
+}
\ No newline at end of file
diff --git a/connectors/rocketmq-connect-neo4j/src/test/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceTaskTest.java b/connectors/rocketmq-connect-neo4j/src/test/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceTaskTest.java
new file mode 100644
index 00000000..f6cb8c00
--- /dev/null
+++ b/connectors/rocketmq-connect-neo4j/src/test/java/org/apache/rocketmq/connect/neo4j/source/Neo4jSourceTaskTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.neo4j.source;
+
+import java.util.List;
+
+import org.apache.rocketmq.connect.neo4j.config.Neo4jBaseConfig;
+import org.apache.rocketmq.connect.neo4j.config.Neo4jConstants;
+import org.apache.rocketmq.connect.neo4j.config.Neo4jSourceConfig;
+import org.apache.rocketmq.connect.neo4j.helper.Neo4jClient;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.internal.DefaultKeyValue;
+
+public class Neo4jSourceTaskTest {
+// private static final String host= "localhost";
+// private static final Integer port = 7687;
+// private static final String db = "test";
+// private static final String user = "test";
+// private static final String password = "root123456";
+//
+// public void testClient() {
+// Neo4jBaseConfig neo4jBaseConfig = new Neo4jSourceConfig();
+// KeyValue config = new DefaultKeyValue();
+// config.put(Neo4jConstants.NEO4J_HOST, host);
+// config.put(Neo4jConstants.NEO4J_PORT, port);
+// config.put(Neo4jConstants.NEO4J_USER, user);
+// config.put(Neo4jConstants.NEO4J_PASSWORD, password);
+// config.put(Neo4jConstants.NEO4J_DB, db);
+// neo4jBaseConfig.load(config);
+// Neo4jClient client = new Neo4jClient(neo4jBaseConfig);
+// final boolean ping = client.ping();
+// System.out.println(ping);
+// }
+//
+// public void testPoll() throws InterruptedException {
+// Neo4jSourceTask task = new Neo4jSourceTask();
+// KeyValue config = new DefaultKeyValue();
+// config.put(Neo4jConstants.NEO4J_HOST, host);
+// config.put(Neo4jConstants.NEO4J_PORT, port);
+// config.put(Neo4jConstants.NEO4J_USER, user);
+// config.put(Neo4jConstants.NEO4J_PASSWORD, password);
+// config.put(Neo4jConstants.NEO4J_DB, db);
+// config.put(Neo4jConstants.NEO4J_TOPIC, "nodeNeo4jTopic");
+// config.put(Neo4jConstants.LABEL_TYPE, "node");
+// config.put(Neo4jConstants.LABELS, "Goods");
+// config.put(Neo4jConstants.COLUMN, "[\n" + " {\n" + " \"name\":\"goodsId\",\n"
+// + " \"type\":\"long\",\n" + " \"columnType\":\"primaryKey\",\n"
+// + " \"valueExtract\":\"#{goodsId}\"\n" + " },\n" + " {\n"
+// + " \"name\":\"label\",\n" + " \"type\":\"string\",\n"
+// + " \"columnType\":\"primaryLabel\"\n" + " },\n" + " {\n"
+// + " \"name\":\"goodsName\",\n" + " \"type\":\"string\",\n"
+// + " \"columnType\":\"nodeProperty\",\n" + " \"valueExtract\":\"#{goodsName}\"\n"
+// + " }\n" + " ]");
+// task.start(config);
+// while (true) {
+// final List connectRecordList = task.poll();
+// for(ConnectRecord record : connectRecordList) {
+// System.out.println(record);
+// }
+// Thread.sleep(3000);
+// }
+// }
+//
+// public static void main(String[] args) throws InterruptedException {
+// Neo4jSourceTaskTest neo4jSourceTaskTest = new Neo4jSourceTaskTest();
+// neo4jSourceTaskTest.testClient();
+// neo4jSourceTaskTest.testPoll();
+// }
+}
\ No newline at end of file