Forem: Tim Kelly

Build a Spring AI MCP Server With MongoDB

Tim Kelly — Tue, 11 Nov 2025 13:54:18 +0000

In this tutorial, we're going to build a Model Context Protocol (MCP) server that connects AI models to MongoDB, using it to power a simple todo list application. If you've been working with AI applications, you've probably run into the challenge of getting LLMs to interact with your actual systems and data. That's exactly what the Model Context Protocol solves! It's quickly becoming the standard way to connect AI models to the real world. Whether you're building agents that need to query databases, access APIs, or interact with your company's internal tools, understanding MCP is becoming essential. By the end of this tutorial, you'll have a working server that exposes MongoDB operations as tools any MCP-compatible AI can use, and you'll understand the patterns for building your own servers for whatever data sources you need to connect.

MongoDB is a natural fit for AI applications—it's flexible, scalable, and has built-in vector search capabilities that make it easy to work with embeddings and semantic search. If you're curious about the AI side of MongoDB and want to dive deeper into vector search, there's a Vector Search Fundamentals course that'll walk you through those concepts (and you can earn a skill badge while you're at it).

If you want to take a look at the code used in this tutorial, check out the GitHub repository.

What is the Model Context Protocol?

The Model Context Protocol (MCP) is an open standard that enables AI applications to securely connect to various data sources and tools. Think of it as a universal adapter that lets your AI models interact with databases, APIs, file systems, and other services in a standardized way. Instead of building custom integrations for every data source, MCP provides a common language that both AI systems and data providers can speak.

At its core, MCP defines how servers expose their capabilities (like database queries or file operations) as tools, resources, and prompts that AI models can discover and use. This makes it incredibly powerful for building context-aware AI applications that need to pull from multiple sources.

The difference between an MCP server and an MCP client

An MCP server is the component that exposes capabilities to AI systems. It's what we'll be building in this tutorial. The server defines tools (functions the AI can call), resources (data the AI can access), and prompts (templated conversations). When we build our Spring AI MCP server, we're creating something that says, "Hey, I can interact with MongoDB for you—here are the operations I support."

An MCP client, on the other hand, is the component that consumes these capabilities. This is typically your AI application or agent that discovers available tools from MCP servers and decides when to use them. The client sends requests to servers and handles the responses. In the broader ecosystem, applications like Claude Desktop or other AI interfaces act as MCP clients that can connect to your servers.

What is Spring AI?

Spring AI is an application framework for AI engineering. Its goal is to apply to the AI domain Spring ecosystem design principles such as portability and modular design and promote using POJOs as the building blocks of an application to the AI domain.

Prerequisites

Java 17+
Maven
A MongoDB Atlas account with a cluster set up
- An M0 free forever tier is perfect
Node (version 22.7.5 or higher is recommended)
- We need this as we will be using the MCP Inspector to test our application

Build a Spring app and our dependencies

To start building our application, we will need to bring in a few dependencies. The easiest way to get started is using Spring Initializr. We are going to need the Model Context Protocol Server and the Spring Web dependencies. We'll be adding the Spring Data MongoDB dependency later on, but we'll keep it to just the first two for now to make getting started easier.

For Project, I have selected Maven; for language, Java; and Spring Boot version 3.5.7 (or the latest stable release). Now, we can generate our Jar and open the downloaded app in our IDE.

Configure our MCP server

First things first, we need to update our Spring AI version to the latest milestone 1.1.0-M3. It is possible to do all this with the latest stable release, but the newer interface provides a much cleaner setup.

Open the POM.XML and update the Spring AI version:

<spring-ai.version>1.1.0-M3</spring-ai.version>

Now, we can define our MCP details in the application.properties file:

spring.application.name=springai-mcp
spring.ai.mcp.server.name=mongodb-mcp
spring.ai.mcp.server.version=1.0.0

spring.ai.mcp.server.protocol=streamable
spring.ai.mcp.server.stdio=false
spring.ai.mcp.server.type=sync

We give our server a name and version—this helps clients identify what they're connecting to. The name mongodb-mcp tells clients this server provides MongoDB capabilities, and the version lets us track updates over time.

We want an HTTP streamable service, no stdio, and synchronous responses. This means our server will communicate over HTTP rather than standard input/output, and it'll handle requests synchronously rather than asynchronously. This keeps things simple for our tutorial, though async operations are definitely an option for production use cases.

Build your first MCP tool

An MCP tool is essentially a function that an AI model can call. When you annotate a method with @McpTool, you're telling the MCP framework, "This is something an AI can use." The AI will be able to see the tool's name, description, and parameters, then decide whether to call it based on what the user is asking for.

Create a class MongoDbTools.java, where we will create a @Component.

We will use the @McpTool annotation. The name we give it will be how the AI identifies and calls this tool—think of it as the function name the AI will invoke. The description is important as it tells the AI what this tool does and when to use it. A good description helps the AI make smart decisions about when to call your tool. Think of this like a prompt template. When we give our application a natural language prompt, these descriptions allow the LLM to discern which tool is best for the job, and orchestrate the tools to achieve what we are asking it to do.

package com.timkelly.springaimcp;

import org.springaicommunity.mcp.annotation.McpTool;
import org.springframework.stereotype.Component;

@Component
public class MongoDbTools {

    // Tools
    @McpTool(name="get-todo-items", description = "I will return the items of a todo list")
    public String getTodoItems() {
        var todo = """
                1. Make Spring AI MCP tutorial

                2. Have a coffee

                3. Walk the dog
                """;

        return todo;
    }
}

This is where we would define @McpResources or @McpPrompts. The @McpResource annotation provides access to resources via URI templates. The @McpPrompt annotation generates prompt messages for AI interactions. These go beyond what we cover in this tutorial, but you can read about them in the Spring AI MCP documentation.

How we will test our MCP server

First, we need to run our application.

mvn spring-boot:run

To test our application, we will use the MCP Inspector. This is an interactive developer tool for testing and debugging MCP servers.

If you want to read more about it, check out the Debugging guide, which covers the Inspector as part of the overall debugging toolkit. For this tutorial, all we are going to do is use it to make sure our application is up and running, exposing the tools we create, and testing those tools.

The Inspector runs with the following:

npx @modelcontextprotocol/inspector

If this is your first time running it, you will likely need to install it:

Need to install the following packages:
@modelcontextprotocol/inspector@0.17.2
Ok to proceed? (y) y

Once this spins up, it will open in the browser.

Now, for the transport type in the top left, we need to change this to Streamable HTTP. This will change the interface slightly, giving us the option to define the URL our app is running on. Mine is localhost:8080, so I will enter http://localhost:8080/mcp. The /mcp is important to add—this is the default endpoint where Spring AI MCP exposes the protocol.

At this point, we can just hit connect. We will receive some server logging in the history section. It will confirm the name of the server, update when tools change, all that good stuff.

From the top bar we can select tools, and here we can list all the tools we have available. If we select a tool, we can then choose to run it in the browser.

If everything is working correctly, we should be able to confirm our output.

Success! Now that we have this confirmed, we can look to add some functionality to our todo list by connecting to MongoDB as our database.

Connect to MongoDB

Add the Spring Data MongoDB dependency to your pom.xml:

<dependency>  
    <groupId>org.springframework.boot</groupId>  
    <artifactId>spring-boot-starter-data-mongodb</artifactId>  
</dependency>

Add connection details to the application.properties file:

#...

# MongoDB Configuration
# 1. Copy your connection string from MongoDB Atlas
# 2. Replace ?appName=Cluster with ?appName=SpringAiMcp, or add it if no appName is present
#
# Before: mongodb+srv://user:pass@cluster.abc12.mongodb.net/?appName=Cluster
# After:  mongodb+srv://user:pass@cluster.abc12.mongodb.net/?appName=SpringAiMcp

spring.data.mongodb.uri=<YOUR-CONNECTION-STRING>
spring.data.mongodb.database=todo

For our connection string, it is best practice to set the app name so we can see clearly what queries or operations are tied to which apps. For this, set the appName to SpringAiMcp by adding it to the end of your connection string. The database name todo is where we'll store all our task data.

Quick connection check (optional)

Run:

mvn spring-boot:run

Look for a line similar to:

Monitor thread successfully connected to server with description ServerDescription{address=...}

If you see authentication or timeout errors, check:

The connection string credentials.
Your IP access list in MongoDB Atlas.
That the appName is added correctly at the end of the URI.

Build our MongoDB tool

Now that we have MongoDB connected, it's time to replace our hardcoded todo list with real database operations. We're going to build out a complete set of tools that let an AI manage tasks in MongoDB—adding new tasks, marking them as complete, and retrieving tasks with various filters.

The beauty of the MCP approach here is that we're not writing AI-specific code. We're just building normal Spring Data repositories and services, then exposing them as tools with a few annotations. The AI doesn't need to know anything about MongoDB connection strings, query syntax, or data modeling—it just needs to know, "Here's a tool called todo-add-task that takes a task name." Our MCP server handles all the translation between the AI's requests and the actual database operations.

We'll structure this in layers, following Spring best practices. First, we'll define our data model and repository for database access. Then, we'll create a service layer to handle our business logic. Finally, we'll expose these operations as MCP tools that AI models can discover and use. This keeps our code clean, testable, and easy to extend with new functionality later.

Let's start by setting up our data model.

Define our Task model and repository

First, let's create our Task model in Task.java. This represents a single todo item in our MongoDB collection:

package com.timkelly.springaimcp;  

import org.bson.types.ObjectId;  
import org.springframework.data.annotation.Id;  
import org.springframework.data.mongodb.core.mapping.Document;  

@Document(collection = "tasks")  
public class Task {  
    @Id  
    private ObjectId id;  
    private String name;  
    private boolean completed;  

    public Task(ObjectId id, String name) {  
        this.id = id;  
        this.name = name;
    }  

    public ObjectId getId() {  
        return id;  
    }  

    public String getName() {  
        return name;  
    }  

    public void setName(String name) {  
        this.name = name;  
    }  

    public boolean isCompleted() {  
        return completed;  
    }  

    public void setCompleted(boolean completed) {  
        this.completed = completed;  
    }  
}

Now, create our repository interface in TodoRepository.java. Spring Data MongoDB will handle the implementation for us—this is one of the best features of Spring Data. We just define an interface that extends MongoRepository, and Spring will automatically generate all the common CRUD operations (create, read, update, delete) at runtime. No need to write boilerplate code for basic database operations like findAll(), save(), or deleteById()—they're all provided out of the box.

package com.timkelly.springaimcp;  

import org.bson.types.ObjectId;  
import org.springframework.data.mongodb.repository.MongoRepository;  
import org.springframework.stereotype.Repository;  

import java.util.List;  

@Repository  
public interface TodoRepository extends MongoRepository<Task, ObjectId> {  
}

The generic types <Task, ObjectId> tell Spring what entity we're working with and what type the ID field is. Spring Data MongoDB is smart enough to map our Java objects to BSON documents in MongoDB and back again, handling all the serialization for us.

Creating our add task tool

Let's create a service layer to handle our business logic. Create TodoService.java:

package com.timkelly.springaimcp;  

import org.bson.types.ObjectId;  
import org.springframework.stereotype.Service;  

import java.util.List;  

@Service  
public class TodoService {  

    private final TodoRepository todoRepository;  

    public TodoService(TodoRepository todoRepository) {  
        this.todoRepository = todoRepository;  
    }  

    public void addTask(String name) {  
        todoRepository.save(new Task(new ObjectId(), name));  
    }  
}

Now, update our MongoDbTools.java class to wire in the service and add our first real tool:

package com.timkelly.springaimcp;  

import org.springaicommunity.mcp.annotation.McpTool;  
import org.springaicommunity.mcp.annotation.McpToolParam;  
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;  
import org.springframework.stereotype.Component;  

import java.util.List;  

@Component  
@ConditionalOnBean(TodoService.class)  
public class MongoDbTools {  

    private final TodoService todoService;  

    public MongoDbTools(TodoService todoService) {  
        this.todoService = todoService;  
    }  

    @McpTool(  
            name = "todo-add-task",  
            description = "Add a new to-do task to MongoDB"  
    )  
    public String addTask(  
            @McpToolParam(  
                    description = "The name or description of the new task",  
                    required = true  
            ) String name  
    ) {  
        todoService.addTask(name);  
        return "Task added successfully: " + name;  
    }  
}

The @McpToolParam annotation allows us to provide metadata about each parameter. The description helps the AI understand what to pass in, and marking it as required ensures the AI knows this parameter can't be optional. The better we describe our parameters, the better the AI can use our tools correctly.

Update our tasks

Let's add a query method to our TodoRepository.java so we can find tasks by name:

package com.timkelly.springaimcp;  

import org.bson.types.ObjectId;  
import org.springframework.data.mongodb.repository.MongoRepository;
import org.springframework.data.mongodb.repository.Update; 
import org.springframework.stereotype.Repository;

@Repository  
public interface TodoRepository extends MongoRepository<Task, ObjectId> {  
    @Update("{ '$set' : { 'completed' : ?0 } }")
    void updateCompletedByName(String name, boolean completed);
}

This is another powerful Spring Data feature—query derivation from method names. Spring Data MongoDB will parse the method name findByName and automatically generate the MongoDB query for us. It sees "findBy" and knows we want to query, then "Name" tells it which field to match against. At runtime, this becomes a query like db.tasks.find({ name: "some name" }) without us writing any query code. As long as we follow Spring Data's naming conventions, we can create complex queries just by naming our methods correctly—things like findByCompletedTrueAndNameContaining would work exactly as you'd expect.

Now, add the completeTask method to TodoService.java:

   public void setCompletedByName(String name, boolean completed) {
        todoRepository.updateCompletedByName(name, completed);
    }

And add the corresponding tool to MongoDbTools.java:

   @McpTool(
            name = "todo-complete-task",
            description = "Mark a to-do task as complete by name"
    )
    public String completeTask(
            @McpToolParam(
                    description = "The name of the task to mark as complete or incomplete",
                    required = true
            ) String name,
            @McpToolParam(
                    description = "The status of the task, either complete(true) or incomplete(false)"
            ) boolean status
    ) {
        todoService.setCompletedByName(name, status);
        return "Marked task as " + status + ": " + name;
    }
}

Creating our get tasks tools

Let's add more query methods to our TodoRepository.java to support filtering by completion status:

package com.timkelly.springaimcp;  

import org.bson.types.ObjectId;  
import org.springframework.data.mongodb.repository.MongoRepository;
import org.springframework.data.mongodb.repository.Update;  
import org.springframework.stereotype.Repository;  

import java.util.List;  

@Repository  
public interface TodoRepository extends MongoRepository<Task, ObjectId> {  
    @Update("{ '$set' : { 'completed' : ?1 } }")
    void updateCompletedByName(String name, boolean completed);
    List<Task> findByCompletedTrue();
    List<Task> findByCompletedFalse();
}

Add the getTasks method to TodoService.java:

public List<Task> getTasks(String filter) {  
    if (filter == null || filter.isBlank() || filter.equalsIgnoreCase("all")) {  
        return todoRepository.findAll();  
    }  

    return switch (filter.toLowerCase()) {  
        case "incomplete" -> todoRepository.findByCompletedFalse();  
        case "complete" -> todoRepository.findByCompletedTrue();  
        default -> todoRepository.findAll();  
    };
}

And finally, add the tool to MongoDbTools.java:

@McpTool(  
        name = "todo-get-tasks",  
        description = "Retrieve to-do list items, optionally filtered by completion status"  
)  
public List<Task> getTasks(  
        @McpToolParam(  
                description = "Filter tasks by completion status: 'complete', 'incomplete', or 'all' (default: 'all')",  
                required = false  
        ) String filter  
) {  
    return todoService.getTasks(filter);
}

Testing our application

Make sure to set your MongoDB connection string before running the application:

mvn spring-boot:run

Now, if we go back to our MCP Inspector, we can see our new tools listed in the tools section.

You can test each one individually—try adding a task, marking it complete, and retrieving tasks with different filters.

The Inspector will show you the requests being sent and the responses coming back, making it easy to verify everything is working as expected.

And there you have it! You've built a fully functional MCP server that exposes MongoDB operations as tools that AI models can use. This same pattern can be extended to build servers for all kinds of data sources and operations—the possibilities are pretty much endless.

Conclusion

Congrats! You've just built a fully functional MCP server that bridges the gap between AI models and MongoDB. What started as a simple, hardcoded todo list evolved into a real application with persistent storage, CRUD operations, and a clean architecture that any AI client can interact with.

The power of what you've built goes way beyond just a todo list. You now understand the fundamental pattern for exposing any data source or service to AI models through MCP. The @McpTool, @McpToolParam, and the other MCP annotations give you a consistent way to make anything accessible to AI.

What's really cool is how little AI-specific code we actually wrote. Most of what we built was just normal Spring Boot application code—models, repositories, services. The MCP layer was just a thin wrapper on top that made everything discoverable and callable by AI models. This means you can take existing Spring applications and MCP-enable them without major rewrites.

From here, you could extend this in tons of directions. Add authentication so different users have their own task lists. Create more complex queries with date filters or priority sorting. Build out @McpResources to expose collection schemas or statistics. Add @McpPrompts to help guide AI models on how to best use your tools. The Spring AI MCP framework gives you all the building blocks you need.

The Model Context Protocol is still relatively new, but it's quickly becoming the standard way to connect AI models to real-world data and systems. Getting comfortable with building MCP servers now puts you ahead of the curve as this ecosystem continues to grow. Whether you're building internal tools for your team, creating services for AI agents, or just experimenting with what's possible when AI can access real data, MCP is the bridge that makes it all work.

If you found this tutorial useful, check out my other tutorial, Secure Local RAG With Role-Based Access: Spring AI, Ollama, & MongoDB.

Build Your First AI Agent With MongoDB and LangChain4j

Tim Kelly — Thu, 16 Oct 2025 15:54:23 +0000

AI agents are everywhere right now. We've all heard the pitch: They reason about problems, use tools autonomously, and chain together multiple steps to accomplish goals without constant hand-holding. They're being deployed to book flights, analyze data pipelines, and handle customer support—taking on tasks that previously required either rigid automation or human intervention.

This tutorial sits as a nice introduction to building your first AI agent with LangChain4j. We'll create a movie recommendation agent that:

Understands natural language descriptions of plots ("a sci-fi movie about rebels fighting an empire").
Searches semantically through a database using vector embeddings.
Calls external APIs to fetch real-time streaming availability.
Orchestrates these steps autonomously based on the query.
Returns clean, conversational answers instead of raw JSON.

This agent uses MongoDB Atlas for vector search, LangChain4j's agentic framework for orchestration, and the Watchmode API for streaming data. By the end, you'll understand not just how to wire up these components, but how agentic systems actually work: how LLMs plan multi-step workflows, how tools share state, and how to give your agent instructions without hardcoding every possible scenario.

Vector search is what will allow us to search our data with natural language, and if you'd like to learn more and earn a skills badge, check out our Vector Search Fundamentals skills badge.

If you just want the code, it's available in this GitHub repo.

What is an AI agent?

An AI agent is a system that can take in information about its environment, decide what to do next, and work towards a goal, with minimum human intervention. Instead of behaving like a static chatbot that only answers questions, an agent can plan, use tools (like databases or APIs), and adapt based on results.

While the defining traits of an AI agent are the ability to reason and act, there’s no single paradigm for how an agent must function. One popular approach is the ReAct framework (Reasoning + Acting), where the agent:

Thinks about the problem.
Takes an action (e.g., querying a search engine).
Observes the result.
Repeats until it can deliver a complete answer.

In short:

Chatbot: Answers questions based on what it already knows
AI agent: Reasons about what it needs to learn, uses tools to gather that information, and works toward a goal

For our movie recommendation agent, we'll use a supervisor pattern, where a planning model orchestrates multiple specialized tools to find movies by plot description, look up streaming availability, and return a clean answer to the user.

What is LangChain4j?

LangChain4j is an open-source Java library designed to simplify building LLM-powered applications. It provides a unified interface for working with multiple LLM providers (OpenAI, Anthropic, etc.) and vector stores like MongoDB Atlas.

While the name invites comparison to the Python-based LangChain project, LangChain4j is more of a fusion, drawing inspiration from Haystack, LlamaIndex, and the wider AI community, all while staying laser-focused on the needs of Java developers.

Development kicked off in early 2023 during the ChatGPT boom, and while the project is still evolving, its core functionality is stable and production-ready. If you're exploring LLM-powered applications in Java, LangChain4j is a pragmatic and actively maintained option.

For this tutorial, we're specifically using the langchain4j-agentic module, which provides abstractions for building agentic systems, including the supervisor pattern that will orchestrate our movie search workflow.

Prerequisites

Before we get started, here's what you'll need:

Java 8 or later installed and ready to go (I'm using Java 24)
Maven for building the project (version 3.9.10 or later recommended)
A MongoDB Atlas cluster—a free M0 tier is perfect for this tutorial. If you need help setting one up, check out the Get Started with Atlas guide.
API keys:
- Voyage AI for generating embeddings (sign up here)
- You can use the OpenAI API for both the embeddings and chat model, if you prefer.
- OpenAI for the planning model (get your key here)
- Watchmode for streaming availability data (register here)

Make sure your IP address is whitelisted in your MongoDB Atlas cluster's network access settings, and create a database user with read/write permissions.

Our dataset

For this tutorial, we're using the IMDB Top 1000 Movies dataset from Kaggle. Download the CSV file and place it in your project's src/main/resources directory, naming it imdb_top_1000.csv.

This dataset includes movie titles, plot overviews, genres, directors, and IMDB ratings. We'll be embedding the overview field (the plot description) so users can search for movies semantically—e.g., "Find me a sci-fi movie about rebels fighting an empire"—instead of needing to remember the exact title.

Creating our app

Let's scaffold a new Maven project. Create a pom.xml file with the following dependencies:


    <dependencies>
        <!-- LangChain4j core -->
        <dependency>
            <groupId>dev.langchain4j</groupId>
            <artifactId>langchain4j</artifactId>
            <version>1.5.0</version>
        </dependency>

        <!-- LangChain4j agentic module -->
        <dependency>
            <groupId>dev.langchain4j</groupId>
            <artifactId>langchain4j-agentic</artifactId>
            <version>1.5.0-beta11</version>
        </dependency>

        <!-- OpenAI integration for the planning model -->
        <dependency>
            <groupId>dev.langchain4j</groupId>
            <artifactId>langchain4j-open-ai</artifactId>
            <version>1.4.0</version>
        </dependency>

        <!-- MongoDB Atlas integration -->
        <dependency>
            <groupId>dev.langchain4j</groupId>
            <artifactId>langchain4j-mongodb-atlas</artifactId>
            <version>1.5.0-beta11</version>
        </dependency>

        <!-- Voyage AI for embeddings -->
        <dependency>
            <groupId>dev.langchain4j</groupId>
            <artifactId>langchain4j-voyage-ai</artifactId>
            <version>1.5.0-beta11</version>
        </dependency>

        <!-- MongoDB Java Driver -->
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>mongodb-driver-sync</artifactId>
            <version>5.5.1</version>
        </dependency>

        <!-- OpenCSV for parsing the IMDB dataset -->
        <dependency>
            <groupId>com.opencsv</groupId>
            <artifactId>opencsv</artifactId>
            <version>5.8</version>
        </dependency>

        <!-- Jackson for parsing JSON responses from Watchmode -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.17.2</version>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>dev.langchain4j</groupId>
                <artifactId>langchain4j-bom</artifactId>
                <version>1.5.0-beta11</version>
                <type>pom</type>
            </dependency>
        </dependencies>
    </dependencyManagement>
</project>

These dependencies give us everything we need: the LangChain4j agentic framework, MongoDB and OpenAI integrations, embedding support via Voyage AI, and utilities for parsing CSV and JSON data.

Connecting to MongoDB

Now, let's create the main application class that will handle connecting to MongoDB, loading our movie data, and setting up our agent.

Create a file at MovieAgentApp.java:

package com.mongodb.movieagent;

import com.mongodb.client.*;
import dev.langchain4j.model.voyageai.VoyageAiEmbeddingModel;
import dev.langchain4j.store.embedding.mongodb.IndexMapping;
import dev.langchain4j.store.embedding.mongodb.MongoDbEmbeddingStore;
import org.bson.Document;

import java.util.HashSet;

public class MovieAgentApp {

    public static final String databaseName = "movie_search";
    public static final String collectionName = "movies";
    public static final String indexName = "vector_index";

    public static void main(String[] args) throws InterruptedException {
        String embeddingApiKey = System.getenv("VOYAGE_AI_KEY");
        String mongodbUri = System.getenv("MONGODB_URI");
        String watchmodeKey = System.getenv("WATCHMODE_KEY");
        String openAiKey = System.getenv("OPENAI_KEY");

        MongoClient mongoClient = MongoClients.create(mongodbUri);

        VoyageAiEmbeddingModel embeddingModel = VoyageAiEmbeddingModel.builder()
                .apiKey(embeddingApiKey)
                .modelName("voyage-3")
                .build();
    }
}

This sets up our connection to MongoDB using the connection string from our environment variables. We're also instantiating the Voyage AI embedding model, which we'll use to convert movie plot descriptions into vector embeddings.

The voyage-3 model generates 1024-dimensional embeddings, which we'll need to specify when creating our vector search index. To learn more about Voyage AI's models, check out their blog post about voyage-3.

Now, we will be using the Voyage AI model for generating our embeddings, and the separate OpenAI model for planning and orchestrating our AI agent. If you prefer, you can use OpenAI for both the embedding and planning. Just swap out your VoyageAiEmbeddingModel for the OpenAiEmbeddingModel. This does not go both ways, as Voyage AI is not supported as a chat model in LangChain4j.

Next, we'll configure our embedding store and automatically create a vector search index if one doesn't already exist:

        IndexMapping indexMapping = IndexMapping.builder()
                .dimension(embeddingModel.dimension())
                .metadataFieldNames(new HashSet<>())
                .build();

        MongoDbEmbeddingStore embeddingStore = MongoDbEmbeddingStore.builder()
                .databaseName(databaseName)
                .collectionName(collectionName)
                .createIndex(checkIndexExists(mongoClient))
                .indexName(indexName)
                .indexMapping(indexMapping)
                .fromClient(mongoClient)
                .build();

        if(checkDataExists(mongoClient)) {
            loadDataFromCSV(embeddingStore, embeddingModel);
        }

Let's break down what's happening here:

The IndexMapping tells LangChain4j how to configure the vector search index. We're setting the dimension to match our embedding model (1024 for voyage-3) and leaving metadataFieldNames empty since we don't need to filter on metadata fields for this example.

The MongoDbEmbeddingStore builder does several things:

Points to our movie_search.movies collection
Checks if an index already exists using our helper method checkIndexExists()
If no index exists, automatically creates one with the name vector_index
Uses our IndexMapping to define the index structure

The resulting vector search index looks like this:

{
  "fields": [
    {
      "type": "vector",
      "path": "embedding",
      "numDimensions": 1024,
      "similarity": "cosine"
    }
  ]
}

Finally, we check if the collection already has data using checkDataExists(), and if it's empty, we load and embed the movie dataset from our CSV file.

Now, let's add those helper methods at the bottom of the class:

    public static void loadDataFromCSV(
            MongoDbEmbeddingStore embeddingStore,
            VoyageAiEmbeddingModel embeddingModel
    ) throws InterruptedException {
        System.out.println("Loading data...");

        MovieEmbeddingService embeddingService = new MovieEmbeddingService(embeddingStore, embeddingModel);
        embeddingService.ingestMoviesFromCsv();

        System.out.println("Movie data loaded successfully!");
        System.out.println("Waiting 5 seconds for indexing to complete...");
        Thread.sleep(5000);
    }

    public static boolean checkDataExists(MongoClient mongoClient) {
        MongoCollection<Document> collection = mongoClient
            .getDatabase(databaseName)
            .getCollection(collectionName);
        return collection.find().first() == null;
    }

    public static boolean checkIndexExists(MongoClient mongoClient) {
        MongoCollection<Document> collection = mongoClient
            .getDatabase(databaseName)
            .getCollection(collectionName);

        try(MongoCursor<Document> indexes = collection.listIndexes().iterator()) {
            while (indexes.hasNext()) {
                Document index = indexes.next();
                if (indexName.equals(index.getString(indexName))) {
                    return false;
                }
            }
        }
        return true;
    }
}

The checkIndexExists() method iterates through all indexes on the collection and returns false if it finds one named vector_index, telling LangChain4j to skip index creation. The checkDataExists() method simply checks if the collection is empty.

The five-second sleep after loading data gives MongoDB Atlas time to build the vector search index. Atlas indexing is eventually consistent, so this wait ensures our index is queryable before we start searching. In production, you'd want more robust index readiness checking, but for a tutorial, a brief sleep does the job.

Importing our data

Now, we need to actually load the IMDB dataset, parse it, and convert the plot descriptions into vector embeddings. We'll create two classes: Movie to represent each row in the CSV, and MovieEmbeddingService to handle the embedding and storage logic.

Creating the Movie model

Create Movie.java:

package com.mongodb.movieagent;  

import com.opencsv.bean.CsvBindByPosition;  

public class Movie {  

    @CsvBindByPosition(position = 0)  
    private String posterLink;  

    @CsvBindByPosition(position = 1)  
    private String title;  

    @CsvBindByPosition(position = 2)  
    private String year;  

    @CsvBindByPosition(position = 3)  
    private String certificate;  

    @CsvBindByPosition(position = 4)  
    private String runtime;  

    @CsvBindByPosition(position = 5)  
    private String genre;  

    @CsvBindByPosition(position = 6)  
    private String imdbRating;  

    @CsvBindByPosition(position = 7)  
    private String overview;  

    @CsvBindByPosition(position = 8)  
    private String metaScore;  

    @CsvBindByPosition(position = 9)  
    private String director;  

    @CsvBindByPosition(position = 10)  
    private String star1;  

    @CsvBindByPosition(position = 11)  
    private String star2;  

    @CsvBindByPosition(position = 12)  
    private String star3;  

    @CsvBindByPosition(position = 13)  
    private String star4;  

    @CsvBindByPosition(position = 14)  
    private String numberOfVotes;  

    @CsvBindByPosition(position = 15)  
    private String gross;  

    public Movie() {}  

    public String getPosterLink() { return posterLink; }  
    public void setPosterLink(String posterLink) { this.posterLink = posterLink; }  

    public String getTitle() { return title; }  
    public void setTitle(String title) { this.title = title; }  

    public String getYear() { return year; }  
    public void setYear(String year) { this.year = year; }  

    public String getCertificate() { return certificate; }  
    public void setCertificate(String certificate) { this.certificate = certificate; }  

    public String getRuntime() { return runtime; }  
    public void setRuntime(String runtime) { this.runtime = runtime; }  

    public String getGenre() { return genre; }  
    public void setGenre(String genre) { this.genre = genre; }  

    public String getImdbRating() { return imdbRating; }  
    public void setImdbRating(String imdbRating) { this.imdbRating = imdbRating; }  

    public String getOverview() { return overview; }  
    public void setOverview(String overview) { this.overview = overview; }  

    public String getMetaScore() { return metaScore; }  
    public void setMetaScore(String metaScore) { this.metaScore = metaScore; }  

    public String getDirector() { return director; }  
    public void setDirector(String director) { this.director = director; }  

    public String getStar1() { return star1; }  
    public void setStar1(String star1) { this.star1 = star1; }  

    public String getStar2() { return star2; }  
    public void setStar2(String star2) { this.star2 = star2; }  

    public String getStar3() { return star3; }  
    public void setStar3(String star3) { this.star3 = star3; }  

    public String getStar4() { return star4; }  
    public void setStar4(String star4) { this.star4 = star4; }  

    public String getNumberOfVotes() { return numberOfVotes; }  
    public void setNumberOfVotes(String numberOfVotes) { this.numberOfVotes = numberOfVotes; }  

    public String getGross() { return gross; }  
    public void setGross(String gross) { this.gross = gross; }  

    @Override  
    public String toString() {  
        return String.format("%s (%s) - %s - Rating: %s\nGenre: %s | Director: %s\n%s",  
                title, year, genre, imdbRating, genre, director, overview);  
    }  
}

The @CsvBindByPosition annotations from OpenCSV map each CSV column to a field in our Movie class. The most important field for our purposes is overview, which contains the plot description we'll embed for semantic search.

Creating embeddings for vector search

Now, let's create the service that reads the CSV, generates embeddings, and stores them in MongoDB. Create MovieEmbeddingService.java:

package com.mongodb.movieagent;

import com.opencsv.bean.CsvToBeanBuilder;
import dev.langchain4j.data.embedding.Embedding;
import dev.langchain4j.data.document.Metadata;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.store.embedding.EmbeddingStore;

import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.*;

public class MovieEmbeddingService {

    private final EmbeddingStore<TextSegment> embeddingStore;
    private final EmbeddingModel embeddingModel;

    public MovieEmbeddingService(EmbeddingStore<TextSegment> embeddingStore, 
                                  EmbeddingModel embeddingModel) {
        this.embeddingStore = embeddingStore;
        this.embeddingModel = embeddingModel;
    }

    public void ingestMoviesFromCsv() {
        try (InputStream inputStream = getClass()
                .getClassLoader()
                .getResourceAsStream("imdb_top_1000.csv")) {

            if (inputStream == null) {
                throw new RuntimeException("imdb_top_1000.csv not found");
            }

            List<Movie> movies = new CsvToBeanBuilder<Movie>(new InputStreamReader(inputStream))
                    .withType(Movie.class)
                    .build()
                    .parse();

            System.out.println("Processing " + movies.size() + " movies...");

            for (Movie movie : movies) {
                if (movie.getTitle() == null || movie.getOverview() == null) {
                    continue;
                }

                Metadata metadata = getMetadata(movie);
                TextSegment segment = TextSegment.from(movie.getOverview(), metadata);
                Embedding embedding = embeddingModel.embed(segment).content();

                embeddingStore.add(embedding, segment);
                System.out.println("Stored: " + movie.getTitle());
            }
        } catch (Exception e) {
            System.err.println("Error processing CSV: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static Metadata getMetadata(Movie movie) {
        Map<String, Object> metadataMap = new HashMap<>();
        metadataMap.put("title", movie.getTitle());
        metadataMap.put("year", movie.getYear());
        metadataMap.put("genre", movie.getGenre());
        metadataMap.put("director", movie.getDirector());
        metadataMap.put("imdbRating", movie.getImdbRating());
        metadataMap.put("star1", movie.getStar1());
        metadataMap.put("star2", movie.getStar2());

        return new Metadata(metadataMap);
    }
}

Let's walk through what's happening here:

CSV parsing: We use OpenCSV's CsvToBeanBuilder to read imdb_top_1000.csv from our resources directory and parse it into a list of Movie objects.
Embedding generation: For each movie, we:

Extract the overview (plot description).
Create a TextSegment containing the overview text and metadata.
Pass the segment to embeddingModel.embed()(our Voyage AI model) to generate a 1024-dimensional vector.
Store both the embedding and the segment (with its metadata) in MongoDB.

Metadata storage: We attach key movie information as metadata (title, year, genre, etc.). This gets stored alongside the embedding in MongoDB, so when we search, we get back not just vectors but all the movie details we need.

The beauty of this approach is that we're only embedding the overview field. When a user searches for "rebels fighting an empire," the semantic search matches that against movie plot descriptions, not titles or actor names. Users can describe what they remember about a movie without needing to know its exact name.

If you want to learn more about how to use MongoDB with LangChain4j, check out our documentation.

Creating our agent

Now comes the fun part: building the agentic system. We'll create three specialized tools that our supervisor agent can use, then wire them together.

Understanding the architecture

Our agent follows the supervisor pattern, where a planning model (powered by GPT-4o-mini) decides which tools to call and in what order. Here's the flow:

User query: "Find me a sci-fi movie about rebels fighting an empire and tell me where to stream it in GB"
Supervisor analyzes the query and generates a plan
Tool 1 (MongoDB search): Searches for movies matching the plot description
Tool 2 (Watchmode search): Gets the Watchmode ID for that movie
Tool 3 (Watchmode sources): Fetches streaming availability for that ID and region
Supervisor synthesizes all the results into a clean, human-readable response

The supervisor doesn't hardcode this workflow—it figures out the steps based on the context we provide it. This is the key difference between a traditional workflow and an agentic system: The LLM is doing the orchestration.

Note: We use a different model for embedding and for our planning model because different models have different strengths. We could easily use ChatGPT for both embedding and chat/planning, but Voyage AI is not supported as a chat model.

Defining tools

In LangChain4j's agentic module, a tool is just a Java method annotated with @Agent. The annotation tells the supervisor what the tool does, and the framework handles the rest—automatically converting method parameters to tool inputs and outputs to shared state.

Let's create our three tools.

Tool 1: MongoDB Movie Search

Create MongoDbMovieSearchTool.java:

package com.mongodb.movieagent;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.data.embedding.Embedding;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.service.V;
import dev.langchain4j.store.embedding.EmbeddingSearchRequest;
import dev.langchain4j.store.embedding.EmbeddingSearchResult;
import dev.langchain4j.store.embedding.mongodb.MongoDbEmbeddingStore;

public class MongoDbMovieSearchTool {

    private final MongoDbEmbeddingStore store;
    private final EmbeddingModel embeddingModel;

    public MongoDbMovieSearchTool(EmbeddingModel embeddingModel, 
                                   MongoDbEmbeddingStore store) {
        this.store = store;
        this.embeddingModel = embeddingModel;
    }

    @Agent(value = "Search movies in MongoDB by semantic description",
            outputName = "movieTitle")
    public String search(@V("query") String query) {
        try {
            Embedding queryEmbedding = embeddingModel.embed(query).content();

            EmbeddingSearchResult<TextSegment> result = store.search(
                    EmbeddingSearchRequest.builder()
                            .queryEmbedding(queryEmbedding)
                            .maxResults(1)
                            .build()
            );

            if (!result.matches().isEmpty()) {
                TextSegment doc = result.matches().getFirst().embedded();
                System.out.println(doc.toString());
                return doc.metadata().getString("title");
            }

            return "No matching movie found in MongoDB.";
        } catch (Exception e) {
            return "Error searching MongoDB: " + e.getMessage();
        }
    }
}

The @Agent annotation does two important things:

value: Describes what this tool does. The supervisor uses this description to decide when to call it.
outputName: Specifies that the result should be stored in a shared variable called "movieTitle", which other tools can access.

The @V("query") annotation on the parameter tells LangChain4j to look for a variable named "query" in the shared state (called the AgenticScope) and pass it to this method.

When the supervisor calls this tool with a plot description like "rebels fighting an empire," we:

Convert the query to an embedding.
Perform a vector search in MongoDB to find the most similar movie plot.
Return the movie title, which gets stored as "movieTitle" in the shared state.

Tool 2: Watchmode Search

Create WatchmodeSearchTool.java:

package com.mongodb.movieagent;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.V;

import java.net.URI;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;

public class WatchmodeSearchTool {

    private final HttpClient http = HttpClient.newHttpClient();
    private final String apiKey;

    public WatchmodeSearchTool(String apiKey) {
        this.apiKey = apiKey;
    }

    @Agent(value = "Find Watchmode ID for a given movie title", 
           outputName = "watchmodeId")
    public String getWatchmodeId(@V("title") String title) {
        try {
            String url = String.format(
                    "https://api.watchmode.com/v1/search/?apiKey=%s&search_field=name&search_value=%s&types=movie",
                    apiKey, URLEncoder.encode(title, StandardCharsets.UTF_8)
            );

            HttpRequest req = HttpRequest.newBuilder()
                .uri(URI.create(url))
                .GET()
                .build();
            HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString());

            JsonNode root = new ObjectMapper().readTree(resp.body());
            JsonNode results = root.get("title_results");

            if (results != null && !results.isEmpty()) {
                int id = results.get(0).get("id").asInt();
                System.out.println("Watchmode ID: " + id);
                return String.valueOf(id);
            }

            return "No Watchmode ID found for " + title;

        } catch (Exception e) {
            return "Error retrieving ID: " + e.getMessage();
        }
    }
}

This tool takes a movie title (like "Star Wars") and queries the Watchmode API to get its internal ID. That ID is then stored as "watchmodeId" in the shared state, ready for the next tool to use.

Watchmode's API documentation has more details on what is available with the API, but the key thing to know is that their /search/ endpoint returns a list of potential matches, and we're grabbing the first one.

Tool 3: Watchmode Sources

Create WatchmodeSourcesTool.java:

package com.mongodb.movieagent;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.V;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

public class WatchmodeSourcesTool {

    private final HttpClient http = HttpClient.newHttpClient();
    private final String apiKey;

    public WatchmodeSourcesTool(String apiKey) {
        this.apiKey = apiKey;
    }

    @Agent(value = "Get streaming availability for a Watchmode ID",
            outputName = "streamingSources")
    public String getSources(@V("watchmodeId") String id, 
                             @V("region") String region) {
        try {
            String url = String.format(
                    "https://api.watchmode.com/v1/title/%s/sources/?apiKey=%s&regions=%s",
                    id, apiKey, region
            );

            HttpRequest req = HttpRequest.newBuilder()
                .uri(URI.create(url))
                .GET()
                .build();
            HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString());

            return resp.body();

        } catch (Exception e) {
            return "Error retrieving sources: " + e.getMessage();
        }
    }
}

This tool takes the watchmodeId from the previous step and a region code (like "GB" for Great Britain) and fetches the streaming availability. It returns the raw JSON response, which the supervisor will interpret to give the user a clean answer.

Notice how this tool expects two inputs: watchmodeId (from the previous tool) and region (from the original user query). The supervisor figures out which variables to pass to each tool based on their parameter names.

Creating the Supervisor

Now, let's wire everything together. First, create a simple interface for the supervisor:

MovieSupervisor.java:

package com.mongodb.movieagent;

import dev.langchain4j.agentic.Agent;
import dev.langchain4j.service.V;

public interface MovieSupervisor {
    @Agent
    String invoke(@V("request") String request);
}

This interface defines the entry point for our agent. Users will call invoke() with a natural language query, and the supervisor will orchestrate the tools to produce an answer.

Back in our MovieAgentApp.java, let's instantiate the planning model and build the supervisor (add this after the embedding store setup):

ChatModel planningModel = OpenAiChatModel.builder()
        .apiKey(openAiKey)
        .modelName("gpt-4o-mini")
        .build();

MongoDbMovieSearchTool mongoSearch =
        new MongoDbMovieSearchTool(embeddingModel, embeddingStore);
WatchmodeSearchTool watchmodeSearch = new WatchmodeSearchTool(watchmodeKey);
WatchmodeSourcesTool watchmodeSources = new WatchmodeSourcesTool(watchmodeKey);

MovieSupervisor supervisor = AgenticServices
        .supervisorBuilder(MovieSupervisor.class)
        .subAgents(mongoSearch, watchmodeSearch, watchmodeSources)
        .supervisorContext("""
            You are a movie assistant.
            1. If the user gives a plot/description, call MongoDbMovieSearchTool to find the movie.
            2. Use WatchmodeSearchTool with the title to get a Watchmode ID.
            3. Use WatchmodeSourcesTool with that ID and the user's region to get streaming info.
            4. Return a clean human-readable response.
            """)
        .chatModel(planningModel)
        .responseStrategy(SupervisorResponseStrategy.SUMMARY)
        .build();

Let's break down what's happening in this supervisor configuration:

The planning model: We're using gpt-4o-mini as our planning model. This is the "brain" that decides which tools to call and when. It's cheaper and faster than GPT-4, but still capable enough to handle multi-step reasoning.

Sub-agents: We register our three tools with the supervisor. Each tool's @Agent annotation provides a description that the planning model uses to decide when to invoke it.

Supervisor context: This is where things get interesting. The context is essentially a set of instructions for the planning model, telling it how to orchestrate the tools. It's similar to a system prompt in a regular chat application, but here it's guiding planning rather than just responding.

Think of the supervisor context as a recipe: "If the user describes a plot, search MongoDB first. Then use that title to get a Watchmode ID. Finally, fetch streaming sources." The LLM follows these steps dynamically, adapting based on what it finds. We can see how this would allow the planner to orchestrate things if the user didn't want to get where everything streams (only do steps 1, 2, and 4) or already had the movie title (skip step 1).

Response strategy: We're using SupervisorResponseStrategy.SUMMARY, which tells the supervisor to return a synthesized summary of all the tool calls rather than just the last tool's raw output. This means instead of getting back raw JSON from Watchmode, we get a clean answer like "Star Wars is available on Disney+ in GB."

The alternative strategies are:

LAST: Returns the output of the last tool called (useful when the final tool produces the answer).
SCORED: Uses another LLM call to score both the summary and the last tool output, returning whichever is better.

For this use case, SUMMARY makes sense because we want the agent to interpret the streaming data and present it nicely to the user.

Testing our movie recommendations

Now, let's put it all together and test our agent. Add this to the end of your main() method in MovieAgentApp.java:

        String query = "Find me a sci-fi movie about rebels fighting an empire in space and tell me where to stream it in GB.";
        String result = supervisor.invoke(query);

        System.out.println("Agent Response:\n" + result);
    }
}

Run your application with:

MONGODB_URI=$MONGODB_URI \
VOYAGE_AI_KEY=$VOYAGE_AI_KEY \
WATCHMODE_KEY=$WATCHMODE_KEY \
OPENAI_KEY=$OPENAI_API_KEY \
mvn compile exec:java -Dexec.mainClass=com.mongodb.movieagent.MovieAgentApp

On first run, you'll see output like:

Loading data...
Processing 1000 movies...
Stored: The Shawshank Redemption
Stored: The Godfather
...
Movie data loaded successfully!
Waiting 5 seconds for indexing to complete...

Then the agent will process your query. You'll see trace output showing which tools are being called:

Watchmode ID: 12345
Agent Response:
Star Wars: Episode IV - A New Hope is available to stream on Disney+ in Great Britain. 
You can also rent it on Amazon Prime Video for £3.49.

What just happened?

The supervisor analyzed your query and identified that you were describing a plot.
It called MongoDbMovieSearchTool, embedding "rebels fighting an empire in space" and searching MongoDB.
MongoDB returned "Star Wars: Episode IV - A New Hope" as the best match.
The supervisor then called WatchmodeSearchTool with that title to get the Watchmode ID.
Finally, it called WatchmodeSourcesTool with the ID and region "GB."
The planning model synthesized all three tool outputs into a clean, conversational response.

Try other queries to see how the agent handles different scenarios:

// Search by genre and mood
String query = "I want a dark crime thriller about an undercover cop, where can I watch it in the US?";

// Search by vague description
String query = "That movie with the kid who sees dead people, streaming options in US?";

// Mix of specifics
String query = "Christopher Nolan movie about dreams within dreams, available in GB?";

The beauty of this approach is that you never hardcoded the workflow. The supervisor figures out the steps based on your context instructions. If you wanted to add a fourth tool (say, fetching movie reviews), you'd just add it to the subAgents() list and update the supervisor context—no need to rewrite orchestration logic.

Understanding the AgenticScope

Behind the scenes, LangChain4j manages something called the AgenticScope, which is essentially a shared key-value store for the conversation. When MongoDbMovieSearchTool returns a title, it's stored as "movieTitle" in this scope. When WatchmodeSearchTool needs a title, it reads from "movieTitle". The supervisor handles all this variable passing automatically based on the @V annotations you defined on each tool's parameters.

If you wanted to inspect or debug what's in the scope at any point, you could implement the AgenticScopeAccess interface on your supervisor, but for this tutorial, the automatic variable passing is all you need.

Conclusion

We've built a genuine AI agent that can understand natural language movie queries, search semantically through plot descriptions, and fetch real-time streaming availability—all without hardcoding a single workflow step.

By combining MongoDB Atlas for vector search with LangChain4j's agentic framework, we created a system that:

Understands vague descriptions like "rebels fighting an empire" and maps them to actual movies.
Autonomously decides which tools to call and in what order.
Handles multi-step reasoning (search → get ID → fetch sources).
Returns clean, conversational responses instead of raw API data.

The supervisor pattern we used here is just one of many agentic architectures. You could extend this with additional tools (fetching reviews, checking rental prices, filtering by age rating), add memory so the agent remembers past conversations, or even create tool-specific agents that the supervisor coordinates.

If you're curious about building more complex agentic systems, check out the LangChain4j agentic documentation and explore patterns like loops, conditionals, and human-in-the-loop workflows.

For more MongoDB and AI tutorials, see my other articles on Building RAG Applications with Spring AI and MongoDB or dive into the MongoDB Atlas Vector Search documentation.

Secure Your Spring API With JWT and MongoDB

Tim Kelly — Thu, 18 Sep 2025 16:29:05 +0000

APIs don’t need to remember who you are if you bring your proof every time. That’s the idea behind JSON Web Token (JWT)—a compact, signed token that carries just enough claims about the user to authorize a request. The server signs it, the client sends it in the Authorization header, and any service with the public key can verify it. It fits SPAs, mobile apps, and microservices, scales cleanly, and makes rotation and revocation straightforward once you add refresh tokens or an auth server.

We’re going to build a small, secure API with Spring Security and store user data in MongoDB. Spring Security already knows how to handle JWTs via the OAuth2 Resource Server support, so we’ll lean on that instead of writing custom filters.

The flow is simple: Register a user, log in to get a signed JWT, and use that token to call protected endpoints. This is a starter, enough to get a real REST API locked down with stateless auth, and you can layer on refresh tokens, key rotation, and an external authorization server later. The code is all available on GitHub.

Note: While we are working on security at the application level, you can learn more about security at the database level at MongoDB University.

Prerequisites

A MongoDB Atlas account with a cluster set up
- A free M0 cluster is perfect for this
Java 17+
Maven (I use version 3.9.10)

Dependencies (Spring Initializr)

Our first step is creating the application. The easiest way is with Spring Initializr: Pick Maven, Java 21, Jar.

Add these starters:

Spring Web
Spring Data MongoDB
OAuth2 Resource Server (this brings in Spring Security and the Nimbus JWT pieces)
Validation

Our pom.xml will include:

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
  </dependency>

  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb</artifactId>
  </dependency>

  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-oauth2-resource-server</artifactId>
  </dependency>

  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-validation</artifactId>
  </dependency>
</dependencies>

That’s all we need: Web for the controllers, MongoDB for persistence, the resource server starter for JWT verification and the security stack, and Validation for request DTOs.

Configuring MongoDB to store user information

Download the project from Initializr and open it in your IDE. Start by pointing Spring at MongoDB. In application.properties, use your MongoDB Atlas connection string and name the database users:

spring.data.mongodb.uri=${MONGODB_URI}  
spring.data.mongodb.database=users

spring.data.mongodb.auto-index-creation=true

We're also telling MongoDB that we can create indexes in the app if we use them and they do not yet exist. We will annotate these using the @index notation later. This will be useful when we want to create a unique index for our usernames to prevent duplicates.

We’re going to store user records in MongoDB. Create a model package and add a UserAccount document that represents what we’ll persist: a unique username, a BCrypt-hashed password, and a set of roles. We don't use the roles in this application, but this can be built upon to make some elements of the API only available to some user groups.

package com.mongodb.springjwt.model;  

import org.springframework.data.annotation.Id;  
import org.springframework.data.mongodb.core.index.Indexed;  
import org.springframework.data.mongodb.core.mapping.Document;  

import java.util.Set;  

@Document("users")  
public class UserAccount {  
    @Id private String id;  

    @Indexed(unique = true)  
    private String username;  

    private String password;        // BCrypt-hashed  
    private Set<String> roles;      // e.g. ["USER", "ADMIN"]  

    public String getId() {  
        return id;  
    }  

    public void setId(String id) {  
        this.id = id;  
    }  

    public String getUsername() {  
        return username;  
    }  

    public void setUsername(String username) {  
        this.username = username;  
    }  

    public String getPassword() {  
        return password;  
    }  

    public void setPassword(String password) {  
        this.password = password;  
    }  

    public Set<String> getRoles() {  
        return roles;  
    }  

    public void setRoles(Set<String> roles) {  
        this.roles = roles;  
    }  
}

The @Document("users") annotation tells Spring Data MongoDB to map this class to the users collection and the @Indexed(unique = true) enforces unique usernames at the database levels.

Next up is a repository so the app can talk to the database. Create a repository package and add UserAccountRepository. Spring Data will generate the queries for us based on method names, so we can look users up by username and quickly check if one already exists.

package com.mongodb.springjwt.repository;  

import com.mongodb.springjwt.model.UserAccount;  
import org.springframework.data.mongodb.repository.MongoRepository;  

import java.util.Optional;  

public interface UserAccountRepository extends MongoRepository<UserAccount, String> {  
    Optional<UserAccount> findByUsername(String username);  
    boolean existsByUsername(String username);  
}

Now, we need to teach Spring Security how to load users from MongoDB. We do that by implementing UserDetailsService and adapting my UserAccount document into Spring Security’s User type.

package com.mongodb.springjwt.service;  

import com.mongodb.springjwt.repository.UserAccountRepository;  
import org.springframework.security.core.userdetails.*;  
import org.springframework.stereotype.Service;  

@Service  
public class MongoUserDetailsService implements UserDetailsService {  

    private final UserAccountRepository repo;  

    public MongoUserDetailsService(UserAccountRepository repo) {  
        this.repo = repo;  
    }  

    @Override  
    public UserDetails loadUserByUsername(String username) throws UsernameNotFoundException {  
        var user = repo.findByUsername(username)  
                .orElseThrow(() -> new UsernameNotFoundException("User not found"));  

        return User.withUsername(user.getUsername())  
                .password(user.getPassword())  
                .roles(user.getRoles().toArray(String[]::new))  
                .build();  
    }  
}

On login, Spring calls loadUserByUsername, we fetch the record, and if it exists, we hand back a User with the stored BCrypt hash for the password and the roles assigned to that user. If not, we throw UsernameNotFoundException. The DaoAuthenticationProvider takes it from there and compares the raw password to the hash using the PasswordEncoder.

One thing to note: .roles(...) expects bare role names like USER and adds the ROLE_ prefix for us, so Set.of("USER","ADMIN") in the database becomes authorities ROLE_USER and ROLE_ADMIN at runtime.

Our secure API

We're adding a tiny controller with one endpoint to prove the security is actually doing something. It lives under /api/secure and returns a greeting to whoever is authenticated. The Principal comes from the SecurityContext that Spring builds after a valid JWT is verified.

Create a package called controller and a new class SecureController:

package com.mongodb.springjwt.controller;  

import org.springframework.web.bind.annotation.GetMapping;  
import org.springframework.web.bind.annotation.RequestMapping;  
import org.springframework.web.bind.annotation.RestController;  

import java.security.Principal;  

@RestController  
@RequestMapping("/api/secure")  
public class SecureController {  
    @GetMapping("/hello")  
    public String hello(Principal principal) {  
        return "Hello, %s".formatted(principal.getName());  
    }  
}

If we hit /api/secure/hello without a token, we’ll get a 401 Unauthorized because the resource server is in play and everything is locked down by default. When we have everything in place, we'll be able to log in to get a JWT, send it in Authorization: Bearer <token>, and the same call responds with Hello, <username>.

Security configuration

Here’s how we will wire security in this project. Let’s create a config package with a single SecurityConfig that defines the filter chain, the JWT components, and the authentication manager we'll use for the JSON login flow.

package com.mongodb.springjwt.config;  

import com.nimbusds.jose.jwk.*;  
import com.nimbusds.jose.jwk.source.*;  
import com.nimbusds.jose.proc.SecurityContext;  
import org.springframework.context.annotation.*;  
import org.springframework.security.authentication.AuthenticationManager;  
import org.springframework.security.config.Customizer;  
import org.springframework.security.config.annotation.authentication.configuration.AuthenticationConfiguration;  
import org.springframework.security.config.annotation.web.builders.HttpSecurity;  
import org.springframework.security.config.annotation.web.configurers.AbstractHttpConfigurer;  
import org.springframework.security.config.http.SessionCreationPolicy;  
import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;  
import org.springframework.security.crypto.password.PasswordEncoder;  
import org.springframework.security.oauth2.jwt.*;  
import org.springframework.security.web.SecurityFilterChain;  

import java.security.interfaces.RSAPrivateKey;  
import java.security.interfaces.RSAPublicKey;  

@Configuration  
public class SecurityConfig {  

    @Bean  
    public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {  
        return http  
                .csrf(AbstractHttpConfigurer::disable)  
                .authorizeHttpRequests(auth -> auth  
                        .requestMatchers("/auth/token", "/auth/register").permitAll()  
                        .anyRequest().authenticated())  
                .oauth2ResourceServer(oauth2 -> oauth2.jwt(Customizer.withDefaults()))  
                .sessionManagement(sm -> sm.sessionCreationPolicy(SessionCreationPolicy.STATELESS))  
                .build();  
    } 

        @Bean  
    AuthenticationManager authenticationManager(AuthenticationConfiguration cfg) throws Exception {  
        return cfg.getAuthenticationManager();  
    }  
}

We're disabling CSRF in this tutorial because we've made this service stateless and use Bearer tokens, not cookies. If we were using server sessions, disabling CSRF would be a mistake; with stateless JWTs, it’s the right call.

The authorization rules are simple on purpose. Everything is locked down by default, with only /auth/register and /auth/token open so a client can create an account and exchange credentials for a token. The session policy is stateless to make every request self-contained. No server memory of prior requests, no hidden state.

The oauth2ResourceServer piece is what actually validates tokens on protected requests. That one line adds the bearer token filter and hands it a JwtDecoder, which we will implement soon. On each call, it pulls the token from the header, checks the signature and expiry, and only then lets the controller run. We're using self-signed JWTs here: a private key to issue them at the login endpoint, the public key to verify them on every other request. It’s simple and elegant for a single service or a starter project.

We also expose an AuthenticationManager so the /auth/token endpoint can accept a JSON body and call authenticate(...) to verify the username and password against MongoDB before minting a JWT. I chose a JSON contract for login because it’s easier to evolve, adding captcha or device ID info or whatever we need.

Storing passwords securely

We're are going to add another bean to hash our passwords:

    @Bean PasswordEncoder passwordEncoder() {  
        return new BCryptPasswordEncoder();  
    }

We need a PasswordEncoder (using BCryptPasswordEncoder) because we want our user passwords to live in MongoDB as hashes, not plaintext.

If we grow our application to include refresh tokens or multiple services, we would want to split responsibilities. Put login, refresh, and key rotation in a dedicated authorization server and expose a JSON Web Key (JWK) endpoint; point our resource servers at that and keep them focused on validating tokens and serving data. Having one issuer we could reuse across different applications would mean a smaller attack surface.

We need a JWT decoder for our resource server, but before we do that, we need some RSA keys.

Creating RSA keys

We can sign JWTs with a shared secret (HS256) or with a keypair (RS256). We're using asymmetric keys so the app signs with a private key and verifies with the public key, which plays nicer with multiple services and future key rotation.

Create src/main/resources/certs, then generate keys with OpenSSL by running the following commands in that new folder location:

openssl genrsa -out keypair.pem 2048

This will generate with an encryption level of 2048, which is fine for this demo. Do bear in mind that this is not enough for long-term security, and RSA 2028 is marked for depreciation by NIST by 2030.

This will generate the pair of keys that we can now extract the public key from, and place in a public.pem file by running the following command in the same folder:

openssl rsa -in keypair.pem -pubout -out public.pem

If we ran our code from here, we would get an error that the private key is in the wrong format because openssl genrsa writes a legacy PKCS#1 private key, but Java/Spring/Nimbus expect a PKCS#8 key.

To extract the private key in the format we need, we run:

openssl pkcs8 -topk8 -inform PEM -outform PEM -nocrypt -in keypair.pem -out private.pem

We should now have public.pem and private.pem, and we can safely delete the keypair.pem file.

└── src/ 
   └── main/
     └── resources/certs/
         keypair.pem
         private.pem 
         public.pem

We need a way to access these keys in our application. Let's configure our application so we can bind them into the app, keep the files local in dev, and move the locations to environment variables in prod.

Create a class RsaKeyProperties in config package:

package com.mongodb.springjwt.config;  

import org.springframework.boot.context.properties.ConfigurationProperties;  

import java.security.interfaces.RSAPrivateKey;  
import java.security.interfaces.RSAPublicKey;  

@ConfigurationProperties(prefix = "rsa")  
public record RsaKeyProperties(RSAPublicKey publicKey, RSAPrivateKey privateKey) {}

In our main SpringJwtApplication class, we need to enable this by adding @EnableConfigurationProperties(RsaKeyProperties.class):

package com.mongodb.springjwt;  

import com.mongodb.springjwt.config.RsaKeyProperties;  
import org.springframework.boot.SpringApplication;  
import org.springframework.boot.autoconfigure.SpringBootApplication;  
import org.springframework.boot.context.properties.EnableConfigurationProperties;  

@SpringBootApplication  
@EnableConfigurationProperties(RsaKeyProperties.class)  
public class SpringJwtApplication {  

    public static void main(String[] args) {  
        SpringApplication.run(SpringJwtApplication.class, args);  
    }  

}

And in application properties:

rsa.private-key=classpath:certs/private.pem  
rsa.public-key=classpath:certs/public.pem

Now that we can access our RSA keys in our app, we can create our JWT encoders and decoders.

Encoding and decoding JWT

Back in SecurityConfig, at the top of the class can, pull the keys in via the constructor:

    private final RSAPublicKey publicKey;  
    private final RSAPrivateKey privateKey;  

    public SecurityConfig(RsaKeyProperties keys) {  
        this.publicKey = keys.publicKey();  
        this.privateKey = keys.privateKey();  
    }

Then, define the JWT encoder and decoder:

    @Bean  
    JwtDecoder jwtDecoder() {  
        return NimbusJwtDecoder.withPublicKey(publicKey).build();  
    }  

    @Bean  
    JwtEncoder jwtEncoder() {  
        JWK jwk = new RSAKey.Builder(publicKey).privateKey(privateKey).build();  
        JWKSource<SecurityContext> jwks = new ImmutableJWKSet<>(new JWKSet(jwk));  
        return new NimbusJwtEncoder(jwks);  
    }

We'll be using the NimbusJwtDecoder and NimbusJwtEncoder, that we have from our OAuth resource server. NimbusJwtDecoder verifies incoming tokens with the public key; NimbusJwtEncoder signs new tokens with the private key. When a user logs in successfully, we’ll return a token produced by this encoder.

Our next step is a small service that builds the claims and calls the encoder to mint that JWT.

Creating tokens

I keep token creation in a small service so the controller doesn’t worry about JWT details. Create a package service and create a TokenService class:

package com.mongodb.springjwt.service;  

import org.springframework.security.core.Authentication;  
import org.springframework.security.core.GrantedAuthority;  
import org.springframework.security.oauth2.jwt.*;  
import org.springframework.stereotype.Service;  

import java.time.Instant;  
import java.time.temporal.ChronoUnit;  
import java.util.stream.Collectors;  

@Service  
public class TokenService {  
    private final JwtEncoder encoder;  

    public TokenService(JwtEncoder encoder) { 
        this.encoder = encoder; 
    }  

    public String generate(Authentication auth) {  
        Instant now = Instant.now();  
        String scope = auth.getAuthorities().stream()  
                .map(GrantedAuthority::getAuthority)  
                .collect(Collectors.joining(" "));  

        var claims = JwtClaimsSet.builder()  
                .issuer("self")  
                .issuedAt(now)  
                .expiresAt(now.plus(1, ChronoUnit.HOURS)) // short TTL is good  
                .subject(auth.getName())  
                .claim("scope", scope)  
                .build();  

        return encoder.encode(JwtEncoderParameters.from(claims)).getTokenValue();  
    }  
}

The Authentication that reaches this method represents a successfully logged-in user. We take their authorities and flatten them into a single space-delimited scope string, then build a claim set that says who issued the token (self-assigned), when it was issued, when it expires (one hour), who it’s about (subject is our Principal.getName), and what scopes it carries.

The encoder signs that claim set with the private key and hands us back a compact JWT string, which is what we return to the caller. A one-hour expiry keeps access tokens short-lived and easy to rotate later.

Authorization controller

Time to add an auth API so clients can register and exchange credentials for a token. We keep this in a dedicated new class AuthController in the controller package under /auth. It accepts JSON, checks credentials with the AuthenticationManager, and mints a JWT with the TokenService:

package com.mongodb.springjwt.controller;  

import com.mongodb.springjwt.model.UserAccount;  
import com.mongodb.springjwt.repository.UserAccountRepository;  
import com.mongodb.springjwt.service.TokenService;  
import jakarta.validation.constraints.NotBlank;  
import org.springframework.http.ResponseEntity;  
import org.springframework.security.authentication.*;  
import org.springframework.security.crypto.password.PasswordEncoder;  
import org.springframework.web.bind.annotation.*;  

import java.util.Set;  

@RestController  
@RequestMapping("/auth")  
public class AuthController {  

    private final UserAccountRepository users;  
    private final PasswordEncoder encoder;  
    private final AuthenticationManager authManager;  
    private final TokenService tokens;  

    public AuthController(UserAccountRepository users, PasswordEncoder encoder,  
                          AuthenticationManager authManager, TokenService tokens) {  
        this.users = users;  
        this.encoder = encoder;  
        this.authManager = authManager;  
        this.tokens = tokens;  
    }  

    public record RegisterRequest(@NotBlank String username, @NotBlank String password) {}  
    public record LoginRequest(@NotBlank String username, @NotBlank String password) {}  
    public record JwtResponse(String token) {}  

    @PostMapping("/register")  
    public ResponseEntity<?> register(@RequestBody @Valid RegisterRequest req) {  
        if (users.existsByUsername(req.username())) return ResponseEntity.badRequest().body("Username taken");  
        var u = new UserAccount();  
        u.setUsername(req.username());  
        u.setPassword(encoder.encode(req.password()));  
        u.setRoles(Set.of("USER"));  
        users.save(u);  
        return ResponseEntity.ok("Registered");  
    }  

    @PostMapping("/token")  
    public ResponseEntity<JwtResponse> token(@RequestBody @Valid LoginRequest req) {  
        var auth = authManager.authenticate(new UsernamePasswordAuthenticationToken(req.username(), req.password()));  
        return ResponseEntity.ok(new JwtResponse(tokens.generate(auth)));  
    }  
}

The request records keep things tight and make sure username and password aren’t empty. Registration hashes the password with BCrypt and seeds a basic USER role before saving to MongoDB.

The token endpoint runs the credential check through Spring Security; on success, it returns a signed JWT. Hit a secured route without that token and you’ll get a 401; include Authorization: Bearer <token> and the same call goes through.

It would be worth setting up an @RestControllerAdvice to shape 400/401/403 responses, which makes DX nicer than plain strings, but we'll leave that out of this tutorial for simplicity.

Testing with curl

Start the app and use these calls from a terminal. I’m keeping them simple and in the exact order you’ll need.

curl -i -X POST http://localhost:8080/auth/register \
  -H "Content-Type: application/json" \
  -d '{"username":"tim","password":"secret"}'

And we can see it in our database:

TOKEN=$(curl -s -X POST http://localhost:8080/auth/token \
  -H "Content-Type: application/json" \
  -d '{"username":"tim","password":"secret"}' | sed -E 's/.*"token":"([^"]+)".*/\1/')
echo "$TOKEN"

Call the secured endpoint with the token:

curl -i http://localhost:8080/api/secure/hello \
  -H "Authorization: Bearer $TOKEN"

Try the same endpoint without a token to see the 401:

curl -i http://localhost:8080/api/secure/hello

If you want to sanity-check bad creds, this returns 401:

curl -i -X POST http://localhost:8080/auth/token \
  -H "Content-Type: application/json" \
  -d '{"username":"tim","password":"wrong"}'

Conclusion

We built a small but complete secure API: users in MongoDB, passwords hashed with BCrypt, a JSON login that mints short-lived RSA-signed JWTs, and protected routes enforced by Spring Security’s OAuth2 Resource Server. MongoDB fits well here because a user is just a simple document (username, hash, roles), unique indexes prevent duplicates, and you can easily extend the schema or add TTL-based refresh tokens later.

Spring Security did the heavy lifting: AuthenticationManager + your Mongo-backed UserDetailsService verified credentials, PasswordEncoder handled hashing, and the bearer token filter validated each JWT and built a SecurityContext per request. It’s all stateless (no sessions to juggle) and ready for you to grow with issuer/audience checks, JWKS + rotation, and refresh tokens when you need them.

Quantize Your Vectors, Speed Up Your Java AI Applications

Tim Kelly — Mon, 08 Sep 2025 14:38:13 +0000

Vector quantization is the process of shrinking full fidelity vectors into fewer bits. It reduces the memory required to store each vector by storing the reduced representation of our vector instead. This reduces our resource consumption and allows for a more efficient application, at the cost of recall. MongoDB recommends quantization for applications with large numbers of vectors, i.e., over 100,000.

With MongoDB Atlas Vector Search, we can automatically quantize our float vector embeddings. It also supports ingesting and indexing pre-quantized scalar and binary vectors from certain embedding models.

First, we'll take a look into how vector quantization works, and then we'll implement it using the MongoDB Java sync driver. If you want to check out the code for this tutorial, it is all available in this GitHub repository.

Scalar quantization

Scalar quantization involves first identifying the minimum and maximum values for each dimension of the indexed vectors to establish a range of values for these dimensions. The range is then divided into equally sized bins. Each float is mapped into a bin to convert our previous float values into discrete integers. In MongoDB Atlas Vector Search, this scalar quantization reduces the vector embedding's RAM cost to almost 25% (1/3.75) of the pre-quantization cost.

Binary quantization

When an embedding is normalized to length 1, such as with OpenAI's text-embedding-3-large, we can use binary quantization. Binary quantization involves assuming a midpoint of 0 for each dimension in the embedding. For each value in the vector, we then assign a binary value of 1 if the value is greater than the midpoint or 0 if the value is less than or equal to the midpoint.

In MongoDB Atlas Vector Search, this binary quantization reduces the vector embeddings RAM usage to just over 4% (1/24) of the pre-quantization cost. The reason it's not 1/32 is because the data structure containing the Hierarchical Navigable Small Worlds graph itself, separate from the vector values, isn't compressed.

When you run a query, Atlas Vector Search first converts the query’s float values into a binary vector, using the same midpoint as the index to enable efficient comparisons with stored binary vectors. After this initial match, it rescores the candidates by re-checking them against the original float values from the binary index to refine the ranking. The full fidelity vectors are kept in a separate on-disk data structure and are accessed only during rescoring when binary quantization is enabled, or when you perform exact search against binary or scalar quantized vectors.

Requirements

In order to automatically quantize vectors, or ingest quantized vectors, there are some requirements that need to be met. The following table from the MongoDB vector quantization docs lays out these requirements:

Requirement	For `int1` Ingestion	For `int8` Ingestion	For Automatic Scalar Quantization	For Automatic Binary Quantization
Requires index definition settings	No	No	Yes	Yes
Requires BSON `binData` format	Yes	Yes	No	No
Storage on mongod	`binData(int1)`	`binData(int8)`	`binData(float32)` `array(double)\`	`binData(float32)` `array(double)\`
Supported similarity method	`euclidean`	`cosine` `euclidean\` `dotProduct\`	`cosine` `euclidean\` `dotProduct\`	`cosine` `euclidean\` `dotProduct\`
Supported number of dimensions	Multiple of 8	1 to 8192	1 to 8192	1 to 8192
Supports ANN and ENN search	Yes	Yes	Yes	Yes

Note: MongoDB Atlas stores all floating-point values as the double data type internally. Therefore, both 32-bit and 64-bit embeddings are compatible with automatic quantization without conversion.

Our Java application

For this tutorial, we'll walk through setting up our Java application and MongoDB database to embed some sample data, and run a vector search query on it. We'll take two main approaches: automatic quantization of our vectors, and ingesting pre-quantized vectors.

Prerequisites

A MongoDB Atlas account with a cluster set up
- A free M0 cluster is perfect for this
Java on your machine, set up and ready to go (I use Java version 21)
Maven (I use version 3.9.10)
A Voyage AI API token
- Other embedding models work—just make sure it supports vector quantization. Check out a list of supported models in our documentation.

Before we dive into the implementation, start by creating a new Maven application. Once the project is set up, open the application.properties file and add the necessary configuration values for connection with MongoDB and Voyage AI, or add them as environment variables.

VOYAGE_API_KEY=YOUR_VOYAGE_AI_API_KEY
MONGODB_URI=YOUR_MONGODB_CONNECTION_STRING

We'll need to bring in our dependencies to our pom.xml file:

<dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.2</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.19.2</version>
        </dependency>
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>mongodb-driver-sync</artifactId>
            <version>5.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>2.0.16</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>2.0.16</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.json</groupId>
            <artifactId>json</artifactId>
            <version>20250517</version>
        </dependency>
        <dependency>
            <groupId>com.squareup.okhttp3</groupId>
            <artifactId>okhttp</artifactId>
            <version>4.12.0</version>
        </dependency>
    </dependencies>

Let’s quickly break down what each of these dependencies is doing in our project:

JUnit: Provides a simple testing framework so we can write unit tests for our application. While not strictly required for this tutorial, it’s good practice to keep tests in place as you iterate.
Jackson Databind: Used to map JSON responses (like the embeddings we get back from Voyage AI) into Java objects. It also makes it easy to serialize Java objects into JSON when needed.
MongoDB Driver Sync: The official synchronous Java Driver for MongoDB. This is how our Java application will connect to Atlas, insert documents, and run vector search queries.
SLF4J API and SLF4J Simple: SLF4J is a logging facade that gives us a consistent API for logging. The “Simple” binding is a lightweight implementation that prints logs to the console. Together, these help us see what’s going on under the hood when we run our app.
JSON: A lightweight library for constructing and manipulating JSON. We’ll use this when parsing small JSON objects or building payloads, especially when interacting with the embedding API.
OkHttp: A modern, efficient HTTP client for Java. We’ll use this to make requests to the Voyage AI API to fetch embeddings for our sample data.

Setting up our main class

For the sake of simplicity, we'll keep the logic for this application in the one Main class. This Main class sets up everything the app needs before we write any logic:

Environment-driven config: VOYAGE_API_KEY and MONGODB_URI are read from environment variables so you don’t hardcode secrets. If either is missing, the app fails fast with a clear error.
Project constants: Database/collection/index names (test, demo, vector_index) are what we'll use to connect to our MongoDB database and configure our vector search index.
HTTP details: VOYAGE_API_URL points at Voyage AI’s embeddings endpoint. The connection/read timeouts keep our HTTP calls predictable.
Toy dataset + query: DATA is a small list of documents we’ll embed and store; QUERY_TEXT is what we’ll use to run a vector search later. I was not too creative here, but you can absolutely plug in your own data, and even make the query text interactive if you want to play around with it.

public class Main {

    // Configurations
    private static final String VOYAGE_API_KEY = System.getenv("VOYAGE_API_KEY");
    private static final String MONGODB_URI = System.getenv("MONGODB_URI");
    private static final String DB_NAME = "test";
    private static final String COLLECTION_NAME = "demo";
    private static final String INDEX_NAME = "vector_index";
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    // Voyage AI API Endpoint
    private static final String VOYAGE_API_URL = "https://api.voyageai.com/v1/embeddings";

    // Timeout values for API requests
    private static final int CONNECTION_TIMEOUT = 30;
    private static final int READ_TIMEOUT = 60;

    // Sample Data
    private static final List<String> DATA = List.of(
            "The Great Wall of China is visible from space.",
            "The Eiffel Tower was completed in Paris in 1889.",
            "Mount Everest is the highest peak on Earth at 8,848m.",
            "Shakespeare wrote 37 plays and 154 sonnets during his lifetime.",
            "The Mona Lisa was painted by Leonardo da Vinci."
    );

    private static final String QUERY_TEXT = "Country landmarks";

    public static void main(String[] args) {
        if (VOYAGE_API_KEY == null || VOYAGE_API_KEY.isEmpty()) {
            throw new RuntimeException("API key not found. Set VOYAGE_API_KEY in your environment.");
        }
        if (MONGODB_URI == null || MONGODB_URI.isEmpty()) {
            throw new RuntimeException("MongoDB URI not found. Set MONGODB_URI in your environment.");
        }

        // we'll be adding more code here
    }

    // and here
}

We’ll add the HTTP call to Voyage, the MongoDB write, and the vector search inside main where the comments indicate.

Response records

We are going to create two records to handle our responses from the Voyage AI API. The first record is going to be called ResponseDouble.

When we use automatic quantization in MongoDB Atlas, we send and store float vectors (received as doubles in Java). This record mirrors the Voyage AI JSON shape for float embeddings:

data[i].embedding is List<Double>, which maps cleanly to MongoDB’s array<double> (what Atlas expects when it will auto-quantize for us).
This path keeps the original full-fidelity vectors available for rescoring while the index stores the quantized form in RAM.

package com.timkelly;  

import java.util.List;  

public record ResponseDouble (  
    String object,  
    List<DataItem> data,  
    String model,  
    Usage usage  
) {  
        public record DataItem(  
                String object,  
                List<Double> embedding, // stored as doubles in MongoDB  
                int index  
        ) {}  
        public record Usage(  
                int total_tokens  
        ) {}  
}

We'll use this model whenever we plan to enable quantization.enabled: true with "type": "scalar" or "binary" in our MongoDB Atlas Search index definition. We'll go more into depth with this later in the tutorial.

For pre-quantized ingestion, some embedding providers return reduced-precision vectors directly:

int8 (scalar-quantized) → a byte[] per vector
int1 (binary-quantized/packed bits) → also a byte[], but bit-packed

We are going to set up a ResponseBytes record for this:

package com.timkelly;  

import java.util.List;  

public record ResponseBytes(  
        String object,  
        List<DataItem> data,  
        String model,  
        Usage usage  
) {  
    public record DataItem(  
            String object,  
            byte[] embedding, // int8 values or packed bits  
            int index  
    ) {}  
    public record Usage(  
            int total_tokens  
    ) {}  
}

This record captures that response shape. When we insert into MongoDB, we’ll store the bytes as BSON binData so Atlas can index them as quantized vectors without extra conversion. One of the reasons we are using Voyage AI here is because we can get these pre-quantized values returned. If you are using a different model, your responses will look different, but there are numerous other models out there that can pre-quantize our vectors.

How to enable automatic quantization of vectors

We can configure MongoDB Atlas Vector Search to automatically quantize float vector embeddings in our collection to reduced representation types, such as int8 (scalar) and binary in our vector indexes.

{
  "fields":[
    {
      "type": "vector",
      "path": "<field-to-index>",
      "numDimensions": <number-of-dimensions>,
      "similarity": "euclidean | cosine | dotProduct",
      "quantization": "none | scalar | binary",
      "hnswOptions": {
        "maxEdges": <number-of-connected-neighbors>,
        "numEdgeCandidates": <number-of-nearest-neighbors>
      }
    },
    {
      "type": "filter",
      "path": "<field-to-index>"
    },
    ...
  ]
}

To set or change the quantization type, we specify a quantization field value of either scalar or binary in our index definition. This triggers an index rebuild similar to any other index definition change. The specified quantization type applies to all indexed vectors and query vectors at query-time. We don't need to change our query as our query vectors are automatically quantized. We'll create our index in the code using the MongoDB Java Driver, but first, we'll create a method to get our embeddings for our data.

Embedding our data

Underneath our main method, we'll create a method embedDataAndCreateDocument that will take in a list of strings (our data to embed) and return a list of BSON documents with all of our embeddings ready to be queried using our automatic scalar or binary quantized vectors.

private static List<Document> embedDataAndCreateDocument(List<String> data) {  
    OkHttpClient client = new OkHttpClient.Builder()  
            .connectTimeout(CONNECTION_TIMEOUT, TimeUnit.SECONDS)  
            .readTimeout(READ_TIMEOUT, TimeUnit.SECONDS)  
            .build();

    try {  
        ResponseDouble baselineValues = embedBaselineDoubles(client, data);   

        // Create a list of BSON documents containing all embeddings  
        return createCombinedEmbeddingsDocument(  
                data,  
                baselineValues);  
    } catch (IOException e) {  
        throw new RuntimeException("Error fetching embedding: ", e);  
    }  
}

First, we'll create an OkHttpClient to connect to our Voyage AI API. We'll then query our API and return our response to our ResponseDouble record. We'll define our embedBaselineDoubles method to take in this client and the data to embed.

Getting our embeddings with Voyage AI

Now, we are going to need a method to communicate with the Voyage AI API. Let's define a method sendRequest() that will allow us to do this. We are going to reuse this method throughout this tutorial, so we will add configuration for the output type in the method parameters, along with the client and data to embed.

// Send API request to Voyage AI  
private static String sendRequest(  
        OkHttpClient client,  
        List<String> data,  
        String outputDataType  
) throws IOException {  

    String requestBody = new JSONObject()  
            .put("input", data)  
            .put("model", "voyage-3-large")  
            .put("input_type", "document")  
            .put("output_dtype", outputDataType)  
            .put("output_dimension", 1024)  
            .toString();  

    Request request = new Request.Builder()  
            .url(VOYAGE_API_URL)  
            .post(RequestBody.create(requestBody, MediaType.get("application/json")))  
            .addHeader("Authorization", "Bearer " + VOYAGE_API_KEY)  
            .build();  

    try (Response response = client.newCall(request).execute()) {  
        if (!response.isSuccessful()) {  
            throw new IOException("API error: HTTP " + response.code() + " - " + response.message());  
        }  

        if (response.body() == null) {  
            throw new IOException("Empty response body");  
        }  

        String json = response.body().string();  
        System.out.println("Raw API JSON:\n" + json);  
        return json;  
    }  
}

We'll also be defining the embedding model to use, the input data type, and the output dimensions we need. It's important that the output dimension we use uses the same number as the dimensions we will specify in our vector search index.

After this, we can define our method embedBaselineDoubles. We will use this to call our sendRequest and define our output data type to float. In MongoDB, we'll be storing this output as doubles, hence the naming conventions I use.

private static ResponseDouble embedBaselineDoubles(OkHttpClient client, List<String> data) throws IOException {  
    String outputDataType = "float";  
    return OBJECT_MAPPER.readValue(sendRequest(client, data, outputDataType), ResponseDouble.class);  
}

This method takes a list of strings, sends them to our embedding service requesting the float-based embeddings, and then deserializes the JSON response into a ResponseDouble object.

Creating our BSON document

Lastly, we can define our method createCombinedEmbeddingsDocument that takes these embeddings and converts them to our document to store in MongoDB, that provides our response for the embedDataAndCreateDocument method.

// Create a list of BSON document containing all embedding responses  
private static List<Document> createCombinedEmbeddingsDocument(  
        List<String> texts,  
        ResponseDouble floatResponse 
) {  
    List<Document> embeddingsList = new ArrayList<>();  

    for (int i = 0; i < texts.size(); i++) {  
        Document embeddingDoc = new Document()  
                .append("text", texts.get(i))  
                .append("embeddings_float32", floatResponse.data().get(i).embedding())  
                .append("embeddings_auto_scalar", floatResponse.data().get(i).embedding())  
                .append("embeddings_auto_binary", floatResponse.data().get(i).embedding());  
        embeddingsList.add(embeddingDoc);  
    }  

    return embeddingsList;  
}

This method takes our texts and embeddings and builds MongoDB documents that include three parallel embedding fields:

One for float32 storage
One for auto scalar quantization
One for auto binary quantization

Even though they all store the same raw values right now, MongoDB will use different index definitions (float32, scalar quantized, binary quantized) depending on which field we target.

Storing in MongoDB

After all this, we can define our storeAllEmbeddings method. This will take in our MongoClient we define in our main method, as well as the list of embeddings, and insert them into the database and collection we define as constants at the start of our class.

public static void storeAllEmbeddings(MongoCollection<Document> collection, List<Document> document) {  
collection.insertMany(document); 
System.out.println("Inserted embeddings documents into MongoDB");
}

We can create our embeddings, create our documents, and store them in MongoDB, but how can we use vector search to query our data? Well, we need to define a vector search index.

Creating our Vector Search index

We'll create a method setupVectorSearchIndex that we can define our index:

// Create the Vector Search index  
public static void setupVectorSearchIndex(MongoClient client) throws InterruptedException {  
    MongoDatabase database = client.getDatabase(DB_NAME);  
    MongoCollection<Document> collection = database.getCollection(COLLECTION_NAME);  

    Bson definition = new Document(  
            "fields",  
            List.of(  
                    new Document("type", "vector")  
                            .append("path", "embeddings_float32")  
                            .append("numDimensions", 1024)  
                            .append("similarity", "dotProduct"),  
                    new Document("type", "vector")  
                            .append("path", "embeddings_auto_scalar")  
                            .append("numDimensions", 1024)  
                            .append("similarity", "dotProduct")  
                            .append("quantization", "scalar"),  
                    new Document("type", "vector")  
                            .append("path", "embeddings_auto_binary")  
                            .append("numDimensions", 1024)  
                            .append("similarity", "dotProduct")  
                            .append("quantization", "binary")
            )  
    );  

    SearchIndexModel indexModel = new SearchIndexModel(  
            INDEX_NAME,  
            definition,  
            SearchIndexType.vectorSearch()  
    );  

    List<String> result = collection.createSearchIndexes(Collections.singletonList(indexModel));

    waitForIndex(collection, INDEX_NAME);  
}

This method wires up the vector search index so we can actually query our embeddings. We grab the collection, then define an index with three vector fields:

embeddings_float32: Raw float vectors (stored as doubles in BSON). No quantization.
embeddings_auto_scalar: Same vectors, but the index applies automatic scalar quantization.
embeddings_auto_binary: Same vectors, but with automatic binary quantization.

Each one uses numDimensions: 1024 and similarity: dotProduct. You’d adjust numDimensions if your model outputs a different size.

The reason for three fields is simple: It lets you compare trade-offs.

No quantization gives the highest recall but eats the most memory.
Scalar cuts memory roughly to a quarter while keeping recall high.
Binary is the leanest and fastest, but recall can dip depending on the data.

MongoDB handles the quantization inside the index. In your documents, you still just store doubles.

Then, we can add our waitForIndex method that will poll the database until the build is complete before moving on:

// Wait for the index build to complete  
public static <T> void waitForIndex(final MongoCollection<T> collection, final String indexName) {  
    long startTime = System.nanoTime();  
    long timeoutNanos = TimeUnit.SECONDS.toNanos(60);  
    while (System.nanoTime() - startTime < timeoutNanos) {  
        Document indexRecord = StreamSupport.stream(collection.listSearchIndexes().spliterator(), false)  
                .filter(index -> indexName.equals(index.getString("name")))  
                .findAny().orElse(null);  
        if (indexRecord != null) {  
            if ("FAILED".equals(indexRecord.getString("status"))) {  
                throw new RuntimeException("Search index has FAILED status.");  
            }  
            if (indexRecord.getBoolean("queryable")) {  
                System.out.println(indexName + " index is ready to query");  
                return;  
            }  
        }  
        try {  
            Thread.sleep(400); // busy-wait, avoid in production  
        } catch (InterruptedException e) {  
            Thread.currentThread().interrupt();  
            throw new RuntimeException(e);  
        }  
    }  
}

Query with Vector Search

To query the data, we will define a runVectorSearchQueries method. We'll be using this method again later to query pre-quantized vectors, but for now, we will just prepare it to query our data that uses the automatically quantized data (as well as a baseline that uses no quantization):

// Run MongoDB vector search queries for all fields using query embeddings  
private static void runVectorSearchQueries(MongoCollection<Document> collection) {  
    OkHttpClient client = new OkHttpClient.Builder()  
            .connectTimeout(CONNECTION_TIMEOUT, TimeUnit.SECONDS)  
            .readTimeout(READ_TIMEOUT, TimeUnit.SECONDS)  
            .build();  

    List<String> queryInput = List.of(QUERY_TEXT);  

    try { 
        ResponseDouble qFloatResp = embedBaselineDoubles(client, queryInput);  
        List<Double> qFloat = qFloatResp.data().getFirst().embedding();

        runVectorSearchDouble(collection, qFloat);

    } catch (IOException e) {  
        throw new RuntimeException("Error embedding query vector: ", e);  
    }  
}

First, we will create an embedding for our query using the embedBaselineDoubles method from earlier. We can extract our vectors from the response. Next, we'll call a runVectorSearchDoubles method. Let's create this method:

private static void runVectorSearchDouble(  
        MongoCollection<Document> collection,  
        List<Double> floatQuery  
) {  
    for (String embeddingType : List.of("embeddings_float32", "embeddings_auto_scalar", "embeddings_auto_binary")) {  
        List<Bson> pipeline = asList(  
                vectorSearch(  
                        fieldPath(embeddingType),  
                        floatQuery,  
                        INDEX_NAME,  
                        2, approximateVectorSearchOptions(5)  
                ),  
                project(fields(  
                        exclude("_id"),  
                        include("text"),  
                        metaVectorSearchScore("vectorSearchScore"))));  

        List<Document> results = collection.aggregate(pipeline).into(new ArrayList<>());  

        System.out.println("Outputting results: ");  
        for (Document result : results) {  
            System.out.println(result.toJson());  
        }  
    }  
}

We loop through each embedding type, embeddings_float32, embeddings_auto_scalar, and embeddings_auto_binary. For each one, we build a pipeline:

vectorSearch stage
- fieldPath(embeddingType) tells MongoDB which embedding field to search.
- floatQuery is our query vector (from the earlier embedBaselineDoubles method).
- INDEX_NAME points to the vector search index we created.
- 2 means return the top two matches.
- approximateVectorSearchOptions(5) says “use ANN with a candidate pool of 5.” That balances speed versus accuracy.
project stage
- Excludes _id (so results are cleaner).
- Includes just text and the computed vectorSearchScore.

We then run the aggregation with collection.aggregate(pipeline) and dump the results into a list. Finally, the results are printed, showing each document’s text and score.

Now, let's update our main method to include all the necessary calls to our methods:

public class Main {

    // Configurations
    private static final String VOYAGE_API_KEY = System.getenv("VOYAGE_API_KEY");
    private static final String MONGODB_URI = System.getenv("MONGODB_URI");
    private static final String DB_NAME = "test";
    private static final String COLLECTION_NAME = "demo";
    private static final String INDEX_NAME = "vector_index";
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    // Voyage AI API Endpoint
    private static final String VOYAGE_API_URL = "https://api.voyageai.com/v1/embeddings";

    // Timeout values for API requests
    private static final int CONNECTION_TIMEOUT = 30;
    private static final int READ_TIMEOUT = 60;

    // Sample Data
    private static final List<String> DATA = List.of(
            "The Great Wall of China is visible from space.",
            "The Eiffel Tower was completed in Paris in 1889.",
            "Mount Everest is the highest peak on Earth at 8,848m.",
            "Shakespeare wrote 37 plays and 154 sonnets during his lifetime.",
            "The Mona Lisa was painted by Leonardo da Vinci."
    );

    private static final String QUERY_TEXT = "Country landmarks";

    public static void main(String[] args) {
        if (VOYAGE_API_KEY == null || VOYAGE_API_KEY.isEmpty()) {
            throw new RuntimeException("API key not found. Set VOYAGE_API_KEY in your environment.");
        }
        if (MONGODB_URI == null || MONGODB_URI.isEmpty()) {
            throw new RuntimeException("MongoDB URI not found. Set MONGODB_URI in your environment.");
        }

        // create embeddings and collect all responses
        System.out.println("Embedding data...");
        List<Document> allEmbeddingsDocuments = embedDataAndCreateDocument(DATA);

        try (MongoClient mongoClient = MongoClients.create(MONGODB_URI)) {
            MongoDatabase database = mongoClient.getDatabase(DB_NAME);
            MongoCollection<Document> collection = database.getCollection(COLLECTION_NAME);
            // store all embeddings in mongodb
            storeAllEmbeddings(collection, allEmbeddingsDocuments);
            // create vector search index
            setupVectorSearchIndex(collection);

            System.out.println("Querying data...");
            runVectorSearchQueries(collection);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

While it is perfectly fine to run this now, and we will use the full functionality of the auto-quantization with MongoDB, we are going to walk through how to update this class to handle pre-quantized vectors.

How to ingest pre-quantized vectors

Pre-quantized vectors are supported by some embedding models, such as Voyage AI’s voyage-3-large. Using them means the model itself does the quantization step for us, so we can store and index the compressed representation directly instead of relying on MongoDB to apply automatic quantization during index build.

This has two main benefits:

Smaller payloads from the API: Instead of receiving large float32 arrays, we get compact int8 or even int1 arrays.
Less memory overhead in MongoDB: We store the already-quantized representation, which keeps both storage size and index RAM cost lower.

MongoDB supports ingestion of:

int8 vectors (stored as BSON binData(int8)).
int1 vectors (stored as BSON binData(int1) with packed bits).

Unlike the auto-quantization paths, these do not require that we define our index specifically for int8 or int1 ingestion.

Embedding our data

Now, to implement this, we will update our code from earlier. First, in embedDataAndCreateDocument, we will update it to embed our data to int8 and int1 and return to our ResponseBytes record.

private static List<Document> embedDataAndCreateDocument(List<String> data) {  
    OkHttpClient client = new OkHttpClient.Builder()  
            .connectTimeout(CONNECTION_TIMEOUT, TimeUnit.SECONDS)  
            .readTimeout(READ_TIMEOUT, TimeUnit.SECONDS)  
            .build();

    try {  
        ResponseDouble baselineValues = embedBaselineDoubles(client, data);  
        ResponseBytes preQuantizedInt8 = embedPreQuantizedInt8(client, data);  
        ResponseBytes preQuantizedInt1 = embedPreQuantizedInt1(client, data);  

        // Create a list of BSON documents containing all embeddings  
        return createCombinedEmbeddingsDocument(  
                data,  
                baselineValues,  
                preQuantizedInt8,  
                preQuantizedInt1);  
    } catch (IOException e) {  
        throw new RuntimeException("Error fetching embedding: ", e);  
    }  
}

We'll be using two new methods to get our pre-quantized data—first, embedPreQuantizedInt8, which works much like our embedBaselineDouble, but passes in the output data type int8 to tell Voyage AI that we want pre-quantized scalar values:

private static ResponseBytes embedPreQuantizedInt8(OkHttpClient client, List<String> data) throws IOException {  
    String outputDataType = "int8";  
    return OBJECT_MAPPER.readValue(sendRequest(client, data, outputDataType), ResponseBytes.class);  
}

We then need an embedPreQuantizedInt1 method to pass in the output type ubinary to get our packed bits response:

private static ResponseBytes embedPreQuantizedInt1(OkHttpClient client, List<String> data) throws IOException {  
    String outputDataType = "ubinary";  
    return OBJECT_MAPPER.readValue(sendRequest(client, data, outputDataType), ResponseBytes.class);  
}

How to store pre-quantized data in MongoDB

We also need to update our createCombinedEmbeddingsDocument to take in these new embeddings to store in MongoDB.

// Create a list of BSON document containing all embedding responses  
private static List<Document> createCombinedEmbeddingsDocument(  
        List<String> texts,  
        ResponseDouble floatResponse,  
        ResponseBytes int8Response,  
        ResponseBytes int1Response  
) {  
    List<Document> embeddingsList = new ArrayList<>();  

    for (int i = 0; i < texts.size(); i++) {  
        Document embeddingDoc = new Document()  
                .append("text", texts.get(i))  
                .append("embeddings_float32", floatResponse.data().get(i).embedding())  
                .append("embeddings_auto_scalar", floatResponse.data().get(i).embedding())  
                .append("embeddings_auto_binary", floatResponse.data().get(i).embedding())  
                .append("embeddings_int8", BinaryVector.int8Vector(int8Response.data().get(i).embedding()))  
                .append("embeddings_int1", BinaryVector.packedBitVector(int1Response.data().get(i).embedding(), (byte) 0));  
        embeddingsList.add(embeddingDoc);  
    }  

    return embeddingsList;  
}

We still store doubles for the auto paths and we now store pre-quantized bytes for int8/int1. MongoDB Atlas treats them as quantized at ingestion, so no conversion.

org.bson.BinaryVector is a helper type in the MongoDB Java Driver. It wraps raw byte[] so MongoDB knows we’re storing a vector type (binData(int8) or binData(int1)) instead of just a random blob of bytes.

MongoDB Atlas Vector Search requires these wrapped types to understand how to treat the data when building a vector index.

BinaryVector.int8Vector(...)

Takes the byte[] from our embedding model that already represents int8-quantized values.
Wraps it as a BSON binData(int8) vector.
MongoDB will store this as a compact 8-bit signed integer vector, and the index will consume it directly with no further conversion.

BinaryVector.packedBitVector(..., (byte) 0)

Takes the byte[] returned for 1-bit embeddings (ubinary).
These vectors are bit-packed. Each bit represents one dimension of the embedding (0 if ≤ midpoint, 1 if > midpoint).
BinaryVector.packedBitVector wraps that array and tags it as BSON binData(int1).
The (byte) 0 is a padding value. If the vector’s number of dimensions isn’t an exact multiple of eight, MongoDB Atlas needs to know how to pad the unused bits in the last byte (usually 0).

Updating our Vector Search index

Now, we need to update the setupVectorSearchIndex to handle these new fields, and prepare them for vector search:

// Create the Vector Search index  
public static void setupVectorSearchIndex(MongoCollection<Document> collection) throws InterruptedException {  
       Bson definition = new Document(  
            "fields",  
            List.of(  
                    new Document("type", "vector")  
                            .append("path", "embeddings_float32")  
                            .append("numDimensions", 1024)  
                            .append("similarity", "dotProduct"),  
                    new Document("type", "vector")  
                            .append("path", "embeddings_auto_scalar")  
                            .append("numDimensions", 1024)  
                            .append("similarity", "dotProduct")  
                            .append("quantization", "scalar"),  
                    new Document("type", "vector")  
                            .append("path", "embeddings_auto_binary")  
                            .append("numDimensions", 1024)  
                            .append("similarity", "dotProduct")  
                            .append("quantization", "binary"),  
                    new Document("type", "vector")  
                            .append("path", "embeddings_int8")  
                            .append("numDimensions", 1024)  
                            .append("similarity", "dotProduct"),  
                    new Document("type", "vector")  
                            .append("path", "embeddings_int1")  
                            .append("numDimensions", 1024)  
                            .append("similarity", "euclidean")  
            )  
    );  

    SearchIndexModel indexModel = new SearchIndexModel(  
            INDEX_NAME,  
            definition,  
            SearchIndexType.vectorSearch()  
    );  

    List<String> result = collection.createSearchIndexes(Collections.singletonList(indexModel));

    waitForIndex(collection, INDEX_NAME);  
}

We've added our two new fields, embeddings_int8 and embeddings_int1, to our index definition. We are still using the same number of dimensions for our embeddings, as this is defined by the model used for embedding, There are two big differences.

We do not need to define quantization.
For int1 vector quantization, we use the euclidean similarity method for searching. This is because int1 does not support dotProduct or cosine similarity methods.

If you have run the code before, and this index already exists on the database, you will need to delete it, or manually update it to avoid any duplicate index errors.

Searching pre-quantized vectors

Now we have our new, pre-quantized data ready, we need to update our runVectorSearchQueries method to embed our queries to these pre-quantised values.

// Run MongoDB vector search queries for all fields using query embeddings  
private static void runVectorSearchQueries(MongoCollection<Document> collection) {  
    OkHttpClient client = new OkHttpClient.Builder()  
            .connectTimeout(CONNECTION_TIMEOUT, TimeUnit.SECONDS)  
            .readTimeout(READ_TIMEOUT, TimeUnit.SECONDS)  
            .build();  

    List<String> queryInput = List.of(QUERY_TEXT);  

    try { 
        ResponseDouble qFloatResp = embedBaselineDoubles(client, queryInput);  
        List<Double> qFloat = qFloatResp.data().getFirst().embedding();  

        ResponseBytes qInt8Resp = embedPreQuantizedInt8(client, queryInput);  
        BinaryVector qInt8 = BinaryVector.int8Vector(qInt8Resp.data().getFirst().embedding());  

        ResponseBytes qInt1Resp = embedPreQuantizedInt1(client, queryInput);  
        BinaryVector qInt1 = BinaryVector.packedBitVector(qInt1Resp.data().getFirst().embedding(), (byte) 0);  

        runVectorSearchDouble(collection, qFloat);  

        runVectorSearchBinary(collection, qInt8, qInt1);  

    } catch (IOException e) {  
        throw new RuntimeException("Error embedding query vector: ", e);  
    }  
}

We'll use our embedPreQuantizedInt8 and embedPreQuantizedInt1 methods to get our embeddings for our query.

Lastly, we need a new method runVectorSearchBinary method to query MongoDB with these values:

private static void runVectorSearchBinary(  
        MongoCollection<Document> collection,  
        BinaryVector qInt8,  
        BinaryVector qInt1  
) {  
    Map<String, BinaryVector> queryByField = new LinkedHashMap<>();  
    queryByField.put("embeddings_int8", qInt8); // int8  
    queryByField.put("embeddings_int1", qInt1); // packed 1-bit  

    for (Map.Entry<String, BinaryVector> entry : queryByField.entrySet()) {  
        List<Bson> pipeline = asList(  
                vectorSearch(  
                        fieldPath(entry.getKey()),  
                        entry.getValue(),  
                        INDEX_NAME,  
                        2, approximateVectorSearchOptions(5)  
                ),  
                project(fields(  
                        exclude("_id"),  
                        include("text"),  
                        metaVectorSearchScore("vectorSearchScore"))));  

        List<Document> results = collection.aggregate(pipeline).into(new ArrayList<>());  

        System.out.println("Outputting results: ");  
        for (Document result : results) {  
            System.out.println(result.toJson());  
        }  
    }  
}

Running our application

We can now run our application with the command:

mvn clean compile exec:java -Dexec.mainClass="com.timkelly.Main"

And we’ll see our results printed to the console:

Outputting results: 
{"text": "The Great Wall of China is visible from space.", "vectorSearchScore": 0.8666857481002808}
{"text": "Mount Everest is the highest peak on Earth at 8,848m.", "vectorSearchScore": 0.8523434400558472}
Outputting results: 
{"text": "The Great Wall of China is visible from space.", "vectorSearchScore": 0.8664929866790771}
{"text": "Mount Everest is the highest peak on Earth at 8,848m.", "vectorSearchScore": 0.8526520729064941}
Outputting results: 
{"text": "The Great Wall of China is visible from space.", "vectorSearchScore": 0.8666857481002808}
{"text": "Mount Everest is the highest peak on Earth at 8,848m.", "vectorSearchScore": 0.8523434400558472}
Outputting results: 
{"text": "The Great Wall of China is visible from space.", "vectorSearchScore": 0.5071338415145874}
{"text": "Mount Everest is the highest peak on Earth at 8,848m.", "vectorSearchScore": 0.5067881345748901}
Outputting results: 
{"text": "The Great Wall of China is visible from space.", "vectorSearchScore": 0.771484375}
{"text": "The Mona Lisa was painted by Leonardo da Vinci.", "vectorSearchScore": 0.7451171875}

You’ll notice that the various methods we used (baseline float32, automatic scalar, automatic binary, and the pre-quantized int8/int1 paths) all return slightly different results and scores.

That’s the trade-off in action:

Float32 (no quantization): Highest fidelity, most memory usage.
Auto-scalar: Scores almost identical to float32, but RAM cost is a fraction.
Auto-binary: Tighter memory, recall dips slightly depending on the dataset.
Pre-quantized int8/int1: Smallest payloads, fastest index use, but you can see the scores shift more noticeably.

This is why MongoDB gives us all these options. We can pick the right balance of accuracy, memory footprint, and query latency for our application.

Conclusion

With this setup, we’ve seen how to embed data, store it in MongoDB, and query it with Atlas Vector Search across multiple quantization strategies. Automatic quantization makes it easy to reduce memory while keeping high recall, while pre-quantized vectors give you the smallest payloads and fastest queries when your embedding model supports them. The trade-offs between accuracy, resource usage, and speed become clear once you run the same query through each index path.

By experimenting with float32, scalar, binary, int8, and int1 vectors side by side, you can choose the right approach for your application’s needs, whether that’s maximum recall, minimal resource footprint, or a balance between the two.

If you found this tutorial useful, and want to learn more about what you can do with MongoDB in Java, check out how I turned my Obsidian vault into a searchable wiki or Secure Local RAG with Role-Based Access: Spring AI, Ollama & MongoDB.

Turning My Obsidian Vault Into a Searchable Wiki With Spring Boot and MongoDB

Tim Kelly — Wed, 13 Aug 2025 10:31:36 +0000

Like any note-taker worth their salt, I’ve built up a cumbersome, often neglected pile of half-formed thoughts or ideas that crossed my mind throughout the day. I’ve always believed there were nuggets of value buried in that chaos.

Among these notes are recipes, tech experiments, a plethora of computer science knowledge from college, and many follow standardized templates I created. I wanted to share them online through a blog, making them searchable for anyone who might find them useful.

So instead of dumping these into a CMS by hand, I decided to repurpose my Obsidian vault into a searchable wiki I can embed on my blog. Since everything is already in Markdown and includes metadata in frontmatter, I could skip the copy-paste slog and get straight to building something useful.

This tutorial walks through how I turned my local notes into a web-accessible knowledge base, with:

MongoDB Atlas Search for full-text, fuzzy, and autocomplete querying.
Spring Boot for ingesting and formatting data, as well as the REST API.
Custom weighting and boosting to prioritise my best and most useful notes.

I’ll be indexing my own technical notes on some frameworks as a demo, and I’ll show you how to expose and query them through an API, so you can do the same with yours.

If you just want the code, it's all available in this Github repo.

What is Obsidian?

Obsidian is a free note-taking app that lets you write and organize your thoughts in plain text using Markdown. It’s designed for people who want to build a personal knowledge base, with features like backlinks and graph views, and it’s designed to help users organize and structure their thoughts and knowledge in a flexible, non-linear way.

I like Obsidian as a note-taking app because it allows a lot of customization, and it’s really just a wrapper for viewing Markdown files. That means I never have to worry about my notes being locked away if Obsidian disappears. They’re still just plain text on disk. It also has a vast community plugin ecosystem and makes it easy to build your own, so I can optimize my workflow exactly how I want.

Why MongoDB Atlas Search makes sense

Since Obsidian stores everything as plain Markdown with optional frontmatter, the structure maps cleanly to documents in MongoDB. I don’t have to force a schema onto my notes and can store multiple formats in the same collection. Each one can have different tags, metadata, or headings, and MongoDB’s flexible model handles that without complaint. But the real reason this setup works so well is MongoDB Atlas Search.

It gives us full-text search, autocomplete, and fuzzy matching, all built in, no separate indexing service or extra infrastructure needed. We can even boost certain fields to prioritize titles, tags, or whatever matters most to us. Want to weigh title matches higher than body content? Or prioritize notes with certain tags? Totally up to us.

The core of it all is the $search stage. For something as messy and inconsistent as a personal vault of thoughts, it’s more than robust enough to build the search for my wiki.

With that out of the way, let’s get into how I actually built it, starting with parsing notes and getting them into MongoDB.

Prerequisites

Before we get started, there are a few things we'll need:

A MongoDB Atlas account with a cluster set up
- A free M0 cluster is perfect for this
Java on your machine, set up and ready to go (I use Java version 24, but for some of the code, you will need at least Java version 22!)
Maven (I use version 3.9.10)
An Obsidian vault, with some notes inside

Creating our Spring app

To get started, we’ll scaffold a simple Spring Boot project using Spring Initializr. We’re keeping things minimal, just enough to expose a REST API and connect to MongoDB.

Here’s what I selected on Spring Initializr:

Project: Maven
Language: Java
Spring Boot: 3.x (any stable version works)
Dependencies:
- Spring Web—to expose our REST API
- Spring Data MongoDB—to interact with our MongoDB collection

Once we've configured those options, give it a name (SpringSearch) and define the group (com.timkelly), click Generate, unzip the project, and open it in our IDE of choice.

From here, we’ll start wiring things up so that Spring can read our notes, store them in MongoDB, and expose them through a clean API.

Storing in MongoDB

Before we start importing notes, let’s make sure our app can actually talk to MongoDB.

In our application.properties, we add the connection string for our MongoDB Atlas cluster, and specify the database we'd like to use:

spring.data.mongodb.uri=YOUR-CONNECTION-STRING
spring.data.mongodb.database=obsidian

Using MongoDB Atlas, you can grab your connection string from the cluster dashboard. Just make sure to whitelist your IP and create a database user.

Now, let’s define a simple repository so Spring Data MongoDB can handle the persistence for us. Create a package repository and in it, we'll define an interface NoteRepository:

package com.timkelly.springsearch.repository;

import com.timkelly.springsearch.model.Note;
import org.springframework.data.mongodb.repository.MongoRepository;
import org.springframework.stereotype.Repository;

import java.util.Optional;

@Repository
public interface NoteRepository extends MongoRepository<Note, String> {
    Optional<Note> findByTitle(String title);
}

This gives us built-in CRUD operations out of the box (thanks, MongoRepository), and we add one custom method: findByTitle(String title), which we’ll use later to check if a note already exists before inserting or updating it.

Ingesting Obsidian notes

For this tutorial, I’m using a few sample notes on tech frameworks, each following a simple Markdown template, that I have included in the Github repo:

---
tags:
  - List
  - of
  - items
created: YYYY-MM-DD HH:mm
---
# Title
Content Lorem Ipsum

This structure gives us a frontmatter block that we’ll treat as metadata. The tags array defines relevant topics for the note (e.g., AI, Java, REST), and created marks when the note was written. Below that, we assume a level-one heading (#) for the title, followed by the main body content.

If your notes are structured differently, you can adjust the parser to suit. There’s nothing stopping you from using the filename as a title, omitting the frontmatter, or extending the metadata. MongoDB is even happy to handle documents with varied structures in the same collection.

These notes will be mapped to MongoDB documents like this:

{
    "title": "String",
    "tags": ["String"],
    "createdAt": "ISODate",
    "content": "String"
}

Now, in order to interact with MongoDB and map our documents properly, we’ll need a Note model.

Create a new class called Note inside a new package named model in our Spring app. We’ll use the @Document annotation from Spring Data MongoDB to let Spring know that this class represents a document in the "notes" collection.

Here’s what it looks like:

package com.timkelly.springsearch.model;  

import org.springframework.data.mongodb.core.mapping.Document;  

import java.util.Date;  

@Document(collection = "notes")
public class Note {
    private String title;
    private String content;
    private String[] tags;
    private Date createdAt;

    public Note() {
    }

    public Note(String title, String content, String[] tags, Date createdAt) {
        this.title = title;
        this.content = content;
        this.tags = tags;
        this.createdAt = createdAt;
    }

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    public String[] getTags() {
        return tags;
    }

    public void setTags(String[] tags) {
        this.tags = tags;
    }

    public Date getCreatedAt() {
        return createdAt;
    }

    public void setCreatedAt(Date createdAt) {
        this.createdAt = createdAt;
    }
}

Nothing fancy—we're defining the fields we want to store in MongoDB, along with a no-args constructor (needed by Spring) and a full constructor for convenience. Then, we generate the usual getters and setters to interact with our object.

Now that we’ve got a Note model and a structure we’re happy with, let’s actually load the files from our Obsidian vault and push them into MongoDB.

First, let's define where our notes live. In our application.properties or application.yml, we need to add the following entry pointing to our notes folder. Here’s my dummy data in the GitHub repo, if you’re following along:

notes.folder.path=dummyData

Next, we’ll bind that value using a simple Spring @ConfigurationProperties class. Create a config package and add the class below, NotesFolderProperties:

package com.timkelly.springsearch.config;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConfigurationProperties(prefix = "notes.folder")
public class NotesFolderProperties {
    private String path;
    public String getPath() { return path; }
    public void setPath(String path) { this.path = path; }
}

This lets us inject the file path wherever we need it.

Parsing Markdown

After our config, we need some logic to parse our Markdown and extract our data. Since the format of our notes is predictable for this example, converting them into our document structure is straightforward. Below is an implementation of a MarkdownNoteParser class, which we can place in a new util package:

package com.timkelly.springsearch.util;  
import com.timkelly.springsearch.model.Note;  
import org.springframework.stereotype.Component;  
import org.yaml.snakeyaml.Yaml;  

import java.text.ParseException;  
import java.text.SimpleDateFormat;  
import java.util.*;  
import java.util.regex.*;  

@Component  
public class MarkdownNoteParser {  

    private static final Pattern FRONTMATTER_PATTERN = Pattern.compile("^---\\s*\\n(.*?)\\n---\\s*\\n", Pattern.DOTALL);  
    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm");  

    public Note parse(String rawMarkdown) {  
        Matcher matcher = FRONTMATTER_PATTERN.matcher(rawMarkdown);  

        Map<String, Object> metadata = new HashMap<>();  
        if (matcher.find()) {  
            String frontmatter = matcher.group(1);  
            rawMarkdown = rawMarkdown.substring(matcher.end());  
            Yaml yaml = new Yaml();  
            metadata = yaml.load(frontmatter);  
        }  

        String title = extractTitle(rawMarkdown);  
        String content = rawMarkdown.trim();  

        String[] tags = Optional.ofNullable(metadata.get("tags"))  
                .map(t -> ((List<?>) t).stream().map(Object::toString).toArray(String[]::new))  
                .orElse(new String[0]);  

        Date createdAt = Optional.ofNullable(metadata.get("created"))  
                .map(Object::toString)  
                .map(this::parseDate)  
                .orElse(null);  

        return new Note(title, content, tags, createdAt);  
    }  

    private String extractTitle(String markdown) {  
        Scanner scanner = new Scanner(markdown);  
        while (scanner.hasNextLine()) {  
            String line = scanner.nextLine().trim();  
            if (line.startsWith("# ")) {  
                return line.substring(2).trim();  
            }  
        }  
        return "Untitled";  
    }  

    private Date parseDate(String dateStr) {  
        try {  
            return DATE_FORMAT.parse(dateStr);  
        } catch (ParseException e) {  
            System.err.println("Failed to parse date: " + dateStr);  
            return null;  
        }  
    }  
}

This class uses YAML to parse the frontmatter metadata block, a regular expression to detect the frontmatter, and a simple scanner to extract the title.

The FRONTMATTER_PATTERN constant defines a regex for identifying YAML frontmatter blocks in the Markdown content. For example, if the frontmatter is:

---
tags:
  - java
  - spring
created: 2025-07-23 12:00
---

It will be extracted as {tags=[java, spring], created="2025-07-23 12:00"}, and the remaining content will be stored in rawMarkdown after the frontmatter is removed.

Next, we use the extractTitle(rawMarkdown) helper method to detect the note’s title, provided the Markdown starts with a # Heading. We also check for a tags field in the metadata, converting it to a string array if present.

For the creation date, parseDate(String dateStr) parses the created value from the metadata into a Date object.

Finally, we trim any extra whitespace from the Markdown body using rawMarkdown.trim() before returning a fully populated Note object.

Importing from our vault

Now that we can parse our Markdown notes, the next step is to scan an entire folder in our vault, parse every .md file, and insert or update each note in MongoDB. For this, we’ll create a service called NoteImportService inside a new service package:

package com.timkelly.springsearch.service;

import com.timkelly.springsearch.config.NotesFolderProperties;
import com.timkelly.springsearch.model.Note;
import com.timkelly.springsearch.repository.NoteRepository;
import com.timkelly.springsearch.util.MarkdownNoteParser;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.nio.file.*;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;

@Service
public class NoteImportService {

    private final NotesFolderProperties props;
    private final MarkdownNoteParser parser;
    private final NoteRepository repository;

    public NoteImportService(NotesFolderProperties props, MarkdownNoteParser parser, NoteRepository repository) {
        this.props = props;
        this.parser = parser;
        this.repository = repository;
    }

    public void importNotes() throws IOException {

        List<Path> files = Files.walk(Paths.get(props.getPath()))
                .filter(p -> p.toString().endsWith(".md"))
                .toList();

        for (Path path : files) {
            String content = Files.readString(path);
            Note newNote = parser.parse(content);

            repository.findByTitle(newNote.getTitle())
                    .ifPresentOrElse(existing -> {
                        boolean changed =
                                !Objects.equals(existing.getContent(), newNote.getContent()) ||
                                        !Arrays.equals(existing.getTags(), newNote.getTags()) ||
                                        !Objects.equals(existing.getCreatedAt(), newNote.getCreatedAt());

                        if (changed) {
                            existing.setContent(newNote.getContent());
                            existing.setTags(newNote.getTags());
                            existing.setCreatedAt(newNote.getCreatedAt());
                            repository.save(existing);
                            System.out.println("Updated: " + existing.getTitle());
                        } else {
                            System.out.println("Skipped (no changes): " + existing.getTitle());  
                        }  
                    }, () -> {  
                        repository.save(newNote);  
                        System.out.println("Inserted: " + newNote.getTitle());  
                    });  
        }  
    }  

}

This service walks through your vault, finds every .md file, and parses it into a Note object. It then checks MongoDB for an existing note with the same title. The repository method findByTitle(...) looks for a document with the same title.

If no note with that title exists, the note is saved as a new document. If a note is found, the service compares three fields to detect changes:

content: Compared using Objects.equals(...).
tags: Compared using Arrays.equals(...).
createdAt: Compared using Objects.equals(...).

If any of these fields differ, the existing note is updated with the new values. Otherwise, the write is skipped to avoid unnecessary database operations. This approach is redimentary, but ensures we can run the import multiple times without creating duplicates or re-writing unchanged notes.

The duplicate detection here is very basic. It assumes that titles are unique across all notes. While this is fine for smaller directories (as in this example), larger vaults may benefit from optimizations such as:

Reducing the number of file checks by caching file modification times, or manually calling the files to upload.
Minimizing database queries by batch operations or indexing our regularly queried fields like title and createdAt.

Search our database with MongoDB Atlas Search

We have our documents in our database, modelled as we wish them to be. Now, it is time to search using the $search operator. This will allow us to do our full-text search, autocomplete, fuzzy matching, and custom boosting. I'll introduce them one after the other so we can have a better idea of how they all work individually, and then wrap it up with all of them in the one search query.

Our Atlas Search index

Before we can perform full-text queries, we need to configure an Atlas Search index for our collection. In our MongoDB Atlas cluster, go to the collection where you will be storing your documents and select the search index tab:

We need to create a new index for the notes collection in a database called obsidian with the following JSON configuration. Since we haven’t run our application yet, we need to manually create the notes collection and the obsidian database. Then, we can add our search index like below:

{
  "mappings": {
    "dynamic": false,
    "fields": {
      "title": [
        { "type": "string" },
        { "type": "autocomplete" }
      ],
      "tags": {
        "type": "string"
      },
      "content": {
        "type": "string"
      }
    }
  }
}

The configuration above enables full-text search on all three fields, and we’ve added an autocomplete mapping on the title field. This gives us the foundation for type-ahead search later.

Simple search

We’ll start with a simple search method that queries across all three fields (title, content, and tags) using the $search operator. We’ll build from here as we add more advanced search capabilities. In the repository package, create a NoteSearchRepository class like below:

package com.timkelly.springsearch.repository;  

import com.timkelly.springsearch.model.Note;  
import org.bson.Document;  
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.data.domain.Sort;  
import org.springframework.data.mongodb.core.MongoTemplate;  
import org.springframework.data.mongodb.core.aggregation.*;  
import org.springframework.stereotype.Repository;  

import java.util.List;  

@Repository  
public class NoteSearchRepository {  

    private final MongoTemplate mongoTemplate;  

    @Autowired  
    public NoteSearchRepository(MongoTemplate mongoTemplate) {  
        this.mongoTemplate = mongoTemplate;
    }

    public List<Note> search(String query) {
        AggregationOperation searchStage = _ -> new Document("$search",
                new Document("index", "default")
                        .append("text", new Document("query", query)
                                .append("path", List.of("title", "content", "tags")))
        );

        Aggregation aggregation = Aggregation.newAggregation(searchStage);  
        return mongoTemplate.aggregate(aggregation, "notes", Note.class).getMappedResults();  
    }

    // we'll add more search methods here...
}

This is our most basic search function. We use MongoTemplate to run an aggregation pipeline that includes a $search stage. The text operator tells MongoDB Atlas Search to look for the query string across the title, content, and tags fields. The results are then mapped directly into our Note model, giving us strongly typed objects ready for use in the rest of the application.

Next, we expose a REST endpoint so our frontend (or even just curl) can hit /api/notes/search?q=... and get matching results. Create a controller package and add a new class, NoteSearchController:

package com.timkelly.springsearch.controller;  

import com.timkelly.springsearch.model.Note;  
import com.timkelly.springsearch.repository.NoteSearchRepository;  
import org.springframework.web.bind.annotation.GetMapping;  
import org.springframework.web.bind.annotation.RequestMapping;  
import org.springframework.web.bind.annotation.RequestParam;  
import org.springframework.web.bind.annotation.RestController;  

import java.util.List;  

@RestController  
@RequestMapping("/api/notes")  
public class NoteSearchController {

    private final NoteSearchRepository searchRepository;

    public NoteSearchController(NoteSearchRepository searchRepository) {
        this.searchRepository = searchRepository;
    }

    @GetMapping("/search")  
    public List<Note> search(@RequestParam String q) {
        return searchRepository.search(q);
    }

    // we'll add more endpoints here
}

Here, we’ve wired up a simple /api/notes/search endpoint that takes a q parameter and passes it along to our repository. The controller returns a list of Note objects as JSON, making it easy for any frontend (or even a quick curl request) to query our indexed notes and get instant results.

Search boosted by title

By default, MongoDB Atlas Search treats all fields equally. But often, a match in the title should matter more than one in the body text. We can achieve this by using the compound operator and applying a boost to the score of title matches. We need to add the following to our NoteSearchRepository, below the earlier code:


    public List<Note> searchBoostedByTitle(String query) {  
        AggregationOperation searchStage = _ -> new Document("$search",  
                new Document("index", "default")  
                        .append("compound", new Document("should", List.of(  
                                new Document("text", new Document("query", query)
                                        .append("path", "title")  
                                        .append("score", new Document("boost", new Document("value", 5)))), // Boost title matches
                                new Document("text", new Document("query", query)
                                        .append("path", "content")),
                                new Document("text", new Document("query", query)
                                        .append("path", "tags"))
                        )))
        );

        AggregationOperation sortStage = Aggregation.sort(Sort.by(Sort.Order.desc("score")));

        Aggregation agg = Aggregation.newAggregation(searchStage, sortStage);

        return mongoTemplate.aggregate(agg, "notes", Note.class).getMappedResults();
    }

And in our NoteSearchController, we wire this up in our new endpoint:

    @GetMapping("/search/boost-title")  
    public List<Note> searchBoostedByTitle(@RequestParam String q) {  
        return searchRepository.searchBoostedByTitle(q);  
    }

Search boosted by title and tags

We can take this further by boosting both the title and tags fields. Here, the title gets the highest weight, tags are slightly less important, and content is left at default. In our NoteSearchRepository, add:

    public List<Note> searchBoostedTitleAndTags(String query) {  
        AggregationOperation searchStage = _ -> new Document("$search",  
                new Document("index", "default")  
                        .append("compound", new Document("should", List.of(  
                                new Document("text", new Document("query", query)  
                                        .append("path", "title")  
                                        .append("score", new Document("boost", new Document("value", 5)))),
                                new Document("text", new Document("query", query)  
                                        .append("path", "tags")  
                                        .append("score", new Document("boost", new Document("value", 3)))),
                                new Document("text", new Document("query", query)  
                                        .append("path", "content"))  
                        )))  
        );  

        Aggregation agg = Aggregation.newAggregation(searchStage);  

        return mongoTemplate.aggregate(agg, "notes", Note.class).getMappedResults();  
    }

And the controller endpoint:

    @GetMapping("/search/boost-title-tags")  
    public List<Note> searchBoostedTitleAndTags(@RequestParam String q) {  
        return searchRepository.searchBoostedTitleAndTags(q);  
    }

Search with autocomplete and fuzzy matching

The boosting lets us prioritise where the words we are searching for appear in our documents, but we're not quite there yet. Searching for exact matches has its limits. Let's bring in autocomplete and fuzzy search to round out our searching experience.

Autocomplete, sometimes called type ahead, lets us complete our search queries without the complete input string. We can use the autocomplete operator if we want our search bar to have a search-as-you-type component to predict words with increasing accuracy as characters are entered in our application's search field.

Fuzzy search brings in typo tolerance to our search mechanics. This way, even if we misspell our query, say Sprong AI, the search term will still try to use inexact matches.

    public List<Note> searchAutocompleteAndFuzzy(String query) {  
        AggregationOperation searchStage = _ -> new Document("$search",  
                new Document("index", "default")  
                        .append("compound", new Document("should", List.of(  
                                new Document("autocomplete", new Document()  
                                        .append("query", query)  
                                        .append("path", "title")  
                                ),  
                                new Document("text", new Document()  
                                        .append("query", query)  
                                        .append("path", "title")  
                                        .append("fuzzy", new Document("maxEdits", 2))  
                                )  
                        )))  
        );  

        AggregationOperation limitStage = Aggregation.limit(10);  

        Aggregation aggregation = Aggregation.newAggregation(searchStage, limitStage);  

        return mongoTemplate.aggregate(aggregation, "notes", Note.class).getMappedResults();  
    }

And the controller endpoint:

    @GetMapping("/search/autocomplete-fuzzy")  
    public List<Note> searchAutocompleteAndFuzzy(@RequestParam String q) {  
        return searchRepository.searchAutocompleteAndFuzzy(q);  
    }

Search with autocomplete, fuzzy matching, and boosting

So we've brought them in step by step, but what does this all look like together? In our NoteSearchRepository, we need to add a searchAutocompleteFuzzyBoosted(String query) like in the code below:

    public List<Note> searchAutocompleteFuzzyBoosted(String query) {
        AggregationOperation searchStage = _ -> new Document("$search",
                new Document("index", "default")
                        .append("compound", new Document("should", List.of(  
                                // Autocomplete on title (highest boost)  
                                new Document("autocomplete", new Document()
                                        .append("query", query)
                                        .append("path", "title")
                                        .append("score", new Document("boost", new Document("value", 6)))
                                ),
                                // Fuzzy text search on title
                                new Document("text", new Document()  
                                        .append("query", query)
                                        .append("path", "title")
                                        .append("fuzzy", new Document("maxEdits", 2))
                                        .append("score", new Document("boost", new Document("value", 5)))
                                ),  
                                // Fuzzy text search on tags  
                                new Document("text", new Document()  
                                        .append("query", query)  
                                        .append("path", "tags")  
                                        .append("fuzzy", new Document("maxEdits", 1))
                                        .append("score", new Document("boost", new Document("value", 3)))
                                ),  
                                // Fuzzy text search on content (no boost = default weight)  
                                new Document("text", new Document()  
                                        .append("query", query)  
                                        .append("path", "content")  
                                        .append("fuzzy", new Document("maxEdits", 1))  
                                )
                        )))
        );

        AggregationOperation sortByScore = Aggregation.sort(Sort.by(Sort.Order.desc("score")));
        AggregationOperation limitStage = Aggregation.limit(10);

        Aggregation aggregation = Aggregation.newAggregation(searchStage, sortByScore, limitStage);

        return mongoTemplate.aggregate(aggregation, "notes", Note.class).getMappedResults();
    }

And our last bit of code:

    @GetMapping("/search/autocomplete-fuzzy-boosted")
    public List<Note> searchAutocompleteFuzzyBoosted(@RequestParam String q) {
        return searchRepository.searchAutocompleteFuzzyBoosted(q);
    }

Testing our search

With all of our endpoints in place, it’s time to see them in action. Start your Spring Boot app:

mvn clean compile
mvn spring-boot:run

You can now test each search feature from your browser or using curl. For example:

Basic full-text search:

curl "http://localhost:8080/api/notes/search?q=Spring"

Boosted by title:

curl "http://localhost:8080/api/notes/search/boost-title?q=Spring"

Boosted by title and tags:

curl "http://localhost:8080/api/notes/search/boost-title-tags?q=Spring"

Autocomplete and fuzzy search (try misspelling the query):

curl "http://localhost:8080/api/notes/search/autocomplete-fuzzy?q=Sprong"

All-in-one (autocomplete + fuzzy + boosting):

curl "http://localhost:8080/api/notes/search/autocomplete-fuzzy-boosted?q=Sprong"

If you hit these endpoints from your browser (e.g., http://localhost:8080/api/notes/search?q=Spring Boot), you’ll see the raw JSON response, but this is exactly what a frontend search bar would consume.

Conclusion

What began as a neglected pile of half-thoughts is now a structured, searchable knowledge base, thanks to MongoDB Atlas Search and Spring Boot. We built a REST API that can handle everything from simple keyword queries to fuzzy, autocomplete-driven searches with custom boosting, making my notes far more discoverable.

By combining Atlas Search’s operators with Spring Boot’s simplicity, I’ve turned my Obsidian vault into a living wiki I can share online, without the manual hassle of a CMS. If you’ve ever wanted to make your notes or knowledge base truly searchable, this approach is simple to replicate.

If you found this useful, check out my other tutorials on MongoDB with Spring, like Spring AI and MongoDB: How to Build RAG Applications or Building a Real-Time Market Data Aggregator with Kafka and MongoDB. If you want to learn more about Search with MongoDB, check out this short skills course which covers everything you need to know.

Secure Local RAG with Role-Based Access: Spring AI, Ollama & MongoDB

Tim Kelly — Tue, 24 Jun 2025 11:30:00 +0000

Retrieval-augmented generation (RAG) gives our language models the power to fetch more relevant, timely, and informed responses. But here's the catch: What if you're not thrilled about shipping your highly sensitive company information off to some remote server farm in a distant country, just relying on a "promise" it won’t be used to train someone else's LLM?

Or maybe your organization is a bit more complex. Your legal team shouldn’t have access to the sales team's confidential data, and that overly curious intern definitely shouldn't be poking around executive-level information. In cases like these, you need RAG—but you need it secure, precise, and completely under your control.

In this tutorial, we’ll build a secure, locally-managed RAG solution that incorporates user-based access control. If you just want the code, check out the GitHub repo.

Prerequisites

Before we get started, there are a few things we'll need:

MongoDB Atlas account with a cluster set up (M0 free tier is perfect)
- If you need help getting started, check out our simple guides.
Docker installed
Java 17+ (I use 21)
Maven 3.9.6+

What is Ollama?

Ollama is an open-source tool that allows you to run LLMs locally on your machine. By running models locally, we can maintain full data ownership and avoid the potential security risks associated with cloud storage. Offline AI tools like Ollama are not only faster and more reliable, but also come with no token costs and no usage fees (basically free).

Setting up Ollama in Docker

Before we can actually run anything, we need to get Ollama up and running locally. This is what will power both our embedding model and chat model. We’ll keep things fully local so none of our documents ever leave our machine.

First, create a simple docker-compose.yml file:

version: '3.8'  

services:  
  ollama:  
    image: ollama/ollama  
    container_name: ollama  
    ports:  
      - "11434:11434"  
    volumes:  
      - ollama_data:/root/.ollama  
    restart: unless-stopped  
    environment:  
      - OLLAMA_HOST=0.0.0.0  

volumes:  
  ollama_data:

This will spin up Ollama inside Docker and map port 11434 to your local machine, which is what Spring AI will connect to later.

Now, let’s start everything up and pull the models we’ll be using:

docker compose up -d

With Ollama running, pull down the models:

# Pull the chat model
docker exec -it ollama ollama pull llama3.2

# Pull the embedding model
docker exec -it ollama ollama pull nomic-embed-text

# Verify models are installed
docker exec -it ollama ollama list

That’s it. We’re fully set up and ready to start wiring Ollama into Spring AI.

Getting our app

Before we start actually coding, we need to set up our dependencies and configure our application.properties. We'll build a simple little API to interact with our application, and while plenty of embedding and language models are out there, we'll stick with nomic-embed-text for embeddings and llama3.2 as our chat model.

Dependencies

To set up our project, the easiest approach is to use Spring Initializr. We will be selecting Spring Boot version 3.5.0 or later, Maven as our build tool, Java as the language, and setting the Java version to 21 with JAR packaging.

For dependencies, we'll need Spring Web to handle our REST API, MongoDB Atlas Vector Search for efficient retrieval, and Spring AI Ollama for integration with our embedding and language models.

After all that, we can generate our application, and open it in an IDE of our liking. Double check you have the necessary dependencies. You should see something like this in your pom.xml once Spring Initializr generates your project:

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-advisors-vector-store</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-starter-model-ollama</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-starter-vector-store-mongodb-atlas</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.ai</groupId>
            <artifactId>spring-ai-bom</artifactId>
            <version>${spring-ai.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

application.properties

Now, let’s configure our application.properties. This is where we wire everything together—MongoDB, Ollama, and a bit of Spring AI logging so we can peek under the hood while we build.

First, the basic Spring and MongoDB config:

spring.application.name=securerag

spring.data.mongodb.uri=${MONGODB_URI}
spring.data.mongodb.database=rag
server.port=8000

logging.level.org.springframework.ai=DEBUG
logging.level.org.springframework.data.mongodb=DEBUG

Next, we configure MongoDB Vector Search. This is what allows Spring AI to store and query our document embeddings inside MongoDB Atlas:

spring.ai.vectorstore.mongodb.collection-name=vector_store
spring.ai.vectorstore.mongodb.initialize-schema=true
spring.ai.vectorstore.mongodb.metadata-fields-to-filter=department,access_level,roles

Here, we’re telling Spring AI:

Which collection to use for storing our vector embeddings (vector_store).
Which metadata fields we’ll later want to use for filtering search results (department, access_level, and roles).
Whether or not to initialize the schema (initialize-schema=true). This simply means Spring AI will create the collection and index automatically if it doesn’t already exist, so we’re ready to run semantic search right out of the gate.

And finally, we wire up Ollama. This points Spring AI to the local Ollama instance we’ll be running, and tells it which models to use for embedding and chat:

spring.ai.ollama.base-url=http://localhost:11434
spring.ai.ollama.embedding.model=nomic-embed-text
spring.ai.ollama.chat.options.model=llama3.2

Since different models excel at different tasks, we’re using nomic-embed-text for generating our embeddings (this helps turn documents into vector representations), and llama3.2 for the actual chat responses. Using separate models here gives us better results on both sides—retrieval and generation.

The business logic

Now that we’ve got everything wired up, let’s look at how the actual chat flow works behind the scenes. This is where we combine retrieval, filtering, and our chat model into one secure conversation pipeline.

The `ChatService`

We’ll start by building a simple Spring @Service to contain our main logic:

package com.mongodb.securerag;  

import org.springframework.ai.chat.client.ChatClient;  
import org.springframework.ai.chat.client.advisor.vectorstore.QuestionAnswerAdvisor;  
import org.springframework.ai.chat.model.ChatModel;  
import org.springframework.ai.vectorstore.SearchRequest;  
import org.springframework.ai.vectorstore.VectorStore;  
import org.springframework.stereotype.Service;  

@Service  
public class ChatService {  

    private final ChatModel chatModel;  
    private final VectorStore vectorStore;  

    ChatService(ChatModel chatModel, VectorStore vectorStore) {  
        this.chatModel = chatModel;  
        this.vectorStore = vectorStore;  
    }

    // We'll add our business methods here next
}

Spring will automatically inject the ChatModel (which talks to Ollama) and the VectorStore (which talks to MongoDB Atlas Vector Search). This allows us to easily wire together search and chat in a clean service layer.

Sending secure messages

Now, let’s build the core method that handles chat requests, performs semantic search, and applies access control at query time. We’ll add the following to the ChatService class:

public String sendSecureMessage(String message, String userRole, String department) {  
    // Build the base QuestionAnswerAdvisor — this wires up vector search for retrieval
    QuestionAnswerAdvisor advisor = QuestionAnswerAdvisor.builder(vectorStore)  
            .searchRequest(SearchRequest.builder()  
                    .topK(5)  // Fetch top 5 most relevant documents
                    .build())  
            .build();  

    // Build the ChatClient using the advisor for retrieval-augmented context
    ChatClient filteredChatClient = ChatClient.builder(chatModel)  
            .defaultAdvisors(advisor)  
            .build();  

    // Dynamically build the filtering expression based on user role & department
    String filterExpression = createAccessFilterExpression(userRole, department);  

    // Attach the filter expression to this specific chat request
    return filteredChatClient.prompt()  
            .user(message)  
            .advisors(a -> a.param(QuestionAnswerAdvisor.FILTER_EXPRESSION, filterExpression))  
            .call()  
            .content();  
}

A few key things happen here:

We build a QuestionAnswerAdvisor that handles the vector search piece.
We dynamically generate our security filter per request, based on who’s asking.
Spring AI lets us inject this filter expression into each call to enforce access control at retrieval time.

Building the filter expression

Let’s generate that dynamic filter that decides which documents each user can see:

private String createAccessFilterExpression(String userRole, String department) {  
    List<String> conditions = new ArrayList<>();  

    // Public documents are always accessible
    conditions.add("access_level == 'public'");  

    // Add role-based conditions if the user has a specific role
    if (userRole != null && !userRole.equalsIgnoreCase("public")) {  
        StringBuilder roleCondition = new StringBuilder();  
        roleCondition.append("(roles in ['").append(userRole).append("', 'all_employees'])");  

        // Further restrict by department if provided
        if (department != null && !department.trim().isEmpty()) {  
            roleCondition.append(" && (department == '").append(department).append("')");  
        }  

        conditions.add(roleCondition.toString());  
    }  

    // Join everything together into one filter string
    return String.join(" || ", conditions);  
}

Quick summary:

If we’re public: We only get public docs.
If we have a role: We get both public docs and documents matching our role (and optionally, department).
This gives us fine-grained access control entirely at query time, no additional layers needed.

Under the hood, Spring AI takes this filter string and parses it using its built-in filter expression language. This syntax isn’t raw MongoDB query language, but rather a simple boolean expression format supported by Spring AI. Operators like ==, in, &&, ||, and parentheses are translated by Spring AI into proper MongoDB $vectorSearch filters when the query is executed. For example, our filter string:

access_level == 'public' || (roles in ['hr_manager', 'all_employees']) && (department == 'HR')

Would be automatically parsed into the filter expression:

{
  "$or": [
    { "access_level": { "$eq": "public" } },
    {
      "$and": [
        { "roles": { "$in": ["hr_manager", "all_employees"] } },
        { "department": { "$eq": "HR" } }
      ]
    }
  ]
}

The key is that Spring AI expects the filter string to follow this limited expression grammar. You cannot send full MongoDB queries or arbitrary text. As long as your string follows this format, Spring AI handles parsing and conversion automatically.

Creating an API

Alright, now that the brains of the system are built, we need a way to actually talk to it. Let’s wire up a super simple REST API so we can hit it with requests and see our secure RAG flow in action.

The request payload

We’ll create a plain old Java object to handle chat requests coming into our app. Nothing fancy, just a little DTO that carries the message, role, and department info along for the ride:

package com.mongodb.securerag;  

public class ChatRequest {  
    private String message;  
    private String userRole;  
    private String department;  

    public ChatRequest() {}  // default constructor needed for Spring to deserialize  

    public ChatRequest(String message, String userRole, String department) {  
        this.message = message;  
        this.userRole = userRole;  
        this.department = department;  
    }  

    public String getMessage() { return message; }  
    public void setMessage(String message) { this.message = message; }  

    public String getUserRole() { return userRole; }  
    public void setUserRole(String userRole) { this.userRole = userRole; }  

    public String getDepartment() { return department; }  
    public void setDepartment(String department) { this.department = department; }  
}

Basically:

message: what the user is asking
userRole: who’s asking
department: which department they're in, if relevant

Spring will automatically turn incoming JSON into this object for us. Easy.

The controller

This is where we expose our endpoints, just a small API surface so we can send chat requests to the back end:

package com.mongodb.securerag;  

import org.springframework.web.bind.annotation.*;  

@RestController  
@RequestMapping("/chat")  
public class ChatController {  

    private final ChatService chatService;  

    public ChatController(ChatService chatService) {  
        this.chatService = chatService;  
    }  

    @GetMapping("/{message}")  
    public String sendMessage(@PathVariable String message,  
                              @RequestParam(defaultValue = "public") String userRole,  
                              @RequestParam(required = false) String department) {  
        return chatService.sendSecureMessage(message, userRole, department);  
    }

    @PostMapping("/secure")  
    public String sendSecureMessage(@RequestBody ChatRequest request) {  
        return chatService.sendSecureMessage(request.getMessage(), request.getUserRole(), request.getDepartment());  
    }  
}

Here’s what’s going on:

The GET endpoint makes it easy to test quickly—just drop a URL in your browser or hit it with curl.
The POST endpoint gives us a bit more flexibility—we can send in the full request body with role and department data.
Both routes hand off to our ChatService, which handles the filtering logic and secure retrieval behind the scenes.

Perfect, here’s a little insert you can drop right after you explain the controller:

Quick note on GET vs POST

You might be wondering: Why do we have both a GET and a POST endpoint for this?

The GET endpoint is mainly here for convenience. It makes testing easier when you're still building. You can fire off quick curl commands or hit it directly from your browser to sanity check things.
The POST endpoint is better suited for real usage. It allows us to send the full chat request as JSON, which is much cleaner if you're dealing with longer queries, special characters, or calling the API from an actual front end or another system. It also gives us room to grow later if we want to pass in more fields.
Technically speaking, since this isn't a simple "read" operation (we're doing retrieval, filtering, embedding comparisons, and inference), many would argue this should be POST-only once you go into production.

For now, we’ll keep both while we’re building, but if you were turning this into a real product, you’d likely just expose the POST endpoint.

And that’s it. The API layer is done. You’ve now got a secure RAG system you can hit with real HTTP requests.

Load some sample data

Before we can test anything, we need some documents loaded into our vector store—otherwise, our model will just sit there politely answering nothing.

We’ll load up some simple HR, Finance, Sales, and Executive documents, each tagged with different access levels, roles, and departments. This will give us enough data to see our filtering in action.

Here’s our data loader:

package com.mongodb.securerag;

import org.springframework.ai.document.Document;
import org.springframework.ai.vectorstore.VectorStore;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
public class DocumentLoader implements CommandLineRunner {

    private final VectorStore vectorStore;

    public DocumentLoader(VectorStore vectorStore) {
        this.vectorStore = vectorStore;
    }

    @Override
    public void run(String... args) throws Exception {
        loadSampleDocuments();
    }

    public void loadSampleDocuments() {
        List<Document> documents = new ArrayList<>();

        // HR Documents
        documents.add(new Document(
                "Employee Handbook - Vacation Policy: All full-time employees are entitled to 15 days of paid vacation per year.",
                Map.of(
                        "department", "HR",
                        "access_level", "confidential",
                        "roles", Arrays.asList("hr_manager", "hr_coordinator"),
                        "title", "Vacation Policy"
                )
        ));

        documents.add(new Document(
                "Salary Review Process: Annual salary reviews are conducted in Q4 of each year.",
                Map.of(
                        "department", "HR",
                        "access_level", "restricted",
                        "roles", Arrays.asList("hr_manager", "executive"),
                        "title", "Salary Review Process"
                )
        ));

        // Finance Documents
        documents.add(new Document(
                "Q3 Budget Report: Total revenue for Q3 was $2.4M.",
                Map.of(
                        "department", "Finance",
                        "access_level", "confidential",
                        "roles", Arrays.asList("finance_manager", "cfo", "executive"),
                        "title", "Q3 Budget Report"
                )
        ));

        documents.add(new Document(
                "Expense Policy: All business expenses over $100 require manager approval.",
                Map.of(
                        "department", "Finance",
                        "access_level", "public",
                        "roles", Arrays.asList("all_employees"),
                        "title", "Expense Policy"
                )
        ));

        // Public Company Information
        documents.add(new Document(
                "Company Mission Statement: Our mission is to democratize access to artificial intelligence.",
                Map.of(
                        "department", "General",
                        "access_level", "public",
                        "roles", Arrays.asList("all_employees", "public"),
                        "title", "Company Mission Statement"
                )
        ));

        documents.add(new Document(
                "Office Hours and Location: Our main office is located at 123 Tech Street, San Francisco, CA.",
                Map.of(
                        "department", "General",
                        "access_level", "public",
                        "roles", Arrays.asList("all_employees", "public"),
                        "title", "Office Information"
                )
        ));

        // Sales Documents
        documents.add(new Document(
                "Q4 Sales Targets: Individual sales targets for Q4 are set at $500K per sales representative.",
                Map.of(
                        "department", "Sales",
                        "access_level", "confidential",
                        "roles", Arrays.asList("sales_rep", "sales_manager"),
                        "title", "Q4 Sales Targets"
                )
        ));

        // Executive Documents
        documents.add(new Document(
                "Strategic Planning 2025: Key initiatives include international expansion and AI model optimization.",
                Map.of(
                        "department", "Executive",
                        "access_level", "restricted",
                        "roles", Arrays.asList("ceo", "cto", "cfo", "executive"),
                        "title", "Strategic Planning 2025"
                )
        ));

        System.out.println("Loading " + documents.size() + " sample documents...");
        vectorStore.add(documents);
        System.out.println("Sample documents loaded successfully!");
    }
}

We inject the VectorStore directly into our loader, and on application startup (using CommandLineRunner), it loads all the documents automatically. Each document contains the main text plus metadata fields: access_level, roles, department, and title. This metadata structure maps directly to the filters we wrote earlier in createAccessFilterExpression().

This metadata is what we’ll later use to apply filtering logic during search. Public documents will always be visible to any user, while the restricted documents will only appear if the user has the appropriate role and department permissions.

That’s it! Our app will automatically load these into MongoDB Atlas when we start it up.

Testing it out

Now, let’s actually give this thing a spin. We can run it with:

export MONGODB_URI="YOUR_CONNECTION_STRING>"

mvn spring-boot:run

With our app running locally, we can hit the API directly using curl and see our secure retrieval logic in action. First, make sure to give it a moment to insert our sample data and create our embeddings.

Let’s try a simple public query where no role is provided. Anyone should be able to access this:

curl "http://localhost:8000/chat/What%20is%20our%20mission?userRole=public"

And voila!

Based on the provided context, the company's mission statement is:

"Our mission is to democratize access to artificial intelligence by building tools that make AI accessible to businesses of all sizes. We believe in ethical AI development and transparent business practices."

This appears to be a repeated statement, but it provides the core of the company's mission and values.

We get back our mission statement, as expected, pulled from the public data. The model also surfaces a bit of related context (office hours, expense policy, etc.), since it's retrieving semantically similar documents.

Next, let’s simulate an HR manager querying for vacation policy. This time, we pass both the role and department to unlock access to more restricted content:

curl "http://localhost:8000/chat/vacation%20policy?userRole=hr_manager&department=HR"

And we get back a fairly well informed response:

Based on the provided context, I see that there are multiple policies mentioned:

1. Vacation Policy: 15 days of paid vacation per year, with a maximum carryover of 5 unused days to the next year.

2. Expense Policy: All expenses over $100 require manager approval and must be submitted within 30 days of the expense date.

3. Office Hours and Location: The main office is located at 123 Tech Street, San Francisco, CA, with office hours from Monday to Friday, 9 AM to 6 PM Pacific Time.

If you could provide more context or clarify your question, I\'d be happy to try and assist further!

Now, we’re seeing both the vacation policy and other relevant internal documents that the HR manager is allowed to see. The access filter is working exactly as intended.

Finally, let’s check what happens if a public user asks the same vacation policy question:

curl "http://localhost:8000/chat/vacation%20policy?userRole=public"

As expected, the system correctly returns no private HR data, and limits the response to public information only.

I don't see any user comments or questions regarding a "vacation policy". The provided text only includes information about an expense policy, office hours and location, and company mission statement. Could you please provide more context or clarify what you would like to know about a vacation policy? I'll do my best to help.

Since this user isn’t authorized for confidential HR documents, the system correctly avoids leaking private data. Instead, the model only references public information and doesn’t return any HR policy details.

Conclusion

And we’ve built a fully functional, secure, locally-managed RAG system using Spring AI, Ollama, and MongoDB Atlas. The key here wasn’t just getting RAG working (plenty of tutorials stop there), but actually layering access control directly into the retrieval step itself. This way, we’re not handing the language model private data it was never supposed to see in the first place. We also don’t have to pay for token usage, so we aren’t paying for each request.

With this foundation, you can easily extend things further: add more roles, more metadata, more complex filters, richer documents, or swap in different models depending on your use case. Most importantly, you're in full control. No third-party model provider ever sees your sensitive documents, but you still get the power of retrieval-augmented LLM responses for your users.

Check out some of my other tutorials, like Building a Real-Time AI Fraud Detection System with Spring Kafka and MongoDB or Your Guide to Optimizing Slow Queries, and learn more about what you can do with MongoDB and Java.

Understanding BSON for Java Developers: A Beginner’s Guide to MongoDB’s Data Format

Tim Kelly — Thu, 12 Jun 2025 08:47:33 +0000

When working with MongoDB, it’s easy to think you’re dealing with JSON. After all, the queries, documents, and API responses all look like JSON. But MongoDB is not storing JSON. It’s storing BSON—a binary format designed for efficient storage and fast traversal.

BSON (Binary JSON) is more than just a binary version of JSON. It introduces additional data types like ObjectId, Decimal128, and Timestamp, allowing MongoDB to handle more complex data structures and ensure data integrity. While we might rarely interact with raw BSON directly, understanding how MongoDB stores and processes BSON documents can help us write more efficient queries, handle data conversions properly, and debug unexpected behavior.

In this guide, we’ll take a look at some of BSON’s key concepts, how it maps to Java types, and how we can work with BSON in the MongoDB Java driver. We’ll look at creating basic documents, working with nested structures, handling raw BSON, querying with BSON, and finally, mapping Java objects to BSON using POJOs.

By the end, you’ll have a clear understanding of how MongoDB leverages BSON under the hood and how you can work with it effectively in Java. Let’s get started. If you just want to check out the code, pop over to the GitHub repository.

What is BSON?

BSON stands for Binary JSON. It’s a binary-encoded serialization of JSON-like documents. It’s everything we like about JSON, just more efficient and type-rich—optimized for speed and storage in MongoDB.

While we interact with MongoDB using JSON-like queries, the documents themselves are stored and transmitted as BSON behind the scenes.

Why not just JSON?

JSON is pretty great. It's human-readable, flexible, and widely adopted. But it’s not ideal for databases. Here’s why MongoDB uses BSON instead:

Speed: BSON can be parsed faster than JSON.
Rich types: BSON supports additional data types like Date, Decimal128, and ObjectId.
Traversability: BSON includes length prefixes, making it easier for MongoDB to jump between fields during queries.

Here’s an example of what a document {“hello”: “world”} would look like in BSON, with length prefixes.

{"hello": "world"} →
\x16\x00\x00\x00           // total document size
\x02                       // 0x02 = type String
hello\x00                  // field name
\x06\x00\x00\x00world\x00  // field value
\x00                       // 0x00 = type EOO ('end of object')

>Note: A BSON document has a size limit of 16MB on MongoDB.

BSON vs. JSON

Feature	JSON	BSON
Format	Text-based	Binary
Readability	Human-readable	Machine-efficient
Data types	Limited (no dates, binary, etc.)	Rich and explicit (e.g., `ObjectId`, `Date`)
Speed	Slower to parse	Faster to parse
Size	Often smaller	Slightly larger due to type metadata

Common BSON data types (and their Java equivalents)

BSON type	Description	Java equivalent
String	UTF-8 string	`String`
Int32 / Int64	32-bit / 64-bit integers	`int`, `long`
Double	64-bit float	`double`
Boolean	`true` / `false`	`boolean`
Date	Epoch millis	`java.util.Date`
ObjectId	12-byte unique identifier	`org.bson.types.ObjectId`
Binary	Byte array	`byte[]`
Document	Embedded object	`org.bson.Document`
Array	List of values	`List<?>`

BSON and MongoDB internals

BSON is how MongoDB stores documents on disk.
BSON is how MongoDB communicates between client and server.
Indexes, metadata, and replication all operate on BSON.

Our Java driver handles the BSON encoding and decoding transparently. But if we're building performance-sensitive applications or exploring custom serialization, or we’re even just curious, it's worth understanding.

Setup and project structure

In order to follow along with the code, make sure you have Java 24 and Maven installed. You will also need a MongoDB cluster set up. A MongoDB M0 free-forever tier is perfect for this.

Project structure:

/src
  └── main
      └── java
          └── com
              └── mongodb
                  ├── Main.java
                  ├── User.java
                  └── Address.java
pom.xml

Maven dependency (pom.xml):

<dependencies>
    <dependency>  
        <groupId>org.mongodb</groupId>  
        <artifactId>mongodb-driver-sync</artifactId>  
        <version>5.4.0</version>  
    </dependency>
</dependencies>

BSON data types and document creation

In MongoDB’s Java driver, we can interact directly with BSON types using classes like BsonString, BsonInt32, and BsonObjectId. However, we rarely do this. Instead, we work with standard Java types, and MongoDB automatically handles the conversion to BSON.

Here’s a look at BSON-specific types:

BsonString bsonString = new BsonString("Hello BSON");
BsonInt32 bsonInt32 = new BsonInt32(42);
BsonInt64 bsonInt64 = new BsonInt64(9876543210L);
BsonDecimal128 bsonDecimal = new BsonDecimal128(new Decimal128(new BigDecimal("12345.678")));
BsonDateTime bsonDate = new BsonDateTime(new Date().getTime());
BsonBinary bsonBinary = new BsonBinary("binary data".getBytes());
BsonObjectId bsonObjectId = new BsonObjectId(new ObjectId());
BsonTimestamp bsonTimestamp = new BsonTimestamp();

In practice, we work with familiar Java types. The driver converts String, int, Date, and other common types to their BSON equivalents behind the scenes.

For example, when we call new Date() in Java, MongoDB stores it as a BSON Date type, represented as milliseconds since the epoch:

Document doc = new Document("created", new Date()); collection.insertOne(doc);

Internally, MongoDB stores it like this:

{ "created": { "$date": "2025-05-09T12:34:56.789Z" }

Understanding this conversion helps when we’re debugging data types or working with MongoDB tools that expose BSON representations.

When interacting with MongoDB, we typically work with the Document class rather than BSON-specific types. The Document class represents a BSON document and allows us to structure data using standard Java types.

Here’s how we create a basic document:

private static void createBasicDocument() {
    System.out.println("\n--- Basic Document Creation ---");

    Document doc = new Document("name", "John Doe")
            .append("age", 30)
            .append("isMember", true)
            .append("joined", new Date())
            .append("_id", new ObjectId());

    collection.insertOne(doc);
    System.out.println("Inserted Document: " + doc.toJson());
}

In this method, we create a Document object with standard Java types: a String, an int, a boolean, a Date, and an ObjectId. The MongoDB driver automatically converts these to BSON types when the document is inserted into the collection.

Inserted Document: {"_id": "64e99b0b4321a1b9a6e4d8c7", "name": "John Doe", "age": 30, "isMember": true, "joined": "2025-05-09T12:34:56.789Z"}

The ObjectId is generated automatically, if not provided. The Date is converted to a BSON Date type, stored as milliseconds since the epoch.

Nested fields and arrays

MongoDB allows for nested structures and arrays, making it easy to represent complex data within a single document. This structure gives us more flexibility in how we want to model our data than traditional relational databases, where data would typically be spread across multiple tables. In MongoDB, we can embed related data directly within the document, creating hierarchical structures that are easy to query and manipulate.

In Java, we use the Document class to create nested structures and arrays. Each nested level is represented by another Document, and arrays are represented by Java List objects. When we work with nested documents and arrays in MongoDB, each level of nesting is still BSON, not JSON. This matters because MongoDB uses BSON-specific types like Decimal128 and Date, even in nested structures:

private static void createNestedDocument() {
    System.out.println("\n--- Nested Fields and Arrays ---");

    Document nestedDoc = new Document("user", "Alice")
            .append("balance", new Decimal128(new BigDecimal("12345.67")))
            .append("contacts", Arrays.asList("123-456-7890", "987-654-3210"))
            .append("address", new Document("city", "Dublin").append("postalCode", "D02"))
            .append("tags", Arrays.asList("premium", "verified"))
            .append("activity", new Document("login", new Date()).append("status", "active"));

    collection.insertOne(nestedDoc);
    System.out.println("Inserted Nested Document: " + nestedDoc.toJson());
}

In this method, we are creating a document that represents a user with several fields:

"user" is a simple string field.
"balance" is a Decimal128 value, which is used for financial calculations to prevent precision loss.
"contacts" is a list of strings, representing multiple contact numbers.
"address" is a nested document containing "city" and "postalCode".
"tags" is an array of strings, useful for categorizing or labeling documents.
"activity" is another nested document containing a login timestamp and a status field.

When the above method is executed, the document is converted to BSON and inserted into the MongoDB collection. The output of the inserted document will look like this:

Inserted Nested Document: {
  "user": "Alice",
  "balance": 12345.67,
  "contacts": ["123-456-7890", "987-654-3210"],
  "address": {
    "city": "Dublin",
    "postalCode": "D02"
  },
  "tags": ["premium", "verified"],
  "activity": {
    "login": "2025-05-09T12:34:56.789Z",
    "status": "active"
  }
}

The nested structure is straightforward to read and query. Each level of nesting is a separate Document object, and arrays are automatically converted to BSON arrays.

Why use nested structures?

Nested documents provide a way to keep related data together, minimizing the number of queries needed to access complete data sets. Instead of joining tables, we can query a single document to retrieve user details, address information, and recent activity. This approach is particularly useful when dealing with hierarchical data, embedded lists, or object relationships that are tightly coupled.

Raw BSON manipulation

So far, we’ve relied on the Document class to handle BSON conversion. But what if we need direct control over BSON structure? That’s where raw BSON manipulation comes in. MongoDB’s Java driver provides a higher-level Document class that abstracts away BSON specifics, allowing us to work with Java types like String, int, and Date. However, sometimes, we may need to interact directly with BSON data. This can be useful when dealing with binary data, timestamps, or performing low-level optimizations.

Binary data: BSON allows for binary storage (BsonBinary). This is useful for handling images, files, or encrypted data.
Timestamps: BSON BsonTimestamp includes both a time and an increment value, making it useful for tracking operations in oplogs or distributed systems.

The BsonDocument class provides a more granular way to construct BSON documents using BSON-specific types such as BsonString, BsonInt32, and BsonBinary. Unlike the Document class, BsonDocument requires explicit type declarations for each field, making it more verbose but also more explicit.

The following method constructs a BSON document directly using BSON-specific classes:

private static void demonstrateRawBSON() {
    System.out.println("\n--- Raw BSON Manipulation ---");

    BsonDocument rawBson = new BsonDocument()
            .append("title", new BsonString("Raw BSON Example"))
            .append("value", new BsonInt32(100))
            .append("binaryData", new BsonBinary("raw data".getBytes()))
            .append("timestamp", new BsonTimestamp())
            .append("array", new BsonArray(Arrays.asList(
                    new BsonInt32(1), 
                    new BsonInt32(2), 
                    new BsonInt32(3)
            )));

    System.out.println("Raw BSON: " + rawBson.toJson());
}

In this example, we create a BSON document using explicit BSON classes:

"title" is a BsonString, representing a UTF-8 string.
"value" is a BsonInt32, a 32-bit integer.
"binaryData" is a BsonBinary, representing raw byte data as a base64-encoded string.
"timestamp" is a BsonTimestamp, containing both a Unix timestamp and an increment counter.
"array" is a BsonArray, holding multiple BsonInt32 values.

Each append() call explicitly defines the BSON type, making this method more verbose than using the Document class but also more precise.

Raw BSON: {
  "title": "Raw BSON Example",
  "value": 100,
  "binaryData": {
    "$binary": "cmF3IGRhdGE=",
    "$type": "00"
  },
  "timestamp": {
    "$timestamp": {
      "t": 1650034567,
      "i": 1
    }
  },
  "array": [1, 2, 3]
}

The binaryData field is represented as a base64-encoded string with a type identifier (00 for generic binary data). The timestamp field includes both a time value (t) and an increment (i), useful for replication and internal operations.

Direct BSON manipulation is not typically necessary for most MongoDB operations. For most use cases, the Document class is sufficient, and a lot more intuitive. The BsonDocument class is there when we need more precise control over BSON data or when working with advanced MongoDB features like oplog processing or custom serialization.

Querying with BSON

When querying MongoDB, we typically use the Filters class.The Filters class provides static factory methods for all the MongoDB query operators. Each method returns an instance of the BSON type, which we can pass to any method that expects a query filter. These filters work with BSON data but allow us to write queries using Java types and let the MongoDB driver handle the conversion to BSON.

Let’s take a look at a simple equality filter. The following method demonstrates how to query for specific documents using Filters.eq() and Filters.exists():

private static void queryWithBSON() {
    System.out.println("\n--- Querying with BSON ---");

    // Find the first document where the "user" field is "Alice"
    Document result = collection.find(Filters.eq("user", "Alice")).first();
    if (result != null) {
        System.out.println("Found Document: " + result.toJson());
    }

    // Find all documents that contain the "activity" field
    List<Document> results = collection.find(Filters.exists("activity")).into(new ArrayList<>());
    System.out.println("Documents with 'activity' field:");
    results.forEach(doc -> System.out.println(doc.toJson()));
}

The Filters.eq() method creates a simple equality query, matching documents where the user field is "Alice". This query is not searching for a string, but for a BSON BsonString. Similarly, when querying for dates, BSON expects a Date type, not a string. This query is similar to the following MongoDB query in the shell:

db.collection.find({ "user": "Alice" })

The Filters.exists() method finds documents that contain a specific field, regardless of its value. In this case, we are searching for all documents that have the "activity" field. This query is equivalent to:

db.collection.find({ "activity": { $exists: true } })

If a document with "user": "Alice" exists, the output will look something like this:

--- Querying with BSON ---
Found Document: {
  "_id": "64e99b0b4321a1b9a6e4d8c7",
  "user": "Alice",
  "balance": 12345.67,
  "contacts": ["123-456-7890", "987-654-3210"],
  "address": {
    "city": "Dublin",
    "postalCode": "D02"
  },
  "tags": ["premium", "verified"],
  "activity": {
    "login": "2025-05-09T12:34:56.789Z",
    "status": "active"
  }
}

If there are multiple documents containing the "activity" field, each will be printed as a separate JSON object.

The Filters class provides a range of query operators, allowing us to construct complex queries using methods like eq(), exists(), gt(), lt(), and in(). These methods allow us to write type-safe queries without dealing directly with BSON objects.

This approach keeps the syntax concise and consistent, making the most of the Java types while MongoDB handles the BSON conversion automatically.

Aggregation with BSON

MongoDB’s aggregation framework allows for complex data processing, transforming documents in a collection through a series of stages like filtering, grouping, and projecting. While we usually interact with MongoDB through Java types like String or int, the aggregation framework operates directly on BSON data, making it important for us to understand how BSON types are handled in aggregation operations.

In the Java driver, we construct aggregation pipelines using the Aggregates class, where each stage is represented as a BSON operation. Each stage in the pipeline processes BSON documents, applying transformations and producing a new BSON structure for the next stage.

Each stage processes BSON data directly. MongoDB maintains BSON data types throughout the pipeline, preventing data loss and ensuring accuracy in operations like $sum and $avg.

For instance, if we aggregate on a Decimal128 field, the aggregation framework maintains the precision:

List<Bson> pipeline = List.of(     Aggregates.group("$user", Accumulators.sum("totalBalance", "$balance")) );

If balance is stored as a Decimal128, the aggregation framework sums it as a Decimal128. This is crucial for financial calculations where precision matters.

In this example, we will build an aggregation pipeline that:

Filters documents to include only those with a balance field.
Groups documents by the user field, calculating the sum of all balance values for each user.
Projects the output to include the user and the calculated totalBalance.

private static void aggregateWithBSON() {
    System.out.println("\n--- Aggregation with BSON ---");

    List<Bson> pipeline = List.of(
        Aggregates.match(Filters.exists("balance")),
        Aggregates.group("$user", Accumulators.sum("totalBalance", "$balance")),
        Aggregates.project(new Document("user", "$_id").append("totalBalance", 1))
    );

    List<Document> results = collection.aggregate(pipeline).into(new ArrayList<>());
    System.out.println("Aggregation Results:");
    results.forEach(doc -> System.out.println(doc.toJson()));
}

Match stage: The first stage filters documents based on the existence of the balance field:

Aggregates.match(Filters.exists("balance"))

This generates a BSON structure similar to:

{ "$match": { "balance": { "$exists": true } } }

The Filters.exists() method constructs a BSON document using the $exists operator, targeting BSON fields directly. If balance is a Decimal128 type, it remains a Decimal128 throughout the pipeline, maintaining precision.

Group stage: The group stage aggregates documents by a specified field. In BSON, the _id field represents the grouping key. Here, we use the user field as the key and calculate the sum of the balance field:

Aggregates.group("$user", Accumulators.sum("totalBalance", "$balance"))

The resulting BSON structure:

{
  "$group": {
    "_id": "$user",
    "totalBalance": { "$sum": "$balance" }
  }
}

In this stage, the key (_id) is defined as the user field. The balance field is aggregated using the $sum accumulator, and the result is stored in a new BSON field called totalBalance.

Project stage: In the project stage, we transform the structure of the BSON document, selecting specific fields and renaming them as needed:

Aggregates.project(new Document("user", "$_id").append("totalBalance", 1))

The resulting BSON structure:

{
  "$project": {
    "user": "$_id",
    "totalBalance": 1
  }
}

This operation renames the _id field to user and includes the totalBalance field in the output. Notice that _id is no longer a BSON ObjectId but a value derived from the group key, in this case, a String.

If the collection contains the following documents:

{ "user": "Alice", "balance": 5000 }
{ "user": "Alice", "balance": 3000 }
{ "user": "Bob", "balance": 7000 }
{ "user": "Alice", "balance": 2500 }

The output of the aggregation pipeline will look like this:

Aggregation Results:
{"user": "Alice", "totalBalance": 10500}
{"user": "Bob", "totalBalance": 7000}

Each document in the output is a BSON object resulting from the aggregation pipeline. The user field is derived from the grouping key, and totalBalance is the calculated sum of all balance values per user.

POJO mapping: Bridging Java and BSON

As we’ve seen, BSON is the native data format for storing documents in MongoDB. While BSON extends JSON with additional data types, Java developers typically don’t need to work directly with raw BSON. Instead, we can work with familiar Java objects and let the MongoDB driver handle the BSON conversion behind the scenes. This is where POJO (Plain Old Java Object) mapping comes into play.

POJOs are often used for data encapsulation, which is the practice of separating business logic from data representation. If you want a deep understanding of POJO mapping with MongoDB, check out our guide, but we will go over the basics and get up and running with a simple example.

The PojoCodecProvider allows MongoDB to automatically map Java objects to BSON documents and back. This not only simplifies data handling but also keeps our data model consistent with our Java classes.

Why POJO mapping?

Without POJO mapping, we would need to manually convert Java objects into Document objects and vice versa. This is error-prone and can quickly become cumbersome as our data model grows more complex.

POJO mapping abstracts away the BSON conversion process. We define our Java classes, and the MongoDB driver handles the rest.

Setting up POJO mapping in Java

Before we define our Java classes, we need to configure the PojoCodecProvider. This codec provider registers our Java classes for automatic BSON mapping.

import org.bson.codecs.configuration.CodecProvider;  
import org.bson.codecs.configuration.CodecRegistry;  
import org.bson.codecs.pojo.PojoCodecProvider;  

import static org.bson.codecs.configuration.CodecRegistries.fromProviders;  
import static org.bson.codecs.configuration.CodecRegistries.fromRegistries;  
import static com.mongodb.MongoClientSettings.getDefaultCodecRegistry;  

public class CodecSetup {  

    public static CodecRegistry getPojoCodecRegistry() {  
        CodecProvider pojoCodecProvider = PojoCodecProvider.builder()  
                .automatic(true)
                .build();  

        return fromRegistries(getDefaultCodecRegistry(), fromProviders(pojoCodecProvider));  
    }  
}

The codec registry combines the default BSON codecs with our custom POJO codecs.

Defining a POJO class: User

Now, let’s define a simple User class that MongoDB will automatically map to BSON.

package com.mongodb;  

import org.bson.types.ObjectId;  
import org.bson.codecs.pojo.annotations.BsonId;  
import org.bson.codecs.pojo.annotations.BsonProperty;  

public class User {  

    @BsonId  
    private ObjectId id;  

    @BsonProperty("username")  
    private String name;  

    private int age;  

    @BsonProperty("member")  
    private boolean isMember;  

    public User() {  
        // No-arg constructor required for POJO mapping  
    }  

    public User(String name, int age, boolean isMember) {  
        this.name = name;  
        this.age = age;  
        this.isMember = isMember;  
    }  

    public ObjectId getId() {  
        return id;  
    }  

    public void setId(ObjectId id) {  
        this.id = id;  
    }  

    public String getName() {  
        return name;  
    }  

    public void setName(String name) {  
        this.name = name;  
    }  

    public int getAge() {  
        return age;  
    }  

    public void setAge(int age) {  
        this.age = age;  
    }  

    public boolean isMember() {  
        return isMember;  
    }  

    public void setMember(boolean isMember) {  
        this.isMember = isMember;  
    }  

    @Override  
    public String toString() {  
        return "User [id=" + id + ", username=" + name + ", age=" + age + ", member=" + isMember + "]";  
    }  
}

@BsonId: Marks the id field as the BSON _id field
@BsonProperty: Maps the name field to the BSON key username, and isMember to member

Inserting and querying POJOs

Now that we have our codec set up and our POJO classes defined, let’s see how we can insert and query these objects.


// ...
private static MongoCollection<User> userCollection;  

public static void main(String[] args) {  

    // ...
    CodecRegistry codecRegistry = CodecSetup.getPojoCodecRegistry();  

    try (MongoClient client = MongoClients.create(CONNECTION_STRING)) {    

        // ...

        userCollection = database.getCollection("users", User.class).withCodecRegistry(codecRegistry);  
        demonstratePojoMapping();  
    }  
}

private static void demonstratePojoMapping() {  
    System.out.println("\n--- POJO Mapping ---");  

    User user = new User("John Doe", 30, true);  
    userCollection.insertOne(user);  
    System.out.println("Inserted User: " + user);  

    List<User> users = new ArrayList<>();  
    userCollection.find().into(users);  
    System.out.println("Retrieved Users: " + users);  
}

First, we make sure we register our codec. We then connect to a User collection with this codec.
When the User object is inserted, MongoDB will automatically convert it to a BSON document.
When querying, MongoDB will deserialize the BSON back into a User object, maintaining type integrity.

BSON representation of the user document

When the User object is inserted, MongoDB stores it as a BSON document. The BSON representation will look like this:

{
    "_id":{ "$oid":"68211b91221c1ce15490b565" },
    "age":{ "$numberInt":"30" },
    "member":true,
    "username":"John Doe"
}

Notice that:

The id field is mapped to the BSON _id field.
The name field is renamed to username.

Even though we’re working with plain Java objects, MongoDB is still converting these to BSON. This automatic conversion ensures that data types are preserved when stored in MongoDB.

Why use POJO mapping?

Cleaner code: POJO mapping eliminates the need to manually convert Java objects to BSON Document objects, reducing boilerplate code.
Type safety: MongoDB automatically handles type conversions, ensuring that data is deserialized to the correct Java type (e.g., ObjectId, Date).
Nested structures: Complex nested objects are easily represented using embedded BSON documents, maintaining data structure and hierarchy.

What about custom conversions?

The default POJO mapping behavior is plenty sufficient for most use cases, but MongoDB also provides options for advanced customization. We can define custom codecs, register additional conventions, or handle abstract types and enums using advanced configuration. For more advanced scenarios, check out our PojoCodecProvider documentation.

Conclusion

BSON is at the core of how MongoDB stores and transmits data. It extends JSON with additional data types like ObjectId, Decimal128, and Timestamp, allowing MongoDB to handle richer and more complex data structures. While BSON is a binary format optimized for storage and traversal, the Java driver abstracts most of its complexity, allowing developers to work with familiar Java types.

Throughout this tutorial, we explored how BSON types map to Java types and how we can interact with BSON data using the Document class and the PojoCodecProvider. We saw how nested structures, arrays, and raw BSON objects are constructed and manipulated, and how aggregation pipelines process BSON data directly within the database.

While most operations in MongoDB can be performed using Java objects and the Document class, understanding BSON is essential for tasks involving binary data, precision calculations with Decimal128, and operations that require explicit data types like BsonTimestamp.

In every operation—document creation, querying, aggregation—BSON is working behind the scenes, ensuring that data types are consistent, optimized, and capable of handling complex structures. By understanding how Java types map to BSON, we can write more predictable queries, prevent data type mismatches, and take full advantage of MongoDB’s type system. For a deeper dive, explore the BSON Specification.

Spring AI and MongoDB: How to Build RAG Applications

Tim Kelly — Tue, 20 May 2025 11:01:09 +0000

Co-authored with Josh Long, Spring Developer Advocate on the Spring team and Youtuber at YouTube.com/@coffeesoftware.

Spring AI 1.0 is now available after a significant development period influenced by rapid advancements in the AI field. Spring AI is a new project from the Spring team that gives you structured, declarative access to large language models, vector search, image generation, and more—directly from your Spring Boot apps. The release includes numerous essential new features for AI engineers.

Java and Spring are in a prime spot to jump on this whole AI wave. Tons of companies are running their stuff on Spring Boot, which makes it super easy to plug AI into what they're already doing. You can basically link up your business logic and data right to those AI models without too much hassle.

If you just want the code for this tutorial, check out our GitHub repository.

What is Spring AI?

Spring AI provides support for various AI models and technologies. Image models can generate images given text prompts. Transcription models can take audio and convert them to text. Embedding models convert arbitrary data into vectors, which are data types optimized for semantic similarity search. Chat models should be familiar! You’ve no doubt even had a brief conversation with one somewhere. Chat models are where most of the fanfare seems to be in the AI space. They're awesome. You can get them to help you correct a document or write a poem. (Just don’t ask them to tell a joke… yet.) They’re awesome but they have some issues.

(The picture shown is used with permission from the Spring AI team lead Dr. Mark Pollack)

AI models are powerful but inherently limited in a few key areas. Spring AI addresses these limitations with targeted solutions:

Chat models and system prompts: Chat models are open-ended and prone to distraction. To guide their responses, you provide a system prompt—a structured directive that sets the tone and context for all interactions.
Memory management: AI models have no inherent memory. To maintain context across interactions, you need to implement memory mechanisms that correlate past and current messages.
Tool calling: Models operate in isolated environments. However, with tool calling, you can expose specific functions that the model can invoke when necessary, allowing for multi-turn interactions without custom orchestration logic.
Data access and prompt engineering: Models don’t have access to your proprietary databases—nor would you want them to by default. Instead, you can inject context by appending relevant information to the prompt. But there’s a limit to how much data you can include.
Retrieval-augmented generation (RAG): Instead of stuffing the prompt with data, use a vector store like MongoDB Atlas to retrieve only the most relevant information. This targeted approach reduces complexity and response size, enhancing both accuracy and cost-efficiency.
Response validation: AI models can confidently produce incorrect information. Evaluation models can validate or cross-check responses to mitigate these hallucinations.
Integration and workflows with MCP: No AI system functions in isolation. With Model Context Protocol (MCP), you can connect your Spring AI applications to other MCP-enabled services, regardless of language or platform, creating structured, goal-driven workflows.

You can leverage familiar Spring Boot conventions with Spring AI’s streamlined autoconfigurations, accessible through Spring Initializr. Spring AI provides convenient Spring Boot autoconfigurations that give you the convention-over-configuration setup you’ve come to know and expect. And Spring AI supports observability with Spring Boot’s Actuator and the Micrometer project. It plays well with GraalVM and virtual threads, too, allowing you to build super fast and efficient AI applications that scale.

Integrating Spring AI with MongoDB

Whether you're new to MongoDB or already building apps with it, MongoDB offers a modern, flexible foundation for AI-driven development. As a document database, MongoDB stores data in a JSON-like format, making it a natural fit for handling unstructured and semi-structured data—like user input, chat messages, or documents.

With Atlas Vector Search, MongoDB extends this flexibility by letting you store and search vector embeddings alongside your application data—all in the same place. Vector embeddings are just numerical representations of your data. For example, a product description or support ticket might be turned into a 512-dimensional vector—a list of numbers capturing its semantic meaning. This simplifies your architecture: no need for a separate vector database, custom pipelines, or syncing between systems.

If you're using Java, the MongoDB Java Sync Driver and Spring AI make it easy to integrate semantic search and LLM-based features directly into your application.

What is retrieval-augmented generation (RAG)?

Retrieval-augmented generation is a powerful pattern for improving the quality of large language model (LLM) responses. It works by pairing a language model with a document retrieval system—like MongoDB Atlas Vector Search—to ground the model's output in real, up-to-date information.

Why use RAG? Working with LLMs alone comes with limitations:

Outdated knowledge: LLMs are trained on static snapshots of the internet and may not know about recent developments.
Lack of context: They can’t access your private or domain-specific data unless you provide it.
Hallucinations: Without access to authoritative context, LLMs may generate plausible-sounding but incorrect information.

RAG addresses these gaps by introducing real-world context into the generation process:

Ingest: Convert your internal documents or domain-specific data into vector embeddings and store them in a vector store like MongoDB Atlas.
Retrieve: When a user asks a question, use MongoDB Atlas Vector Search to find semantically relevant documents based on the meaning of the query—not just keyword matching.
Generate: Feed the retrieved documents to the LLM as context, so it can generate accurate, relevant responses.

This architecture is great for building chatbots, document Q&A systems, or internal assistants that can reason over your organization's knowledge, not just what the model was originally trained on.

By combining Spring AI’s APIs with MongoDB Atlas Vector Search, you can build intelligent, production-ready applications without leaving the Java ecosystem. Let’s start wiring up your first RAG-powered AI app.

Here’s a diagram outlining how our app will work. The user sends a question to the chat endpoint. The question is embedded using our AI model, and this embedding is then used to retrieve relevant documents in our database using vector search. These documents, as well as system prompt and optional chat memory, are applied to maintain context. These documents are injected into the prompt, and the LLM generates a context-aware response, which is then returned to the user.

Building the application: Meet the dogs

We're building a fictitious dog adoption agency called Pooch Palace. It's like a shelter where you can find and adopt dogs online! And just like most shelters, people are going to want to talk to somebody, to interview the dogs. We're going to build a service to facilitate that: an assistant.

Our goal is to build an assistant to help us find the dog of (somebody's!) dream, Prancer, who is described rather hilariously as “a demonic, neurotic, man-hating, animal-hating, children-hating dog that looks like a gremlin.” This dog’s awesome. You might’ve heard about him. He went viral a few years ago when his owner was looking to find a new home for him. The ad was hysterical! Here’s the original post, in Buzzfeed News, in USA Today, and in the New York Times.

We’re going to build a simple HTTP endpoint that will use the Spring AI integration with an LLM. In this case, we’ll use Open AI, though you can use anything you’d like, including Ollama, Amazon Bedrock, Google Gemini, HuggingFace, and scores of others. It’s all supported through Spring AI to ask the AI model to help us find the dog of our dreams by analysing our question and deciding—after looking at the dogs in the shelter (and in our database)—which might be the best match for us.

Getting started: Project setup

To follow along, you’ll need Java version 24 and Maven on your machine ready to run. You’ll also need a MongoDB Atlas cluster set up (the M0 free tier will work), and an OpenAI account with an API key ready to go.

Go to Spring Initializr, create a new project with the artifact ID mongodb, and add the following dependencies:

Web
Actuator
GraalVM
Devtools
OpenAI
MongoDB Atlas Vector Database

We’re using Java 24 and Apache Maven, but you can adjust as needed. Download the .zip file and open it in your preferred IDE. We need to put the following properties into our application.properties:

spring.application.name=mongodb

# ai
spring.ai.openai.api-key=<YOUR_OPENAI_CREDENTIAL_HERE>

# mongodb 
spring.data.mongodb.uri=<YOUR_MONGODB_ATLAS_CONNECTION_STRING_HERE>
spring.data.mongodb.database=rag

# mongodb atlas vector store
spring.ai.vectorstore.mongodb.collection-name=vector_store
spring.ai.vectorstore.mongodb.initialize-schema=true

# scale
spring.threads.virtual.enabled=true

# health
management.endpoints.web.exposure.include=*
management.endpoint.health.show-details=always

Let’s walk through some of the configuration values. The ai section defines an OpenAI credential. You can and should externalize this as an environment variable, like this: SPRING_AI_OPENAI_API_KEY. This way, you never risk accidentally checking in the credentials to your GitHub repository.

In the mongodb section, we've got the usual bread-and-butter connectivity details. Again, you’ll get this from your MongoDB Atlas account and you’d do well to externalize it into an environment variable.

Then, in the mongodb atlas vector store section, we tell MongoDB Atlas Vector Search to initialize a collection called vector_store for us. Setting initialize-schema to true tells Spring AI to create the vector search index on our collection.

In the scale section, we tell Spring Boot to configure Java’s vaunted virtual threads. We’ll revisit this later.

In the health section, we tell Spring Boot’s Actuator to expose some details around the state of the application. We’ll revisit this later.

Initializing the vector store

Now, let’s get some code going. In the main class, called MongodbApplication.java in our example, we'll initialize the collection by reading some data from a .json file into the MongDB Atlas Vector Store. This .json file is available in the GitHub repo. We need to download it and add it to the resources folder in our application with the name dogs.json.

Now, we’ll open our MongodbApplication class, and add the necessary hints to read in our new resource.

@ImportRuntimeHints(MongoApplication.Hints.class)
@SpringBootApplication
public class MongodbApplication {

    public static void main(String[] args) {
        SpringApplication.run(MongoApplication.class, args);
    }

    static final Resource DOGS_JSON_FILE = new ClassPathResource("/dogs.json");

    static class Hints implements RuntimeHintsRegistrar {

        @Override
        public void registerHints(RuntimeHints hints, ClassLoader classLoader) {
            hints.resources().registerResource(DOGS_JSON_FILE);
        }
    }

    // TBD
}

And now, we need to add this data to our vector store. Add the following code to the MongodbApplication class, and we’ll take a look at what it does:

@Bean
ApplicationRunner mongoDbInitialzier(MongoTemplate template,
                                         VectorStore vectorStore,
                                         @Value("${spring.ai.vectorstore.mongodb.collection-name}") String collectionName,
                                         ObjectMapper objectMapper) {
        return args -> {

            if (template.collectionExists(collectionName) && template.estimatedCount(collectionName) > 0)
                return;

            var documentData = DOGS_JSON_FILE.getContentAsString(Charset.defaultCharset());
            var jsonNode = objectMapper.readTree(documentData);
            jsonNode.spliterator().forEachRemaining(jsonNode1 -> {
                var id = jsonNode1.get("id").intValue();
                var name = jsonNode1.get("name").textValue();
                var description = jsonNode1.get("description").textValue();
                var dogument = new Document("id: %s, name: %s, description: %s".formatted(
                        id, name, description
                ));
                vectorStore.add(List.of(dogument));
            });
        };
 }

This runner ingests data from the dogs.json file on the classpath and writes it to MongoDB Atlas using the Spring AI VectorStore interface. Each record is written as a simple string—id: %s, name: %s, description: %s. The format doesn’t matter as long as it’s consistent. This is what we’ll be using for generating our embedding, so make sure it includes all the data you want to reference.

The VectorStore handles both embedding generation and storage in one step. It uses an OpenAI-backed EmbeddingModel behind the scenes to convert each string into a vector, then stores that vector in MongoDB Atlas.

If the target collection doesn’t already exist, Spring AI will create it and automatically define a vector index on the embedding field. This means the collection and its index are ready for similarity search immediately after ingestion, with no manual setup required.

Building the assistant controller

Now, the easy part: building the assistant! We’ll define the AdoptionController. This is a simple Spring MVC controller that responds to requests and interacts with the AI model to recommend dogs based on user input.


@Controller
@ResponseBody
class AdoptionController {

    private final ChatClient ai;
    private final InMemoryChatMemoryRepository memoryRepository;

    AdoptionController(ChatClient.Builder ai, VectorStore vectorStore) {
        var system = """
                You are an AI-powered assistant to help people adopt a dog from the adoption
                agency named Pooch Palace with locations in Antwerp, Seoul, Tokyo, Singapore, Paris,
                Mumbai, New Delhi, Barcelona, San Francisco, and London. Information about the dogs available
                will be presented below. If there is no information, then return a polite response suggesting we
                don't have any dogs available.
                """;

        this.memoryRepository = new InMemoryChatMemoryRepository();

        this.ai = ai
                .defaultSystem(system)
                .defaultAdvisors(new QuestionAnswerAdvisor(vectorStore))
                .build();
    }

    /// TBD
}

So far, all we’ve defined is a constructor into which we’ve injected the Spring AI ChatClient.Builder. The ChatClient is the center piece API for interacting with the auto-configured ChatModel representing our connection to OpenAI (or Ollama or whatever).

We want all interactions with the chat model to have some sensible defaults, including a system prompt. The system prompt gives the model a framing through which it should attempt to answer all subsequent questions. So, in this case, we’ve instructed it to answer all questions as though it were working at our fictitious adoption agency called Pooch Palace. As a result, it should politely decline or deflect any questions that don’t relate to our dog adoption agency.

In addition, we use a QuestionAnswerAdvisor backed by a MongoDB VectorStore. This ensures that the model attempts to answer questions based only on relevant documents—specifically, the vector-matched entries from our database of adoptable dogs.

We’ve also configured an advisor. Advisors are a Spring AI concept. Think of them as filters, or interceptors. They pre- and post-process requests made to the large language model. Here, the QuestionAnswerAdvisor lives in a separate dependency we need to explicitly add to our pom.xml:

<dependency>
   <groupId>org.springframework.ai</groupId>
   <artifactId>spring-ai-advisors-vector-store</artifactId>
</dependency>

This advisor will search the MongoDB Atlas vector store on each request and do a similarity search, looking for data that might be germane to the user’s request. When somebody comes along inquiring about dogs of a particular characteristic, it will look for dogs that might match the characteristic. The results we get back will be a subselection of all the dogs available for adoption. We’ll transmit only that relevant subselection to the model. Remember, each interaction with a model has costs—both dollars and cents (or Euros!) and complexity costs. So the less data we send, the better. The model, then, takes our system prompt, the user’s question, and the included data from the vector store into consideration when formulating its response.

Models don't remember things. Think of them like Dory the goldfish from Finding Nemo. So we configure one more advisor which keeps a running transcript of the conversation between you and it and then reminds the model of what’s been said. This way, it'll remember you from one request to another.

Now, we’ll add an endpoint to our AdoptionController to communicate with our AI assistant, with a little bit of memory. This will allow our assistant to maintain a local chat log (10 messages back), to keep track of the conversation, and provide a bit more context to our LLM.

   @GetMapping("/{user}/dogs/assistant")
    String inquire(@PathVariable String user, @RequestParam String question) {

        var memory = MessageWindowChatMemory.builder()
                .chatMemoryRepository(memoryRepository)
                .maxMessages(10)
                .build();

        var memoryAdvisor = MessageChatMemoryAdvisor.builder(memory).build();

        return this.ai
                .prompt()
                .user(question)
                .advisors(a -> a
                        .advisors(memoryAdvisor)
                        .param(ChatMemory.CONVERSATION_ID, user)
                )
                .call()
                .content();
    }

With all things taken together, you can now launch the program and then try it out:

http :8080/tkelly/dogs/assistant question==”do you have any neurotic dogs?”

With everything in place, it should return that there’s this dog named Prancer we might be interested in. (Boy, would we!)

Yes, we have a neurotic dog available for adoption. Meet Prancer, a demonic, neurotic dog who is not fond of humans, animals, or children, and resembles a gremlin. If you're interested in Prancer or would like more information, feel free to ask!

And there you have it. In no time at all, we went from zero to AI hero, making it easier to connect people with the dogs of their dreams… or nightmares!

Bring AI to production

Spring AI does more than just simple pet projects. It is designed for enterprise applications, and to fit in with your existing Spring applications. Transitioning to production involves ensuring security, scalability, and observability. Let’s break it down:

Security

It’s trivial to use Spring Security to lock down this web application. You could use the authenticated Principal#getName to use as the conversation ID too. What about the data stored in the collections? You can use MongoDB’s robust data encryption at rest.

Scalability

Each time you make an HTTP request to an LLM (or many databases), you’re blocking IO. This is a waste of a perfectly good thread. Threads aren't meant to just sit idle, waiting. Java 21 gives us virtual threads, which—for sufficiently IO-bound services—can dramatically improve scalability.

Enable them in application.properties with:

spring.threads.virtual.enabled=true

Instead of waiting for IO to complete, virtual threads free up resources, allowing your application to scale more effectively under load.

GraalVM native images

GraalVM compiles Java applications to native executables, reducing startup time and memory usage. Here’s how to compile your Spring AI app.

If you’ve got that set up as your SDK, you can turn this Spring AI application into an operating system and architecture specific native image with ease:

./mvnw -DskipTests -Pnative native:compile

This takes a minute or so on most machines but once it is done, you can run the binary with ease.

./target/mongodb

This program will start up in a fraction of the time that it did on the JVM. You might want to comment on the ApplicationRunner we created earlier since it's going to do IO on startup time, significantly delaying that startup. On my machine, it starts up in less than a tenth of a second.

Even better, you should observe that the application takes a very small fraction of the RAM it would've otherwise taken on the JVM.

Containerize for Docker

“That’s all very well and good,” you might say, “but I need to get this running on any cloud platform, and that means getting it into the shape of a Docker image.” Easy!

./mvnw -DskipTests -Pnative spring-boot:build-image

Stand back. This might take another minute still. When it finishes, you’ll see it’s printed out the name of the Docker image that’s been generated.

You can run it, remembering to override the hosts and ports of things it would've referenced on your host.

We’re on macOS, and amazingly, this application, when run on a macOS virtual machine emulating Linux, runs even faster (and right from the jump, too) then it would’ve on macOS directly. Amazing!

Observability

This application’s so darn good I’ll bet it’ll make headlines, just like our dog Prancer. And when that happens, you’d be well advised to keep an eye on your system resources and, importantly, the token count. All requests to an LLM have a cost—at least one of complexity, if not dollars and cents.

Spring Boot Actuator, powered by Micrometer, provides metrics out of the box.

Access the metrics endpoint to monitor token usage and other key indicators:

http://localhost:8080/actuator/metrics

Nice! You can use Micrometer to forward those metrics to your MongoDB time-series database to get a single pane of glass, a dashboard.

MongoDB Atlas: Fully managed and scalable

MongoDB Atlas is the managed cloud service for MongoDB, offering scalability and high availability across multiple cloud providers. With Atlas, MongoDB handles backups, monitoring, upgrades, and patches—letting you focus on building your application, not maintaining infrastructure. If managing databases isn’t core to your business, why not delegate it to the people who built MongoDB?

Conclusion

There you have it! You've just built a production-worthy, AI-ready, Spring AI- and MongoDB-powered application in no time at all. We’ve only begun to scratch the surface. Check out Spring AI 1.0 at the Spring Initializr today and learn more about MongoDB Atlas and the vector store support.

If you want to see more about what you can do with MongoDB, Spring, and AI, check out our tutorial on building a real-time AI fraud detection system with Spring Kafka and MongoDB. If you’re more interested in just getting started with Spring and MongoDB, we have a tutorial on that too!

Building a Real-Time Market Data Aggregator with Kafka and MongoDB

Tim Kelly — Thu, 15 May 2025 12:26:42 +0000

It starts the same way every time: an RSS feed, a CSV dump, a dashboard wired together with hope and duct tape. Maybe the data’s late. Maybe the moving average lags just enough to miss the signal. Or maybe it’s just another system straining under the weight of “real-time” without the architecture to back it up.

In this tutorial, we’ll build something better. Using Spring Kafka, we’ll stream simulated live stock data into a Kafka topic. From there, Kafka Streams will compute the Relative Strength Index (RSI) on the fly, and we’ll persist it to MongoDB using Spring Data MongoDB. On the historical side, we’ll use MongoDB’s aggregation framework to calculate and analyze simple and exponential moving averages (SMA and EMA) using the MongoDB aggregation pipeline—no third-party data warehouse required.

By the end, we’ll have a real-time financial pipeline that’s compact and fast. Feel free to clone the GitHub repository, and use the front end I reference throughout the tutorial.

Prerequisites

Before getting started, make sure you have the following installed and ready:

Java 17+ (I use 21): Download Java or use your preferred distribution
Maven 3.9+: Download Maven
MongoDB Atlas cluster, a free tier M0 will work perfectly
Apache Kafka 3.9+: (locally installed or accessible remotely) Download Kafka

Creating our Spring application

To keep things simple, we’ll use Spring Initializr to scaffold our project.

Choose the following options:

Project: Maven
Language: Java
Spring Boot Version: 3.4.4
Group: com.mongodb (our your own preferred group)
Artifact: FinancialAggregator
Name: FinancialAggregator
Java version: 21
Dependencies:
- Spring Web
- Spring Data MongoDB
- Spring for Apache Kafka
- Kafka Streams

Here’s what each dependency will do for us:

Spring Web: Gives us everything we need to expose REST APIs with Spring Boot. We’ll use this to serve our financial data through HTTP endpoints.
Spring Data MongoDB: Provides convenient, idiomatic access to MongoDB using Spring Data repositories. It allows us to interact with our database in a way that feels familiar as a Spring developer.
Spring for Apache Kafka: Adds Spring Boot integration for Kafka, including KafkaTemplate for sending messages and configuration support for producers and consumers.
Kafka Streams: The actual stream processing engine. This powers our calculations on our stock data in real time. Spring will wire this into the app using @EnableKafkaStreams.

With this setup, we’ll have a data pipeline that simulates live financial data, processes it on the fly, stores it, and exposes it, as well as allowing us to analyze trends in historic data.

Once generated, extract the zip and open it in your preferred IDE.

Setting up MongoDB

Before we can start processing and storing data, we need a MongoDB cluster ready to receive it.

If you don’t already have a MongoDB Atlas cluster, you can follow this quick start guide to set one up for free. Once our cluster is running, we can use either the Atlas UI or mongosh to configure the necessary collections.

What MongoDB does in this project

MongoDB acts as our primary storage engine. It stores both:

Real-time RSI data produced by Kafka Streams.
Historical stock market data used to calculate SMA and EMA via aggregations.

We’ll take advantage of time series collections in MongoDB, which are optimized for timestamped data like stock prices. This lets us efficiently query, sort, and analyze trends over time.

Creating time series collections in MongoDB Atlas

Once your cluster is ready, follow these steps:

Open the MongoDB Atlas UI and navigate to your cluster.
Click "Browse Collections."
Create a new database named Stock_Market_Data.
Inside this database, create two time series collections:
- Collection Name: Live_Data
  - Time Field: date
  - Granularity: Seconds
- Collection Name: Historical_Data
  - Time Field: date
  - Metafield: company (we'll see why this is important later)
  - Granularity: Seconds

That’s all you need to start writing and reading real-time stock data with MongoDB. Next, we’ll simulate live market data and send it to Kafka.

Configuring our MongoDB connection

From the Atlas UI, we can get our connection string, and bring it back to our application properties.

spring.application.name=Financial Aggregator  

spring.data.mongodb.uri=<Your-Connection-String>
spring.data.mongodb.database=Stock_Market_Data

We'll also name the database we're using, Stock_Market_Data.

Creating our live data

Our live data won’t be truly live. Instead, we’ll simulate and stream it into the system. There are plenty of APIs out there that offer real-time stock data, but to keep this demo self-contained and avoid external dependencies and key generation/management, we’ll mock the data ourselves. That said, if you’d prefer to plug in a real API, go for it—this setup won’t get in your way.

Our live data model

Start by creating a package called model in your application. Inside it, add a new class named LiveStockData. I came to understand far too late in the game that this could be confused with an agricultural term, but alas.

This class is our POJO—used to map simulated stock data to documents in MongoDB:

package com.mongodb.financialaggregator.model;  

import org.bson.types.ObjectId;  
import org.springframework.data.annotation.Id;  
import org.springframework.data.mongodb.core.mapping.Document;  
import org.springframework.data.mongodb.core.mapping.TimeSeries;  

import java.util.Date;  

@Document(collection = "Live_Data")  
@TimeSeries(timeField = "date")  
public class LiveStockData {  

    @Id  
    private ObjectId id;  
    private Date date;  
    private String company;  
    private Double closeLast;  
    private Double open;  
    private Double low;  
    private Double volume;  
    private Double high;  
    private Double rsi;  

    public LiveStockData() {}  

    public ObjectId getId() { return id; }  
    public void setId(ObjectId id) { this.id = id; }  

    public Date getDate() { return date; }  
    public void setDate(Date date) { this.date = date; }  

    public String getCompany() { return company; }  
    public void setCompany(String company) { this.company = company; }  

    public Double getCloseLast() { return closeLast; }  
    public void setCloseLast(Double closeLast) { this.closeLast = closeLast; }  

    public Double getOpen() { return open; }  
    public void setOpen(Double open) { this.open = open; }  

    public Double getLow() { return low; }  
    public void setLow(Double low) { this.low = low; }  

    public Double getVolume() { return volume; }  
    public void setVolume(Double volume) { this.volume = volume; }  

    public Double getHigh() { return high; }  
    public void setHigh(Double high) { this.high = high; }  

    public Double getRsi() { return rsi; }  
    public void setRsi(Double rsi) { this.rsi = rsi; }  
}

This class gives us enough fields to mimic the structure of a real stock market API and power a candlestick chart. It also includes a field for the RSI (Relative Strength Index), which we'll calculate later.

A lot of this is fairly standard getter/setters, but let's take a look at the annotations we're adding to let Spring know what to do with this class.

@Document(collection = "Live_Data")

This tells Spring Data MongoDB to map this class to the Live_Data collection in MongoDB.

@TimeSeries(timeField = "date")

This annotation enables MongoDB’s time series capabilities. We’re specifying date as the time field, which allows us to efficiently store and query stock data over time.

Configure Kafka producer

Next, we’ll set up a Kafka producer to stream our simulated stock data. Create a new config package, and inside it, add a class named KafkaProducerConfig.

The Kafka producer is responsible for publishing messages, in our case, LiveStockData objects, to a Kafka topic.

package com.mongodb.financialaggregator.config;  

import com.mongodb.financialaggregator.model.LiveStockData;  
import org.apache.kafka.clients.producer.ProducerConfig;  
import org.apache.kafka.common.serialization.StringSerializer;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  
import org.springframework.kafka.core.*;  
import org.springframework.kafka.support.serializer.JsonSerializer;  

import java.util.HashMap;  
import java.util.Map;  

@Configuration  
public class KafkaProducerConfig {  

    @Bean  
    public ProducerFactory<String, LiveStockData> producerFactory() {  
        Map<String, Object> config = new HashMap<>();  
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");  
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);  
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);  
        return new DefaultKafkaProducerFactory<>(config);  
    }  

    @Bean  
    public KafkaTemplate<String, LiveStockData> kafkaTemplate() {  
        return new KafkaTemplate<>(producerFactory());  
    }  
}

Here, we configure a ProducerFactory that knows how to serialize our data using JsonSerializer. We also state we will be talking to the broker on localhost:9092. The KafkaTemplate is a convenient way to send messages to Kafka topics from anywhere in the app.

The @Configuration annotation just tells Spring this is a configuration class.

Configure Kafka streams

Before we can wire up the stream logic, we need to tell Kafka Streams how to serialize and deserialize our custom LiveStockData objects. To do this, we'll create a custom SerDe (serializer/deserializer).

Create a utility package, and inside it, add the LiveStockDataSerde class:

package com.mongodb.financialaggregator.utility;  

import com.mongodb.financialaggregator.model.LiveStockData;  
import org.apache.kafka.common.serialization.Serdes;  
import org.springframework.kafka.support.serializer.JsonDeserializer;  
import org.springframework.kafka.support.serializer.JsonSerializer;  

public class LiveStockDataSerde extends Serdes.WrapperSerde<LiveStockData> {  
    public LiveStockDataSerde() {  
        super(new JsonSerializer<>(), new JsonDeserializer<>(LiveStockData.class));  
    }  
}

We’ll use this class when defining our stream topology to ensure Kafka Streams can work directly with our domain objects without manual conversion—basically, so Kafka Streams knows how to convert bytes to Java objects and back.

In the same config package as our KafkaProducerConfig, we also need to create a KafkaStreamsConfig class. The KafkaStreamConfig will configure the Kafka Streams API to consume our messages and process them in real time.

package com.mongodb.financialaggregator.config;  

import org.apache.kafka.common.serialization.Serdes;  
import org.apache.kafka.streams.StreamsConfig;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  
import org.springframework.kafka.annotation.EnableKafkaStreams;  
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;  
import org.springframework.kafka.config.KafkaStreamsConfiguration;  
import org.springframework.kafka.support.serializer.JsonDeserializer;  

import java.util.HashMap;  
import java.util.Map;  


@Configuration  
@EnableKafkaStreams  
public class KafkaStreamsConfig {  

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)  
    KafkaStreamsConfiguration kStreamsConfig() {  
        Map<String, Object> props = new HashMap<>();  
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "rsi-streams-app");  
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");  
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());  
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.mongodb.financialaggregator.model");  
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.mongodb.financialaggregator.model.LiveStockData");  
        props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false);  

        return new KafkaStreamsConfiguration(props);  
    }  
}

This sets up the Kafka Streams runtime with the basic configurations it needs: a unique application ID, Kafka broker address, and the expected serialization types. It also ensures that our stream can deserialize LiveStockData objects correctly.

@EnableKafkaStreams tells Spring Boot to activate Kafka Streams support.

We also have some important configurations set in our @Bean. KafkaStreamsConfiguration

APPLICATION_ID_CONFIG: A unique name for this Kafka Streams application. This is required. This ID is how Kafka tracks our app’s state and progress across restarts. Changing it will reset our app’s stream state, including any local state stores or offsets.
BOOTSTRAP_SERVERS_CONFIG: Where to find our Kafka cluster.
DEFAULT_KEY_SERDE_CLASS_CONFIG: Uses Serdes.String(). We expect keys to be strings.
JsonDeserializer config: Ensures Kafka Streams knows how to read our custom object (LiveStockData).

If you're new to Kafka, you might be wondering, "Why not just use a Kafka Consumer?" Well Kafka Streams isn’t just a consumer—it’s a full stream processing library built on top of Kafka. It lets us define processing logic as a topology of transformations, stateful operations, and sinks. In this case, we’ll use it to calculate RSI values as they come in.

Simulating live stock data

Now, I’m not going to pretend this is the most impressive stock market simulator in the world. It’s not. But it gets the job done. It gives us consistent, dynamic data to push into Kafka and eventually graph. That’s all we need.

I've added some logic here just to generate some fluctuating data for an imaginary company called "Doof."

We'll create a service package and add a class named SimulateLiveData.

package com.mongodb.financialaggregator.service;

import com.mongodb.financialaggregator.model.LiveStockData;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.*;
import java.util.Date;
import java.util.Random;

@Service
public class SimulateLiveData {

    private Date currentSimulatedDate;
    private double previousClose;
    private final KafkaTemplate<String, LiveStockData> kafkaTemplate;

    private static final String COMPANY = "DOOF";
    private static final double INITIAL_OPEN = 100.0;
    private static final double INITIAL_VOLUME = 1000.0;

    private final double trendSlope = 0.002;
    private boolean trendingUp = true;
    private int trendCounter = 0;

    private final int oscillationPeriod = 30;
    private final double oscillationAmplitude = 1.5;

    private final Random random = new Random();

    public SimulateLiveData(KafkaTemplate<String, LiveStockData> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        this.currentSimulatedDate = Date.from(LocalDate.now()
                .atStartOfDay(ZoneId.systemDefault())
                .toInstant());
        this.previousClose = INITIAL_OPEN;
    }

    public LiveStockData simulateLiveData() {
        LiveStockData stock = new LiveStockData();

        stock.setDate(currentSimulatedDate);
        stock.setCompany(COMPANY);

        double trendAdjustment = trendingUp ? trendSlope : -trendSlope;
        double oscillation = oscillationAmplitude *
                Math.sin((2 * Math.PI * trendCounter) / oscillationPeriod);
        double marketBaseline = previousClose * (1 + trendAdjustment) + oscillation;

        boolean hasVolatility = random.nextDouble() < 0.1;
        double volatilityBoost = hasVolatility ? (randomBetween(-0.1, 0.1)) : 0;

        double open = marketBaseline * (1 + volatilityBoost);
        double close = open * (1 + randomBetween(-0.05, 0.05));
        double high = Math.max(open, close) * (1 + randomBetween(0.01, 0.1));
        double low = Math.min(open, close) * (1 - randomBetween(0.01, 0.1));
        double volume = INITIAL_VOLUME * (1 + randomBetween(-0.5, 0.5));

        stock.setOpen(round(open));
        stock.setCloseLast(round(close));
        stock.setHigh(round(high));
        stock.setLow(round(low));
        stock.setVolume(round(volume));

        previousClose = close;
        trendCounter++;

        // Advance to the next simulated day
        LocalDate next = currentSimulatedDate.toInstant()
                .atZone(ZoneId.systemDefault())
                .toLocalDate()
                .plusDays(1);
        currentSimulatedDate = Date.from(next.atStartOfDay(ZoneId.systemDefault()).toInstant());

        // Flip the trend every 60 days
        if (trendCounter % 60 == 0) {
            trendingUp = !trendingUp;
        }

        return stock;
    }

    private double randomBetween(double min, double max) {
        return min + (max - min) * random.nextDouble();
    }

    private double round(double value) {
        return BigDecimal.valueOf(value)
                .setScale(2, RoundingMode.HALF_UP)
                .doubleValue();
    }
}

This function simulates somewhat real-looking stock data with some randomness, gentle trends, and a hint of periodic volatility. It resets daily and flips direction every 60 “days” to prevent things from spiraling to the moon or zero.

Scheduling the simulation

Now, let’s wire it up so that we publish this fake stock data to Kafka automatically. I've arbitrarily decided once per second.

Inside the same SimulateLiveData class, add:

@Scheduled(fixedRate = 1000)
public void simulateEverySecond() {
    LiveStockData data = simulateLiveData();

    System.out.println("========== STOCK DEBUG INFO ==========");
    System.out.println("Company:       " + data.getCompany());
    System.out.println("Date (raw):    " + data.getDate());
    System.out.println("Date (class):  " + (data.getDate() != null ? data.getDate().getClass().getName() : "null"));
    System.out.println("Open:          " + data.getOpen());
    System.out.println("Close/Last:    " + data.getCloseLast());
    System.out.println("High:          " + data.getHigh());
    System.out.println("Low:           " + data.getLow());
    System.out.println("Volume:        " + data.getVolume());
    System.out.println("======================================");

    try {
        kafkaTemplate.send("stock-prices", data.getCompany(), data);
        System.out.println("Sent successfully.\n");
    } catch (Exception e) {
        System.err.println("Insert failed: " + e.getMessage());
        e.printStackTrace();
    }
}

Thanks to @Scheduled(fixedRate = 1000), this method is triggered every second. It generates a new stock datapoint, prints it out for debugging, and sends it to the stock-prices Kafka topic using the KafkaTemplate.

I've added some logging and error handling just to help with the clarity when we're running the application.

Finally, make sure you enable scheduling in your application entrypoint by adding @EnableScheduling:

package com.mongodb.financialaggregator;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class FinancialAggregatorApplication {

    public static void main(String[] args) {
        SpringApplication.run(FinancialAggregatorApplication.class, args);
    }
}

With this in place, our app will start streaming simulated stock data to Kafka as soon as we run it.

Perform calculations in the stream processor pipeline

The stream processor allows us to consume the stock price data flowing through Kafka, calculate real-time analytics, and persist our enriched results to MongoDB.

We'll use this to calculate some Relative Strength Indexes (RSIs). The RSI is a momentum indicator used in financial analysis. RSI measures the speed and magnitude of a security's recent price changes to detect overbought or oversold conditions in the price of that security. It is an oscillatory metric (displayed as a line graph) on a scale of 0 to 100. We can add them to our documents before we store them in MongoDB.

Using `MongoRepository`

To save our live stock data into MongoDB, we’ll use Spring Data’s MongoRepository interface. This gives us a ready-to-go set of CRUD operations with minimal setup. No boilerplate, just clean data access.

Create a repository package and add a new interface called LiveDataRepository:

package com.mongodb.financialaggregator.repository;

import com.mongodb.financialaggregator.model.LiveStockData;
import org.bson.types.ObjectId;
import org.springframework.data.mongodb.repository.MongoRepository;

public interface LiveDataRepository extends MongoRepository<LiveStockData, ObjectId> {
}

Here, we're extending MongoRepository, which is a Spring Data interface that automatically provides methods like save(), findAll(), deleteById(), and more.

We specify LiveStockData as the type we’re working with, and ObjectId as the ID type (since that’s what we used in our model).

We can also define custom query methods here later, if needed (e.g., findByCompany(String company)), but for now, the built-in methods are more than enough.

The RSI calculator

In the Utility package, add a new class called RsiCalculator:

package com.mongodb.financialaggregator.utility;  

import java.util.List;  

public class RsiCalculator {  

    public static double calculate(List<Double> prices, int period) {  
        if (prices.size() < period + 1) return 0.0;  

        double gain = 0.0;  
        double loss = 0.0;  
        for (int i = 1; i <= period; i++) {  
            double diff = prices.get(i) - prices.get(i - 1);  
            if (diff > 0) gain += diff;  
            else loss -= diff;  
        }  

        if (loss == 0) return 100.0;  
        double rs = gain / loss;  
        return 100.0 - (100.0 / (1.0 + rs));  
    }  
}

Here, you can see that we take a rolling window of prices and compute RSI using the standard formula. If there's no loss (i.e., price only went up), the RSI maxes out at 100. Otherwise, it calculates the average gain/loss ratio over the given period.

Now, in our service package, we'll add a class RsiStreamProcessor. Here, we will define our hook into the Kafka Streams pipeline, apply the RSI calculation, and persist the result to MongoDB.

package com.mongodb.financialaggregator.service;  

import com.mongodb.financialaggregator.model.LiveStockData;  
import com.mongodb.financialaggregator.repository.LiveDataRepository;  
import com.mongodb.financialaggregator.utility.LiveStockDataSerde;  
import com.mongodb.financialaggregator.utility.RsiCalculator;  
import org.apache.kafka.common.serialization.Serdes;  
import org.apache.kafka.streams.StreamsBuilder;  
import org.apache.kafka.streams.kstream.Consumed;  
import org.apache.kafka.streams.kstream.KStream;  
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.stereotype.Component;  

import java.util.ArrayDeque;  
import java.util.ArrayList;  
import java.util.Deque;  
import java.util.Map;  
import java.util.concurrent.ConcurrentHashMap;  

@Component  
public class RsiStreamProcessor {  

    private static final int RSI_PERIOD = 14;  
    private final LiveDataRepository liveDataRepository;  
    private final Map<String, Deque<Double>> priceHistory = new ConcurrentHashMap<>();  

    public RsiStreamProcessor(LiveDataRepository liveDataRepository, StreamsBuilder streamsBuilder) {  
        this.liveDataRepository = liveDataRepository;  
        buildPipeline(streamsBuilder);  
    }  

    public void buildPipeline(StreamsBuilder builder) {  
        LiveStockDataSerde liveSerde = new LiveStockDataSerde();  
        KStream<String, LiveStockData> stream = builder.stream(  
                "stock-prices",  
                Consumed.with(Serdes.String(), liveSerde)  
        );  

        stream.peek((key, data) -> {  
            System.out.println("Received from Kafka: " + data);  

            if (data == null || data.getCloseLast() == null || data.getCompany() == null) {  
                return;  
            }  

            String symbol = data.getCompany();  
            double close = data.getCloseLast();  

            Deque<Double> prices = priceHistory.computeIfAbsent(symbol, k -> new ArrayDeque<>());  
            prices.addLast(close);  
            if (prices.size() > RSI_PERIOD + 1) prices.removeFirst();  

            if (prices.size() == RSI_PERIOD + 1) {  
                double rsi = RsiCalculator.calculate(new ArrayList<>(prices), RSI_PERIOD);  
                data.setRsi(rsi);  
            }  

            liveDataRepository.save(data);  
        });  
    }  
}

You can see in the buildPipeline we first validate the data. Then, we track a sliding window of the last 15 prices. If we have enough data (at least 15 days to look through), we compute the RSI and attach it to the stock object. At the end of all of this, we can then save our data into our database with the RSI calculated.

The @Component annotation tells Spring this is a managed bean and should be picked up during component scanning.

This gives us a fully reactive stream: Every second, simulated data flows in, RSI is computed in real time, and the result is persisted—all without writing a line of boilerplate consumer logic.

Serve up our data in an API

We have our data processed and funnelled into our database. Let's look at getting some of that data out.

We’ll expose a REST endpoint so we (or our front end) can query the latest stock data and indicators like RSI.

In our service package, let’s create a new class, FinancialService:

package com.mongodb.financialaggregator.service;  

import com.mongodb.financialaggregator.model.*;  
import com.mongodb.financialaggregator.repository.LiveDataRepository;  
import org.springframework.stereotype.Service;  

import java.util.Date;  
import java.util.List;  

@Service  
public class FinancialService {  

    private LiveDataRepository liveDataRepository;

    public FinancialService(LiveDataRepository liveDataRepository) {
        this.liveDataRepository = liveDataRepository; 
    }  

    public List<LiveStockData> getLiveData() {  
        return liveDataRepository.findAll();  
    }  
}

Nothing fancy—we inject the LiveDataRepository and expose a getLiveData() method that fetches everything from the Live_Data collection using findAll().

In a new package, controller, create a class called FinancialController. This is the part of the app that receives HTTP requests and sends responses back.

package com.mongodb.financialaggregator.controller;  

import com.mongodb.financialaggregator.model.LiveStockData;   
import com.mongodb.financialaggregator.service.FinancialService;  
import org.springframework.web.bind.annotation.*;  

import java.util.Date;  
import java.util.List;  

@CrossOrigin(origins = "http://localhost:5173") // Adjust if needed  
@RestController  
@RequestMapping(path = "/stocks")  
public class FinancialController {  
    private final FinancialService financialService;  

    public FinancialController(FinancialService financialService) {  
        this.financialService = financialService;  
    }  

    // test  
    // GET http://localhost:8080/stocks/live    
    // curl -v "http://localhost:8080/stocks/live"    
    @GetMapping(value= "/live", produces = "application/json")  
    public List<LiveStockData> getLiveData() {  
        return financialService.getLiveData();  
    };  
}

This controller:

Maps the /stocks route to this controller.
Exposes a /stocks/live endpoint that returns a JSON list of all stored LiveStockData documents.
Uses @CrossOrigin to allow local frontend dev (like our React/Vite on port 5173) to make requests during development.

We'll be back here later to add more endpoints. Now that we have our live data being fed to our database and presented in our API, it's time to look at some of the aggregations we can do on data in our database.

Perform aggregations on our historical data

Our stock market data model

package com.mongodb.financialaggregator.model;  

import org.bson.types.ObjectId;  
import org.springframework.data.annotation.Id;  
import org.springframework.data.mongodb.core.mapping.Document;  
import org.springframework.data.mongodb.core.mapping.TimeSeries;  

import java.util.Date;  

@Document(collection = "Historical_Data")  
@TimeSeries(collection = "Historical_Data", timeField = "Date", metaField = "Company")  
public class StockMarketData {  

    @Id  
    private ObjectId id;  
    private Date date;  
    private String company;  
    private Double closeLast;  
    private Double open;  
    private Double low;  
    private Double volume;  
    private Double high;  

    public StockMarketData() {  
    }  

    public StockMarketData(ObjectId id, Date date, String company, Double closeLast, Double open, Double low,Double volume, Double high) {  
        this.id = id;  
        this.date = date;  
        this.company = company;  
        this.closeLast = closeLast;  
        this.open = open;  
        this.low = low;  
        this.volume = volume;  
        this.high = high;  
    }  

    public StockMarketData(Date date, String company, Double closeLast, Double open, Double low,Double volume, Double high) {  
        this.date = date;  
        this.company = company;  
        this.closeLast = closeLast;  
        this.open = open;  
        this.low = low;  
        this.volume = volume;  
        this.high = high;  
    }  

    public ObjectId getId() {  
        return id;  
    }  

    public void setId(ObjectId id) {  
        this.id = id;  
    }  

    public Date getDate() {  
        return date;  
    }  

    public void setDate(Date date) {  
        this.date = date;  
    }  

    public String getCompany() {  
        return company;  
    }  

    public void setCompany(String company) {  
        this.company = company;  
    }  

    public Double getCloseLast() {  
        return closeLast;  
    }  

    public void setCloseLast(Double closeLast) {  
        this.closeLast = closeLast;  
    }  

    public Double getOpen() {  
        return open;  
    }  

    public void setOpen(Double open) {  
        this.open = open;  
    }  

    public Double getLow() {  
        return low;  
    }  

    public void setLow(Double low) {  
        this.low = low;  
    }  

    public Double getVolume() {  
        return volume;  
    }  

    public void setVolume(Double volume) {  
        this.volume = volume;  
    }  

    public Double getHigh() {  
        return high;  
    }  

    public void setHigh(Double high) {  
        this.high = high;  
    }  
}

Read in historical data

In addition to our real-time stream, we’ll bring in some historical stock data for context and longer-term trend analysis, showing off a few of MongoDB's aggregations specifically for analysing trends in financial data.

We’re using the Stock Market: Historical Data of Top 10 Companies dataset from Kaggle. It contains 25,161 rows, each representing stock market activity for a major company on a specific date between July 2013 and July 2023. The dataset includes daily open, high, low, close prices, and trading volumes for companies like Apple, Starbucks, Microsoft, Cisco, Qualcomm, Meta, Amazon, Tesla, AMD, and Netflix.

Download the dataset and place it in your project’s resources directory with the filename data.csv.

Creating a repository for historical data

To load this data into MongoDB, we’ll use a MongoRepository just like we did for live data.

Create a repository package and add an interface called FinancialRepository:

package com.mongodb.financialaggregator.repository;

import com.mongodb.financialaggregator.model.StockMarketData;
import org.bson.types.ObjectId;
import org.springframework.data.mongodb.repository.MongoRepository;

import java.util.Date;
import java.util.List;

public interface FinancialRepository extends MongoRepository<StockMarketData, ObjectId> {
    List<StockMarketData> findAllByCompanyAndDateBetween(String company, Date startDate, Date endDate);
}

This interface gives us access to save() and saveAll() for inserting data, and we’ve also added a custom method—

findAllByCompanyAndDateBetween(...)—which Spring Data will automatically implement for us. This will be handy when calculating things like SMA and EMA over a time range.

Reading and importing the CSV

Now, let’s read the historical CSV and insert it into MongoDB.

Create a class called ReadInHistoricalData in your service package:

package com.mongodb.financialaggregator.service;

import com.mongodb.financialaggregator.model.StockMarketData;
import com.mongodb.financialaggregator.repository.FinancialRepository;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

@Service
public class ReadInHistoricalData implements ApplicationRunner {

    private final FinancialRepository financialRepository;
    private static final int BATCH_SIZE = 500; // Insert in chunks for performance

    public ReadInHistoricalData(FinancialRepository financialRepository) {
        this.financialRepository = financialRepository;
    }

    @Override
    public void run(org.springframework.boot.ApplicationArguments args) {
        loadData();
    }

    public void loadData() {
        String fileName = "data.csv";

        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(new ClassPathResource(fileName).getInputStream()))) {

            // Skip header
            reader.readLine();

            List<StockMarketData> batch = new ArrayList<>();
            String line;

            while ((line = reader.readLine()) != null) {
                StockMarketData data = parseCsvLine(line);
                if (data != null) {
                    batch.add(data);
                }

                if (batch.size() == BATCH_SIZE) {
                    financialRepository.saveAll(batch);
                    batch.clear();
                }
            }

            // Save any remaining data
            if (!batch.isEmpty()) {
                financialRepository.saveAll(batch);
            }

            System.out.println("Data import completed successfully.");
        } catch (Exception e) {
            System.err.println("Error loading CSV: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private StockMarketData parseCsvLine(String line) {
        try {
            String[] fields = line.split(",");

            if (fields.length < 7) {
                System.err.println("Skipping invalid line: " + line);
                return null;
            }

            String company = fields[0].trim();
            Date date = parseDate(fields[1].trim());
            Double closeLast = parseDouble(fields[2]);
            Double volume = parseDouble(fields[3]);
            Double open = parseDouble(fields[4]);
            Double high = parseDouble(fields[5]);
            Double low = parseDouble(fields[6]);

            if (date == null) {
                System.err.println("Skipping row due to invalid date: " + line);
                return null;
            }

            return new StockMarketData(date, company, closeLast, open, low, volume, high);
        } catch (Exception e) {
            System.err.println("Error parsing line: " + line);
            return null;
        }
    }

    private Date parseDate(String dateStr) {
        String[] dateFormats = {"MM/dd/yyyy", "MM-dd-yyyy"};

        for (String format : dateFormats) {
            try {
                return new SimpleDateFormat(format, Locale.US).parse(dateStr);
            } catch (ParseException ignored) {}
        }

        System.err.println("Unrecognized date format: " + dateStr);
        return null;
    }

    private Double parseDouble(String value) {
        try {
            return Double.parseDouble(value.replace("$", "").trim());
        } catch (NumberFormatException e) {
            return null;
        }
    }
}

This does a few important things:

It uses ApplicationRunner to run once at startup and load the data (which we’ll remove later).
It reads the data.csv file from resources using a buffered reader.
It parses and batches records into chunks of 500 for efficient writes.
It handles some common data parsing issues (like dollar signs and inconsistent date formats).

Tip: One-time data load

Once the data is successfully loaded into MongoDB, we can prevent it from running again on every startup by removing the implements ApplicationRunner line and deleting this method:

@Override
public void run(org.springframework.boot.ApplicationArguments args) {
    loadData();
}

This will leave us with a reusable loadData() method we can call manually if needed.

Custom financial repository

We already have a basic repository for loading data. But now we want to calculate things like moving averages, and that’s where things get a little more advanced. Instead of fetching raw documents and calculating these averages manually in memory in Java, we’re going to lean on MongoDB’s aggregation framework to do the heavy lifting for us.

Let’s start by creating a few records in our model package.

First, we'll define a simple record for the company names. Create a record called CompanyDTO:

package com.mongodb.financialaggregator.model;  

public record CompanyDTO(String name) {}

We'll use this to retrieve a list of unique company names for our front end, that we can use to filter our stock information for graphing.

Next, we need a record called SimpleMovingAverageDTO:

package com.mongodb.financialaggregator.model;  

import java.util.Date;  

public record SimpleMovingAverageDTO(double sma, Date date) {}

Lastly, let’s define a DTO first called ExponentialMovingAverageDTO:

package com.mongodb.financialaggregator.model;  

import java.util.Date;  

public record ExponentialMovingAverageDTO(double ema, Date date) {}

Up next, we'll create our interface in the repository package called CustomFinancialRepository, where we'll define our methods.

package com.mongodb.financialaggregator.repository;  

import com.mongodb.financialaggregator.model.CompanyDTO;  
import com.mongodb.financialaggregator.model.ExponentialMovingAverageDTO;  
import com.mongodb.financialaggregator.model.SimpleMovingAverageDTO;  

import java.util.Date;  
import java.util.List;  

public interface CustomFinancialRepository {  
    List<CompanyDTO> getAllCompanies();  
    List<SimpleMovingAverageDTO> getSimpleMovingAverage(String company, Date startDate, Date endDate);  
    List<ExponentialMovingAverageDTO> getExponentialMovingAverage(String company, Date startDate, Date endDate);  
}

Now, we’ll set up our custom repository implementation. This is where we’ll drop in our aggregation logic:

package com.mongodb.financialaggregator.repository;  

import com.mongodb.financialaggregator.model.CompanyDTO;  
import com.mongodb.financialaggregator.model.ExponentialMovingAverageDTO;  
import com.mongodb.financialaggregator.model.SimpleMovingAverageDTO;  
import org.springframework.data.domain.Sort;  
import org.springframework.data.mongodb.core.MongoTemplate;  
import org.springframework.data.mongodb.core.aggregation.*;  
import org.springframework.data.mongodb.core.query.Criteria;  
import org.springframework.stereotype.Repository;  

import java.util.Date;  
import java.util.List;  

import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;  

@Repository  
public class CustomFinancialRepositoryImplementation implements CustomFinancialRepository {  

    private final int EXPONENTIAL_MOVING_AVERAGE_PERIOD = 20;  
    private final int SIMPLE_MOVING_AVERAGE_PERIOD = 4;  

    private final MongoTemplate mongoTemplate;  

    public CustomFinancialRepositoryImplementation(MongoTemplate mongoTemplate) {  
        this.mongoTemplate = mongoTemplate;  
    }  

    @Override  
    public List<CompanyDTO> getAllCompanies() {  
        Aggregation getListOfUniqueCompanies = newAggregation(  
                group("company").first("company").as("name"),  
                sort(Sort.by("name"))  
        );  

        return mongoTemplate.aggregate(  
                getListOfUniqueCompanies, "Historical_Data", CompanyDTO.class  
        ).getMappedResults();  
    }  

}

The getAllCompanies() method grabs all unique companies in our Historical_Data collection. We’re grouping by the "company" field and grabbing the first one per group. It’s a clean way to get a list of available stock symbols to work with.

We also define some values we'll use for windows when calculating our moving averages.

    private final int EXPONENTIAL_MOVING_AVERAGE_PERIOD = 20;  
    private final int SIMPLE_MOVING_AVERAGE_PERIOD = 4;

Now, let’s move into the actual indicators.

Simple Moving Average

A Simple Moving Average (SMA) is what it sounds like. It takes the average closing price over a rolling window. This smooths out daily fluctuations and helps us see the trend over time.

Now, the actual implementation. We will add the following method to our CustomFinancialRepositoryImplementation implementation:

    @Override  
    public List<SimpleMovingAverageDTO> getSimpleMovingAverage(String company, Date startDate, Date endDate) {  
        Criteria criteria = Criteria  
                .where("company").is(company)  
                .and("date").gte(startDate).lte(endDate);  

        // Define SMA calculation (5-day moving average)  
        SetWindowFieldsOperation smaOperation = SetWindowFieldsOperation.builder()  
                .partitionBy("company") // Ensures calculations are per company  
                .sortBy(Sort.by(Sort.Direction.ASC, "date")) // Sort ascending by Date  
                .output(AccumulatorOperators.Avg.avgOf("closeLast")) // Compute SMA using avg()  
                .within(SetWindowFieldsOperation.Windows.documents(-SIMPLE_MOVING_AVERAGE_PERIOD, 0)) // Defines a 5-day window (4 previous + current)  
                .as("sma") // Output field named "sma"  
                .build();  

        Aggregation aggregation = newAggregation(  
                match(criteria),  
                smaOperation,  
                project("sma").and("date").as("date") // Keep only SMA and Date fields  
        );  

        return mongoTemplate.aggregate(  
                aggregation, "Historical_Data", SimpleMovingAverageDTO.class  
        ).getMappedResults();  
    }

We’re using MongoDB’s $setWindowFields operator here. This allows us to do rolling window calculations directly inside the database. We’re asking it to average the closeLast value for each document, across a sliding five-day window (the current doc plus the four before it), for a single company.

The result? We get a list of dates with a corresponding sma value.

Exponential Moving Average

The Exponential Moving Average is similar to the SMA, but instead of giving every day equal weight, it weights recent days more heavily. This makes the EMA a bit more reactive to recent price movement, which traders often prefer.

And now, let’s add the method that calculates it to our CustomFinancialRepositoryImplementation:

    @Override  
    public List<ExponentialMovingAverageDTO> getExponentialMovingAverage(String company, Date startDate, Date endDate) {  
        Criteria criteria = Criteria  
                .where("company").is(company)  
                .and("date").gte(startDate).lte(endDate);  

        SetWindowFieldsOperation emaOperation = SetWindowFieldsOperation.builder()  
                .partitionBy("company")  
                .sortBy(Sort.by(Sort.Direction.ASC, "date"))  
                .output(AccumulatorOperators.ExpMovingAvg.expMovingAvgOf("closeLast").n(EXPONENTIAL_MOVING_AVERAGE_PERIOD))  
                .as("ema")  
                .build();  

        Aggregation aggregation = newAggregation(  
                match(criteria),  
                emaOperation,  
                project("ema").and("date").as("date")  
        );  

        return mongoTemplate.aggregate(  
                aggregation, "Historical_Data", ExponentialMovingAverageDTO.class  
        ).getMappedResults();  
    }

This is almost identical to the SMA implementation, except we swap out $avgOf for $expMovingAvgOf, and give it a period (n) to define how far back to weigh things. MongoDB handles the weighting math, so we don’t need to.

Let’s test it

Now, with the aggregations done, all we need is a service to expose these methods to our controllers. This is the part of the app that acts as a go-between: It calls our repositories, applies any business logic (if needed), and returns results to the controller so they can be exposed through our API.

Update our FinancialService class

Here’s the full and final FinancialService class:

package com.mongodb.financialaggregator.service;  

import com.mongodb.financialaggregator.model.*;  
import com.mongodb.financialaggregator.repository.CustomFinancialRepository;  
import com.mongodb.financialaggregator.repository.FinancialRepository;  
import com.mongodb.financialaggregator.repository.LiveDataRepository;  
import org.springframework.stereotype.Service;  

import java.util.Date;  
import java.util.List;  

@Service  
public class FinancialService {  

    private FinancialRepository financialRepository;  
    private CustomFinancialRepository customFinancialRepository;  
    private LiveDataRepository liveDataRepository;  

    public FinancialService(CustomFinancialRepository customFinancialRepository,  
                            FinancialRepository financialRepository,  
                            LiveDataRepository liveDataRepository) {  
        this.customFinancialRepository = customFinancialRepository;  
        this.financialRepository = financialRepository;  
        this.liveDataRepository = liveDataRepository;  
    }  

    public List<String> getAllCompanies() {  
        List<CompanyDTO> companiesList = customFinancialRepository.getAllCompanies();  

        return companiesList.stream()  
                .map(CompanyDTO::name)  
                .toList();  
    }  

    public List<StockMarketData> getStockDataBetweenDates(String company, Date startDate, Date endDate) {  
        return financialRepository.findAllByCompanyAndDateBetween(company, startDate, endDate);  
    }  

    public List<SimpleMovingAverageDTO> getSimpleMovingAverage(String companyName, Date startDate, Date endDate) {  
        return customFinancialRepository.getSimpleMovingAverage(companyName, startDate, endDate);  
    }  

    public List<ExponentialMovingAverageDTO> getExponentialMovingAverage(String companyName, Date startDate, Date endDate) {  
        return customFinancialRepository.getExponentialMovingAverage(companyName, startDate, endDate);  
    }  

    public List<LiveStockData> getLiveData() {  
        return liveDataRepository.findAll();  
    }  
}

We've added our new repositories to the constructor, so we can access the methods. We've also called the methods we created in these repositories:

getAllCompanies() fetches a list of unique company names from our historical dataset.
getStockDataBetweenDates(...) pulls raw historical records between two dates, no aggregations applied.
getSimpleMovingAverage(...) and getExponentialMovingAverage(...) call the aggregation pipelines we built in the custom repository.

Adding our endpoints

Let's go back to the FinancialController we built earlier and add some endpoints for our new methods.

This controller will expose everything we’ve been working on—historical data queries, technical indicators, and even our live Kafka-driven data.

package com.mongodb.financialaggregator.controller;  

import com.mongodb.financialaggregator.model.ExponentialMovingAverageDTO;  
import com.mongodb.financialaggregator.model.LiveStockData;  
import com.mongodb.financialaggregator.model.SimpleMovingAverageDTO;  
import com.mongodb.financialaggregator.model.StockMarketData;  
import com.mongodb.financialaggregator.service.FinancialService;  
import org.springframework.format.annotation.DateTimeFormat;  
import org.springframework.web.bind.annotation.*;  

import java.util.Date;  
import java.util.List;  

@CrossOrigin(origins = "http://localhost:5173") // Adjust if needed  
@RestController  
@RequestMapping(path = "/stocks")  
public class FinancialController {  
    private final FinancialService financialService;  

    public FinancialController(FinancialService financialService) {  
        this.financialService = financialService;  
    }  

    // test  
    // GET http://localhost:8080/stocks/companies    // curl -v "http://localhost:8080/stocks/companies"    @GetMapping(value = "/companies", produces = "application/json")  
    public List<String> getAllCompanies() {  
        return financialService.getAllCompanies();  
    }  

    // test  
    // GET http://localhost:8080/stocks/companyStocksBetween?company=AAPL&startDate=2022-17-10&endDate=2023-07-17    // curl -v "http://localhost:8080/stocks/companyStocksBetween?company=AAPL&startDate=2022-07-17&endDate=2023-07-17"    @GetMapping(value = "/companyStocksBetween", produces = "application/json")  
    public List<StockMarketData> getStockDataBetweenDates(  
            @RequestParam String company,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date startDate,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date endDate) {  

        return financialService.getStockDataBetweenDates(company, startDate, endDate);  
    };  

    // test  
    // GET http://localhost:8080/stocks/sma?company=AAPL&startDate=2023-07-10&endDate=2023-07-17    // curl -v "http://localhost:8080/stocks/sma?company=AAPL&startDate=2022-07-10&endDate=2023-07-17"    @GetMapping(value = "/sma", produces = "application/json")  
    public List<SimpleMovingAverageDTO> getSmaForCompany(  
            @RequestParam String company,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date startDate,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date endDate) {  

        return financialService.getSimpleMovingAverage(company, startDate, endDate);  
    };  

    // test  
    // GET http://localhost:8080/stocks/ema?company=AAPL&startDate=2023-07-10&endDate=2023-07-17    // curl -v "http://localhost:8080/stocks/sma?company=AAPL&startDate=2023-07-10&endDate=2023-07-17"    @GetMapping(value= "/ema", produces = "application/json")  
    public List<ExponentialMovingAverageDTO> getEmaForCompany(  
            @RequestParam String company,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date startDate,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date endDate) {  

        return financialService.getExponentialMovingAverage(company, startDate, endDate);  
    };  

    // test  
    // GET http://localhost:8080/stocks/live    // curl -v "http://localhost:8080/stocks/live"    @GetMapping(value= "/live", produces = "application/json")  
    public List<LiveStockData> getLiveData() {  
        return financialService.getLiveData();  
    };  

}

Every method here maps directly to one in FinancialService. We keep this controller thin—just parsing request parameters, calling the service, and returning the result.

Let's walk through the new endpoints we just added.

    public List<String> getAllCompanies() {  
        return financialService.getAllCompanies();  
    }

This endpoint returns a list of all the unique companies in our historical dataset.

Behind the scenes, the FinancialService calls a MongoDB aggregation that groups documents by the company field, sorts them alphabetically, and maps them into a CompanyDTO. This method simply extracts the .name field from each of those records and returns the result as a List<String>.

    public List<StockMarketData> getStockDataBetweenDates(  
            @RequestParam String company,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date startDate,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date endDate) {  

        return financialService.getStockDataBetweenDates(company, startDate, endDate);  
    };

This one gives us the raw historical stock data for a given company between two dates.

The @RequestParam annotations pull values straight from the query string.
The @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) ensures that Spring can parse the date strings correctly, e.g., 2023-01-01.
The FinancialService calls a Spring Data repository method that uses a MongoDB query filter: company == ... and date >= startDate && <= endDate.

We'll use this to populate the candlestick graph in the front end.

    public List<SimpleMovingAverageDTO> getSmaForCompany(  
            @RequestParam String company,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date startDate,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date endDate) {  

        return financialService.getSimpleMovingAverage(company, startDate, endDate);  
    };

This endpoint returns a Simple Moving Average for a given company over a specific date range.

This method uses the MongoDB aggregation pipeline we set up to compute a rolling average of the closeLast field, and returns an array we can plot as a line graph.

Each record in the response includes:

A sma value (the average).
The date it corresponds to.

    public List<ExponentialMovingAverageDTO> getEmaForCompany(  
            @RequestParam String company,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date startDate,  
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) Date endDate) {  

        return financialService.getExponentialMovingAverage(company, startDate, endDate);  
    };

This one’s similar to the SMA endpoint but calculates an EMA instead.

The key difference is that recent values are weighted more heavily, but it still returns that array we can plot as a line graph.

This is useful for traders looking for momentum shifts or for graphing smoother, more sensitive trendlines.

Running our application

If you’re using Kafka 3.x with KRaft mode (no ZooKeeper), you must initialize your Kafka log directory with kafka-storage.sh format before starting the server. This only needs to be done once per broker.

Feel free to check a full detailed walkthrough on getting started with Kafka, but we’ll just do what we need to get started.

Choose any UUID for your cluster (or generate one). You can generate one like this:

CLUSTER_ID=$(bin/kafka-storage.sh random-uuid) echo $CLUSTER_ID

Run this command to format the KRaft storage:

bin/kafka-storage.sh format -t $CLUSTER_ID -c config/kraft/server.properties

This writes the metadata to the log dirs and makes the broker startable.

Now, start the Kafka server:

bin/kafka-server-start.sh config/kraft/server.properties

You should see logs like: INFO [KafkaServer id=0] started (kafka.server.KafkaServer).

You can now run your Spring application, and after we give some time for everything to start up, we should start to see some of our logs:

Received from Kafka: com.mongodb.financialaggregator.model.LiveStockData@2b0aaf9c
========== STOCK DEBUG INFO ==========
Company:       DOOF
Date (raw):    Sun Jun 22 00:00:00 IST 2025
Date (class):  java.util.Date
Open:          111.12
Close/Last:    115.26
High:          123.71
Low:           102.81
Volume:        1054.14
======================================
Sent successfully.

Received from Kafka: com.mongodb.financialaggregator.model.LiveStockData@398892bb
========== STOCK DEBUG INFO ==========
Company:       DOOF
Date (raw):    Mon Jun 23 00:00:00 IST 2025
Date (class):  java.util.Date
Open:          116.53
Close/Last:    111.57
High:          127.41
Low:           106.09
Volume:        1282.71
======================================
Sent successfully.

Received from Kafka: com.mongodb.financialaggregator.model.LiveStockData@31ca10ba
========== STOCK DEBUG INFO ==========
Company:       DOOF
Date (raw):    Tue Jun 24 00:00:00 IST 2025
Date (class):  java.util.Date
Open:          112.77
Close/Last:    117.03
High:          128.63
Low:           106.69
Volume:        1418.38
======================================
Sent successfully.

And we can verify this in our MongoDB collection. Our Historical_Data collection will take a little longer to populate, but it shouldn't be more than a few seconds until we start to see data in our Live_Data.

If we want to test out our endpoints, spin up the React front end I provided in the GitHub repo that allows us to look at our data and aggregations, graphed.

If you want to test out the endpoints manually, we can run these curl commands:

# All companies
curl -v http://localhost:8080/stocks/companies

# Historical data for AAPL
curl -v "http://localhost:8080/stocks/companyStocksBetween?company=AAPL&startDate=2023-01-01&endDate=2023-02-01"

# SMA
curl -v "http://localhost:8080/stocks/sma?company=AAPL&startDate=2023-01-01&endDate=2023-02-01"

# EMA
curl -v "http://localhost:8080/stocks/ema?company=AAPL&startDate=2023-01-01&endDate=2023-02-01"

# Live data
curl -v http://localhost:8080/stocks/live

Conclusion

This isn’t just another toy project pretending to be real-time. We’ve built a proper streaming architecture—one that ingests live data, enriches it on the fly with RSI, persists it with zero ceremony using Spring Data MongoDB, and exposes it through a clean, RESTful interface. On the historical side, we’ve leaned on MongoDB’s aggregation framework to calculate indicators like SMA and EMA directly in the database, avoiding the overhead of processing everything in memory or shipping it off to a separate analytics engine.

If you want to learn more about what can be done with MongoDB and Kafka, check out my tutorial, Building a Real-Time AI Fraud Detection System with Spring Kafka and MongoDB, or learn how to create an app in Demand Forecasting With MongoDB Time Series and Edge-Server over on Medium.

Building a Real-Time AI Fraud Detection System with Spring Kafka and MongoDB

Tim Kelly — Wed, 16 Apr 2025 14:00:00 +0000

In this tutorial, we'll build a real-time fraud detection system using MongoDB Atlas Vector Search, Apache Kafka, and AI-generated embeddings. We'll demonstrate how MongoDB Atlas Vector Search can be used to detect anomalies in a stream of financial transactions by analyzing a user's transaction history and identifying suspicious behavior based on LLM-generated embeddings.

Our solution will monitor MongoDB Change Streams using the Java synchronous driver, triggering vector searches on each new transaction to detect potential fraud. While this approach works well for our demo, and for many use cases, we’ll also discuss its limitations. Throughout the tutorial, I’ll cover alternative strategies to optimize performance, whether you need higher transaction throughput, faster fraud detection, or a more scalable architecture.

If you want to view the completed code for this tutorial, you can find it on GitHub.

What we are building

Our real-time fraud detection pipeline will work like this:

Generate customer profiles: Create synthetic users with predictable spending habits and store them in MongoDB.
Produce transactions: A Kafka producer generates transactions, computes AI embeddings, and sends them to Kafka.
Consume transactions: A Kafka consumer reads transactions from Kafka and records them in MongoDB.
Monitor in real-time: The application listens to MongoDB Change Streams to detect new transactions as they arrive.
Run vector search: Each new transaction is compared against the user’s historical transactions using vector search.
Detect fraud: A transaction is marked as fraud if:
- No similar transactions exist for that user.
- Any of the returned similar transactions are already flagged as fraud.

This pipeline ensures real-time anomaly detection using AI-powered embeddings and vector search.

Prerequisites

To follow along, you'll need:

A MongoDB Atlas account with a free-forever M0 (or higher) cluster set up.
Kafka 3.5+ (local installation or Docker).
Java 21 set up on your machine.
OpenAI API key (for embeddings).
Maven 3.9+.

Create our MongoDB database

Log into MongoDB Atlas and create a free-forever M0 cluster. Create a new database called fraud. In this database, we need to create two collections.

customers
transactions

For this simple demo, we are just going to create basic collections because, as of writing this, time series collections do not support search indexes. If you want to learn more about setting these up for your Java application, check out our tutorial Handle Time Series Data with MongoDB.

Create a Vector Search index

We are going to create a vector search index on our transactions collection. This will allow us to find semantically similar documents, to help us identify anomalies in our customer transactions, and mark these as fraud.

How does this work? Well, we create vector embeddings (numerical representations of our data) of our transactions, to store in our vector store, the transactions collection. We get these by sending the information from our transactions to an embedding model, like OpenAI's text-embedding-3-small. These embeddings look like an array of floating point numbers, e.g., [2.0457, -3.1417895, ...].

We then use MongoDB's Atlas Vector Search, which uses a nearest neighbour algorithm to decide what transactions are semantically similar. These transactions are returned, and if the new transaction does not match any past transactions for that user, or if it matches transactions marked fraud, it could be an anomaly (potential fraud).

You can learn the in-depth steps to set up a Vector Search index in Atlas Vector Search Quick Start, but briefly:

Navigate to MongoDB Atlas → Collections → Click transactions.
Select → Atlas Search → Create Search Index + Vector Search.

Use the following JSON:

{
  "fields": [
    {
      "type": "vector",
      "path": "embedding",
      "numDimensions": 1536,
      "similarity": "dotProduct"
    }
  ]
}

Click create index.

We are using 1536 for our number of dimensions because of the embedding model we will be using. OpenAI provides the text-embedding-3-small for embedding (among others). This is a general purpose model embedding text, but using a model tailored for financial fraud detection, especially a custom model tailored for and trained on your data, will lead to better results. Whether you're looking for faster embeddings or more accurate predictions, you can gain some control of that here.

Since OpenAI embeddings are optimized for using the dot product algorithm, that is the similarity method we will stick with. It's also worth noting that if fraudulent transactions tend to have stronger signals (e.g., very different merchants, categories, amounts), dot product helps capture that magnitude, along with the direction. Essentially, it naturally emphasises these stronger signals. In contrast, cosine similarity normalizes all vectors to unit length, effectively discarding this useful magnitude information. This means it would treat a small anomaly and a large one as equally “different” in direction, which can dilute the signal strength we care about when identifying fraud.

Create a Spring application

We are going to create our application using Spring Initializr. Here, we need to set our project to Maven, our language to Java, and our Spring Boot version to 3.4.2 (most recent stable release as of writing). We'll also set our app name—mine is frauddetector—and for packaging, we are going to make a jar, and Java version 21.

We will need a couple of packages too.

Spring Web
Spring for Apache Kafka
Spring Data MongoDB
OpenAI

Download and unzip your Spring application. Open the application in the IDE of your choosing. Open the pom.xml file and add the necessary Jackson dependency.

<dependency>  
    <groupId>com.fasterxml.jackson.core</groupId>  
    <artifactId>jackson-databind</artifactId>  
</dependency>

Setting up configuration

Now, open up the application and go to the application.properties. Here, we'll add the various connection strings, tokens, and configuration settings for our dependencies.

MongoDB configuration

Spring Boot will automatically configure MongoDB, so we don’t have to manually create a MongoClient.

spring.application.name=frauddetector
spring.data.mongodb.uri=<YOUR_CONNECTION_STRING>
spring.data.mongodb.database=fraud

Just add your connection string and the database name and we're ready to go.

Spring AI configuration

We use Spring AI with OpenAI's API to generate text embeddings for transaction similarity detection. We will use these embeddings for vector search, allowing us to compare transactions based on their semantic meaning and patterns (overall vibe), rather than raw data.

spring.ai.openai.api-key=<YOUR_OPEN_AI_API_KEY>
spring.ai.openai.embedding.options.model=text-embedding-3-small

text-embedding-3-small is a lightweight embedding model optimized for low-latency vector generation. Alternatively, you could use "text-embedding-3-large" for higher accuracy but at increased cost and latency.

We will also create a Config package and add a MongoDBConfig class.

package com.mongodb.frauddetector.config;

import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import org.bson.Document;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MongoDBConfig {

    private static final String DATABASE_NAME = "fraud";
    private static final String TRANSACTIONS_COLLECTION = "transactions";

    @Bean
    public MongoDatabase fraudDatabase(MongoClient mongoClient) {
        return mongoClient.getDatabase(DATABASE_NAME);
    }

    @Bean
    public MongoCollection<Document> transactionsCollection(MongoDatabase fraudDatabase) {
        return fraudDatabase.getCollection(TRANSACTIONS_COLLECTION);
    }
}

Here, we will add some methods to connect to our MongoDB database that we can reuse through our application.

Kafka configuration

Kafka is a distributed messaging system that allows producers to send messages and consumers to process them asynchronously. In our fraud detection system, we use Kafka to handle real-time transaction ingestion.

Spring Boot provides built-in Kafka support via spring-kafka, allowing us to configure Kafka producers and consumers using application properties.

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=fraud-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=com.mongodb.frauddetector.model

If you are unfamiliar with Apache Kafka, this might look like a wall of gibberish, so let's break down what each property does

Connecting to Kafka

spring.kafka.bootstrap-servers=localhost:9092
- Specifies the Kafka broker address (default is localhost:9092 if running locally).
- The producer and consumer will use this to send and receive messages.

Kafka Producer configuration

spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
- Converts the message key (used for partitioning) into a string.
- We aren't explicitly setting keys, so Kafka will auto-generate them.
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
- Serializes Java objects (transactions) into JSON before sending them to Kafka.
- Required because we send complex objects like Transaction.

Kafka Consumer configuration

spring.kafka.consumer.bootstrap-servers=localhost:9092
- Specifies which Kafka broker the consumer should listen to.
spring.kafka.consumer.group-id=fraud-group
- Defines a consumer group (fraud-group).
- Multiple consumers in this group can share the workload and balance message consumption.
- If a consumer fails, another instance in the group will resume from where it left off.
spring.kafka.consumer.auto-offset-reset=earliest
- Determines where to start consuming messages if no offset exists.
- Options:
- earliest: Start from the beginning of the topic (useful for debugging).
- latest: Only read new messages (default behavior).

Kafka Consumer deserialization

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
- Deserializes message keys into Strings.
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
- Deserializes Kafka messages from JSON into Java objects.
- This is necessary since our transactions are sent as JSON.
spring.kafka.consumer.properties.spring.json.trusted-packages=com.mongodb.frauddetector.model
- Ensures only trusted packages are deserialized.
- Prevents security vulnerabilities from untrusted deserialization.

How it all comes together:

Kafka Producer sends transaction events as JSON to the "transactions" topic.
Kafka Broker routes these events.
Kafka Consumer reads these messages, deserializes them into Transaction objects, and stores them in MongoDB.
MongoDB Change Streams detect new transactions and trigger fraud detection.

Generate our synthetic customer profiles

To make our life simple, we're going to start by adding a few enums. We'll be using these when you're generating our transactions, and defining what is "normal" behaviour for our sample customers. Create a package enums in the Spring app to store these.

First, we'll create a Category enum.

package com.mongodb.frauddetector.enums;  

public enum Category {  
    RETAIL, TECH, GROCERY;  
}

Next, a Currency enum.

package com.mongodb.frauddetector.enums;  

public enum Currency {  
    EUR, USD, GBP;  
}

And finally, a slightly more exciting Merchant enum.

package com.mongodb.frauddetector.enums;  

import java.util.List;  
import java.util.Map;  
import java.util.Random;  

public enum Merchant {  
    // Retail Merchants  
    AMAZON, WALMART, BEST_BUY, TARGET, COSTCO, ETSY, EBAY, IKEA,  

    // Tech Merchants  
    APPLE, MICROSOFT, GOOGLE,  

    // Grocery Merchants  
    DUNNES_STORES, LIDL, TESCO;  

    private static final Random RANDOM = new Random();  

    private static final Map<Category, List<Merchant>> CATEGORY_MERCHANTS = Map.of(  
            Category.RETAIL, List.of(AMAZON, WALMART, BEST_BUY, TARGET, COSTCO, ETSY, EBAY, IKEA),  
            Category.TECH, List.of(APPLE, MICROSOFT, GOOGLE),  
            Category.GROCERY, List.of(DUNNES_STORES, LIDL, TESCO)  
    );  

    public static Merchant getRandomMerchant(Category category) {  
        List<Merchant> merchants = CATEGORY_MERCHANTS.get(category);  
        return merchants.get(RANDOM.nextInt(merchants.size()));  
    }  
}

Here, we have a bit of logic to map our merchants to different categories, and a function to get a random merchant, to help with our transaction generation later (for testing and demo purposes).

The customer model

Our customer model is going to outline some information about our sample customers that will help us to define their spending habits. Since we’re building a fraud detection system, we need a way to recognize when a user makes a purchase that doesn't align with their usual behavior. Create a model package and add a Customer class.

package com.mongodb.frauddetector.model;  

import com.mongodb.frauddetector.enums.Category;  
import com.mongodb.frauddetector.enums.Currency;  
import com.mongodb.frauddetector.enums.Merchant;  
import org.springframework.data.annotation.Id;  
import org.springframework.data.mongodb.core.mapping.Document;  

import java.util.List;  
import java.util.Random;  

@Document(collection = "customers")  
public class Customer {  
    @Id  
    private String id;  
    private final String userId;  
    private final List<Merchant> merchants; // Trusted merchants  
    private final List<Category> categories; // Trusted categories  
    private final Double meanSpending;  
    private final Double spendingStdDev;  
    private final Currency preferredCurrency;  

    public Customer(String userId, List<Merchant> merchants, List<Category> categories,  
                    Double meanSpending, Double spendingStdDev, Currency preferredCurrency) {  
        this.userId = userId;  
        this.merchants = merchants;  
        this.categories = categories;  
        this.meanSpending = meanSpending;  
        this.spendingStdDev = spendingStdDev;  
        this.preferredCurrency = preferredCurrency;  
    }  

    public String getId() { return id; }  
    public String getUserId() { return userId; }  
    public List<Merchant> getMerchants() { return merchants; }  
    public List<Category> getCategories() { return categories; }  
    public Double getMeanSpending() { return meanSpending; }  
    public Double getSpendingStdDev() { return spendingStdDev; }  
    public Currency getPreferredCurrency() { return preferredCurrency; }  

}

Each customer in our system has:

A unique user ID: This lets us associate transactions with a specific person.
Trusted merchants: The places they frequently shop.
Trusted categories: The types of purchases they regularly make.
Average spending habits: Their typical purchase amount and the standard deviation.
Preferred currency: The currency they usually transact in.

This allows us to establish a baseline for their spending behavior. If a new transaction deviates too much, we can flag it as potentially fraudulent. We need to add a couple of helper functions to help us in our data generation. We’ll create the getFrequentCategory() method, to randomly select one of the user's preferred categories. Open the Customer class and add the following method:

    public Category getFrequentCategory() {  
        Random random = new Random();  
        return categories.get(random.nextInt(categories.size()));  
    }

We’ll also add the getUnfrequentCategory() method, to pick a category they don’t usually spend money on—which could be a red flag.

    public Category getUnfrequentCategory() {  
        // Get all categories from the enum  
        List<Category> allCategories = List.of(Category.values());  

        // Filter out frequent categories  
        List<Category> infrequentCategories = allCategories.stream()  
                .filter(category -> !categories.contains(category)) 
                .toList();  

        // Pick a random category from the remaining ones  
        Random random = new Random();  
        return infrequentCategories.get(random.nextInt(infrequentCategories.size()));  
    }

If a user who only shops at grocery stores suddenly makes a high-ticket tech purchase, we might need to flag that as fraud.

Currency changes can be another strong indicator of fraud. If a user always transacts in USD, but suddenly makes a purchase in EUR, that could signal fraudulent activity.

    public Currency getRandomSuspiciousCurrency() {  
        // Get all categories from the enum  
        List<Currency> allCurrency = List.of(Currency.values());  

        // Filter out frequent categories  
        List<Currency> infrequentCurrency = allCurrency.stream()  
                .filter(currency -> !(preferredCurrency == currency)) 
                .toList();  

        // Pick a random category from the remaining ones  
        Random random = new Random();  
        return infrequentCurrency.get(random.nextInt(infrequentCurrency.size()));  
    }

This method helps simulate a realistic fraud scenario, where someone might steal a card and use it overseas.

Customer seeding

Before we can generate transactions, we need some customers in our database. These aren't just random users—we're giving them specific spending habits, preferred merchants, and common transaction patterns. This will help us create realistic data for fraud detection. Create a service package and add a CustomerSeeder class. Copy the following code into it, then we'll break down what is happening.

package com.mongodb.frauddetector.service;  

import com.mongodb.frauddetector.enums.Category;  
import com.mongodb.frauddetector.enums.Currency;  
import com.mongodb.frauddetector.enums.Merchant;  
import com.mongodb.frauddetector.model.Customer;  
import jakarta.annotation.PostConstruct;  
import org.slf4j.Logger;  
import org.slf4j.LoggerFactory;  
import org.springframework.data.mongodb.core.MongoTemplate;  
import org.springframework.stereotype.Service;  

import java.util.List;  

@Service  
public class CustomerSeeder {  
    private static final Logger logger = LoggerFactory.getLogger(CustomerSeeder.class);  
    private final MongoTemplate mongoTemplate;  

    public CustomerSeeder(MongoTemplate mongoTemplate) {  
        this.mongoTemplate = mongoTemplate;  
    }  

    @PostConstruct  
    public void seedCustomers() {  
        if (mongoTemplate.getCollection("customers").countDocuments() > 0) {  
            logger.info("Customers already exist. Skipping seed.");  
            return;  
        }  

        List<Customer> customers = List.of(  
                new Customer("user_1", List.of(Merchant.AMAZON, Merchant.BEST_BUY), List.of(Category.TECH, Category.RETAIL), 150.0, 30.0, Currency.USD),  
                new Customer("user_2", List.of(Merchant.WALMART, Merchant.TARGET, Merchant.DUNNES_STORES), List.of(Category.RETAIL, Category.GROCERY), 80.0, 20.0, Currency.USD),  
                new Customer("user_3", List.of(Merchant.APPLE, Merchant.MICROSOFT), List.of(Category.TECH), 250.0, 50.0, Currency.EUR),  
                new Customer("user_4", List.of(Merchant.ETSY, Merchant.BEST_BUY), List.of(Category.RETAIL), 100.0, 25.0, Currency.EUR),  
                new Customer("user_5", List.of(Merchant.ETSY, Merchant.EBAY), List.of(Category.RETAIL), 90.0, 20.0, Currency.GBP),  
                new Customer("user_6", List.of(Merchant.TESCO, Merchant.DUNNES_STORES), List.of(Category.GROCERY), 40.0, 10.0, Currency.EUR),  
                new Customer("user_7", List.of(Merchant.LIDL, Merchant.COSTCO), List.of(Category.GROCERY), 35.0, 8.0, Currency.EUR),  
                new Customer("user_8",  List.of(Merchant.GOOGLE, Merchant.MICROSOFT), List.of(Category.TECH), 15.0, 5.0, Currency.USD),  
                new Customer("user_9", List.of(Merchant.EBAY, Merchant.ETSY), List.of(Category.RETAIL), 60.0, 15.0, Currency.GBP),  
                new Customer("user_10", List.of(Merchant.COSTCO, Merchant.IKEA), List.of(Category.RETAIL), 25.0, 7.0, Currency.GBP)  
        );  

        mongoTemplate.insertAll(customers);  
        logger.info("Customers seeded successfully!");  
    }  
}

Now, let’s walk through the implementation.

You might be wondering why we’re not using MongoRepository for this. While MongoRepository is great for simple CRUD operations, it doesn't provide full control over how we seed our database.

Using MongoTemplate, we can directly check if data exists before inserting (so we don’t create duplicates).

We use Spring’s @PostConstruct annotation to run the seedCustomers() method automatically when the application starts. This ensures our database is populated before transactions start flowing.

@PostConstruct
public void seedCustomers() {

If customer data already exists in the MongoDB collection, we skip seeding.
Otherwise, we generate 10 customers, each with:
- A user ID.
- A list of trusted merchants.
- A list of frequent categories.
- A mean spending amount and standard deviation.
- A preferred currency.

Once the data is created, we use:

mongoTemplate.insertAll(customers);

Thanks to the magic of Spring and our @PostContruct annotation, mongoTemplate will bulk insert everything into our MongoDB database just by running the application.

How the sample customers are structured

Each customer is designed to behave predictably. For the sake of this demonstration, very predictably:

Tech shoppers: Spend more on Amazon, Best Buy, and Apple.
Grocery shoppers: Frequent Tesco, Dunnes Stores, or Lidl.
General retail shoppers: Shop at Etsy, eBay, or Walmart.
Spending amounts vary but follow a general pattern.

With this customer seeding in place:

We have a structured dataset to test fraud detection.
Customers behave in predictable ways, making anomalies easier to catch.

Once the customers are in MongoDB, we can move on to generating transactions.

The transaction model

The Transaction model represents individual financial transactions in our system. These transactions are generated based on our synthetic customer profiles and are stored in MongoDB. This model plays a key role in detecting anomalies by allowing us to compare each new transaction against a user’s historical behavior.

Create a Transaction class inside the model package:

package com.mongodb.frauddetector.model;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.mongodb.frauddetector.enums.Category;
import com.mongodb.frauddetector.enums.Currency;
import com.mongodb.frauddetector.enums.Merchant;
import org.bson.BsonDocument;
import org.bson.codecs.configuration.CodecRegistry;
import org.bson.conversions.Bson;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
import java.time.Instant;
import java.util.Random;
import java.util.UUID;

import static com.mongodb.frauddetector.enums.Merchant.getRandomMerchant;

@JsonIgnoreProperties(ignoreUnknown = true)
@Document(collection = "transactions")
public class Transaction {
    @Id
    private String id;
    @JsonProperty("transaction_id")
    private String transactionId;
    private String userId;
    private double amount;
    private Currency currency;
    private Instant timestamp;
    private Merchant merchant;
    private Category category;
    private boolean isFraud;
    private float[] embedding = {};

    public Transaction() {}

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getTransactionId() {
        return transactionId;
    }

    public void setTransactionId(String transactionId) {
        this.transactionId = transactionId;
    }

    public String getUserId() {
        return userId;
    }

    public void setUserId(String userId) {
        this.userId = userId;
    }

    public double getAmount() {
        return amount;
    }

    public void setAmount(double amount) {
        this.amount = amount;
    }

    public Currency getCurrency() {
        return currency;
    }

    public void setCurrency(Currency currency) {
        this.currency = currency;
    }

    public Instant getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(Instant timestamp) {
        this.timestamp = timestamp;
    }

    public Merchant getMerchant() {
        return merchant;
    }

    public void setMerchant(Merchant merchant) {
        this.merchant = merchant;
    }

    public Category getCategory() {
        return category;
    }

    public void setCategory(Category category) {
        this.category = category;
    }

    public boolean isFraud() {
        return isFraud;
    }

    public void setFraud(boolean fraud) {
        isFraud = fraud;
    }

    public float[] getEmbedding() {
        return embedding;
    }

    public void setEmbedding(float[] embedding) {
        this.embedding = embedding;
    }

}

Each transaction has:

A unique transaction ID: Used to track and reference individual transactions.
A user ID: Links the transaction to a specific customer.
Amount: The total cost of the transaction.
Currency: The currency in which the transaction was made.
Timestamp: The exact time the transaction took place.
Merchant: The vendor where the transaction occurred.
Category: The type of purchase made.
Fraud flag: A boolean value indicating whether the transaction is marked as fraudulent.
Embedding: A numerical representation of the transaction generated via AI embeddings, which we’ll use for vector search.

At the top of our class, we specify which collection we want our financial transactions stored in, @Document(collection = "transactions"). Spring Data MongoDB handles the BSON serialization for us.

To detect fraudulent transactions, we’ll use vector search to compare transactions based on their overall context rather than exact matching. The generateEmbeddingText() method creates a string representation of the transaction, which we’ll later convert into an embedding.

public String generateEmbeddingText() {
    return userId + " " + amount + " " + currency + " " + merchant + " " + category;
}

Generating random transactions

We need to simulate realistic transaction data for our customers. The generateRandom() method creates our synthetic transactions that either align with or deviate from a user's normal spending patterns.

public static Transaction generateRandom(Customer customer) {
    Random random = new Random();

    // Generate a normal or suspicious transaction
    boolean isSuspicious = random.nextDouble() < 0.1; // 10% chance of fraud

    double amount = isSuspicious ? customer.getMeanSpending() * (2 + random.nextDouble()) // Unusually large
            : customer.getMeanSpending() * (0.5 + random.nextDouble());

    Category category = isSuspicious ? customer.getUnfrequentCategory() : customer.getFrequentCategory();

    Merchant merchant = getRandomMerchant(category);

    Currency currency = isSuspicious ? customer.getRandomSuspiciousCurrency()
            : customer.getPreferredCurrency();

    return new Transaction(
            null,
            UUID.randomUUID().toString(),
            customer.getUserId(),
            amount,
            currency,
            Instant.now(),
            merchant,
            category,
            false
    );
}

This method randomly decides whether the transaction is suspicious (10% chance). Feel free to adjust this to suit your use case you want to test for. It then sets the amount to either a normal range or an unusually high value. Next, we choose a category and merchant based on the user’s normal spending behavior (or select a category they don’t usually shop in if the transaction is suspicious). Lastly, we select a currency, potentially choosing an unusual one for suspicious transactions.

This Transaction model is the backbone of our fraud detection pipeline. Once generated, transactions will be sent to Kafka, processed, and stored in MongoDB, where we’ll use MongoDB Change Streams and Vector Search to flag potential fraud.

How do we create our AI embeddings?

For this example, we are going to use the OpenAI API embedding model for no other reason than it's simple. It is general purpose so it will work, and Spring AI makes it very straightforward to set up. Now, I highly recommend looking at a tailored embedding model for fraud detection for your application. If you want to learn how to do this, check out my tutorial on how to train your own fraud detection model in Java.

With our API key and embedding model configured in the application.properties from earlier, we need to create a config package, and add a new OpenAIConfig class.

Here, we just need to create a configuration class that takes in the API key, and sets up the OpenAI embedding model using Spring AI.

package com.mongodb.frauddetector.config;  
import org.springframework.ai.embedding.EmbeddingModel;  
import org.springframework.ai.openai.OpenAiEmbeddingModel;  
import org.springframework.ai.openai.api.OpenAiApi;  
import org.springframework.beans.factory.annotation.Value;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  

@Configuration  
public class OpenAIConfig {  

    @Value("${spring.ai.openai.api-key}")  
    private String apiKey;  

    @Bean  
    public EmbeddingModel embeddingModel() {  
        return new OpenAiEmbeddingModel(new OpenAiApi(apiKey));  
    }  
}

Now, our embedding model is ready to be called upon for generating our vectors to store alongside our transactions in MongoDB.

With our embedding model set up, in our Service package, we can add our new class, EmbeddingGenerator.

package com.mongodb.frauddetector.service;

import org.springframework.ai.embedding.EmbeddingModel;
import org.springframework.stereotype.Component;

@Component
public class EmbeddingGenerator {
    private final EmbeddingModel embeddingModel;

    public EmbeddingGenerator(EmbeddingModel embeddingModel) {
        this.embeddingModel = embeddingModel;
    }

    public float[] getEmbedding(String transaction) {
        float[] embeddingResponse = this.embeddingModel.embed(transaction);
        return embeddingResponse;
    }
}

This will take the information we want to include in our embedding for each transaction, and generate this embedding using our embedding model.

The @Component notation allows Spring to detect our custom beans automatically. Basically, it lets us instantiate a class and inject any specified dependencies into it, and inject them wherever needed.

How to simply interact with our database with Spring Data MongoDB

For our simple CRUD operations for our customers and transactions, finds, and inserts mostly, we can use MongoRepository to get this done nice and simple. Create a package repository.

First, we'll create a CustomerRepository interface.

package com.mongodb.frauddetector.repository;  

import com.mongodb.frauddetector.model.Customer;  
import org.springframework.data.mongodb.repository.MongoRepository;  

public interface CustomerRepository extends MongoRepository<Customer, String> {  
}

Next, we'll create a TransactionRepository interface.

package com.mongodb.frauddetector.repository;  

import com.mongodb.frauddetector.model.Transaction;  
import org.springframework.data.mongodb.repository.MongoRepository;  

public interface TransactionRepository extends MongoRepository<Transaction, String> {  
}

We don't need to do anything else. MongoRepository provides us with all the CRUD operations we need. We don't need to create any custom query implementations. If you want to see more about what MongoRepository provides, and the differences between MongoRepository and MongoTemplate, check out our article about getting started with Spring Data MongoDB.

Seed the transactions for our customers

Now that we have customers in our database, we need to generate some transactions for them. These transactions will serve as our baseline data, allowing us to compare future transactions against historical spending patterns.

Create a TransactionSeeder class inside the service package and copy the following code.

package com.mongodb.frauddetector.service;

import com.mongodb.client.MongoCollection;
import com.mongodb.frauddetector.model.Customer;
import com.mongodb.frauddetector.model.Transaction;
import com.mongodb.frauddetector.repository.CustomerRepository;
import com.mongodb.frauddetector.repository.TransactionRepository;
import jakarta.annotation.PostConstruct;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

@Service
public class TransactionSeeder {
    private static final Logger logger = LoggerFactory.getLogger(TransactionSeeder.class);
    private final CustomerRepository customerRepository;
    private final EmbeddingGenerator embeddingGenerator;
    private final TransactionRepository transactionRepository;
    private final TransactionChangeStreamListener transactionChangeStreamListener;
    private final MongoCollection<Document> transactionsCollection;

    public TransactionSeeder(CustomerRepository customerRepository,
                             EmbeddingGenerator embeddingGenerator, TransactionRepository transactionRepository,
                             TransactionChangeStreamListener transactionChangeStreamListener, MongoCollection<Document> transactionsCollection) {
        this.transactionsCollection = transactionsCollection;
        this.transactionRepository = transactionRepository;
        this.customerRepository = customerRepository;
        this.embeddingGenerator = embeddingGenerator;
        this.transactionChangeStreamListener = transactionChangeStreamListener;
    }

}

This service is responsible for generating sample transactions for each customer and creating the embeddings, then storing them in MongoDB.

It is also responsible for starting the Change Stream listener to detect real-time transactions after the database has been seeded. We make sure to do this after the database has been seeded. We are again accessing the MongoDB Java driver directly, to allow us to verify if the collection has already been seeded.

Next, we need a post construct method, seedTransactions().

   @PostConstruct
    public void seedTransactions() {
        if (transactionsCollection.countDocuments() > 0) {
            logger.info("Transactions already seeded.");
            return;
        }

        List<Customer> customers = customerRepository.findAll();
        List<Transaction> transactions = new ArrayList<>();

        for (Customer customer : customers) {
            for (int i = 0; i < 10; i++) {
                Transaction transaction = Transaction.generateRandom(customer);
                String embeddingText = transaction.generateEmbeddingText();
                float[] embedding = embeddingGenerator.getEmbedding(embeddingText);
                transaction.setEmbedding(embedding);
                transactions.add(transaction);
            }
        }

        transactionRepository.saveAll(transactions);
        logger.info("Seeded 100 transactions.");

        transactionChangeStreamListener.startListening();
        logger.info("Change Stream Listener Started.");
    }

Once the service is initialized, the @PostConstruct method runs, ensuring that our transactions are created before the system starts processing new ones. We check if the transactions collection already contains data. If it does, we skip the seeding process. Otherwise, we generate 10 transactions per customer.

We fetch all customers from the database. Each customer gets a fixed amount of 10 historical transactions, and each transaction is randomly generated based on the customer's typical spending habits.

Finally, we batch insert our transactions, and we start listening for real-time transactions so that new ones can be evaluated for fraud detection.

How we'll send our transactions with a Kafka Producer

Now that we've seeded our customer data and generated initial transactions, we need a way to continuously produce transactions in real time. We'll be using a Kafka Producer for this. The producer will generate synthetic transactions, compute their embeddings, and send them to Kafka for further processing.

Kafka follows a publish-subscribe model:

Producers send messages to a Kafka topic (in this case, "transactions").
Consumers subscribe to that topic and process incoming messages.
Topics store and distribute messages to consumers, ensuring scalability.

Create a TransactionProducer class inside the service package and copy the following code.

package com.mongodb.frauddetector.service;  

import com.mongodb.frauddetector.model.Customer;  
import com.mongodb.frauddetector.model.Transaction;  
import com.mongodb.frauddetector.repository.CustomerRepository;  
import jakarta.annotation.PostConstruct;  
import org.slf4j.Logger;  
import org.slf4j.LoggerFactory;  
import org.springframework.kafka.core.KafkaTemplate;  
import org.springframework.scheduling.annotation.Scheduled;  
import org.springframework.stereotype.Service;  
import java.util.List;  
import java.util.Random;  

@Service  
public class TransactionProducer {  
    private static final Logger logger = LoggerFactory.getLogger(TransactionProducer.class);  

    private static final String TOPIC = "transactions";  
    private final EmbeddingGenerator embeddingGenerator;  
    private final KafkaTemplate<String, Transaction> kafkaTemplate;  
    private List<Customer> customers;  
    private final Random random = new Random();  
    private final CustomerRepository customerRepository;  

    public TransactionProducer(KafkaTemplate<String, Transaction> kafkaTemplate, EmbeddingGenerator embeddingGenerator, CustomerRepository customerRepository) {  
        this.kafkaTemplate = kafkaTemplate;  
        this.embeddingGenerator = embeddingGenerator;  
        this.customerRepository = customerRepository;  
    }
}

Spring's Kafka template is used to send messages to a Kafka topic. Next, we need a method to fetch customer data to create transactions. Create a @PostConstruct method.

    @PostConstruct  
    public void loadCustomers() {  
        customers = customerRepository.findAll();  
        if (customers.isEmpty()) {  
            logger.error("Warning: No customers found! Transactions may fail.");  
        } else {  
            logger.info("Cached {} customers for transaction generation.", customers.size());  
        }  
    }

We cache customer data in memory when the application starts. This avoids repeated database lookups and speeds up transaction generation. Now, we need a method to generate our synthetic transactions. We will use the @Scheduled annotation to create a generateAndSendTransaction() method to run every 100ms.

    @Scheduled(fixedRate = 100)  
    public void generateAndSendTransaction() {  
        if (customers == null || customers.isEmpty()) {  
            logger.error("No customers available. Skipping transaction generation.");  
            return;  
        }  
        Transaction transaction = Transaction.generateRandom(customers.get(random.nextInt(customers.size())));  
        String embeddingText = transaction.generateEmbeddingText();  
        transaction.setEmbedding(embeddingGenerator.getEmbedding(embeddingText));  
        kafkaTemplate.send(TOPIC, transaction.getTransactionId(), transaction);  
        logger.info("Transaction sent to topic {}", TOPIC);  
    }

This randomly selects a customer, creates a transaction, and generates an embedding. It then sends the transaction to Kafka.

We also need to add @EnableScheduling to our FrauddetectorApplication class to allow us to schedule these transactions for our demo.

package com.mongodb.frauddetector;  

import org.springframework.boot.SpringApplication;  
import org.springframework.boot.autoconfigure.SpringBootApplication;  
import org.springframework.scheduling.annotation.EnableScheduling;  

@SpringBootApplication  
@EnableScheduling  
public class FrauddetectorApplication {  

    public static void main(String[] args) {  
       SpringApplication.run(FrauddetectorApplication.class, args);  
    }  

}

Now that our Kafka Producer is generating transactions and sending them to a Kafka topic, we need a way to consume them and store them in MongoDB. This is where the Kafka Consumer comes in.

Ingest our transactions

This step is fairly simple using spring-kafka. In the service package, add a TransactionConsumer class. We'll set up a simple Kafka listener to automatically process incoming transaction messages.

package com.mongodb.frauddetector.service;  

import com.mongodb.frauddetector.model.Transaction;  
import com.mongodb.frauddetector.repository.TransactionRepository;  
import org.springframework.kafka.annotation.KafkaListener;  
import org.springframework.stereotype.Service;  

@Service  
public class TransactionConsumer {  

    private final TransactionRepository transactionRepository;  

    public TransactionConsumer(TransactionRepository transactionRepository) {  
        this.transactionRepository = transactionRepository;  
    }  

    @KafkaListener(topics = "transactions", groupId = "fraud-group")  
    public void consumeTransaction(Transaction transaction) {  
        transactionRepository.save(transaction);  
    }  
}

How Kafka Consumers work

Kafka follows a publish-subscribe model:

Producers publish messages to a topic (in this case, "transactions").
Consumers subscribe to that topic and process incoming messages.
Consumers can be grouped into consumer groups, where each instance reads a portion of the messages.

For our system:

Our Kafka Producer generates transactions and publishes them to Kafka.
Our Kafka Consumer listens to the "transactions" topic.
Each transaction is stored in MongoDB via TransactionRepository.

With our transactions being produced and consumed, we can now look at how we will monitor incoming transactions on our MongoDB database.

Monitor our database with Change Streams

Once transactions are stored in MongoDB, we need a way to detect new transactions in real time and trigger fraud detection immediately. Instead of repeatedly querying the database for new data, we can use MongoDB Change Streams to efficiently listen for changes.

What is a change stream?

MongoDB Change Streams provide a real-time feed that allows applications to listen for inserts, updates, deletes, and replacements happening in a collection or database. It works similarly to an event listener, triggering an action whenever a change occurs.

Instead of running queries to check for new transactions, Change Streams push updates to our application as soon as they occur, making fraud detection more efficient and reducing unnecessary database calls.

So why use Change Streams?

Real-time monitoring: Detects fraud immediately after a transaction is inserted.
Push-based updates: Eliminates the need for polling or manual querying.
Optimized performance: Runs efficiently, even for high-velocity transaction data.
Built-in scalability: Works across sharded clusters and replica sets.

To monitor transactions, we create a TransactionChangeStreamListener service in our service package.

package com.mongodb.frauddetector.service;  

import com.mongodb.client.*;  
import com.mongodb.client.model.Aggregates;  
import com.mongodb.client.model.Filters;  
import com.mongodb.client.model.changestream.ChangeStreamDocument;  
import org.bson.Document;  
import org.bson.conversions.Bson;  
import org.slf4j.Logger;  
import org.slf4j.LoggerFactory;  
import org.springframework.stereotype.Service;  

import java.util.List;  
import java.util.concurrent.Executors;  
import java.util.concurrent.ExecutorService;  

@Service  
public class TransactionChangeStreamListener {  
    private static final Logger logger = LoggerFactory.getLogger(TransactionChangeStreamListener.class);  

   private final TransactionVectorSearchService vectorSearchService;
    private final ExecutorService executorService = Executors.newSingleThreadExecutor(); // Keeps it synchronous
    private final MongoCollection<Document> transactionsCollection;

    public TransactionChangeStreamListener(TransactionVectorSearchService vectorSearchService, MongoCollection<Document> transactionsCollection) {
        this.transactionsCollection = transactionsCollection;
        this.vectorSearchService = vectorSearchService;
    }
}

We need to create a method startListening() to begin monitoring our database.

public void startListening() {
        executorService.submit(() -> {
            // Filter to only listen for INSERT operations
            List<Bson> pipeline = List.of(Aggregates.match(Filters.eq("operationType", "insert")));

            try (MongoCursor<ChangeStreamDocument<Document>> cursor = transactionsCollection.watch(pipeline).iterator()) {
                while (cursor.hasNext()) {
                    ChangeStreamDocument<Document> change = cursor.next();
                    Document transactionDoc = change.getFullDocument();

                    if (transactionDoc != null) {
                        logger.info("New transaction detected: {}", transactionDoc.getString("transactionId"));

                        List<Double> embedding = transactionDoc.getList("embedding", Double.class);
                        if (embedding != null) {
                            logger.info("Performing vector search");
                            vectorSearchService.evaluateTransactionFraud(transactionDoc);
                        } else {
                            logger.error("Warning: Transaction does not contain an embedding field.");
                        }
                    }
                }
            }
        });
    }

Now, the change stream monitors all operations on our collection. We don't care about the updates or deletes for our use case, so we use the aggregation pipeline to filter for only inserts.

List<Bson> pipeline = List.of(Aggregates.match(Filters.eq("operationType", "insert")));

With each incoming transaction, we extract the embedding and use it to run a vector search (we are going to implement this next). If this returns similar transactions marked fraud, we are going to mark this new transaction as fraud. If we don't find any similar transactions for that customer, we are also going to mark it as fraud. We will go more into the implementation of this logic in the next section.

Fraud detection with vector search

Fraud detection in this system relies on MongoDB Atlas Vector Search, which allows us to compare transaction embeddings against historical transactions to determine if a new transaction is suspicious. Create a class TransactionVectorSearchService in the service package.

package com.mongodb.frauddetector.service;

import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Updates;
import com.mongodb.client.model.search.VectorSearchOptions;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Arrays;

import static com.mongodb.client.model.search.SearchPath.fieldPath;

@Service  
public class TransactionVectorSearchService {  
    private static final Logger logger = LoggerFactory.getLogger(TransactionVectorSearchService.class);  

    private final MongoCollection<Document> transactionCollection;  
    private static final String VECTOR_INDEX_NAME = "vector_index"; // Ensure this matches your Atlas index name  
    private static final int SEARCH_LIMIT = 5; // Number of similar transactions to retrieve  
    private static final int NUM_CANDIDATES = 50; // Number of approximate neighbors to consider  

   public TransactionVectorSearchService(MongoCollection<Document> transactionCollection) {
        this.transactionCollection = transactionCollection;
    }
}

The TransactionVectorSearchService is responsible for:

Running vector search on each new transaction.
Retrieving similar transactions from the database.
Determining fraud likelihood based on similarity.

Now, this first method we will create is the method we are calling in our Change Stream listener. Create the evaluateTransactionFraud() method that will take in our transaction document as a parameter.

public void evaluateTransactionFraud(Document transactionDoc) {  
        String transactionId = transactionDoc.getString("transactionId");  
        String userId = transactionDoc.getString("userId");  
        List<Double> embedding = transactionDoc.getList("embedding", Double.class);  

        // Run vector search to find similar transactions  
        List<Document> similarTransactions = findSimilarTransactions(embedding, userId);  

        // If no similar transactions exist for this user OR any of them are fraud -> Mark as fraud  
        boolean isFraud = similarTransactions.isEmpty() ||  
                similarTransactions.stream().anyMatch(doc -> doc.getBoolean("isFraud", false));  

        if (isFraud) {  
            markTransactionAsFraud(transactionId);  
        }  
    }

First, we're going to extract our transactionId, our UserId, and our Embedding. Next, we'll call a helper function findSimilarTransactions(), and pass in our embeddings and the user.

    List<Document> findSimilarTransactions(List<Double> embedding, String userId) {  
        Bson vectorSearch = Aggregates.vectorSearch(  
                fieldPath("embedding"),  
                embedding,  
                VECTOR_INDEX_NAME,  
                SEARCH_LIMIT,  
                VectorSearchOptions.approximateVectorSearchOptions(NUM_CANDIDATES)  
        );  

        Bson matchUser = Aggregates.match(Filters.eq("userId", userId));  

        return transactionCollection.aggregate(Arrays.asList(vectorSearch, matchUser))  
                .into(new ArrayList<>());  
    }

In findSimilarTransactions(), we create an aggregation pipeline. Here, we set up our vector search to return the top five similar transactions, and filter by the user. We return this to our evaluateTransactionFraud() method, where we run some fraud detection logic.

        boolean isFraud = similarTransactions.isEmpty() ||  
                similarTransactions.stream().anyMatch(doc -> doc.getBoolean("isFraud", false));

This will return isFraud = True if we can't find any similar transactions for that user, or if any of the returned transactions are marked as fraud. If this is the case, we call the method, markTransactionAsFraud(), and pass in the transactionId.

    private void markTransactionAsFraud(String transactionId) {  
        transactionCollection.updateOne(  
                Filters.eq("transactionId", transactionId),  
                Updates.set("isFraud", true)  
        );  
        logger.info("Transaction marked as fraud: {}", transactionId);  
    }

This will update the document in our MongoDB collection.

Now, this logic works perfectly fine for my simple little demo, but will it work for you? Maybe fraudulent transactions look remarkably similar across all your customers. In this case, it would be advisable to not filter by user. Maybe a larger number of transactions need to be passed back to get a more thorough look at your users’ spending history. It might also make sense to maintain a list of likely fraudulent vendors to automatically flag. In the next section, we go through some of the limitations of this implementation and discuss how you may tailor it to your use case.

Limitations of our application

While vector search is a powerful tool for detecting fraudulent transactions, there are a few notable limitations:

MongoDB Atlas Vector Search does not support batching multiple queries. This means each transaction must be evaluated individually, which can lead to performance bottlenecks in high-throughput environments.
Since each transaction must be queried separately, this approach may become slow when processing hundreds upon thousands of transactions per second.

Atlas Triggers for monitoring incoming transactions

To reduce latency and avoid excessive document transfers between the application and MongoDB, leverage MongoDB Atlas Triggers. Instead of running vector search from the application, we can move that logic directly into MongoDB using an Atlas trigger.

Benefits
- Reduces latency. Since the vector search runs within MongoDB, we avoid the overhead of sending documents back and forth between the database and the application.
- Offloads processing to MongoDB. This means the application doesn't need to handle the change stream, freeing up resources for other tasks.
- Ensures real-time fraud detection. As soon as a transaction is stored, it is immediately analyzed for fraud inside MongoDB.
Limitations
- Atlas Triggers have a hard timeout of 30 seconds, which may be problematic for high volumes of transactions or complex fraud detection logic.
- Less flexible than application-side logic. If we need customized fraud detection workflows, it may be harder to modify compared to handling it in our application.
- The trigger will still need to run a vector search for each transaction, which means batching remains a challenge.
Implementation
- MongoDB Atlas supports triggers that can execute logic when new documents are inserted into a collection.
- Instead of listening for changes in the application, we can set up a trigger on the transactions collection.
- When a new transaction is inserted, the trigger can run a vector search within MongoDB and flag the transaction as fraud if necessary.

Use Case	Best Approach
Low-latency fraud detection	MongoDB Trigger
Need for flexible fraud detection rules	Application Logic
High transaction volume (thousands/sec)	Triggers (30s limit)—consider batching in-memory
Want to avoid excessive application queries	MongoDB Trigger

Run our application

So with our fraud detection pipeline set up, it's time to run our application. First, we need to start our Kafka.

Starting Apache Kafka

Before starting Kafka for the first time, we need to format its metadata log directory. This only needs to be done once.

export KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties

This initializes the internal metadata quorum and sets up Kafka’s KRaft storage directory. Now, we start the Kafka broker, using our config.

bin/kafka-server-start.sh config/server.properties

The broker handles all messaging: storing, forwarding, and distributing messages between producers and consumers. Since we’re running Kafka in standalone mode, this single broker is responsible for everything. If everything starts successfully, you should see Kafka logs indicating that the broker is up and running.

Lastly, we need to open up a terminal and create our transactions topic.

bin/kafka-topics.sh --create \
  --topic transactions \
  --bootstrap-server localhost:9092 \
  --partitions 1 \
  --replication-factor 1

This command creates a topic named transactions. It uses one partition for simplicity, but in production, you would likely want more for scalability. It also sets the replication factor to 1, since we're running only a single broker.

Run our Spring app

Now that we have our Kafka up and running, time to run our app.

mvn clean spring-boot:run

There will be some start-up time, about 100 seconds, to generate the sample data and populate the database. If everything is working, you should see logs indicating:

Customers have been seeded.
Transactions are being generated and sent to Kafka.
Kafka Consumer is saving transactions in MongoDB.
Change Stream is monitoring new transactions.
Vector Search is detecting fraud in real-time.

hangeStreamListener  : Performing vector search
2025-02-24T15:29:56.479Z  INFO 81047 --- [frauddetector] [   scheduling-1] c.m.f.service.TransactionProducer        : Transaction sent to topic transactions
2025-02-24T15:29:56.520Z  INFO 81047 --- [frauddetector] [pool-2-thread-1] c.m.f.s.TransactionChangeStreamListener  : New transaction detected: 3ddc2181-5bb3-42b8-8b7b-953aaf098429
2025-02-24T15:29:56.520Z  INFO 81047 --- [frauddetector] [pool-2-thread-1] c.m.f.s.TransactionChangeStreamListener  : Performing vector search
2025-02-24T15:29:56.628Z  INFO 81047 --- [frauddetector] [pool-2-thread-1] c.m.f.s.TransactionVectorSearchService   : Transaction marked as fraud: 3ddc2181-5bb3-42b8-8b7b-953aaf098429
2025-02-24T15:29:56.839Z  INFO 81047 --- [frauddetector] [   scheduling-1] c.m.f.service.TransactionProducer        : Transaction sent to topic transactions
2025-02-24T15:29:56.881Z  INFO 81047 --- [frauddetector] [pool-2-thread-1] c.m.f.s.TransactionChangeStreamListener  : New transaction detected: 7354568d-7a39-4708-904b-b14c913ce2d0
2025-02-24T15:29:56.881Z  INFO 81047 --- [frauddetector] [pool-2-thread-1] c.m.f.s.TransactionChangeStreamListener  : Performing vector search
2025-02-24T15:29:57.050Z  INFO 81047 --- [frauddetector] [   scheduling-1] c.m.f.service.TransactionProducer        : Transaction sent to topic transactions

Conclusion

So, we now have a Spring app that can stream in financial transactions from Apache Kafka, and upon insertion into the database, can use Atlas Vector Search to indicate whether they appear to be fraudulent.

If you found this tutorial useful, check out more about what you can do with MongoDB in Java, such as building invincible applications with Temporal and MongoDB.

How to Make a RAG Application With LangChain4j

Tim Kelly — Thu, 13 Feb 2025 17:13:54 +0000

Retrieval-augmented generation, or RAG, introduces some serious capabilities to your large language models (LLMs). These applications can answer questions about your specific corpus of knowledge, while leveraging all the nuance and sophistication of a traditional LLM.

This tutorial will take you through the ins and outs of creating a Q&A chatbot using RAG. The application will:

Retrieve data from a MongoDB Atlas database.
Embed and store documents as vector embeddings.
Use LangChain4J to query the database and augment LLM prompts with the retrieved data.
Enable secure, scalable, and efficient AI-powered applications.

If you want to see the completed application, it is available in the GitHub repository.

Why use RAG?

RAG works by retrieving relevant data from your knowledge base and using that information to enrich the input to the LLM. Here are some of the major benefits:

Sensitive data management: RAG allows you to use sensitive or proprietary data without incorporating it into the LLM’s training set. This ensures data privacy and security while still enabling intelligent responses, particularly useful for instances such as GDPR compliance.
Real-time updates: Instead of retraining the model to include the latest information, an expensive process, RAG enables real-time updates by pulling current data from your knowledge base.
Increased relevance: By grounding responses in your own corpus of data, RAG ensures answers are both accurate and contextually relevant.

Use cases for RAG

RAG is well-suited for a wide variety of applications, including:

Q&A applications: Answer user queries with precision, grounded in your company’s documentation, FAQs, or internal knowledge base.
Customer support chatbots: Personalize interactions by referencing customer history, CRM data, and past interactions.
Dynamic BI tools: Enable business intelligence applications to provide insights using live operational data from databases or spreadsheets.

LangChain4J for RAG

LangChain4J is a Java-based library designed to simplify the integration of LLMs into Java applications by abstracting away a lot of the necessary components in your AI applications. It offers an extensive toolbox for building applications powered by retrieval-augmented generation, enabling us to build quicker and create modular applications.

LangChain4J provides the building blocks to streamline your RAG implementation while maintaining full control over the underlying architecture.

MongoDB for RAG

MongoDB is an ideal database for RAG implementations due to its:

Native vector search: Store and query vector embeddings directly within MongoDB alongside your operational data, enabling retrieval of relevant context.
Flexible schema: Add new fields or adjust data models easily without the absolute hassle and mayhem of a data migration.
Scalability: Handle high throughput and large datasets with MongoDB’s horizontal scaling.
Operational efficiency: Use MongoDB’s aggregation pipelines, time-series collections, and multimodal capabilities to support both RAG and non-RAG workloads.

Prerequisites

For this tutorial, you will need:

Java 21 or higher.
Maven or Gradle (for managing dependencies):
- We use Maven for this tutorial.
A MongoDB Atlas account with a live cluster.
An OpenAI API key.

Our dependencies

First things first, let's add our dependencies to our POM:

<dependencies>  
    <dependency>  
        <groupId>dev.langchain4j</groupId>  
        <artifactId>langchain4j-open-ai</artifactId>  
        <version>1.0.0-alpha1</version>  
    </dependency>  
    <dependency>  
        <groupId>dev.langchain4j</groupId>  
        <artifactId>langchain4j-mongodb-atlas</artifactId>  
        <version>1.0.0-alpha1</version>  
    </dependency>  
    <dependency>  
        <groupId>dev.langchain4j</groupId>  
        <artifactId>langchain4j</artifactId>  
        <version>1.0.0-alpha1</version>  
    </dependency>  
    <dependency>  
        <groupId>com.fasterxml.jackson.core</groupId>  
        <artifactId>jackson-databind</artifactId>  
        <version>2.18.1</version>  
    </dependency>  
</dependencies>

langchain4j-open-ai:
- Embedding generation: Allows you to use OpenAI's embedding models, like text-embedding-ada-002, for transforming textual data into vector representations.
- Chat model integration: Supports communication with OpenAI’s GPT models (e.g., GPT-3.5, GPT-4), enabling conversational AI capabilities.
- Simplified API calls: Abstracts the complexity of interacting with the OpenAI API, reducing boilerplate code and improving developer productivity.
langchain4j-mongodb-atlas:
- Embedding store management: Simplifies storing and retrieving embeddings in MongoDB, making it an ideal solution for RAG applications.
- Vector search support: Enables high-performance vector similarity queries using MongoDB Atlas's native capabilities.
- Metadata handling: Allows storing and querying additional metadata associated with embeddings, which is vital for building rich, context-aware systems.
langchain4j: Provides the tools for building RAG workflows. It includes:
- Classes for text segmentation, document splitting, and chunking, which help break down large documents into manageable pieces.
- Utilities for connecting and orchestrating various components like embedding models, vector stores, and content retrievers.
Jackson Databind simplifies the process of loading and handling JSON data.

Setting up MongoDB and our embedding store

To make our retrieval-augmented generation application work effectively, we need a robust and scalable solution for storing and querying embeddings. MongoDB, with its Atlas Search capabilities, serves as the backbone for this task. In this section, we’ll walk through how to set up MongoDB and configure an embedding store using LangChain4J’s MongoDB integration.

MongoDB setup

The first step is to initialize a connection to our MongoDB cluster. We use the MongoClient from the MongoDB Java driver to connect to our database:

package com.mongodb;  

import com.fasterxml.jackson.databind.JsonNode;  
import com.fasterxml.jackson.databind.ObjectMapper;  
import com.mongodb.client.MongoClient;  
import com.mongodb.client.MongoClients;  
import com.mongodb.client.model.CreateCollectionOptions;  
import dev.langchain4j.data.document.Document;  
import dev.langchain4j.data.document.DocumentSplitter;  
import dev.langchain4j.data.document.Metadata;  
import dev.langchain4j.data.embedding.Embedding;  
import dev.langchain4j.data.segment.TextSegment;  
import dev.langchain4j.model.chat.ChatLanguageModel;  
import dev.langchain4j.model.openai.OpenAiChatModel;  
import dev.langchain4j.model.openai.OpenAiEmbeddingModel;  
import dev.langchain4j.model.openai.OpenAiEmbeddingModelName;  
import dev.langchain4j.model.openai.OpenAiTokenizer;  
import dev.langchain4j.rag.content.retriever.ContentRetriever;  
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;  
import dev.langchain4j.store.embedding.*;  
import dev.langchain4j.store.embedding.mongodb.IndexMapping;  
import dev.langchain4j.store.embedding.mongodb.MongoDbEmbeddingStore;  
import dev.langchain4j.service.AiServices;  
import org.bson.conversions.Bson;  
import dev.langchain4j.data.document.splitter.DocumentSplitters;

public class LangChainRagApp {  

    public static void main(String[] args) {  
        try {  
            // MongoDB setup  
            MongoClient mongoClient = MongoClients.create("CONNECTION_URI");  

        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}

Replace "CONNECTION_URI" with your actual MongoDB connection string, which includes your database credentials and cluster information. This connection will be used to interact with the database and perform operations like storing and retrieving embeddings.

We are also adding in our whole host of imports we will use throughout this tutorial. Don't worry, we will go through all these as we add them to our application.

Configuring the embedding store

The embedding store is the corpus of knowledge of our RAG application, where all embeddings and their associated metadata are stored. Let's add a method and call it from our main:

private static EmbeddingStore<TextSegment> createEmbeddingStore(MongoClient mongoClient) {
    String databaseName = "rag_app";
    String collectionName = "embeddings";
    String indexName = "embedding";
    Long maxResultRatio = 10L;
    CreateCollectionOptions createCollectionOptions = new CreateCollectionOptions();
    Bson filter = null;
    Set<String> metadataFields = new HashSet<>();
    IndexMapping indexMapping = new IndexMapping(1536, metadataFields);
    Boolean createIndex = true;

    return new MongoDbEmbeddingStore(
            mongoClient,
            databaseName,
            collectionName,
            indexName,
            maxResultRatio,
            createCollectionOptions,
            filter,
            indexMapping,
            createIndex
    );
}

Let's explore the parameters we set up:

Database name (databaseName) We specify "rag_app" as the database where our embeddings will be stored. You can rename this to suit your application.
Collection name (collectionName) The collection "embeddings" will hold the embedding data and metadata. Collections in MongoDB are analogous to tables in relational databases.
Index name (indexName) The "embedding" index enables efficient vector search operations. This index is crucial for retrieving relevant embeddings quickly based on similarity scores.
Max result ratio (maxResultRatio) This defines the maximum number of results to return during retrieval, keeping the results manageable.
Create collection options (createCollectionOptions) Options for creating the collection can be customized here. For example, you could configure specific validation rules or shard keys.
Filter (filter) Currently set to null, this can be used to define custom filtering criteria if needed for specific retrieval operations.
Metadata fields (metadataFields) A set of metadata field names that can be indexed alongside embeddings for richer search capabilities, this allows for queries based on both vector similarity and metadata.
Index mapping (indexMapping) This maps the dimensionality of the embedding vectors (e.g., 1536 for OpenAI’s text-embedding-ada-002). This ensures compatibility with the vector model being used.
Create index (createIndex) When set to true, this flag ensures that the necessary index for vector searches is created automatically.

In the main method, we call this method and assign the result to an EmbeddingStore instance:

package com.mongodb;  

public class LangChainRagApp {  

    public static void main(String[] args) {  
        try {  
            // MongoDB setup  
            MongoClient mongoClient = MongoClients.create("CONNECTION_URI");  

            // Embedding Store  
EmbeddingStore<TextSegment> embeddingStore = createEmbeddingStore(mongoClient);

        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}

This embeddingStore is now ready to store, retrieve, and manage our embeddings, with all the beauty and benefits of MongoDB behind it.

Creating our embedding model

The embedding model is the engine that converts raw text into numerical representations, also known as embeddings. These embeddings are high-dimensional representations of our data that capture the semantic meaning of text, making them the foundation for similarity searches in a retrieval-augmented generation application.

In this section, we set up an embedding model using OpenAI's text-embedding-ada-002. To configure the embedding model, we use LangChain4J's OpenAiEmbeddingModel builder, which abstracts the complexities of interacting with OpenAI's API. Here’s the implementation:

package com.mongodb;  

public class LangChainRagApp {  

    public static void main(String[] args) {  
        try {  
            // ...

            // Embedding Model setup  
OpenAiEmbeddingModel embeddingModel = OpenAiEmbeddingModel.builder()  
        .apiKey("OPEN_AI_API_KEY")  
        .modelName(OpenAiEmbeddingModelName.TEXT_EMBEDDING_ADA_002)  
        .build();

        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}

// ...

API key (apiKey) The API key provides access to OpenAI’s services. Replace "OPEN_AI_API_KEY" with your actual OpenAI API key.
Model name (modelName) We specify OpenAiEmbeddingModelName.TEXT_EMBEDDING_ADA_002. This model offers:
- High dimensionality (1536 dimensions): Captures detailed semantic information.
- General-purpose embeddings: Suitable for a multitude of embedding tasks like document retrieval, clustering, and classification.

This embedding model is critical for generating vector representations of the text data we work with.

Configuring our chat model

In a retrieval-augmented generation application, the chat model serves as the conversational engine. It generates our context-aware, human-like responses based on the user’s query and retrieved content. For this tutorial, we configure a chat model using OpenAI's GPT-4 (other AI models are available), simplified by LangChain4J’s straightforward API.

package com.mongodb;  

public class LangChainRagApp {  

    public static void main(String[] args) {  
        try {  
            // ...

            // Chat Model setup
            ChatLanguageModel chatModel = OpenAiChatModel.builder()
                    .apiKey("OPEN_AI_API_KEY")
                    .modelName("gpt-4")
                    .build();

        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}

// ...

Replace the API key, just as before. We also specify the model name here.

The chat model becomes the core of answering user queries in the RAG flow:

Retrieve relevant content: The embedding store retrieves relevant documents based on the user’s query.
Generate a response: The chat model uses the retrieved content as context to generate a detailed and accurate answer.

For instance, a query like:

"How does Atlas Vector Search work?"

Would involve retrieving our related embeddings about Atlas Vector Search from the MongoDB vector store, and then GPT-4 would generate a response using that context.

How to load our data

We are going to be loading our data, which we can download from MongoDB's Hugging Face. It is a collection of approximately 600 articles and tutorials from MongoDB's Developer Center. We are going to place this devcenter-content-snapshot.2024-05-20 1.json file into the resources folder.

Now, we need a method loadJsonDocuments to handle our logic. The loadJsonDocuments method handles loading and processing the dataset. It reads the JSON file, extracts relevant content (title, body, metadata), and splits it into smaller segments for embedding.

private static List<TextSegment> loadJsonDocuments(String resourcePath, int maxTokensPerChunk, int overlapTokens) throws IOException {
    List<TextSegment> textSegments = new ArrayList<>();

    // Load file from resources using the ClassLoader
    InputStream inputStream = LangChainRagApp.class.getClassLoader().getResourceAsStream(resourcePath);

    if (inputStream == null) {
        throw new FileNotFoundException("Resource not found: " + resourcePath);
    }

    // Jackson ObjectMapper
    ObjectMapper objectMapper = new ObjectMapper();
    BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));

    // Batch size for processing
    int batchSize = 500;  // Adjust batch size as needed
    List<Document> batch = new ArrayList<>();

    String line;
    while ((line = reader.readLine()) != null) {
        JsonNode jsonNode = objectMapper.readTree(line);

        String title = jsonNode.path("title").asText(null);
        String body = jsonNode.path("body").asText(null);
        JsonNode metadataNode = jsonNode.path("metadata");

        if (body != null) {
            String text = (title != null ? title + "\n\n" + body : body);

            Metadata metadata = new Metadata();
            if (metadataNode != null && metadataNode.isObject()) {
                Iterator<String> fieldNames = metadataNode.fieldNames();
                while (fieldNames.hasNext()) {
                    String fieldName = fieldNames.next();
                    metadata.put(fieldName, metadataNode.path(fieldName).asText());
                }
            }

            Document document = Document.from(text, metadata);
            batch.add(document);

            // If batch size is reached, process the batch
            if (batch.size() >= batchSize) {
                textSegments.addAll(splitIntoChunks(batch, maxTokensPerChunk, overlapTokens));
                batch.clear();
            }
        }
    }

    // Process remaining documents in the last batch
    if (!batch.isEmpty()) {
        textSegments.addAll(splitIntoChunks(batch, maxTokensPerChunk, overlapTokens));
    }

    return textSegments;
}

Documents need to be divided into smaller chunks to fit within the token limits of our embedding model. We achieve this using the splitIntoChunks method. Here, we will use DocumentSplitter, a tool provided to us by LangChain4j to divide our documents into manageable chunks, while maintaining the original context they provide.


private static List<TextSegment> splitIntoChunks(List<Document> documents, int maxTokensPerChunk, int overlapTokens) {  
    // Create a tokenizer for OpenAI  
    OpenAiTokenizer tokenizer = new OpenAiTokenizer(OpenAiEmbeddingModelName.TEXT_EMBEDDING_ADA_002);  

    // Create a recursive document splitter with the specified token size and overlap  
    DocumentSplitter splitter = DocumentSplitters.recursive(  
            maxTokensPerChunk,  
            overlapTokens,  
            tokenizer  
    );  

    List<TextSegment> allSegments = new ArrayList<>();  
    for (Document document : documents) {  
        List<TextSegment> segments = splitter.split(document);  
        allSegments.addAll(segments);  
    }  

    return allSegments;  
}

Parameters

maxTokensPerChunk: Maximum tokens allowed in each segment. This ensures compatibility with the model’s token limit.
overlapTokens: Number of overlapping tokens between consecutive chunks. Overlaps help preserve context across segments.

Now, time to add this to our main method. The main method orchestrates the entire process: loading the data, embedding it, and storing it in the embedding store.

package com.mongodb;  

public class LangChainRagApp {  

    public static void main(String[] args) {  
        try {  
            // ...

            // Load documents
            String resourcePath = "devcenter-content-snapshot.2024-05-20.json";
            List<TextSegment> documents = loadJsonDocuments(resourcePath, 800, 200);

            System.out.println("Loaded " + documents.size() + " documents");

            for (int i = 0; i < documents.size()/10; i++) {
                TextSegment segment = documents.get(i);
                Embedding embedding = embeddingModel.embed(segment.text()).content();
                embeddingStore.add(embedding, segment);
            }

            System.out.println("Stored embeddings");

        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}

// ...

I added a few comments here to help us track our progress as we ingest our data. I also adjusted to only intake the first 10% of the documents. When I did this with the entire dataset, it took 30+ minutes to load in all the data on my slow internet. Feel free to adjust this, as the more data ingested, the more potentially accurate the answers.

Creating our content retriever

In our retrieval-augmented generation application, the Content Retriever fetches the most relevant content from a data source based on our user query. LangChain4J provides an abstraction for this, allowing us to connect our embedding store and embedding model to retrieve content.

We use the EmbeddingStoreContentRetriever to retrieve content from the embedding store by embedding the user query and finding the most relevant matches.

package com.mongodb;  

public class LangChainRagApp {  

    public static void main(String[] args) {  
        try {  
            // ...

            // Content Retriever
            ContentRetriever contentRetriever = EmbeddingStoreContentRetriever.builder()
                    .embeddingStore(embeddingStore)
                    .embeddingModel(embeddingModel)
                    .maxResults(5)
                    .minScore(0.75)
                    .build();

        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}

// ...

Let’s break down what makes this Content Retriever:

embeddingStore: This is our corpus of data we set up earlier. It’s where all the vectorized representations of our documents live.
embeddingModel: This is the brains behind the operation. It’s the same model we used to create the embeddings (e.g., text-embedding-ada-002). By using the same model here, we ensure that the user’s query is embedded in the same "language" as the stored content.
maxResults: Setting maxResults to 5 means the retriever will hand us up to five of the most relevant matches for your query.
minScore: This is your quality filter. By setting a minScore of 0.75, we’re saying, "Don’t bother showing me anything that’s not highly relevant." If none of the results meet this threshold, we’ll get an empty list instead of cluttered, irrelevant data.

By tweaking these parameters, we can fine-tune how your retriever performs, ensuring it delivers exactly what we need!

Asking questions

Time to put the pieces together. We need a way to bring all our components together to query our enhanced LLM. First, we need to create an interface for our assistant. Create an interface, as shown below.

package com.mongodb;  

public interface Assistant {  

    String answer(String question);  
}

Keeping it very simple, we just want to provide a question and get an answer. Next, we need to create and call our assistant in our main class.

package com.mongodb;  

public class LangChainRagApp {  

    public static void main(String[] args) {  
        try {  
            // ...

            // Assistant
            Assistant assistant = AiServices.builder(Assistant.class)
                    .chatLanguageModel(chatModel)
                    .contentRetriever(contentRetriever)
                    .build();

            String output = assistant.answer("How to use Atlas Triggers and AI to summarise AirBnB reviews?");

            System.out.println(output);

        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
}

// ...

Now, in this implementation, I kept it very simple and kept the query in line. There is nothing stopping you from implementing the querying system in an API, or in terminal, or any other way you can imagine. Let's take a look at our reply:

To summarise Airbnb reviews using MongoDB Atlas Triggers and OpenAI, follow these steps:

1. **Prerequisites**: Set up an App Services application to link to the cluster with Airbnb data. Also, create an OpenAI account with API access.

2. **Set up Secrets and Values**: In App Services, create a secret named `openAIKey` with your OpenAI API key. Then, create a linked value named `OpenAIKey` and link it to the secret.

3. **Trigger code**: The trigger listens for changes in the sample_airbnb.listingsAndReviews collection. When a new review is detected, it samples up to 50 reviews, sends them to OpenAI's API for summarisation, and updates the original document with the summarised content and tags. The trigger reacts to updates that are marked with the `"process" : false` flag, which indicates that a summary hasn't been created for the batch of reviews yet.

4. **Sample Reviews Function**: To avoid overloading the API with too many reviews, a function called `sampleReviews` is defined that randomly samples up to 50 reviews.

5. **API Interaction**: Using the `context.http.post` method, the API request is sent to the OpenAI API.

6. **Updating the Original Document**: Once a successful response from the API is received, the trigger updates the original document with the summarised content, negative tags (neg_tags), positive tags (pos_tags), and a process flag set to true.

7. **Displaying the Data**: Once the data is added to the documents, it can be displayed in a VUE application by adding an HTML template.

By combining MongoDB Atlas triggers with OpenAI's powerful models, large volumes of reviews can be processed and analysed in real-time. This not only provides concise summaries of reviews but also categorises them into positive and negative tags, offering valuable insights to property hosts and potential renters.

This is a well informed response, and actually references the information available in the original tutorial, Using MongoDB Atlas Triggers to Summarize Airbnb Reviews With OpenAI. Want the code? Just ask in the query. It will tailor the responses to exactly what you ask!

Conclusion

There we have it—we used MongoDB with LangChain4j to create a simple RAG application. LangChain4j abstracted away a lot of the steps along the way, from segmenting our data, to connecting to our MongoDB database and embedding model.

If you found this tutorial useful, head over to the Developer Center and check out some of our other tutorials, such as Terraforming AI Workflows: RAG With MongoDB Atlas and Spring AI, or head over to LangChain4j to learn more about what you can do with MongoDB and AI in Java.