More tests for Deduplication Executor

@@ -56,14 +56,14 @@ public class EventDeduplicationExecutor<P> {
                             log.info("[{}] Executing task: {}", name, params);
                             log.info("[{}] Executing task: {}", name, params);
                             function.accept(params);
                             function.accept(params);
                         } catch (Throwable e) {
                         } catch (Throwable e) {
-                            log.warn("Failed to process task with parameters: {}", params, e);
+                            log.warn("[{}] Failed to process task with parameters: {}", name, params, e);
                             throw e;
                             throw e;
                         } finally {
                         } finally {
                             unlockAndProcessIfAny();
                             unlockAndProcessIfAny();
                         }
                         }
                     });
                     });
                 } catch (Throwable e) {
                 } catch (Throwable e) {
-                    log.warn("Failed to submit task with parameters: {}", params, e);
+                    log.warn("[{}] Failed to submit task with parameters: {}", name, params, e);
                     unlockAndProcessIfAny();
                     unlockAndProcessIfAny();
                     throw e;
                     throw e;
                 }
                 }

@@ -41,15 +41,24 @@ public class EventDeduplicationExecutorTest {
         periodicFlow(MoreExecutors.newDirectExecutorService());
         periodicFlow(MoreExecutors.newDirectExecutorService());
     }
     }
 
 
+    @Test
+    public void testExceptionFlowSameThread() throws InterruptedException {
+        exceptionFlow(MoreExecutors.newDirectExecutorService());
+    }
 
 
     @Test
     @Test
     public void testSimpleFlowSingleThread() throws InterruptedException {
     public void testSimpleFlowSingleThread() throws InterruptedException {
-        simpleFlow(Executors.newFixedThreadPool(1));
+        simpleFlow(Executors.newSingleThreadExecutor());
     }
     }
 
 
     @Test
     @Test
     public void testPeriodicFlowSingleThread() throws InterruptedException {
     public void testPeriodicFlowSingleThread() throws InterruptedException {
-        periodicFlow(Executors.newFixedThreadPool(1));
+        periodicFlow(Executors.newSingleThreadExecutor());
+    }
+
+    @Test
+    public void testExceptionFlowSingleThread() throws InterruptedException {
+        exceptionFlow(Executors.newSingleThreadExecutor());
     }
     }
 
 
     @Test
     @Test
@@ -62,6 +71,11 @@ public class EventDeduplicationExecutorTest {
         periodicFlow(Executors.newFixedThreadPool(3));
         periodicFlow(Executors.newFixedThreadPool(3));
     }
     }
 
 
+    @Test
+    public void testExceptionFlowMultiThread() throws InterruptedException {
+        exceptionFlow(Executors.newFixedThreadPool(3));
+    }
+
     private void simpleFlow(ExecutorService executorService) throws InterruptedException {
     private void simpleFlow(ExecutorService executorService) throws InterruptedException {
         try {
         try {
             Consumer<String> function = Mockito.spy(StringConsumer.class);
             Consumer<String> function = Mockito.spy(StringConsumer.class);
@@ -105,6 +119,28 @@ public class EventDeduplicationExecutorTest {
         }
         }
     }
     }
 
 
+    private void exceptionFlow(ExecutorService executorService) throws InterruptedException {
+        try {
+            Consumer<String> function = Mockito.spy(StringConsumer.class);
+            EventDeduplicationExecutor<String> executor = new EventDeduplicationExecutor<>(EventDeduplicationExecutorTest.class.getSimpleName(), executorService, function);
+
+            String params1 = "params1";
+            String params2 = "params2";
+            String params3 = "params3";
+
+            Mockito.doThrow(new RuntimeException()).when(function).accept("params1");
+            executor.submit(params1);
+            executor.submit(params2);
+            Thread.sleep(500);
+            executor.submit(params3);
+            Thread.sleep(500);
+            Mockito.verify(function).accept(params2);
+            Mockito.verify(function).accept(params3);
+        } finally {
+            executorService.shutdownNow();
+        }
+    }
+
     public static class StringConsumer implements Consumer<String> {
     public static class StringConsumer implements Consumer<String> {
         @Override
         @Override
         public void accept(String s) {
         public void accept(String s) {