Exception chaining

.gitignore

@@ -1,6 +1,10 @@
+.idea/
+*.iml
+*.ipr
 dist/
 target/
 lib_managed/
 src_managed/
 project/boot/
 project/plugins/project/
+querulous.tmproj

README.markdown

@@ -154,7 +154,7 @@ Add the following dependency and repository stanzas to your project's configurat
 
     <repository>
       <id>twitter.com</id>
-      <url>http://www.lag.net/nest</url>
+      <url>http://maven.twttr.com/</url>
     </repository>
 
 ### Ivy
@@ -165,7 +165,7 @@ Add the following dependency to ivy.xml
 
 and the following repository to ivysettings.xml
 
-    <ibiblio name="twitter.com" m2compatible="true" root="http://www.lag.net/nest/" />
+    <ibiblio name="twitter.com" m2compatible="true" root="http://maven.twttr.com/" />
 
 ## Running Tests
 

config/test.scala

@@ -0,0 +1,15 @@
+import com.twitter.querulous.config.Connection
+
+new Connection {
+  val hostnames = Seq("localhost")
+  val database = "db_test"
+  val username = {
+    val userEnv = System.getenv("DB_USERNAME")
+    if (userEnv == null) "root" else userEnv
+  }
+
+  val password = {
+    val passEnv = System.getenv("DB_PASSWORD")
+    if (passEnv == null) "" else passEnv
+  }
+}

project/build.properties

@@ -1,8 +1,8 @@
 #Project properties
-#Fri Sep 17 11:09:45 PDT 2010
+#Mon Feb 28 13:01:14 PST 2011
 project.organization=com.twitter
 project.name=querulous
 sbt.version=0.7.4
-project.version=1.2.2
-build.scala.versions=2.7.7
+project.version=2.0.2-SNAPSHOT
+build.scala.versions=2.8.1
 project.initialize=false

project/build/QuerulousProject.scala

@@ -1,30 +1,23 @@
 import sbt._
+import Process._
 import com.twitter.sbt._
 
-
 class QuerulousProject(info: ProjectInfo) extends StandardProject(info) with SubversionPublisher {
-  val vscaladoc = "org.scala-tools" % "vscaladoc" % "1.1-md-3"
-  val configgy  = "net.lag" % "configgy" % "1.5.2"
+  override def filterScalaJars = false
+
+  val util      = "com.twitter" % "util" % "1.6.4"
+
   val dbcp      = "commons-dbcp" % "commons-dbcp" % "1.4"
-  val mysqljdbc = "mysql" % "mysql-connector-java" % "5.1.6"
+  val mysqljdbc = "mysql" % "mysql-connector-java" % "5.1.13"
   val pool      = "commons-pool" % "commons-pool" % "1.5.4"
-  val xrayspecs = "com.twitter" % "xrayspecs" % "1.0.7"
-
-  val hamcrest  = "org.hamcrest" % "hamcrest-all" % "1.1" % "test"
-  val specs     = "org.scala-tools.testing" % "specs" % "1.6.2.1" % "test"
-  val objenesis = "org.objenesis" % "objenesis" % "1.1" % "test"
-  val jmock     = "org.jmock" % "jmock" % "2.4.0" % "test"
-  val cglib     = "cglib" % "cglib" % "2.1_3" % "test"
-  val asm       = "asm" % "asm" %  "1.5.3" % "test"
 
-  override def pomExtra =
-    <licenses>
-      <license>
-        <name>Apache 2</name>
-        <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
-        <distribution>repo</distribution>
-      </license>
-    </licenses>
+  val scalaTools = "org.scala-lang" % "scala-compiler" % "2.8.1" % "test"
+  val hamcrest   = "org.hamcrest" % "hamcrest-all" % "1.1" % "test"
+  val specs      = "org.scala-tools.testing" % "specs_2.8.0" % "1.6.5" % "test"
+  val objenesis  = "org.objenesis" % "objenesis" % "1.1" % "test"
+  val jmock      = "org.jmock" % "jmock" % "2.4.0" % "test"
+  val cglib      = "cglib" % "cglib" % "2.1_3" % "test"
+  val asm        = "asm" % "asm" %  "1.5.3" % "test"
 
   override def subversionRepository = Some("http://svn.local.twitter.com/maven-public/")
 }

project/plugins/Plugins.scala

@@ -1,6 +1,6 @@
 import sbt._
 
 class Plugins(info: ProjectInfo) extends PluginDefinition(info) {
-  val twitterNest = "com.twitter" at "http://www.lag.net/nest"
-  val defaultProject = "com.twitter" % "standard-project" % "0.7.1"
+  val twitter = "twitter.com" at "http://maven.twttr.com/"
+  val defaultProject = "com.twitter" % "standard-project" % "0.7.17"
 }

src/main/scala/com/twitter/querulous/AutoDisabler.scala

@@ -1,15 +1,15 @@
 package com.twitter.querulous
 
-import com.twitter.xrayspecs.{Time, Duration}
-import com.twitter.xrayspecs.TimeConversions._
+import com.twitter.util.{Time, Duration}
+import com.twitter.util.TimeConversions._
 import java.sql.{SQLException, SQLIntegrityConstraintViolationException}
 
 
 trait AutoDisabler {
   protected val disableErrorCount: Int
   protected val disableDuration: Duration
 
-  private var disabledUntil: Time = Time.never
+  private var disabledUntil: Time = Time.epoch
   private var consecutiveErrors = 0
 
   protected def throwIfDisabled(throwMessage: String): Unit = {

src/main/scala/com/twitter/querulous/ConnectionDestroying.scala

@@ -4,12 +4,10 @@ import java.sql.Connection
 import org.apache.commons.dbcp.{DelegatingConnection => DBCPConnection}
 import com.mysql.jdbc.{ConnectionImpl => MySQLConnection}
 
-
 // Emergency connection destruction toolkit
-
 trait ConnectionDestroying {
   def destroyConnection(conn: Connection) {
-    if ( !conn.isClosed )
+    if (!conn.isClosed)
       conn match {
         case c: DBCPConnection =>
           destroyDbcpWrappedConnection(c)
@@ -22,21 +20,18 @@ trait ConnectionDestroying {
   def destroyDbcpWrappedConnection(conn: DBCPConnection) {
     val inner = conn.getInnermostDelegate
 
-    if ( inner != null ) {
+    if (inner ne null) {
       destroyConnection(inner)
     } else {
-      // this should never happen if we use our own ApachePoolingDatabase to get connections.
-      error("Could not get access to the delegate connection. Make sure the dbcp connection pool allows access to underlying connections.")
+      // might just be a race; move on.
+      return
     }
 
     // "close" the wrapper so that it updates its internal bookkeeping, just do it
-    try { conn.close } catch { case _ => }
+    try { conn.close() } catch { case _ => }
   }
 
   def destroyMysqlConnection(conn: MySQLConnection) {
-    val abort = Class.forName("com.mysql.jdbc.ConnectionImpl").getDeclaredMethod("abortInternal")
-    abort.setAccessible(true)
-
-    abort.invoke(conn)
+    conn.abortInternal()
   }
 }

src/main/scala/com/twitter/querulous/FutureTimeout.scala

@@ -2,7 +2,7 @@ package com.twitter.querulous
 
 import java.util.concurrent.{ThreadFactory, TimeoutException => JTimeoutException, _}
 import java.util.concurrent.atomic.AtomicInteger
-import com.twitter.xrayspecs.Duration
+import com.twitter.util.Duration
 
 class FutureTimeout(poolSize: Int, queueSize: Int) {
   object DaemonThreadFactory extends ThreadFactory {
@@ -20,7 +20,7 @@ class FutureTimeout(poolSize: Int, queueSize: Int) {
       thread
     }
   }
-  private val executor = new ThreadPoolExecutor(poolSize, poolSize, 0, TimeUnit.SECONDS,
+  private val executor = new ThreadPoolExecutor(1, poolSize, 60, TimeUnit.SECONDS,
                                                 new LinkedBlockingQueue[Runnable](queueSize),
                                                 DaemonThreadFactory)
 

src/main/scala/com/twitter/querulous/StatsCollector.scala

@@ -4,3 +4,8 @@ trait StatsCollector {
   def incr(name: String, count: Int)
   def time[A](name: String)(f: => A): A
 }
+
+object NullStatsCollector extends StatsCollector {
+  def incr(name: String, count: Int) {}
+  def time[A](name: String)(f: => A): A = f
+}

src/main/scala/com/twitter/querulous/Timeout.scala

@@ -1,7 +1,7 @@
 package com.twitter.querulous
 
 import java.util.{Timer, TimerTask}
-import com.twitter.xrayspecs.Duration
+import com.twitter.util.Duration
 
 
 class TimeoutException extends Exception
@@ -20,7 +20,9 @@ object Timeout {
     } finally {
       task map { t =>
         t.cancel()
-        timer.purge()
+        // TODO(benjy): Timer is not optimized to deal with large numbers of cancellations: it releases and reacquires its monitor
+        // on every task, cancelled or not, when it could quickly skip over all cancelled tasks in a single monitor region.
+        // This may not be a problem, but it's something to be aware of.
       }
       if (cancelled) throw new TimeoutException
     }
@@ -32,7 +34,15 @@ object Timeout {
 
   private def schedule(timer: Timer, timeout: Duration, f: => Unit) = {
     val task = new TimerTask() {
-      override def run() { f }
+      override def run() {
+        try {
+          f
+        } catch {
+          case e: Throwable =>
+            error("Timer task tried to throw an exception: " + e.toString())
+            e.printStackTrace(System.err)
+        }
+      }
     }
     timer.schedule(task, timeout.inMillis)
     task

src/main/scala/com/twitter/querulous/config/Database.scala

@@ -0,0 +1,125 @@
+package com.twitter.querulous.config
+
+import com.twitter.querulous._
+import com.twitter.util.Duration
+import com.twitter.util.TimeConversions._
+import database._
+
+
+class ApachePoolingDatabase {
+  var sizeMin: Int = 10
+  var sizeMax: Int = 10
+  var testIdle: Duration = 1.second
+  var maxWait: Duration = 10.millis
+  var minEvictableIdle: Duration = 60.seconds
+  var testOnBorrow: Boolean = false
+}
+
+class TimingOutDatabase {
+  var poolSize: Int = 10
+  var queueSize: Int = 10000
+  var open: Duration = 1.second
+}
+
+trait AutoDisablingDatabase {
+  def errorCount: Int
+  def interval: Duration
+}
+
+class Database {
+  var pool: Option[ApachePoolingDatabase] = None
+  def pool_=(p: ApachePoolingDatabase) { pool = Some(p) }
+  var autoDisable: Option[AutoDisablingDatabase] = None
+  def autoDisable_=(a: AutoDisablingDatabase) { autoDisable = Some(a) }
+  var timeout: Option[TimingOutDatabase] = None
+  def timeout_=(t: TimingOutDatabase) { timeout = Some(t) }
+  var memoize: Boolean = true
+
+  def apply(stats: StatsCollector): DatabaseFactory = {
+    var factory: DatabaseFactory = pool.map(apacheConfig =>
+      new ApachePoolingDatabaseFactory(
+        apacheConfig.sizeMin,
+        apacheConfig.sizeMax,
+        apacheConfig.testIdle,
+        apacheConfig.maxWait,
+        apacheConfig.testOnBorrow,
+        apacheConfig.minEvictableIdle)
+    ).getOrElse(new SingleConnectionDatabaseFactory)
+
+    timeout.foreach { timeoutConfig =>
+      factory = new TimingOutDatabaseFactory(factory,
+        timeoutConfig.poolSize,
+        timeoutConfig.queueSize,
+        timeoutConfig.open,
+        timeoutConfig.poolSize)
+    }
+
+    if (stats ne NullStatsCollector) {
+      factory = new StatsCollectingDatabaseFactory(factory, stats)
+    }
+
+    autoDisable.foreach { disable =>
+      factory = new AutoDisablingDatabaseFactory(factory, disable.errorCount, disable.interval)
+    }
+
+    if (memoize) {
+      factory = new MemoizingDatabaseFactory(factory)
+    }
+
+    factory
+  }
+
+  def apply(): DatabaseFactory = apply(NullStatsCollector)
+}
+
+trait Connection {
+  def hostnames: Seq[String]
+  def database: String
+  def username: String
+  def password: String
+  var urlOptions: Map[String, String] = Map()
+
+  def withHost(newHost: String) = {
+    val current = this
+    new Connection {
+      def hostnames = Seq(newHost)
+      def database = current.database
+      def username = current.username
+      def password = current.password
+      urlOptions = current.urlOptions
+    }
+  }
+
+  def withHosts(newHosts: Seq[String]) = {
+    val current = this
+    new Connection {
+      def hostnames = newHosts
+      def database = current.database
+      def username = current.username
+      def password = current.password
+      urlOptions = current.urlOptions
+    }
+  }
+
+  def withDatabase(newDatabase: String) = {
+    val current = this
+    new Connection {
+      def hostnames = current.hostnames
+      def database = newDatabase
+      def username = current.username
+      def password = current.password
+      urlOptions = current.urlOptions
+    }
+  }
+
+  def withoutDatabase = {
+    val current = this
+    new Connection {
+      def hostnames = current.hostnames
+      def database = null
+      def username = current.username
+      def password = current.password
+      urlOptions = current.urlOptions
+    }
+  }
+}