Implement BufferedTokenizer to return an iterable that can verify size limit for every token emitted (#17229)

Reimplements BufferedTokenizer to leverage pure Java classes instead of use JRuby runtime's classes.

Reimplement the `BufferedTokenizerExt` in pure Java using an iterable, while the `BufferedTokenizerExt` is a shell around this new class.

The principal method `extract`, which dice the data by separator, instead of returning a `RubyArray` now return an  `Iterable` which applies the size limit check.
The returned iterable is made of  `DataSplitter` iterator which accumulates data in a `StringBuilder` and then dice the data by separator. This iterator moves the size check limit on the read side (before was on the write phase), however, it implements a logic to avoid accumulating segments when the size limit has been already passed on during the writing side.

To be compliant with some usage patterns which expect an `empty?` method to be present in the returned object from `extract`, like [this](https://github.com/logstash-plugins/logstash-input-file/blob/55a4a7099f05f29351672417036c1342850c7adc/lib/filewatch/watched_file.rb#L250), the `extract` method of the `BufferedTokenizerExt` return an `Iterable` adapter custom class with such method.

Author: andsel

logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright Elasticsearch B.V. and/or 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.benchmark;
+
+import org.logstash.common.BufferedTokenizer;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.concurrent.TimeUnit;
+
+
+@Warmup(iterations = 3, time = 100, timeUnit = TimeUnit.MILLISECONDS)
+@Measurement(iterations = 10, time = 3000, timeUnit = TimeUnit.MILLISECONDS)
+@Fork(1)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@State(Scope.Thread)
+public class BufferedTokenizerBenchmark {
+
+    private BufferedTokenizer sut;
+    private String singleTokenPerFragment;
+    private String multipleTokensPerFragment;
+    private String multipleTokensSpreadMultipleFragments_1;
+    private String multipleTokensSpreadMultipleFragments_2;
+    private String multipleTokensSpreadMultipleFragments_3;
+
+    @Setup(Level.Invocation)
+    public void setUp() {
+        sut = new BufferedTokenizer();
+        singleTokenPerFragment = "a".repeat(512) + "\n";
+
+        multipleTokensPerFragment = "a".repeat(512) + "\n" + "b".repeat(512) + "\n" + "c".repeat(512) + "\n";
+
+        multipleTokensSpreadMultipleFragments_1 = "a".repeat(512) + "\n" + "b".repeat(512) + "\n" + "c".repeat(256);
+        multipleTokensSpreadMultipleFragments_2 = "c".repeat(256) + "\n" + "d".repeat(512) + "\n" + "e".repeat(256);
+        multipleTokensSpreadMultipleFragments_3 = "f".repeat(256) + "\n" + "g".repeat(512) + "\n" + "h".repeat(512) + "\n";
+    }
+
+    @Benchmark
+    public final void onlyOneTokenPerFragment(Blackhole blackhole) {
+        Iterable<String> tokens = sut.extract(singleTokenPerFragment);
+        tokens.forEach(blackhole::consume);
+        blackhole.consume(tokens);
+    }
+
+    @Benchmark
+    public final void multipleTokenPerFragment(Blackhole blackhole) {
+        Iterable<String> tokens = sut.extract(multipleTokensPerFragment);
+        tokens.forEach(blackhole::consume);
+        blackhole.consume(tokens);
+    }
+
+    @Benchmark
+    public final void multipleTokensCrossingMultipleFragments(Blackhole blackhole) {
+        Iterable<String> tokens = sut.extract(multipleTokensSpreadMultipleFragments_1);
+        tokens.forEach(t -> {});
+        blackhole.consume(tokens);
+
+        tokens = sut.extract(multipleTokensSpreadMultipleFragments_2);
+        tokens.forEach(t -> {});
+        blackhole.consume(tokens);
+
+        tokens = sut.extract(multipleTokensSpreadMultipleFragments_3);
+        tokens.forEach(blackhole::consume);
+        blackhole.consume(tokens);
+    }
+}

logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerExtBenchmark.java

@@ -20,27 +20,37 @@
 describe  FileWatch::BufferedTokenizer  do
   subject { FileWatch::BufferedTokenizer.new }
 
+
+  # A matcher that ensures the result of BufferedTokenizer#extract "quacks like" an expected ruby Array in two respects:
+  #  - #empty? -> boolean: true indicates that the _next_ Enumerable#each will emit zero items.
+  #  - #entries -> Array: the ordered entries
+  def emit_exactly(expected_array)
+    # note: order matters; Iterator#each and the methods that delegate to it consume the iterator
+    have_attributes(:empty? => expected_array.empty?,
+                    :entries => expected_array.entries) # consumes iterator, must be done last
+  end
+
   it "should tokenize a single token" do
-    expect(subject.extract("foo\n")).to eq(["foo"])
+    expect(subject.extract("foo\n")).to emit_exactly(["foo"])
   end
 
   it "should merge multiple token" do
-    expect(subject.extract("foo")).to eq([])
-    expect(subject.extract("bar\n")).to eq(["foobar"])
+    expect(subject.extract("foo")).to emit_exactly([])
+    expect(subject.extract("bar\n")).to emit_exactly(["foobar"])
   end
 
   it "should tokenize multiple token" do
-    expect(subject.extract("foo\nbar\n")).to eq(["foo", "bar"])
+    expect(subject.extract("foo\nbar\n")).to emit_exactly(["foo", "bar"])
   end
 
   it "should ignore empty payload" do
-    expect(subject.extract("")).to eq([])
-    expect(subject.extract("foo\nbar")).to eq(["foo"])
+    expect(subject.extract("")).to emit_exactly([])
+    expect(subject.extract("foo\nbar")).to emit_exactly(["foo"])
   end
 
   it "should tokenize empty payload with newline" do
-    expect(subject.extract("\n")).to eq([""])
-    expect(subject.extract("\n\n\n")).to eq(["", "", ""])
+    expect(subject.extract("\n")).to emit_exactly([""])
+    expect(subject.extract("\n\n\n")).to emit_exactly(["", "", ""])
   end
 
   describe 'flush' do
@@ -83,12 +93,12 @@
     let(:delimiter) { "||" }
 
     it "should tokenize multiple token" do
-      expect(subject.extract("foo||b|r||")).to eq(["foo", "b|r"])
+      expect(subject.extract("foo||b|r||")).to emit_exactly(["foo", "b|r"])
     end
 
     it "should ignore empty payload" do
-      expect(subject.extract("")).to eq([])
-      expect(subject.extract("foo||bar")).to eq(["foo"])
+      expect(subject.extract("")).to emit_exactly([])
+      expect(subject.extract("foo||bar")).to emit_exactly(["foo"])
     end
   end
 end

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

@@ -34,6 +34,7 @@
 import org.logstash.ackedqueue.ext.JRubyWrappedAckedQueueExt;
 import org.logstash.common.AbstractDeadLetterQueueWriterExt;
 import org.logstash.common.BufferedTokenizerExt;
+import org.logstash.common.BufferedTokenizer;
 import org.logstash.config.ir.compiler.AbstractFilterDelegatorExt;
 import org.logstash.config.ir.compiler.AbstractOutputDelegatorExt;
 import org.logstash.config.ir.compiler.FilterDelegatorExt;