Skip to content

Commit ffbaca4

Browse files
authored
Adding wk size to message size calculation along with a buffer (#476)
* Adding wk size to message size calculation along with a buffer
1 parent d3e0d86 commit ffbaca4

File tree

2 files changed

+14
-19
lines changed

2 files changed

+14
-19
lines changed

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

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -154,7 +154,8 @@ public int messageSizeInBytes(Message message) {
154154
private Boolean isBackPressuredAfterSize(int incomingSize) {
155155
int POISON_BYTE_SIZE = messageSizeInBytes(FlushMessage.POISON);
156156
int sizeAfterAdd = this.currentQueueSizeInBytes + incomingSize + POISON_BYTE_SIZE;
157-
return sizeAfterAdd >= Math.min(this.maximumQueueByteSize, BATCH_MAX_SIZE);
157+
// Leave a 10% buffer since the unsynchronized enqueue could add multiple at a time
158+
return sizeAfterAdd >= Math.min(this.maximumQueueByteSize, BATCH_MAX_SIZE) * 0.9;
158159
}
159160

160161
public boolean offer(Message message) {
@@ -176,8 +177,6 @@ public void enqueue(Message message) {
176177
// @jorgen25 check if message is below 32kb limit for individual messages, no need to check
177178
// for extra characters
178179
if (messageByteSize <= MSG_MAX_SIZE) {
179-
// messageQueue.put(message);
180-
181180
if (isBackPressuredAfterSize(messageByteSize)) {
182181
this.currentQueueSizeInBytes = messageByteSize;
183182
messageQueue.put(FlushMessage.POISON);
@@ -443,13 +442,13 @@ public static class BatchUtility {
443442
* "userId":"jorgen25","integrations":{"someKey":{"data":"aaaaa"}},"previousId":"foo"},{"type":"alias",
444443
* "messageId":"57b0ceb4-a1cf-4599-9fba-0a44c7041004","timestamp":"Nov 18, 2021, 2:45:07 PM",
445444
* "userId":"jorgen25","integrations":{"someKey":{"data":"aaaaa"}},"previousId":"foo"}],
446-
* "sentAt":"Nov 18, 2021, 2:45:07
447-
* PM","context":{"library":{"name":"analytics-java","version":"3.1.3"}},"sequence":1}
445+
* "sentAt":"Nov 18, 2021, 2:45:07 PM","context":{"library":{"name":"analytics-java",
446+
* "version":"3.1.3"}},"sequence":1,"writeKey":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"}
448447
*
449-
* <p>total size of batch : 886
448+
* <p>total size of batch : 932
450449
*
451450
* <p>BREAKDOWN: {"batch":[MESSAGE1,MESSAGE2,MESSAGE3,MESSAGE4],"sentAt":"MMM dd, yyyy, HH:mm:ss
452-
* tt","context":CONTEXT,"sequence":1}
451+
* tt","context":CONTEXT,"sequence":1,"writeKey":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"}
453452
*
454453
* <p>so we need to account for: 1 -message size: 189 * 4 = 756 2 -context object size = 55 in
455454
* this sample -> 756 + 55 = 811 3 -Metadata (This has the sent data/sequence characters) +
@@ -463,12 +462,14 @@ public static class BatchUtility {
463462
* <p>so formulae to determine the expected default size of the batch is
464463
*
465464
* @return: defaultSize = messages size + context size + metadata size + comma number + sequence
466-
* digits
465+
* digits + writekey + buffer
467466
* @return
468467
*/
469468
private static int getBatchDefaultSize(int contextSize, int currentMessageNumber) {
470-
// sample data: {"batch":[],"sentAt":"MMM dd, yyyy, HH:mm:ss tt","context":,"sequence":1} - 73
471-
int metadataExtraCharsSize = 73;
469+
// sample data: {"batch":[],"sentAt":"MMM dd, yyyy, HH:mm:ss tt","context":,"sequence":1,
470+
// "writeKey":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"} - 119
471+
// Don't need to squeeze everything possible into a batch, adding a buffer
472+
int metadataExtraCharsSize = 119 + 1024;
472473
int commaNumber = currentMessageNumber - 1;
473474

474475
return contextSize

analytics/src/test/java/com/segment/analytics/internal/AnalyticsClientTest.java

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -875,7 +875,7 @@ public void submitBatchBelowThreshold() throws InterruptedException, IllegalArgu
875875
new Gson());
876876

877877
Map<String, String> properties = new HashMap<String, String>();
878-
properties.put("property3", generateDataOfSizeSpecialChars(MAX_MSG_SIZE, true));
878+
properties.put("property3", generateDataOfSizeSpecialChars(((int) (MAX_MSG_SIZE * 0.9)), true));
879879

880880
for (int i = 0; i < 15; i++) {
881881
TrackMessage bigMessage =
@@ -929,13 +929,7 @@ public void submitBatchAboveThreshold() throws InterruptedException, IllegalArgu
929929
client.shutdown();
930930
while (!isShutDown.get()) {}
931931

932-
/**
933-
* modified from expected 8 to expected 7 times, since we removed the inner loop. The inner loop
934-
* was forcing to message list created from the queue to keep making batches even if its a 1
935-
* message batch until the message list is empty, that was forcing the code to make one last
936-
* batch of 1 msg in size bumping the number of times a batch would be submitted from 7 to 8
937-
*/
938-
verify(networkExecutor, times(7)).submit(any(Runnable.class));
932+
verify(networkExecutor, times(8)).submit(any(Runnable.class));
939933
}
940934

941935
@Test
@@ -970,6 +964,6 @@ public void submitManySmallMessagesBatchAboveThreshold() throws InterruptedExcep
970964
client.shutdown();
971965
while (!isShutDown.get()) {}
972966

973-
verify(networkExecutor, times(19)).submit(any(Runnable.class));
967+
verify(networkExecutor, times(21)).submit(any(Runnable.class));
974968
}
975969
}

0 commit comments

Comments
 (0)