Custom Blocking Queue — From Problem to Production-Grade Solution

Java Interview Deep Dive

Implement a bounded blocking queue (like ArrayBlockingQueue). Twist: Writers must be able to cancel their attempt if waiting too long.

Real-world relevance: Producer-consumer patterns, bounded buffer implementations, timeout handling, cancellation support, concurrent data structures. Core skills: Blocking queue implementation, timeout mechanisms, cancellation patterns, concurrent programming, bounded buffer design.

Full story for non-members | Grab My Microservices E-Book | Youtube | LinkedIn | Book a 1:1 Meeting

The Interview Setup (Scenario)

Interviewer: “We need a custom blocking queue implementation. It should be bounded like ArrayBlockingQueue, but with a twist — writers should be able to cancel their attempt if they’re waiting too long. How would you approach this?”

Candidate: “Interesting! This is essentially a producer-consumer pattern with timeout and cancellation support. I’d need to implement a bounded circular buffer with blocking operations, but add timeout mechanisms for writers. The key challenge is handling the cancellation gracefully without leaving the queue in an inconsistent state.”

Interviewer: “Good thinking! What data structure would you use for the underlying storage?”

Candidate: “I’d use a circular array with two pointers — head and tail — to track where to read and write. This gives us O-1 enqueue and dequeue operations. I’d also need to track the current size to know when the queue is full or empty.”

Interviewer: “Excellent! Now, how would you handle the blocking when the queue is full?”

Candidate: “I’d use a ReentrantLock with two Condition objects — one for when the queue is full (notFull) and one for when it’s empty (notEmpty). Writers would wait on notFull.await() when the queue is full, and readers would signal notFull.signal() when they remove an element.”

Interviewer: “Good! Now for the twist — how would you implement the cancellation mechanism?”

Candidate: “I’d use a timeout-based approach with tryLock() or await(timeout). The writer would wait on the condition with a timeout, and if it expires, they can cancel. I’d also need to handle the case where a writer is interrupted while waiting — they should be able to cancel cleanly.”

Interviewer: “What about thread safety? How would you ensure the queue operations are atomic?”

Candidate: “I’d use the ReentrantLock to protect all queue operations. The lock would be acquired before any modification and released in a finally block. The Condition objects would be created from this same lock, ensuring proper synchronization between waiting and signaling.”

Interviewer: “Perfect! Now show me the implementation with proper timeout and cancellation support.”

Solution Design

The key insight is implementing a bounded circular buffer with proper synchronization using ReentrantLock and Condition objects, while adding timeout mechanisms that allow writers to cancel their attempts gracefully.

Visual Representation

Queue State Management

Final Correct Code Implementation

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
public class CustomBlockingQueue<T> {

    private final T[] elements;
    private final int capacity;
    private final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;

    private int head;        // Read position
    private int tail;        // Write position
    private int size;        // Current number of elements
    private final AtomicLong totalOffers = new AtomicLong(0);
    private final AtomicLong totalPolls = new AtomicLong(0);
    private final AtomicLong cancelledOffers = new AtomicLong(0);

    @SuppressWarnings("unchecked")
    public CustomBlockingQueue(int capacity) {
        if (capacity <= 0) {
            throw new IllegalArgumentException("Capacity must be positive");
        }
        this.capacity = capacity;
        this.elements = (T[]) new Object[capacity];
        this.lock = new ReentrantLock();
        this.notEmpty = lock.newCondition();
        this.notFull = lock.newCondition();
        this.head = 0;
        this.tail = 0;
        this.size = 0;
    }

    // Basic blocking operations
    public void put(T element) throws InterruptedException {
        if (element == null) {
            throw new NullPointerException("Element cannot be null");
        }

        lock.lock();
        try {
            while (size == capacity) {
                notFull.await(); // Wait until there's space
            }
            enqueue(element);
        } finally {
            lock.unlock();
        }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (size == 0) {
                notEmpty.await(); // Wait until there's an element
            }
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    // Non-blocking operations
    public boolean offer(T element) {
        if (element == null) {
            return false;
        }

        lock.lock();
        try {
            if (size == capacity) {
                return false; // Queue is full
            }
            enqueue(element);
            return true;
        } finally {
            lock.unlock();
        }
    }

    public T poll() {
        lock.lock();
        try {
            if (size == 0) {
                return null; // Queue is empty
            }
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    // TIMEOUT OPERATIONS WITH CANCELLATION SUPPORT
    public boolean offer(T element, long timeout, TimeUnit unit) throws InterruptedException {
        if (element == null) {
            return false;
        }

        long nanos = unit.toNanos(timeout);
        lock.lock();
        try {
            while (size == capacity) {
                if (nanos <= 0) {
                    // Timeout expired - cancellation successful
                    cancelledOffers.incrementAndGet();
                    return false;
                }
                nanos = notFull.awaitNanos(nanos);
            }
            enqueue(element);
            return true;
        } finally {
            lock.unlock();
        }
    }

    public T poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        lock.lock();
        try {
            while (size == 0) {
                if (nanos <= 0) {
                    return null; // Timeout expired
                }
                nanos = notEmpty.awaitNanos(nanos);
            }
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    // CANCELLATION SUPPORT - Try to offer with immediate cancellation check
    public boolean tryOffer(T element, long timeout, TimeUnit unit) throws InterruptedException {
        if (element == null) {
            return false;
        }

        long nanos = unit.toNanos(timeout);
        long startTime = System.nanoTime();

        lock.lock();
        try {
            while (size == capacity) {
                if (nanos <= 0) {
                    cancelledOffers.incrementAndGet();
                    return false;
                }

                // Check if we should cancel based on elapsed time
                long elapsed = System.nanoTime() - startTime;
                if (elapsed >= unit.toNanos(timeout)) {
                    cancelledOffers.incrementAndGet();
                    return false;
                }

                nanos = notFull.awaitNanos(nanos);
            }
            enqueue(element);
            return true;
        } finally {
            lock.unlock();
        }
    }

    // ADVANCED CANCELLATION - Cancel based on external condition
    public static class CancellableOffer<T> {
        private final T element;
        private volatile boolean cancelled = false;
        private final long startTime;
        private final long timeoutNanos;

        public CancellableOffer(T element, long timeout, TimeUnit unit) {
            this.element = element;
            this.timeoutNanos = unit.toNanos(timeout);
            this.startTime = System.nanoTime();
        }

        public void cancel() {
            this.cancelled = true;
        }

        public boolean isCancelled() {
            return cancelled || (System.nanoTime() - startTime) >= timeoutNanos;
        }

        public T getElement() {
            return element;
        }
    }

    public boolean offerWithCancellation(CancellableOffer<T> cancellableOffer) throws InterruptedException {
        if (cancellableOffer == null || cancellableOffer.getElement() == null) {
            return false;
        }

        lock.lock();
        try {
            while (size == capacity) {
                if (cancellableOffer.isCancelled()) {
                    cancelledOffers.incrementAndGet();
                    return false;
                }
                notFull.await(100, TimeUnit.MILLISECONDS); // Small wait to check cancellation
            }
            enqueue(cancellableOffer.getElement());
            return true;
        } finally {
            lock.unlock();
        }
    }

    // Queue state operations
    public int size() {
        lock.lock();
        try {
            return size;
        } finally {
            lock.unlock();
        }
    }

    public boolean isEmpty() {
        lock.lock();
        try {
            return size == 0;
        } finally {
            lock.unlock();
        }
    }

    public boolean isFull() {
        lock.lock();
        try {
            return size == capacity;
        } finally {
            lock.unlock();
        }
    }

    public int remainingCapacity() {
        lock.lock();
        try {
            return capacity - size;
        } finally {
            lock.unlock();
        }
    }

    // Statistics
    public long getTotalOffers() {
        return totalOffers.get();
    }

    public long getTotalPolls() {
        return totalPolls.get();
    }

    public long getCancelledOffers() {
        return cancelledOffers.get();
    }

    public double getCancellationRate() {
        long total = totalOffers.get();
        return total == 0 ? 0.0 : (double) cancelledOffers.get() / total;
    }

    // Private helper methods
    private void enqueue(T element) {
        elements[tail] = element;
        tail = (tail + 1) % capacity;
        size++;
        totalOffers.incrementAndGet();
        notEmpty.signal(); // Signal that an element is available
    }

    private T dequeue() {
        T element = elements[head];
        elements[head] = null; // Help GC
        head = (head + 1) % capacity;
        size--;
        totalPolls.incrementAndGet();
        notFull.signal(); // Signal that space is available
        return element;
    }

    // Clear the queue
    public void clear() {
        lock.lock();
        try {
            for (int i = 0; i < size; i++) {
                elements[(head + i) % capacity] = null;
            }
            head = 0;
            tail = 0;
            size = 0;
            notFull.signalAll(); // Signal all waiting writers
        } finally {
            lock.unlock();
        }
    }

    // Drain operations
    public int drainTo(java.util.Collection<? super T> collection) {
        return drainTo(collection, Integer.MAX_VALUE);
    }

    public int drainTo(java.util.Collection<? super T> collection, int maxElements) {
        if (collection == null) {
            throw new NullPointerException("Collection cannot be null");
        }
        if (collection == this) {
            throw new IllegalArgumentException("Cannot drain to self");
        }

        lock.lock();
        try {
            int n = Math.min(maxElements, size);
            for (int i = 0; i < n; i++) {
                T element = elements[head];
                elements[head] = null;
                head = (head + 1) % capacity;
                collection.add(element);
            }
            size -= n;
            notFull.signalAll(); // Signal all waiting writers
            return n;
        } finally {
            lock.unlock();
        }
    }
}
// DEMONSTRATION: Show the queue in action with cancellation
public class CustomBlockingQueueDemo {

    public static void demonstrateBasicOperations() throws InterruptedException {
        System.out.println("=== BASIC OPERATIONS DEMO ===");

        CustomBlockingQueue<String> queue = new CustomBlockingQueue<>(3);

        // Basic offer/poll
        System.out.println("Queue size: " + queue.size());
        System.out.println("Is empty: " + queue.isEmpty());

        queue.offer("First");
        queue.offer("Second");
        queue.offer("Third");

        System.out.println("After adding 3 elements:");
        System.out.println("Queue size: " + queue.size());
        System.out.println("Is full: " + queue.isFull());
        System.out.println("Remaining capacity: " + queue.remainingCapacity());

        // Poll elements
        System.out.println("Polled: " + queue.poll());
        System.out.println("Polled: " + queue.poll());
        System.out.println("Polled: " + queue.poll());

        System.out.println("After polling all elements:");
        System.out.println("Queue size: " + queue.size());
        System.out.println("Is empty: " + queue.isEmpty());
    }

    public static void demonstrateTimeoutAndCancellation() throws InterruptedException {
        System.out.println("\n=== TIMEOUT AND CANCELLATION DEMO ===");

        CustomBlockingQueue<Integer> queue = new CustomBlockingQueue<>(2);

        // Fill the queue
        queue.offer(1);
        queue.offer(2);

        // Try to offer with timeout - should fail after 1 second
        System.out.println("Attempting to offer with 1 second timeout...");
        long startTime = System.currentTimeMillis();
        boolean success = queue.offer(3, 1, TimeUnit.SECONDS);
        long endTime = System.currentTimeMillis();

        System.out.println("Offer result: " + success);
        System.out.println("Time taken: " + (endTime - startTime) + " ms");
        System.out.println("Cancellation rate: " + String.format("%.2f%%", queue.getCancellationRate() * 100));

        // Remove an element and try again
        queue.poll();
        System.out.println("After removing one element, attempting to offer...");
        success = queue.offer(3, 1, TimeUnit.SECONDS);
        System.out.println("Offer result: " + success);
    }

    public static void demonstrateConcurrentUsage() throws InterruptedException {
        System.out.println("\n=== CONCURRENT USAGE DEMO ===");

        CustomBlockingQueue<String> queue = new CustomBlockingQueue<>(5);

        // Producer thread
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    boolean success = queue.offer("Item-" + i, 2, TimeUnit.SECONDS);
                    if (success) {
                        System.out.println("Produced: Item-" + i);
                    } else {
                        System.out.println("Failed to produce: Item-" + i + " (timeout/cancelled)");
                    }
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // Consumer thread
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    String item = queue.poll(2, TimeUnit.SECONDS);
                    if (item != null) {
                        System.out.println("Consumed: " + item);
                    } else {
                        System.out.println("Failed to consume (timeout)");
                    }
                    Thread.sleep(150);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();

        System.out.println("Final queue size: " + queue.size());
        System.out.println("Total offers: " + queue.getTotalOffers());
        System.out.println("Total polls: " + queue.getTotalPolls());
        System.out.println("Cancelled offers: " + queue.getCancelledOffers());
        System.out.println("Cancellation rate: " + String.format("%.2f%%", queue.getCancellationRate() * 100));
    }

    public static void demonstrateAdvancedCancellation() throws InterruptedException {
        System.out.println("\n=== ADVANCED CANCELLATION DEMO ===");

        CustomBlockingQueue<Integer> queue = new CustomBlockingQueue<>(1);

        // Fill the queue
        queue.offer(1);

        // Create a cancellable offer
        CustomBlockingQueue.CancellableOffer<Integer> cancellableOffer = 
            new CustomBlockingQueue.CancellableOffer<>(2, 5, TimeUnit.SECONDS);

        // Start a thread that will cancel the offer after 1 second
        Thread canceller = new Thread(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("Cancelling the offer...");
                cancellableOffer.cancel();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        canceller.start();

        // Try to offer with cancellation support
        System.out.println("Attempting to offer with cancellation support...");
        boolean success = queue.offerWithCancellation(cancellableOffer);
        System.out.println("Offer result: " + success);

        canceller.join();
    }

    public static void main(String[] args) throws InterruptedException {
        demonstrateBasicOperations();
        demonstrateTimeoutAndCancellation();
        demonstrateConcurrentUsage();
        demonstrateAdvancedCancellation();

        System.out.println("\n=== SUMMARY ===");
        System.out.println("1. Implemented bounded circular buffer with ReentrantLock and Conditions");
        System.out.println("2. Added timeout support for blocking operations");
        System.out.println("3. Implemented cancellation mechanisms for writers");
        System.out.println("4. Provided statistics and monitoring capabilities");
        System.out.println("5. Demonstrated thread-safe concurrent usage");
    }
}

Key Features: Bounded circular buffer, timeout operations, cancellation support, concurrent safety, statistics tracking

Edge Cases & Testing

What Could Break

• Null elements: Queue should reject null values
• Negative timeouts: Should handle gracefully
• Interrupted threads: Proper cleanup on interruption
• Concurrent modifications: Thread safety during size changes
• Memory leaks: Proper cleanup of removed elements

Testing Scenarios

@Test
public void testBasicOperations() {
    CustomBlockingQueue<String> queue = new CustomBlockingQueue<>(3);

    assertTrue(queue.isEmpty());
    assertEquals(0, queue.size());
    assertEquals(3, queue.remainingCapacity());

    assertTrue(queue.offer("test"));
    assertFalse(queue.isEmpty());
    assertEquals(1, queue.size());
    assertEquals(2, queue.remainingCapacity());

    assertEquals("test", queue.poll());
    assertTrue(queue.isEmpty());
}

@Test
public void testTimeoutOperations() throws InterruptedException {
    CustomBlockingQueue<Integer> queue = new CustomBlockingQueue<>(1);

    // Fill the queue
    queue.offer(1);

    // Try to offer with timeout - should fail
    long startTime = System.currentTimeMillis();
    boolean success = queue.offer(2, 100, TimeUnit.MILLISECONDS);
    long endTime = System.currentTimeMillis();

    assertFalse(success);
    assertTrue("Should timeout within reasonable time", (endTime - startTime) < 200);
}
@Test
public void testCancellation() throws InterruptedException {
    CustomBlockingQueue<Integer> queue = new CustomBlockingQueue<>(1);
    queue.offer(1);

    CustomBlockingQueue.CancellableOffer<Integer> cancellableOffer = 
        new CustomBlockingQueue.CancellableOffer<>(2, 5, TimeUnit.SECONDS);

    // Cancel immediately
    cancellableOffer.cancel();

    boolean success = queue.offerWithCancellation(cancellableOffer);
    assertFalse(success);
    assertEquals(1, queue.getCancelledOffers());
}
@Test
public void testConcurrentAccess() throws InterruptedException {
    CustomBlockingQueue<Integer> queue = new CustomBlockingQueue<>(100);

    // Multiple producers and consumers
    Thread[] producers = new Thread[5];
    Thread[] consumers = new Thread[5];

    for (int i = 0; i < 5; i++) {
        final int producerId = i;
        producers[i] = new Thread(() -> {
            for (int j = 0; j < 20; j++) {
                try {
                    queue.offer(producerId * 100 + j, 1, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });

        consumers[i] = new Thread(() -> {
            for (int j = 0; j < 20; j++) {
                try {
                    queue.poll(1, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });
    }

    // Start all threads
    for (int i = 0; i < 5; i++) {
        producers[i].start();
        consumers[i].start();
    }

    // Wait for completion
    for (int i = 0; i < 5; i++) {
        producers[i].join();
        consumers[i].start();
    }

    // Verify final state
    assertEquals(0, queue.size());
    assertTrue(queue.isEmpty());
}

Real-World Adaptation

Production Considerations

• Monitoring: Track cancellation rates and timeout patterns
• Capacity tuning: Adjust queue size based on producer/consumer rates
• Timeout values: Set appropriate timeouts for your use case
• Backpressure: Use cancellation to implement backpressure strategies

Performance Optimizations

• Lock-free alternatives: Consider LMAX Disruptor for ultra-high performance
• Batch operations: Implement drainTo for bulk processing
• Memory pools: Reuse objects to reduce GC pressure

Pro Tips for Interviews

How to Talk Through the Problem

1. Start with the data structure: Circular buffer with head/tail pointers
2. Explain synchronization: ReentrantLock with Condition objects
3. Address the twist: Timeout mechanisms and cancellation support
4. Discuss edge cases: Null values, interrupted threads, memory management
5. Consider alternatives: Different cancellation strategies

Common Pitfalls to Avoid

• Not handling interrupted threads properly
• Forgetting to signal conditions after state changes
• Not considering memory leaks from removed elements
• Ignoring the bounded nature of the queue

If You Don’t Know the Answer

1. Start with a simple array-based queue
2. Add basic synchronization
3. Think about how to implement blocking
4. Consider timeout mechanisms

Conclusion

This problem demonstrates the importance of understanding concurrent data structures and implementing proper timeout and cancellation mechanisms. The key insight is that a blocking queue needs both proper synchronization and graceful handling of cancellation scenarios.

Key learnings: Circular buffers provide efficient bounded queues, Conditions enable proper thread coordination, timeout mechanisms prevent indefinite blocking, cancellation support enables responsive systems.

Related topics: Concurrent programming, producer-consumer patterns, bounded buffers, timeout handling, cancellation patterns.

==========================================

Check out the collection below for similar stories

[embed]Java Interview QnA | Java Deep Dive Lets deep dive into java and understand the basic to advanced concepts. Be updated with the latest features added in…codefarm0.medium.com

If you found this useful, please do clap the story and follow me for more such interesting and informative stories!