avoid unnecessary string conversion

atlas-eval/src/main/scala/com/netflix/atlas/eval/model/LwcMessages.scala

@@ -18,6 +18,7 @@ package com.netflix.atlas.eval.model
 import akka.util.ByteString
 import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.core.JsonToken
+import com.netflix.atlas.akka.ByteStringInputStream
 import com.netflix.atlas.akka.DiagnosticMessage
 import com.netflix.atlas.core.util.SmallHashMap
 import com.netflix.atlas.json.Json
@@ -29,14 +30,24 @@ import com.netflix.atlas.json.JsonSupport
   */
 object LwcMessages {
 
+  /**
+    * Parse the message string into an internal model object based on the type.
+    */
+  def parse(msg: ByteString): AnyRef = {
+    parse(Json.newJsonParser(new ByteStringInputStream(msg)))
+  }
+
   /**
     * Parse the message string into an internal model object based on the type.
     */
   def parse(msg: String): AnyRef = {
+    parse(Json.newJsonParser(msg))
+  }
+
+  private def parse(parser: JsonParser): AnyRef = {
     // This is a performance critical part of the code so the parsing is done by
     // hand rather than using ObjectMapper to minimize allocations and get peak
     // performance.
-    val parser = Json.newJsonParser(msg)
     try {
       // All
       var typeDesc: String = null

atlas-eval/src/main/scala/com/netflix/atlas/eval/stream/LwcToAggrDatapoint.scala

@@ -60,7 +60,7 @@ private[stream] class LwcToAggrDatapoint(context: StreamContext)
       override def onPush(): Unit = {
         val message = grab(in)
         try {
-          val parsedMsg = LwcMessages.parse(message.utf8String)
+          val parsedMsg = LwcMessages.parse(message)
           parsedMsg match {
             case sb: LwcSubscription      => updateState(sb)
             case dp: LwcDatapoint         => pushDatapoint(dp)

atlas-jmh/src/main/scala/com/netflix/atlas/eval/model/LwcMessagesParse.scala

@@ -15,6 +15,7 @@
  */
 package com.netflix.atlas.eval.model
 
+import akka.util.ByteString
 import com.netflix.atlas.json.Json
 import org.openjdk.jmh.annotations.Benchmark
 import org.openjdk.jmh.annotations.Scope
@@ -26,12 +27,22 @@ import org.openjdk.jmh.infra.Blackhole
   * > jmh:run -prof gc -wi 10 -i 10 -f1 -t1 .*LwcMessagesParse.*
   * ```
   *
-  * Results:
+  * Throughput:
   *
   * ```
-  * Benchmark                          Mode  Cnt        Score       Error   Units
+  * Benchmark                          Mode  Cnt        Score        Error   Units
   * parseDatapoint                    thrpt    5  1228277.874 ± 43946.724   ops/s
-  * parseDatapoint       gc.alloc.rate.norm    5     1632.000 ±     0.001    B/op
+  * parseDatapointByteString          thrpt    5  1381682.708 ±  74560.401   ops/s
+  * parseDatapointByteStringUTF8      thrpt    5  1019061.212 ±  40733.781   ops/s
+  * ```
+  *
+  * Allocations:
+  *
+  * ```
+  * Benchmark                          Mode  Cnt        Score        Error   Units
+  * parseDatapoint                    alloc    5     1632.000 ±      0.001    B/op
+  * parseDatapointByteString          alloc    5     1816.000 ±      0.001    B/op
+  * parseDatapointByteStringUTF8      alloc    5     2128.000 ±      0.001    B/op
   * ```
   **/
 @State(Scope.Thread)
@@ -47,9 +58,20 @@ class LwcMessagesParse {
 
   private val datapoint = LwcDatapoint(1234567890L, "i-12345", tags, 42.0)
   private val json = Json.encode(datapoint)
+  private val bytes = ByteString(json)
 
   @Benchmark
   def parseDatapoint(bh: Blackhole): Unit = {
     bh.consume(LwcMessages.parse(json))
   }
+
+  @Benchmark
+  def parseDatapointByteString(bh: Blackhole): Unit = {
+    bh.consume(LwcMessages.parse(bytes))
+  }
+
+  @Benchmark
+  def parseDatapointByteStringUTF8(bh: Blackhole): Unit = {
+    bh.consume(LwcMessages.parse(bytes.utf8String))
+  }
 }