CT-163 Optimize Event.estimatedSerializedBytes (#40)

* Change `Event.countJsonEscapedCharacters` from stream to for loop for performance
* Use Jackson `SerializedString` for Event message

Author: Oliver Hsu

src/main/java/com/scalyr/integrations/kafka/AddEventsClient.java

@@ -519,7 +519,11 @@ public void serialize(Event event, JsonGenerator jsonGenerator, SerializerProvid
      */
     private void writeEventAttrs(Event event, JsonGenerator jsonGenerator) throws IOException {
       jsonGenerator.writeObjectFieldStart("attrs");
-      jsonGenerator.writeStringField("message", event.getMessage());
+      if (event.getSerializedMessage() != null) {
+        jsonGenerator.writeFieldName("message");
+        byte[] quotedMsg = event.getSerializedMessage().asQuotedUTF8();
+        jsonGenerator.writeRawUTF8String(quotedMsg, 0, quotedMsg.length);
+      }
 
       // Write additional attrs
       if (event.getAdditionalAttrs() != null) {

src/main/java/com/scalyr/integrations/kafka/Event.java

@@ -16,14 +16,14 @@
 
 package com.scalyr.integrations.kafka;
 
+import com.fasterxml.jackson.core.io.SerializedString;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableSet;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
-import java.util.stream.IntStream;
 
 /**
  * Abstraction for a Scalyr Event.
@@ -48,7 +48,7 @@ public class Event {
 
   // Event level fields
   private long timestamp;
-  private String message;
+  private SerializedString message;
   private Map<String, Object> additionalAttrs;
 
   // Cached estimated event size
@@ -107,7 +107,7 @@ public Event setTimestamp(long timestamp) {
   }
 
   public Event setMessage(String message) {
-    this.message = message;
+    this.message = message == null ? null : new SerializedString(message);
     return this;
   }
 
@@ -170,7 +170,9 @@ public long getTimestamp() {
     return timestamp;
   }
 
-  public String getMessage() {
+  public String getMessage() { return message == null ? null : message.getValue(); }
+
+  public SerializedString getSerializedMessage() {
     return message;
   }
 
@@ -187,7 +189,7 @@ public int estimatedSerializedBytes() {
       return estimatedSizeBytes;
     }
 
-    int size = Strings.isNullOrEmpty(getMessage()) ? 0 : estimateEscapedStringSize(getMessage());
+    int size = Strings.isNullOrEmpty(getMessage()) ? 0 : getSerializedMessage().asQuotedUTF8().length;
     size += getTopic().length();
     size += EVENT_SERIALIZATION_OVERHEAD_BYTES;
 
@@ -222,11 +224,14 @@ private int estimateEscapedStringSize(String s) {
     return s.length();
   }
 
-  private int countJsonEscapedCharacters(String s) {
-    return (int)IntStream.range(0, s.length())
-      .mapToObj(s::charAt)
-      .filter(JSON_ESCAPED_CHARS::contains)
-      .count();
+  public int countJsonEscapedCharacters(String s) {
+    int escapedChars = 0;
+    for (int i = 0; i < s.length(); i++) {
+      if (JSON_ESCAPED_CHARS.contains(s.charAt(i))) {
+        escapedChars++;
+      }
+    }
+    return escapedChars;
   }
 
   /**

src/test/java/com/scalyr/integrations/kafka/AddEventsClientTest.java

@@ -121,6 +121,16 @@ public void addEventsRequestTest() throws IOException {
     addEventsRequestTest(100, 5, 3, 2);
   }
 
+  /**
+   * Verify JSON in Event message serialization
+   */
+  @Test
+  public void jsonMessageTest() throws IOException {
+    final String jsonMsg = "{\"k1\":\"v1\", \"k2\":\"v2\", \"k3\": 1.0, \"k4\": {\"k1\":\"v1\", \"k2\":\"v2\"}}";
+    List<Event> events = TestUtils.createTestEvents(10, jsonMsg, 1, 1, 1);
+    createAndVerifyAddEventsRequest(events);
+  }
+
   /**
    * Create `numEvents` with the specified `numServers`, `numLogFiles`, `numParsers`
    * and verify the AddEventsRequest is serialized correctly to JSON.

src/test/java/com/scalyr/integrations/kafka/EventBufferTest.java

@@ -62,6 +62,23 @@ public void testJsonEventMsg() throws Exception {
     assertEquals(actualSerializedBytes, estimatedSerializedBytes, actualSerializedBytes * deltaPercent);
   }
 
+  /**
+   * Test add events payload estimate with no message attribute.
+   */
+  @Test
+  public void testNullMessage() throws Exception {
+    final int numEvents = 1000;
+    EventBuffer eventBuffer = new EventBuffer();
+    TestUtils.createTestEvents(numEvents, null, 100, 1, 1)
+      .forEach(eventBuffer::addEvent);
+
+    final int estimatedSerializedBytes = eventBuffer.estimatedSerializedBytes();
+    final int actualSerializedBytes = actualSerializedSize(eventBuffer.getEvents());
+    // message field is not included in serialized payload when null, although estimates still include `message`
+    // allow bigger delta to account for this
+    assertEquals(actualSerializedBytes, estimatedSerializedBytes, actualSerializedBytes * deltaPercent * 2);
+  }
+
   /**
    * Calculate add events payload serialized size.
    */

src/test/java/com/scalyr/integrations/kafka/TestUtils.java

@@ -138,6 +138,7 @@ public static Stream<Event> createTestEventStream(int numEvents, String msg, int
           .addAdditionalAttr("app", "test")
           .addAdditionalAttr("isTest", true)
           .addAdditionalAttr("version", 2.3)
+          .addAdditionalAttr("jsonAttr", "{\"k1\":\"v1\"}")
           .setEnrichmentAttrs(TestValues.ENRICHMENT_VALUE_MAP);
       });
   }