ensure inputSize state value is reset during buftok.flush (#16760) (#16769)

(cherry picked from commit e36cacedc8)

Co-authored-by: João Duarte <jsvd@users.noreply.github.com>

logstash-core/spec/logstash/util/buftok_spec.rb

@@ -43,6 +43,35 @@ describe  FileWatch::BufferedTokenizer  do
     expect(subject.extract("\n\n\n")).to eq(["", "", ""])
   end
 
+  describe 'flush' do
+    let(:data) { "content without a delimiter" }
+    before(:each) do
+      subject.extract(data)
+    end
+
+    it "emits the contents of the buffer" do
+      expect(subject.flush).to eq(data)
+    end
+
+    it "resets the state of the buffer" do
+      subject.flush
+      expect(subject).to be_empty
+    end
+
+    context 'with decode_size_limit_bytes' do
+      subject { FileWatch::BufferedTokenizer.new("\n", 100) }
+
+      it "emits the contents of the buffer" do
+        expect(subject.flush).to eq(data)
+      end
+
+      it "resets the state of the buffer" do
+        subject.flush
+        expect(subject).to be_empty
+      end
+    end
+  end
+
   context 'with delimiter' do
     subject { FileWatch::BufferedTokenizer.new(delimiter) }
 

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

@@ -106,12 +106,13 @@ public class BufferedTokenizerExt extends RubyObject {
     public IRubyObject flush(final ThreadContext context) {
         final IRubyObject buffer = input.join(context);
         input.clear();
+        inputSize = 0;
         return buffer;
     }
 
     @JRubyMethod(name = "empty?")
     public IRubyObject isEmpty(final ThreadContext context) {
-        return input.empty_p();
+        return RubyUtil.RUBY.newBoolean(input.isEmpty() && (inputSize == 0));
     }
 
 }