Standardize on configuration files and use of Tranquilizer.

- Introduce TranquilityConfig.read and DruidBeams.fromConfig.
- Switch all modules from simple tranquilizer to Tranquilizer + flush.
- Switch Java/Scala code examples to Tranquilizer + flush.
- Adjust default config files to have more consistent formatting.
- Allow 'dataSources' to be a list instead of a map in config files.

Also:
- Include full date in bundled logback.xml

Fixes #103

README.md

@@ -16,9 +16,8 @@ General:
 
 - [Overview](docs/overview.md) - Introduction to Tranquility concepts, including details about how it creates and
   manages Druid tasks.
-- [DruidBeams](docs/druidbeams.md) - The first step to using Tranquility is to configure it appropriately for your
-  Druid dataSource. For most modules (including core, storm, spark, and samza) this is generally done through the
-  DruidBeams builder object.
+- [Configuration](docs/configuration.md) - The first step to using Tranquility is to configure it appropriately for your
+  Druid dataSource. This is generally done through a configuration file, with optional refinements in code.
 - [Troubleshooting](docs/trouble.md) - Solutions to common problems.
 
 Modules:

build.sbt

@@ -15,6 +15,7 @@ val jacksonOneVersion = "1.9.13"
 val jacksonTwoVersion = "2.4.6"
 val jacksonTwoModuleScalaVersion = "2.4.5"
 val druidVersion = "0.8.2"
+val guiceVersion = "4.0"
 val flinkVersion = "0.10.1"
 val finagleVersion = "6.31.0"
 val twitterUtilVersion = "6.30.0"
@@ -63,9 +64,9 @@ val coreDependencies = Seq(
   "com.fasterxml.jackson.module" %% "jackson-module-scala" % jacksonTwoModuleScalaVersion force()
 ) ++ Seq(
   dependOnDruid("druid-server"),
-  "com.google.inject" % "guice" % "4.0-beta" force(),
-  "com.google.inject.extensions" % "guice-servlet" % "4.0-beta" force(),
-  "com.google.inject.extensions" % "guice-multibindings" % "4.0-beta" force(),
+  "com.google.inject" % "guice" % guiceVersion force(),
+  "com.google.inject.extensions" % "guice-servlet" % guiceVersion force(),
+  "com.google.inject.extensions" % "guice-multibindings" % guiceVersion force(),
   "javax.validation" % "validation-api" % "1.1.0.Final" force()
 )
 

core/src/main/scala/com/metamx/tranquility/config/DataSourceConfig.scala

@@ -0,0 +1,56 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.metamx.tranquility.config
+
+import com.metamx.common.scala.untyped.Dict
+import com.metamx.common.scala.untyped.dict
+import com.metamx.common.scala.untyped.str
+import com.metamx.tranquility.tranquilizer.Tranquilizer
+
+case class DataSourceConfig[T <: PropertiesBasedConfig](
+  dataSource: String,
+  propertiesBasedConfig: T,
+  specMap: Dict
+)
+{
+  validate()
+
+  def tranquilizerBuilder(): Tranquilizer.Builder = {
+    Tranquilizer.builder()
+      .maxBatchSize(propertiesBasedConfig.tranquilityMaxBatchSize)
+      .maxPendingBatches(propertiesBasedConfig.tranquilityMaxPendingBatches)
+      .lingerMillis(propertiesBasedConfig.tranquilityLingerMillis)
+  }
+
+  private def validate(): Unit = {
+    // Sanity check: two ways of providing dataSource, they must match
+    val specDataSource = try {
+      str(dict(specMap("dataSchema"))("dataSource"))
+    }
+    catch {
+      case e: ClassCastException =>
+        throw new IllegalArgumentException(s"spec[$dataSource] is missing 'dataSource' inside 'dataSchema'")
+    }
+    require(
+      dataSource == specDataSource,
+      s"dataSource[$dataSource] did not match spec[$specDataSource]"
+    )
+  }
+}

core/src/main/scala/com/metamx/tranquility/config/config.scala → core/src/main/scala/com/metamx/tranquility/config/PropertiesBasedConfig.scala

@@ -25,17 +25,22 @@ import com.metamx.common.scala.untyped.Dict
 import com.metamx.tranquility.beam.ClusteredBeamTuning
 import com.metamx.tranquility.druid.DruidBeamConfig
 import com.metamx.tranquility.tranquilizer.Tranquilizer
-import io.druid.segment.realtime.FireDepartment
+import java.util.Properties
 import org.joda.time.Period
 import org.skife.config.Config
+import org.skife.config.ConfigurationObjectFactory
 
-abstract class TranquilityConfig(globalProperties: Set[String]) extends DiscoConfig
+abstract class PropertiesBasedConfig(
+  private[config] val globalPropertyNames: Set[String]
+) extends DiscoConfig
 {
+  private var props: Properties = null
+
   def this() {
     this(Set[String]())
   }
 
-  val GlobalProperties = Set[String]() ++ globalProperties
+  def properties: Properties = props
 
   @Config(Array("druid.selectors.indexing.serviceName"))
   def druidIndexingServiceName: String = "druid/overlord"
@@ -101,4 +106,20 @@ abstract class TranquilityConfig(globalProperties: Set[String]) extends DiscoCon
   override def discoAnnounce: Option[DiscoAnnounceConfig] = None
 }
 
-case class DataSourceConfig[T <: TranquilityConfig](config: T, dataSource: String, specMap: Dict)
+object PropertiesBasedConfig
+{
+  def fromDict[ConfigType <: PropertiesBasedConfig](
+    d: Dict,
+    clazz: Class[ConfigType]
+  ): ConfigType =
+  {
+    val properties = new Properties
+    for ((k, v) <- d if v != null) {
+      properties.setProperty(k, String.valueOf(v))
+    }
+    val configFactory = new ConfigurationObjectFactory(properties)
+    val config = configFactory.build(clazz)
+    config.props = properties
+    config
+  }
+}