Message structure

The message key structure depends on kafka.streams.log.compaction.strategy.

With delete is a string: "${meta.txId + meta.txEventId}-${meta.txEventId}".

"[txId+txEventId] - txEventId "

where:

  • txId identifies the transaction that affected the entity

  • txEventId is a counter that identifies the internal order in which Neo4j handled the specific event

  • [txId+txEventId] is the numeric sum of the two previous values

 

Instead with compact:

In case of node without constrained label the key is the string value of node id.

In case of node with constrained label, the key is a json with {ids: mapOfConstaint , labels: listOfLabels}

For example, with this configuration:

streams.source.topic.nodes.<TOPIC_NAME>=Person{*}
kafka.streams.log.compaction.strategy=compact

this constraint:

CREATE CONSTRAINT ON (p:Person) ASSERT p.name IS UNIQUE

and this query:

CREATE (:Person {name:'Sherlock', surname: 'Holmes'})

We obtain this key:

{"ids": {"name": "Sherlock"}, "labels": ["Person"]}

 

Otherwise, with the same configuration and query as above, but with the constraint:

CREATE CONSTRAINT ON (p:Person) ASSERT (p.name, p.surname) IS NODE KEY

We obtain this key:

{"ids": {"surname": "Holmes", "name":  "Sherlock"}, "labels": ["Person"]}

 

In case of relationship, the key is a json with {start: START_NODE , end: END_NODE, label: typeOfRelationship}
START_NODE and END_NODE node follow the same rule as above.

For example, with this configuration:

streams.source.topic.nodes.<TOPIC_NAME>=Person{*}
streams.source.topic.relationships.<TOPIC_NAME>=Person{*}
kafka.streams.log.compaction.strategy=compact

these constraints:

CREATE CONSTRAINT ON (p:Person) ASSERT p.name IS UNIQUE;
CREATE CONSTRAINT ON (p:Product) ASSERT p.code IS UNIQUE;

and these queries:

CREATE (:Person {name:'Pippo'});
CREATE (p:Product {code:'1367', name: 'Notebook'});
MATCH (pe:Person {name:'Pippo'}), (pr:Product {name:'Notebook'}) MERGE (pe)-[:BUYS]->(pr);

We obtain this key:

{"start": {"ids": {"name":  "Pippo"}, "labels": ["Person"]}, "end": {"ids": {"code":  "1367"}, "labels": ["Product"]},
 "label": "BUYS"}

 

Otherwise, with this configuration:

streams.source.topic.nodes.<TOPIC_NAME>=Person{*}
streams.source.topic.relationships.<TOPIC_NAME>=Person{*}
kafka.streams.log.compaction.strategy=compact

without constraints, and with these queries:

CREATE (:Person {name:'Pippo'})

We obtain this key:

{"start": "0", "end": "1", "label": "BUYS"}

In case of relationships with multiple constraints on start or end node, the ids fields depend on streams.source.topic.relationships.<TOPIC_NAME>.key_strategy config. See 'key-strategy' section to more details