Add Headers from Kafka Records to Mongo Field using Sink Connector: A Step-by-Step Guide

Introduction

In today’s data-driven world, integrating multiple systems is essential to unlock insights and make data-driven decisions. Apache Kafka and MongoDB are two popular technologies used for handling real-time data and storing large amounts of data, respectively. However, integrating these systems can be a daunting task, especially when it comes to preserving valuable metadata. In this article, we’ll explore how to add headers from Kafka records to a Mongo field using a Sink Connector, making it easy to preserve and utilize this valuable metadata.

The Problem: Losing Valuable Metadata

When integrating Kafka and MongoDB, one common issue is losing valuable metadata from Kafka records. This metadata, stored in the headers of Kafka records, provides crucial context about the data, such as the source system, event type, and timestamp. However, when using a traditional Sink Connector, this metadata is often lost during the data transfer process.

This loss of metadata can have severe consequences, making it challenging to:

• Track data lineage and provenance
• Implement data quality checks
• Perform auditing and logging

The Solution: Using a Custom Sink Connector

To preserve the valuable metadata from Kafka records, we can create a custom Sink Connector that adds the headers to a MongoDB field. This approach allows us to utilize the metadata in MongoDB, enabling better data management and analysis.

Prerequisites

Before we dive into the implementation, ensure you have the following:

• Kafka cluster with a topic containing records with headers
• MongoDB instance with a collection
• Confluent Platform or a Kafka distribution with Sink Connector support
• Java 8 or higher installed
• Maven or another build tool installed

Implementing the Custom Sink Connector

Let’s create a custom Sink Connector that adds the headers from Kafka records to a MongoDB field. We’ll use Java and the Confluent Kafka Connect API to build the connector.

Create a new Maven project and add the required dependencies:

<dependencies>
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-connect-api</artifactId>
    <version>2.5.0</version>
  </dependency>
  <dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongo-java-driver</artifactId>
    <version>3.12.7</version>
  </dependency>
</dependencies>

Create a new Java class that extends the `SinkConnector` class from the Kafka Connect API:

public class MongoSinkConnector extends SinkConnector {
 
    @Override
    public String version() {
        return "1.0";
    }
 
    @Override
    public Class<?> getKeyConverter() {
        return StringConverter.class;
    }
 
    @Override
    public Class<?> getValueConverter() {
        return JsonConverter.class;
    }
 
    @Override
    public void start(Map<String, String> props) {
        // Initialize the MongoDB client
        MongoClient mongoClient = new MongoClient(props.get("mongodb.url"));
        MongoDatabase database = mongoClient.getDatabase(props.get("mongodb.database"));
        MongoCollection<Document> collection = database.getCollection(props.get("mongodb.collection"));
 
        // Create a MongoDB Bulk Write Operation
        BulkWriteOperation bulkWriteOperation = new BulkWriteOperationBuilder(collection, true).build();
    }
 
    @Override
    public void put(Collection<SinkRecord> records) {
        for (SinkRecord record : records) {
            // Get the Kafka record headers
            Headers headers = record.headers();
 
            // Create a new MongoDB document
            Document document = new Document();
            document.put("key", record.key());
            document.put("value", record.value());
 
            // Add the Kafka headers to the MongoDB document
            Document headersDocument = new Document();
            for (Header header : headers) {
                headersDocument.put(header.key(), header.value());
            }
            document.put("headers", headersDocument);
 
            // Add the document to the Bulk Write Operation
            bulkWriteOperation.execute(new InsertOneModel<>(document));
        }
    }
 
    @Override
    public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // Flush the Bulk Write Operation
        bulkWriteOperation.execute();
    }
 
    @Override
    public void close() {
        // Close the MongoDB client
        mongoClient.close();
    }
}

name=mongo-sink
connector.class=com.example.MongoSinkConnector
tasks.max=1
 
mongodb.url=mongodb://localhost:27017
mongodb.database=mydatabase
mongodb.collection=mycollection

confluent local connect create --name mongo-sink --connector-class com.example.MongoSinkConnector --config mongo-sink.properties

confluent local connect start mongo-sink

kafka-console-producer --bootstrap-server localhost:9092 --topic mytopic

mongo
use mydatabase
db.mycollection.find().pretty()