CD Rental System, and Making It Thread-Safe

Published 3 October 2026

This was round 2 of an Okta loop (Java, DSA / low-level design), and it went badly. The problem was tracking CD rentals across several shops. I struggled with the design, couldn't agree with the interviewer on an approach, and ran out of time before a clean implementation. I was still scheduled for the next round, which suggests Okta weighs the whole loop rather than cutting candidates after one weak interview.

The single-threaded data structure turns out to be simple. The part worth studying is the follow-up, "make it thread-safe", because there's more to plan for there than just "add a lock".

Problem statement

There are n shops, identified by an integer shopId. Each shop may stock copies of various CDs, identified by an integer cdId, each at a shop-specific rental price. The initial inventory is a list of entries [shopId, cdId, price], and each (shopId, cdId) pair appears at most once.

Implement CdRentalSystem:

• CdRentalSystem(int n, int[][] entries) sets up the inventory. All copies start available.
• List<Integer> search(int cdId) returns the IDs of shops with an unrented copy of cdId, sorted by price ascending, ties broken by shopId ascending.
• void rent(int shopId, int cdId) rents that CD from that shop.
• void unrent(int shopId, int cdId) returns a rented CD to its shop.

Requirements: make all three operations efficient and state their complexity; write clean, object-oriented Java; handle a CD stocked nowhere, renting something already rented, and an empty search.

Follow-ups: limit search to the top k; add report(), returning the cheapest rented CDs across all shops; make it thread-safe.

It's very close to LeetCode 1912, Design Movie Rental System.

The single-threaded solution

Keep a sorted set of available copies per CD, ordered by (price, shopId). search reads the front of that set, rent removes a copy, unrent puts it back. To find a copy's price from (shopId, cdId), keep a map from that pair to the listing.

For report(), keep one more sorted set of rented listings. rent moves a listing from available[cdId] to rented, unrent moves it back, and report(k) reads the first k.

A listing is a record, and its ordering lives with it:

import java.util.Comparator;

record Listing(int price, int shopId, int cdId) {
    // TreeSet and ConcurrentSkipListSet decide "same element" with this, not equals(),
    // so it has to compare every field that tells two listings apart.
    static final Comparator<Listing> ORDER = Comparator
            .comparingInt(Listing::price)
            .thenComparingInt(Listing::shopId)
            .thenComparingInt(Listing::cdId);
}

The comparator has to include every field. A TreeSet decides that two elements are "the same" by comparing them, not by calling equals(). Leave a field out and distinct listings silently collapse into one:

• With (price, shopId), one shop renting two different CDs at the same price: the second add returns false and that rental disappears from report().
• With (price, cdId), my first idea for the rented set: two shops renting the same CD at the same price collide the same way.

I checked both: each second add returns false.

import java.util.*;

class CdRentalSystem {
    private final Map<Long, Listing> listings = new HashMap<>();            // (shop, cd) -> listing
    private final Map<Integer, TreeSet<Listing>> available = new HashMap<>(); // cd -> unrented copies
    private final TreeSet<Listing> rented = new TreeSet<>(Listing.ORDER);

    CdRentalSystem(int n, int[][] entries) {
        for (int[] e : entries) {
            Listing l = new Listing(e[2], e[0], e[1]);
            listings.put(key(e[0], e[1]), l);
            available.computeIfAbsent(l.cdId(), cd -> new TreeSet<>(Listing.ORDER)).add(l);
        }
    }

    List<Integer> search(int cdId, int k) {
        List<Integer> shops = new ArrayList<>();
        for (Listing l : available.getOrDefault(cdId, new TreeSet<>(Listing.ORDER))) {
            if (shops.size() == k) break;
            shops.add(l.shopId());
        }
        return shops;
    }

    boolean rent(int shopId, int cdId) {
        Listing l = listings.get(key(shopId, cdId));
        if (l == null || !available.get(cdId).remove(l)) return false; // not stocked, or already out
        rented.add(l);
        return true;
    }

    boolean unrent(int shopId, int cdId) {
        Listing l = listings.get(key(shopId, cdId));
        if (l == null || !rented.remove(l)) return false;               // not stocked, or not out
        available.get(cdId).add(l);
        return true;
    }

    List<Listing> report(int k) {
        return rented.stream().limit(k).toList();
    }

    private static long key(int shopId, int cdId) {
        return ((long) shopId << 32) | (cdId & 0xffffffffL);
    }
}

Operation	Cost
search(cdId, k)	O(log n + k)
rent / unrent	O(log n): one removal and one insertion
report(k)	O(log n + k)

rent and unrent return false instead of assuming valid input. That covers a CD stocked nowhere and renting something that's already out. There's no deep copy in search: shop IDs are immutable, so a new list is enough.

What can go wrong with threads

"Strap a lock on every method" is a correct first answer. Before choosing anything cleverer, here's what you're actually protecting against. Deadlock is on the list, but it's rarely where the real bugs are.

1. Check-then-act. if (available.contains(x)) { available.remove(x); ... } lets two threads both see the copy and both rent it. Checking availability and claiming the copy have to be one atomic step.
2. An invariant spanning two structures. Every listing should be in exactly one of available and rented. rent removes from one and then adds to the other, so in between, the listing is in neither.
3. Visibility and safe publication. A plain HashMap filled lazily from several threads is a data race, and the Java memory model doesn't promise other threads see a fully built object. Either build every map in the constructor and never change its shape, or use ConcurrentHashMap.
4. Iterating while others write. A plain TreeSet iterated during a write can throw ConcurrentModificationException or return garbage. ConcurrentSkipListSet iterators never throw, but they're weakly consistent: not a snapshot.
5. Liveness other than deadlock. A non-fair ReentrantReadWriteLock under heavy reads can starve writers. Striping doesn't help when everyone rents the same hit CD. CAS retry loops and tryLock-and-back-off schemes can livelock.
6. Lock scope. No I/O, logging or callbacks while holding a lock.

Step 1: one read-write lock

Wrap the single-threaded class. search and report take the read lock and run in parallel. rent and unrent take the write lock.

import java.util.List;
import java.util.concurrent.locks.ReentrantReadWriteLock;

class LockedCdRentalSystem {
    private final CdRentalSystem inner;
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    LockedCdRentalSystem(int n, int[][] entries) {
        inner = new CdRentalSystem(n, entries);
    }

    List<Integer> search(int cdId, int k) {
        lock.readLock().lock();
        try { return inner.search(cdId, k); } finally { lock.readLock().unlock(); }
    }

    List<Listing> report(int k) {
        lock.readLock().lock();
        try { return inner.report(k); } finally { lock.readLock().unlock(); }
    }

    boolean rent(int shopId, int cdId) {
        lock.writeLock().lock();
        try { return inner.rent(shopId, cdId); } finally { lock.writeLock().unlock(); }
    }

    boolean unrent(int shopId, int cdId) {
        lock.writeLock().lock();
        try { return inner.unrent(shopId, cdId); } finally { lock.writeLock().unlock(); }
    }
}

It's obviously correct: every operation sees a consistent state, and with one lock there's nothing to deadlock on. The cost is that every rental in the system queues behind every other, even for unrelated CDs. If writes are frequent, a non-fair lock can starve them; new ReentrantReadWriteLock(true) is fair at the cost of throughput.

Step 2: a lock per CD

rent(shop, cd) only touches available[cd] and the global rented set. So give each CD its own lock, and make rented a ConcurrentSkipListSet, which is safe without one. A fixed array of 64 lock stripes avoids creating a lock per CD; unrelated CDs that hash to the same stripe just share it.

import java.util.*;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.locks.ReentrantLock;

class StripedCdRentalSystem {
    private final Map<Long, Listing> listings = new HashMap<>();
    private final Map<Integer, TreeSet<Listing>> available = new HashMap<>();
    private final ConcurrentSkipListSet<Listing> rented = new ConcurrentSkipListSet<>(Listing.ORDER);
    private final ReentrantLock[] stripes = new ReentrantLock[64];

    StripedCdRentalSystem(int n, int[][] entries) {
        // The maps are final fields, filled here and never reshaped afterwards. Final-field
        // semantics guarantee every thread sees them fully built, so plain HashMaps are safe.
        for (int i = 0; i < stripes.length; i++) stripes[i] = new ReentrantLock();
        for (int[] e : entries) {
            Listing l = new Listing(e[2], e[0], e[1]);
            listings.put(key(e[0], e[1]), l);
            available.computeIfAbsent(l.cdId(), cd -> new TreeSet<>(Listing.ORDER)).add(l);
        }
    }

    private ReentrantLock lockFor(int cdId) {
        return stripes[Math.floorMod(cdId, stripes.length)];
    }

    List<Integer> search(int cdId, int k) {
        TreeSet<Listing> copies = available.get(cdId);
        if (copies == null) return List.of();
        ReentrantLock lock = lockFor(cdId);
        lock.lock();
        try {
            List<Integer> shops = new ArrayList<>();
            for (Listing l : copies) {
                if (shops.size() == k) break;
                shops.add(l.shopId());
            }
            return shops;
        } finally {
            lock.unlock();
        }
    }

    boolean rent(int shopId, int cdId) {
        Listing l = listings.get(key(shopId, cdId));
        if (l == null) return false;
        ReentrantLock lock = lockFor(cdId);
        lock.lock();
        try {
            if (!available.get(cdId).remove(l)) return false; // check and claim in one step
            rented.add(l);
            return true;
        } finally {
            lock.unlock();
        }
    }

    boolean unrent(int shopId, int cdId) {
        Listing l = listings.get(key(shopId, cdId));
        if (l == null) return false;
        ReentrantLock lock = lockFor(cdId);
        lock.lock();
        try {
            if (!rented.remove(l)) return false;
            available.get(cdId).add(l);
            return true;
        } finally {
            lock.unlock();
        }
    }

    List<Listing> report(int k) {
        return rented.stream().limit(k).toList(); // no lock: weakly consistent
    }

    private static long key(int shopId, int cdId) {
        return ((long) shopId << 32) | (cdId & 0xffffffffL);
    }
}

Why this can't deadlock. Deadlock needs a thread to wait for one lock while holding another, in a cycle. Here, rent, unrent and search each hold exactly one lock, and report holds none; ConcurrentSkipListSet never blocks. So no thread ever waits while holding a lock.

That stops being true the moment you protect rented with a lock instead. Then rent holds the CD lock and the rented lock, and any operation that takes them in the opposite order can deadlock against it. The fix is a fixed lock order everyone follows, such as "a CD's lock before the rented lock", or "lower cdId first" for an operation that needs two CDs. Interviewers want to hear that rule said out loud.

Step 3: no locks at all

Make the per-CD sets ConcurrentSkipListSet too, and let remove() be the claim. It returns true for exactly one thread, so whoever wins the removal owns the copy.

import java.util.*;
import java.util.concurrent.ConcurrentSkipListSet;

class LockFreeCdRentalSystem {
    private final Map<Long, Listing> listings = new HashMap<>();
    private final Map<Integer, ConcurrentSkipListSet<Listing>> available = new HashMap<>();
    private final ConcurrentSkipListSet<Listing> rented = new ConcurrentSkipListSet<>(Listing.ORDER);

    LockFreeCdRentalSystem(int n, int[][] entries) {
        for (int[] e : entries) {
            Listing l = new Listing(e[2], e[0], e[1]);
            listings.put(key(e[0], e[1]), l);
            available.computeIfAbsent(l.cdId(), cd -> new ConcurrentSkipListSet<>(Listing.ORDER)).add(l);
        }
    }

    List<Integer> search(int cdId, int k) {
        ConcurrentSkipListSet<Listing> copies = available.get(cdId);
        if (copies == null) return List.of();
        return copies.stream().limit(k).map(Listing::shopId).toList();
    }

    boolean rent(int shopId, int cdId) {
        Listing l = listings.get(key(shopId, cdId));
        // remove() succeeds for exactly one thread: winning it is the claim.
        if (l == null || !available.get(cdId).remove(l)) return false;
        rented.add(l);
        return true;
    }

    boolean unrent(int shopId, int cdId) {
        Listing l = listings.get(key(shopId, cdId));
        if (l == null || !rented.remove(l)) return false;
        available.get(cdId).add(l);
        return true;
    }

    List<Listing> report(int k) {
        return rented.stream().limit(k).toList();
    }

    private static long key(int shopId, int cdId) {
        return ((long) shopId << 32) | (cdId & 0xffffffffL);
    }
}

The listing in neither set

In steps 2 and 3, rent removes the listing from available[cd] and only then adds it to rented. Between the two, the listing is in neither set. unrent has the same gap in reverse.

When does that matter? Not for search or report on their own, since each reads only one set. It matters for an operation that reads both, such as "how many copies of this CD exist" or a consistency audit, which can come up one short.

It's tempting to test this by calling search and then report and checking that the listing is in one of them. That test is wrong. I tried it: the "missing" count was non-zero for the step-1 version too, because two separate calls are never atomic together, whatever locks each takes. A caller who needs one consistent view needs a single method that gives it.

There are three ways to handle it:

• Accept it and say so. report() is eventually consistent: it can briefly miss a rental that's in progress.
• Make readers take the same locks in the same order. A reader that takes the CD's lock and then the rented lock can never see the gap, because rent holds the CD's lock for the whole move. The Rust version below does this and tests it.
• Keep one source of truth. A ConcurrentHashMap<(shop, cd), State> where one CAS turns AVAILABLE into RENTED. The two sorted sets become indexes derived from it, and readers double-check what they find against the state map.

Does it actually work?

I ran each version with 8 threads fighting over two copies of the same CD for one second. Each thread rents a copy, checks that nobody else holds it, and returns it. I added two broken variants for comparison: check-then-act on concurrent sets (every single call is thread-safe, the sequence isn't), and the single-threaded class with no lock at all.

Version	Successful rents	Double rents
Check-then-act on ConcurrentSkipListSet	4,560,578	286,715
No lock at all	2,870	489
Step 1: read-write lock	1,963,367	0
Step 2: lock per CD	2,377,830	0
Step 3: lock-free	2,273,499	0

The unlocked version didn't just double-rent: it mostly stopped working, managing 2,870 rents in a second against around two million for the safe versions. All the versions also agree with a brute-force model on 2,000 random single-threaded operation sequences.

The first row is the one to remember. Thread-safe building blocks don't make a thread-safe operation.

A Rust version

Rust changes which of these mistakes you can make. Here's the step-2 design using only the standard library: a Mutex per CD, and a Mutex around the rented set, since std has no concurrent skip list. That brings back the two-lock case, so the lock order is written down and followed.

use std::collections::{BTreeSet, HashMap};
use std::sync::Mutex;

// derive(Ord) compares fields in declaration order: price, then shop_id, then cd_id.
// Every field takes part, so two different listings can never compare equal.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct Listing {
    pub price: u32,
    pub shop_id: u32,
    pub cd_id: u32,
}

pub struct CdRentalSystem {
    listings: HashMap<(u32, u32), Listing>,            // (shop, cd) -> listing
    available: HashMap<u32, Mutex<BTreeSet<Listing>>>, // one lock per CD
    rented: Mutex<BTreeSet<Listing>>,
}

// Lock order: a CD's lock, then `rented`. Nothing takes them the other way round.
impl CdRentalSystem {
    pub fn new(entries: &[[u32; 3]]) -> Self {
        let mut listings = HashMap::new();
        let mut available: HashMap<u32, Mutex<BTreeSet<Listing>>> = HashMap::new();
        for &[shop_id, cd_id, price] in entries {
            let l = Listing { price, shop_id, cd_id };
            listings.insert((shop_id, cd_id), l);
            available.entry(cd_id).or_default().get_mut().unwrap().insert(l);
        }
        Self { listings, available, rented: Mutex::new(BTreeSet::new()) }
    }

    pub fn search(&self, cd_id: u32, k: usize) -> Vec<u32> {
        match self.available.get(&cd_id) {
            Some(copies) => copies.lock().unwrap().iter().take(k).map(|l| l.shop_id).collect(),
            None => Vec::new(),
        }
    }

    pub fn rent(&self, shop_id: u32, cd_id: u32) -> bool {
        let Some(&l) = self.listings.get(&(shop_id, cd_id)) else { return false };
        let mut copies = self.available[&cd_id].lock().unwrap();
        if !copies.remove(&l) {
            return false; // check and claim in one step, under the CD's lock
        }
        self.rented.lock().unwrap().insert(l);
        true
    }

    pub fn unrent(&self, shop_id: u32, cd_id: u32) -> bool {
        let Some(&l) = self.listings.get(&(shop_id, cd_id)) else { return false };
        let mut copies = self.available[&cd_id].lock().unwrap();
        if !self.rented.lock().unwrap().remove(&l) {
            return false;
        }
        copies.insert(l);
        true
    }

    pub fn report(&self, k: usize) -> Vec<Listing> {
        self.rented.lock().unwrap().iter().take(k).copied().collect()
    }
}

Share it as Arc<CdRentalSystem>; every method takes &self.

What the compiler catches. The visibility bug, mistake 3, doesn't compile. Mutating shared state through &self needs interior mutability, and the single-threaded kind (RefCell) isn't Sync. Share one across threads and you get:

error[E0277]: `RefCell<BTreeSet<u32>>` cannot be shared between threads safely

The HashMaps of listings and per-CD locks need no lock of their own: they're built in new() and only read afterwards, and the compiler can prove it. derive(Ord) compares fields in declaration order, so putting every field in the struct gives you the all-fields comparator for free.

What it doesn't catch. Logic races. This compiles cleanly and double-rented 1,788 times in the same one-second test, against 0 for the version above:

fn rent(&self, l: Listing) -> bool {
    if !self.available.lock().unwrap().contains(&l) { return false; }
    self.available.lock().unwrap().remove(&l); // a second lock: another thread got in between
    self.rented.lock().unwrap().insert(l);
    true
}

Rust also doesn't stop deadlocks. Lock order is still on you. And a lock guard lives longer than it looks. In if let Some(x) = m.lock().unwrap().first().copied() { ... } the guard stays held for the whole body, in both edition 2021 and 2024 (2024 only drops it before an else). Locking m again inside deadlocks. Bind the value with let first, so the guard is dropped at the end of that statement.

Holding the CD lock across the move. rent holds the CD's lock while it touches rented, so a reader that takes the CD lock and then the rented lock never sees the gap. I checked that with an audit method counting a CD's copies across both sets: it read exactly 1 on about 5 million audits while another thread rented and returned the copy in a loop.

Critique

• Ask about copies first. If a shop can hold several copies of a CD, the model changes: you need a count per (shop, cd), and a listing leaves available only when the count reaches 0. That's the kind of requirement where interviewer and candidate stop agreeing. In the first two minutes, ask about copy counts, whether rent can fail, and how big k is.
• void rent with "assume it's available" is a trap under concurrency. With two threads, the precondition can't be checked before the call and still hold during it. Returning boolean lets the claim and the check be the same step.
• Striping trades precision for memory. With 64 stripes, unrelated CDs share locks, and a hit CD is still one lock that every renter queues on. Neither more stripes nor a lock per CD fixes a hot key. Step 3 stops renters blocking each other, but they still contend on the same set; sharding that CD's copies is the real fix.
• report() is a design choice, not a detail. Exact and point-in-time means a global lock, versioning or snapshots. Cheap and lock-free means weakly consistent. Pick one and say it.
• In the interview, give deadlock 30 seconds. State the lock-order rule and move on. Spend the time on check-then-act and the cross-structure invariant, because that's where the bugs in the table above came from.