I needed to source data from a MongoDB  change stream- using Kafka. The official Mongo DB connector is on https://www.confluent.io/hub/mongodb/kafka-connect-mongodb and is straightforward to install.

If you’re database server has an internet connection , use the confluent-hub method. Otherwise there is a manual install method – outlined in the link above.

This is an example of using the confluent-hub method

confluent-hub install mongodb/kafka-connect-mongodb:0.1

Once you’ve installed the connector stop | start the Confluent Connect

confluent stop connect 

confluent start connect 

You should now see a new entry in the Connector list : MongoDBConnector

For the connection to work there is some preliminary work required .Firstly make sure there is a MongoDB change stream available. From Mongo 3.6 it is possible to set up a changestream on a standalone MongoDB – as long as a Replica Set is initiated.

There were a number of issues I faced when setting up the Connector

1) Connection URl – I played around with a few various url options – but this is the one that worked for me

mongodb://my_usr:my_pws@myserver:myport/my_authentication_db

e.g mongodb://jackv:uuY256@10.12.151.21:27017/db1

This format worked for the setup. Note the reference to the authentication db. This user’s authentication database was set up as a different database to the “admin” db.

2) String errors when reading the topic. I initially setup the Mongo connection  with no values for :

Key Converter Class

Value Converter class 

When I attempted to read the data there were multiple conversion issues

I edited the values of key Converter Class and Value Converter Class to  “org.apache.kafka.connect.storage.StringConverter”

When I restarted the Source Connector I could now read the data through  ksql PRINT functionality.

As I start to  work through the process of reading and directing the data to a Sink connector , flattening  the JSON format and other related tasks.  I’ll add more details .

Author: Jack Vamvas