# Moving changed records from neo4j to kafka

**URL:** https://community.neo4j.com/t/moving-changed-records-from-neo4j-to-kafka/59770
**Category:** General
**Tags:** migrated
**Created:** [August 10, 2022, 3:15pm UTC](https://community.neo4j.com/t/moving-changed-records-from-neo4j-to-kafka/59770 "2022-08-10T15:15:48Z")
**Posts on this page:** 2
**Page:** 1

<div class="post-metadata">

### Author: ![serengil](https://avatars.discourse-cdn.com/v4/letter/s/439d5e/32.png) [@serengil](https://community.neo4j.com/u/serengil)
#### Post date: [August 10, 2022, 3:15pm UTC](https://community.neo4j.com/t/moving-changed-records-from-neo4j-to-kafka/59770/1 "2022-08-10T15:15:48Z")

</div>

I want to move changes records from neo4j to kafka.

I can do it with Neo4j Streams plugin well but it is [announced](https://github.com/neo4j-contrib/neo4j-streams) that the plugin is going to be deprecated after version 5.

So, it is recommended to use kafka connect neo4j connector for this task. According to the official [tutorial](https://neo4j.com/labs/kafka/4.1/kafka-connect/#kafka-connect-source-instance), I need to create a connector on the kafka side similar to following configuration. Based on my experiments, this really checks the timestamp property of TestSource node. If I set timestamp property of the node when I update it, it really moves data from neo4j to kafka. However, if I don't set timestamp property when I update the node, nothing is going to move kafka. Neo4j does not provide a auto-filled property similar to relational databases.

```auto
{
  "name": "Neo4jSourceConnectorAVRO",
  "config": {
    "topic": "my-topic",
    "connector.class": "streams.kafka.connect.source.Neo4jSourceConnector",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema_registry:8081",
    "value.converter.schema.registry.url": "http://schema_registry:8081",
    "neo4j.server.uri": "bolt://neo4j:7687",
    "neo4j.authentication.basic.username": "neo4j",
    "neo4j.authentication.basic.password": "connect",
    "neo4j.encryption.enabled": false,
    "neo4j.streaming.poll.interval.msecs": 5000,
    "neo4j.streaming.property": "timestamp",
    "neo4j.streaming.from": "NOW",
    "neo4j.enforce.schema": true,
    "neo4j.source.query": "MATCH (ts:TestSource) WHERE ts.timestamp > $lastCheck RETURN ts.name AS name, ts.timestamp AS timestamp"
  }
}

```

The question is that cannot I move changes records from neo4j to kafka with kafka connect neo4j connector if I do not have a last update property in my nodes?

---

<div class="post-metadata">

### Author: ![serengil](https://avatars.discourse-cdn.com/v4/letter/s/439d5e/32.png) [@serengil](https://community.neo4j.com/u/serengil)
#### Post date: [August 11, 2022, 9:13am UTC](https://community.neo4j.com/t/moving-changed-records-from-neo4j-to-kafka/59770/2 "2022-08-11T09:13:29Z")

</div>

I plan to set this timestamp property if it is not set in the update statement with [triggers](https://neo4j.com/labs/apoc/4.4/background-operations/triggers/). In that way, kafka connect neo4j connector will be able to track changes.

```auto
CALL apoc.trigger.add('setLastUpdate', "
UNWIND keys($assignedNodeProperties) AS k
UNWIND $assignedNodeProperties[k] AS map
WITH map.node AS node
MATCH (n)
WHERE id(n) = id(node) AND NOT 'timestamp' in keys($assignedNodeProperties)
SET n.timestamp= timestamp()
", {phase: 'afterAsync'})

```

This triggger will add timestamp property for all node types. You can customize the node type in the line 5 as MATCH (n:SpecificNode).

Still, you have an alternative approach, I will be appreciate if you share here.