order service 12

Story 12: Implement Graceful Shutdown and Resource Cleanup

// GracefulShutdownProcessor.java
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.*;
import java.io.*;
import java.nio.file.*;

public class GracefulShutdownProcessor {
    private final ConcurrentHashMap<String, AtomicInteger> inventory;
    private final ThreadPoolExecutor mainExecutor;
    private final ScheduledExecutorService scheduledExecutor;
    private final ExecutorService criticalExecutor;
    
    // Shutdown coordination
    private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false);
    private final CountDownLatch shutdownLatch = new CountDownLatch(1);
    private final Set<CompletableFuture<?>> inFlightRequests = ConcurrentHashMap.newKeySet();
    
    // Persistent state management
    private final BlockingQueue<PersistentState> stateQueue;
    private final Thread persistenceThread;
    
    // Metrics for shutdown reporting
    private final AtomicInteger totalProcessed = new AtomicInteger(0);
    private final AtomicInteger rejectedDuringShutdown = new AtomicInteger(0);
    private final AtomicInteger completedDuringShutdown = new AtomicInteger(0);
    
    // Services
    private final PaymentService paymentService;
    private final NotificationService notificationService;
    
    // Configuration
    private static final int SHUTDOWN_TIMEOUT_SECONDS = 30;
    private static final String STATE_FILE = "order_processor_state.dat";

    public GracefulShutdownProcessor() {
        inventory = new ConcurrentHashMap<>();
        initializeInventory();
        
        // Configure executors with custom rejection handlers
        int cpuCores = Runtime.getRuntime().availableProcessors();
        
        mainExecutor = new ThreadPoolExecutor(
            cpuCores,
            cpuCores * 2,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000),
            new ThreadFactory() {
                private final AtomicInteger threadNumber = new AtomicInteger(1);
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r, "Order-Worker-" + threadNumber.getAndIncrement());
                    t.setUncaughtExceptionHandler((thread, ex) -> {
                        System.err.println("Uncaught exception in " + thread.getName() + ": " + ex);
                        ex.printStackTrace();
                    });
                    return t;
                }
            },
            new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    if (!shutdownInitiated.get()) {
                        // Normal rejection - queue full
                        System.err.println("Task rejected - queue full. Consider increasing capacity.");
                    } else {
                        // Rejection during shutdown
                        rejectedDuringShutdown.incrementAndGet();
                        System.out.println("Task rejected during shutdown.");
                    }
                }
            }
        );
        
        // Critical tasks executor - for important cleanup operations
        criticalExecutor = Executors.newCachedThreadPool(r -> {
            Thread t = new Thread(r, "Critical-Task");
            t.setPriority(Thread.MAX_PRIORITY);
            return t;
        });
        
        scheduledExecutor = Executors.newScheduledThreadPool(2);
        
        // State persistence
        stateQueue = new LinkedBlockingQueue<>();
        persistenceThread = new Thread(this::runPersistence, "State-Persistence");
        persistenceThread.start();
        
        // Services
        paymentService = new PaymentService();
        notificationService = new NotificationService();
        
        // Register shutdown hook
        registerShutdownHook();
        
        // Start monitoring
        startMonitoring();
        
        // Load previous state if exists
        loadState();
        
        System.out.println("✅ GracefulShutdownProcessor initialized");
    }

    private void initializeInventory() {
        inventory.put("Laptop", new AtomicInteger(1000));
        inventory.put("Mouse", new AtomicInteger(2000));
        inventory.put("Keyboard", new AtomicInteger(1500));
        inventory.put("Monitor", new AtomicInteger(750));
        inventory.put("iPhone", new AtomicInteger(1000));
        inventory.put("AirPods", new AtomicInteger(1200));
    }

    private void registerShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("\n🛑 Shutdown hook triggered");
            try {
                initiateGracefulShutdown();
            } catch (Exception e) {
                System.err.println("Error during shutdown: " + e);
                e.printStackTrace();
            }
        }, "Shutdown-Hook"));
    }

    private void startMonitoring() {
        // Monitor in-flight requests
        scheduledExecutor.scheduleAtFixedRate(() -> {
            if (!shutdownInitiated.get()) {
                System.out.printf("📊 Active requests: %d, Total processed: %d\n",
                    inFlightRequests.size(), totalProcessed.get());
            }
        }, 5, 5, TimeUnit.SECONDS);
        
        // Periodic state save
        scheduledExecutor.scheduleAtFixedRate(this::saveState, 30, 30, TimeUnit.SECONDS);
    }

    public CompletableFuture<OrderResult> processOrder(Order order) {
        // Check if shutdown is in progress
        if (shutdownInitiated.get()) {
            rejectedDuringShutdown.incrementAndGet();
            return CompletableFuture.completedFuture(
                new OrderResult(order.getOrderId(), false, "System shutting down"));
        }
        
        // Track this request
        CompletableFuture<OrderResult> future = new CompletableFuture<>();
        inFlightRequests.add(future);
        
        // Process order
        CompletableFuture<OrderResult> processingFuture = CompletableFuture
            .supplyAsync(() -> {
                System.out.println("Processing order: " + order.getOrderId());
                return validateOrder(order);
            }, mainExecutor)
            .thenCompose(validationResult -> {
                if (!validationResult.isSuccess() || shutdownInitiated.get()) {
                    return CompletableFuture.completedFuture(validationResult);
                }
                
                return CompletableFuture.supplyAsync(() -> 
                    updateInventory(order), mainExecutor);
            })
            .thenCompose(inventoryResult -> {
                if (!inventoryResult.isSuccess() || shutdownInitiated.get()) {
                    return CompletableFuture.completedFuture(inventoryResult);
                }
                
                // Process payment
                return paymentService.processPaymentAsync(order, mainExecutor)
                    .thenApply(paymentSuccess -> {
                        if (!paymentSuccess) {
                            rollbackInventory(order);
                            return new OrderResult(order.getOrderId(), false, "Payment failed");
                        }
                        
                        // Queue state for persistence
                        stateQueue.offer(new PersistentState(order, "COMPLETED"));
                        
                        return new OrderResult(order.getOrderId(), true, "Order completed");
                    });
            })
            .whenComplete((result, throwable) -> {
                // Remove from tracking
                inFlightRequests.remove(future);
                totalProcessed.incrementAndGet();
                
                if (shutdownInitiated.get()) {
                    completedDuringShutdown.incrementAndGet();
                }
                
                // Complete the tracking future
                if (throwable != null) {
                    future.completeExceptionally(throwable);
                } else {
                    future.complete(result);
                }
            });
        
        return future;
    }

    public void initiateGracefulShutdown() throws InterruptedException {
        if (shutdownInitiated.compareAndSet(false, true)) {
            System.out.println("\n🔄 Initiating graceful shutdown...");
            long startTime = System.currentTimeMillis();
            
            // Step 1: Stop accepting new requests
            System.out.println("1️⃣ Stopping new request acceptance");
            
            // Step 2: Cancel scheduled tasks
            System.out.println("2️⃣ Cancelling scheduled tasks");
            scheduledExecutor.shutdown();
            
            // Step 3: Wait for in-flight requests
            System.out.println("3️⃣ Waiting for " + inFlightRequests.size() + " in-flight requests...");
            
            // Create a future that completes when all in-flight requests are done
            CompletableFuture<Void> allInFlight = CompletableFuture.allOf(
                inFlightRequests.toArray(new CompletableFuture[0])
            );
            
            try {
                allInFlight.get(SHUTDOWN_TIMEOUT_SECONDS / 2, TimeUnit.SECONDS);
                System.out.println("✅ All in-flight requests completed");
            } catch (TimeoutException e) {
                System.out.println("⚠️  Timeout waiting for in-flight requests. " + 
                                 inFlightRequests.size() + " still pending.");
                
                // Cancel remaining requests
                inFlightRequests.forEach(f -> f.cancel(true));
            } catch (ExecutionException e) {
                System.err.println("Error during in-flight request completion: " + e);
            }
            
            // Step 4: Shutdown executors
            System.out.println("4️⃣ Shutting down executors");
            mainExecutor.shutdown();
            
            if (!mainExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                System.out.println("⚠️  Force shutting down main executor");
                List<Runnable> pending = mainExecutor.shutdownNow();
                System.out.println("   Cancelled " + pending.size() + " pending tasks");
            }
            
            // Step 5: Save final state
            System.out.println("5️⃣ Saving final state");
            saveState();
            
            // Step 6: Shutdown persistence
            System.out.println("6️⃣ Finalizing persistence");
            stateQueue.offer(new PersistentState(null, "SHUTDOWN")); // Poison pill
            persistenceThread.join(5000);
            
            // Step 7: Cleanup critical resources
            System.out.println("7️⃣ Cleaning up critical resources");
            cleanupResources();
            
            // Step 8: Final report
            long shutdownTime = System.currentTimeMillis() - startTime;
            printShutdownReport(shutdownTime);
            
            // Signal shutdown complete
            shutdownLatch.countDown();
            
            // Shutdown critical executor last
            criticalExecutor.shutdown();
            criticalExecutor.awaitTermination(5, TimeUnit.SECONDS);
        }
    }

    private void runPersistence() {
        try {
            while (true) {
                PersistentState state = stateQueue.take();
                
                // Check for shutdown signal
                if (state.order == null && "SHUTDOWN".equals(state.status)) {
                    System.out.println("Persistence thread shutting down");
                    break;
                }
                
                // Simulate writing to persistent storage
                Thread.sleep(10);
                
                // In real implementation, would write to database or file
                System.out.println("💾 Persisted order: " + state.order.getOrderId());
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("Persistence thread interrupted");
        }
    }

    private void saveState() {
        try {
            criticalExecutor.submit(() -> {
                try (ObjectOutputStream oos = new ObjectOutputStream(
                        Files.newOutputStream(Paths.get(STATE_FILE)))) {
                    
                    // Save inventory state
                    Map<String, Integer> inventorySnapshot = new HashMap<>();
                    inventory.forEach((k, v) -> inventorySnapshot.put(k, v.get()));
                    
                    oos.writeObject(inventorySnapshot);
                    oos.writeInt(totalProcessed.get());
                    oos.writeLong(System.currentTimeMillis());
                    
                    System.out.println("💾 State saved successfully");
                } catch (IOException e) {
                    System.err.println("Failed to save state: " + e);
                }
            }).get(5, TimeUnit.SECONDS);
        } catch (Exception e) {
            System.err.println("Error during state save: " + e);
        }
    }

    @SuppressWarnings("unchecked")
    private void loadState() {
        Path stateFile = Paths.get(STATE_FILE);
        if (Files.exists(stateFile)) {
            try (ObjectInputStream ois = new ObjectInputStream(
                    Files.newInputStream(stateFile))) {
                
                Map<String, Integer> savedInventory = (Map<String, Integer>) ois.readObject();
                savedInventory.forEach((k, v) -> {
                    AtomicInteger current = inventory.get(k);
                    if (current != null) {
                        current.set(v);
                    }
                });
                
                int savedProcessed = ois.readInt();
                long savedTime = ois.readLong();
                
                totalProcessed.set(savedProcessed);
                
                System.out.printf("📂 Loaded previous state: %d orders, saved at %s\n",
                    savedProcessed, new Date(savedTime));
            } catch (Exception e) {
                System.err.println("Failed to load state: " + e);
            }
        }
    }

    private void cleanupResources() {
        try {
            // Close database connections, file handles, etc.
            System.out.println("🧹 Cleaning up resources...");
            
            // Example: Close any open connections
            Thread.sleep(100); // Simulate cleanup
            
            System.out.println("✅ Resources cleaned up");
        } catch (Exception e) {
            System.err.println("Error during resource cleanup: " + e);
        }
    }

    private void printShutdownReport(long shutdownTime) {
        System.out.println("\n========== SHUTDOWN REPORT ==========");
        System.out.println("Total orders processed: " + totalProcessed.get());
        System.out.println("Orders completed during shutdown: " + completedDuringShutdown.get());
        System.out.println("Orders rejected during shutdown: " + rejectedDuringShutdown.get());
        System.out.println("Shutdown time: " + shutdownTime + " ms");
        System.out.println("Final inventory state:");
        inventory.forEach((item, count) -> 
            System.out.println("  " + item + ": " + count.get()));
        System.out.println("====================================\n");
    }

    public void awaitShutdown() throws InterruptedException {
        shutdownLatch.await();
    }

    private OrderResult validateOrder(Order order) {
        simulateWork(20);
        return new OrderResult(order.getOrderId(), true, "Valid");
    }

    private OrderResult updateInventory(Order order) {
        for (String item : order.getItems()) {
            AtomicInteger stock = inventory.get(item);
            if (stock == null || stock.get() < 1) {
                return new OrderResult(order.getOrderId(), false, "No stock");
            }
            stock.decrementAndGet();
        }
        return new OrderResult(order.getOrderId(), true, "Inventory updated");
    }

    private void rollbackInventory(Order order) {
        for (String item : order.getItems()) {
            AtomicInteger stock = inventory.get(item);
            if (stock != null) {
                stock.incrementAndGet();
            }
        }
    }

    private void simulateWork(int ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // Helper class for state persistence
    private static class PersistentState {
        final Order order;
        final String status;
        final long timestamp;

        PersistentState(Order order, String status) {
            this.order = order;
            this.status = status;
            this.timestamp = System.currentTimeMillis();
        }
    }
}

// ShutdownDemo.java - Demonstrates graceful shutdown
import java.util.*;
import java.util.concurrent.*;

public class ShutdownDemo {
    public static void main(String[] args) throws Exception {
        GracefulShutdownProcessor processor = new GracefulShutdownProcessor();
        
        System.out.println("🚀 Starting Graceful Shutdown Demo\n");
        
        // Start order generation threads
        ExecutorService orderGenerators = Executors.newFixedThreadPool(3);
        AtomicBoolean keepGenerating = new AtomicBoolean(true);
        AtomicInteger orderId = new AtomicInteger(7000);
        
        // Generator 1: Regular orders
        orderGenerators.submit(() -> {
            Random random = new Random();
            while (keepGenerating.get()) {
                try {
                    Order order = createRandomOrder(orderId.incrementAndGet(), random);
                    processor.processOrder(order);
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
        
        // Generator 2: Burst orders
        orderGenerators.submit(() -> {
            Random random = new Random();
            while (keepGenerating.get()) {
                try {
                    // Generate burst of 10 orders
                    for (int i = 0; i < 10; i++) {
                        Order order = createRandomOrder(orderId.incrementAndGet(), random);
                        processor.processOrder(order);
                    }
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
        
        // Simulate different shutdown scenarios
        Scanner scanner = new Scanner(System.in);
        System.out.println("\nCommands:");
        System.out.println("  1 - Normal shutdown");
        System.out.println("  2 - Emergency shutdown (Ctrl+C simulation)");
        System.out.println("  3 - Continue running");
        System.out.println("  q - Quit\n");
        
        while (true) {
            System.out.print("Enter command: ");
            String command = scanner.nextLine().trim();
            
            switch (command) {
                case "1":
                    System.out.println("\n📋 Initiating normal shutdown...");
                    keepGenerating.set(false);
                    orderGenerators.shutdown();
                    
                    // Wait a bit for pending orders
                    Thread.sleep(1000);
                    
                    // Initiate graceful shutdown
                    processor.initiateGracefulShutdown();
                    processor.awaitShutdown();
                    return;
                    
                case "2":
                    System.out.println("\n🚨 Simulating emergency shutdown (Ctrl+C)...");
                    keepGenerating.set(false);
                    
                    // Simulate abrupt termination by calling shutdown hook
                    Runtime.getRuntime().exit(0);
                    break;
                    
                case "3":
                    System.out.println("Continuing operation...");
                    break;
                    
                case "q":
                    keepGenerating.set(false);
                    orderGenerators.shutdownNow();
                    return;
                    
                default:
                    System.out.println("Unknown command");
            }
        }
    }
    
    private static Order createRandomOrder(int orderId, Random random) {
        String[] items = {"Laptop", "Mouse", "Keyboard", "Monitor", "iPhone", "AirPods"};
        List<String> orderItems = new ArrayList<>();
        
        int itemCount = 1 + random.nextInt(3);
        for (int i = 0; i < itemCount; i++) {
            orderItems.add(items[random.nextInt(items.length)]);
        }
        
        return new Order(orderId, 700 + random.nextInt(100), 
            orderItems, random.nextBoolean(), 500 + random.nextDouble() * 1000);
    }
}

// ShutdownTest.java - Unit tests for shutdown scenarios
import java.util.concurrent.*;
import java.util.*;

public class ShutdownTest {
    public static void main(String[] args) throws Exception {
        System.out.println("🧪 Testing Graceful Shutdown Scenarios\n");
        
        testNormalShutdown();
        testShutdownWithInFlightRequests();
        testShutdownUnderLoad();
        
        System.out.println("\n✅ All shutdown tests completed");
    }
    
    private static void testNormalShutdown() throws Exception {
        System.out.println("Test 1: Normal Shutdown");
        System.out.println("------------------------");
        
        GracefulShutdownProcessor processor = new GracefulShutdownProcessor();
        
        // Process some orders
        List<CompletableFuture<OrderResult>> futures = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Order order = new Order(8000 + i, 800, Arrays.asList("Laptop"), true, 999.99);
            futures.add(processor.processOrder(order));
        }
        
        // Wait for completion
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
        
        // Shutdown
        processor.initiateGracefulShutdown();
        processor.awaitShutdown();
        
        System.out.println("✅ Normal shutdown completed\n");
    }
    
    private static void testShutdownWithInFlightRequests() throws Exception {
        System.out.println("Test 2: Shutdown with In-Flight Requests");
        System.out.println("----------------------------------------");
        
        GracefulShutdownProcessor processor = new GracefulShutdownProcessor();
        
        // Start many orders but don't wait
        List<CompletableFuture<OrderResult>> futures = new ArrayList<>();
        for (int i = 0; i < 50; i++) {
            Order order = new Order(9000 + i, 900, Arrays.asList("iPhone", "AirPods"), false, 1299.99);
            futures.add(processor.processOrder(order));
        }
        
        // Immediately start shutdown
        Thread.sleep(100); // Let some orders start processing
        processor.initiateGracefulShutdown();
        processor.awaitShutdown();
        
        // Check how many completed
        long completed = futures.stream().filter(CompletableFuture::isDone).count();
        System.out.println("Completed: " + completed + " out of 50");
        System.out.println("✅ Shutdown with in-flight requests completed\n");
    }
    
    private static void testShutdownUnderLoad() throws Exception {
        System.out.println("Test 3: Shutdown Under Heavy Load");
        System.out.println("---------------------------------");
        
        GracefulShutdownProcessor processor = new GracefulShutdownProcessor();
        AtomicBoolean keepRunning = new AtomicBoolean(true);
        
        // Create load generator thread
        Thread loadGenerator = new Thread(() -> {
            int orderId = 10000;
            Random random = new Random();
            
            while (keepRunning.get()) {
                try {
                    Order order = new Order(orderId++, 1000, 
                        Arrays.asList("Monitor", "Keyboard"), true, 599.99);
                    processor.processOrder(order);
                    Thread.sleep(10); // High frequency
                } catch (InterruptedException e) {
                    break;
                }
            }
        });
        
        loadGenerator.start();
        
        // Let it run under load
        Thread.sleep(2000);
        
        // Stop load and shutdown
        keepRunning.set(false);
        loadGenerator.interrupt();
        
        processor.initiateGracefulShutdown();
        processor.awaitShutdown();
        
        System.out.println("✅ Shutdown under load completed\n");
    }
}

Key Concepts in Story 12:

1. Shutdown Hooks – JVM callbacks for cleanup during termination
2. In-Flight Request Tracking – Monitor and wait for active requests
3. Graceful Executor Shutdown – shutdown() vs shutdownNow()
4. State Persistence – Save critical state before shutdown
5. Timeout Mechanisms – Prevent indefinite waiting during shutdown
6. Resource Cleanup – Proper cleanup of connections, files, etc.
7. Shutdown Coordination – Multi-phase shutdown process

What this achieves:

• No data loss during shutdown
• Completes in-flight requests when possible
• Saves state for recovery
• Handles both graceful and abrupt shutdowns
• Provides detailed shutdown reporting
• Prevents resource leaks

Sample Output:

🚀 Starting Graceful Shutdown Demo

📊 Active requests: 15, Total processed: 127

📋 Initiating normal shutdown...

🔄 Initiating graceful shutdown...
1️⃣ Stopping new request acceptance
2️⃣ Cancelling scheduled tasks
3️⃣ Waiting for 15 in-flight requests...
✅ All in-flight requests completed
4️⃣ Shutting down executors
5️⃣ Saving final state
💾 State saved successfully
6️⃣ Finalizing persistence
7️⃣ Cleaning up critical resources
✅ Resources cleaned up

========== SHUTDOWN REPORT ==========
Total orders processed: 142
Orders completed during shutdown: 15
Orders rejected during shutdown: 3
Shutdown time: 2847 ms
Final inventory state:
  Laptop: 876
  Mouse: 1923
  Keyboard: 1411
  Monitor: 698
  iPhone: 945
  AirPods: 1156
====================================

Benefits:

• Data integrity maintained during shutdown
• Graceful handling of system termination
• Recovery capability through state persistence
• Clear visibility into shutdown process
• Prevents abrupt termination issues

This completes the implementation of all 12 stories in the epic! The system now has comprehensive multi-threading capabilities with proper shutdown handling.