diff --git a/README.md b/README.md
index b2b44bd46..45731e912 100644
--- a/README.md
+++ b/README.md
@@ -364,7 +364,70 @@ This has only been tested with Confluent Cloud but technically all SASL authenti
 
 [Back to top](#toc)
 
+# MongoDB Integration
 
+## Description and Configuration
+
+To sink streamed kafka topic data to a MongoDB database, a kafka connect and MongoDB instance can be deployed for the ODE. By running the provided docker compose [file](./docker-compose-mongo.yml) the following topics will be streamed to MongoDB:
+
+- OdeRawEncodedBSMJson
+- OdeBsmJson
+- OdeRawEncodedMAPJson
+- OdeMapJson
+- OdeRawEncodedSPATJson
+- OdeSpatJson
+- OdeRawEncodedTIMJson
+- OdeTimJson
+- OdeRawEncodedPsmJson
+- OdePsmJson
+
+The configuration that defines this is in the jpo-s3-deposit submodule [here](jpo-s3-deposit\mongo-connector\connect_start.sh). This script is attached to the `connect` container as a volume and if you would like to sink different topics then feel free to make a copy of the `connect_start.sh` script and attach it to the `connect` container to the following path: `/scripts/connect_start.sh`.
+
+## Environment variables
+
+### Purpose & Usage
+
+- The `MONGO_IP` environment variable is used to define the IP address of the MongoDB container. This can be configured to use a remote MongoDB instance instead of using the provided docker deployed container.
+
+- The `MONGO_DB_NAME` environmental variable defines the name of the DB created in MongoDB. This variable is used for both configuring user permission access as well as a destination for the connectors defined in the `connect` container.
+
+- The `MONGO_ADMIN_DB_USER` and `MONGO_ADMIN_DB_PASS` define the credentials for the `admin` MongoDB user. This user has full control of the cluster and the password must be securely set for production deployments.
+
+- The `MONGO_ODE_DB_USER` and `MONGO_ODE_DB_PASS` define the credentials for the `ode` MongoDB user. This user has `readWrite` permissions to the `MONGO_DB_NAME` database.
+
+- The `MONGO_URI` environmental variable contains the complete connection string used to connect to the MongoDB when creating connectors in the `connect` container.
+
+- The `MONGO_COLLECTION_TTL` environmental variable configures the Time To Live (TTL) for created TTL indexes. Setting this value too high will result in much more storage usage.
+
+### Values
+In order to utilize Confluent Cloud:
+
+- `MONGO_IP` must be set to the IP address of the MongoDB container. This can be left as `${DOCKER_HOST_IP}` for deployments using the provided MongoDB instance included in the docker-compose file.
+
+- `MONGO_DB_NAME` configures the created DB name in MongoDB.
+
+- `MONGO_ADMIN_DB_USER` configures the MongoDB admin user's name.
+
+- `MONGO_ADMIN_DB_PASS` configures the MongoDB admin user's name. This must be changed to a more secure password for production deployments.
+
+- `MONGO_ODE_DB_USER` configures the username of the initialized user with `readwrite` access to the initialized database.
+
+- `MONGO_ODE_DB_PASS` configures the password of the initialized user with `readwrite` access to the initialized database.
+
+- `MONGO_URI` defines the connection URI used by the kafka connect instance. MongoDB connection URI options are documented [here](https://www.mongodb.com/docs/manual/reference/connection-string/)
+
+- `MONGO_COLLECTION_TTL` sets the Time To Live (TTL) for the created TTL indexes.
+
+
+## Mongo Docker Compose File
+
+There is a provided docker-compose [file](docker-compose-mongo.yml) that spins up a MongoDB instance with a kafka connect service. There is also a initialization container that configures the RBAC and replica set of the MongoDB container. 
+
+## Note
+
+Kafka connect is being used for MongoDB in this implimentation but it can interact with many types of databases, here is further documentation for [kafka connect](https://docs.confluent.io/platform/current/connect/index.html)
+
+[Back to top](#toc)
 
 <!--
 #########################################
diff --git a/docker-compose-mongo.yml b/docker-compose-mongo.yml
new file mode 100644
index 000000000..6ae0d0d21
--- /dev/null
+++ b/docker-compose-mongo.yml
@@ -0,0 +1,92 @@
+version: '3'
+
+include:
+  - docker-compose.yml
+
+services:
+  mongo:
+    image: mongo:7
+    container_name: mongo 
+    restart: always
+    ports:
+      - "27017:27017"
+    environment:
+      MONGO_INITDB_ROOT_USERNAME: ${MONGO_ADMIN_DB_USER}
+      MONGO_INITDB_ROOT_PASSWORD: ${MONGO_ADMIN_DB_PASS}
+      MONGO_INITDB_DATABASE: admin 
+    entrypoint:
+      - bash
+      - -c
+      - |
+          openssl rand -base64 741 > /mongo_keyfile
+          chmod 400 /mongo_keyfile
+          chown 999:999 /mongo_keyfile
+          exec docker-entrypoint.sh $$@
+    command: "mongod --bind_ip_all --replSet rs0 --keyFile /mongo_keyfile"
+    volumes:
+      - mongo_data:/data/db
+    healthcheck:
+      test: | 
+        echo 'db.runCommand("ping").ok' | mongosh localhost:27017/test --quiet
+      interval: 10s
+      start_period: 30s
+
+  mongo-setup:
+    image: mongo:7
+    container_name: mongo_setup
+    depends_on:
+      - mongo
+    restart: on-failure
+    environment:
+      MONGO_ADMIN_DB_USER: ${MONGO_ADMIN_DB_USER}
+      MONGO_ADMIN_DB_PASS: ${MONGO_ADMIN_DB_PASS}
+      MONGO_DB_NAME: ${MONGO_DB_NAME}
+      MONGO_ODE_DB_USER: ${MONGO_ODE_DB_USER}
+      MONGO_ODE_DB_PASS: ${MONGO_ODE_DB_PASS}
+      MONGO_COLLECTION_TTL: ${MONGO_COLLECTION_TTL}
+    entrypoint: ["/bin/bash", "setup_mongo.sh"]
+    volumes:
+      - ./scripts/mongo/setup_mongo.sh:/setup_mongo.sh
+      - ./scripts/mongo/create_indexes.js:/create_indexes.js
+
+
+  connect:
+    image: kafka-connect:latest
+    restart: always
+    build:
+      context: ./jpo-s3-deposit/mongo-connector
+      dockerfile: Dockerfile
+    ports:
+      - "8083:8083"
+    depends_on:
+      mongo:
+        condition: service_healthy
+    environment:
+      MONGO_URI: ${MONGO_URI}
+      MONGO_DB_NAME: ${MONGO_DB_NAME}
+      CONNECT_BOOTSTRAP_SERVERS: ${DOCKER_HOST_IP}:9092
+      CONNECT_REST_ADVERTISED_HOST_NAME: connect
+      CONNECT_REST_PORT: 8083
+      CONNECT_GROUP_ID: compose-connect-group
+      CONNECT_CONFIG_STORAGE_TOPIC: topic.kafka-connect-configs
+      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_CONFIG_STORAGE_CLEANUP_POLICY: compact
+      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
+      CONNECT_OFFSET_STORAGE_TOPIC: topic.kafka-connect-offsets
+      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_OFFSET_STORAGE_CLEANUP_POLICY: compact
+      CONNECT_STATUS_STORAGE_TOPIC: topic.kafka-connect-status
+      CONNECT_STATUS_STORAGE_CLEANUP_POLICY: compact
+      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_LOG4J_ROOT_LOGLEVEL: "ERROR"
+      CONNECT_LOG4J_LOGGERS: "org.apache.kafka.connect.runtime.rest=ERROR,org.reflections=ERROR,com.mongodb.kafka=ERROR"
+      CONNECT_PLUGIN_PATH: /usr/share/confluent-hub-components
+    volumes:
+      - ./jpo-s3-deposit/mongo-connector/connect_start.sh:/scripts/connect_start.sh
+
+volumes:
+  mongo_data:
\ No newline at end of file
diff --git a/docker-compose.yml b/docker-compose.yml
index 5a6d29c1c..f61839bff 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -6,7 +6,7 @@ services:
     ports:
       - "9092:9092"
     volumes:
-      - "${DOCKER_SHARED_VOLUME}:/bitnami"
+      - kafka:/bitnami
     environment:
       KAFKA_ENABLE_KRAFT: "yes"
       KAFKA_CFG_PROCESS_ROLES: "broker,controller"
@@ -268,3 +268,7 @@ services:
       options:
         max-size: "10m"  
         max-file: "5"
+
+volumes:
+  kafka:
+    {}
\ No newline at end of file
diff --git a/sample.env b/sample.env
index 8ca50533a..c48737ea6 100644
--- a/sample.env
+++ b/sample.env
@@ -116,6 +116,16 @@ RDE_TIM_GROUP=group_rde_tim
 ## Required if using SDX depositor module (REST interface)
 SDW_API_KEY=
 
+# Required MONGODB Variables
+MONGO_IP=${DOCKER_HOST_IP}
+MONGO_DB_NAME=ode
+MONGO_ADMIN_DB_USER=admin
+MONGO_ADMIN_DB_PASS=password
+MONGO_ODE_DB_USER=ode
+MONGO_ODE_DB_PASS=password
+MONGO_URI=mongodb://${MONGO_ODE_DB_USER}:${MONGO_ODE_DB_PASS}@${MONGO_IP}:27017/?directConnection=true
+MONGO_COLLECTION_TTL=7 # days
+
 ## Optional overrides
 #SDW_DESTINATION_URL=
 #SDW_GROUP_ID=
diff --git a/scripts/mongo/create_indexes.js b/scripts/mongo/create_indexes.js
new file mode 100644
index 000000000..04487a537
--- /dev/null
+++ b/scripts/mongo/create_indexes.js
@@ -0,0 +1,206 @@
+// Create indexes on all collections
+
+/*
+This script is responsible for initializing the replica set, creating collections, adding indexes and TTLs
+*/
+console.log("Running create_indexes.js");
+
+const ode_db = process.env.MONGO_DB_NAME;
+const ode_user = process.env.MONGO_ODE_DB_USER;
+const ode_pass = process.env.MONGO_ODE_DB_PASS;
+
+const ttlInDays = process.env.MONGO_COLLECTION_TTL; // TTL in days
+const expire_seconds = ttlInDays * 24 * 60 * 60;
+const retry_milliseconds = 5000;
+
+console.log("ODE DB Name: " + ode_db);
+
+try {
+    console.log("Initializing replica set...");
+
+    var config = {
+        "_id": "rs0",
+        "version": 1,
+        "members": [
+        {
+            "_id": 0,
+            "host": "mongo:27017",
+            "priority": 2
+        },
+        ]
+    };
+    rs.initiate(config, { force: true });
+    rs.status();
+} catch(e) {
+    rs.status().ok
+}
+
+// name -> collection name
+// ttlField -> field to perform ttl on 
+// timeField -> field to index for time queries
+
+const collections = [
+    {name: "OdeBsmJson", ttlField: "recordGeneratedAt", timeField: "metadata.odeReceivedAt"},
+    {name: "OdeRawEncodedBSMJson", ttlField: "recordGeneratedAt", timeField: "none"},
+
+    {name: "OdeMapJson", ttlField: "recordGeneratedAt", timeField: "metadata.odeReceivedAt"},
+    {name: "OdeRawEncodedMAPJson", ttlField: "recordGeneratedAt", timeField: "none"},
+
+    {name: "OdeSpatJson", ttlField: "recordGeneratedAt", timeField: "metadata.odeReceivedAt"},
+    {name: "OdeRawEncodedSPATJson", ttlField: "recordGeneratedAt", timeField: "none"},
+    
+    {name: "OdeTimJson", ttlField: "recordGeneratedAt", timeField: "metadata.odeReceivedAt"},
+    {name: "OdeRawEncodedTIMJson", ttlField: "recordGeneratedAt", timeField: "none"},
+
+    {name: "OdePsmJson", ttlField: "recordGeneratedAt", timeField: "metadata.odeReceivedAt"},
+    {name: "OdeRawEncodedPsmJson", ttlField: "recordGeneratedAt", timeField: "none"},
+];
+
+// Function to check if the replica set is ready
+function isReplicaSetReady() {
+    let status;
+    try {
+        status = rs.status();
+    } catch (error) {
+        console.error("Error getting replica set status: " + error);
+        return false;
+    }
+
+    // Check if the replica set has a primary
+    if (!status.hasOwnProperty('myState') || status.myState !== 1) {
+        console.log("Replica set is not ready yet");
+        return false;
+    }
+
+    console.log("Replica set is ready");
+    return true;
+}
+
+try{
+
+    // Wait for the replica set to be ready
+    while (!isReplicaSetReady()) {
+        sleep(retry_milliseconds);
+    }
+    sleep(retry_milliseconds);
+    // creates another user
+    console.log("Creating ODE user...");
+    admin = db.getSiblingDB("admin");
+    // Check if user already exists
+    var user = admin.getUser(ode_user);
+    if (user == null) {
+        admin.createUser(
+            {
+                user: ode_user,
+                pwd: ode_pass,
+                roles: [
+                    { role: "readWrite", db: ode_db },
+                ]
+            }
+        );
+    } else {
+        console.log("User \"" + ode_user + "\" already exists.");
+    }
+
+} catch (error) {
+    print("Error connecting to the MongoDB instance: " + error);
+}
+
+
+// Wait for the collections to exist in mongo before trying to create indexes on them
+let missing_collection_count;
+const db = db.getSiblingDB(ode_db);
+do {
+    try {
+        missing_collection_count = 0;
+        const collection_names = db.getCollectionNames();
+        for (collection of collections) {
+            console.log("Creating Indexes for Collection" + collection["name"]);
+            // Create Collection if It doesn't exist
+            let created = false;
+            if(!collection_names.includes(collection.name)){
+                created = createCollection(collection);
+                // created = true;
+            }else{
+                created = true;
+            }
+
+            if(created){
+                if (collection.hasOwnProperty('ttlField') && collection.ttlField !== 'none') {
+                    createTTLIndex(collection);  
+                }
+
+
+            }else{
+                missing_collection_count++;
+                console.log("Collection " + collection.name + " does not exist yet");
+            }
+        }
+        if (missing_collection_count > 0) {
+            print("Waiting on " + missing_collection_count + " collections to be created...will try again in " + retry_milliseconds + " ms");
+            sleep(retry_milliseconds);
+        }
+    } catch (err) {
+        console.log("Error while setting up TTL indexes in collections");
+        console.log(rs.status());
+        console.error(err);
+        sleep(retry_milliseconds);
+    }
+} while (missing_collection_count > 0);
+
+console.log("Finished Creating All TTL indexes");
+
+
+function createCollection(collection){
+    try {
+        db.createCollection(collection.name);
+        return true;
+    } catch (err) {
+        console.log("Unable to Create Collection: " + collection.name);
+        console.log(err);
+        return false;
+    }
+}
+
+// Create TTL Indexes
+function createTTLIndex(collection) {
+    if (ttlIndexExists(collection)) {
+        console.log("TTL index already exists for " + collection.name);
+        return;
+    }
+
+    const collection_name = collection.name;
+    const timeField = collection.ttlField;
+
+    console.log(
+        "Creating TTL index for " + collection_name + " to remove documents after " +
+        expire_seconds +
+        " seconds"
+    );
+
+    try {
+        var index_json = {};
+        index_json[timeField] = 1;
+        db[collection_name].createIndex(index_json,
+            {expireAfterSeconds: expire_seconds}
+        );
+        console.log("Created TTL index for " + collection_name + " using the field: " + timeField + " as the timestamp");
+    } catch (err) {
+        var pattern_json = {};
+        pattern_json[timeField] = 1;
+        db.runCommand({
+            "collMod": collection_name,
+            "index": {
+                keyPattern: pattern_json,
+                expireAfterSeconds: expire_seconds
+            }
+        });
+        console.log("Updated TTL index for " + collection_name + " using the field: " + timeField + " as the timestamp");
+    }
+
+}
+
+
+function ttlIndexExists(collection) {
+    return db[collection.name].getIndexes().find((idx) => idx.hasOwnProperty('expireAfterSeconds')) !== undefined;
+}
\ No newline at end of file
diff --git a/scripts/mongo/setup_mongo.sh b/scripts/mongo/setup_mongo.sh
new file mode 100644
index 000000000..13cb4e7b7
--- /dev/null
+++ b/scripts/mongo/setup_mongo.sh
@@ -0,0 +1,11 @@
+#!/bin/bash 
+
+until mongosh --host mongo:27017 --eval 'quit(db.runCommand({ ping: 1 }).ok ? 0 : 2)' &>/dev/null; do
+  sleep 1
+done
+
+echo "MongoDB is up and running!"
+
+cd /
+
+mongosh -u $MONGO_ADMIN_DB_USER -p $MONGO_ADMIN_DB_PASS --authenticationDatabase admin --host mongo:27017 /create_indexes.js