Neo4j kafka sink not initializing/working when trying to stream data

IISuperLuminaLII · May 14, 2021, 7:29pm

Trying to configure a neo4j kafka sink to consume json data from a microservice. Currently after restarting local neo4j client, I do not see any debug messages that indicate the kafka neo4j streams loading. I have provided my neo4j conf and debug trace files. I am pretty new to neo4j so please let me know if I have made any mistakes on either the configuration or neo4j side of things.
`
If it helps I am following this guide to set everything up, unfortunately my company uses 4.2.1 so couldn't use the docker compose to show use case.
Streaming Graphs: Combining Kafka and Neo4j

`
This is my docker of confluent platforms kafka docker compose. This is the docker-compose ps command. I can access everything from localhost:9092 for kafka(tested using kafka-console-consumer --topic friends --bootstrap-server localhost:9092 --from-beginning and it works after producing documents to the topic)

     Name                    Command               State                       Ports                     
---------------------------------------------------------------------------------------------------------
broker            /etc/confluent/docker/run        Up      0.0.0.0:9092->9092/tcp, 0.0.0.0:9101->9101/tcp
connect           /etc/confluent/docker/run        Up      0.0.0.0:8083->8083/tcp, 9092/tcp              
control-center    /etc/confluent/docker/run        Up      0.0.0.0:9021->9021/tcp                        
ksql-datagen      bash -c echo Waiting for K ...   Up                                                    
ksqldb-cli        /bin/sh                          Up                                                    
ksqldb-server     /etc/confluent/docker/run        Up      0.0.0.0:8088->8088/tcp                        
rest-proxy        /etc/confluent/docker/run        Up      0.0.0.0:8082->8082/tcp                        
schema-registry   /etc/confluent/docker/run        Up      0.0.0.0:8081->8081/tcp                        
zookeeper         /etc/confluent/docker/run        Up      0.0.0.0:2181->2181/tcp, 2888/tcp, 3888/tcp

`
I am using neo4j 4.2.1 with Streams 4.0.7 versions

kafka.bootstrap.servers=localhost:9092
kafka.auto.offset.reset=earliest
kafka.group.id=neo4j
kafka.enable.auto.commit=true
kafka.key.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
kafka.value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

streams.sink.topic.cypher.friends=MERGE (p1:Person { name: event.initiated }) MERGE (p2:Person { name: event.accepted }) CREATE (p1)-[:FRIENDS { when: event.date }]->(p2)
streams.sink.enabled=true

`
This is my configuration in debug trace

2021-05-14 07:43:51.278+0000 INFO  Starting...
2021-05-14 07:43:54.047+0000 INFO  ======== Neo4j 4.2.1 ========
2021-05-14 07:43:56.353+0000 DEBUG [neo4j/2c0e4668] Adding listener for event: CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.353+0000 INFO  [neo4j/2c0e4668] Adding listener for event type CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.359+0000 DEBUG [neo4j/2c0e4668] Adding listener for event: CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.359+0000 INFO  [neo4j/2c0e4668] Adding listener for event type CONFIGURATION_INITIALIZED
2021-05-14 07:43:57.337+0000 DEBUG [neo4j/2c0e4668] Starting StreamsConfig
2021-05-14 07:43:57.344+0000 DEBUG [neo4j/2c0e4668] Waiting for the Neo4j instance to be ready...
2021-05-14 07:43:57.806+0000 INFO  [neo4j/2c0e4668] Starting the connector lifecycle listener
2021-05-14 07:43:57.807+0000 INFO  [neo4j/2c0e4668] StreamsConfig started
2021-05-14 07:43:57.808+0000 DEBUG [neo4j/2c0e4668] Starting StreamsConfig
2021-05-14 07:43:58.588+0000 INFO  Called db.clearQueryCaches(): Query caches successfully cleared of 1 queries.
2021-05-14 07:43:56.353+0000 DEBUG [neo4j/2c0e4668] Adding listener for event: CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.353+0000 INFO  [neo4j/2c0e4668] Adding listener for event type CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.359+0000 DEBUG [neo4j/2c0e4668] Adding listener for event: CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.359+0000 INFO  [neo4j/2c0e4668] Adding listener for event type CONFIGURATION_INITIALIZED
2021-05-14 07:43:57.337+0000 DEBUG [neo4j/2c0e4668] Starting StreamsConfig
2021-05-14 07:43:57.344+0000 DEBUG [neo4j/2c0e4668] Waiting for the Neo4j instance to be ready...
2021-05-14 07:43:57.806+0000 INFO  [neo4j/2c0e4668] Starting the connector lifecycle listener
2021-05-14 07:43:57.807+0000 INFO  [neo4j/2c0e4668] StreamsConfig started
2021-05-14 07:43:57.808+0000 DEBUG [neo4j/2c0e4668] Starting StreamsConfig
2021-05-14 07:43:58.588+0000 INFO  Called db.clearQueryCaches(): Query caches successfully cleared of 1 queries.
2021-05-14 07:43:56.353+0000 DEBUG [neo4j/2c0e4668] Adding listener for event: CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.353+0000 INFO  [neo4j/2c0e4668] Adding listener for event type CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.359+0000 DEBUG [neo4j/2c0e4668] Adding listener for event: CONFIGURATION_INITIALIZED
2021-05-14 07:43:56.359+0000 INFO  [neo4j/2c0e4668] Adding listener for event type CONFIGURATION_INITIALIZED
2021-05-14 07:43:57.337+0000 DEBUG [neo4j/2c0e4668] Starting StreamsConfig
2021-05-14 07:43:57.344+0000 DEBUG [neo4j/2c0e4668] Waiting for the Neo4j instance to be ready...
2021-05-14 07:43:57.806+0000 INFO  [neo4j/2c0e4668] Starting the connector lifecycle listener
2021-05-14 07:43:57.807+0000 INFO  [neo4j/2c0e4668] StreamsConfig started
2021-05-14 07:43:57.808+0000 DEBUG [neo4j/2c0e4668] Starting StreamsConfig
2021-05-14 07:43:58.588+0000 INFO  Called db.clearQueryCaches(): Query caches successfully cleared of 1 queries.
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] Detected new event change in configuration CONFIGURATION_INITIALIZED
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] The event tree is [CONFIGURATION_INITIALIZED]
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] The listenerMap contains the following listeners: [CONFIGURATION_INITIALIZED]
2021-05-14 07:44:07.812+0000 DEBUG [neo4j/2c0e4668] [Sink] Configuration is empty
2021-05-14 07:44:07.812+0000 DEBUG [neo4j/2c0e4668] [Source] Configuration is empty
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] Detected new event change in configuration CONFIGURATION_INITIALIZED
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] The event tree is [CONFIGURATION_INITIALIZED]
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] The listenerMap contains the following listeners: [CONFIGURATION_INITIALIZED]
2021-05-14 07:44:07.812+0000 DEBUG [neo4j/2c0e4668] [Sink] Configuration is empty
2021-05-14 07:44:07.812+0000 DEBUG [neo4j/2c0e4668] [Source] Configuration is empty
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] Detected new event change in configuration CONFIGURATION_INITIALIZED
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] The event tree is [CONFIGURATION_INITIALIZED]
2021-05-14 07:44:07.811+0000 DEBUG [neo4j/2c0e4668] The listenerMap contains the following listeners: [CONFIGURATION_INITIALIZED]
2021-05-14 07:44:07.812+0000 DEBUG [neo4j/2c0e4668] [Sink] Configuration is empty
2021-05-14 07:44:07.812+0000 DEBUG [neo4j/2c0e4668] [Source] Configuration is empty
2021-05-14 07:44:09.797+0000 DEBUG Reading users from /Users/***/Library/Application Support/Neo4j Desktop/Application/relate-data/dbmss/dbms-9ceb05f0-7575-4c16-967e-a7b7943a5a1d/data/dbms/auth.ini
2021-05-14 07:44:09.814+0000 INFO  Sending metrics to CSV file at /Users/***/Library/Application Support/Neo4j Desktop/Application/relate-data/dbmss/dbms-9ceb05f0-7575-4c16-967e-a7b7943a5a1d/metrics
2021-05-14 07:44:09.828+0000 INFO  Bolt enabled on localhost:7687.
2021-05-14 07:44:09.850+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.ClusteringDatabaseService] at [/db]
2021-05-14 07:44:09.850+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.ClusteringDbmsService] at [/dbms]
2021-05-14 07:44:09.851+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.LegacyClusteringRedirectService] at [/db/manage]
2021-05-14 07:44:09.857+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.rest.discovery.DiscoveryService] at [/]
2021-05-14 07:44:09.857+0000 DEBUG Adding JAXRS classes [class com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider] at [/]
2021-05-14 07:44:09.860+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.http.cypher.CypherResource, class org.neo4j.server.http.cypher.format.input.json.JsonMessageBodyReader, class org.neo4j.server.http.cypher.format.output.json.JsonMessageBodyWriter, class org.neo4j.server.http.cypher.format.output.eventsource.EventSourceMessageBodyWriter] at [/db]
2021-05-14 07:44:09.860+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.http.cypher.LegacyTransactionService, class org.neo4j.server.http.cypher.format.input.json.JsonMessageBodyReader, class org.neo4j.server.http.cypher.format.output.json.JsonMessageBodyWriter] at [/db/data]
2021-05-14 07:44:09.931+0000 DEBUG Mounting servlet at [/dbms]
2021-05-14 07:44:10.019+0000 DEBUG Mounting servlet at [/db/manage]
2021-05-14 07:44:10.019+0000 DEBUG Mounting servlet at [/db/data]
2021-05-14 07:44:10.020+0000 DEBUG Mounting servlet at [/db]
2021-05-14 07:44:10.043+0000 DEBUG Mounting servlet at [/]
2021-05-14 07:44:10.790+0000 INFO  Remote interface available at http://localhost:7474/
2021-05-14 07:44:10.790+0000 INFO  Started.
2021-05-14 07:44:09.797+0000 DEBUG Reading users from /Users/***/Library/Application Support/Neo4j Desktop/Application/relate-data/dbmss/dbms-9ceb05f0-7575-4c16-967e-a7b7943a5a1d/data/dbms/auth.ini
2021-05-14 07:44:09.814+0000 INFO  Sending metrics to CSV file at /Users/****/Library/Application Support/Neo4j Desktop/Application/relate-data/dbmss/dbms-9ceb05f0-7575-4c16-967e-a7b7943a5a1d/metrics
2021-05-14 07:44:09.828+0000 INFO  Bolt enabled on localhost:7687.
2021-05-14 07:44:09.850+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.ClusteringDatabaseService] at [/db]
2021-05-14 07:44:09.850+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.ClusteringDbmsService] at [/dbms]
2021-05-14 07:44:09.851+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.LegacyClusteringRedirectService] at [/db/manage]
2021-05-14 07:44:09.857+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.rest.discovery.DiscoveryService] at [/]
2021-05-14 07:44:09.857+0000 DEBUG Adding JAXRS classes [class com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider] at [/]
2021-05-14 07:44:09.860+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.http.cypher.CypherResource, class org.neo4j.server.http.cypher.format.input.json.JsonMessageBodyReader, class org.neo4j.server.http.cypher.format.output.json.JsonMessageBodyWriter, class org.neo4j.server.http.cypher.format.output.eventsource.EventSourceMessageBodyWriter] at [/db]
2021-05-14 07:44:09.860+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.http.cypher.LegacyTransactionService, class org.neo4j.server.http.cypher.format.input.json.JsonMessageBodyReader, class org.neo4j.server.http.cypher.format.output.json.JsonMessageBodyWriter] at [/db/data]
2021-05-14 07:44:09.931+0000 DEBUG Mounting servlet at [/dbms]
2021-05-14 07:44:10.019+0000 DEBUG Mounting servlet at [/db/manage]
2021-05-14 07:44:10.019+0000 DEBUG Mounting servlet at [/db/data]
2021-05-14 07:44:10.020+0000 DEBUG Mounting servlet at [/db]
2021-05-14 07:44:10.043+0000 DEBUG Mounting servlet at [/]
2021-05-14 07:44:10.790+0000 INFO  Remote interface available at http://localhost:7474/
2021-05-14 07:44:10.790+0000 INFO  Started.
2021-05-14 07:44:09.797+0000 DEBUG Reading users from /Users/****/Library/Application Support/Neo4j Desktop/Application/relate-data/dbmss/dbms-9ceb05f0-7575-4c16-967e-a7b7943a5a1d/data/dbms/auth.ini
2021-05-14 07:44:09.814+0000 INFO  Sending metrics to CSV file at /Users/****/Library/Application Support/Neo4j Desktop/Application/relate-data/dbmss/dbms-9ceb05f0-7575-4c16-967e-a7b7943a5a1d/metrics
2021-05-14 07:44:09.828+0000 INFO  Bolt enabled on localhost:7687.
2021-05-14 07:44:09.850+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.ClusteringDatabaseService] at [/db]
2021-05-14 07:44:09.850+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.ClusteringDbmsService] at [/dbms]
2021-05-14 07:44:09.851+0000 DEBUG Adding JAXRS classes [class com.neo4j.server.rest.causalclustering.LegacyClusteringRedirectService] at [/db/manage]
2021-05-14 07:44:09.857+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.rest.discovery.DiscoveryService] at [/]
2021-05-14 07:44:09.857+0000 DEBUG Adding JAXRS classes [class com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider] at [/]
2021-05-14 07:44:09.860+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.http.cypher.CypherResource, class org.neo4j.server.http.cypher.format.input.json.JsonMessageBodyReader, class org.neo4j.server.http.cypher.format.output.json.JsonMessageBodyWriter, class org.neo4j.server.http.cypher.format.output.eventsource.EventSourceMessageBodyWriter] at [/db]
2021-05-14 07:44:09.860+0000 DEBUG Adding JAXRS classes [class org.neo4j.server.http.cypher.LegacyTransactionService, class org.neo4j.server.http.cypher.format.input.json.JsonMessageBodyReader, class org.neo4j.server.http.cypher.format.output.json.JsonMessageBodyWriter] at [/db/data]
2021-05-14 07:44:09.931+0000 DEBUG Mounting servlet at [/dbms]
2021-05-14 07:44:10.019+0000 DEBUG Mounting servlet at [/db/manage]
2021-05-14 07:44:10.019+0000 DEBUG Mounting servlet at [/db/data]
2021-05-14 07:44:10.020+0000 DEBUG Mounting servlet at [/db]
2021-05-14 07:44:10.043+0000 DEBUG Mounting servlet at [/]
2021-05-14 07:44:10.790+0000 INFO  Remote interface available at http://localhost:7474/
2021-05-14 07:44:10.790+0000 INFO  Started.

IISuperLuminaLII · May 14, 2021, 9:09pm

Found the solution myself. Basically, you will need to create a new streams.conf file and add the connection configurations in that conf file, or else it will not start the kafka neo4j streams connection.

Topic		Replies	Views
How to configure Neo4j sink instances with Kafka and neo4j-streams? Stream Processing kafka	1	1719	April 1, 2019
Neo4J Streams Neo4j Graph Platform neo4j-streams	3	542	January 26, 2022
Neo4j kafka sink plugin Stream Processing	0	383	May 4, 2021
Neo4j Sink not recieving Kafka Topic events ETL-Tool kafka , operations , knowledge-base , cloud	4	1084	September 15, 2019
NEO4J Sink and Source connector Neo4j Graph Platform migrated	0	237	December 4, 2022

Neo4j kafka sink not initializing/working when trying to stream data

Related topics