Update MMX libraries and replace scala_tools.time (#220)

* * Removed scala 2.10 support
* Updated samza and kafka versions

* Updated documentation and sbt version

* Removed JDK7 since new samza uses JDK8 features

* * Update MMX libraries versions
* Replaced scala_tools.time by nscala_time since scala_tools.time doesn't support Scala 2.12 and nscala_time is being actively maintained

* Fix for sigar

* Update twitter util and finagle

* Updated scala-util and removed resolver for sigar

* Updated plugin versions

Author: logarithm

.travis.yml

@@ -1,10 +1,8 @@
 language: scala
 scala:
-   - 2.10.5
-   - 2.11.7
+   - 2.11.8
 sudo: false
 jdk:
   - oraclejdk8
-  - oraclejdk7
 script:
   - sbt ++$TRAVIS_SCALA_VERSION -Dfile.encoding=UTF8 -J-XX:ReservedCodeCacheSize=256M -J-Xms512m -J-Xmx512m test

CONTRIBUTING.md

@@ -3,8 +3,9 @@
 Tranquility generally follows the same contributing guidelines as Druid. See here for Druid's guidelines:
 https://github.com/druid-io/druid/blob/master/CONTRIBUTING.md
 
-The Druid [intellij_formatting.jar](https://github.com/druid-io/druid/raw/master/intellij_formatting.jar) contains a
-Scala codestyle as well, which you can use to ensure your Scala code is formatted according to project conventions.
+The Druid [druid_intellij_formatting.xml](https://github.com/druid-io/druid/raw/master/druid_intellij_formatting.xml) 
+contains a Scala codestyle as well, which you can use to ensure your Scala code is formatted according to project 
+conventions.
 
 The [Druid CLA](http://druid.io/community/cla.html) applies to Tranquility as well. Please fill it out when
 contributing to either project. The same form does apply to both projects, so there is no need to fill it out twice.

README.md

@@ -41,35 +41,33 @@ Central to make them easy to include. The current stable versions are:
 <dependency>
   <groupId>io.druid</groupId>
   <artifactId>tranquility-core_2.11</artifactId>
-  <version>0.8.1</version>
+  <version>0.9.0</version>
 </dependency>
 <dependency>
   <groupId>io.druid</groupId>
-  <artifactId>tranquility-samza_2.10</artifactId>
-  <version>0.8.1</version>
+  <artifactId>tranquility-samza_2.11</artifactId>
+  <version>0.9.0</version>
 </dependency>
 <dependency>
   <groupId>io.druid</groupId>
   <artifactId>tranquility-spark_2.11</artifactId>
-  <version>0.8.1</version>
+  <version>0.9.0</version>
 </dependency>
 <dependency>
   <groupId>io.druid</groupId>
   <artifactId>tranquility-storm_2.11</artifactId>
-  <version>0.8.1</version>
+  <version>0.9.0</version>
 </dependency>
 <dependency>
   <groupId>io.druid</groupId>
   <artifactId>tranquility-flink_2.11</artifactId>
-  <version>0.8.1</version>
+  <version>0.9.0</version>
 </dependency>
 ```
 
 You only need to include the modules you are actually using.
 
-All Tranquility modules are built for both Scala 2.10 and 2.11, except for the Samza module, which is only built for
-Scala 2.10. If you're using Scala for your own code, you should choose the Tranquility build that matches your version
-of Scala. Otherwise, Scala 2.11 is recommended.
+All Tranquility modules are built for Scala 2.11.
 
 This version is built to work with Druid 0.7.x and 0.8.x. If you are using Druid 0.6.x, you may want to use Tranquility
 v0.3.2, which is the most recent version built for use with Druid 0.6.x.
@@ -85,10 +83,10 @@ run them directly. The distribution also includes the [Core API](docs/core.md) a
 them rather than get them through Maven.
 
 The current distribution is:
-[tranquility-distribution-0.8.1](http://static.druid.io/tranquility/releases/tranquility-distribution-0.8.1.tgz).
+[tranquility-distribution-0.9.0](http://static.druid.io/tranquility/releases/tranquility-distribution-0.9.0.tgz).
 
 To use it, first download it and then unpack it into your directory of choice by running
-`tar -xzf tranquility-distribution-0.8.1.tgz`.
+`tar -xzf tranquility-distribution-0.9.0.tgz`.
 
 ### How to Contribute
 

build.sbt

@@ -1,6 +1,4 @@
-scalaVersion := "2.10.6"
-
-crossScalaVersions := Seq("2.10.6", "2.11.8")
+scalaVersion in ThisBuild := "2.11.8"
 
 // Disable parallel execution, the various Druid oriented tests need to claim ports
 parallelExecution in ThisBuild := false
@@ -18,14 +16,14 @@ val druidVersion = "0.9.2"
 val curatorVersion = "2.12.0"
 val guiceVersion = "4.0"
 val flinkVersion = "1.0.3"
-val finagleVersion = "6.31.0"
-val twitterUtilVersion = "6.30.0"
-val samzaVersion = "0.8.0"
+val finagleVersion = "6.43.0"
+val twitterUtilVersion = "6.42.0"
+val samzaVersion = "0.12.0"
 val sparkVersion = "1.6.0"
 val scalatraVersion = "2.3.1"
 val jettyVersion = "9.2.5.v20141112"
 val apacheHttpVersion = "4.3.3"
-val kafkaVersion = "0.8.2.2"
+val kafkaVersion = "0.10.1.1"
 val airlineVersion = "0.7"
 
 def dependOnDruid(artifact: String) = {
@@ -45,8 +43,8 @@ def dependOnDruid(artifact: String) = {
 }
 
 val coreDependencies = Seq(
-  "com.metamx" %% "scala-util" % "1.11.6" exclude("log4j", "log4j") force(),
-  "com.metamx" % "java-util" % "0.27.10" exclude("log4j", "log4j") force(),
+  "com.metamx" %% "scala-util" % "1.13.2" exclude("log4j", "log4j") force(),
+  "com.metamx" % "java-util" % "0.28.2" exclude("log4j", "log4j") force(),
   "io.netty" % "netty" % "3.10.5.Final" force(),
   "org.apache.curator" % "curator-client" % curatorVersion force(),
   "org.apache.curator" % "curator-framework" % curatorVersion force(),
@@ -55,8 +53,8 @@ val coreDependencies = Seq(
   "com.twitter" %% "util-core" % twitterUtilVersion force(),
   "com.twitter" %% "finagle-core" % finagleVersion force(),
   "com.twitter" %% "finagle-http" % finagleVersion force(),
-  "org.slf4j" % "slf4j-api" % "1.7.12" force() force(),
-  "org.slf4j" % "jul-to-slf4j" % "1.7.12" force() force(),
+  "org.slf4j" % "slf4j-api" % "1.7.25" force() force(),
+  "org.slf4j" % "jul-to-slf4j" % "1.7.25" force() force(),
   "org.apache.httpcomponents" % "httpclient" % apacheHttpVersion force(),
   "org.apache.httpcomponents" % "httpcore" % apacheHttpVersion force(),
 
@@ -88,7 +86,7 @@ val loggingDependencies = Seq(
   "org.slf4j" % "jul-to-slf4j" % "1.7.12"
 )
 
-def flinkDependencies(scalaVersion: String) = {
+val flinkDependencies = {
   Seq(
     "org.apache.flink" %% "flink-streaming-scala" % flinkVersion % "optional"
     exclude("log4j", "log4j")
@@ -146,19 +144,21 @@ val coreTestDependencies = Seq(
   "org.slf4j" % "jul-to-slf4j" % "1.7.12" % "test"
 ) ++ loggingDependencies.map(_ % "test")
 
-def flinkTestDependencies(scalaVersion: String) = {
+val flinkTestDependencies = {
   Seq("org.apache.flink" % "flink-core" % flinkVersion % "test" classifier "tests",
     "org.apache.flink" %% "flink-runtime" % flinkVersion % "test" classifier "tests",
     "org.apache.flink" %% "flink-test-utils" % flinkVersion % "test"
   ).map(_ exclude("log4j", "log4j") exclude("org.slf4j", "slf4j-log4j12") force()) ++
     loggingDependencies.map(_ % "test")
 }
 
-// Force 2.10 here, makes update resolution happy, but since w'ere not building for 2.11
-// we won't end up in runtime version hell by doing this.
-val samzaTestDependencies = Seq(
-  "org.apache.samza" % "samza-core_2.10" % samzaVersion % "test"
-)
+val samzaTestDependencies = {
+  Seq(
+    "org.apache.samza" %% "samza-core" % samzaVersion % "test",
+    "org.apache.samza" %% "samza-kafka" % samzaVersion % "test"
+  ).map(_ exclude("log4j", "log4j") exclude("org.slf4j", "slf4j-log4j12") force()) ++
+    loggingDependencies.map(_ % "test")
+}
 
 val serverTestDependencies = Seq(
   "org.scalatra" %% "scalatra-test" % scalatraVersion % "test"
@@ -204,8 +204,8 @@ lazy val commonSettings = Seq(
       </developers>),
 
   fork in Test := true
-) ++ releaseSettings ++ net.virtualvoid.sbt.graph.Plugin.graphSettings ++ Seq(
-  ReleaseKeys.publishArtifactsAction := PgpKeys.publishSigned.value
+) ++ Seq(
+  releasePublishArtifactsAction := PgpKeys.publishSigned.value
 )
 
 lazy val root = project.in(file("."))
@@ -222,8 +222,7 @@ lazy val core = project.in(file("core"))
 lazy val flink = project.in(file("flink"))
   .settings(commonSettings: _*)
   .settings(name := "tranquility-flink")
-  .settings(libraryDependencies <++= scalaVersion(flinkDependencies))
-  .settings(libraryDependencies <++= scalaVersion(flinkTestDependencies))
+  .settings(libraryDependencies ++= (flinkDependencies ++ flinkTestDependencies))
   .dependsOn(core % "test->test;compile->compile")
 
 lazy val spark = project.in(file("spark"))
@@ -243,10 +242,6 @@ lazy val samza = project.in(file("samza"))
   .settings(commonSettings: _*)
   .settings(name := "tranquility-samza")
   .settings(libraryDependencies ++= (samzaDependencies ++ samzaTestDependencies))
-  // don't compile or publish for Scala > 2.10
-  .settings((skip in compile) := scalaVersion { sv => !sv.startsWith("2.10.") }.value)
-  .settings((skip in test) := scalaVersion { sv => !sv.startsWith("2.10.") }.value)
-  .settings(publishArtifact <<= scalaVersion { sv => sv.startsWith("2.10.") })
   .settings(publishArtifact in Test := false)
   .dependsOn(core % "test->test;compile->compile")
 

core/src/main/scala/com/metamx/tranquility/beam/BeamMaker.scala

@@ -18,8 +18,8 @@
  */
 package com.metamx.tranquility.beam
 
+import com.github.nscala_time.time.Imports._
 import com.metamx.common.scala.untyped._
-import org.scala_tools.time.Imports._
 
 /**
  * Makes beams for particular intervals and partition numbers. Can also serialize and deserialize beam representations.

core/src/main/scala/com/metamx/tranquility/beam/ClusteredBeam.scala

@@ -19,6 +19,7 @@
 package com.metamx.tranquility.beam
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import com.github.nscala_time.time.Imports._
 import com.google.common.util.concurrent.ThreadFactoryBuilder
 import com.metamx.common.scala.Logging
 import com.metamx.common.scala.Predef._
@@ -30,22 +31,17 @@ import com.metamx.common.scala.timekeeper.Timekeeper
 import com.metamx.common.scala.untyped._
 import com.metamx.emitter.service.ServiceEmitter
 import com.metamx.tranquility.typeclass.Timestamper
-import com.twitter.util.Future
-import com.twitter.util.FuturePool
-import com.twitter.util.Promise
-import com.twitter.util.Return
-import com.twitter.util.Throw
+import com.twitter.util._
 import java.util.UUID
 import java.util.concurrent.Executors
 import java.util.concurrent.atomic.AtomicBoolean
 import org.apache.curator.framework.CuratorFramework
 import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex
 import org.apache.zookeeper.KeeperException.NodeExistsException
+import org.joda.time.chrono.ISOChronology
 import org.joda.time.DateTime
 import org.joda.time.DateTimeZone
 import org.joda.time.Interval
-import org.joda.time.chrono.ISOChronology
-import org.scala_tools.time.Implicits._
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.language.reflectiveCalls
@@ -190,16 +186,16 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
   private[this] def beam(timestamp: DateTime, now: DateTime): Future[Beam[EventType]] = {
     val bucket = tuning.segmentBucket(timestamp)
     val creationInterval = new Interval(
-      tuning.segmentBucket(now - tuning.windowPeriod).start.millis,
-      tuning.segmentBucket(Seq(now + tuning.warmingPeriod, now + tuning.windowPeriod).maxBy(_.millis)).end.millis,
+      tuning.segmentBucket(now - tuning.windowPeriod).start.getMillis,
+      tuning.segmentBucket(Seq(now + tuning.warmingPeriod, now + tuning.windowPeriod).maxBy(_.getMillis)).end.getMillis,
       ISOChronology.getInstanceUTC
     )
     val windowInterval = new Interval(
-      tuning.segmentBucket(now - tuning.windowPeriod).start.millis,
-      tuning.segmentBucket(now + tuning.windowPeriod).end.millis,
+      tuning.segmentBucket(now - tuning.windowPeriod).start.getMillis,
+      tuning.segmentBucket(now + tuning.windowPeriod).end.getMillis,
       ISOChronology.getInstanceUTC
     )
-    val futureBeamOption = beams.get(timestamp.millis) match {
+    val futureBeamOption = beams.get(timestamp.getMillis) match {
       case _ if !open => Future.value(None)
       case Some(x) if windowInterval.overlaps(bucket) => Future.value(Some(x))
       case Some(x) => Future.value(None)
@@ -210,7 +206,7 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
         // This could be more efficient, but it's happening infrequently so it's probably not a big deal.
         data.modify {
           prev =>
-            val prevBeamDicts = prev.beamDictss.getOrElse(timestamp.millis, Nil)
+            val prevBeamDicts = prev.beamDictss.getOrElse(timestamp.getMillis, Nil)
             if (prevBeamDicts.size >= tuning.partitions) {
               log.info(
                 "Merged beam already created for identifier[%s] timestamp[%s], with sufficient partitions (target = %d, actual = %d)",
@@ -236,8 +232,8 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
               val numSegmentsToCover = tuning.minSegmentsPerBeam +
                 rand.nextInt(tuning.maxSegmentsPerBeam - tuning.minSegmentsPerBeam + 1)
               val intervalToCover = new Interval(
-                timestamp.millis,
-                tuning.segmentGranularity.increment(timestamp, numSegmentsToCover).millis,
+                timestamp.getMillis,
+                tuning.segmentGranularity.increment(timestamp, numSegmentsToCover).getMillis,
                 ISOChronology.getInstanceUTC
               )
               val timestampsToCover = tuning.segmentGranularity.getIterable(intervalToCover).asScala.map(_.start)
@@ -249,7 +245,7 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
                   // Expire old beamDicts
                   tuning.segmentGranularity.increment(new DateTime(millis)) + tuning.windowPeriod < now
               }) ++ (for (ts <- timestampsToCover) yield {
-                val tsPrevDicts = prev.beamDictss.getOrElse(ts.millis, Nil)
+                val tsPrevDicts = prev.beamDictss.getOrElse(ts.getMillis, Nil)
                 log.info(
                   "Creating new merged beam for identifier[%s] timestamp[%s] (target = %d, actual = %d)",
                   identifier,
@@ -272,10 +268,10 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
                       }
                     )
                 })
-                (ts.millis, tsNewDicts)
+                (ts.getMillis, tsNewDicts)
               })
               val newLatestCloseTime = new DateTime(
-                (Seq(prev.latestCloseTime.millis) ++ (prev.beamDictss.keySet -- newBeamDictss.keySet)).max,
+                (Seq(prev.latestCloseTime.getMillis) ++ (prev.beamDictss.keySet -- newBeamDictss.keySet)).max,
                 ISOChronology.getInstanceUTC
               )
               ClusteredBeamMeta(
@@ -298,12 +294,12 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
               // Only add the beams we actually wanted at this time. This is because there might be other beams in ZK
               // that we don't want to add just yet, on account of maybe they need their partitions expanded (this only
               // happens when they are the necessary ones).
-              if (!beams.contains(timestamp.millis) && meta.beamDictss.contains(timestamp.millis)) {
-                val beamDicts = meta.beamDictss(timestamp.millis)
+              if (!beams.contains(timestamp.getMillis) && meta.beamDictss.contains(timestamp.getMillis)) {
+                val beamDicts = meta.beamDictss(timestamp.getMillis)
                 log.info("Adding beams for identifier[%s] timestamp[%s]: %s", identifier, timestamp, beamDicts)
                 // Should have better handling of unparseable zk data. Changing BeamMaker implementations currently
                 // just causes exceptions until the old dicts are cleared out.
-                beams(timestamp.millis) = beamMergeFn(
+                beams(timestamp.getMillis) = beamMergeFn(
                   beamDicts.zipWithIndex map {
                     case (beamDict, partitionNum) =>
                       val decorate = beamDecorateFn(tuning.segmentBucket(timestamp), partitionNum)
@@ -319,7 +315,7 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
                 beams.remove(timestamp)
               }
               // Return requested beam. It may not have actually been created, so it's an Option.
-              beams.get(timestamp.millis) ifEmpty {
+              beams.get(timestamp.getMillis) ifEmpty {
                 log.info(
                   "Turns out we decided not to actually make beams for identifier[%s] timestamp[%s]. Returning None.",
                   identifier,
@@ -352,7 +348,7 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
     val grouped: Seq[(DateTime, IndexedSeq[(EventType, Promise[SendResult])])] = (eventsWithPromises groupBy {
       case (event, promise) =>
         tuning.segmentBucket(timestamper(event)).start
-    }).toSeq.sortBy(_._1.millis)
+    }).toSeq.sortBy(_._1.getMillis)
     // Possibly warm up future beams
     def toBeWarmed(dt: DateTime, end: DateTime): List[DateTime] = {
       if (dt <= end) {
@@ -362,7 +358,7 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
       }
     }
     val latestEventTimestamp: Option[DateTime] = grouped.lastOption map { case (truncatedTimestamp, group) =>
-      val event: EventType = group.maxBy(tuple => timestamper(tuple._1).millis)._1
+      val event: EventType = group.maxBy(tuple => timestamper(tuple._1).getMillis)._1
       timestamper(event)
     }
     val warmingBeams: Future[Seq[Beam[EventType]]] = Future.collect(
@@ -406,13 +402,13 @@ class ClusteredBeam[EventType: Timestamper, InnerBeamType <: Beam[EventType]](
                   data.modify {
                     prev =>
                       ClusteredBeamMeta(
-                        Seq(prev.latestCloseTime, timestamp).maxBy(_.millis),
-                        prev.beamDictss - timestamp.millis
+                        Seq(prev.latestCloseTime, timestamp).maxBy(_.getMillis),
+                        prev.beamDictss - timestamp.getMillis
                       )
                   } onSuccess {
                     meta =>
                       beamWriteMonitor.synchronized {
-                        beams.remove(timestamp.millis)
+                        beams.remove(timestamp.getMillis)
                       }
                   } map (_ => SendResult.Dropped)
                 } else {
@@ -479,7 +475,7 @@ case class ClusteredBeamMeta(latestCloseTime: DateTime, beamDictss: Map[Long, Se
     Dict(
       // latestTime is only being written for backwards compatibility
       "latestTime" -> new DateTime(
-        (Seq(latestCloseTime.millis) ++ beamDictss.map(_._1)).max,
+        (Seq(latestCloseTime.getMillis) ++ beamDictss.map(_._1)).max,
         ISOChronology.getInstanceUTC
       ).toString(),
       "latestCloseTime" -> latestCloseTime.toString(),
@@ -499,7 +495,7 @@ object ClusteredBeamMeta
         case (k, vs) =>
           val ts = new DateTime(k, ISOChronology.getInstanceUTC)
           val beamDicts = list(vs) map (dict(_))
-          (ts.millis, beamDicts)
+          (ts.getMillis, beamDicts)
       }
       val latestCloseTime = new DateTime(d.getOrElse("latestCloseTime", 0L), ISOChronology.getInstanceUTC)
       Right(ClusteredBeamMeta(latestCloseTime, beams))