Revert "Revert "finagle/finagle-core: Fix bug that causes backup requests to be disabled when configured via stack param in MethodBuilder""

This reverts commit 6b36c8bc38dd53e554f255a78b958c42cd032f3e.

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

Author: noctella

CHANGELOG.rst

@@ -40,6 +40,9 @@ Bug Fixes
 * finagle-memcached: Fixed support for running memcached tests with external memcached. Added README with
   instructions under finagle/finagle-memcached. ``PHAB_ID=D1120240``
 * finagle-core: Fixed a bug in `ExponentialJitteredBackoff` where `rng` can be 0. ``PHAB_ID=D1178528``
+* finagle-core: When `c.t.f.client.BackupRequestFilter` is configured via stack params,
+  use this configuration when using MethodBuilder if `idempotent`/`nonIdempotent` have
+  not been set. ``PHAB_ID=D1189436``
 
 
 Breaking API Changes

finagle-core/src/main/scala/com/twitter/finagle/client/MethodBuilder.scala

@@ -123,7 +123,7 @@ object MethodBuilder {
     retry: MethodBuilderRetry.Config,
     timeout: MethodBuilderTimeout.Config,
     filter: Filter.TypeAgnostic = Filter.typeAgnosticIdentity,
-    backup: BackupRequestFilter.Param = BackupRequestFilter.Param.Disabled)
+    backup: Option[BackupRequestFilter.Param] = None)
 
   /** Used by the `ClientRegistry` */
   private[client] val RegistryKey = "methods"
@@ -396,7 +396,7 @@ final class MethodBuilder[Req, Rep] private[finagle] (
       dest,
       stack,
       params,
-      config.copy(backup = brfParam)
+      config.copy(backup = Some(brfParam))
     )
 
     base.withRetry.forClassifier(idempotentedConfigClassifier)
@@ -436,7 +436,7 @@ final class MethodBuilder[Req, Rep] private[finagle] (
       dest,
       stack,
       params,
-      config.copy(backup = BackupRequestFilter.Param.Disabled)
+      config.copy(backup = Some(BackupRequestFilter.Param.Disabled))
     ).withRetry.forClassifier(nonidempotentedConfigClassifier)
   }
 
@@ -677,9 +677,16 @@ final class MethodBuilder[Req, Rep] private[finagle] (
 
     val underlying = methodPool.get
 
-    val backupRequestParams = params +
-      config.backup +
-      param.ResponseClassifier(config.retry.responseClassifier)
+    // If the user has not configured the backup requests via MethodBuilder (via `idempotent`,
+    // of `nonIdempotent`), use the configuration from the params (Disabled if none available).
+    val backupRequestParams = config.backup match {
+      case Some(backupRequestParam) =>
+        params +
+          backupRequestParam +
+          param.ResponseClassifier(config.retry.responseClassifier)
+      case None =>
+        params + param.ResponseClassifier(config.retry.responseClassifier)
+    }
 
     // register BackupRequestFilter under the same prefixes as other method entries
     val prefixes = Seq(registryEntry().addr) ++ registryKeyPrefix(name)

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

@@ -658,16 +658,19 @@ class MethodBuilderTest
 
     val stackClient = TestStackClient(stack, params)
     val initialMethodBuilder = MethodBuilder.from("with backups", stackClient)
+    assert(initialMethodBuilder.config.backup == None)
+
     val methodBuilder =
-      initialMethodBuilder.withConfig(initialMethodBuilder.config.copy(backup = configuredBrfParam))
+      initialMethodBuilder.withConfig(
+        initialMethodBuilder.config.copy(backup = Some(configuredBrfParam)))
 
     // Ensure BRF is configured before calling `nonIdempotent`
-    assert(methodBuilder.config.backup == configuredBrfParam)
+    assert(methodBuilder.config.backup == Some(configuredBrfParam))
 
     val nonIdempotentClient = methodBuilder.nonIdempotent
 
     // Ensure BRF is disabled after calling `nonIdempotent`
-    assert(nonIdempotentClient.config.backup == BackupRequestFilter.Disabled)
+    assert(nonIdempotentClient.config.backup == Some(BackupRequestFilter.Disabled))
   }
 
   test("nonIdempotent client keeps existing ResponseClassifier in params ") {
@@ -796,8 +799,9 @@ class MethodBuilderTest
       .idempotent(1.percent, sendInterrupts = true, classifier)
 
     mb.config.backup match {
-      case BackupRequestFilter.Param
-            .Configured(maxExtraLoadTunable, sendInterrupts, minSendBackupAfterMs) =>
+      case Some(
+            BackupRequestFilter.Param
+              .Configured(maxExtraLoadTunable, sendInterrupts, minSendBackupAfterMs)) =>
         assert(
           maxExtraLoadTunable().get == 1.percent && sendInterrupts && minSendBackupAfterMs == 1)
       case _ => fail("BackupRequestFilter not configured")
@@ -862,8 +866,9 @@ class MethodBuilderTest
       .idempotent(tunable, sendInterrupts = true, ResponseClassifier.Default)
 
     assert(
-      mb.config.backup == BackupRequestFilter
-        .Configured(tunable, sendInterrupts = true)
+      mb.config.backup == Some(
+        BackupRequestFilter
+          .Configured(tunable, sendInterrupts = true))
     )
   }
 
@@ -880,12 +885,13 @@ class MethodBuilderTest
       .idempotent(tunable, sendInterrupts = true, ResponseClassifier.Default)
 
     assert(
-      mb.config.backup == BackupRequestFilter
-        .Configured(tunable, sendInterrupts = true)
+      mb.config.backup == Some(
+        BackupRequestFilter
+          .Configured(tunable, sendInterrupts = true))
     )
 
     val nonIdempotentMB = mb.nonIdempotent
-    assert(nonIdempotentMB.config.backup == BackupRequestFilter.Disabled)
+    assert(nonIdempotentMB.config.backup == Some(BackupRequestFilter.Disabled))
   }
 
   test("idempotent combines existing classifier with new one") {
@@ -1122,6 +1128,100 @@ class MethodBuilderTest
     assert(client.params[Retries.Budget].retryBudget eq retryBudget)
   }
 
+  test(
+    "BackupRequestFilter is configured with passed-in stack params when not configured in MethodBuilder") {
+    val stats = new InMemoryStatsReceiver()
+    val timer = new MockTimer()
+    val params =
+      Stack.Params.empty +
+        param.Stats(stats) +
+        param.Timer(timer) +
+        BackupRequestFilter.Configured(maxExtraLoad = 0.01, sendInterrupts = true)
+
+    val svc: Service[Int, Int] = Service.mk { i =>
+      Future.value(i)
+    }
+
+    val stack = Stack.leaf(Stack.Role("test"), ServiceFactory.const(svc))
+
+    val stackClient = TestStackClient(stack, Stack.Params.empty).withParams(params)
+    val mb = MethodBuilder.from("mb", stackClient)
+
+    Time.withCurrentTimeFrozen { tc =>
+      val client = mb.newService("a_client")
+      awaitResult(client(1))
+
+      tc.advance(10.seconds)
+      timer.tick()
+      assert(stats.stats.contains(Seq("mb", "a_client", "backups", "send_backup_after_ms")))
+    }
+  }
+
+  test(
+    "BackupRequestFilter is configured with MethodBuilder idempotent configuration when also configured via stack params") {
+    val stats = new InMemoryStatsReceiver()
+    val timer = new MockTimer()
+    val params =
+      Stack.Params.empty +
+        param.Stats(stats) +
+        param.Timer(timer) +
+        BackupRequestFilter.Disabled
+
+    val svc: Service[Int, Int] = Service.mk { i =>
+      Future.value(i)
+    }
+
+    val stack = Stack.leaf(Stack.Role("test"), ServiceFactory.const(svc))
+
+    val stackClient = TestStackClient(stack, Stack.Params.empty).withParams(params)
+    val classifier: ResponseClassifier = ResponseClassifier.named("foo") {
+      case ReqRep(_, Throw(_: IndividualRequestTimeoutException)) =>
+        ResponseClass.RetryableFailure
+    }
+
+    // MB config should take precedence
+    val mb = MethodBuilder.from("mb", stackClient).idempotent(0.05, true, classifier)
+
+    Time.withCurrentTimeFrozen { tc =>
+      val client = mb.newService("a_client")
+      awaitResult(client(1))
+
+      tc.advance(10.seconds)
+      timer.tick()
+      assert(stats.stats.contains(Seq("mb", "a_client", "backups", "send_backup_after_ms")))
+    }
+  }
+
+  test(
+    "BackupRequestFilter is configured with MethodBuilder nonIdempotent configuration when also configured via stack params") {
+    val stats = new InMemoryStatsReceiver()
+    val timer = new MockTimer()
+    val params =
+      Stack.Params.empty +
+        param.Stats(stats) +
+        param.Timer(timer) +
+        BackupRequestFilter.Configured(maxExtraLoad = 0.01, sendInterrupts = true)
+
+    val svc: Service[Int, Int] = Service.mk { i =>
+      Future.value(i)
+    }
+
+    val stack = Stack.leaf(Stack.Role("test"), ServiceFactory.const(svc))
+
+    val stackClient = TestStackClient(stack, Stack.Params.empty).withParams(params)
+    // MB config should take precedence
+    val mb = MethodBuilder.from("mb", stackClient).nonIdempotent
+
+    Time.withCurrentTimeFrozen { tc =>
+      val client = mb.newService("a_client")
+      awaitResult(client(1))
+
+      tc.advance(10.seconds)
+      timer.tick()
+      assert(!stats.stats.contains(Seq("mb", "a_client", "backups", "send_backup_after_ms")))
+    }
+  }
+
   test("shares RetryBudget between methods") {
     val params = StackClient.defaultParams