[Feature][Connector-v2] Neo4j sink connector (#2434)

* ConnectorV2 - Neo4j Sink

docs/en/connector-v2/sink/Neo4j.md

@@ -0,0 +1,82 @@
+# Neo4j
+
+> Neo4j sink connector
+
+## Description
+
+Write data to Neo4j. 
+
+`neo4j-java-driver` version 4.4.9
+
+## Options
+
+| name                       | type   | required | default value |
+|----------------------------|--------|----------|---------------|
+| uri                        | String | Yes      | -             |
+| username                   | String | No       | -             |
+| password                   | String | No       | -             |
+| bearer_token               | String | No       | -             |
+| kerberos_ticket            | String | No       | -             |
+| database                   | String | Yes      | -             |
+| query                      | String | Yes      | -             |
+| queryParamPosition         | Object | Yes      | -             |
+| max_transaction_retry_time | Long   | No       | 30            |
+| max_connection_timeout     | Long   | No       | 30            |
+
+
+### uri [string]
+The URI of the Neo4j database. Refer to a case: `neo4j://localhost:7687`
+
+### username [string]
+username of the Neo4j
+
+### password [string]
+password of the Neo4j. required if `username` is provided
+
+### bearer_token [string]
+base64 encoded bearer token of the Neo4j. for Auth. 
+
+### kerberos_ticket [string]
+base64 encoded kerberos ticket of the Neo4j. for Auth.
+
+### database [string]
+database name.
+
+### query [string]
+Query statement. contain parameter placeholders that are substituted with the corresponding values at runtime
+
+### queryParamPosition [object]
+position mapping information for query parameters.
+
+key name is parameter placeholder name.
+
+associated value is position of field in input data row. 
+
+
+### max_transaction_retry_time [long]
+maximum transaction retry time(seconds). transaction fail if exceeded
+
+### max_connection_timeout [long]
+The maximum amount of time to wait for a TCP connection to be established (seconds)
+
+
+## Example
+```
+sink {
+  Neo4j {
+    uri = "neo4j://localhost:7687"
+    username = "neo4j"
+    password = "1234"
+    database = "neo4j"
+
+    max_transaction_retry_time = 10
+    max_connection_timeout = 10
+
+    query = "CREATE (a:Person {name: $name, age: $age})"
+    queryParamPosition = {
+        name = 0
+        age = 1
+    }
+  }
+}
+```

plugin-mapping.properties

@@ -117,3 +117,4 @@ seatunnel.sink.DingTalk = connector-dingtalk
 seatunnel.sink.elasticsearch = connector-elasticsearch
 seatunnel.source.IoTDB = connector-iotdb
 seatunnel.sink.IoTDB = connector-iotdb
+seatunnel.sink.Neo4j = connector-neo4j

pom.xml

@@ -227,6 +227,7 @@
         <checker.qual.version>3.10.0</checker.qual.version>
         <iotdb.version>0.13.1</iotdb.version>
         <awaitility.version>4.2.0</awaitility.version>
+        <neo4j-java-driver.version>4.4.9</neo4j-java-driver.version>
     </properties>
 
     <dependencyManagement>

seatunnel-connectors-v2-dist/pom.xml

@@ -131,6 +131,11 @@
             <artifactId>connector-iotdb</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-neo4j</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 
     <build>

seatunnel-connectors-v2/connector-neo4j/pom.xml

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

seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/DriverBuilder.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.seatunnel.connectors.seatunnel.neo4j.config;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.neo4j.driver.AuthTokens;
+import org.neo4j.driver.Config;
+import org.neo4j.driver.Driver;
+import org.neo4j.driver.GraphDatabase;
+
+import java.io.Serializable;
+import java.net.URI;
+import java.util.concurrent.TimeUnit;
+
+@Getter
+@Setter
+public class DriverBuilder implements Serializable {
+    private final URI uri;
+    private String username;
+    private String password;
+    private String bearerToken;
+    private String kerberosTicket;
+    private String database;
+
+    private Long maxTransactionRetryTimeSeconds;
+    private Long maxConnectionTimeoutSeconds;
+
+    public static DriverBuilder create(URI uri) {
+        return new DriverBuilder(uri);
+    }
+
+    private DriverBuilder(URI uri) {
+        this.uri = uri;
+    }
+
+    public Driver build() {
+        final Config.ConfigBuilder configBuilder = Config.builder()
+                .withMaxConnectionPoolSize(1);
+        if (maxConnectionTimeoutSeconds != null) {
+            configBuilder
+                    .withConnectionAcquisitionTimeout(maxConnectionTimeoutSeconds * 2, TimeUnit.SECONDS)
+                    .withConnectionTimeout(maxConnectionTimeoutSeconds, TimeUnit.SECONDS);
+        }
+        if (maxTransactionRetryTimeSeconds != null) {
+            configBuilder
+                    .withMaxTransactionRetryTime(maxTransactionRetryTimeSeconds, TimeUnit.SECONDS);
+        }
+        Config config = configBuilder
+                .build();
+
+        if (username != null) {
+            return GraphDatabase.driver(uri, AuthTokens.basic(username, password), config);
+        } else if (bearerToken != null) {
+            return GraphDatabase.driver(uri, AuthTokens.bearer(bearerToken), config);
+        } else if (kerberosTicket != null){
+            return GraphDatabase.driver(uri, AuthTokens.kerberos(kerberosTicket), config);
+        }
+        throw new IllegalArgumentException("Invalid Field");
+    }
+}

seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jConfig.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.neo4j.config;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.io.Serializable;
+import java.util.Map;
+
+@Getter
+@Setter
+public class Neo4jConfig implements Serializable {
+
+    public static final String PLUGIN_NAME = "Neo4j";
+    public static final String KEY_NEO4J_URI = "uri";
+    public static final String KEY_USERNAME = "username";
+    public static final String KEY_PASSWORD = "password";
+    public static final String KEY_BEARER_TOKEN = "bearer_token";
+    public static final String KEY_KERBEROS_TICKET = "kerberos_ticket"; // Base64 encoded
+
+    public static final String KEY_DATABASE = "database";
+    public static final String KEY_QUERY = "query";
+    public static final String KEY_QUERY_PARAM_POSITION = "queryParamPosition";
+    public static final String KEY_MAX_TRANSACTION_RETRY_TIME = "max_transaction_retry_time";
+    public static final String KEY_MAX_CONNECTION_TIMEOUT = "max_connection_timeout";
+
+
+    private DriverBuilder driverBuilder;
+    private String query;
+    private Map<String, Object> queryParamPosition;
+
+}