Using Docker-Compose Kafka-connect-sink for neo4j

[Schiewatir]

Hi,

I use the docker-compose file to spin up everything, I'm really wondering if nobody encountered this error before, every time I send the settings to create a neo4j connector by this ;

{ "name": "Neo4jSinkConnector", "config": { "topics": "my-topic", "connector.class": "streams.kafka.connect.sink.Neo4jSinkConnector", "errors.retry.timeout": "-1", "errors.retry.delay.max.ms": "1000", "errors.tolerance": "all", "errors.log.enable": true, "errors.log.include.messages": true, "neo4j.server.uri": "bolt://neo4j:7687", "neo4j.authentication.basic.username": "neo4j", "neo4j.authentication.basic.password": "connect", "neo4j.encryption.enabled": false, "neo4j.topic.cypher.my-topic": "Create (:person)" } }

then the Kafka-Connect container raises an error like this:

[2020-05-28 20:20:00,329] ERROR WorkerConnector{id=Neo4jSinkConnector} Error while starting connector (org.apache.kafka.connect.runtime.WorkerConnector) connect | java.util.ServiceConfigurationError: kotlin.reflect.jvm.internal.impl.builtins.BuiltInsLoader: Provider kotlin.reflect.jvm.internal.impl.serialization.deserialization.builtins.BuiltInsLoaderImpl not a subtype connect | at java.util.ServiceLoader.fail(ServiceLoader.java:239) connect | at java.util.ServiceLoader.access$300(ServiceLoader.java:185) connect | at java.util.ServiceLoader$LazyIterator.nextService(ServiceLoader.java:376) connect | at java.util.ServiceLoader$LazyIterator.next(ServiceLoader.java:404) connect | at java.util.ServiceLoader$1.next(ServiceLoader.java:480) connect | at kotlin.collections.CollectionsKt___CollectionsKt.first(_Collections.kt:185) connect | at kotlin.reflect.jvm.internal.impl.builtins.BuiltInsLoader$Companion.(Buil....

any idea how can I solve this.

[SmithJohnTaylor]

I'm also seeing this error with v1.0.7. In addition, logs show:

java.lang.NoClassDefFoundError: Could not initialize class kotlin.reflect.jvm.internal.impl.builtins.BuiltInsLoader
    at kotlin.reflect.jvm.internal.impl.builtins.KotlinBuiltIns.createBuiltInsModule(KotlinBuiltIns.java:148)
    at kotlin.reflect.jvm.internal.impl.platform.JvmBuiltIns.<init>(JvmBuiltIns.kt:56)
    at kotlin.reflect.jvm.internal.impl.platform.JvmBuiltIns.<init>(JvmBuiltIns.kt:31)
    at kotlin.reflect.jvm.internal.components.RuntimeModuleData$Companion.create(RuntimeModuleData.kt:57)
    at kotlin.reflect.jvm.internal.ModuleByClassLoaderKt.getOrCreateModule(moduleByClassLoader.kt:58)
    at kotlin.reflect.jvm.internal.KDeclarationContainerImpl$Data$moduleData$2.invoke(KDeclarationContainerImpl.kt:35)
    at kotlin.reflect.jvm.internal.KDeclarationContainerImpl$Data$moduleData$2.invoke(KDeclarationContainerImpl.kt:32)
    at kotlin.reflect.jvm.internal.ReflectProperties$LazySoftVal.invoke(ReflectProperties.java:93)
    at kotlin.reflect.jvm.internal.ReflectProperties$Val.getValue(ReflectProperties.java:32)
    at kotlin.reflect.jvm.internal.KDeclarationContainerImpl$Data.getModuleData(KDeclarationContainerImpl.kt)
    at kotlin.reflect.jvm.internal.KClassImpl$Data$descriptor$2.invoke(KClassImpl.kt:46)
    at kotlin.reflect.jvm.internal.KClassImpl$Data$descriptor$2.invoke(KClassImpl.kt:43)
    at kotlin.reflect.jvm.internal.ReflectProperties$LazySoftVal.invoke(ReflectProperties.java:93)
    at kotlin.reflect.jvm.internal.ReflectProperties$Val.getValue(ReflectProperties.java:32)
    at kotlin.reflect.jvm.internal.KClassImpl$Data.getDescriptor(KClassImpl.kt)
    at kotlin.reflect.jvm.internal.KClassImpl.getDescriptor(KClassImpl.kt:172)
    at kotlin.reflect.jvm.internal.KClassImpl.getConstructorDescriptors(KClassImpl.kt:186)
    at kotlin.reflect.jvm.internal.KClassImpl$Data$constructors$2.invoke(KClassImpl.kt:90)
    at kotlin.reflect.jvm.internal.KClassImpl$Data$constructors$2.invoke(KClassImpl.kt:43)
    at kotlin.reflect.jvm.internal.ReflectProperties$LazySoftVal.invoke(ReflectProperties.java:93)
    at kotlin.reflect.jvm.internal.ReflectProperties$Val.getValue(ReflectProperties.java:32)
    at kotlin.reflect.jvm.internal.KClassImpl$Data.getConstructors(KClassImpl.kt)
    at kotlin.reflect.jvm.internal.KClassImpl.getConstructors(KClassImpl.kt:222)
    at streams.kafka.connect.sink.Neo4jSinkConnectorConfig.validateAllTopics(Neo4jSinkConnectorConfig.kt:347)
    at streams.kafka.connect.sink.Neo4jSinkConnectorConfig.<init>(Neo4jSinkConnectorConfig.kt:111)
    at streams.kafka.connect.sink.Neo4jSinkConnector.start(Neo4jSinkConnector.kt:25)
    at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:110)
    at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:135)
    at org.apache.kafka.connect.runtime.WorkerConnector.transitionTo(WorkerConnector.java:195)
    at org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:257)
    at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:1190)
    at org.apache.kafka.connect.runtime.distributed.DistributedHerder.processConnectorConfigUpdates(DistributedHerder.java:548)
    at org.apache.kafka.connect.runtime.distributed.DistributedHerder.tick(DistributedHerder.java:395)
    at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:289)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)```

[conker84]

Can you please try with this release? https://github.com/neo4j-contrib/neo4j-streams/releases/download/4.0.2/neo4j-kafka-connect-neo4j-1.0.7.zip

[moxious]

@SmithJohnTaylor @arash-ariana sorry for the time lag, but is there any follow-up on this ticket?

[conker84]

@SmithJohnTaylor @arash-ariana can you please try with v1.0.9?

[antonio-urbano]

@SmithJohnTaylor @arash-ariana can you please try with v1.0.9?

I'm using version 1.0.9 and I still have the same error. Any suggestion on how to solve it? This is the error log:

{"name":"Neo4jSinkConnector","connector":{"state":"FAILED","worker_id":"connect:8083","trace":"java.lang.NoClassDefFoundError: Could not initialize class kotlin.reflect.jvm.internal.impl.builtins.BuiltInsLoader\n\tat kotlin.reflect.jvm.internal.impl.builtins.KotlinBuiltIns.createBuiltInsModule(KotlinBuiltIns.java:148)\n\tat kotlin.reflect.jvm.internal.impl.platform.JvmBuiltIns.(JvmBuiltIns.kt:56)\n\tat kotlin.reflect.jvm.internal.impl.platform.JvmBuiltIns.(JvmBuiltIns.kt:31)\n\tat kotlin.reflect.jvm.internal.components.RuntimeModuleData$Companion.create(RuntimeModuleData.kt:57)\n\tat kotlin.reflect.jvm.internal.ModuleByClassLoaderKt.getOrCreateModule(moduleByClassLoader.kt:58)\n\tat kotlin.reflect.jvm.internal.KDeclarationContainerImpl$Data$moduleData$2.invoke(KDeclarationContainerImpl.kt:35)\n\tat kotlin.reflect.jvm.internal.KDeclarationContainerImpl$Data$moduleData$2.invoke(KDeclarationContainerImpl.kt:32)\n\tat kotlin.reflect.jvm.internal.ReflectProperties$LazySoftVal.invoke(ReflectProperties.java:93)\n\tat kotlin.reflect.jvm.internal.ReflectProperties$Val.getValue(ReflectProperties.java:32)\n\tat kotlin.reflect.jvm.internal.KDeclarationContainerImpl$Data.getModuleData(KDeclarationContainerImpl.kt)\n\tat kotlin.reflect.jvm.internal.KClassImpl$Data$descriptor$2.invoke(KClassImpl.kt:46)\n\tat kotlin.reflect.jvm.internal.KClassImpl$Data$descriptor$2.invoke(KClassImpl.kt:43)\n\tat kotlin.reflect.jvm.internal.ReflectProperties$LazySoftVal.invoke(ReflectProperties.java:93)\n\tat kotlin.reflect.jvm.internal.ReflectProperties$Val.getValue(ReflectProperties.java:32)\n\tat kotlin.reflect.jvm.internal.KClassImpl$Data.getDescriptor(KClassImpl.kt)\n\tat kotlin.reflect.jvm.internal.KClassImpl.getDescriptor(KClassImpl.kt:172)\n\tat kotlin.reflect.jvm.internal.KClassImpl.getConstructorDescriptors(KClassImpl.kt:186)\n\tat kotlin.reflect.jvm.internal.KClassImpl$Data$constructors$2.invoke(KClassImpl.kt:90)\n\tat kotlin.reflect.jvm.internal.KClassImpl$Data$constructors$2.invoke(KClassImpl.kt:43)\n\tat kotlin.reflect.jvm.internal.ReflectProperties$LazySoftVal.invoke(ReflectProperties.java:93)\n\tat kotlin.reflect.jvm.internal.ReflectProperties$Val.getValue(ReflectProperties.java:32)\n\tat kotlin.reflect.jvm.internal.KClassImpl$Data.getConstructors(KClassImpl.kt)\n\tat kotlin.reflect.jvm.internal.KClassImpl.getConstructors(KClassImpl.kt:222)\n\tat streams.kafka.connect.sink.Neo4jSinkConnectorConfig.validateAllTopics(Neo4jSinkConnectorConfig.kt:347)\n\tat streams.kafka.connect.sink.Neo4jSinkConnectorConfig.(Neo4jSinkConnectorConfig.kt:111)\n\tat streams.kafka.connect.sink.Neo4jSinkConnector.start(Neo4jSinkConnector.kt:25)\n\tat org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:185)\n\tat org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:210)\n\tat org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:349)\n\tat org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:332)\n\tat org.apache.kafka.connect.runtime.WorkerConnector.doRun(WorkerConnector.java:140)\n\tat org.apache.kafka.connect.runtime.WorkerConnector.run(WorkerConnector.java:117)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:834)\n"},"tasks":[],"type":"sink"}

[freid001-els]

Im using v5.0.2 and having the same problem. Is there any solution to this?

[ali-ince]

would anyone happy to share their docker-compose file for us to reproduce this behaviour?

[Emrehzl94]

Closing this due to lack of response, you can try latest version of Kafka Connector; https://github.com/neo4j/neo4j-kafka-connector and if anything comes up, please feel free to create an issue.