In the last article, we have seen How to modify Kafka Connectors configuration using kcctl. In this article, we will see how to republish existing data using Apache kafka connect. In a kafka driven architecture, there is always a case where destination system request to send original data again due to some reason or due to some issue in processing the messages for the first time. This requires source system to produce messages even though there are no further changes to make. All this can be avoided through the use of connector. You just have to integrate connector with your producer and consumer systems.

Not only that, you can also use kafka connect to consume messages from some specific offset and republish messages. You can do all sorts of things by just integrating kafka connect in your event based architecture. I am going to demonstrate one such example here. I already have a setup where the stream of events would be triggered from the changes done in the MongoDB. I have configured a source connector to listen to those events and publish messages on a topic called cyber.col.topic from where destination consumer would process all the messages.

 

How to republish existing data using Apache Kafka Connect

Also Read: How to Restart Apache Kafka Connectors?

Step 1: Switch to context

First, you have to switch context to the place where connectors are running. Run kcctl config set-context <context_name> --cluster=<cluster_name> --user=<user_name> --password=<password_name> command to set the context. In my case, my connectors are configured under cyber context so I am switching to that context by providing cluster information and required credentials as shown below.

cyberithub@macos1066 % kcctl config set-context cyber --cluster=https://cyberithub-kafka-connect-cyber.com --username=abfgfdc --password=cyberpass

 

Step 2: Stop Connector

Then you have to check the status of connector. If it is running then you have to stop the connector which you would like to use for republishing the data. Run kcctl stop connector <connector_name> command. In our case, we are republishing data through connector cyberithub-connector-republisher so we are using kcctl stop connector cyberithub-connector-republisher command as shown below.

cyberithub@macos1066 % kcctl stop connector cyberithub-connector-republisher
Stopped connector cyberithub-connector-republisher

 

Step 3: Prepare JSON file

Once the connector is stopped, below json file can be prepared to republish the required data. You can either publish a single object or multiple through one json file. In my case, I am republishing a data called cyberID through connector using below JSON file.

cyberithub@macos1066 % vi cyberithubdata.json
{
   "name": "cyberithub-connector-republisher",
   "config": {
       "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
       "tasks.max": "1",
       "connection.uri": "mongodb://${file:/usr/share/cyberithub/secrets.properties:username}:${file:/usr/share/cyberithub/secrets.properties:password}@cyberithub.example.server.com:27017/?ssl=true,
       "database": "testdb",
       "collection": "testcol",
       "topic.prefix": "cyber.col",
       "errors.tolerance": "none",
       "change.stream.full.document": "updateLookup",
       "startup.mode": "copy_existing",
       "startup.mode.copy.existing.pipeline": "[{ '$match': {$and:[ {'_id':ObjectId('cyberID')} ]} }]",
       "pipeline": "[{ '$match': { 'operationType': 'insert' }}]",
       "producer.override.max.request.size": "4081428"
   }
}

You can also republish multiple data in a single file by using multiple ObjectId section as shown in below JSON file.

cyberithub@macos1066 % vi cyberithubdata.json
{
   "name": "cyberithub-connector-republisher",
   "config": {
       "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
       "tasks.max": "1",
       "connection.uri": "mongodb://${file:/usr/share/cyberithub/secrets.properties:username}:${file:/usr/share/cyberithub/secrets.properties:password}@cyberithub.example.server.com:27017/?ssl=true,
       "database": "testdb",
       "collection": "testcol",
       "topic.prefix": "cyber.col",
       "errors.tolerance": "none",
       "change.stream.full.document": "updateLookup",
       "startup.mode": "copy_existing",
       "startup.mode.copy.existing.pipeline": "[{ '$match': { '$data': { '$in': [ ObjectId('cyberSecret'), ObjectId('cyberID') ] } } }]",
       "pipeline": "[{ '$match': { 'operationType': 'insert' }}]",
       "producer.override.max.request.size": "4081428"
    }
}

• name :  Name of the kafka connect connector
• config : Contains all configured properties of the connector
• connector.class : Specify connector implementation class
• tasks.max : Maximum number of kafka connect tasks
• connection.uri : MongoDB connection URI for connector to connect
• database : Name of MongoDB database from which the connector reads the data
• collection : To specify name of collection
• topic.prefix : Kafka topic name prefix used in records
• errors.tolerance : This property is used to handle record processing errors
• change.stream.full.document : This property is used to configure if MongoDB change stream events include full document associated with an event.
• startup.mode : This parameter tells how MongoDB read source connector read data once it starts
• startup.mode.copy.existing.pipeline : This configuration parameters is only used when startup.mode is set to copy_existing to copy existing documents along with the processing of streams.
• pipeline : This parameter decides which MongoDB change stream events will be sent to kafka
• producer.override.max.request.size : This parameter is used to specify maximum size of request kafka connect can produce overriding the size of producer message request.

 

Step 4: Apply the configuration

Once JSON file is ready, you can apply the configuration in required context by using kcctl apply -f <json_file> command as shown below. Our configuration has been saved in cyberithubdata.json file so to apply all the configuration from this file we are running kubectl apply -f cyberithubdata.json command as shown below.

cyberithub@macos1066 % kcctl apply -f cyberithubdata.json
Updated connector cyberithub-connector-republisher

 

Step 5: Delete Offset

Before republishing the data, you have to make sure to delete offset from your connector by using kubectl delete offset <connector_name> command as shown below. In our scenario connector name is cyberithub-connector-republisher so we are deleting the offsets by running kcctl delete offsets cyberithub-connector-republisher command as shown below.

cyberithub@macos1066 % kcctl delete offsets cyberithub-connector-republisher
The Connect framework-managed offsets for this connector have been reset successfully. However, if this connector manages offsets externally, they will need to be manually reset in the system that the connector uses.

 

Step 6: Verify Offset

Once deleted, you can verify the offset again by using kubectl get offsets cyberithub-connector-republisher command as shown below. You should see empty offsets as shown in below output. If it is all clear then we are good to proceed.

cyberithub@macos1066 % kcctl get offsets cyberithub-connector-republisher
{
  "offsets" : [ ]
}

 

Step 7: Resume Connector

Once all offsets are cleared, you are now ready to republish your data by starting the connector using kcctl resume connector <connector_name> command as shown below. Since in my case, connector name is cyberithub-connector-republisher then I am running kcctl resume connector cyberithub-connector-republisher command as shown below.

cyberithub@macos1066 % kcctl resume connector cyberithub-connector-republisher
Resumed connector cyberithub-connector-republisher

 

Step 8: Verify data

Once connector comes back to running state, you can verify the published data over the topic. All the data should be published over the topic. You can directly query the topic to check the published data.