fix : fix tests

Author: ovklm

src/main/java/org/alxkm/patterns/leaderfollower/LeaderFollowerPattern.java

@@ -8,6 +8,8 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.Condition;
 import java.util.function.Consumer;
+import java.util.List;
+import java.util.ArrayList;
 
 /**
  * Leader-Follower Pattern implementation.
@@ -38,6 +40,9 @@ public class LeaderFollowerPattern<T> {
     private final AtomicInteger processedEvents;
     private final AtomicInteger leaderPromotions;
 
+    // Thread management
+    private final List<WorkerThread> workerThreads;
+    
     /**
      * Creates a new Leader-Follower thread pool.
      * 
@@ -61,10 +66,11 @@ public LeaderFollowerPattern(int threadPoolSize, Consumer<T> eventProcessor) {
         this.leaderLock = new ReentrantLock();
         this.followerCondition = leaderLock.newCondition();
         this.currentLeader = null;
-        this.followerQueue = new LinkedBlockingQueue<>();
+        this.followerQueue = new LinkedBlockingQueue<>(threadPoolSize);
 
         this.processedEvents = new AtomicInteger(0);
         this.leaderPromotions = new AtomicInteger(0);
+        this.workerThreads = new ArrayList<>(threadPoolSize);
     }
 
     /**
@@ -76,6 +82,7 @@ private class WorkerThread extends Thread {
         public WorkerThread(int threadId) {
             this.threadId = threadId;
             setName("LeaderFollower-Worker-" + threadId);
+            setDaemon(true); // Make daemon thread to prevent JVM hanging
         }
 
         @Override
@@ -84,23 +91,37 @@ public void run() {
 
             try {
                 while (isRunning.get()) {
-                    if (becomeLeader()) {
-                        // I am the leader, wait for and process events
-                        processAsLeader();
-                    } else {
-                        // I am a follower, wait to be promoted
-                        waitAsFollower();
+                    try {
+                        if (becomeLeader()) {
+                            // I am the leader, wait for and process events
+                            processAsLeader();
+                        } else {
+                            // I am a follower, wait to be promoted
+                            waitAsFollower();
+                        }
+                    } catch (InterruptedException e) {
+                        // Exit if interrupted
+                        Thread.currentThread().interrupt();
+                        break;
                     }
 
                     // Check if we should stop
                     if (!isRunning.get()) {
                         break;
                     }
                 }
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
             } finally {
                 activeThreads.decrementAndGet();
+                // Remove self from follower queue on exit
+                leaderLock.lock();
+                try {
+                    followerQueue.remove(Thread.currentThread());
+                    if (currentLeader == Thread.currentThread()) {
+                        currentLeader = null;
+                    }
+                } finally {
+                    leaderLock.unlock();
+                }
             }
         }
 
@@ -115,8 +136,10 @@ private boolean becomeLeader() {
                     leaderPromotions.incrementAndGet();
                     return true;
                 } else {
-                    // Add self to follower queue
-                    followerQueue.offer(Thread.currentThread());
+                    // Add self to follower queue only if not already there
+                    if (!followerQueue.contains(Thread.currentThread())) {
+                        followerQueue.offer(Thread.currentThread());
+                    }
                     return false;
                 }
             } finally {
@@ -143,9 +166,6 @@ private void processAsLeader() throws InterruptedException {
 
                         // After processing, promote a follower to leader and step down
                         promoteFollowerToLeader();
-                        
-                        // Add self back to follower queue
-                        followerQueue.offer(Thread.currentThread());
                         break; // Exit leader role
                     }
                 }
@@ -219,9 +239,14 @@ private void stepDownAsLeader() {
      */
     public void start() {
         if (isRunning.compareAndSet(false, true)) {
+            // Clear any previous threads
+            workerThreads.clear();
+            
             // Start worker threads
             for (int i = 0; i < threadPoolSize; i++) {
-                new WorkerThread(i).start();
+                WorkerThread thread = new WorkerThread(i);
+                workerThreads.add(thread);
+                thread.start();
             }
 
             // Wait for threads to initialize with timeout
@@ -277,6 +302,11 @@ public void shutdown() {
             leaderLock.unlock();
         }
 
+        // Interrupt all threads
+        for (WorkerThread thread : workerThreads) {
+            thread.interrupt();
+        }
+        
         // Wait for all threads to finish
         long timeout = System.currentTimeMillis() + 5000; // 5 second timeout
         while (activeThreads.get() > 0 && System.currentTimeMillis() < timeout) {
@@ -306,6 +336,11 @@ public void shutdownNow() {
         } finally {
             leaderLock.unlock();
         }
+        
+        // Interrupt all threads immediately
+        for (WorkerThread thread : workerThreads) {
+            thread.interrupt();
+        }
     }
 
     /**