Sink kafka topic to mongodb

Question

I have a project where I need to get data from JSON files using java and sink it into kafka topic, and then sink that data from the topic to mongodb. I have found the kafka-mongodb connector, but the documentation is available only to connect using confluent plateform. I have tried:

Download mongo-kafka-connect-1.2.0.jar from Maven.
Put the file in /kafka/plugins
Added this line "plugin.path=C:\kafka\plugins" in connect-standalone.properties.
created MongoSinkConnector.properties.

name=mongo-sink
topics=test
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1
key.ignore=true

# Specific global MongoDB Sink Connector configuration
connection.uri=mongodb://localhost:27017
database=student_kafka
collection=students
max.num.retries=3
retries.defer.timeout=5000
type.name=kafka-connect

and than I ran the command

.\bin\windows\connect-standalone.bat .\config\connect-standalone.properties .\config\MongoSinkConnector.properties

I got this error

[2020-08-09 20:18:30,329] ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone)
java.util.concurrent.ExecutionException: java.lang.BootstrapMethodError: java.lang.NoClassDefFoundError: com/mongodb/ConnectionString
        at org.apache.kafka.connect.util.ConvertingFutureCallback.result(ConvertingFutureCallback.java:115)
        at org.apache.kafka.connect.util.ConvertingFutureCallback.get(ConvertingFutureCallback.java:99)
        at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:118)
Caused by: java.lang.BootstrapMethodError: java.lang.NoClassDefFoundError: com/mongodb/ConnectionString
        at com.mongodb.kafka.connect.sink.MongoSinkConfig.createConfigDef(MongoSinkConfig.java:248)
        at com.mongodb.kafka.connect.sink.MongoSinkConfig.<clinit>(MongoSinkConfig.java:139)
        at com.mongodb.kafka.connect.MongoSinkConnector.config(MongoSinkConnector.java:72)
        at org.apache.kafka.connect.runtime.AbstractHerder.validateConnectorConfig(AbstractHerder.java:366)
        at org.apache.kafka.connect.runtime.AbstractHerder.lambda$validateConnectorConfig$1(AbstractHerder.java:326)
        at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
        at java.util.concurrent.FutureTask.run(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
        at java.lang.Thread.run(Unknown Source)
Caused by: java.lang.NoClassDefFoundError: com/mongodb/ConnectionString
        ... 10 more
Caused by: java.lang.ClassNotFoundException: com.mongodb.ConnectionString
        at java.net.URLClassLoader.findClass(Unknown Source)
        at java.lang.ClassLoader.loadClass(Unknown Source)
        at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source)
        at java.lang.ClassLoader.loadClass(Unknown Source)
        ... 10 more

EDIT: Thanks to BogdanSucaciu for the help, I have found a solution for this.

You need to add the following jars into kafka/lib folder.

mongodb-driver-3.12.7.jar and mongodb-driver-core-3.12.7.jar and mongo-java-driver-3.12.6.jar and mongo-kafka-connect-1.0.1.jar.
PS: I have had some problems working with the latest mongo-kafka-connect. so I had to work with this version.

can you post the worker log at startup? it should contain something like Loading plugin from: ... — BogdanSucaciu
I can't find that line, here is a full log after I execute command:pastebin.pl/view/c97dbf03 Thank you again — tariq m'ghari
are you sure it is the full log? It should start with something like this Kafka Connect standalone worker initializing ... — BogdanSucaciu
interesting, not even the loggers are getting loaded... Can you try using a different connector ( e.g: file-source-connector ) ? I want to determine if the problem is the connector itself or your setup ( Java / OS ) — BogdanSucaciu

BogdanSucaciu BogdanSucaciu · Accepted Answer · 2020-08-09T21:14:49

You are missing the MongoDB driver. The MongoDB connector jar contains only the classes relevant to Kafka Connect but it still needs a driver to be able to connect to a MongoDB instance. You would need to download that driver and copy the jar file to the same path where you've published your connector ( C:\kafka\plugins ).

To keep things clean you should also create another folder inside that plugins directory ( e.g: C:\kafka\plugins\mongodb ) and move all the stuff relevant to this connector there.

Later Edit:

I went through an old(er) setup that I had with Kafka Connect and MongoDB Sink connector and I found the following jars:

This makes me believe that the kafka-connect-mongdb jar and the mongodb-driver won't be enough. You can give it a try though.

Sink kafka topic to mongodb

1 Answers