finagle-thrift: Expose withExecutionOffloaded with type signatures

Problem
-------

Client's and server's methods
`c.t.f.p.CommonParams.withExecutionOffloaded(pool: FuturePool): A` and
`c.t.f.p.CommonParams.withExecutionOffloaded(executor: ExecutorService): A`
are not accessible from java due to missing type information.
See https://issues.scala-lang.org/browse/SI-8905

Solution
--------

Override methods and call super to forward the parameters.

Result
------

`withExecutionOffloaded` will be accessible from java code.

JIRA Issues: MOBDR-6124

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

finagle-http/src/main/scala/com/twitter/finagle/Http.scala

@@ -17,8 +17,9 @@ import com.twitter.finagle.stats.{ExceptionStatsHandler, StatsReceiver}
 import com.twitter.finagle.toggle.Toggle
 import com.twitter.finagle.tracing._
 import com.twitter.finagle.transport.{Transport, TransportContext}
-import com.twitter.util.{Duration, Future, Monitor, StorageUnit}
+import com.twitter.util.{Duration, Future, FuturePool, Monitor, StorageUnit}
 import java.net.SocketAddress
+import java.util.concurrent.ExecutorService
 
 /**
  * A rich HTTP/1.1 client with a *very* basic URL fetcher. (It does not handle
@@ -327,6 +328,10 @@ object Http extends Client[Request, Response] with HttpRichClient with Server[Re
       fn: Stack[ServiceFactory[Request, Response]] => Stack[ServiceFactory[Request, Response]]
     ): Client =
       super.withStack(fn)
+    override def withExecutionOffloaded(executor: ExecutorService): Client =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Client =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Client = super.configured(psp)
     override def configuredParams(newParams: Stack.Params): Client =
       super.configuredParams(newParams)
@@ -562,7 +567,10 @@ object Http extends Client[Request, Response] with HttpRichClient with Server[Re
       fn: Stack[ServiceFactory[Request, Response]] => Stack[ServiceFactory[Request, Response]]
     ): Server =
       super.withStack(fn)
-
+    override def withExecutionOffloaded(executor: ExecutorService): Server =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Server =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Server = super.configured(psp)
     override def configuredParams(newParams: Stack.Params): Server =
       super.configuredParams(newParams)

finagle-memcached/src/main/scala/com/twitter/finagle/Memcached.scala

@@ -53,6 +53,7 @@ import com.twitter.io.Buf
 import com.twitter.util._
 import com.twitter.util.registry.GlobalRegistry
 import java.net.SocketAddress
+import java.util.concurrent.ExecutorService
 
 /**
  * Factory methods to build a finagle-memcached client.
@@ -352,6 +353,10 @@ object Memcached extends finagle.Client[Command, Response] with finagle.Server[C
       fn: Stack[ServiceFactory[Command, Response]] => Stack[ServiceFactory[Command, Response]]
     ): Client =
       super.withStack(fn)
+    override def withExecutionOffloaded(executor: ExecutorService): Client =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Client =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Client = super.configured(psp)
     override def filtered(filter: Filter[Command, Response, Command, Response]): Client =
       super.filtered(filter)
@@ -424,7 +429,10 @@ object Memcached extends finagle.Client[Command, Response] with finagle.Server[C
       fn: Stack[ServiceFactory[Command, Response]] => Stack[ServiceFactory[Command, Response]]
     ): Server =
       super.withStack(fn)
-
+    override def withExecutionOffloaded(executor: ExecutorService): Server =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Server =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Server = super.configured(psp)
   }
 

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

@@ -5,8 +5,8 @@ import com.twitter.finagle.mysql._
 import com.twitter.finagle.mysql.param._
 import com.twitter.finagle.mysql.transport.Packet
 import com.twitter.finagle.param.{
-  ExceptionStatsHandler => _,
   Monitor => ParamMonitor,
+  ExceptionStatsHandler => _,
   ResponseClassifier => _,
   Tracer => _,
   _
@@ -15,8 +15,9 @@ import com.twitter.finagle.service.{ResponseClassifier, RetryBudget}
 import com.twitter.finagle.stats.{ExceptionStatsHandler, NullStatsReceiver, StatsReceiver}
 import com.twitter.finagle.tracing.Tracer
 import com.twitter.finagle.transport.{Transport, TransportContext}
-import com.twitter.util.{Duration, Monitor}
+import com.twitter.util.{Duration, FuturePool, Monitor}
 import java.net.SocketAddress
+import java.util.concurrent.ExecutorService
 
 /**
  * Supplements a [[com.twitter.finagle.Client]] with convenient
@@ -253,6 +254,10 @@ object Mysql extends com.twitter.finagle.Client[Request, Result] with MysqlRichC
       fn: Stack[ServiceFactory[Request, Result]] => Stack[ServiceFactory[Request, Result]]
     ): Client =
       super.withStack(fn)
+    override def withExecutionOffloaded(executor: ExecutorService): Client =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Client =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Client = super.configured(psp)
     override def filtered(filter: Filter[Request, Result, Request, Result]): Client =
       super.filtered(filter)

finagle-redis/src/main/scala/com/twitter/finagle/Redis.scala

@@ -14,16 +14,16 @@ import com.twitter.finagle.param.{
 }
 import com.twitter.finagle.redis.RedisPartitioningService
 import com.twitter.finagle.redis.exp.{ConnectionInitCommand, RedisPool}
+import com.twitter.finagle.redis.param.{Database, Password}
 import com.twitter.finagle.redis.protocol.{Command, Reply, StageTransport}
 import com.twitter.finagle.service.{ResponseClassifier, RetryBudget}
 import com.twitter.finagle.stats.{ExceptionStatsHandler, StatsReceiver}
 import com.twitter.finagle.tracing.Tracer
 import com.twitter.finagle.transport.{Transport, TransportContext}
 import com.twitter.io.Buf
-import com.twitter.util.{Duration, Monitor}
+import com.twitter.util.{Duration, FuturePool, Monitor}
 import java.net.SocketAddress
-
-import com.twitter.finagle.redis.param.{Database, Password}
+import java.util.concurrent.ExecutorService
 
 trait RedisRichClient { self: Client[Command, Reply] =>
 
@@ -152,6 +152,10 @@ object Redis extends Client[Command, Reply] with RedisRichClient {
       fn: Stack[ServiceFactory[Command, Reply]] => Stack[ServiceFactory[Command, Reply]]
     ): Client =
       super.withStack(fn)
+    override def withExecutionOffloaded(executor: ExecutorService): Client =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Client =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Client = super.configured(psp)
     override def filtered(filter: Filter[Command, Reply, Command, Reply]): Client =
       super.filtered(filter)

finagle-thrift/src/main/scala/com/twitter/finagle/Thrift.scala

@@ -27,8 +27,9 @@ import com.twitter.finagle.thrift.{ClientId => FinagleClientId, _}
 import com.twitter.finagle.tracing.Tracer
 import com.twitter.finagle.transport.{Transport, TransportContext}
 import com.twitter.scrooge.TReusableBuffer
-import com.twitter.util.{Closable, Duration, Future, Monitor}
+import com.twitter.util.{Closable, Duration, Future, FuturePool, Monitor}
 import java.net.SocketAddress
+import java.util.concurrent.ExecutorService
 import org.apache.thrift.protocol.TProtocolFactory
 
 /**
@@ -418,6 +419,10 @@ object Thrift
       ]
     ): Client =
       super.withStack(fn)
+    override def withExecutionOffloaded(executor: ExecutorService): Client =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Client =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Client = super.configured(psp)
     override def filtered(
       filter: Filter[ThriftClientRequest, Array[Byte], ThriftClientRequest, Array[Byte]]
@@ -599,7 +604,10 @@ object Thrift
       ]
     ): Server =
       super.withStack(fn)
-
+    override def withExecutionOffloaded(executor: ExecutorService): Server =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Server =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Server = super.configured(psp)
   }
 

finagle-thriftmux/src/main/scala/com/twitter/finagle/ThriftMux.scala

@@ -41,6 +41,7 @@ import com.twitter.io.Buf
 import com.twitter.scrooge.TReusableBuffer
 import com.twitter.util._
 import java.net.SocketAddress
+import java.util.concurrent.ExecutorService
 import org.apache.thrift.TException
 import org.apache.thrift.protocol.TProtocolFactory
 
@@ -420,7 +421,10 @@ object ThriftMux
     override def withRetryBudget(budget: RetryBudget): Client = super.withRetryBudget(budget)
     override def withRetryBackoff(backoff: Stream[Duration]): Client =
       super.withRetryBackoff(backoff)
-
+    override def withExecutionOffloaded(executor: ExecutorService): Client =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Client =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Client = super.configured(psp)
   }
 
@@ -668,7 +672,10 @@ object ThriftMux
     override def withExceptionStatsHandler(exceptionStatsHandler: ExceptionStatsHandler): Server =
       super.withExceptionStatsHandler(exceptionStatsHandler)
     override def withRequestTimeout(timeout: Duration): Server = super.withRequestTimeout(timeout)
-
+    override def withExecutionOffloaded(executor: ExecutorService): Server =
+      super.withExecutionOffloaded(executor)
+    override def withExecutionOffloaded(pool: FuturePool): Server =
+      super.withExecutionOffloaded(pool)
     override def configured[P](psp: (P, Stack.Param[P])): Server = super.configured(psp)
   }