finagle-mysql: Don't bother rolling back if the mysql session is alre…

…ady closed

Problem

Trying to send the rollback query to a closed session is wasteful and can
cause unnecessary log entries.

Solution

If the mysql session is closed skip the rollback and send the poisoned
request and close the service.

JIRA Issues: CSL-9170

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

CHANGELOG.rst

@@ -10,17 +10,21 @@ Unreleased
 Runtime Behavior Changes
 ~~~~~~~~~~~~~~~~~~~~~~~~
 
+* finagle-core: Per-method metrics on MethodBuilder are now created lazily, so if you have
+  methods that you don't use, the associated metrics won't be exported.  ``PHAB_ID=D400382``
+
+* finagle-mysql: The RollbackFactory no longer attempts to roll back if the underlying
+  session is closed since it is highly unlikely to succeed. It now simply poisons the
+  session and calls close. ``PHAB_ID=D408155``
+
 * finagle-netty4: Change the 'connection_requests' metric to debug verbosity.
   ``PHAB_ID=D391289``
 
 * finagle-thrift: Per-method metrics are now created lazily, so if you have methods on a Thrift
   service that you don't use, the associated metrics won't be exported.  ``PHAB_ID=D400382``
 
-* finagle-core: Per-method metrics on MethodBuilder are now created lazily, so if you have
-  methods that you don't use, the associated metrics won't be exported.  ``PHAB_ID=D400382``
-
- * finagle-zipkin-core: Tracing produces microsecond resolution timestamps in JDK9 or later. 
-   ``PHAB_ID=D400661``
+* finagle-zipkin-core: Tracing produces microsecond resolution timestamps in JDK9 or later. 
+  ``PHAB_ID=D400661``
 
 Breaking API Changes
 ~~~~~~~~~~~~~~~~~~~~

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/RollbackFactory.scala

@@ -40,7 +40,13 @@ final class RollbackFactory(client: ServiceFactory[Request, Result], statsReceiv
 
   private[this] def wrap(underlying: Service[Request, Result]): Service[Request, Result] =
     new ServiceProxy[Request, Result](underlying) {
+
       override def close(deadline: Time): Future[Unit] = {
+        if (self.status == Status.Closed) poisonAndClose(deadline)
+        else rollback(deadline)
+      }
+
+      private[this] def rollback(deadline: Time): Future[Unit] = {
         val elapsed = Stopwatch.start()
         self(RollbackQuery).transform { result =>
           rollbackLatencyStat.add(elapsed().inMillis)

finagle-mysql/src/test/scala/com/twitter/finagle/mysql/unit/RollbackFactoryTest.scala

@@ -2,7 +2,13 @@ package com.twitter.finagle.mysql
 
 import com.twitter.conversions.DurationOps._
 import com.twitter.finagle.stats.NullStatsReceiver
-import com.twitter.finagle.{ChannelClosedException, ClientConnection, Service, ServiceFactory}
+import com.twitter.finagle.{
+  ChannelClosedException,
+  ClientConnection,
+  Service,
+  ServiceFactory,
+  Status
+}
 import com.twitter.util.{Await, Awaitable, Future, Time}
 import org.scalatest.FunSuite
 
@@ -125,4 +131,33 @@ class RollbackFactoryTest extends FunSuite {
     assert(requests == Seq(QueryRequest("1"), PoisonConnectionRequest))
     assert(closeCalled)
   }
+
+  test("the rollback query is omitted if the underlying service already has status closed") {
+    var requests: Seq[Request] = Seq.empty
+    var closeCalled = false
+    val client: ServiceFactory[Request, Result] = new ServiceFactory[Request, Result] {
+      private[this] val svc: Service[Request, Result] = new Service[Request, Result] {
+        def apply(req: Request): Future[EOF] = {
+          requests = requests :+ req
+          Future.exception(new ChannelClosedException())
+        }
+
+        override def close(when: Time): Future[Unit] = {
+          closeCalled = true
+          Future.Done
+        }
+
+        override def status: Status = Status.Closed
+      }
+      def apply(c: ClientConnection): Future[Service[Request, Result]] = Future.value(svc)
+      def close(deadline: Time): Future[Unit] = svc.close(deadline)
+    }
+
+    val rollbackClient = new RollbackFactory(client, NullStatsReceiver)
+
+    await(rollbackClient().flatMap(_.close()))
+
+    assert(requests == Seq(PoisonConnectionRequest))
+    assert(closeCalled)
+  }
 }