Fixed Github Issue #524

Author: neelkanth-kaushik

analytics/src/main/java/com/segment/analytics/internal/AnalyticsClient.java

@@ -159,7 +159,8 @@ public int messageSizeInBytes(Message message) {
   private Boolean isBackPressuredAfterSize(int incomingSize) {
     int POISON_BYTE_SIZE = messageSizeInBytes(FlushMessage.POISON);
     int sizeAfterAdd = this.currentQueueSizeInBytes + incomingSize + POISON_BYTE_SIZE;
-    // Leave a 10% buffer since the unsynchronized enqueue could add multiple at a time
+    // Leave a 10% buffer since the unsynchronized enqueue could add multiple at a
+    // time
     return sizeAfterAdd >= Math.min(this.maximumQueueByteSize, BATCH_MAX_SIZE) * 0.9;
   }
 
@@ -174,12 +175,14 @@ public void enqueue(Message message) {
     }
 
     try {
-      // @jorgen25 message here could be regular msg, POISON or STOP. Only do regular logic if its
+      // @jorgen25 message here could be regular msg, POISON or STOP. Only do regular
+      // logic if its
       // valid message
       if (message != StopMessage.STOP && message != FlushMessage.POISON) {
         int messageByteSize = messageSizeInBytes(message);
 
-        // @jorgen25 check if message is below 32kb limit for individual messages, no need to check
+        // @jorgen25 check if message is below 32kb limit for individual messages, no
+        // need to check
         // for extra characters
         if (messageByteSize <= MSG_MAX_SIZE) {
           if (isBackPressuredAfterSize(messageByteSize)) {
@@ -234,7 +237,9 @@ public void shutdown() {
   }
 
   /**
-   * Wait for the looper to complete processing all messages before proceedin with shutdown. This prevents the race condition where the network executor is shut down before the looper finishes
+   * Wait for the looper to complete processing all messages before proceedin with
+   * shutdown. This prevents the race condition where the network executor is shut
+   * down before the looper finishes
    * submitting all batches.
    */
   private void waitForLooperCompletion() {
@@ -292,7 +297,8 @@ public void shutdownAndWait(ExecutorService executor, String name) {
   }
 
   /**
-   * Looper runs on a background thread and takes messages from the queue. Once it collects enough
+   * Looper runs on a background thread and takes messages from the queue. Once it
+   * collects enough
    * messages, it triggers a flush.
    */
   class Looper implements Runnable {
@@ -320,18 +326,20 @@ public void run() {
               log.print(VERBOSE, "Flushing messages.");
             }
           } else {
-            // we do  +1 because we are accounting for this new message we just took from the queue
+            // we do +1 because we are accounting for this new message we just took from the
+            // queue
             // which is not in list yet
-            // need to check if this message is going to make us go over the limit considering
+            // need to check if this message is going to make us go over the limit
+            // considering
             // default batch size as well
-            int defaultBatchSize =
-                BatchUtility.getBatchDefaultSize(contextSize, messages.size() + 1);
+            int defaultBatchSize = BatchUtility.getBatchDefaultSize(contextSize, messages.size() + 1);
             int msgSize = messageSizeInBytes(message);
             if (currentBatchSize.get() + msgSize + defaultBatchSize <= BATCH_MAX_SIZE) {
               messages.add(message);
               currentBatchSize.addAndGet(msgSize);
             } else {
-              // put message that did not make the cut this time back on the queue, we already took
+              // put message that did not make the cut this time back on the queue, we already
+              // took
               // this message if we dont put it back its lost
               // we take care of that after submitting the batch
               batchSizeLimitReached = true;
@@ -368,7 +376,8 @@ public void run() {
             currentBatchSize.set(0);
             messages.clear();
             if (batchSizeLimitReached) {
-              // If this is true that means the last message that would make us go over the limit
+              // If this is true that means the last message that would make us go over the
+              // limit
               // was not added,
               // add it to the now cleared messages list so its not lost
               messages.add(message);
@@ -385,12 +394,11 @@ public void run() {
   }
 
   static class BatchUploadTask implements Runnable {
-    private static final Backo BACKO =
-        Backo.builder() //
-            .base(TimeUnit.SECONDS, 15) //
-            .cap(TimeUnit.HOURS, 1) //
-            .jitter(1) //
-            .build();
+    private static final Backo BACKO = Backo.builder() //
+        .base(TimeUnit.SECONDS, 15) //
+        .cap(TimeUnit.HOURS, 1) //
+        .jitter(1) //
+        .build();
 
     private final AnalyticsClient client;
     private final Backo backo;
@@ -416,7 +424,10 @@ private void notifyCallbacksWithException(Batch batch, Exception exception) {
       }
     }
 
-    /** Returns {@code true} to indicate a batch should be retried. {@code false} otherwise. */
+    /**
+     * Returns {@code true} to indicate a batch should be retried. {@code false}
+     * otherwise.
+     */
     boolean upload() {
       client.log.print(VERBOSE, "Uploading batch %s.", batch.sequence());
 
@@ -469,7 +480,8 @@ public void run() {
       int attempt = 0;
       for (; attempt <= maxRetries; attempt++) {
         boolean retry = upload();
-        if (!retry) return;
+        if (!retry)
+          return;
         try {
           backo.sleep(attempt);
         } catch (InterruptedException e) {
@@ -492,45 +504,65 @@ private static boolean is5xx(int status) {
   public static class BatchUtility {
 
     /**
-     * Method to determine what is the expected default size of the batch regardless of messages
+     * Method to determine what is the expected default size of the batch regardless
+     * of messages
      *
-     * <p>Sample batch:
+     * <p>
+     * Sample batch:
      * {"batch":[{"type":"alias","messageId":"fc9198f9-d827-47fb-96c8-095bd3405d93","timestamp":"Nov
      * 18, 2021, 2:45:07
      * PM","userId":"jorgen25","integrations":{"someKey":{"data":"aaaaa"}},"previousId":"foo"},{"type":"alias",
-     * "messageId":"3ce6f88c-36cb-4991-83f8-157e10261a89","timestamp":"Nov 18, 2021, 2:45:07
+     * "messageId":"3ce6f88c-36cb-4991-83f8-157e10261a89","timestamp":"Nov 18, 2021,
+     * 2:45:07
      * PM","userId":"jorgen25",
      * "integrations":{"someKey":{"data":"aaaaa"}},"previousId":"foo"},{"type":"alias",
-     * "messageId":"a328d339-899a-4a14-9835-ec91e303ac4d","timestamp":"Nov 18, 2021, 2:45:07 PM",
+     * "messageId":"a328d339-899a-4a14-9835-ec91e303ac4d","timestamp":"Nov 18, 2021,
+     * 2:45:07 PM",
      * "userId":"jorgen25","integrations":{"someKey":{"data":"aaaaa"}},"previousId":"foo"},{"type":"alias",
-     * "messageId":"57b0ceb4-a1cf-4599-9fba-0a44c7041004","timestamp":"Nov 18, 2021, 2:45:07 PM",
+     * "messageId":"57b0ceb4-a1cf-4599-9fba-0a44c7041004","timestamp":"Nov 18, 2021,
+     * 2:45:07 PM",
      * "userId":"jorgen25","integrations":{"someKey":{"data":"aaaaa"}},"previousId":"foo"}],
-     * "sentAt":"Nov 18, 2021, 2:45:07 PM","context":{"library":{"name":"analytics-java",
+     * "sentAt":"Nov 18, 2021, 2:45:07
+     * PM","context":{"library":{"name":"analytics-java",
      * "version":"3.1.3"}},"sequence":1,"writeKey":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"}
      *
-     * <p>total size of batch : 932
+     * <p>
+     * total size of batch : 932
      *
-     * <p>BREAKDOWN: {"batch":[MESSAGE1,MESSAGE2,MESSAGE3,MESSAGE4],"sentAt":"MMM dd, yyyy, HH:mm:ss
+     * <p>
+     * BREAKDOWN: {"batch":[MESSAGE1,MESSAGE2,MESSAGE3,MESSAGE4],"sentAt":"MMM dd,
+     * yyyy, HH:mm:ss
      * tt","context":CONTEXT,"sequence":1,"writeKey":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"}
      *
-     * <p>so we need to account for: 1 -message size: 189 * 4 = 756 2 -context object size = 55 in
-     * this sample -> 756 + 55 = 811 3 -Metadata (This has the sent data/sequence characters) +
-     * extra chars (these are chars like "batch":[] or "context": etc and will be pretty much the
-     * same length in every batch -> size is 73 --> 811 + 73 = 884 (well 72 actually, char 73 is the
-     * sequence digit which we account for in point 5) 4 -Commas between each message, the total
-     * number of commas is number_of_msgs - 1 = 3 -> 884 + 3 = 887 (sample is 886 because the hour
-     * in sentData this time happens to be 2:45 but it could be 12:45 5 -Sequence Number increments
+     * <p>
+     * so we need to account for: 1 -message size: 189 * 4 = 756 2 -context object
+     * size = 55 in
+     * this sample -> 756 + 55 = 811 3 -Metadata (This has the sent data/sequence
+     * characters) +
+     * extra chars (these are chars like "batch":[] or "context": etc and will be
+     * pretty much the
+     * same length in every batch -> size is 73 --> 811 + 73 = 884 (well 72
+     * actually, char 73 is the
+     * sequence digit which we account for in point 5) 4 -Commas between each
+     * message, the total
+     * number of commas is number_of_msgs - 1 = 3 -> 884 + 3 = 887 (sample is 886
+     * because the hour
+     * in sentData this time happens to be 2:45 but it could be 12:45 5 -Sequence
+     * Number increments
      * with every batch created
      *
-     * <p>so formulae to determine the expected default size of the batch is
+     * <p>
+     * so formulae to determine the expected default size of the batch is
      *
-     * @return: defaultSize = messages size + context size + metadata size + comma number + sequence
-     *     digits + writekey + buffer
+     * @return: defaultSize = messages size + context size + metadata size + comma
+     *          number + sequence
+     *          digits + writekey + buffer
      * @return
      */
     private static int getBatchDefaultSize(int contextSize, int currentMessageNumber) {
-      // sample data: {"batch":[],"sentAt":"MMM dd, yyyy, HH:mm:ss tt","context":,"sequence":1,
-      //   "writeKey":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"} - 119
+      // sample data: {"batch":[],"sentAt":"MMM dd, yyyy, HH:mm:ss
+      // tt","context":,"sequence":1,
+      // "writeKey":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"} - 119
       // Don't need to squeeze everything possible into a batch, adding a buffer
       int metadataExtraCharsSize = 119 + 1024;
       int commaNumber = currentMessageNumber - 1;