mongodb: Spring Data MongoDB transactions, aggregation and compound index (post 26)

README.md

@@ -5,7 +5,7 @@ Companion code for a series of posts on [ankurm.com](https://ankurm.com) coverin
 | Module | Post | Stack |
 |---|---|---|
 | [`time-series/`](./time-series) | Time-Series Data in Java: QuestDB vs InfluxDB vs TimescaleDB | QuestDB 10.0.1, InfluxDB 3 Core 3.12.0, TimescaleDB 2.25.1 (PG17) |
-| `mongodb/` | Spring Boot 4 with MongoDB: Spring Data MongoDB, Aggregations and Transactions | MongoDB 8 replica set, Spring Data MongoDB |
+| [`mongodb/`](./mongodb) | Spring Boot 4 with MongoDB: Spring Data MongoDB, Aggregations and Transactions | MongoDB 9.0.2 single-node replica set, Spring Data MongoDB |
 | `elasticsearch/` | Full-Text Search with Spring Data Elasticsearch on Spring Boot 4 | Elasticsearch 9, Spring Data Elasticsearch |
 | `distributed-cache/` | Distributed Caching in Spring Boot: Hazelcast vs Infinispan vs Redis | Hazelcast, Infinispan, Redis |
 

mongodb/docs/output/26-index-explain.txt

@@ -0,0 +1,2 @@
+query on indexed fields (customerId, createdAt): winning plan stage = FETCH IXSCAN
+query on unindexed field (total): winning plan stage = COLLSCAN

mongodb/docs/output/26-order-transaction-commit.txt

@@ -0,0 +1,2 @@
+order_id=6acfd210bf211a8919681ce9 status=PLACED total=199.90
+inventory_before=50 inventory_after=40 (decremented by order quantity)

mongodb/docs/output/26-order-transaction-rollback.txt

@@ -0,0 +1,3 @@
+attempted_quantity=999 available=5 -> Insufficient stock for WIDGET-2: available=5 requested=999
+inventory_before=5 inventory_after=5 (unchanged - transaction rolled back)
+orders_collection_count=0 (no order document was inserted)

mongodb/docs/output/26-revenue-aggregation.txt

@@ -0,0 +1,2 @@
+status=PLACED total_revenue=350.00 order_count=2
+status=REJECTED_INSUFFICIENT_STOCK total_revenue=40.00 order_count=1

mongodb/pom.xml

@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>4.1.1</version>
+  </parent>
+
+  <groupId>com.ankurm</groupId>
+  <artifactId>mongodb-demo</artifactId>
+  <version>1.0.0</version>
+  <packaging>jar</packaging>
+
+  <properties>
+    <java.version>25</java.version>
+    <maven.compiler.release>25</maven.compiler.release>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-data-mongodb</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-testcontainers</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>testcontainers-mongodb</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>testcontainers-junit-jupiter</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>org.testcontainers</groupId>
+        <artifactId>testcontainers-bom</artifactId>
+        <version>2.0.5</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-maven-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>

mongodb/src/main/java/com/ankurm/mongodb/InventoryItem.java

@@ -0,0 +1,14 @@
+package com.ankurm.mongodb;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.annotation.Version;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+/**
+ * Stock for one SKU in one warehouse. @Version gives optimistic locking on
+ * top of the multi-document transaction used when placing an order - see
+ * OrderService.placeOrder().
+ */
+@Document("inventory")
+public record InventoryItem(@Id String id, String sku, String warehouse, int quantityOnHand, @Version Long version) {
+}

mongodb/src/main/java/com/ankurm/mongodb/InventoryRepository.java

@@ -0,0 +1,9 @@
+package com.ankurm.mongodb;
+
+import org.springframework.data.mongodb.repository.MongoRepository;
+
+import java.util.Optional;
+
+public interface InventoryRepository extends MongoRepository<InventoryItem, String> {
+    Optional<InventoryItem> findBySkuAndWarehouse(String sku, String warehouse);
+}

mongodb/src/main/java/com/ankurm/mongodb/MongoConfig.java

@@ -0,0 +1,21 @@
+package com.ankurm.mongodb;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.mongodb.MongoDatabaseFactory;
+import org.springframework.data.mongodb.MongoTransactionManager;
+
+/**
+ * Spring Boot does NOT auto-configure a PlatformTransactionManager for
+ * MongoDB the way it does for JDBC - without this bean, @Transactional on
+ * OrderService is a silent no-op (no error, no rollback, just two
+ * independent writes). See the post's transactions section.
+ */
+@Configuration
+public class MongoConfig {
+
+    @Bean
+    MongoTransactionManager transactionManager(MongoDatabaseFactory dbFactory) {
+        return new MongoTransactionManager(dbFactory);
+    }
+}

mongodb/src/main/java/com/ankurm/mongodb/MongoDemoApplication.java

@@ -0,0 +1,7 @@
+package com.ankurm.mongodb;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+@SpringBootApplication
+public class MongoDemoApplication {
+    public static void main(String[] args) { SpringApplication.run(MongoDemoApplication.class, args); }
+}

mongodb/src/main/java/com/ankurm/mongodb/Order.java

@@ -0,0 +1,30 @@
+package com.ankurm.mongodb;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.index.CompoundIndex;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.util.List;
+
+/**
+ * An order document. The compound index on (customerId, createdAt) is what
+ * turns "this customer's recent orders" from a collection scan into an
+ * index scan - see the post's "indexes" section for the explain() output
+ * that proves it.
+ */
+@Document("orders")
+@CompoundIndex(name = "customer_created_idx", def = "{'customerId': 1, 'createdAt': -1}")
+public record Order(
+        @Id String id,
+        String customerId,
+        List<OrderItem> items,
+        OrderStatus status,
+        BigDecimal total,
+        Instant createdAt
+) {
+    public record OrderItem(String sku, int quantity, BigDecimal unitPrice) {}
+
+    public enum OrderStatus { PLACED, REJECTED_INSUFFICIENT_STOCK }
+}

mongodb/src/main/java/com/ankurm/mongodb/OrderAnalytics.java

@@ -0,0 +1,45 @@
+package com.ankurm.mongodb;
+
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
+import org.springframework.stereotype.Service;
+
+import java.math.BigDecimal;
+import java.util.List;
+
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
+
+/**
+ * Derived queries and @Query cover point lookups; an aggregation pipeline
+ * is for reshaping many documents into a report MongoRepository has no
+ * vocabulary for - here, revenue and order count per status.
+ */
+@Service
+public class OrderAnalytics {
+
+    private final MongoTemplate mongoTemplate;
+
+    public OrderAnalytics(MongoTemplate mongoTemplate) {
+        this.mongoTemplate = mongoTemplate;
+    }
+
+    public record RevenueByStatus(String status, BigDecimal totalRevenue, long orderCount) {}
+
+    public List<RevenueByStatus> revenueByStatus() {
+        Aggregation pipeline = newAggregation(
+                group("status")
+                        .sum("total").as("totalRevenue")
+                        .count().as("orderCount"),
+                project("totalRevenue", "orderCount").and("status").previousOperation(),
+                sort(org.springframework.data.domain.Sort.Direction.DESC, "totalRevenue")
+        );
+        AggregationResults<RevenueByStatusRaw> results =
+                mongoTemplate.aggregate(pipeline, "orders", RevenueByStatusRaw.class);
+        return results.getMappedResults().stream()
+                .map(r -> new RevenueByStatus(r.status(), r.totalRevenue(), r.orderCount()))
+                .toList();
+    }
+
+    record RevenueByStatusRaw(String status, BigDecimal totalRevenue, long orderCount) {}
+}

mongodb/src/main/java/com/ankurm/mongodb/OrderRepository.java

@@ -0,0 +1,13 @@
+package com.ankurm.mongodb;
+
+import org.springframework.data.mongodb.repository.MongoRepository;
+
+import java.util.List;
+
+public interface OrderRepository extends MongoRepository<Order, String> {
+
+    // Derived query - served by the customerId/createdAt compound index.
+    List<Order> findByCustomerIdOrderByCreatedAtDesc(String customerId);
+
+    List<Order> findByStatus(Order.OrderStatus status);
+}

mongodb/src/main/java/com/ankurm/mongodb/OrderService.java

@@ -0,0 +1,53 @@
+package com.ankurm.mongodb;
+
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.util.List;
+
+/**
+ * Placing an order touches two collections - inventory (decrement stock)
+ * and orders (insert the order) - and both must succeed or neither does.
+ * Without the transaction, a crash between the two writes leaves inventory
+ * decremented with no order to show for it.
+ */
+@Service
+public class OrderService {
+
+    private final OrderRepository orders;
+    private final InventoryRepository inventory;
+
+    public OrderService(OrderRepository orders, InventoryRepository inventory) {
+        this.orders = orders;
+        this.inventory = inventory;
+    }
+
+    @Transactional
+    public Order placeOrder(String customerId, String warehouse, List<Order.OrderItem> items) {
+        for (Order.OrderItem item : items) {
+            InventoryItem stock = inventory.findBySkuAndWarehouse(item.sku(), warehouse)
+                    .orElseThrow(() -> new IllegalStateException("No stock record for " + item.sku()));
+            if (stock.quantityOnHand() < item.quantity()) {
+                // Throwing here rolls back any inventory decrements already
+                // applied earlier in this loop, in the same transaction.
+                throw new InsufficientStockException(item.sku(), stock.quantityOnHand(), item.quantity());
+            }
+            inventory.save(new InventoryItem(stock.id(), stock.sku(), stock.warehouse(),
+                    stock.quantityOnHand() - item.quantity(), stock.version()));
+        }
+
+        BigDecimal total = items.stream()
+                .map(i -> i.unitPrice().multiply(BigDecimal.valueOf(i.quantity())))
+                .reduce(BigDecimal.ZERO, BigDecimal::add);
+
+        return orders.save(new Order(null, customerId, items, Order.OrderStatus.PLACED, total, Instant.now()));
+    }
+
+    public static class InsufficientStockException extends RuntimeException {
+        public InsufficientStockException(String sku, int available, int requested) {
+            super("Insufficient stock for %s: available=%d requested=%d".formatted(sku, available, requested));
+        }
+    }
+}

mongodb/src/test/java/com/ankurm/mongodb/MongoDemoTest.java

@@ -0,0 +1,157 @@
+package com.ankurm.mongodb;
+
+import com.mongodb.client.MongoCollection;
+import org.bson.Document;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.testcontainers.containers.MongoDBContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Runs the repository, aggregation, transaction and index behaviour
+ * against a real MongoDB 9.0.2 single-node replica set (Testcontainers'
+ * MongoDBContainer configures replica-set mode automatically - no
+ * .withReplicaSet() call needed on this version, unlike what older
+ * tutorials show). Every number in the post's transcripts comes from
+ * these tests.
+ */
+@Testcontainers
+@SpringBootTest
+class MongoDemoTest {
+
+    @Container
+    @ServiceConnection
+    static MongoDBContainer mongo = new MongoDBContainer("mongo:9.0.2");
+
+    @Autowired OrderRepository orders;
+    @Autowired InventoryRepository inventory;
+    @Autowired OrderService orderService;
+    @Autowired OrderAnalytics analytics;
+    @Autowired MongoTemplate mongoTemplate;
+
+    @BeforeEach
+    void cleanDatabase() {
+        mongoTemplate.getDb().drop();
+    }
+
+    @Test
+    void transactionCommitsBothWritesTogether() {
+        inventory.save(new InventoryItem(null, "WIDGET-1", "blr-wh", 50, null));
+
+        Order placed = orderService.placeOrder("cust-1", "blr-wh",
+                List.of(new Order.OrderItem("WIDGET-1", 10, new BigDecimal("19.99"))));
+
+        InventoryItem afterOrder = inventory.findBySkuAndWarehouse("WIDGET-1", "blr-wh").orElseThrow();
+
+        Transcript t = new Transcript("26-order-transaction-commit.txt");
+        t.line("order_id=%s status=%s total=%s", placed.id(), placed.status(), placed.total());
+        t.line("inventory_before=50 inventory_after=%d (decremented by order quantity)", afterOrder.quantityOnHand());
+        t.write();
+
+        assertEquals(Order.OrderStatus.PLACED, placed.status());
+        assertEquals(40, afterOrder.quantityOnHand());
+        assertEquals(1, orders.count());
+    }
+
+    @Test
+    void transactionRollsBackBothWritesOnInsufficientStock() {
+        inventory.save(new InventoryItem(null, "WIDGET-2", "blr-wh", 5, null));
+
+        OrderService.InsufficientStockException ex = assertThrows(OrderService.InsufficientStockException.class,
+                () -> orderService.placeOrder("cust-2", "blr-wh",
+                        List.of(new Order.OrderItem("WIDGET-2", 999, new BigDecimal("5.00")))));
+
+        InventoryItem afterFailedOrder = inventory.findBySkuAndWarehouse("WIDGET-2", "blr-wh").orElseThrow();
+
+        Transcript t = new Transcript("26-order-transaction-rollback.txt");
+        t.line("attempted_quantity=999 available=5 -> %s", ex.getMessage());
+        t.line("inventory_before=5 inventory_after=%d (unchanged - transaction rolled back)", afterFailedOrder.quantityOnHand());
+        t.line("orders_collection_count=%d (no order document was inserted)", orders.count());
+        t.write();
+
+        assertEquals(5, afterFailedOrder.quantityOnHand());
+        assertEquals(0, orders.count());
+    }
+
+    @Test
+    void aggregationGroupsRevenueByStatus() {
+        Instant now = Instant.now();
+        orders.saveAll(List.of(
+                new Order(null, "cust-3", List.of(), Order.OrderStatus.PLACED, new BigDecimal("100.00"), now),
+                new Order(null, "cust-4", List.of(), Order.OrderStatus.PLACED, new BigDecimal("250.00"), now),
+                new Order(null, "cust-5", List.of(), Order.OrderStatus.REJECTED_INSUFFICIENT_STOCK, new BigDecimal("40.00"), now)
+        ));
+
+        List<OrderAnalytics.RevenueByStatus> result = analytics.revenueByStatus();
+
+        Transcript t = new Transcript("26-revenue-aggregation.txt");
+        for (OrderAnalytics.RevenueByStatus r : result) {
+            t.line("status=%s total_revenue=%s order_count=%d", r.status(), r.totalRevenue(), r.orderCount());
+        }
+        t.write();
+
+        assertEquals(2, result.size());
+        OrderAnalytics.RevenueByStatus placed = result.stream()
+                .filter(r -> r.status().equals("PLACED")).findFirst().orElseThrow();
+        assertEquals(new BigDecimal("350.00"), placed.totalRevenue());
+        assertEquals(2, placed.orderCount());
+    }
+
+    @Test
+    void compoundIndexTurnsTheQueryIntoAnIndexScan() {
+        // Explicit, rather than relying on spring.data.mongodb.auto-index-creation
+        // (false by default) - see the post's indexes section.
+        mongoTemplate.indexOps(Order.class).createIndex(
+                new org.springframework.data.mongodb.core.index.CompoundIndexDefinition(
+                        new Document("customerId", 1).append("createdAt", -1)).named("customer_created_idx"));
+
+        Instant now = Instant.now();
+        for (int i = 0; i < 500; i++) {
+            orders.save(new Order(null, "cust-" + (i % 5), List.of(), Order.OrderStatus.PLACED,
+                    BigDecimal.valueOf(i), now.minusSeconds(i)));
+        }
+
+        MongoCollection<Document> raw = mongoTemplate.getCollection("orders");
+
+        Document indexedExplain = raw.find(new Document("customerId", "cust-0")
+                        .append("createdAt", new Document("$lte", now)))
+                .sort(new Document("createdAt", -1))
+                .explain();
+        String indexedStage = winningPlanStage(indexedExplain);
+
+        Document unindexedExplain = raw.find(new Document("total", new Document("$gt", 490))).explain();
+        String unindexedStage = winningPlanStage(unindexedExplain);
+
+        Transcript t = new Transcript("26-index-explain.txt");
+        t.line("query on indexed fields (customerId, createdAt): winning plan stage = %s", indexedStage);
+        t.line("query on unindexed field (total): winning plan stage = %s", unindexedStage);
+        t.write();
+
+        assertTrue(indexedStage.contains("IXSCAN"), "expected an index scan, got: " + indexedStage);
+        assertTrue(unindexedStage.contains("COLLSCAN"), "expected a collection scan, got: " + unindexedStage);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static String winningPlanStage(Document explain) {
+        Document queryPlanner = (Document) explain.get("queryPlanner");
+        Document winningPlan = (Document) queryPlanner.get("winningPlan");
+        StringBuilder stages = new StringBuilder();
+        Document stage = winningPlan;
+        while (stage != null) {
+            stages.append(stage.getString("stage")).append(" ");
+            stage = (Document) stage.get("inputStage");
+        }
+        return stages.toString().trim();
+    }
+}

mongodb/src/test/java/com/ankurm/mongodb/Transcript.java

@@ -0,0 +1,40 @@
+package com.ankurm.mongodb;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Writes docs/output/NN-*.txt while a test asserts the same numbers, so
+ * every figure quoted in the blog post is backed by an assertion that
+ * fails the build when it stops being true.
+ */
+final class Transcript {
+    private final List<String> lines = new ArrayList<>();
+    private final Path file;
+
+    Transcript(String name) {
+        this.file = Path.of("docs/output/" + name);
+    }
+
+    Transcript line(String s) {
+        lines.add(s);
+        System.out.println(s);
+        return this;
+    }
+
+    Transcript line(String fmt, Object... args) {
+        return line(fmt.formatted(args));
+    }
+
+    void write() {
+        try {
+            Files.createDirectories(file.getParent());
+            Files.writeString(file, String.join("\n", lines) + "\n");
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}