BufferedTokenizer: size limit check in flush and logging when drop data (#19312)

Updates the BufferedTokenizer flush method to raise an error if the remainder of the token is oversized, respect to the size limit.
Adds a warning log when data is dropped in the accumulation phase.

Co-authored-by: Álex Cámara Lara <alex.camara@elastic.co>

Author: andsel

logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java

@@ -20,6 +20,8 @@
 
 import java.util.Iterator;
 import java.util.NoSuchElementException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 public class BufferedTokenizer {
 
@@ -40,12 +42,15 @@ public boolean hasNext() {
     }
 
     static class DataSplitter implements Iterator<String> {
+        private static final Logger logger = LogManager.getLogger(BufferedTokenizer.class);
+
         private final String separator;
         private int currentIdx = 0;
         private int nextSeparatorIdx = -1;
         private final StringBuilder accumulator = new StringBuilder();
         private final int sizeLimit;
         private int lastFragmentSize = 0;
+        private long droppedBytesCount = 0;
 
         DataSplitter(String separator) {
             this.separator = separator;
@@ -109,10 +114,17 @@ public void append(String data) {
             if (isSizeLimitSet()) {
                 if (!data.contains(separator) && lastFragmentSize > sizeLimit) {
                     // stop accumulating if last fragments already reached the sizeLimit
+                    droppedBytesCount += data.length();
                     return;
                 }
 
-                // we know that data contains at least one separator or that we haven't yet reached the first separator instance, update lastFragmentSize
+                if (droppedBytesCount > 0) {
+                    logger.warn("Input buffer exceeded the sizeLimit of {} and dropped {} bytes from an oversized token", sizeLimit, droppedBytesCount);
+                    droppedBytesCount = 0;
+                }
+
+                // we know that data contains at least one separator or that we haven't yet reached
+                // the first separator instance, update lastFragmentSize
                 int lastSeparatorIdx = data.lastIndexOf(separator);
                 if (lastSeparatorIdx == -1) {
                     lastFragmentSize += data.length();
@@ -128,11 +140,27 @@ private boolean isSizeLimitSet() {
         }
 
         public String flush() {
+            if (isSizeLimitSet()) {
+                if (droppedBytesCount > 0) {
+                    logger.warn("Input buffer exceeded the sizeLimit of {} and dropped {} bytes from an oversized token", sizeLimit,  droppedBytesCount);
+                }
+                if (lastFragmentSize > sizeLimit) {
+                    cleanAccumulatorState();
+                    throw new IllegalStateException("input buffer full, consumed token which exceeded the sizeLimit " + sizeLimit);
+                }
+            }
             final String flushed = accumulator.substring(currentIdx);
+            // empty the accumulator
+            cleanAccumulatorState();
+            return flushed;
+        }
+
+        private void cleanAccumulatorState() {
             // empty the accumulator
             accumulator.setLength(0);
             currentIdx = 0;
-            return flushed;
+            lastFragmentSize = 0;
+            droppedBytesCount = 0;
         }
 
         // considered empty if caught up to the accumulator

logstash-core/src/test/java/org/logstash/common/BufferedTokenizerDroppedDataLoggingTest.java

@@ -0,0 +1,166 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.logstash.common;
+
+import org.apache.logging.log4j.core.test.appender.ListAppender;
+import org.apache.logging.log4j.core.test.junit.LoggerContextRule;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Verifies that BufferedTokenizer logs a WARN message with the dropped byte count whenever
+ * buffered data is discarded due to the sizeLimit being exceeded.
+ *
+ * Two trigger points are tested:
+ *  1. When a separator arrives after a sequence of dropped fragments (recovery in append).
+ *  2. When flush() is called while dropped data is outstanding.
+ */
+public final class BufferedTokenizerDroppedDataLoggingTest {
+
+    private static final String CONFIG = "log4j2-test1.xml";
+
+    @ClassRule
+    public static LoggerContextRule CTX = new LoggerContextRule(CONFIG);
+
+    private ListAppender appender;
+    private BufferedTokenizer sut;
+
+    @Before
+    public void setUp() {
+        appender = CTX.getListAppender("EventLogger").clear();
+        sut = new BufferedTokenizer("\n", 10);
+    }
+
+    @Test
+    public void givenDroppedFragmentsWhenSeparatorArrivesInAppendThenWarnIsLoggedWithDroppedByteCount() {
+        // "01234567890" (11 chars) — NOT dropped because lastFragmentSize starts at 0
+        sut.extract("01234567890");
+
+        // "AAAAA" (5 chars, no sep) — dropped: lastFragmentSize(11) > sizeLimit(10)
+        sut.extract("AAAAA");
+
+        // "BBBBB" (5 chars, no sep) — dropped: still 11 > 10
+        sut.extract("BBBBB");
+
+        // No warn yet — separator hasn't arrived
+        assertTrue("No warning should be emitted before a separator is seen",
+                warnMessages().stream().noneMatch(m -> m.contains("dropped")));
+
+        // "\nCC" contains a separator → recovery triggers the warn with 10 dropped bytes
+        sut.extract("\nCC");
+
+        assertEquals(1, warnMessages().size());
+        assertThat("Warning must report 10 dropped bytes (AAAAA + BBBBB)", first(warnMessages()), containsString("dropped 10 bytes"));
+    }
+
+    @Test
+    public void givenMultipleBatchesOfDroppedDataWhenSeparatorArrivesRepeatedly_ThenEachBatchIsLoggedSeparately() {
+        // First overrun: 11 chars, no sep → accumulated (lastFragmentSize = 11)
+        sut.extract("01234567890");
+
+        // Drop 3 bytes
+        sut.extract("AAA");
+        // Recovery: separator seen → warn("3 bytes dropped")
+        sut.extract("\n");
+
+        assertEquals(1, warnMessages().size());
+        assertThat("First warn should report 3 dropped bytes", first(warnMessages()), containsString("dropped 3 bytes"));
+
+        // Start a new overrun: 11 chars again → accumulated on top of existing content
+        sut.extract("01234567890");
+        // Drop 7 bytes
+        sut.extract("BBBBBBB");
+        // Recovery: separator seen → warn("7 bytes dropped")
+        sut.extract("\n");
+
+        assertEquals(2, warnMessages().size());
+        assertThat("Second warn should report 7 dropped bytes", last(warnMessages()), containsString("dropped 7 bytes"));
+    }
+    
+    @Test
+    public void givenDroppedFragmentsWhenFlushIsInvokedThenWarnIsLoggedWithDroppedByteCountBeforeThrowing() {
+        // "01234567890" (11 chars) — accumulated, lastFragmentSize = 11
+        sut.extract("01234567890");
+
+        // Drop 4 bytes
+        sut.extract("CCCC");
+        // Drop 6 bytes
+        sut.extract("DDDDDD");
+
+        // No warn yet — separator hasn't arrived and flush not called
+        assertTrue("No warning before flush", warnMessages().stream().noneMatch(m -> m.contains("dropped")));
+
+        // flush() must warn about dropped data then throw for the overrun partial token
+        assertThrows(IllegalStateException.class, () -> sut.flush());
+
+        assertEquals(1, warnMessages().size());
+        assertThat("Warning must report 10 dropped bytes (CCCC + DDDDDD)", first(warnMessages()), containsString("dropped 10 bytes"));
+    }
+
+    @Test
+    public void givenDroppedFragmentsWhenFlushIsInvokedThenWarnPrecedesTheException() {
+        sut.extract("01234567890"); // accumulated, lastFragmentSize = 11
+        sut.extract("EEE");        // dropped (3 bytes)
+
+        assertThrows(IllegalStateException.class, () -> sut.flush());
+
+        // The warn must have been emitted (before the exception propagated)
+        assertEquals(1, warnMessages().size());
+        assertThat("Dropped-data warn must be logged even when flush throws", first(warnMessages()), containsString("dropped 3 bytes"));
+    }
+
+    @Test
+    public void givenNoDroppedDataWhenFlushIsInvokedThenNoWarnIsLogged() {
+        sut.extract("short");
+        sut.flush();
+
+        assertTrue("No dropped-data warn should appear when nothing was dropped",
+                warnMessages().stream().noneMatch(m -> m.contains("dropped")));
+    }
+
+    @Test
+    public void givenNoDroppedDataWhenSeparatorArrivesNoWarnIsLogged() {
+        sut.extract("hello\nworld");
+
+        assertTrue("No dropped-data warn should appear for normal tokenization",
+                warnMessages().stream().noneMatch(m -> m.contains("dropped")));
+    }
+
+    private List<String> warnMessages() {
+        return appender.getMessages();
+    }
+
+    private static String first(List<String> list) {
+        return list.get(0);
+    }
+
+    private static String last(List<String> list) {
+        return list.get(list.size() - 1);
+    }
+}

logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java

@@ -25,12 +25,15 @@
 
 import java.util.Iterator;
 import java.util.List;
+import java.util.NoSuchElementException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.emptyOrNullString;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.lessThan;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.logstash.common.BufferedTokenizerTest.toList;
@@ -47,7 +50,32 @@ public void setUp() {
     private void initSUTWithSizeLimit(int sizeLimit) {
         sut = new BufferedTokenizer("\n", sizeLimit);
     }
-
+    
+    @Test
+    public void givenOversizedFragmentWithoutSeparatorWhenFlushIsInvokedThenThrows() {
+        // Provide an overrun fragment without delimiter, from this point on 
+        // the BufferedTokenizer start dropping data because already passed the 
+        // sizeLimit.
+        Iterable<String> it = sut.extract("01234567890");
+        Iterator<String> ite = it.iterator();
+        verifyNoTokensAvailableOnReadSide(ite);
+
+        // Provide another fragment which is inside the sizeLimit and DO NOT contain a delimiter.
+        // Reuse the previous iterator, it's the same returned by this call.
+        sut.extract("AAAAA");
+        verifyNoTokensAvailableOnReadSide(ite);
+        
+        // Exercise flush and expect it throws an exception for the overrun partial token
+        Exception thrownException = assertThrows(IllegalStateException.class, () -> sut.flush());
+        assertThat(thrownException.getMessage(), containsString("input buffer full"));
+    }
+    
+    private void verifyNoTokensAvailableOnReadSide(Iterator<String> ite) {
+        assertFalse(ite.hasNext());
+        Exception thrownException = assertThrows(NoSuchElementException.class, ite::next);
+        assertThat(thrownException.getMessage(), is(emptyOrNullString()));
+    }
+ 
     @Test
     public void givenTokenWithinSizeLimitWhenExtractedThenReturnTokens() {
         List<String> tokens = toList(sut.extract("foo\nbar\n"));
@@ -146,7 +174,8 @@ public void givenSequenceOfFragmentsWithoutSeparatorThenDoesntGenerateOutOfMemor
 
         // with the second fragment passed to extract it overrun the sizeLimit, the tokenizer
         // drop starting from the third fragment
-        assertThat("Accumulator include only a part of an exploding payload", sut.flush().length(), is(lessThan(neverEndingData.length() * 3)));
+        Exception thrownException = assertThrows(IllegalStateException.class, () -> sut.flush());
+        assertThat(thrownException.getMessage(), containsString("input buffer full"));
 
         Iterable<String> tokensIterable = sut.extract("\nbbb\n");
         Iterator<String> tokensIterator = tokensIterable.iterator();