Fix bug in CompressionUtils that fails with certain delimiters (#154)

* Add logging to SinkClient to track uncleaned connections * Only count resubmission from worker failures * Fix CompressionUtil bug with certain delimiter * fix annotation * update test * spotless Co-authored-by: Calvin Cheung <ccheung@netflix.com>
Netflix · Mar 25, 2022 · da92977 · da92977
1 parent cc66c61
commit da92977
Show file tree

Hide file tree

Showing 2 changed files with 72 additions and 25 deletions.
diff --git a/mantis-common/src/main/java/io/mantisrx/common/compression/CompressionUtils.java b/mantis-common/src/main/java/io/mantisrx/common/compression/CompressionUtils.java
@@ -217,10 +217,19 @@ static List<MantisServerSentEvent> tokenize(BufferedReader bf, String delimiter)
             for (int i = 0; i < line.length(); i++) {
                 if (line.charAt(i) != delimiterArray[delimiterCount]) {
                     if (delimiterCount > 0) {
-                        for (int j = 0; j < delimiterCount; ++j) {
-                            sb.append(delimiterArray[j]);
+                        boolean prefixMatch = true;
+                        for (int j = delimiterCount - 1; j >= 0; j--) {
+                            if (line.charAt(i) != delimiterArray[j]) {
+                                prefixMatch = false;
+                                break;
+                            }
+                        }
+                        if (!prefixMatch) {
+                            for (int j = 0; j < delimiterCount; ++j) {
+                                sb.append(delimiterArray[j]);
+                            }
+                            delimiterCount = 0;
                         }
-                        delimiterCount = 0;
                     }
                     if (line.charAt(i) != delimiterArray[delimiterCount]) {
                         sb.append(line.charAt(i));

diff --git a/mantis-common/src/test/java/io/mantisrx/common/compression/CompressionUtilsTest.java b/mantis-common/src/test/java/io/mantisrx/common/compression/CompressionUtilsTest.java
@@ -19,31 +19,29 @@
 
 import io.mantisrx.common.MantisServerSentEvent;
 import java.io.BufferedReader;
-import java.io.IOException;
 import java.io.StringReader;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
-import org.junit.Assert;
+import java.util.stream.Collectors;
 import org.junit.Test;
 
 public class CompressionUtilsTest {
 
-    @Test public void shouldTokenizeWithEventsContainingPartialDelimiterMatches() {
+    @Test
+    public void shouldTokenizeWithEventsContainingPartialDelimiterMatches() throws Exception {
         String testInput = "ab$cdef$$$ghi$jkl$$$lmno$$pqrst$";
-        try (BufferedReader reader = new BufferedReader(new StringReader(testInput))) {
-            List<MantisServerSentEvent> result = CompressionUtils.tokenize(reader);
-
-            assertEquals(result.size(), 3);
-            assertEquals(result.get(0).getEventAsString(), "ab$cdef");
-            assertEquals(result.get(1).getEventAsString(), "ghi$jkl");
-            assertEquals(result.get(2).getEventAsString(), "lmno$$pqrst$");
+        BufferedReader reader = new BufferedReader(new StringReader(testInput));
+        List<MantisServerSentEvent> result = CompressionUtils.tokenize(reader);
 
-        } catch (IOException ex) {
-            Assert.fail("Tokenization threw an IO exception that was unexpected");
-        }
+        assertEquals(result.size(), 3);
+        assertEquals(result.get(0).getEventAsString(), "ab$cdef");
+        assertEquals(result.get(1).getEventAsString(), "ghi$jkl");
+        assertEquals(result.get(2).getEventAsString(), "lmno$$pqrst$");
     }
 
-    @Test public void shouldTokenizeWithEventsContainingPartialDelimiterMatchesWithCustomDelimiter() {
+    @Test
+    public void shouldTokenizeWithEventsContainingPartialDelimiterMatchesWithCustomDelimiter() throws Exception {
         String delimiter = "a04f0418-bdff-4f53-af7d-9f5a093b9d65";
 
         String event1 = "ab" + delimiter.substring(0, 9) + "cdef";
@@ -54,17 +52,57 @@ public class CompressionUtilsTest {
                 + event2
                 + delimiter
                 + event3;
-        try (BufferedReader reader = new BufferedReader(new StringReader(testInput))) {
-            List<MantisServerSentEvent> result = CompressionUtils.tokenize(reader, delimiter);
+        BufferedReader reader = new BufferedReader(new StringReader(testInput));
+        List<MantisServerSentEvent> result = CompressionUtils.tokenize(reader, delimiter);
+
+        List<String> actual = result.stream().map(e -> e.getEventAsString()).collect(Collectors.toList());
+        assertEquals("Delimiter: '" + delimiter + "'", Arrays.asList(event1,event2,event3), actual);
+    }
+
+    @Test
+    public void testDelimiterWiithPrefixMatchingEndOfMEssage() throws Exception {
+        // Delimiter starts with 'c', event1 ends with 'c'
+        String delimiter = "ccd";
+
+        String event1 = "abc";
+        String event2 = "def";
+        String event3 = "ghi";
+        String testInput = event1
+                + delimiter
+                + event2
+                + delimiter
+                + event3;
+        BufferedReader reader = new BufferedReader(new StringReader(testInput));
+        List<MantisServerSentEvent> result = CompressionUtils.tokenize(reader, delimiter);
+
+        List<String> actual = result.stream().map(e -> e.getEventAsString()).collect(Collectors.toList());
+        assertEquals("Delimiter: '" + delimiter + "'", Arrays.asList(event1,event2,event3), actual);
+    }
 
-            assertEquals("Delimiter: '" + delimiter + "'", result.size(), 3);
-            assertEquals(result.get(0).getEventAsString(), event1);
-            assertEquals(result.get(1).getEventAsString(), event2);
-            assertEquals(result.get(2).getEventAsString(), event3);
+    @Test
+    public void testMultiline() throws Exception {
+        String delimiter = "ccd";
 
-        } catch (IOException ex) {
-            Assert.fail("Tokenization threw an IO exception that was unexpected");
+        String event1 = "abc";
+        String event2 = "def";
+        String event3 = "ghi";
+        StringBuffer buf = new StringBuffer();
+        String testInput = event1
+                + delimiter
+                + event2
+                + delimiter
+                + event3;
+        // Turn input into 1 character per line
+        for (int i = 0; i < testInput.length(); i++) {
+            buf.append(testInput.charAt(i)).append("\n");
         }
+        testInput = buf.toString();
+
+        BufferedReader reader = new BufferedReader(new StringReader(testInput));
+        List<MantisServerSentEvent> result = CompressionUtils.tokenize(reader, delimiter);
+
+        List<String> actual = result.stream().map(e -> e.getEventAsString()).collect(Collectors.toList());
+        assertEquals("Delimiter: '" + delimiter + "'", Arrays.asList(event1,event2,event3), actual);
     }
 
     @Test