Adding Task<>.recover(String, Class<> exceptionFilter, Function1<>) method to provide recovery with failure filter (#299)

Author: sleepytomcat

CHANGELOG.md

@@ -1,7 +1,8 @@
 v5.1.3
 ------
-* Fix for multiple javadoc warnings coming from Task.java 
-
+* Fix for multiple javadoc warnings coming from Task.java
+* Add recover() method with exception filter
+ 
 v5.1.2
 ------
 - Migrate the ParSeq release process from Bintray to JFrog Artifactory.

subprojects/parseq/src/main/java/com/linkedin/parseq/Task.java

@@ -637,11 +637,46 @@ default <R> Task<R> andThen(final Task<R> task) {
    */
   default Task<T> recover(final String desc, final Function1<Throwable, T> func) {
     ArgumentUtil.requireNotNull(func, "function");
+    return recover(desc, Throwable.class, func);
+  }
+
+  /**
+   * Equivalent to {@code recover("recover", func)}.
+   * @see #recover(String, Function1)
+   */
+  default Task<T> recover(final Function1<Throwable, T> func) {
+    return recover("recover: " + _taskDescriptor.getDescription(func.getClass().getName()), Throwable.class, func);
+  }
+
+  /**
+   * Creates a new task that will handle failure of this task.
+   *
+   * Early completion due to cancellation is not considered to be a failure and the recovery function is not called.
+   * If this task competed with failure, and failure type matches the exception class the provided, that failure recovery
+   * function is called.
+   *
+   * If this task completes successfully, then recovery function is not called.
+   *
+   * Note that the first call to <code>recover()</code> with matching exception class provided would result in a recovered Task.
+   * For example, observe the code below:
+   * <pre><code>
+   * Task&lt;String&gt; task = Task.failure(new RuntimeException())
+   *     .recover("try recovering RuntimeException", RuntimeException.class, e -&gt; "recovered") // results in recovery, as exception class matches
+   *     .recover("try recovering Throwable", Throwable.class, e -&gt; "no action"); // no action, as already recovered at previous step
+   * </code></pre>
+   * @param desc description of a recovery function, it will show up in a trace
+   * @param exceptionClass exception class, defines which types of failures would be attempted to recover with {@code func}
+   * @param func recovery function which can complete task with a value depending on failure of this task
+   * @return a new task which can recover from failure of this task
+   */
+  default <X extends Throwable> Task<T> recover(final String desc, final Class<X> exceptionClass, final Function1<? super X, ? extends T> func) {
+    ArgumentUtil.requireNotNull(func, "function");
+    ArgumentUtil.requireNotNull(exceptionClass, "exception class");
     return apply(desc, (src, dst) -> {
       if (src.isFailed()) {
-        if (!(Exceptions.isCancellation(src.getError()))) {
+        if (!(Exceptions.isCancellation(src.getError())) && exceptionClass.isInstance(src.getError())) {
           try {
-            dst.done(func.apply(src.getError()));
+            dst.done(func.apply(exceptionClass.cast(src.getError())));
           } catch (Throwable t) {
             dst.fail(t);
           }
@@ -655,13 +690,14 @@ default Task<T> recover(final String desc, final Function1<Throwable, T> func) {
   }
 
   /**
-   * Equivalent to {@code recover("recover", func)}.
-   * @see #recover(String, Function1)
+   * Equivalent to {@code recover("recover", exceptionClass, func)}.
+   * @see #recover(String, Class, Function1)
    */
-  default Task<T> recover(final Function1<Throwable, T> func) {
-    return recover("recover: " + _taskDescriptor.getDescription(func.getClass().getName()), func);
+  default <X extends Throwable> Task<T> recover(Class<X> exceptionClass, final Function1<? super X, ? extends T> func) {
+    String description = "recover: " + _taskDescriptor.getDescription(func.getClass().getName());
+    return recover(description, exceptionClass, func);
   }
-
+  
   /**
    * Creates a new task which applies a consumer to the exception this
    * task may fail with. It is used in situations where consumer needs

subprojects/parseq/src/test/java/com/linkedin/parseq/AbstractTaskTest.java

@@ -1,23 +1,19 @@
 package com.linkedin.parseq;
 
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertNull;
-import static org.testng.Assert.assertSame;
-import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.fail;
-
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
+import com.linkedin.parseq.trace.Trace;
 import org.testng.annotations.Test;
 
 import com.linkedin.parseq.function.Failure;
 import com.linkedin.parseq.function.Function1;
 import com.linkedin.parseq.function.Success;
 import com.linkedin.parseq.function.Try;
 
+import static org.testng.Assert.*;
+
 
 /**
  * @author Jaroslaw Odzga (jodzga@linkedin.com)
@@ -98,6 +94,56 @@ public void testRecover(int expectedNumberOfTasks) {
     assertEquals(countTasks(failure.getTrace()), expectedNumberOfTasks);
   }
 
+  public void testRecoverWithExceptionFilter(int expectedNumberOfTasks) {
+    Task<Integer> recoverd = getFailureTask()
+            .map("strlen", String::length)
+            .recover("", RuntimeException.class, e -> -1);
+    runAndWait("AbstractTaskTest.recoverd", recoverd);
+    assertEquals((int) recoverd.get(), -1);
+    assertEquals(countTasks(recoverd.getTrace()), expectedNumberOfTasks);
+
+    Task<Integer> notRecovered = getFailureTask()
+            .map("strlen", String::length)
+            .recover("", ArithmeticException.class, e -> -1);
+    runAndWaitException("AbstractTaskTest.testRecoverWithExceptionFilterFailure", notRecovered, RuntimeException.class);
+    assertTrue(notRecovered.isFailed());
+    assertEquals(countTasks(notRecovered.getTrace()), expectedNumberOfTasks);
+
+    Task<String> recoveredExceptionSubclassesForward = getFailureTask() // Failed with RuntimeError
+            .recover("", Throwable.class, e -> "expected-value") // expected to recover, as RuntimeError is subclass of Throwable
+            .recover("", RuntimeException.class, e -> "unexpected-value"); // no action, as already recovered at previous step
+    runAndWait("AbstractTaskTest.recoveredExceptionSubclassesForward", recoveredExceptionSubclassesForward);
+    assertEquals(recoveredExceptionSubclassesForward.get(), "expected-value");
+    assertEquals(countTasks(recoveredExceptionSubclassesForward.getTrace()), expectedNumberOfTasks);
+
+    Task<String> recoveredExceptionSubclassesBackwards = getFailureTask() // Failed with RuntimeError
+            .recover("", RuntimeException.class, e -> "expected-value") // expected to recover, as exception class matches
+            .recover("", Throwable.class, e -> "unexpected-value"); // no action, as already recovered at previous step
+    runAndWait("AbstractTaskTest.recoveredExceptionSubclassesBackwards", recoveredExceptionSubclassesBackwards);
+    assertEquals(recoveredExceptionSubclassesBackwards.get(), "expected-value");  // recovered with expected value
+    assertEquals(countTasks(recoveredExceptionSubclassesBackwards.getTrace()), expectedNumberOfTasks);
+
+    Task<String> recoveredUnrelatedExceptionFirst = getFailureTask() // Failed with RuntimeError
+            .recover("", Error.class, e -> "unexpected-value") // no action, as the failed state is not of type Error
+            .recover("", Throwable.class, e -> "expected-value"); // expected to recover, as exception class matches
+    runAndWait("AbstractTaskTest.recoveredUnrelatedExceptionFirst", recoveredUnrelatedExceptionFirst);
+    assertEquals(recoveredUnrelatedExceptionFirst.get(), "expected-value");  // recovered with expected value
+    assertEquals(countTasks(recoveredUnrelatedExceptionFirst.getTrace()), expectedNumberOfTasks);
+
+    Task<String> recoveredRecoveryThrows = getFailureTask() // Failed with RuntimeError
+            .recover("", RuntimeException.class, e -> {throw new IllegalArgumentException();}) // exception class matches, but recovery throws another exception
+            .recover("", IllegalArgumentException.class, e -> "expected-value"); // expected to recover, as exception class matches (see above)
+    runAndWait("AbstractTaskTest.recoveredRecoveryThrows", recoveredRecoveryThrows);
+    assertEquals(recoveredRecoveryThrows.get(), "expected-value");  // recovered with expected value
+    assertEquals(countTasks(recoveredRecoveryThrows.getTrace()), expectedNumberOfTasks);
+
+    Task<String> failedAfterRecoveryThrows = getFailureTask() // Failed with RuntimeError
+            .recover("", RuntimeException.class, e -> {throw new IllegalArgumentException();}); // throws another exception while recovering
+    IllegalArgumentException ex = runAndWaitException(failedAfterRecoveryThrows, IllegalArgumentException.class);
+    assertNotNull(ex);
+    assertEquals(countTasks(recoveredRecoveryThrows.getTrace()), expectedNumberOfTasks);
+  }
+
   public void testCancelledRecover(int expectedNumberOfTasks) {
     Task<Integer> cancelled = getCancelledTask().map("strlen", String::length).recover(e -> -1);
     try {

subprojects/parseq/src/test/java/com/linkedin/parseq/TestFusionTask.java

@@ -65,6 +65,11 @@ public void testRecover() {
     testRecover(4);
   }
 
+  @Test
+  public void testRecoverWithExceptionFilter() {
+    testRecoverWithExceptionFilter(4);
+  }
+
   @Test
   public void testCancelledRecover() {
     testCancelledRecover(4);

subprojects/parseq/src/test/java/com/linkedin/parseq/TestTask.java

@@ -54,6 +54,11 @@ public void testRecover() {
     testRecover(5);
   }
 
+  @Test
+  public void testRecoverWithExceptionFilter() {
+    testRecoverWithExceptionFilter(5);
+  }
+
   @Test
   public void testCancelledRecover() {
     testCancelledRecover(5);