Collections, Generics
& Concurrency

// ArrayList: single contiguous block
transient Object[] elementData; // heap: [ref0, ref1, ref2, ...]
// grow: Arrays.copyOf(data, newCapacity = size * 3/2 + 1)

// LinkedList: scattered nodes
private static class Node<E> {
    E item;          // 8 bytes ref
    Node<E> next;   // 8 bytes ref — points ANYWHERE in heap
    Node<E> prev;   // 8 bytes ref
    // + 16 bytes object header = ~40 bytes per element
}

// Benchmark: iterating 10M ints
// ArrayList: ~8ms  (prefetcher loads 8 refs per cache line)
// LinkedList: ~80ms (cache miss on every node → main memory)

// Hash spreading — prevents high-bit clustering
static final int hash(Object key) {
    int h;
    return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}

// putVal: plain writes — NOT thread-safe!
if ((p = tab[i = (n - 1) & hash]) == null)
    tab[i] = newNode(hash, key, value, null); // plain store
else {
    for (int binCount = 0;; ++binCount) {
        if ((e = p.next) == null) {
            p.next = newNode(hash, key, value, null);
            if (binCount >= TREEIFY_THRESHOLD - 1) // 8
                treeifyBin(tab, hash); // convert to TreeNode
            break;
        }
    }
}
if (++size > threshold) resize(); // load factor 0.75

// Correct equals + hashCode (Java 16+ records do this automatically)
public boolean equals(Object o) {
    if (!(o instanceof User u)) return false;
    return age == u.age && Objects.equals(name, u.name);
}
public int hashCode() { return Objects.hash(name, age); }

final V putVal(K key, V value, boolean onlyIfAbsent) {
    for (Node<K,V>[] tab = table;;) { // volatile read each iteration
        Node<K,V> f; int n, i, fh;
        if (tab == null) tab = initTable();
        else if ((f = tabAt(tab, i=(n-1)&hash)) == null) {
            // EMPTY: lock-free CAS — no contention path
            if (casTabAt(tab, i, null, new Node<>(hash,key,value)))
                break;
        } else {
            // COLLISION: lock on first node only
            synchronized (f) {
                // traverse list or tree, update or insert
            }
        }
    }
    addCount(1L, binCount); // LongAdder-style striped counter
}

// CounterCell — padded to avoid false sharing between CPUs
@jdk.internal.vm.annotation.Contended
static final class CounterCell {
    volatile long value; // each on its own cache line
}

// TreeMap — range queries (e.g. Snowflake partition metadata)
TreeMap<Long, Partition> partitions = new TreeMap<>();
NavigableMap<Long, Partition> range =
    partitions.subMap(startTs, true, endTs, true); // O(log n) view

// LinkedHashMap LRU Cache
class LRUCache<K,V> extends LinkedHashMap<K,V> {
    private final int cap;
    LRUCache(int cap) { super(cap, 0.75f, true); this.cap=cap; }
    protected boolean removeEldestEntry(Map.Entry<K,V> e) {
        return size() > cap; // evict LRU when over capacity
    }
}

// PriorityQueue — rate limiter sorted by expiry window
PriorityQueue<Request> pq =
    new PriorityQueue<>(Comparator.comparingLong(Request::getExpiry));
pq.peek();     // earliest expiry, null if empty
pq.element(); // earliest expiry, throws NoSuchElementException
pq.poll();     // removes+returns earliest, null if empty
pq.remove();  // removes+returns earliest, throws if empty

// Simplified stale entry cleanup (runs on map operations)
private void expungeStaleEntries() {
    for (Object x; (x = queue.poll()) != null; ) {
        synchronized (queue) {
            Entry<K,V> e = (Entry<K,V>) x;
            int i = indexFor(e.hash, table.length);
            // unlink entry from bucket chain
        }
    }
}

// ⚠ Not thread-safe! Wrap with synchronizedMap if needed
Map<Key, Metadata> cache =
    Collections.synchronizedMap(new WeakHashMap<>());

// Alternative: Use Guava Cache or Caffeine with weakKeys()
Cache<Key, Value> c = Caffeine.newBuilder()
    .weakKeys().softValues().build();

public boolean add(E e) {
    synchronized (lock) {
        Object[] es = getArray();           // volatile read
        int len = es.length;
        Object[] newEs = Arrays.copyOf(es, len + 1); // O(n) copy
        newEs[len] = e;
        setArray(newEs);                  // volatile write → HB guarantee
        return true;
    }
}

// Spring ApplicationEventMulticaster uses it internally:
// private final Set<ApplicationListener<?>> applicationListeners
//    = new CopyOnWriteArraySet<>();   (backed by COWAL)

// Producer-consumer with LinkedBlockingQueue
BlockingQueue<Task> queue = new LinkedBlockingQueue<>(1000);
queue.put(task);               // blocks if full
Task t = queue.take();         // blocks until available
queue.offer(task, 1, TimeUnit.SECONDS); // timeout version

// DelayQueue for session expiry
class SessionEvent implements Delayed {
    private final long expiryNanos;
    public long getDelay(TimeUnit u) {
        return u.convert(expiryNanos - System.nanoTime(), TimeUnit.NANOSECONDS);
    }
    public int compareTo(Delayed o) {
        return Long.compare(expiryNanos, ((SessionEvent)o).expiryNanos);
    }
}
DelayQueue<SessionEvent> expiry = new DelayQueue<>();
SessionEvent expired = expiry.take(); // blocks until delay expires

// ✅ Safe: iterator.remove() (syncs expectedModCount)
Iterator<String> it = list.iterator();
while (it.hasNext())
    if (shouldRemove(it.next())) it.remove();

// ✅ Safe: removeIf (Java 8, uses modCount internally)
list.removeIf(s -> s.startsWith("X"));

// ✅ Safe: CopyOnWriteArrayList (snapshot iterator)
new CopyOnWriteArrayList<>(list).iterator();

// ✅ Safe: collect to new list then remove
List<String> toRemove = list.stream()
    .filter(this::shouldRemove).toList();
list.removeAll(toRemove);

// ❌ Broken: modifying list during forEach
list.forEach(s -> if (bad) list.remove(s)); // CME!

// Concurrent event timeline (used in Kafka consumer internals)
ConcurrentSkipListMap<Long, Event> timeline = new ConcurrentSkipListMap<>();
timeline.put(System.currentTimeMillis(), event);

// Thread-safe range query — no external lock needed
ConcurrentNavigableMap<Long,Event> lastMinute =
    timeline.tailMap(System.currentTimeMillis() - 60_000);

// ConcurrentSkipListSet for sorted deduplication
ConcurrentSkipListSet<String> tags = new ConcurrentSkipListSet<>();
tags.add("java"); tags.add("kafka");
tags.higher("j"); // "java" — lock-free O(log n)

// Bridge method — compiler generated
class Base<T> { void set(T t) {} }
class Sub extends Base<String> {
    void set(String s) { /* override */ }
    // compiler adds: void set(Object o){set((String)o);} — BRIDGE
}

// PECS: Collections.copy uses both wildcards
public static <T> void copy(
    List<? super T> dest,    // consumer: write T into it
    List<? extends T> src)   // producer: read T from it
{ ... }

// Why can't add to ? extends Number?
List<? extends Number> nums = new ArrayList<Integer>();
nums.add(1.5); // COMPILE ERROR — could be List<Integer>, adding Double is wrong

// Heap pollution
List[] raw = new List[1];
List<String>[] polluted = raw;      // unchecked
raw[0] = List.of(42);               // Integer list in String[] slot
String s = polluted[0].get(0);     // CCE here, not at assignment!

// EnumMap — state machine or permission flags
enum State { PENDING, PROCESSING, DONE, FAILED }
EnumMap<State, Color> stateColors = new EnumMap<>(State.class);
stateColors.put(State.DONE, Color.GREEN);
// Internal: Object[4] indexed by State.ordinal() — fastest possible Map

// IdentityHashMap — serialization cycle detection
// (e.g. Java ObjectOutputStream uses IdentityHashMap internally)
IdentityHashMap<Object, Integer> visited = new IdentityHashMap<>();
void serialize(Object obj) {
    if (visited.containsKey(obj)) return; // cycle detected by identity
    visited.put(obj, nextId++);
    // serialize obj...
}

// Spring AOP uses IdentityHashMap for proxy caching:
// two proxied objects wrapping same target should be distinct

// Circular buffer — head/tail wrap around
// addFirst: head moves left (wrapping), addLast: tail moves right
//  [_, _, 3, 2, 1, _, _]  head=2, tail=5
//  addFirst(4): head=(2-1)&7=1  →  [_, 4, 3, 2, 1, _, _]

// Use as Stack (LIFO)
Deque<Integer> stack = new ArrayDeque<>();
stack.push(1); stack.push(2);
stack.pop();   // 2  (LIFO)
stack.peek();  // 1  (no remove)

// Use as Queue (FIFO)
Deque<Integer> queue = new ArrayDeque<>();
queue.offer(1); queue.offer(2);
queue.poll();  // 1  (FIFO)

// BFS/DFS algorithm template
Deque<Node> frontier = new ArrayDeque<>();
frontier.offerLast(root);     // BFS: poll from front
// frontier.offerFirst(root); // DFS: poll from front too

// Volatile flag with non-volatile data — safe publication pattern
class Publisher {
    int data = 0;          // plain field
    volatile boolean ready = false;

    // Thread A
    void publish() {
        data = 42;           // plain write, but...
        ready = true;        // volatile write = StoreStore barrier above
        // data=42 happens-before ready=true happens-before Thread B read
    }

    // Thread B
    void consume() {
        if (ready) {          // volatile read
            assert data == 42; // guaranteed: HB chain intact
        }
    }
}

// NOT thread-safe — not atomic
volatile int counter = 0;
counter++;  // READ(0) ... WRITE(1) — race condition window!
// Fix: AtomicInteger.incrementAndGet() — single CAS instruction

class Foo {
    synchronized void        a() {} // lock = this (Foo instance)
    static synchronized void b() {} // lock = Foo.class

    void c() {
        synchronized(this)       {} // same as a()
        synchronized(Foo.class) {} // same as b()
        synchronized(lockObj)    {} // private lock object — best practice
    }
}
// Thread states under synchronized:
// Waiting for monitor entry → BLOCKED
// In Object.wait() → WAITING
// In Object.wait(timeout) → TIMED_WAITING

protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState(); // volatile read
    if (c == 0) {
        // UNFAIR: no queue check — barge in
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    } else if (current == getExclusiveOwnerThread()) {
        // REENTRANT: same thread holds lock
        setState(c + acquires); // plain write — already own the lock
        return true;
    }
    return false; // enqueue and park
}

// Condition example — bounded buffer
ReentrantLock lock = new ReentrantLock();
Condition notFull  = lock.newCondition();
Condition notEmpty = lock.newCondition();

// producer
lock.lock();
try {
    while (isFull()) notFull.await(); // releases lock, waits
    buf.add(item);
    notEmpty.signal(); // wake one consumer
} finally { lock.unlock(); }

class Point {
    double x, y;
    final StampedLock sl = new StampedLock();

    double distanceFromOrigin() { // called very frequently
        long stamp = sl.tryOptimisticRead(); // no lock — just a version
        double cx = x, cy = y;              // read into locals

        if (!sl.validate(stamp)) {          // write happened? re-read
            stamp = sl.readLock();         // now hold shared lock
            try { cx = x; cy = y; }
            finally { sl.unlockRead(stamp); }
        }
        return Math.hypot(cx, cy);
    }

    void move(double dx, double dy) {     // rare
        long stamp = sl.writeLock();
        try { x += dx; y += dy; }
        finally { sl.unlockWrite(stamp); }
    }
}

@Component
public class CorrelationFilter implements Filter {
    public void doFilter(ServletRequest req, ...) {
        String id = Optional
            .ofNullable(((HttpServletRequest)req).getHeader("X-Correlation-ID"))
            .orElse(UUID.randomUUID().toString());
        try {
            MDC.put("correlationId", id);
            chain.doFilter(req, res);
        } finally {
            MDC.clear(); // MUST clear — thread returns to pool!
        }
    }
}

// Propagating MDC to async tasks (thread pool)
Map<String,String> ctx = MDC.getCopyOfContextMap();
executor.submit(() -> {
    try {
        if (ctx != null) MDC.setContextMap(ctx); // restore in new thread
        doWork();
    } finally { MDC.clear(); }
});

// Detection: ThreadMXBean
ThreadMXBean tmx = ManagementFactory.getThreadMXBean();
long[] ids = tmx.findDeadlockedThreads(); // monitors + synchronizers
if (ids != null) {
    ThreadInfo[] infos = tmx.getThreadInfo(ids, true, true);
    for (ThreadInfo ti : infos)
        log.error("DEADLOCK: {} blocked on {}", ti.getThreadName(), ti.getLockName());
}

// Prevention 1: consistent lock ordering (break circular wait)
void transfer(Account a, Account b) {
    Account first  = a.getId() < b.getId() ? a : b; // always lower id first
    Account second = a.getId() < b.getId() ? b : a;
    synchronized(first) { synchronized(second) { /* transfer */ } }
}

// Prevention 2: tryLock with timeout (break hold-and-wait)
if (lockA.tryLock(100, TimeUnit.MS)) {
    try {
        if (lockB.tryLock(100, TimeUnit.MS)) {
            try { /* critical section */ }
            finally { lockB.unlock(); }
        }
    } finally { lockA.unlock(); }
}

// AtomicInteger CAS loop (incrementAndGet internals)
public final int incrementAndGet() {
    for (;;) {
        int cur = get();          // volatile read
        int next = cur + 1;
        if (compareAndSet(cur, next)) // CAS: atomic read+write
            return next;
        // retry if CAS failed (another thread incremented)
    }
}

// ABA fix: AtomicStampedReference
AtomicStampedReference<Node> head =
    new AtomicStampedReference<>(node, 0);
int[] stampHolder = new int[1];
Node curr = head.get(stampHolder);
int stamp = stampHolder[0];
head.compareAndSet(curr, newNode, stamp, stamp + 1);
// Now A→B→A is detected: stamp incremented each change

// LongAdder for high-throughput counters (metrics, rate limiters)
LongAdder hitCount = new LongAdder();
hitCount.increment(); // writes to thread-local Cell — no contention
long total = hitCount.sum(); // aggregate — NOT atomic snapshot

// Raw ThreadPoolExecutor with bounded queue
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10,                              // corePoolSize
    50,                              // maximumPoolSize
    60L, TimeUnit.SECONDS,           // keepAlive (non-core threads)
    new ArrayBlockingQueue<>(200), // bounded queue (backpressure)
    new ThreadPoolExecutor.CallerRunsPolicy() // backpressure via caller
);

// Spring @Async production configuration
@Configuration @EnableAsync
public class AsyncConfig {
    @Bean public Executor asyncExecutor() {
        ThreadPoolTaskExecutor ex = new ThreadPoolTaskExecutor();
        ex.setCorePoolSize(10);
        ex.setMaxPoolSize(50);
        ex.setQueueCapacity(200);
        ex.setThreadNamePrefix("async-worker-");
        ex.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        ex.setWaitForTasksToCompleteOnShutdown(true);
        ex.initialize(); return ex;
    }
}

// thenApply vs thenCompose
CF<String> step1 = CF.supplyAsync(() -> "hello");
// thenApply — sync transform (no new async step)
CF<Integer> r1 = step1.thenApply(String::length);  // CF<Integer> ✓
// thenCompose — flatten when next step is async
CF<User> r2 = step1.thenCompose(name ->
    userService.lookupAsync(name));  // CF<User> ✓, not CF<CF<User>>

// allOf — 3 parallel calls, aggregate
CF<User>    cfU  = getUserAsync(id);
CF<Orders> cfO  = getOrdersAsync(id);
CF<Account> cfAcc = getAccountAsync(id);

CF.allOf(cfU, cfO, cfAcc)
  .orTimeout(2, TimeUnit.SECONDS)  // Java 9+
  .thenApply(v ->
      new Dashboard(cfU.join(), cfO.join(), cfAcc.join())) // safe, done
  .exceptionally(ex -> Dashboard.empty())  // on timeout/error
  .handle((res, ex) -> {                   // always runs (try-finally)
      metrics.record(res, ex);
      return res != null ? res : Dashboard.empty();
  });

// completeOnTimeout — return default instead of throwing
cfU.completeOnTimeout(User.anonymous(), 500, TimeUnit.MS);

class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 1_000;
    int[] arr; int lo, hi;

    @Override
    protected Long compute() {
        if (hi - lo <= THRESHOLD) {
            long sum = 0;
            for (int i = lo; i < hi; i++) sum += arr[i];
            return sum;
        }
        int mid = (lo + hi) / 2;
        SumTask left  = new SumTask(arr, lo, mid);
        SumTask right = new SumTask(arr, mid, hi);
        left.fork();                        // push left to deque tail
        return right.compute() + left.join(); // compute right inline
    }
}

// Custom FJP to isolate parallel stream from commonPool
ForkJoinPool custom = new ForkJoinPool(8);
long result = custom.submit(() ->
    bigList.parallelStream().mapToLong(Item::getValue).sum()
).get();
custom.shutdown();

class CachedData {
    volatile boolean cacheValid = false;
    final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    Object[] data;

    void processCachedData() {
        rwl.readLock().lock();
        try {
            if (!cacheValid) {
                rwl.readLock().unlock(); // drop read before acquiring write
                rwl.writeLock().lock();
                try {
                    if (!cacheValid) {    // double-check after write lock
                        data = loadData();
                        cacheValid = true;
                    }
                    rwl.readLock().lock(); // DOWNGRADE: acquire read before releasing write
                } finally { rwl.writeLock().unlock(); } // release write
                // now holding read lock only — safe to proceed
            }
            use(data); // process under read lock
        } finally { rwl.readLock().unlock(); }
    }
}

// CountDownLatch — await all services to start
CountDownLatch ready = new CountDownLatch(3);
// each service calls ready.countDown() after init
ready.await(); // main waits — all writes before countDown() are visible

// CyclicBarrier — parallel batch processing
CyclicBarrier barrier = new CyclicBarrier(4,
    () -> log.info("Batch complete, merging results"));
// each worker calls barrier.await() when done with its shard
// barrier.reset() or it auto-resets for next batch round

// Semaphore — connection pool rate limiter
Semaphore semaphore = new Semaphore(10, true); // 10 permits, fair
semaphore.acquire(); // blocks if 10 already acquired
try { useResource(); } finally { semaphore.release(); }

// Phaser — dynamic multi-phase simulation
Phaser ph = new Phaser(1); // 1 initial party (orchestrator)
ph.register();                // add party dynamically
ph.arriveAndAwaitAdvance(); // participate and wait for phase N
ph.arriveAndDeregister();   // leave phaser

class MyClass {
    volatile int state = 0;
    static final VarHandle STATE;
    static {
        STATE = MethodHandles.lookup()
            .findVarHandle(MyClass.class, "state", int.class);
    }

    void demonstrate() {
        STATE.set(this, 1);                        // plain write
        int v = (int) STATE.getAcquire(this);      // acquire read
        STATE.setRelease(this, 2);                // release write
        STATE.compareAndSet(this, 0, 1);           // CAS volatile
        STATE.getAndAdd(this, 1);                  // atomic increment
        STATE.compareAndExchangeAcquire(this,0,1); // CAS + acquire fence
    }
}

// ❌ False sharing: a and b may share same 64-byte cache line
class Counters {
    volatile long a = 0; // Thread 1 writes
    volatile long b = 0; // Thread 2 writes — same cache line!
}

// ✅ Fix with @Contended (adds ~128 bytes padding around each)
class Counters {
    @jdk.internal.vm.annotation.Contended
    volatile long a = 0; // own 128-byte padded cache line
    @jdk.internal.vm.annotation.Contended
    volatile long b = 0; // own 128-byte padded cache line
}

// JDK ConcurrentHashMap uses this for its striped counter
@Contended static final class CounterCell {
    volatile long value; // each CounterCell on its own cache line
    // Without @Contended: 16 threads → 16 writes → all invalidate same line
    // With @Contended: each thread hits its own cache line → linear throughput
}

// Manual padding (pre-Java 9, ugly but works)
long p1,p2,p3,p4,p5,p6,p7; // 56 bytes prefix
volatile long value;         // 8 bytes on its own line
long q1,q2,q3,q4,q5,q6,q7; // 56 bytes suffix

// park/unpark — AQS uses this for all lock implementations
Thread waiter = new Thread(() -> {
    System.out.println("parking...");
    LockSupport.park();          // blocks — may have spurious wakeups
    System.out.println("unparked");
});
waiter.start();
Thread.sleep(100);
// data write before unpark — waiter will see it
LockSupport.unpark(waiter);    // wake waiter — HB established

// unpark BEFORE park — still works
LockSupport.unpark(waiter);   // permit granted now
LockSupport.park(waiter);     // returns immediately — permit consumed

// Spurious wakeup guard (always check condition in loop)
while (!condition) {
    LockSupport.park(this);     // 'this' = blocker object for jstack
    if (Thread.interrupted())
        throw new InterruptedException();
}

// Pattern 1: DCL with volatile (acceptable)
public class Singleton {
    private static volatile Singleton instance; // volatile REQUIRED
    public static Singleton get() {
        if (instance == null) {              // fast path (no lock)
            synchronized (Singleton.class) {
                if (instance == null)        // re-check under lock
                    instance = new Singleton();
            }
        }
        return instance;
    }
}

// Pattern 2: Initialization-on-Demand Holder (preferred)
// Class loading is thread-safe by JVM spec (§12.4.2)
public class Singleton2 {
    private static class Holder {
        static final Singleton2 INSTANCE = new Singleton2(); // lazy+safe
    }
    public static Singleton2 get() { return Holder.INSTANCE; }
}

// Pattern 3: Enum singleton (simplest, serialization-safe)
public enum Config {
    INSTANCE;
    public void doWork() { ... } // guaranteed single instance
}
Config.INSTANCE.doWork();

// Exchanger: zero-copy double buffer pattern
Exchanger<List<Event>> ex = new Exchanger<>();

// Producer
new Thread(() -> {
    List<Event> buf = new ArrayList<>();
    while(true) {
        buf.add(nextEvent());
        if (buf.size() == 100) {
            buf = ex.exchange(buf); // swap full for empty list
            buf.clear();
        }
    }
});

// Consumer
new Thread(() -> {
    List<Event> buf = new ArrayList<>();
    while(true) {
        buf = ex.exchange(buf);    // get full, return empty
        buf.forEach(this::process);
        buf.clear();
    }
});

// ❌ AtomicReference: 1 extra heap object per node
class Node<T> {
    T value;
    AtomicReference<Node<T>> next = new AtomicReference<>();
    // 24+ bytes per node just for the AtomicReference wrapper
}

// ✅ ARFU: one shared updater, zero per-node overhead
class Node<T> {
    T value;
    volatile Node<T> next;   // just a volatile reference

    static final AtomicReferenceFieldUpdater<Node, Node> NEXT =
        AtomicReferenceFieldUpdater.newUpdater(Node.class, Node.class, "next");

    boolean casNext(Node<T> expect, Node<T> update) {
        return NEXT.compareAndSet(this, expect, update);
    }
}

// VarHandle alternative (Java 9+, preferred)
static final VarHandle NEXT_VH = MethodHandles.lookup()
    .findVarHandle(Node.class, "next", Node.class);

// Create virtual threads
Thread vt = Thread.ofVirtual().name("vt-1").start(() -> {
    // Blocking IO here is fine — carrier freed while blocked
    var result = dbService.queryBlocking(id); // unmounts, doesn't block carrier
});

// ExecutorService with virtual threads — no pooling needed!
try (var exec = Executors.newVirtualThreadPerTaskExecutor()) {
    for (var req : requests)
        exec.submit(() -> handleRequest(req)); // 1M concurrent = fine
}

// ⚠ PINNING: synchronized + blocking = carrier pinned!
synchronized(this) {
    Thread.sleep(1000); // carrier thread blocked — pinning!
}
// ✅ Fix: use ReentrantLock
lock.lock();
try { Thread.sleep(1000); } // carrier freed — no pinning
finally { lock.unlock(); }

// Spring Boot 3.2+ application.yml
// spring.threads.virtual.enabled: true

// StructuredTaskScope: ShutdownOnFailure pattern
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<User>    userTask = scope.fork(() -> fetchUser(id));
    Subtask<Orders> orderTask = scope.fork(() -> fetchOrders(id));

    scope.join(); // wait for all tasks
    scope.throwIfFailed(); // propagate first failure, cancels others

    return new Dashboard(userTask.get(), orderTask.get());
} // scope closes: all forked threads guaranteed complete or cancelled

// ScopedValue: immutable per-request context
static final ScopedValue<RequestContext> CTX = ScopedValue.newInstance();

// Set at request entry point
ScopedValue.where(CTX, new RequestContext(correlationId))
    .run(() -> handleRequest());

// Read anywhere in the call tree (including child virtual threads)
void deepMethod() {
    String id = CTX.get().correlationId(); // no ThreadLocal leak possible
}

// Only 4 filter calls — stops at first match ≥ 10
List.of(1,3,5,10,20).stream()
    .filter(n -> { System.out.print(n + " "); return n >= 10; })
    .findFirst(); // prints: 1 3 5 10  (not 20!)

// Intermediate ops are STATELESS vs STATEFUL:
// Stateless: filter, map, flatMap, peek — no ordering dependency
// Stateful: sorted, distinct, limit — need to see data before outputting

// Custom Spliterator for custom data source
class RangeSpliterator implements Spliterator.OfInt {
    int current, end;
    public boolean tryAdvance(IntConsumer action) {
        if (current < end) { action.accept(current++); return true; }
        return false;
    }
    public Spliterator.OfInt trySplit() {
        int mid = (current + end) / 2;
        if (mid == current) return null; // too small to split
        int lo = current; current = mid;
        return new RangeSpliterator(lo, mid); // left half to new thread
    }
}

// groupingBy with downstream: dept → avg salary
Map<String, Double> avgSalary = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::getDept,
        Collectors.averagingDouble(Employee::getSalary)));

// toMap with merge (avoid key collision exception)
Map<String, Long> wordCount = words.stream()
    .collect(Collectors.toMap(w -> w, w -> 1L, Long::sum));

// teeing (Java 12): two collectors in one pass
record Stats(double avg, Optional<Employee> highest) {}
Stats stats = employees.stream().collect(Collectors.teeing(
    Collectors.averagingDouble(Employee::getSalary),
    Collectors.maxBy(Comparator.comparingDouble(Employee::getSalary)),
    Stats::new));

// Custom Collector: immutable set with running total
Collector<Integer, ?, Map<String,Object>> custom =
    Collector.of(
        () -> new int[]{0, 0},             // [sum, count]
        (arr, n) -> { arr[0]+=n; arr[1]++; }, // accumulate
        (a, b) -> new int[]{a[0]+b[0], a[1]+b[1]}, // combine (parallel)
        arr -> Map.of("sum", arr[0], "avg", arr[1]>0 ? arr[0]/arr[1] : 0) // finisher
    );

// String methods
"  hello  ".strip();         // "hello" — Unicode-aware (vs trim)
"  ".isBlank();             // true — whitespace check
"a\nb\nc".lines().toList(); // ["a","b","c"]
"ab".repeat(3);             // "ababab"
"".stripLeading();           // strip only from left/right

// var in lambda parameters (Java 11)
(var s, var t) -> s + t      // allows @NotNull var s — for annotations

// Files.readString / writeString
String content = Files.readString(Path.of("config.json"));
Files.writeString(Path.of("out.txt"), "hello", StandardOpenOption.CREATE);

// New HttpClient (non-blocking)
HttpClient client = HttpClient.newBuilder()
    .connectTimeout(Duration.ofSeconds(5))
    .build();
HttpRequest req = HttpRequest.newBuilder()
    .uri(URI.create("https://api.example.com/data"))
    .header("Authorization", "Bearer " + token)
    .GET().build();

// Async HTTP call
CompletableFuture<String> body = client
    .sendAsync(req, HttpResponse.BodyHandlers.ofString())
    .thenApply(HttpResponse::body);

// Collection.toArray with type
String[] arr = list.toArray(String[]::new); // no reflection needed

// Record — immutable data carrier
record Point(double x, double y) {
    // compact canonical constructor for validation
    Point { if (x < 0) throw new IllegalArgumentException(); }
    double distance() { return Math.hypot(x, y); } // extra methods OK
}
Point p = new Point(3, 4);
p.x(); p.y(); p.distance(); // accessors — no getX()

// Sealed class + exhaustive pattern matching switch
sealed interface Shape
    permits Circle, Rectangle, Triangle {}

record Circle(double r) implements Shape {}
record Rectangle(double w, double h) implements Shape {}
record Triangle(double base, double h) implements Shape {}

double area(Shape s) {
    return switch (s) { // Java 21: exhaustive — no default needed
        case Circle(var r)         -> Math.PI * r * r;
        case Rectangle(var w, var h) -> w * h;
        case Triangle(var b, var h)  -> 0.5 * b * h;
    };
}

// Text blocks (Java 15)
String json = """
    {
        "name": "Bheemesh",
        "role": "Senior Java Developer"
    }
    """; // indentation stripped automatically

// Pattern Matching for switch (JEP 441 — finalized)
Object obj = getResult();
String desc = switch (obj) {
    case Integer i    -> "int: " + i;
    case String s     -> "str: " + s;
    case Long l when l > 1000 -> "big long"; // guard clause
    case null         -> "null";
    default           -> "unknown";
};

// SequencedCollection (JEP 431 — finalized)
// New interface for ordered collections with defined first/last
SequencedCollection<String> seq = new ArrayList<>(List.of("a","b","c"));
seq.getFirst();           // "a" — unified API across List/Deque/LinkedHashSet
seq.getLast();            // "c"
seq.addFirst("z");        // ["z","a","b","c"]
seq.reversed();           // reversed view ["c","b","a","z"]

// SequencedMap — firstEntry / lastEntry
SequencedMap<String,Integer> sm =
    new LinkedHashMap<>(Map.of("a",1,"b",2));
sm.firstEntry();  sm.lastEntry();

// Record Patterns in switch (JEP 440 — finalized)
if (shape instanceof Circle(var r) && r > 10) {
    // r is in scope here — pattern match + destructuring
}

// Unnamed Classes and Instance Main (Java 21 preview)
// void main() { System.out.println("Hello"); } — no class needed

// ✅ Correct — return type only
public Optional<User> findById(Long id) { ... }

// map vs flatMap
Optional<String> name = findUser(id)
    .map(User::getName);            // Optional<String>
Optional<Address> addr = findUser(id)
    .flatMap(User::getOptionalAddr); // flatten Optional<Optional<Address>>

// orElse: ALWAYS evaluates even if present
String s = opt.orElse(expensiveDefault()); // called unconditionally!
// orElseGet: LAZY — only calls if empty
String s2 = opt.orElseGet(() -> expensiveDefault()); // only if empty

// Java 9+: ifPresentOrElse
opt.ifPresentOrElse(v -> log.info("found {}", v),
                    () -> log.warn("not found"));

// Java 9+: or — supply alternative Optional
Optional<User> user = findInDB(id)
    .or(() -> findInCache(id));   // fallback chain

// Java 9+: stream — bridge to Stream pipeline
List<User> users = ids.stream()
    .map(this::findById)          // Stream<Optional<User>>
    .flatMap(Optional::stream)   // Java 9: unwrap non-empty
    .toList();

// Java 7 circular list (simplified transfer logic)
void transfer(Entry[] newTable) {
    for (Entry<K,V> e : table) {
        while (e != null) {
            Entry<K,V> next = e.next;
            e.next = newTable[i]; // Thread A reverses here
            newTable[i] = e;      // Thread B does same → A.next=B, B.next=A
            e = next;             // → CIRCULAR → infinite loop on get()
        }
    }
}

// Java 8: table reference overwrite data loss
// Thread A: newTab = new Node[32]; ... table = newTab; ✓
// Thread B: newTab = new Node[32]; ... table = newTab; ← overwrites A!
// → A's entries inserted before B's resize are LOST

// ✅ Always use ConcurrentHashMap
Map<K,V> safe = new ConcurrentHashMap<>();
// CAS + per-bucket sync: zero data loss, no infinite loops

class BrokenSingleton {
    private static BrokenSingleton instance; // NOT volatile
    private int value = 42;

    public static BrokenSingleton get() {
        if (instance == null) {
            synchronized(BrokenSingleton.class) {
                if (instance == null)
                    instance = new BrokenSingleton(); // BUG 1
            }
        }
        return instance; // BUG 2
    }

    void use() {
        BrokenSingleton s = get();
        assert s.value == 42; // Can FAIL!
    }
}

// BUG 1: new BrokenSingleton() may be reordered by JIT:
//   a) allocate memory
//   b) write reference to instance  ← Thread B sees non-null here
//   c) initialize fields (value=42)  ← NOT YET DONE
// Thread B reads instance != null, skips sync, reads value = 0!

// BUG 2: Without volatile, Thread B may read stale null
// from its CPU cache even after Thread A wrote it — visibility!

// FIX: private static volatile BrokenSingleton instance;
// volatile write establishes HB: constructor end HB volatile write HB every read

CompletableFuture<Integer> idFuture =
    CompletableFuture.supplyAsync(() -> 42);

// Q1: What is the type of r1?
var r1 = idFuture.thenApply(id ->
    CompletableFuture.supplyAsync(() -> "User:" + id));
// A: CompletableFuture<CompletableFuture<String>> — NESTED! Wrong.
// You'd need r1.join().join() — bad.

// Q2: What is the type of r2?
var r2 = idFuture.thenCompose(id ->
    CompletableFuture.supplyAsync(() -> "User:" + id));
// A: CompletableFuture<String> — FLATTENED. Correct!
// thenCompose = flatMap in Stream / Mono.flatMap in Reactor

// Rule: thenApply when fn returns T (not CF)
//       thenCompose when fn returns CompletionStage<T>

// Q3: What thread executes thenApply?
var r3 = idFuture.thenApply(id -> id * 2);
// A: The thread that completed idFuture (ForkJoinPool worker)
//    OR the calling thread if already complete at registration time
// thenApplyAsync: ALWAYS a new ForkJoinPool thread

// ❌ BUG: ArrayList not thread-safe — concurrent add causes corruption
List<Integer> results = new ArrayList<>();
IntStream.range(0, 1000).parallel()
    .forEach(results::add);
// results.size() may be 983, 995, or throw ArrayIndexOutOfBoundsException
// Two threads expand array simultaneously → corrupted state

// ❌ BUG: shared counter — race condition
int[] count = {0};
IntStream.range(0, 1000).parallel()
    .forEach(i -> count[0]++); // lost updates!

// ✅ FIX 1: use collect() — thread-safe combiner
List<Integer> safe = IntStream.range(0, 1000).parallel()
    .boxed()
    .collect(Collectors.toList()); // always correct — thread-safe merge

// ✅ FIX 2: thread-safe collection
List<Integer> safe2 = new CopyOnWriteArrayList<>();
// or Vector, or ConcurrentLinkedQueue

// ✅ FIX 3: LongAdder for counting
LongAdder cnt = new LongAdder();
IntStream.range(0, 1000).parallel().forEach(i -> cnt.increment());
cnt.sum(); // always 1000

// Puzzle 1: Stream reuse — throws IllegalStateException
Stream<String> s = Stream.of("a","b","c");
s.forEach(System.out::println);
s.forEach(System.out::println); // IllegalStateException: stream already operated
// Streams are single-use pipelines

// Puzzle 2: stateful lambda in parallel (non-deterministic)
List<Integer> seen = new ArrayList<>();
IntStream.range(0,10).parallel()
    .filter(i -> { seen.add(i); return i%2==0; }) // STATEFUL SIDE EFFECT
    .count();
// seen.size() may be 6, 8, 10 — non-deterministic
// filter may be called in any order, possibly concurrently

// Puzzle 3: sorted + parallel order
List.of(3,1,4,1,5).parallelStream()
    .sorted()
    .forEachOrdered(System.out::println); // [1,1,3,4,5] — ordered despite parallel
// sorted() is stateful — must see all before outputting
// forEachOrdered preserves encounter order in parallel

// Puzzle 4: peek() in parallel
long count = List.of(1,2,3).parallelStream()
    .peek(System.out::println)  // may print in any order
    .count();                   // count = 3, always correct

// Thread-safe word counter via merge
ConcurrentHashMap<String, Long> wordCount = new ConcurrentHashMap<>();

// merge: if absent insert 1L; if present add 1L (atomic)
words.parallelStream()
    .forEach(w -> wordCount.merge(w, 1L, Long::sum));

// compute — atomic read-modify-write for complex state
wordCount.compute("hello", (k, v) -> v == null ? 1L : v + 1L);

// computeIfAbsent — lazy initialization pattern
ConcurrentHashMap<String, List<Event>> groups = new ConcurrentHashMap<>();
groups.computeIfAbsent(key, k -> new CopyOnWriteArrayList<>())
      .add(event); // thread-safe group creation

// ⚠ DON'T do this — not atomic, race between get + put
if (!map.containsKey(k)) map.put(k, new ArrayList<>()); // RACE!

// frequency — Java 8 reduce + CHM
Map<String, Long> freq = words.stream()
    .collect(Collectors.groupingByConcurrent(
        w -> w, Collectors.counting()));  // parallel-safe

// Rate limiter with Semaphore (N requests/sec)
class SemaphoreRateLimiter {
    private final Semaphore sem;
    SemaphoreRateLimiter(int rps) {
        sem = new Semaphore(0);
        Executors.newSingleThreadScheduledExecutor()
            .scheduleAtFixedRate(() -> sem.release(rps), 0, 1, TimeUnit.SECONDS);
    }
    void acquire() throws InterruptedException { sem.acquire(); }
}

// Concurrent cache with computeIfAbsent + expiry
class TimedCache<K,V> {
    record Entry<V>(V value, long expiryMs) {}
    private final ConcurrentHashMap<K, Entry<V>> map = new ConcurrentHashMap<>();

    V get(K key, Supplier<V> loader, long ttlMs) {
        long now = System.currentTimeMillis();
        return map.compute(key, (k, e) -> {
            if (e != null && e.expiryMs() > now) return e; // fresh
            return new Entry<>(loader.get(), now + ttlMs);  // load + refresh
        }).value();
    }
}

// Token bucket rate limiter (AtomicLong based)
class TokenBucket {
    private final int capacity;
    private final AtomicLong tokens;
    private volatile long lastRefill = System.nanoTime();

    synchronized boolean tryConsume() {
        refill();
        if (tokens.get() > 0) { tokens.decrementAndGet(); return true; }
        return false;
    }
    private void refill() {
        long now = System.nanoTime();
        long add = (now - lastRefill) / 1_000_000; // per ms
        if (add > 0) {
            tokens.set(Math.min(capacity, tokens.get() + add));
            lastRefill = now;
        }
    }
}

// MDC-aware TaskDecorator for @Async propagation
@Bean
public Executor asyncExecutor() {
    ThreadPoolTaskExecutor ex = new ThreadPoolTaskExecutor();
    ex.setTaskDecorator(runnable -> {
        Map<String,String> ctx = MDC.getCopyOfContextMap(); // capture
        return () -> {
            try {
                if (ctx != null) MDC.setContextMap(ctx); // restore in new thread
                runnable.run();
            } finally { MDC.clear(); } // always clean up
        };
    });
    ex.setCorePoolSize(10);
    ex.initialize();
    return ex;
}

// Spring Boot 3.2+: enable virtual threads for Tomcat
// application.yml:
// spring:
//   threads:
//     virtual:
//       enabled: true
// This switches Tomcat's request handler from platform thread pool
// to Executors.newVirtualThreadPerTaskExecutor()

// Spring Boot 3.2+: virtual thread bean config
@Bean
public Executor virtualThreadExecutor() {
    return Executors.newVirtualThreadPerTaskExecutor();
}

// ⚠ ThreadLocal caveat with virtual threads:
// Each VT has its own ThreadLocalMap — value not shared, but
// with 100k concurrent VTs, 100k copies of ThreadLocal data in heap
// Use ScopedValue instead for read-only per-request context

// Factory configuration
@Bean
public ConcurrentKafkaListenerContainerFactory<String,Order> factory() {
    var factory = new ConcurrentKafkaListenerContainerFactory<String,Order>();
    factory.setConsumerFactory(consumerFactory);
    factory.setConcurrency(3); // 3 threads = 3 partition consumers
    factory.getContainerProperties()
           .setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    return factory;
}

// Listener with manual offset commit
@KafkaListener(topics = "orders", concurrency = "3")
public void onOrder(ConsumerRecord<String,Order> record,
                     Acknowledgment ack) {
    try {
        processOrder(record.value());
        ack.acknowledge(); // manual commit after success
    } catch (Exception e) {
        log.error("Processing failed: {}", record.offset(), e);
        // don't ack → message retried on next poll
    }
}

// Offset tracking: HashMap<TopicPartition, OffsetAndMetadata>
// Per-thread (per container) — no synchronization needed
// ConcurrentHashMap used at the KafkaConsumer coordinator level