update dependencies

.github/workflows/pr.yml

@@ -8,7 +8,7 @@ jobs:
     strategy:
       matrix:
         java: [17, 21]
-        scala: [2.13.14, 3.4.1]
+        scala: [2.13.15, 3.5.1]
     steps:
       - uses: actions/checkout@v3
       - name: Set up JDK ${{ matrix.java }}

.github/workflows/release.yml

@@ -31,5 +31,5 @@ jobs:
           PGP_PASSPHRASE: ${{ secrets.ORG_SIGNING_PASSWORD }}
         run: |
           git fetch --unshallow --tags
-          cat /dev/null | project/sbt ++2.13.14 clean test +publishSigned
+          cat /dev/null | project/sbt ++2.13.15 clean test +publishSigned
           cat /dev/null | project/sbt sonatypeBundleRelease

.github/workflows/snapshot.yml

@@ -31,4 +31,4 @@ jobs:
           PGP_PASSPHRASE: ${{ secrets.ORG_SIGNING_PASSWORD }}
         run: |
           git fetch --unshallow --tags
-          cat /dev/null | project/sbt ++2.13.14 clean test +publishSigned
+          cat /dev/null | project/sbt ++2.13.15 clean test +publishSigned

atlas-json/src/main/scala/com/netflix/atlas/json/JsonParserHelper.scala

@@ -17,7 +17,6 @@ package com.netflix.atlas.json
 
 import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.core.JsonToken
-import com.fasterxml.jackson.core.io.doubleparser.JavaDoubleParser
 
 object JsonParserHelper {
 
@@ -94,11 +93,16 @@ object JsonParserHelper {
     parser.nextToken() match {
       case VALUE_NUMBER_INT   => parser.getValueAsDouble
       case VALUE_NUMBER_FLOAT => parser.getValueAsDouble
-      case VALUE_STRING       => JavaDoubleParser.parseDouble(parser.getText)
+      case VALUE_STRING       => parseDouble(parser.getText)
       case t                  => fail(parser, s"expected VALUE_NUMBER_FLOAT but received $t")
     }
   }
 
+  private def parseDouble(v: String): Double = {
+    // Common case for string encoding is NaN
+    if (v == "NaN") Double.NaN else java.lang.Double.parseDouble(v)
+  }
+
   def foreachItem[T](parser: JsonParser)(f: => T): Unit = {
     requireNextToken(parser, JsonToken.START_ARRAY)
     var t = parser.nextToken()

atlas-lwc-events/src/main/scala/com/netflix/atlas/lwc/events/DatapointConverter.scala

@@ -94,6 +94,7 @@ private[events] object DatapointConverter {
     v * v
   }
 
+  @scala.annotation.tailrec
   private[events] def toDouble(value: Any, dflt: Any): Double = {
     value match {
       case v: Boolean  => if (v) 1.0 else 0.0
@@ -118,17 +119,17 @@ private[events] object DatapointConverter {
     }
   }
 
-  private[events] def addNaN(value: AtomicDouble, amount: Double): Unit = {
+  private[events] def addNaN(now: Long, value: StepDouble, amount: Double): Unit = {
     if (amount.isNaN)
       return
 
     var set = false
     while (!set) {
-      val v = value.get()
+      val v = value.getCurrent(now)
       if (v.isNaN) {
-        set = value.compareAndSet(v, amount)
+        set = value.compareAndSet(now, v, amount)
       } else {
-        value.addAndGet(amount)
+        value.addAndGet(now, amount)
         set = true
       }
     }
@@ -154,7 +155,7 @@ private[events] object DatapointConverter {
 
     override def update(value: Double): Unit = {
       if (value.isFinite && value >= 0.0) {
-        addNaN(buffer.getCurrent, value)
+        addNaN(params.clock.wallTime(), buffer, value)
       }
     }
 
@@ -169,7 +170,7 @@ private[events] object DatapointConverter {
 
     override def hasNoData: Boolean = {
       val now = params.clock.wallTime()
-      buffer.getCurrent(now).get().isNaN && buffer.poll(now).isNaN
+      buffer.getCurrent(now).isNaN && buffer.poll(now).isNaN
     }
   }
 
@@ -183,7 +184,7 @@ private[events] object DatapointConverter {
     }
 
     override def update(value: Double): Unit = {
-      addNaN(buffer.getCurrent, 1.0)
+      addNaN(params.clock.wallTime(), buffer, 1.0)
     }
 
     override def flush(timestamp: Long): Unit = {
@@ -197,7 +198,7 @@ private[events] object DatapointConverter {
 
     override def hasNoData: Boolean = {
       val now = params.clock.wallTime()
-      buffer.getCurrent(now).get().isNaN && buffer.poll(now).isNaN
+      buffer.getCurrent(now).isNaN && buffer.poll(now).isNaN
     }
   }
 
@@ -211,9 +212,7 @@ private[events] object DatapointConverter {
     }
 
     override def update(value: Double): Unit = {
-      if (value.isFinite) {
-        buffer.getCurrent.max(value)
-      }
+      buffer.max(params.clock.wallTime(), value)
     }
 
     override def flush(timestamp: Long): Unit = {
@@ -227,7 +226,7 @@ private[events] object DatapointConverter {
 
     override def hasNoData: Boolean = {
       val now = params.clock.wallTime()
-      buffer.getCurrent(now).get().isNaN && buffer.poll(now).isNaN
+      buffer.getCurrent(now).isNaN && buffer.poll(now).isNaN
     }
   }
 
@@ -241,9 +240,7 @@ private[events] object DatapointConverter {
     }
 
     override def update(value: Double): Unit = {
-      if (value.isFinite) {
-        min(buffer.getCurrent, value)
-      }
+      buffer.min(params.clock.wallTime(), value)
     }
 
     private def min(current: AtomicDouble, value: Double): Unit = {
@@ -270,7 +267,7 @@ private[events] object DatapointConverter {
 
     override def hasNoData: Boolean = {
       val now = params.clock.wallTime()
-      buffer.getCurrent(now).get().isNaN && buffer.poll(now).isNaN
+      buffer.getCurrent(now).isNaN && buffer.poll(now).isNaN
     }
   }
 

atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/StreamMetadata.scala

@@ -38,16 +38,17 @@ import com.netflix.spectator.impl.StepLong
 case class StreamMetadata(
   streamId: String,
   remoteAddress: String = "unknown",
+  clock: Clock = Clock.SYSTEM,
   receivedMessages: StepLong = new StepLong(0, Clock.SYSTEM, 60_000),
   droppedMessages: StepLong = new StepLong(0, Clock.SYSTEM, 60_000)
 ) extends JsonSupport {
 
   def updateReceived(n: Int): Unit = {
-    receivedMessages.getCurrent.addAndGet(n)
+    receivedMessages.addAndGet(clock.wallTime(), n)
   }
 
   def updateDropped(n: Int): Unit = {
-    droppedMessages.getCurrent.addAndGet(n)
+    droppedMessages.addAndGet(clock.wallTime(), n)
   }
 
   override def hasCustomEncoding: Boolean = true

atlas-lwcapi/src/test/scala/com/netflix/atlas/lwcapi/StreamMetadataSuite.scala

@@ -25,7 +25,13 @@ class StreamMetadataSuite extends FunSuite {
     val clock = new ManualClock()
     val step = 60_000L
     val meta =
-      StreamMetadata("id", "addr", new StepLong(0, clock, step), new StepLong(0, clock, step))
+      StreamMetadata(
+        "id",
+        "addr",
+        clock,
+        new StepLong(0, clock, step),
+        new StepLong(0, clock, step)
+      )
     meta.updateReceived(100)
     meta.updateDropped(2)
 

atlas-lwcapi/src/test/scala/com/netflix/atlas/lwcapi/SubscriptionManagerSuite.scala

@@ -252,7 +252,7 @@ class SubscriptionManagerSuite extends FunSuite {
 
     val sm = new SubscriptionManager[Integer](registry)
     val meta =
-      StreamMetadata("a", "test", new StepLong(0, clock, step), new StepLong(0, clock, step))
+      StreamMetadata("a", "test", clock, new StepLong(0, clock, step), new StepLong(0, clock, step))
     assert(sm.register(meta, 1))
 
     val ok = Id.create("atlas.lwcapi.currentStreams").withTag("state", "ok")

atlas-pekko/src/main/scala/com/netflix/atlas/pekko/RequestHandler.scala

@@ -206,7 +206,7 @@ object RequestHandler extends StrictLogging {
       case e: NoSuchElementException =>
         DiagnosticMessage.error(StatusCodes.NotFound, e)
       case e: EntityStreamSizeException =>
-        DiagnosticMessage.error(StatusCodes.PayloadTooLarge, e)
+        DiagnosticMessage.error(StatusCodes.ContentTooLarge, e)
       case e: CircuitBreakerOpenException =>
         DiagnosticMessage.error(StatusCodes.ServiceUnavailable, e)
       case e: Throwable =>

project/Dependencies.scala

@@ -4,17 +4,17 @@ import sbt._
 
 object Dependencies {
   object Versions {
-    val pekko       = "1.0.3"
-    val pekkoHttpV  = "1.0.1"
-    val iep         = "5.0.26"
-    val jackson     = "2.17.2"
-    val log4j       = "2.23.1"
-    val scala       = "2.13.14"
+    val pekko       = "1.1.2"
+    val pekkoHttpV  = "1.1.0"
+    val iep         = "5.0.27"
+    val jackson     = "2.18.0"
+    val log4j       = "2.24.1"
+    val scala       = "2.13.15"
     val slf4j       = "2.0.16"
-    val spectator   = "1.7.19"
-    val spring      = "6.1.12"
+    val spectator   = "1.8.0"
+    val spring      = "6.1.13"
 
-    val crossScala  = Seq(scala, "3.4.1")
+    val crossScala  = Seq(scala, "3.5.1")
   }
 
   import Versions._
@@ -28,8 +28,8 @@ object Dependencies {
   val pekkoStreamTestkit= "org.apache.pekko" %% "pekko-stream-testkit" % pekko
   val pekkoTestkit      = "org.apache.pekko" %% "pekko-testkit" % pekko
   val caffeine          = "com.github.ben-manes.caffeine" % "caffeine" % "3.1.8"
-  val datasketches      = "org.apache.datasketches" % "datasketches-java" % "6.0.0"
-  val equalsVerifier    = "nl.jqno.equalsverifier" % "equalsverifier" % "3.16.1"
+  val datasketches      = "org.apache.datasketches" % "datasketches-java" % "6.1.1"
+  val equalsVerifier    = "nl.jqno.equalsverifier" % "equalsverifier" % "3.17.1"
   val hikariCP          = "com.zaxxer" % "HikariCP" % "5.1.0"
   val iepLeaderApi      = "com.netflix.iep" % "iep-leader-api" % iep
   val iepLeaderDynamoDb = "com.netflix.iep" % "iep-leader-dynamodb" % iep
@@ -51,10 +51,10 @@ object Dependencies {
   val log4jJcl          = "org.apache.logging.log4j" % "log4j-jcl" % log4j
   val log4jJul          = "org.apache.logging.log4j" % "log4j-jul" % log4j
   val log4jSlf4j        = "org.apache.logging.log4j" % "log4j-slf4j-impl" % log4j
-  val munit             = "org.scalameta" %% "munit" % "1.0.1"
+  val munit             = "org.scalameta" %% "munit" % "1.0.2"
   val postgres          = "org.postgresql" % "postgresql" % "42.7.3"
   val postgresEmbedded  = "io.zonky.test" % "embedded-postgres" % "2.0.7"
-  val roaringBitmap     = "org.roaringbitmap" % "RoaringBitmap" % "1.2.1"
+  val roaringBitmap     = "org.roaringbitmap" % "RoaringBitmap" % "1.3.0"
   val scalaCompat       = "org.scala-lang.modules" %% "scala-collection-compat" % "2.12.0"
   val scalaCompatJdk8   = "org.scala-lang.modules" %% "scala-java8-compat" % "1.0.2"
   val scalaCompiler     = "org.scala-lang" % "scala-compiler"

project/build.properties

@@ -1 +1 @@
-sbt.version=1.10.1
+sbt.version=1.10.2