finagle-mux: Preserve RetryableNackFailure log level when reusing fla…

…gs from context

Problem

If RdispatchNack propagates flags, it should still have similar behavior to if flags aren't propagated.

Solution

Copy the RetryableNackFailure instead of constructing a wholly new Failure.

Result

Instead of using the default log level, we reuse the RetryableNackFailure log level.  Nothing else is changed.

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

finagle-mux/src/main/scala/com/twitter/finagle/mux/ReqRepFilter.scala

@@ -26,7 +26,7 @@ private[finagle] object ReqRepFilter {
 
     case Message.RdispatchNack(_, contexts) =>
       val exn = MuxFailure.fromContexts(contexts) match {
-        case Some(f) => Failure(Failure.RetryableNackFailure.why, f.finagleFlags)
+        case Some(f) => Failure.RetryableNackFailure.withFlags(f.finagleFlags)
         case None => Failure.RetryableNackFailure
       }
       Throw(exn)

finagle-mux/src/test/scala/com/twitter/finagle/mux/ReqRepFilterTest.scala

@@ -0,0 +1,21 @@
+package com.twitter.finagle.mux
+
+import com.twitter.finagle.Failure
+import com.twitter.finagle.mux.transport.{Message, MuxFailure}
+import com.twitter.logging.Level
+import org.scalatest.FunSuite
+
+class ReqRepFilterTest extends FunSuite {
+  test("Hydrate a mux failure from a message with encoded error codes, so that it preserves the" +
+    " flags and uses the right log level") {
+    val muxFailure = new MuxFailure(MuxFailure.NonRetryable | MuxFailure.Rejected)
+    val contexts = muxFailure.contexts
+    val msg = Message.RdispatchNack(1 /* tag */, contexts)
+    val actual = intercept[Failure] {
+      ReqRepFilter.reply(msg).get
+    }
+    assert(actual == Failure.RetryableNackFailure.withFlags(muxFailure.finagleFlags))
+    assert(actual.flags == muxFailure.finagleFlags)
+    assert(actual.logLevel == Level.DEBUG)
+  }
+}