finagle-core: FailureFlags.asNonRetryable no longer modifies non-Fail…

…ureFlags

Problem

Wrapping non-FailureFlags exceptions can lead to surprising behavior
for users.

Solution

Leave them alone.

JIRA Issues: CSL-6847

Differential Revision: https://phabricator.twitter.biz/D216281

CHANGELOG.rst

@@ -7,6 +7,14 @@ Note that ``PHAB_ID=#`` and ``RB_ID=#`` correspond to associated messages in com
 Unreleased
 ----------
 
+Runtime Behavior Changes
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+* finagle-core: When Finagle would exhaust a retry budget with an exception that was
+  not a `FailureFlags`, previously it would wrap that exception with a non-retryable
+  failure. This lead to surprising behavior for users. Those exceptions will no longer
+  be wrapped. ``PHAB_ID=D216281``
+
 18.9.0
 -------
 

finagle-core/src/main/scala/com/twitter/finagle/FailureFlags.scala

@@ -102,13 +102,12 @@ object FailureFlags {
   }
 
   /**
-   * A function for transforming unsuccessful responses into ones that are
-   * flagged as NonRetryable
+   * A function for transforming unsuccessful [[FailureFlags]] responses
+   * into ones that are flagged as [[NonRetryable]].
    */
   private[finagle] def asNonRetryable[Rep](t: Try[Rep]): Future[Rep] = {
     t match {
       case Throw(f: FailureFlags[_]) => Future.exception(f.asNonRetryable)
-      case Throw(exn) => Future.exception(Failure(exn, FailureFlags.NonRetryable))
       case _ => Future.const(t)
     }
   }
@@ -207,8 +206,8 @@ trait FailureFlags[T <: FailureFlags[T]] extends Exception { this: T =>
  * For Java users wanting to implement exceptions that are [[FailureFlags]].
  */
 abstract class AbstractFailureFlags[T <: AbstractFailureFlags[T]]
-  extends Exception
-  with FailureFlags[T] { this: T =>
+    extends Exception
+    with FailureFlags[T] { this: T =>
 
   // Java-friendly forwarders
   // See https://issues.scala-lang.org/browse/SI-8905

finagle-core/src/test/scala/com/twitter/finagle/FailureFlagsTest.scala

@@ -1,18 +1,23 @@
 package com.twitter.finagle
 
+import com.twitter.conversions.time._
+import com.twitter.util.{Await, Future, Throw}
 import org.scalacheck.Gen
 import org.scalatest.FunSuite
 import org.scalatest.prop.GeneratorDrivenPropertyChecks
 
 class FailureFlagsTest extends FunSuite with GeneratorDrivenPropertyChecks {
   import FailureFlags._
 
+  private def await[T](f: Future[T]): T =
+    Await.result(f, 5.seconds)
+
   case class FlagCheck(flags: Long) extends FailureFlags[FlagCheck] {
     protected def copyWithFlags(f: Long): FlagCheck = FlagCheck(f)
   }
 
   private val flag = Gen.oneOf(
-    0L,
+    FailureFlags.Empty,
     FailureFlags.Retryable,
     FailureFlags.Interrupted,
     FailureFlags.Wrapped,
@@ -69,4 +74,30 @@ class FailureFlagsTest extends FunSuite with GeneratorDrivenPropertyChecks {
     assert(copied.getCause == initial.getCause)
     assert(copied.getSuppressed.toSeq == initial.getSuppressed.toSeq)
   }
+
+  test("FailureFlags.asNonRetryable removes retryable flag from FailureFlags exceptions") {
+    val t = Throw(Failure.RetryableNackFailure)
+    val asNonRetryable = intercept[FailureFlags[_]] {
+      await(FailureFlags.asNonRetryable(t))
+    }
+    assert(!asNonRetryable.isFlagged(FailureFlags.Retryable))
+  }
+
+  test("FailureFlags.asNonRetryable on non-retryable FailureFlags exceptions") {
+    val t = Throw(Failure.NonRetryableNackFailure)
+    val asNonRetryable = intercept[FailureFlags[_]] {
+      await(FailureFlags.asNonRetryable(t))
+    }
+    assert(!asNonRetryable.isFlagged(FailureFlags.Retryable))
+  }
+
+  test("FailureFlags.asNonRetryable does not modify non-FailureFlags exceptions") {
+    val ex = new RuntimeException("not a FailureFlags")
+    val t = Throw(ex)
+    val notWrapped = intercept[RuntimeException] {
+      await(FailureFlags.asNonRetryable(t))
+    }
+    assert(ex == notWrapped)
+  }
+
 }

finagle-core/src/test/scala/com/twitter/finagle/client/StackClientTest.scala

@@ -391,7 +391,7 @@ abstract class AbstractStackClientTest
   })
 
   test("service acquisition requeues will close Status.Closed sessions") {
-    val ctx = new RequeueCtx { }
+    val ctx = new RequeueCtx {}
     ctx._svcFacStatus = Status.Open
     ctx._sessionStatus = Status.Closed
     ctx.closeSideEffect = () => ctx._sessionStatus = Status.Open
@@ -408,15 +408,15 @@ abstract class AbstractStackClientTest
     class CountFactory extends ServiceFactory[Unit, Unit] {
       var count = 0
 
-      val service = new Service[Unit, Unit] {
+      val service: Service[Unit, Unit] = new Service[Unit, Unit] {
         def apply(request: Unit): Future[Unit] = {
           count = count + 1
           Future.exception(WriteException(null))
         }
       }
 
-      def apply(conn: ClientConnection) = Future.value(service)
-      def close(deadline: Time) = Future.Done
+      def apply(conn: ClientConnection): Future[Service[Unit, Unit]] = Future.value(service)
+      def close(deadline: Time): Future[Unit] = Future.Done
     }
 
     val fac1 = new CountFactory
@@ -449,7 +449,7 @@ abstract class AbstractStackClientTest
       .replace(
         LoadBalancerFactory.role,
         new Stack.Module1[LoadBalancerFactory.Dest, ServiceFactory[Unit, Unit]] {
-          val role = new Stack.Role("role")
+          val role = Stack.Role("role")
           val description = "description"
           def make(dest: LoadBalancerFactory.Dest, next: ServiceFactory[Unit, Unit]) = {
             val LoadBalancerFactory.Dest(va) = dest
@@ -474,7 +474,7 @@ abstract class AbstractStackClientTest
         )
       )
 
-    intercept[Failure] {
+    intercept[ChannelWriteException] {
       await(service(()))
     }
 
@@ -545,7 +545,7 @@ abstract class AbstractStackClientTest
         StringServer.server.serve(new InetSocketAddress(InetAddress.getLoopbackAddress, 0), echoSvc)
       val ia = server.boundAddress.asInstanceOf[InetSocketAddress]
 
-      val client = new LocalCheckingStringClient(key)
+      val client = LocalCheckingStringClient(key)
         .newService(Name.bound(Address(ia)), "a-label")
 
       val result = await(client("abc"))
@@ -703,14 +703,14 @@ abstract class AbstractStackClientTest
     // We could insert using [[ExceptionSourceFilter]] as the relative insertion point, but we use
     // another module instead so that if the [[ExceptionSourceFilter]] were moved earlier in the
     // stack, this test would fail.
-    val svc = baseClient.withStack(baseClient.stack.insertBefore(
-      ClearContextValueFilter.role, throwsModule))
+    val svc = baseClient
+      .withStack(baseClient.stack.insertBefore(ClearContextValueFilter.role, throwsModule))
       .newService(Name.bound(Address(boundAddress)), label)
 
     val failure = intercept[Failure] {
       await(svc("hello"))
     }
 
-   assert(failure.toString == "Failure(boom!, flags=0x00) with Service -> stringClient")
+    assert(failure.toString == "Failure(boom!, flags=0x00) with Service -> stringClient")
   }
 }

finagle-core/src/test/scala/com/twitter/finagle/service/RetriesTest.scala

@@ -10,7 +10,11 @@ import org.scalatest.FunSuite
 
 class RetriesTest extends FunSuite {
 
-  private[this] class MyRetryEx extends Exception
+  private[this] class MyRetryEx(val flags: Long = FailureFlags.Empty)
+      extends Exception
+      with FailureFlags[MyRetryEx] {
+    protected def copyWithFlags(flags: Long): MyRetryEx = new MyRetryEx(flags)
+  }
   private[this] class AnotherEx extends Exception
 
   private[this] val retryFn: PartialFunction[Try[Nothing], Boolean] = {
@@ -214,7 +218,7 @@ class RetriesTest extends FunSuite {
     val svc: Service[Exception, Int] =
       Await.result(svcFactory(), 5.seconds)
 
-    intercept[Failure] {
+    intercept[MyRetryEx] {
       Await.result(svc(new MyRetryEx()), 5.seconds)
     }
 
@@ -308,7 +312,7 @@ class RetriesTest extends FunSuite {
     val numReqs = 100
     Time.withCurrentTimeFrozen { _ =>
       0.until(numReqs).foreach { _ =>
-        intercept[Failure] {
+        intercept[MyRetryEx] {
           Await.result(svc(new MyRetryEx()), 5.seconds)
         }
       }
@@ -426,12 +430,13 @@ class RetriesTest extends FunSuite {
 
   test("hashCode and equals for Retries.Budget only checks RetryBudget, not Backoff stream") {
     val budget = Retries.Budget(RetryBudget.Empty, Backoff.const(1.second))
-    val budgetWithDifferentBackoffStream = Retries.Budget(RetryBudget.Empty, Backoff.const(2.second))
+    val budgetWithDifferentBackoffStream =
+      Retries.Budget(RetryBudget.Empty, Backoff.const(2.second))
     val differentBudget = Retries.Budget(RetryBudget.Infinite, Backoff.const(3.second))
 
     assert(budget.equals(budget))
     assert(budget.equals(budgetWithDifferentBackoffStream))
-    assert(! budget.equals(differentBudget))
+    assert(!budget.equals(differentBudget))
 
     assert(budget.## == budget.##)
     assert(budget.## == budgetWithDifferentBackoffStream.##)