fixes #375: Source Plugin for Kafka Connect (#496)

* fixes #375: Source Plugin for Kafka Connect

* feedback

* fixed Neo4j version in deprecation message

Author: conker84

common/src/main/kotlin/streams/extensions/CommonExtensions.kt

@@ -8,9 +8,10 @@ import org.apache.avro.generic.IndexedRecord
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.clients.consumer.OffsetAndMetadata
 import org.apache.kafka.common.TopicPartition
+import org.neo4j.driver.types.Relationship
 import org.neo4j.graphdb.Node
-import streams.utils.JSONUtils
 import streams.service.StreamsSinkEntity
+import streams.utils.JSONUtils
 import java.nio.ByteBuffer
 import java.util.*
 import javax.lang.model.SourceVersion
@@ -26,6 +27,22 @@ fun Node.labelNames() : List<String> {
     return this.labels.map { it.name() }
 }
 
+fun org.neo4j.driver.types.Node.asStreamsMap(): Map<String, Any?> {
+    val nodeMap = this.asMap().toMutableMap()
+    nodeMap["<id>"] = this.id()
+    nodeMap["<labels>"] = this.labels()
+    return nodeMap
+}
+
+fun Relationship.asStreamsMap(): Map<String, Any?> {
+    val relMap = this.asMap().toMutableMap()
+    relMap["<id>"] = this.id()
+    relMap["<type>"] = this.type()
+    relMap["<source.id>"] = this.startNodeId()
+    relMap["<target.id>"] = this.endNodeId()
+    return relMap
+}
+
 fun String.toPointCase(): String {
     return this.split("(?<=[a-z])(?=[A-Z])".toRegex()).joinToString(separator = ".").toLowerCase()
 }

common/src/main/kotlin/streams/utils/JSONUtils.kt

@@ -12,6 +12,7 @@ import org.neo4j.driver.internal.value.PointValue
 import org.neo4j.graphdb.spatial.Point
 import org.neo4j.values.storable.CoordinateReferenceSystem
 import streams.events.*
+import streams.extensions.asStreamsMap
 import java.io.IOException
 import java.time.temporal.TemporalAccessor
 import kotlin.reflect.full.isSubclassOf
@@ -43,6 +44,17 @@ fun PointValue.toStreamsPoint(): StreamsPoint {
     }
 }
 
+fun org.neo4j.driver.types.Point.toStreamsPoint(): StreamsPoint {
+    val point = this
+    return when (val crsType = point.srid()) {
+        CoordinateReferenceSystem.Cartesian.code -> StreamsPointCartesian(CoordinateReferenceSystem.Cartesian.name, point.x(), point.y())
+        CoordinateReferenceSystem.Cartesian_3D.code -> StreamsPointCartesian(CoordinateReferenceSystem.Cartesian_3D.name, point.x(), point.y(), point.z())
+        CoordinateReferenceSystem.WGS84.code -> StreamsPointWgs(CoordinateReferenceSystem.WGS84.name, point.x(), point.y())
+        CoordinateReferenceSystem.WGS84_3D.code -> StreamsPointWgs(CoordinateReferenceSystem.WGS84_3D.name, point.x(), point.y(), point.z())
+        else -> throw IllegalArgumentException("Point type $crsType not supported")
+    }
+}
+
 class PointSerializer : JsonSerializer<Point>() {
     @Throws(IOException::class, JsonProcessingException::class)
     override fun serialize(value: Point?, jgen: JsonGenerator,
@@ -54,6 +66,17 @@ class PointSerializer : JsonSerializer<Point>() {
     }
 }
 
+class DriverPointSerializer : JsonSerializer<org.neo4j.driver.types.Point>() {
+    @Throws(IOException::class, JsonProcessingException::class)
+    override fun serialize(value: org.neo4j.driver.types.Point?, jgen: JsonGenerator,
+                           provider: SerializerProvider) {
+        if (value == null) {
+            return
+        }
+        jgen.writeObject(value.toStreamsPoint())
+    }
+}
+
 class PointValueSerializer : JsonSerializer<PointValue>() {
     @Throws(IOException::class, JsonProcessingException::class)
     override fun serialize(value: PointValue?, jgen: JsonGenerator,
@@ -76,6 +99,27 @@ class TemporalAccessorSerializer : JsonSerializer<TemporalAccessor>() {
     }
 }
 
+class DriverNodeSerializer : JsonSerializer<org.neo4j.driver.types.Node>() {
+    @Throws(IOException::class, JsonProcessingException::class)
+    override fun serialize(value: org.neo4j.driver.types.Node?, jgen: JsonGenerator,
+                           provider: SerializerProvider) {
+        if (value == null) {
+            return
+        }
+        jgen.writeObject(value.asStreamsMap())
+    }
+}
+
+class DriverRelationshipSerializer : JsonSerializer<org.neo4j.driver.types.Relationship>() {
+    @Throws(IOException::class, JsonProcessingException::class)
+    override fun serialize(value: org.neo4j.driver.types.Relationship?, jgen: JsonGenerator,
+                           provider: SerializerProvider) {
+        if (value == null) {
+            return
+        }
+        jgen.writeObject(value.asStreamsMap())
+    }
+}
 
 object JSONUtils {
 
@@ -86,6 +130,9 @@ object JSONUtils {
         val module = SimpleModule("Neo4jKafkaSerializer")
         StreamsUtils.ignoreExceptions({ module.addSerializer(Point::class.java, PointSerializer()) }, NoClassDefFoundError::class.java) // in case is loaded from
         StreamsUtils.ignoreExceptions({ module.addSerializer(PointValue::class.java, PointValueSerializer()) }, NoClassDefFoundError::class.java) // in case is loaded from
+        StreamsUtils.ignoreExceptions({ module.addSerializer(org.neo4j.driver.types.Point::class.java, DriverPointSerializer()) }, NoClassDefFoundError::class.java) // in case is loaded from
+        StreamsUtils.ignoreExceptions({ module.addSerializer(org.neo4j.driver.types.Node::class.java, DriverNodeSerializer()) }, NoClassDefFoundError::class.java) // in case is loaded from
+        StreamsUtils.ignoreExceptions({ module.addSerializer(org.neo4j.driver.types.Relationship::class.java, DriverRelationshipSerializer()) }, NoClassDefFoundError::class.java) // in case is loaded from
         module.addSerializer(TemporalAccessor::class.java, TemporalAccessorSerializer())
         OBJECT_MAPPER.registerModule(module)
         OBJECT_MAPPER.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)

common/src/main/kotlin/streams/utils/StreamsUtils.kt

@@ -62,4 +62,10 @@ object StreamsUtils {
 
     fun getName(db: GraphDatabaseService) = db.databaseName()
 
+    fun closeSafetely(closeable: AutoCloseable, onError: (Throwable) -> Unit) = try {
+        closeable.close()
+    } catch (e: Throwable) {
+        onError(e)
+    }
+
 }

doc/docs/modules/ROOT/nav.adoc

@@ -2,11 +2,11 @@
 
 * xref::overview.adoc[Project overview]
 // ** xref::overview.adoc#neo4j_streams_plugin_overview[Neo4j Streams plugin]
-// ** xref::overview.adoc#kafka_connect_plugin_overview[Kafka Connect plugin]
+// ** xref::overview.adoc#kafka_connect_neo4j_connector_overview[Kafka Connect Neo4j Connector]
 
 * xref::quickstart.adoc[Quick Start]
 // ** xref::quickstart.adoc#neo4j_streams_plugin_quickstart[Neo4j Streams plugin]
-// ** xref::quickstart.adoc#kafka_connect_plugin_quickstart[Kafka Connect plugin]
+// ** xref::quickstart.adoc#kafka_connect_neo4j_connector_quickstart[Kafka Connect Neo4j Connector]
 
 * xref::configuration.adoc[The Configuration System]
 // ** xref::configuration.adoc#neo4j_configuration_system[Configuration System overview]
@@ -28,8 +28,8 @@
 // ** xref::procedures.adoc#_streams_publish[streams.publish]
 // ** xref::procedures.adoc#_streams_consume[streams.consume]
 
-* xref::kafka-connect.adoc[Kafka Connect Plugin]
-// ** xref::kafka-connect.adoc#kafka_connect_plugin_install[Plugin installation]
+* xref::kafka-connect.adoc[Kafka Connect Neo4j Connector]
+// ** xref::kafka-connect.adoc#kafka_connect_neo4j_connector_install[Plugin installation]
 // ** xref::kafka-connect.adoc#kafka-connect-sink-instance[Create the Sink Instance]
 // ** xref::kafka-connect.adoc#kafka-connect-sink-strategies[Sink ingestion strategies]
 // ** xref::kafka-connect.adoc#kafka-connect-cud-file-format[How deal with bad data]
@@ -45,7 +45,7 @@
 
 * xref::docker.adoc[Run with Docker]
 // ** xref::docker.adoc#neo4j_streams_docker[Neo4j Streams plugin]
-// ** xref::docker.adoc#docker_kafka_connect[Kafka Connect Plugin]
+// ** xref::docker.adoc#docker_kafka_connect[Kafka Connect Neo4j Connector]
 // ** xref::docker.adoc#docker_streams_cluster[Neo4j Streams with Neo4j Cluster and Kafka Cluster]
 
 * xref::kafka-ssl.adoc[Configure with Kafka over SSL]

doc/docs/modules/ROOT/pages/architecture/pluginvsconnect.adoc

@@ -1,4 +1,4 @@
-= When to use Kafka Connect vs. Neo4j Streams as a Plugin
+= When to use Kafka Connect Neo4j Connector vs. Neo4j Streams as a Plugin
 
 [abstract]
 This section covers how to decide whether to run as a Kafka Connect worker, or as a Neo4j Plugin.
@@ -7,16 +7,18 @@ This section covers how to decide whether to run as a Kafka Connect worker, or a
 
 === Pros
 
-* Processing is outside of Neo4j so that memory & CPU impact doesn't impact Neo4j.  You don't need to size the database with Kafka utilization in mind.
-* Much easier for Kafka pros to manage; they benefit from the Confluent ecosystem, such as connecting the REST API to manipulate connectors, the control center to administer & monitor them.
-* By restarting the worker, you can restart your sink strategy without having downtime for Neo4j.
+* Processing is outside of Neo4j so that memory & CPU impact doesn't impact Neo4j.
+You don't need to size the database with Kafka utilization in mind.
+* Much easier for Kafka pros to manage; they benefit from the Confluent ecosystem,
+such as connecting the REST API to manipulate connectors, the control center to administer & monitor them.
+* By restarting the worker, you can restart your sink/source strategy without having downtime for Neo4j.
 * Upgrade Neo4j-Streams without restarting the cluster
-Strictly an external bolt client, so better overall security management of plugin actions.
+* Strictly an external bolt client, so better overall security management of plugin actions.
 
 === Cons
 
-* You can't do TransactionEventHandlers from outside of the database, so you can only sink to Neo4j, you can't produce from it.
-* If you're using Confluent Cloud, you can't host the connector in the cloud (yet).   So this requires a 3rd piece of architecture:  Confluent Cloud, Neo4j, and the Connect Worker (usually a separate VM)
+* If you're using Confluent Cloud, you can't host the connector in the cloud (yet).
+So this requires a 3rd piece of architecture:  Confluent Cloud, Neo4j, and the Connect Worker (usually a separate VM)
 * Possibly worse throughput due to bolt latency & overhead, and separate network hop.
 
 == Neo4j-Streams Plugin
@@ -31,8 +33,6 @@ Strictly an external bolt client, so better overall security management of plugi
 === Cons
 
 * Memory & CPU consumption on your Neo4j Server
-* Requires restarting Neo4j in order to update your configuration and/or Cypher.
-* Upgrading plugin requires cluster restart
 * Need to track config to be identical across all members in the cluster
 * Lesser ability to manage the plugin because it is running inside of the database and not under a particular user account.
 

doc/docs/modules/ROOT/pages/configuration.adoc

@@ -1,4 +1,9 @@
 [#neo4j_configuration_system]
+
+[NOTE]
+The Neo4j Streams Plugin running inside the Neo4j database is deprecated and will not be supported after version 4.3 of Neo4j.
+We recommend users not to adopt this plugin for new implementations, and to consider migrating to the use of the Kafka Connect Neo4j Connector as a replacement
+
 == Configuration system overview
 
 === Location of Configuration Information

doc/docs/modules/ROOT/pages/consumer.adoc

@@ -10,6 +10,11 @@ Use this section to configure Neo4j to ingest the data from Kafka into Neo4j.
 --
 
 endif::env-docs[]
+
+[NOTE]
+The Neo4j Streams Plugin running inside the Neo4j database is deprecated and will not be supported after version 4.3 of Neo4j.
+We recommend users not to adopt this plugin for new implementations, and to consider migrating to the use of the Kafka Connect Neo4j Connector as a replacement
+
 [[neo4j_streams_sink]]
 Is the Kafka Sink that ingest the data directly into Neo4j
 

doc/docs/modules/ROOT/pages/docker.adoc

@@ -1,5 +1,9 @@
 = Run with Docker
 
+[NOTE]
+The Neo4j Streams Plugin running inside the Neo4j database is deprecated and will not be supported after version 4.3 of Neo4j.
+We recommend users not to adopt this plugin for new implementations, and to consider migrating to the use of the Kafka Connect Neo4j Connector as a replacement
+
 ifdef::env-docs[]
 [abstract]
 --
@@ -277,7 +281,7 @@ You'll see something like this:
 ----
 
 [[docker_kafka_connect]]
-== Kafka Connect plugin
+== Kafka Connect Neo4j Connector
 
 Inside the directory `/kafka-connect-neo4j/docker` you'll find a compose file that allows you to start the whole testing environment:
 

doc/docs/modules/ROOT/pages/faq.adoc

@@ -9,19 +9,16 @@ the source code repository for full terms and conditions.
 
 == How to integrate Neo4j and Kafka
 
-When integrating Neo4j and Kafka using Neo4j Streams plugin or Kafka Connect plugin
+When integrating Neo4j and Kafka using Neo4j Streams plugin or Kafka Connect Neo4j Connector
 is important configure just one of them and not both.
-If you need to load data from Kafka to Neo4j (and not viceversa) you can just use the Kafka Connect plugin.
-If you need to have both sink and source functionalities then you have to use the Neo4j Streams plugin.
-This is the main difference between those plugins.
 
 == About CUD file format
 
 The CUD file format is JSON file that represents Graph Entities (Nodes/Relationships) and how to manage them in term
 of **C**reate/**U**pdate/**D**elete operations.
 So every JSON event represents a single operation.
 For more details about how to use these, please checkout the xref:consumer.adoc#_cud_file_format[CUD File Format] section for the
-Neo4j Streams plugin, and the xref:kafka-connect.adoc#kafka-connect-cud-file-format[CUD File Format] section for Kafka Connect plugin.
+Neo4j Streams plugin, and the xref:kafka-connect.adoc#kafka-connect-cud-file-format[CUD File Format] section for Kafka Connect Neo4j Connector.
 
 == How to ingest events using CDC Schema strategy
 
@@ -45,7 +42,7 @@ streams.sink.topic.cdc.schema=<topic-name>
 streams.sink.enabled=true
 ----
 
-If you decide to use Kafka Connect plugin for the sink instance, then it has to be configured as follow:
+If you decide to use Kafka Connect Neo4j Connector for the sink instance, then it has to be configured as follow:
 
 [source, json]
 ----
@@ -191,3 +188,7 @@ Please see the following link for further details:
 * https://docs.confluent.io/platform/current/kafka/deployment.html#multi-node-configuration[Kafka multi-node configuration]
 * https://docs.confluent.io/platform/current/zookeeper/deployment.html#multi-node-setup[Zookeeper multi-node setup]
 --
+
+== Which way should I run the Neo4j Connector for Apache Kafka: As a database plugin, or using the Kafka Connect Framework?
+
+If you have already implemented the database plugin and are running Neo4j <= 4.2, there is no need to change, all other users, new users, and Neo4j Aura users should implement only the Kafka Connect Neo4j Connector

doc/docs/modules/ROOT/pages/index.adoc

@@ -17,7 +17,7 @@ The guide covers the following areas:
 * xref:producer.adoc[How to configure Neo4j Streams as Source]
 * xref:consumer.adoc[How to configure Neo4j Streams as Sink]
 * xref:procedures.adoc[How to use Neo4j Streams Procedures]
-* xref:kafka-connect.adoc[How to configure Kafka Connect plugin]
+* xref:kafka-connect.adoc[How to configure Kafka Connect Neo4j Connector]
 * xref:neo4j-cluster.adoc[Using with Neo4j Causal Cluster]
 * xref:docker.adoc[Run Neo4j Streams and Kafka connect with Docker]
 * xref:kafka-ssl.adoc[A guidance on how to configure SSL between Kafka and Neo4j]