From 9eab0c819d84b1fc1e0aea52085713ee9f1408ff Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 3 Mar 2025 09:37:17 +0100 Subject: [PATCH 01/30] First rough idea of reimplementation to return an iterator --- .../src/main/java/org/logstash/RubyUtil.java | 12 +- .../org/logstash/common/CustomTokenizer.java | 111 ++++++++++++++++++ 2 files changed, 117 insertions(+), 6 deletions(-) create mode 100644 logstash-core/src/main/java/org/logstash/common/CustomTokenizer.java diff --git a/logstash-core/src/main/java/org/logstash/RubyUtil.java b/logstash-core/src/main/java/org/logstash/RubyUtil.java index 564f51bc08..c79708c833 100644 --- a/logstash-core/src/main/java/org/logstash/RubyUtil.java +++ b/logstash-core/src/main/java/org/logstash/RubyUtil.java @@ -33,7 +33,7 @@ import org.logstash.ackedqueue.ext.JRubyAckedQueueExt; import org.logstash.ackedqueue.ext.JRubyWrappedAckedQueueExt; import org.logstash.common.AbstractDeadLetterQueueWriterExt; -import org.logstash.common.BufferedTokenizerExt; +//import org.logstash.common.BufferedTokenizerExt; import org.logstash.config.ir.compiler.AbstractFilterDelegatorExt; import org.logstash.config.ir.compiler.AbstractOutputDelegatorExt; import org.logstash.config.ir.compiler.FilterDelegatorExt; @@ -152,7 +152,7 @@ public final class RubyUtil { public static final RubyClass OUTPUT_STRATEGY_SHARED; - public static final RubyClass BUFFERED_TOKENIZER; +// public static final RubyClass BUFFERED_TOKENIZER; public static final RubyClass ABSTRACT_METRIC_CLASS; @@ -356,10 +356,10 @@ public final class RubyUtil { OutputStrategyExt.OutputStrategyRegistryExt::new, OutputStrategyExt.OutputStrategyRegistryExt.class ); - BUFFERED_TOKENIZER = RUBY.getOrCreateModule("FileWatch").defineClassUnder( - "BufferedTokenizer", RUBY.getObject(), BufferedTokenizerExt::new - ); - BUFFERED_TOKENIZER.defineAnnotatedMethods(BufferedTokenizerExt.class); +// BUFFERED_TOKENIZER = RUBY.getOrCreateModule("FileWatch").defineClassUnder( +// "BufferedTokenizer", RUBY.getObject(), BufferedTokenizerExt::new +// ); +// BUFFERED_TOKENIZER.defineAnnotatedMethods(BufferedTokenizerExt.class); OUTPUT_DELEGATOR_STRATEGIES = RUBY.defineModuleUnder("OutputDelegatorStrategies", LOGSTASH_MODULE); OUTPUT_STRATEGY_ABSTRACT = OUTPUT_DELEGATOR_STRATEGIES.defineClassUnder( diff --git a/logstash-core/src/main/java/org/logstash/common/CustomTokenizer.java b/logstash-core/src/main/java/org/logstash/common/CustomTokenizer.java new file mode 100644 index 0000000000..333b34ce86 --- /dev/null +++ b/logstash-core/src/main/java/org/logstash/common/CustomTokenizer.java @@ -0,0 +1,111 @@ +package org.logstash.common; + +import java.util.Iterator; +import java.util.NoSuchElementException; + +public class CustomTokenizer { + + private final DataSplitter dataSplitter; + + static class ValueLimitIteratorDecorator implements Iterator { + private final Iterator iterator; + private final int limit = 10; + + ValueLimitIteratorDecorator(Iterator iterator) { + this.iterator = iterator; + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public String next() { + String value = iterator.next(); + if (value.length() > limit) { + throw new IllegalArgumentException("Too long"); + } + return value; + } + } + + static class DataSplitter implements Iterator { + private final String separator; + private int currentIdx = 0; + private final StringBuilder accumulator = new StringBuilder(); + + DataSplitter(String separator) { + this.separator = separator; + } + + @Override + public boolean hasNext() { + int nextIdx = accumulator.indexOf(separator, currentIdx); + if (nextIdx == -1) { + // not found next separator + System.out.println("hasNext return false because next token not found"); + cleanupAccumulator(); + return false; + } else { + return true; + } + } + + @Override + public String next() { + int nextIdx = accumulator.indexOf(separator, currentIdx); + if (nextIdx == -1) { + // not found next separator + cleanupAccumulator(); + throw new NoSuchElementException(); + } else { + String token = accumulator.substring(currentIdx, nextIdx); + currentIdx = nextIdx + separator.length(); + return token; + } + } + + private void cleanupAccumulator() { + accumulator.delete(0, currentIdx); + currentIdx = 0; + } + + public void append(String data) { + accumulator.append(data); + } + + public String flush() { + return accumulator.toString(); + } + + @Override + public String toString() { + return "accumulator=" + accumulator + ", currentIdx=" + currentIdx; + } + } + + public CustomTokenizer(String separator) { + this.dataSplitter = new DataSplitter(separator); + } + + public Iterable extract(String data) { + dataSplitter.append(data); + + return new Iterable() { + @Override + public Iterator iterator() { + return new ValueLimitIteratorDecorator(dataSplitter); + } + }; + } + + public String flush() { + return dataSplitter.flush(); + } + + @Override + public String toString() { + return dataSplitter.toString(); + } +} From 7213acbb6a9d4d18d7cdc73116fd74bc0b2cf681 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 3 Mar 2025 14:57:31 +0100 Subject: [PATCH 02/30] Exposed Java BufferedTokenizer under FileWatch module and adapted tests to use it --- .../benchmark/BufferedTokenizerBenchmark.java | 73 +++++++ .../BufferedTokenizerExtBenchmark.java | 83 -------- logstash-core/lib/logstash/file_watch.rb | 3 + logstash-core/lib/logstash/util.rb | 1 + .../spec/logstash/util/buftok_spec.rb | 28 ++- .../src/main/java/org/logstash/RubyUtil.java | 1 + ...mTokenizer.java => BufferedTokenizer.java} | 35 +++- .../common/BufferedTokenizerExtTest.java | 161 --------------- ...BufferedTokenizerExtWithSizeLimitTest.java | 191 ------------------ .../common/BufferedTokenizerTest.java | 155 ++++++++++++++ ...> BufferedTokenizerWithDelimiterTest.java} | 25 +-- .../BufferedTokenizerWithSizeLimitTest.java | 102 ++++++++++ 12 files changed, 388 insertions(+), 470 deletions(-) create mode 100644 logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java delete mode 100644 logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerExtBenchmark.java create mode 100644 logstash-core/lib/logstash/file_watch.rb rename logstash-core/src/main/java/org/logstash/common/{CustomTokenizer.java => BufferedTokenizer.java} (71%) delete mode 100644 logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtTest.java delete mode 100644 logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtWithSizeLimitTest.java create mode 100644 logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java rename logstash-core/src/test/java/org/logstash/common/{BufferedTokenizerExtWithDelimiterTest.java => BufferedTokenizerWithDelimiterTest.java} (56%) create mode 100644 logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java diff --git a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java new file mode 100644 index 0000000000..5366e0fa04 --- /dev/null +++ b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java @@ -0,0 +1,73 @@ +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 = 100, timeUnit = TimeUnit.MILLISECONDS) +@Fork(1) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@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"; + } + + @SuppressWarnings("unchecked") + @Benchmark + public final void onlyOneTokenPerFragment(Blackhole blackhole) { + Iterable tokens = sut.extract(singleTokenPerFragment); + blackhole.consume(tokens); + } + + @SuppressWarnings("unchecked") + @Benchmark + public final void multipleTokenPerFragment(Blackhole blackhole) { + Iterable tokens = sut.extract(multipleTokensPerFragment); + blackhole.consume(tokens); + } + + @SuppressWarnings("unchecked") + @Benchmark + public final void multipleTokensCrossingMultipleFragments(Blackhole blackhole) { + Iterable tokens = sut.extract(multipleTokensSpreadMultipleFragments_1); + blackhole.consume(tokens); + + tokens = sut.extract(multipleTokensSpreadMultipleFragments_2); + blackhole.consume(tokens); + + tokens = sut.extract(multipleTokensSpreadMultipleFragments_3); + blackhole.consume(tokens); + } +} diff --git a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerExtBenchmark.java b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerExtBenchmark.java deleted file mode 100644 index 5b01cebb3e..0000000000 --- a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerExtBenchmark.java +++ /dev/null @@ -1,83 +0,0 @@ -package org.logstash.benchmark; - -import org.jruby.RubyArray; -import org.jruby.RubyString; -import org.jruby.runtime.ThreadContext; -import org.jruby.runtime.builtin.IRubyObject; -import org.logstash.RubyUtil; -import org.logstash.common.BufferedTokenizerExt; -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; - -import static org.logstash.RubyUtil.RUBY; - -@Warmup(iterations = 3, time = 100, timeUnit = TimeUnit.MILLISECONDS) -@Measurement(iterations = 10, time = 100, timeUnit = TimeUnit.MILLISECONDS) -@Fork(1) -@BenchmarkMode(Mode.Throughput) -@OutputTimeUnit(TimeUnit.NANOSECONDS) -@State(Scope.Thread) -public class BufferedTokenizerExtBenchmark { - - private BufferedTokenizerExt sut; - private ThreadContext context; - private RubyString singleTokenPerFragment; - private RubyString multipleTokensPerFragment; - private RubyString multipleTokensSpreadMultipleFragments_1; - private RubyString multipleTokensSpreadMultipleFragments_2; - private RubyString multipleTokensSpreadMultipleFragments_3; - - @Setup(Level.Invocation) - public void setUp() { - sut = new BufferedTokenizerExt(RubyUtil.RUBY, RubyUtil.BUFFERED_TOKENIZER); - context = RUBY.getCurrentContext(); - IRubyObject[] args = {}; - sut.init(context, args); - singleTokenPerFragment = RubyUtil.RUBY.newString("a".repeat(512) + "\n"); - - multipleTokensPerFragment = RubyUtil.RUBY.newString("a".repeat(512) + "\n" + "b".repeat(512) + "\n" + "c".repeat(512) + "\n"); - - multipleTokensSpreadMultipleFragments_1 = RubyUtil.RUBY.newString("a".repeat(512) + "\n" + "b".repeat(512) + "\n" + "c".repeat(256)); - multipleTokensSpreadMultipleFragments_2 = RubyUtil.RUBY.newString("c".repeat(256) + "\n" + "d".repeat(512) + "\n" + "e".repeat(256)); - multipleTokensSpreadMultipleFragments_3 = RubyUtil.RUBY.newString("f".repeat(256) + "\n" + "g".repeat(512) + "\n" + "h".repeat(512) + "\n"); - } - - @SuppressWarnings("unchecked") - @Benchmark - public final void onlyOneTokenPerFragment(Blackhole blackhole) { - RubyArray tokens = (RubyArray) sut.extract(context, singleTokenPerFragment); - blackhole.consume(tokens); - } - - @SuppressWarnings("unchecked") - @Benchmark - public final void multipleTokenPerFragment(Blackhole blackhole) { - RubyArray tokens = (RubyArray) sut.extract(context, multipleTokensPerFragment); - blackhole.consume(tokens); - } - - @SuppressWarnings("unchecked") - @Benchmark - public final void multipleTokensCrossingMultipleFragments(Blackhole blackhole) { - RubyArray tokens = (RubyArray) sut.extract(context, multipleTokensSpreadMultipleFragments_1); - blackhole.consume(tokens); - - tokens = (RubyArray) sut.extract(context, multipleTokensSpreadMultipleFragments_2); - blackhole.consume(tokens); - - tokens = (RubyArray) sut.extract(context, multipleTokensSpreadMultipleFragments_3); - blackhole.consume(tokens); - } -} diff --git a/logstash-core/lib/logstash/file_watch.rb b/logstash-core/lib/logstash/file_watch.rb new file mode 100644 index 0000000000..f1a73f5592 --- /dev/null +++ b/logstash-core/lib/logstash/file_watch.rb @@ -0,0 +1,3 @@ +module FileWatch + java_import org.logstash.common.BufferedTokenizer +end \ No newline at end of file diff --git a/logstash-core/lib/logstash/util.rb b/logstash-core/lib/logstash/util.rb index 9eac95e1a2..5e1f760f01 100644 --- a/logstash-core/lib/logstash/util.rb +++ b/logstash-core/lib/logstash/util.rb @@ -16,6 +16,7 @@ # under the License. require "logstash/environment" +require "logstash/file_watch" module LogStash::Util UNAME = case RbConfig::CONFIG["host_os"] diff --git a/logstash-core/spec/logstash/util/buftok_spec.rb b/logstash-core/spec/logstash/util/buftok_spec.rb index 3d7b4b1990..4eceb20014 100644 --- a/logstash-core/spec/logstash/util/buftok_spec.rb +++ b/logstash-core/spec/logstash/util/buftok_spec.rb @@ -20,27 +20,33 @@ describe FileWatch::BufferedTokenizer do subject { FileWatch::BufferedTokenizer.new } + def to_list(iterator) + a = [] + iterator.each { |v| a << v } + return a + end + it "should tokenize a single token" do - expect(subject.extract("foo\n")).to eq(["foo"]) + expect(to_list(subject.extract("foo\n"))).to eq(["foo"]) end it "should merge multiple token" do - expect(subject.extract("foo")).to eq([]) - expect(subject.extract("bar\n")).to eq(["foobar"]) + expect(to_list(subject.extract("foo"))).to eq([]) + expect(to_list(subject.extract("bar\n"))).to eq(["foobar"]) end it "should tokenize multiple token" do - expect(subject.extract("foo\nbar\n")).to eq(["foo", "bar"]) + expect(to_list(subject.extract("foo\nbar\n"))).to eq(["foo", "bar"]) end it "should ignore empty payload" do - expect(subject.extract("")).to eq([]) - expect(subject.extract("foo\nbar")).to eq(["foo"]) + expect(to_list(subject.extract(""))).to eq([]) + expect(to_list(subject.extract("foo\nbar"))).to eq(["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(to_list(subject.extract("\n"))).to eq([""]) + expect(to_list(subject.extract("\n\n\n"))).to eq(["", "", ""]) end describe 'flush' do @@ -83,12 +89,12 @@ let(:delimiter) { "||" } it "should tokenize multiple token" do - expect(subject.extract("foo||b|r||")).to eq(["foo", "b|r"]) + expect(to_list(subject.extract("foo||b|r||"))).to eq(["foo", "b|r"]) end it "should ignore empty payload" do - expect(subject.extract("")).to eq([]) - expect(subject.extract("foo||bar")).to eq(["foo"]) + expect(to_list(subject.extract(""))).to eq([]) + expect(to_list(subject.extract("foo||bar"))).to eq(["foo"]) end end end diff --git a/logstash-core/src/main/java/org/logstash/RubyUtil.java b/logstash-core/src/main/java/org/logstash/RubyUtil.java index c79708c833..c3446053c9 100644 --- a/logstash-core/src/main/java/org/logstash/RubyUtil.java +++ b/logstash-core/src/main/java/org/logstash/RubyUtil.java @@ -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; diff --git a/logstash-core/src/main/java/org/logstash/common/CustomTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java similarity index 71% rename from logstash-core/src/main/java/org/logstash/common/CustomTokenizer.java rename to logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 333b34ce86..1576283aa9 100644 --- a/logstash-core/src/main/java/org/logstash/common/CustomTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -3,16 +3,18 @@ import java.util.Iterator; import java.util.NoSuchElementException; -public class CustomTokenizer { +public class BufferedTokenizer { private final DataSplitter dataSplitter; + private Integer sizeLimit; static class ValueLimitIteratorDecorator implements Iterator { private final Iterator iterator; - private final int limit = 10; + private final int limit; - ValueLimitIteratorDecorator(Iterator iterator) { + ValueLimitIteratorDecorator(Iterator iterator, int sizeLimit) { this.iterator = iterator; + this.limit = sizeLimit; } @Override @@ -24,7 +26,7 @@ public boolean hasNext() { public String next() { String value = iterator.next(); if (value.length() > limit) { - throw new IllegalArgumentException("Too long"); + throw new IllegalStateException("input buffer full, consumed token which exceeded the sizeLimit " + limit); } return value; } @@ -85,17 +87,34 @@ public String toString() { } } - public CustomTokenizer(String separator) { + public BufferedTokenizer() { + this("\n"); + } + + public BufferedTokenizer(String separator) { this.dataSplitter = new DataSplitter(separator); } + public BufferedTokenizer(String separator, int sizeLimit) { + if (sizeLimit <= 0) { + throw new IllegalArgumentException("Size limit must be positive"); + } + + this.dataSplitter = new DataSplitter(separator); + this.sizeLimit = sizeLimit; + } + public Iterable extract(String data) { dataSplitter.append(data); return new Iterable() { @Override public Iterator iterator() { - return new ValueLimitIteratorDecorator(dataSplitter); + Iterator returnedIterator = dataSplitter; + if (sizeLimit != null) { + returnedIterator = new ValueLimitIteratorDecorator(returnedIterator, sizeLimit); + } + return returnedIterator; } }; } @@ -108,4 +127,8 @@ public String flush() { public String toString() { return dataSplitter.toString(); } + + public boolean isEmpty() { + return !dataSplitter.hasNext(); + } } diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtTest.java deleted file mode 100644 index 524abb36ed..0000000000 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtTest.java +++ /dev/null @@ -1,161 +0,0 @@ -/* - * 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.jruby.RubyArray; -import org.jruby.RubyEncoding; -import org.jruby.RubyString; -import org.jruby.runtime.ThreadContext; -import org.jruby.runtime.builtin.IRubyObject; -import org.junit.Before; -import org.junit.Test; -import org.logstash.RubyTestBase; -import org.logstash.RubyUtil; - -import java.util.List; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; -import static org.logstash.RubyUtil.RUBY; - -@SuppressWarnings("unchecked") -public final class BufferedTokenizerExtTest extends RubyTestBase { - - private BufferedTokenizerExt sut; - private ThreadContext context; - - @Before - public void setUp() { - sut = new BufferedTokenizerExt(RubyUtil.RUBY, RubyUtil.BUFFERED_TOKENIZER); - context = RUBY.getCurrentContext(); - IRubyObject[] args = {}; - sut.init(context, args); - } - - @Test - public void shouldTokenizeASingleToken() { - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("foo\n")); - - assertEquals(List.of("foo"), tokens); - } - - @Test - public void shouldMergeMultipleToken() { - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("foo")); - assertTrue(tokens.isEmpty()); - - tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("bar\n")); - assertEquals(List.of("foobar"), tokens); - } - - @Test - public void shouldTokenizeMultipleToken() { - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("foo\nbar\n")); - - assertEquals(List.of("foo", "bar"), tokens); - } - - @Test - public void shouldIgnoreEmptyPayload() { - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("")); - assertTrue(tokens.isEmpty()); - - tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("foo\nbar")); - assertEquals(List.of("foo"), tokens); - } - - @Test - public void shouldTokenizeEmptyPayloadWithNewline() { - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("\n")); - assertEquals(List.of(""), tokens); - - tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("\n\n\n")); - assertEquals(List.of("", "", ""), tokens); - } - - @Test - public void shouldNotChangeEncodingOfTokensAfterPartitioning() { - RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x0A, 0x41}); // £ character, newline, A - IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); - RubyArray tokens = (RubyArray)sut.extract(context, rubyInput); - - // read the first token, the £ string - IRubyObject firstToken = tokens.shift(context); - assertEquals("£", firstToken.toString()); - - // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion - RubyEncoding encoding = (RubyEncoding) firstToken.callMethod(context, "encoding"); - assertEquals("ISO-8859-1", encoding.toString()); - } - - @Test - public void shouldNotChangeEncodingOfTokensAfterPartitioningInCaseMultipleExtractionInInvoked() { - RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3}); // £ character - IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); - sut.extract(context, rubyInput); - IRubyObject capitalAInLatin1 = RubyString.newString(RUBY, new byte[]{(byte) 0x41}) - .force_encoding(context, RUBY.newString("ISO8859-1")); - RubyArray tokens = (RubyArray)sut.extract(context, capitalAInLatin1); - assertTrue(tokens.isEmpty()); - - tokens = (RubyArray)sut.extract(context, RubyString.newString(RUBY, new byte[]{(byte) 0x0A})); - - // read the first token, the £ string - IRubyObject firstToken = tokens.shift(context); - assertEquals("£A", firstToken.toString()); - - // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion - RubyEncoding encoding = (RubyEncoding) firstToken.callMethod(context, "encoding"); - assertEquals("ISO-8859-1", encoding.toString()); - } - - @Test - public void shouldNotChangeEncodingOfTokensAfterPartitioningWhenRetrieveLastFlushedToken() { - RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x0A, 0x41}); // £ character, newline, A - IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); - RubyArray tokens = (RubyArray)sut.extract(context, rubyInput); - - // read the first token, the £ string - IRubyObject firstToken = tokens.shift(context); - assertEquals("£", firstToken.toString()); - - // flush and check that the remaining A is still encoded in ISO8859-1 - IRubyObject lastToken = sut.flush(context); - assertEquals("A", lastToken.toString()); - - // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion - RubyEncoding encoding = (RubyEncoding) lastToken.callMethod(context, "encoding"); - assertEquals("ISO-8859-1", encoding.toString()); - } - - @Test - public void givenDirectFlushInvocationUTF8EncodingIsApplied() { - RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x41}); // £ character, A - IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); - - // flush and check that the remaining A is still encoded in ISO8859-1 - IRubyObject lastToken = sut.flush(context); - assertEquals("", lastToken.toString()); - - // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion - RubyEncoding encoding = (RubyEncoding) lastToken.callMethod(context, "encoding"); - assertEquals("UTF-8", encoding.toString()); - } -} \ No newline at end of file diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtWithSizeLimitTest.java deleted file mode 100644 index c25ee01b41..0000000000 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtWithSizeLimitTest.java +++ /dev/null @@ -1,191 +0,0 @@ -/* - * 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.hamcrest.Matchers; -import org.jruby.RubyArray; -import org.jruby.RubyString; -import org.jruby.runtime.ThreadContext; -import org.jruby.runtime.builtin.IRubyObject; -import org.junit.Before; -import org.junit.Test; -import org.logstash.RubyTestBase; -import org.logstash.RubyUtil; -import org.logstash.util.JavaVersion; - -import javax.management.Attribute; -import javax.management.InstanceNotFoundException; -import javax.management.ReflectionException; -import java.lang.management.ManagementFactory; -import java.lang.management.OperatingSystemMXBean; -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.Assume.assumeThat; -import static org.junit.Assume.assumeTrue; -import static org.logstash.RubyUtil.RUBY; - -@SuppressWarnings("unchecked") -public final class BufferedTokenizerExtWithSizeLimitTest extends RubyTestBase { - - public static final int GB = 1024 * 1024 * 1024; - private BufferedTokenizerExt sut; - private ThreadContext context; - - @Before - public void setUp() { - initSUTWithSizeLimit(10); - } - - private void initSUTWithSizeLimit(int sizeLimit) { - sut = new BufferedTokenizerExt(RubyUtil.RUBY, RubyUtil.BUFFERED_TOKENIZER); - context = RUBY.getCurrentContext(); - IRubyObject[] args = {RubyUtil.RUBY.newString("\n"), RubyUtil.RUBY.newFixnum(sizeLimit)}; - sut.init(context, args); - } - - @Test - public void givenTokenWithinSizeLimitWhenExtractedThenReturnTokens() { - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("foo\nbar\n")); - - assertEquals(List.of("foo", "bar"), tokens); - } - - @Test - public void givenTokenExceedingSizeLimitWhenExtractedThenThrowsAnError() { - Exception thrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract(context, RubyUtil.RUBY.newString("this_is_longer_than_10\nkaboom")); - }); - assertThat(thrownException.getMessage(), containsString("input buffer full")); - } - - @Test - public void givenExtractedThrownLimitErrorWhenFeedFreshDataThenReturnTokenStartingFromEndOfOffendingToken() { - Exception thrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract(context, RubyUtil.RUBY.newString("this_is_longer_than_10\nkaboom")); - }); - assertThat(thrownException.getMessage(), containsString("input buffer full")); - - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("\nanother")); - assertEquals("After buffer full error should resume from the end of line", List.of("kaboom"), tokens); - } - - @Test - public void givenExtractInvokedWithDifferentFramingAfterBufferFullErrorTWhenFeedFreshDataThenReturnTokenStartingFromEndOfOffendingToken() { - sut.extract(context, RubyUtil.RUBY.newString("aaaa")); - - Exception thrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract(context, RubyUtil.RUBY.newString("aaaaaaa")); - }); - assertThat(thrownException.getMessage(), containsString("input buffer full")); - - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("aa\nbbbb\nccc")); - assertEquals(List.of("bbbb"), tokens); - } - - @Test - public void giveMultipleSegmentsThatGeneratesMultipleBufferFullErrorsThenIsAbleToRecoverTokenization() { - sut.extract(context, RubyUtil.RUBY.newString("aaaa")); - - //first buffer full on 13 "a" letters - Exception thrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract(context, RubyUtil.RUBY.newString("aaaaaaa")); - }); - assertThat(thrownException.getMessage(), containsString("input buffer full")); - - // second buffer full on 11 "b" letters - Exception secondThrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract(context, RubyUtil.RUBY.newString("aa\nbbbbbbbbbbb\ncc")); - }); - assertThat(secondThrownException.getMessage(), containsString("input buffer full")); - - // now should resemble processing on c and d - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("ccc\nddd\n")); - assertEquals(List.of("ccccc", "ddd"), tokens); - } - - @Test - public void givenTooLongInputExtractDoesntOverflow() { - // This test has proven to go OOM on JDK 11 and JDK 17, also if physical memory is 16GB. - // JDK 21 successfully executes the test due to internal changes or efficiency of G1GC (the default GC). - // Tested also others GC on JDK 11 without any success: - // - ZGC - // - Parallel GC - // - CMS - // remove this code when the minimal JDK version for Logstash is JDK 21 or greater. - - assumeThat("Expect at least JDK 21", JavaVersion.CURRENT, Matchers.greaterThanOrEqualTo(JavaVersion.JAVA_21)); - long expectedNeedHeapMemory = 10L * GB; - // To understand the motivation of 10GB heap please read https://github.com/elastic/logstash/pull/17373#issuecomment-2750378212 - assumeTrue("Skip the test because VM hasn't enough physical memory", hasEnoughPhysicalMemory(expectedNeedHeapMemory)); - - assertEquals("Xmx must equals to what's defined in the Gradle's javaTests task", - expectedNeedHeapMemory, Runtime.getRuntime().maxMemory()); - - // re-init the tokenizer with big sizeLimit - initSUTWithSizeLimit((int) (2L * GB) - 3); - // Integer.MAX_VALUE is 2 * GB - RubyString bigFirstPiece = generateString("a", Integer.MAX_VALUE - 1024); - sut.extract(context, bigFirstPiece); - - // add another small fragment to trigger int overflow - // sizeLimit is (2^32-1)-3 first segment length is (2^32-1) - 1024 second is 1024 +2 - // so the combined length of first and second is > sizeLimit and should throw an exception - // but because of overflow it's negative and happens to be < sizeLimit - Exception thrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract(context, generateString("a", 1024 + 2)); - }); - assertThat(thrownException.getMessage(), containsString("input buffer full")); - } - - private RubyString generateString(String fill, int size) { - return RubyUtil.RUBY.newString(fill.repeat(size)); - } - - private boolean hasEnoughPhysicalMemory(long requiredPhysicalMemory) { - long physicalMemory; - try { - physicalMemory = readPhysicalMemorySize(); - } catch (InstanceNotFoundException | ReflectionException e) { - System.out.println("Can't read attribute JMX OS bean"); - return false; - } catch (IllegalStateException e) { - System.out.println(e.getMessage()); - return false; - } - System.out.println("Physical memory on the VM is: " + physicalMemory + " bytes"); - return physicalMemory > requiredPhysicalMemory; - } - - private long readPhysicalMemorySize() throws ReflectionException, InstanceNotFoundException { - OperatingSystemMXBean op = ManagementFactory.getOperatingSystemMXBean(); - - List attributes = ManagementFactory.getPlatformMBeanServer() - .getAttributes(op.getObjectName(), new String[]{"TotalPhysicalMemorySize"} ).asList(); - if (attributes.isEmpty()) { - throw new IllegalStateException("Attribute TotalPhysicalMemorySize is not available from JMX OS bean"); - } - Attribute a = attributes.get(0); - return (long) (Long) a.getValue(); - } -} \ No newline at end of file diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java new file mode 100644 index 0000000000..43532f5394 --- /dev/null +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java @@ -0,0 +1,155 @@ +/* + * 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.junit.Before; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public final class BufferedTokenizerTest { + + private BufferedTokenizer sut; + + static List toList(Iterable iter) { + List acc = new ArrayList<>(); + iter.forEach(acc::add); + return acc; + } + + @Before + public void setUp() { + sut = new BufferedTokenizer(); + } + + @Test + public void shouldTokenizeASingleToken() { + List tokens = toList(sut.extract("foo\n")); + + assertEquals(List.of("foo"), tokens); + } + + @Test + public void shouldMergeMultipleToken() { + List tokens = toList(sut.extract("foo")); + assertTrue(tokens.isEmpty()); + + tokens = toList(sut.extract("bar\n")); + assertEquals(List.of("foobar"), tokens); + } + + @Test + public void shouldTokenizeMultipleToken() { + List tokens = toList(sut.extract("foo\nbar\n")); + + assertEquals(List.of("foo", "bar"), tokens); + } + + @Test + public void shouldIgnoreEmptyPayload() { + List tokens = toList(sut.extract("")); + assertTrue(tokens.isEmpty()); + + tokens = toList(sut.extract("foo\nbar")); + assertEquals(List.of("foo"), tokens); + } + + @Test + public void shouldTokenizeEmptyPayloadWithNewline() { + List tokens = toList(sut.extract("\n")); + assertEquals(List.of(""), tokens); + + tokens = toList(sut.extract("\n\n\n")); + assertEquals(List.of("", "", ""), tokens); + } + +// @Test +// public void shouldNotChangeEncodingOfTokensAfterPartitioning() { +// RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x0A, 0x41}); // £ character, newline, A +// IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); +// RubyArray tokens = (RubyArray)sut.extract(context, rubyInput); +// +// // read the first token, the £ string +// IRubyObject firstToken = tokens.shift(context); +// assertEquals("£", firstToken.toString()); +// +// // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion +// RubyEncoding encoding = (RubyEncoding) firstToken.callMethod(context, "encoding"); +// assertEquals("ISO-8859-1", encoding.toString()); +// } +// +// @Test +// public void shouldNotChangeEncodingOfTokensAfterPartitioningInCaseMultipleExtractionInInvoked() { +// RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3}); // £ character +// IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); +// sut.extract(context, rubyInput); +// IRubyObject capitalAInLatin1 = RubyString.newString(RUBY, new byte[]{(byte) 0x41}) +// .force_encoding(context, RUBY.newString("ISO8859-1")); +// RubyArray tokens = (RubyArray)sut.extract(context, capitalAInLatin1); +// assertTrue(tokens.isEmpty()); +// +// tokens = (RubyArray)sut.extract(context, RubyString.newString(RUBY, new byte[]{(byte) 0x0A})); +// +// // read the first token, the £ string +// IRubyObject firstToken = tokens.shift(context); +// assertEquals("£A", firstToken.toString()); +// +// // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion +// RubyEncoding encoding = (RubyEncoding) firstToken.callMethod(context, "encoding"); +// assertEquals("ISO-8859-1", encoding.toString()); +// } +// +// @Test +// public void shouldNotChangeEncodingOfTokensAfterPartitioningWhenRetrieveLastFlushedToken() { +// RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x0A, 0x41}); // £ character, newline, A +// IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); +// RubyArray tokens = (RubyArray)sut.extract(context, rubyInput); +// +// // read the first token, the £ string +// IRubyObject firstToken = tokens.shift(context); +// assertEquals("£", firstToken.toString()); +// +// // flush and check that the remaining A is still encoded in ISO8859-1 +// IRubyObject lastToken = sut.flush(context); +// assertEquals("A", lastToken.toString()); +// +// // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion +// RubyEncoding encoding = (RubyEncoding) lastToken.callMethod(context, "encoding"); +// assertEquals("ISO-8859-1", encoding.toString()); +// } +// +// @Test +// public void givenDirectFlushInvocationUTF8EncodingIsApplied() { +// RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x41}); // £ character, A +// IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); +// +// // flush and check that the remaining A is still encoded in ISO8859-1 +// IRubyObject lastToken = sut.flush(context); +// assertEquals("", lastToken.toString()); +// +// // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion +// RubyEncoding encoding = (RubyEncoding) lastToken.callMethod(context, "encoding"); +// assertEquals("UTF-8", encoding.toString()); +// } +} \ No newline at end of file diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtWithDelimiterTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithDelimiterTest.java similarity index 56% rename from logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtWithDelimiterTest.java rename to logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithDelimiterTest.java index 19872e66c3..2b4c38c7e4 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtWithDelimiterTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithDelimiterTest.java @@ -19,48 +19,37 @@ package org.logstash.common; -import org.jruby.RubyArray; -import org.jruby.RubyString; -import org.jruby.runtime.ThreadContext; -import org.jruby.runtime.builtin.IRubyObject; import org.junit.Before; import org.junit.Test; -import org.logstash.RubyTestBase; -import org.logstash.RubyUtil; import java.util.List; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; -import static org.logstash.RubyUtil.RUBY; +import static org.logstash.common.BufferedTokenizerTest.toList; -@SuppressWarnings("unchecked") -public final class BufferedTokenizerExtWithDelimiterTest extends RubyTestBase { +public final class BufferedTokenizerWithDelimiterTest { - private BufferedTokenizerExt sut; - private ThreadContext context; + private BufferedTokenizer sut; @Before public void setUp() { - sut = new BufferedTokenizerExt(RubyUtil.RUBY, RubyUtil.BUFFERED_TOKENIZER); - context = RUBY.getCurrentContext(); - IRubyObject[] args = {RubyUtil.RUBY.newString("||")}; - sut.init(context, args); + sut = new BufferedTokenizer("||"); } @Test public void shouldTokenizeMultipleToken() { - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("foo||b|r||")); + List tokens = toList(sut.extract("foo||b|r||")); assertEquals(List.of("foo", "b|r"), tokens); } @Test public void shouldIgnoreEmptyPayload() { - RubyArray tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("")); + List tokens = toList(sut.extract("")); assertTrue(tokens.isEmpty()); - tokens = (RubyArray) sut.extract(context, RubyUtil.RUBY.newString("foo||bar")); + tokens = toList(sut.extract("foo||bar")); assertEquals(List.of("foo"), tokens); } } diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java new file mode 100644 index 0000000000..780434d071 --- /dev/null +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -0,0 +1,102 @@ +/* + * 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.junit.Before; +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.logstash.common.BufferedTokenizerTest.toList; + +public final class BufferedTokenizerWithSizeLimitTest { + + private BufferedTokenizer sut; + + @Before + public void setUp() { + sut = new BufferedTokenizer("\n", 10); + } + + @Test + public void givenTokenWithinSizeLimitWhenExtractedThenReturnTokens() { + List tokens = toList(sut.extract("foo\nbar\n")); + + assertEquals(List.of("foo", "bar"), tokens); + } + + @Test + public void givenTokenExceedingSizeLimitWhenExtractedThenThrowsAnError() { + Exception thrownException = assertThrows(IllegalStateException.class, () -> { + sut.extract("this_is_longer_than_10\nkaboom").forEach(s -> {}); + }); + assertThat(thrownException.getMessage(), containsString("input buffer full")); + } + + @Test + public void givenExtractedThrownLimitErrorWhenFeedFreshDataThenReturnTokenStartingFromEndOfOffendingToken() { + Exception thrownException = assertThrows(IllegalStateException.class, () -> { + sut.extract("this_is_longer_than_10\nkaboom").forEach(s -> {}); + }); + assertThat(thrownException.getMessage(), containsString("input buffer full")); + + List tokens = toList(sut.extract("\nanother")); + assertEquals("After buffer full error should resume from the end of line", List.of("kaboom"), tokens); + } + + @Test + public void givenExtractInvokedWithDifferentFramingAfterBufferFullErrorTWhenFeedFreshDataThenReturnTokenStartingFromEndOfOffendingToken() { + sut.extract("aaaa"); + + Exception thrownException = assertThrows(IllegalStateException.class, () -> { + sut.extract("aaaaaaa").forEach(s -> {}); + }); + assertThat(thrownException.getMessage(), containsString("input buffer full")); + + List tokens = toList(sut.extract("aa\nbbbb\nccc")); + assertEquals(List.of("bbbb"), tokens); + } + + @Test + public void giveMultipleSegmentsThatGeneratesMultipleBufferFullErrorsThenIsAbleToRecoverTokenization() { + sut.extract("aaaa"); + + //first buffer full on 13 "a" letters + Exception thrownException = assertThrows(IllegalStateException.class, () -> { + sut.extract("aaaaaaa").forEach(s -> {}); + }); + assertThat(thrownException.getMessage(), containsString("input buffer full")); + + // second buffer full on 11 "b" letters + Exception secondThrownException = assertThrows(IllegalStateException.class, () -> { + sut.extract("aa\nbbbbbbbbbbb\ncc"); + }); + assertThat(secondThrownException.getMessage(), containsString("input buffer full")); + + // now should resemble processing on c and d + List tokens = toList(sut.extract("ccc\nddd\n")); + assertEquals(List.of("ccccc", "ddd"), tokens); + } +} \ No newline at end of file From ef5ead05923fd73fcb0ea7efae9270df06a7a45b Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 4 Mar 2025 16:11:37 +0100 Subject: [PATCH 03/30] Moved string encoding logic to outer Ruby extension BufferedTokenizerExt class --- .../benchmark/BufferedTokenizerBenchmark.java | 3 - logstash-core/lib/logstash/file_watch.rb | 2 +- .../src/main/java/org/logstash/RubyUtil.java | 12 +- .../logstash/common/BufferedTokenizer.java | 18 +- .../logstash/common/BufferedTokenizerExt.java | 226 ++++++++++-------- .../common/BufferedTokenizerExtTest.java | 171 +++++++++++++ .../common/BufferedTokenizerTest.java | 71 +----- 7 files changed, 322 insertions(+), 181 deletions(-) create mode 100644 logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtTest.java diff --git a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java index 5366e0fa04..eb0c1df8ff 100644 --- a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java +++ b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java @@ -44,21 +44,18 @@ public void setUp() { multipleTokensSpreadMultipleFragments_3 = "f".repeat(256) + "\n" + "g".repeat(512) + "\n" + "h".repeat(512) + "\n"; } - @SuppressWarnings("unchecked") @Benchmark public final void onlyOneTokenPerFragment(Blackhole blackhole) { Iterable tokens = sut.extract(singleTokenPerFragment); blackhole.consume(tokens); } - @SuppressWarnings("unchecked") @Benchmark public final void multipleTokenPerFragment(Blackhole blackhole) { Iterable tokens = sut.extract(multipleTokensPerFragment); blackhole.consume(tokens); } - @SuppressWarnings("unchecked") @Benchmark public final void multipleTokensCrossingMultipleFragments(Blackhole blackhole) { Iterable tokens = sut.extract(multipleTokensSpreadMultipleFragments_1); diff --git a/logstash-core/lib/logstash/file_watch.rb b/logstash-core/lib/logstash/file_watch.rb index f1a73f5592..a05a8549b5 100644 --- a/logstash-core/lib/logstash/file_watch.rb +++ b/logstash-core/lib/logstash/file_watch.rb @@ -1,3 +1,3 @@ module FileWatch - java_import org.logstash.common.BufferedTokenizer + # java_import org.logstash.common.BufferedTokenizer end \ No newline at end of file diff --git a/logstash-core/src/main/java/org/logstash/RubyUtil.java b/logstash-core/src/main/java/org/logstash/RubyUtil.java index c3446053c9..50ee6d4944 100644 --- a/logstash-core/src/main/java/org/logstash/RubyUtil.java +++ b/logstash-core/src/main/java/org/logstash/RubyUtil.java @@ -33,7 +33,7 @@ import org.logstash.ackedqueue.ext.JRubyAckedQueueExt; import org.logstash.ackedqueue.ext.JRubyWrappedAckedQueueExt; import org.logstash.common.AbstractDeadLetterQueueWriterExt; -//import org.logstash.common.BufferedTokenizerExt; +import org.logstash.common.BufferedTokenizerExt; import org.logstash.common.BufferedTokenizer; import org.logstash.config.ir.compiler.AbstractFilterDelegatorExt; import org.logstash.config.ir.compiler.AbstractOutputDelegatorExt; @@ -153,7 +153,7 @@ public final class RubyUtil { public static final RubyClass OUTPUT_STRATEGY_SHARED; -// public static final RubyClass BUFFERED_TOKENIZER; + public static final RubyClass BUFFERED_TOKENIZER; public static final RubyClass ABSTRACT_METRIC_CLASS; @@ -357,10 +357,10 @@ public final class RubyUtil { OutputStrategyExt.OutputStrategyRegistryExt::new, OutputStrategyExt.OutputStrategyRegistryExt.class ); -// BUFFERED_TOKENIZER = RUBY.getOrCreateModule("FileWatch").defineClassUnder( -// "BufferedTokenizer", RUBY.getObject(), BufferedTokenizerExt::new -// ); -// BUFFERED_TOKENIZER.defineAnnotatedMethods(BufferedTokenizerExt.class); + BUFFERED_TOKENIZER = RUBY.getOrCreateModule("FileWatch").defineClassUnder( + "BufferedTokenizer", RUBY.getObject(), BufferedTokenizerExt::new + ); + BUFFERED_TOKENIZER.defineAnnotatedMethods(BufferedTokenizerExt.class); OUTPUT_DELEGATOR_STRATEGIES = RUBY.defineModuleUnder("OutputDelegatorStrategies", LOGSTASH_MODULE); OUTPUT_STRATEGY_ABSTRACT = OUTPUT_DELEGATOR_STRATEGIES.defineClassUnder( diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 1576283aa9..f10650ea61 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -8,19 +8,26 @@ public class BufferedTokenizer { private final DataSplitter dataSplitter; private Integer sizeLimit; - static class ValueLimitIteratorDecorator implements Iterator { - private final Iterator iterator; - private final int limit; + static abstract class IteratorDecorator implements Iterator { + protected final Iterator iterator; - ValueLimitIteratorDecorator(Iterator iterator, int sizeLimit) { + IteratorDecorator(Iterator iterator) { this.iterator = iterator; - this.limit = sizeLimit; } @Override public boolean hasNext() { return iterator.hasNext(); } + } + + static class ValueLimitIteratorDecorator extends IteratorDecorator { + private final int limit; + + ValueLimitIteratorDecorator(Iterator iterator, int sizeLimit) { + super(iterator); + this.limit = sizeLimit; + } @Override public String next() { @@ -46,7 +53,6 @@ public boolean hasNext() { int nextIdx = accumulator.indexOf(separator, currentIdx); if (nextIdx == -1) { // not found next separator - System.out.println("hasNext return false because next token not found"); cleanupAccumulator(); return false; } else { diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java index 2161dc9004..09b7ee9a41 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java @@ -34,23 +34,24 @@ import org.logstash.RubyUtil; import java.nio.charset.Charset; +import java.util.Iterator; @JRubyClass(name = "BufferedTokenizer") public class BufferedTokenizerExt extends RubyObject { private static final long serialVersionUID = 1L; - private static final RubyString NEW_LINE = (RubyString) RubyUtil.RUBY.newString("\n"). - freeze(RubyUtil.RUBY.getCurrentContext()); - - private @SuppressWarnings("rawtypes") RubyArray input = RubyUtil.RUBY.newArray(); - private StringBuilder headToken = new StringBuilder(); - private RubyString delimiter = NEW_LINE; - private int sizeLimit; - private boolean hasSizeLimit; - private long inputSize; - private boolean bufferFullErrorNotified = false; +// private static final RubyString NEW_LINE = (RubyString) RubyUtil.RUBY.newString("\n"). +// freeze(RubyUtil.RUBY.getCurrentContext()); +// private @SuppressWarnings("rawtypes") RubyArray input = RubyUtil.RUBY.newArray(); +// private StringBuilder headToken = new StringBuilder(); +// private RubyString delimiter = NEW_LINE; +// private int sizeLimit; +// private boolean hasSizeLimit; +// private int inputSize; +// private boolean bufferFullErrorNotified = false; private String encodingName; + private transient BufferedTokenizer tokenizer; public BufferedTokenizerExt(final Ruby runtime, final RubyClass metaClass) { super(runtime, metaClass); @@ -58,18 +59,30 @@ public BufferedTokenizerExt(final Ruby runtime, final RubyClass metaClass) { @JRubyMethod(name = "initialize", optional = 2) public IRubyObject init(final ThreadContext context, IRubyObject[] args) { +// if (args.length >= 1) { +// this.delimiter = args[0].convertToString(); +// } +// if (args.length == 2) { +// final int sizeLimit = args[1].convertToInteger().getIntValue(); +// if (sizeLimit <= 0) { +// throw new IllegalArgumentException("Size limit must be positive"); +// } +// this.sizeLimit = sizeLimit; +// this.hasSizeLimit = true; +// } +// this.inputSize = 0; + + String delimiter = "\n"; if (args.length >= 1) { - this.delimiter = args[0].convertToString(); + delimiter = args[0].convertToString().asJavaString(); } if (args.length == 2) { final int sizeLimit = args[1].convertToInteger().getIntValue(); - if (sizeLimit <= 0) { - throw new IllegalArgumentException("Size limit must be positive"); - } - this.sizeLimit = sizeLimit; - this.hasSizeLimit = true; + this.tokenizer = new BufferedTokenizer(delimiter, sizeLimit); + } else { + this.tokenizer = new BufferedTokenizer(delimiter); } - this.inputSize = 0; + return this; } @@ -86,70 +99,88 @@ public IRubyObject init(final ThreadContext context, IRubyObject[] args) { */ @JRubyMethod @SuppressWarnings("rawtypes") - public RubyArray extract(final ThreadContext context, IRubyObject data) { + public IRubyObject extract(final ThreadContext context, IRubyObject data) { RubyEncoding encoding = (RubyEncoding) data.convertToString().encoding(context); encodingName = encoding.getEncoding().getCharsetName(); - final RubyArray entities = data.convertToString().split(delimiter, -1); - if (!bufferFullErrorNotified) { - input.clear(); - input.concat(entities); - } else { - // after a full buffer signal - if (input.isEmpty()) { - // after a buffer full error, the remaining part of the line, till next delimiter, - // has to be consumed, unless the input buffer doesn't still contain fragments of - // subsequent tokens. - entities.shift(context); - input.concat(entities); - } else { - // merge last of the input with first of incoming data segment - if (!entities.isEmpty()) { - RubyString last = ((RubyString) input.pop(context)); - RubyString nextFirst = ((RubyString) entities.shift(context)); - entities.unshift(last.concat(nextFirst)); - input.concat(entities); - } - } - } - if (hasSizeLimit) { - if (bufferFullErrorNotified) { - bufferFullErrorNotified = false; - if (input.isEmpty()) { - return RubyUtil.RUBY.newArray(); - } - } - final int entitiesSize = ((RubyString) input.first()).size(); - if (inputSize + entitiesSize > sizeLimit) { - bufferFullErrorNotified = true; - headToken = new StringBuilder(); - String errorMessage = String.format("input buffer full, consumed token which exceeded the sizeLimit %d; inputSize: %d, entitiesSize %d", sizeLimit, inputSize, entitiesSize); - inputSize = 0; - input.shift(context); // consume the token fragment that generates the buffer full - throw new IllegalStateException(errorMessage); - } - this.inputSize = inputSize + entitiesSize; - } + Iterable extractor = tokenizer.extract(data.asJavaString()); - if (input.getLength() < 2) { - // this is a specialization case which avoid adding and removing from input accumulator - // when it contains just one element - headToken.append(input.shift(context)); // remove head - return RubyUtil.RUBY.newArray(); - } - if (headToken.length() > 0) { - // if there is a pending token part, merge it with the first token segment present - // in the accumulator, and clean the pending token part. - headToken.append(input.shift(context)); // append buffer to first element and - // create new RubyString with the data specified encoding - RubyString encodedHeadToken = toEncodedRubyString(context, headToken.toString()); - input.unshift(encodedHeadToken); // reinsert it into the array - headToken = new StringBuilder(); - } - headToken.append(input.pop(context)); // put the leftovers in headToken for later - inputSize = headToken.length(); - return input; + IRubyObject rubyIterable = RubyUtil.toRubyObject(new Iterable() { + @Override + public Iterator iterator() { + return new BufferedTokenizer.IteratorDecorator<>(extractor.iterator()) { + @Override + public CharSequence next() { + return toEncodedRubyString(context, iterator.next()); + } + }; + } + }); + + return rubyIterable; + +// final RubyArray entities = data.convertToString().split(delimiter, -1); +// if (!bufferFullErrorNotified) { +// input.clear(); +// input.concat(entities); +// } else { +// // after a full buffer signal +// if (input.isEmpty()) { +// // after a buffer full error, the remaining part of the line, till next delimiter, +// // has to be consumed, unless the input buffer doesn't still contain fragments of +// // subsequent tokens. +// entities.shift(context); +// input.concat(entities); +// } else { +// // merge last of the input with first of incoming data segment +// if (!entities.isEmpty()) { +// RubyString last = ((RubyString) input.pop(context)); +// RubyString nextFirst = ((RubyString) entities.shift(context)); +// entities.unshift(last.concat(nextFirst)); +// input.concat(entities); +// } +// } +// } +// +// if (hasSizeLimit) { +// if (bufferFullErrorNotified) { +// bufferFullErrorNotified = false; +// if (input.isEmpty()) { +// return RubyUtil.RUBY.newArray(); +// } +// } +// final int entitiesSize = ((RubyString) input.first()).size(); +// if (inputSize + entitiesSize > sizeLimit) { +// bufferFullErrorNotified = true; +// headToken = new StringBuilder(); +// String errorMessage = String.format("input buffer full, consumed token which exceeded the sizeLimit %d; inputSize: %d, entitiesSize %d", sizeLimit, inputSize, entitiesSize); +// inputSize = 0; +// input.shift(context); // consume the token fragment that generates the buffer full +// throw new IllegalStateException(errorMessage); +// } +// this.inputSize = inputSize + entitiesSize; +// } +// +// if (input.getLength() < 2) { +// // this is a specialization case which avoid adding and removing from input accumulator +// // when it contains just one element +// headToken.append(input.shift(context)); // remove head +// return RubyUtil.RUBY.newArray(); +// } +// +// if (headToken.length() > 0) { +// // if there is a pending token part, merge it with the first token segment present +// // in the accumulator, and clean the pending token part. +// headToken.append(input.shift(context)); // append buffer to first element and +// // create new RubyString with the data specified encoding +// RubyString encodedHeadToken = toEncodedRubyString(context, headToken.toString()); +// input.unshift(encodedHeadToken); // reinsert it into the array +// headToken = new StringBuilder(); +// } +// headToken.append(input.pop(context)); // put the leftovers in headToken for later +// inputSize = headToken.length(); +// return input; } private RubyString toEncodedRubyString(ThreadContext context, String input) { @@ -168,30 +199,33 @@ private RubyString toEncodedRubyString(ThreadContext context, String input) { */ @JRubyMethod public IRubyObject flush(final ThreadContext context) { - final IRubyObject buffer = RubyUtil.toRubyObject(headToken.toString()); - headToken = new StringBuilder(); - inputSize = 0; - - // create new RubyString with the last data specified encoding, if exists - RubyString encodedHeadToken; - if (encodingName != null) { - encodedHeadToken = toEncodedRubyString(context, buffer.toString()); - } else { - // When used with TCP input it could be that on socket connection the flush method - // is invoked while no invocation of extract, leaving the encoding name unassigned. - // In such case also the headToken must be empty - if (!buffer.toString().isEmpty()) { - throw new IllegalStateException("invoked flush with unassigned encoding but not empty head token, this shouldn't happen"); - } - encodedHeadToken = (RubyString) buffer; - } - - return encodedHeadToken; + String s = tokenizer.flush(); + return RubyUtil.toRubyObject(s); +// final IRubyObject buffer = RubyUtil.toRubyObject(headToken.toString()); +// headToken = new StringBuilder(); +// inputSize = 0; +// +// // create new RubyString with the last data specified encoding, if exists +// RubyString encodedHeadToken; +// if (encodingName != null) { +// encodedHeadToken = toEncodedRubyString(context, buffer.toString()); +// } else { +// // When used with TCP input it could be that on socket connection the flush method +// // is invoked while no invocation of extract, leaving the encoding name unassigned. +// // In such case also the headToken must be empty +// if (!buffer.toString().isEmpty()) { +// throw new IllegalStateException("invoked flush with unassigned encoding but not empty head token, this shouldn't happen"); +// } +// encodedHeadToken = (RubyString) buffer; +// } +// +// return encodedHeadToken; } @JRubyMethod(name = "empty?") public IRubyObject isEmpty(final ThreadContext context) { - return RubyUtil.RUBY.newBoolean(headToken.toString().isEmpty() && (inputSize == 0)); +// return RubyUtil.RUBY.newBoolean(headToken.toString().isEmpty() && (inputSize == 0)); + return RubyUtil.RUBY.newBoolean(tokenizer.isEmpty()); } } diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtTest.java new file mode 100644 index 0000000000..c33042b4ce --- /dev/null +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerExtTest.java @@ -0,0 +1,171 @@ +/* + * 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.jruby.RubyEncoding; +import org.jruby.RubyString; +import org.jruby.runtime.ThreadContext; +import org.jruby.runtime.builtin.IRubyObject; +import org.junit.Before; +import org.junit.Test; +import org.logstash.RubyTestBase; +import org.logstash.RubyUtil; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.logstash.RubyUtil.RUBY; + +@SuppressWarnings("unchecked") +public final class BufferedTokenizerExtTest extends RubyTestBase { + + private BufferedTokenizerExt sut; + private ThreadContext context; + + @Before + public void setUp() { + sut = new BufferedTokenizerExt(RubyUtil.RUBY, RubyUtil.BUFFERED_TOKENIZER); + context = RUBY.getCurrentContext(); + IRubyObject[] args = {}; + sut.init(context, args); + } + + private static List toList(Iterable it) { + List l = new ArrayList<>(); + it.forEach(e -> l.add(e.toString())); + return l; + } + + private static List toList(IRubyObject it) { + return toList(it.toJava(Iterable.class)); + } + + @Test + public void shouldTokenizeASingleToken() { + List tokens = toList(sut.extract(context, RubyUtil.RUBY.newString("foo\n"))); + + assertEquals(List.of("foo"), tokens); + } + + @Test + public void shouldMergeMultipleToken() { + List tokens = toList(sut.extract(context, RubyUtil.RUBY.newString("foo"))); + assertTrue(tokens.isEmpty()); + + tokens = toList(sut.extract(context, RubyUtil.RUBY.newString("bar\n"))); + assertEquals(List.of("foobar"), tokens); + } + + @Test + public void shouldTokenizeMultipleToken() { + List tokens = toList(sut.extract(context, RubyUtil.RUBY.newString("foo\nbar\n"))); + + assertEquals(List.of("foo", "bar"), tokens); + } + + @Test + public void shouldIgnoreEmptyPayload() { + List tokens = toList(sut.extract(context, RubyUtil.RUBY.newString(""))); + assertTrue(tokens.isEmpty()); + + tokens = toList(sut.extract(context, RubyUtil.RUBY.newString("foo\nbar"))); + assertEquals(List.of("foo"), tokens); + } + + @Test + public void shouldTokenizeEmptyPayloadWithNewline() { + List tokens = toList(sut.extract(context, RubyUtil.RUBY.newString("\n"))); + assertEquals(List.of(""), tokens); + + tokens = toList(sut.extract(context, RubyUtil.RUBY.newString("\n\n\n"))); + assertEquals(List.of("", "", ""), tokens); + } + + @Test + public void shouldNotChangeEncodingOfTokensAfterPartitioning() { + RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x0A, 0x41}); // £ character, newline, A + IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); + Iterable tokens = sut.extract(context, rubyInput).toJava(Iterable.class); + + // read the first token, the £ string + RubyString firstToken = tokens.iterator().next(); + assertEquals("£", firstToken.toString()); + + // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion + RubyEncoding encoding = (RubyEncoding) firstToken.callMethod(context, "encoding"); + assertEquals("ISO-8859-1", encoding.toString()); + } + + @Test + public void shouldNotChangeEncodingOfTokensAfterPartitioningInCaseMultipleExtractionInInvoked() { + RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3}); // £ character + IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); + sut.extract(context, rubyInput); + IRubyObject capitalAInLatin1 = RubyString.newString(RUBY, new byte[]{(byte) 0x41}) + .force_encoding(context, RUBY.newString("ISO8859-1")); + List tokensJava = toList(sut.extract(context, capitalAInLatin1)); + assertTrue(tokensJava.isEmpty()); + + Iterable tokens = sut.extract(context, RubyString.newString(RUBY, new byte[]{(byte) 0x0A})).toJava(Iterable.class); + + // read the first token, the £ string + RubyString firstToken = tokens.iterator().next(); + assertEquals("£A", firstToken.toString()); + + // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion + RubyEncoding encoding = (RubyEncoding) firstToken.callMethod(context, "encoding"); + assertEquals("ISO-8859-1", encoding.toString()); + } + + @Test + public void shouldNotChangeEncodingOfTokensAfterPartitioningWhenRetrieveLastFlushedToken() { + RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x0A, 0x41}); // £ character, newline, A + IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); + Iterable tokens = sut.extract(context, rubyInput).toJava(Iterable.class); + + // read the first token, the £ string + RubyString firstToken = tokens.iterator().next(); + assertEquals("£", firstToken.toString()); + + // flush and check that the remaining A is still encoded in ISO8859-1 + IRubyObject lastToken = sut.flush(context); + assertEquals("A", lastToken.toString()); + + // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion + RubyEncoding encoding = (RubyEncoding) lastToken.callMethod(context, "encoding"); + assertEquals("ISO-8859-1", encoding.toString()); + } + + @Test + public void givenDirectFlushInvocationUTF8EncodingIsApplied() { + RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x41}); // £ character, A + IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); + + // flush and check that the remaining A is still encoded in ISO8859-1 + IRubyObject lastToken = sut.flush(context); + assertEquals("", lastToken.toString()); + + // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion + RubyEncoding encoding = (RubyEncoding) lastToken.callMethod(context, "encoding"); + assertEquals("UTF-8", encoding.toString()); + } +} \ No newline at end of file diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java index 43532f5394..6499393b00 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java @@ -23,10 +23,12 @@ import org.junit.Test; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; public final class BufferedTokenizerTest { @@ -83,73 +85,4 @@ public void shouldTokenizeEmptyPayloadWithNewline() { tokens = toList(sut.extract("\n\n\n")); assertEquals(List.of("", "", ""), tokens); } - -// @Test -// public void shouldNotChangeEncodingOfTokensAfterPartitioning() { -// RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x0A, 0x41}); // £ character, newline, A -// IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); -// RubyArray tokens = (RubyArray)sut.extract(context, rubyInput); -// -// // read the first token, the £ string -// IRubyObject firstToken = tokens.shift(context); -// assertEquals("£", firstToken.toString()); -// -// // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion -// RubyEncoding encoding = (RubyEncoding) firstToken.callMethod(context, "encoding"); -// assertEquals("ISO-8859-1", encoding.toString()); -// } -// -// @Test -// public void shouldNotChangeEncodingOfTokensAfterPartitioningInCaseMultipleExtractionInInvoked() { -// RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3}); // £ character -// IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); -// sut.extract(context, rubyInput); -// IRubyObject capitalAInLatin1 = RubyString.newString(RUBY, new byte[]{(byte) 0x41}) -// .force_encoding(context, RUBY.newString("ISO8859-1")); -// RubyArray tokens = (RubyArray)sut.extract(context, capitalAInLatin1); -// assertTrue(tokens.isEmpty()); -// -// tokens = (RubyArray)sut.extract(context, RubyString.newString(RUBY, new byte[]{(byte) 0x0A})); -// -// // read the first token, the £ string -// IRubyObject firstToken = tokens.shift(context); -// assertEquals("£A", firstToken.toString()); -// -// // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion -// RubyEncoding encoding = (RubyEncoding) firstToken.callMethod(context, "encoding"); -// assertEquals("ISO-8859-1", encoding.toString()); -// } -// -// @Test -// public void shouldNotChangeEncodingOfTokensAfterPartitioningWhenRetrieveLastFlushedToken() { -// RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x0A, 0x41}); // £ character, newline, A -// IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); -// RubyArray tokens = (RubyArray)sut.extract(context, rubyInput); -// -// // read the first token, the £ string -// IRubyObject firstToken = tokens.shift(context); -// assertEquals("£", firstToken.toString()); -// -// // flush and check that the remaining A is still encoded in ISO8859-1 -// IRubyObject lastToken = sut.flush(context); -// assertEquals("A", lastToken.toString()); -// -// // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion -// RubyEncoding encoding = (RubyEncoding) lastToken.callMethod(context, "encoding"); -// assertEquals("ISO-8859-1", encoding.toString()); -// } -// -// @Test -// public void givenDirectFlushInvocationUTF8EncodingIsApplied() { -// RubyString rubyString = RubyString.newString(RUBY, new byte[]{(byte) 0xA3, 0x41}); // £ character, A -// IRubyObject rubyInput = rubyString.force_encoding(context, RUBY.newString("ISO8859-1")); -// -// // flush and check that the remaining A is still encoded in ISO8859-1 -// IRubyObject lastToken = sut.flush(context); -// assertEquals("", lastToken.toString()); -// -// // verify encoding "ISO8859-1" is preserved in the Java to Ruby String conversion -// RubyEncoding encoding = (RubyEncoding) lastToken.callMethod(context, "encoding"); -// assertEquals("UTF-8", encoding.toString()); -// } } \ No newline at end of file From 51d8eddb7eb24dfe17642c8f97399c1092c2f368 Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 4 Mar 2025 17:20:15 +0100 Subject: [PATCH 04/30] Fixed flush behavior --- .../org/logstash/common/BufferedTokenizer.java | 2 +- .../org/logstash/common/BufferedTokenizerExt.java | 15 ++++++++++++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index f10650ea61..0f43348dbd 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -84,7 +84,7 @@ public void append(String data) { } public String flush() { - return accumulator.toString(); + return accumulator.substring(currentIdx); } @Override diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java index 09b7ee9a41..acffcbb497 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java @@ -200,7 +200,20 @@ private RubyString toEncodedRubyString(ThreadContext context, String input) { @JRubyMethod public IRubyObject flush(final ThreadContext context) { String s = tokenizer.flush(); - return RubyUtil.toRubyObject(s); + + // create new RubyString with the last data specified encoding, if exists + if (encodingName != null) { + return toEncodedRubyString(context, s); + } else { + // When used with TCP input it could be that on socket connection the flush method + // is invoked while no invocation of extract, leaving the encoding name unassigned. + // In such case also the headToken must be empty + if (!s.isEmpty()) { + throw new IllegalStateException("invoked flush with unassigned encoding but not empty head token, this shouldn't happen"); + } + return RubyUtil.toRubyObject(s); + } + // final IRubyObject buffer = RubyUtil.toRubyObject(headToken.toString()); // headToken = new StringBuilder(); // inputSize = 0; From 950b8d034fea3912038490558911c3342eefeaa7 Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 4 Mar 2025 17:22:27 +0100 Subject: [PATCH 05/30] Minor, removed commented code --- .../logstash/common/BufferedTokenizerExt.java | 108 +----------------- 1 file changed, 1 insertion(+), 107 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java index acffcbb497..a6456bf345 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java @@ -21,7 +21,6 @@ package org.logstash.common; import org.jruby.Ruby; -import org.jruby.RubyArray; import org.jruby.RubyClass; import org.jruby.RubyEncoding; import org.jruby.RubyObject; @@ -41,15 +40,6 @@ public class BufferedTokenizerExt extends RubyObject { private static final long serialVersionUID = 1L; -// private static final RubyString NEW_LINE = (RubyString) RubyUtil.RUBY.newString("\n"). -// freeze(RubyUtil.RUBY.getCurrentContext()); -// private @SuppressWarnings("rawtypes") RubyArray input = RubyUtil.RUBY.newArray(); -// private StringBuilder headToken = new StringBuilder(); -// private RubyString delimiter = NEW_LINE; -// private int sizeLimit; -// private boolean hasSizeLimit; -// private int inputSize; -// private boolean bufferFullErrorNotified = false; private String encodingName; private transient BufferedTokenizer tokenizer; @@ -59,19 +49,6 @@ public BufferedTokenizerExt(final Ruby runtime, final RubyClass metaClass) { @JRubyMethod(name = "initialize", optional = 2) public IRubyObject init(final ThreadContext context, IRubyObject[] args) { -// if (args.length >= 1) { -// this.delimiter = args[0].convertToString(); -// } -// if (args.length == 2) { -// final int sizeLimit = args[1].convertToInteger().getIntValue(); -// if (sizeLimit <= 0) { -// throw new IllegalArgumentException("Size limit must be positive"); -// } -// this.sizeLimit = sizeLimit; -// this.hasSizeLimit = true; -// } -// this.inputSize = 0; - String delimiter = "\n"; if (args.length >= 1) { delimiter = args[0].convertToString().asJavaString(); @@ -105,7 +82,7 @@ public IRubyObject extract(final ThreadContext context, IRubyObject data) { Iterable extractor = tokenizer.extract(data.asJavaString()); - + // return an iterator that does the encoding conversion IRubyObject rubyIterable = RubyUtil.toRubyObject(new Iterable() { @Override public Iterator iterator() { @@ -119,68 +96,6 @@ public CharSequence next() { }); return rubyIterable; - -// final RubyArray entities = data.convertToString().split(delimiter, -1); -// if (!bufferFullErrorNotified) { -// input.clear(); -// input.concat(entities); -// } else { -// // after a full buffer signal -// if (input.isEmpty()) { -// // after a buffer full error, the remaining part of the line, till next delimiter, -// // has to be consumed, unless the input buffer doesn't still contain fragments of -// // subsequent tokens. -// entities.shift(context); -// input.concat(entities); -// } else { -// // merge last of the input with first of incoming data segment -// if (!entities.isEmpty()) { -// RubyString last = ((RubyString) input.pop(context)); -// RubyString nextFirst = ((RubyString) entities.shift(context)); -// entities.unshift(last.concat(nextFirst)); -// input.concat(entities); -// } -// } -// } -// -// if (hasSizeLimit) { -// if (bufferFullErrorNotified) { -// bufferFullErrorNotified = false; -// if (input.isEmpty()) { -// return RubyUtil.RUBY.newArray(); -// } -// } -// final int entitiesSize = ((RubyString) input.first()).size(); -// if (inputSize + entitiesSize > sizeLimit) { -// bufferFullErrorNotified = true; -// headToken = new StringBuilder(); -// String errorMessage = String.format("input buffer full, consumed token which exceeded the sizeLimit %d; inputSize: %d, entitiesSize %d", sizeLimit, inputSize, entitiesSize); -// inputSize = 0; -// input.shift(context); // consume the token fragment that generates the buffer full -// throw new IllegalStateException(errorMessage); -// } -// this.inputSize = inputSize + entitiesSize; -// } -// -// if (input.getLength() < 2) { -// // this is a specialization case which avoid adding and removing from input accumulator -// // when it contains just one element -// headToken.append(input.shift(context)); // remove head -// return RubyUtil.RUBY.newArray(); -// } -// -// if (headToken.length() > 0) { -// // if there is a pending token part, merge it with the first token segment present -// // in the accumulator, and clean the pending token part. -// headToken.append(input.shift(context)); // append buffer to first element and -// // create new RubyString with the data specified encoding -// RubyString encodedHeadToken = toEncodedRubyString(context, headToken.toString()); -// input.unshift(encodedHeadToken); // reinsert it into the array -// headToken = new StringBuilder(); -// } -// headToken.append(input.pop(context)); // put the leftovers in headToken for later -// inputSize = headToken.length(); -// return input; } private RubyString toEncodedRubyString(ThreadContext context, String input) { @@ -213,31 +128,10 @@ public IRubyObject flush(final ThreadContext context) { } return RubyUtil.toRubyObject(s); } - -// final IRubyObject buffer = RubyUtil.toRubyObject(headToken.toString()); -// headToken = new StringBuilder(); -// inputSize = 0; -// -// // create new RubyString with the last data specified encoding, if exists -// RubyString encodedHeadToken; -// if (encodingName != null) { -// encodedHeadToken = toEncodedRubyString(context, buffer.toString()); -// } else { -// // When used with TCP input it could be that on socket connection the flush method -// // is invoked while no invocation of extract, leaving the encoding name unassigned. -// // In such case also the headToken must be empty -// if (!buffer.toString().isEmpty()) { -// throw new IllegalStateException("invoked flush with unassigned encoding but not empty head token, this shouldn't happen"); -// } -// encodedHeadToken = (RubyString) buffer; -// } -// -// return encodedHeadToken; } @JRubyMethod(name = "empty?") public IRubyObject isEmpty(final ThreadContext context) { -// return RubyUtil.RUBY.newBoolean(headToken.toString().isEmpty() && (inputSize == 0)); return RubyUtil.RUBY.newBoolean(tokenizer.isEmpty()); } From 32b05ebd85bb1aaf1b0c9be76470cc600e3fb540 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 5 Mar 2025 10:00:35 +0100 Subject: [PATCH 06/30] Moved creation of iterable upfront in the constructor to be executed just one time --- .../main/java/org/logstash/common/BufferedTokenizer.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 0f43348dbd..a8bd1f98f3 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -6,6 +6,7 @@ public class BufferedTokenizer { private final DataSplitter dataSplitter; + private final Iterable iterable; private Integer sizeLimit; static abstract class IteratorDecorator implements Iterator { @@ -99,6 +100,7 @@ public BufferedTokenizer() { public BufferedTokenizer(String separator) { this.dataSplitter = new DataSplitter(separator); + this.iterable = setupIterable(); } public BufferedTokenizer(String separator, int sizeLimit) { @@ -108,17 +110,22 @@ public BufferedTokenizer(String separator, int sizeLimit) { this.dataSplitter = new DataSplitter(separator); this.sizeLimit = sizeLimit; + this.iterable = setupIterable(); } public Iterable extract(String data) { dataSplitter.append(data); + return iterable; + } + + private Iterable setupIterable() { return new Iterable() { @Override public Iterator iterator() { Iterator returnedIterator = dataSplitter; if (sizeLimit != null) { - returnedIterator = new ValueLimitIteratorDecorator(returnedIterator, sizeLimit); + returnedIterator = new ValueLimitIteratorDecorator(returnedIterator, sizeLimit); } return returnedIterator; } From f4a9d9dd01aacd22c0e366f9a8084b7af99c323a Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 5 Mar 2025 12:24:29 +0100 Subject: [PATCH 07/30] Fixed tests to grab the failure on exceeded size limit on itereation once reached the next separator --- .../BufferedTokenizerWithSizeLimitTest.java | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index 780434d071..3937f4cce1 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -70,12 +70,18 @@ public void givenExtractedThrownLimitErrorWhenFeedFreshDataThenReturnTokenStarti public void givenExtractInvokedWithDifferentFramingAfterBufferFullErrorTWhenFeedFreshDataThenReturnTokenStartingFromEndOfOffendingToken() { sut.extract("aaaa"); + // it goes to 11 on a sizeLimit of 10, but doesn't trigger the exception till the next separator is reached + sut.extract("aaaaaaa").forEach(s -> {}); + + Iterable tokenIterable = sut.extract("aa\nbbbb\nccc"); Exception thrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract("aaaaaaa").forEach(s -> {}); + // now when querying and the next delimiter is present, the error is raised + tokenIterable.forEach(s -> {}); }); assertThat(thrownException.getMessage(), containsString("input buffer full")); - List tokens = toList(sut.extract("aa\nbbbb\nccc")); + // the iteration on token can proceed + List tokens = toList(tokenIterable); assertEquals(List.of("bbbb"), tokens); } @@ -83,15 +89,20 @@ public void givenExtractInvokedWithDifferentFramingAfterBufferFullErrorTWhenFeed public void giveMultipleSegmentsThatGeneratesMultipleBufferFullErrorsThenIsAbleToRecoverTokenization() { sut.extract("aaaa"); + // it goes to 11 on a sizeLimit of 10, but doesn't trigger the exception till the next separator is reached + sut.extract("aaaaaaa").forEach(s -> {}); + + Iterable tokenIterable = sut.extract("aa\nbbbbbbbbbbb\ncc"); + //first buffer full on 13 "a" letters Exception thrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract("aaaaaaa").forEach(s -> {}); + tokenIterable.forEach(s -> {}); }); assertThat(thrownException.getMessage(), containsString("input buffer full")); // second buffer full on 11 "b" letters Exception secondThrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract("aa\nbbbbbbbbbbb\ncc"); + tokenIterable.forEach(s -> {}); }); assertThat(secondThrownException.getMessage(), containsString("input buffer full")); From 3f5b82c43e940720b405f2ad57f8ba53a1fbe220 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 5 Mar 2025 12:46:46 +0100 Subject: [PATCH 08/30] Fixed license headers --- .../benchmark/BufferedTokenizerBenchmark.java | 19 ++++++++++++ logstash-core/lib/logstash/file_watch.rb | 3 -- logstash-core/lib/logstash/util.rb | 1 - .../logstash/common/BufferedTokenizer.java | 31 ++++++++++++++----- .../logstash/common/BufferedTokenizerExt.java | 13 ++------ .../common/BufferedTokenizerTest.java | 2 -- 6 files changed, 45 insertions(+), 24 deletions(-) delete mode 100644 logstash-core/lib/logstash/file_watch.rb diff --git a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java index eb0c1df8ff..b94df65eca 100644 --- a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java +++ b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java @@ -1,3 +1,22 @@ +/* + * 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; diff --git a/logstash-core/lib/logstash/file_watch.rb b/logstash-core/lib/logstash/file_watch.rb deleted file mode 100644 index a05a8549b5..0000000000 --- a/logstash-core/lib/logstash/file_watch.rb +++ /dev/null @@ -1,3 +0,0 @@ -module FileWatch - # java_import org.logstash.common.BufferedTokenizer -end \ No newline at end of file diff --git a/logstash-core/lib/logstash/util.rb b/logstash-core/lib/logstash/util.rb index 5e1f760f01..9eac95e1a2 100644 --- a/logstash-core/lib/logstash/util.rb +++ b/logstash-core/lib/logstash/util.rb @@ -16,7 +16,6 @@ # under the License. require "logstash/environment" -require "logstash/file_watch" module LogStash::Util UNAME = case RbConfig::CONFIG["host_os"] diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index a8bd1f98f3..9b39d97340 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -1,3 +1,21 @@ +/* + * 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.common; import java.util.Iterator; @@ -120,15 +138,12 @@ public Iterable extract(String data) { } private Iterable setupIterable() { - return new Iterable() { - @Override - public Iterator iterator() { - Iterator returnedIterator = dataSplitter; - if (sizeLimit != null) { - returnedIterator = new ValueLimitIteratorDecorator(returnedIterator, sizeLimit); - } - return returnedIterator; + return () -> { + Iterator returnedIterator = dataSplitter; + if (sizeLimit != null) { + returnedIterator = new ValueLimitIteratorDecorator(returnedIterator, sizeLimit); } + return returnedIterator; }; } diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java index a6456bf345..9658ebe1bf 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java @@ -83,19 +83,12 @@ public IRubyObject extract(final ThreadContext context, IRubyObject data) { Iterable extractor = tokenizer.extract(data.asJavaString()); // return an iterator that does the encoding conversion - IRubyObject rubyIterable = RubyUtil.toRubyObject(new Iterable() { + return RubyUtil.toRubyObject((Iterable) () -> new BufferedTokenizer.IteratorDecorator<>(extractor.iterator()) { @Override - public Iterator iterator() { - return new BufferedTokenizer.IteratorDecorator<>(extractor.iterator()) { - @Override - public CharSequence next() { - return toEncodedRubyString(context, iterator.next()); - } - }; + public CharSequence next() { + return toEncodedRubyString(context, iterator.next()); } }); - - return rubyIterable; } private RubyString toEncodedRubyString(ThreadContext context, String input) { diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java index 6499393b00..4c2045618e 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java @@ -23,12 +23,10 @@ import org.junit.Test; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; public final class BufferedTokenizerTest { From e40da7f83d738e13c8caea71fd4ba34be05d4fa0 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 5 Mar 2025 15:56:16 +0100 Subject: [PATCH 09/30] [Test] added test to verify buffer full error is notified not only on the first token --- .../BufferedTokenizerWithSizeLimitTest.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index 3937f4cce1..d7b01686a4 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -23,6 +23,7 @@ import org.junit.Before; import org.junit.Test; +import java.util.Iterator; import java.util.List; import static org.hamcrest.MatcherAssert.assertThat; @@ -110,4 +111,22 @@ public void giveMultipleSegmentsThatGeneratesMultipleBufferFullErrorsThenIsAbleT List tokens = toList(sut.extract("ccc\nddd\n")); assertEquals(List.of("ccccc", "ddd"), tokens); } + + @Test + public void givenFragmentThatHasTheSecondTokenOverrunsSizeLimitThenAnErrorIsThrown() { + Iterable tokensIterable = sut.extract("aaaa\nbbbbbbbbbbb\nccc\n"); + Iterator tokensIterator = tokensIterable.iterator(); + + // first token length = 4, it's ok + assertEquals("aaaa", tokensIterator.next()); + + // second token is an overrun, length = 11 + Exception exception = assertThrows(IllegalStateException.class, () -> { + tokensIterator.next(); + }); + assertThat(exception.getMessage(), containsString("input buffer full")); + + // third token resumes + assertEquals("ccc", tokensIterator.next()); + } } \ No newline at end of file From 49e4d9cae421520657a01ad5c319b9737659ab39 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 5 Mar 2025 16:42:22 +0100 Subject: [PATCH 10/30] Updated benchmark to consume effectively the iterator --- .../org/logstash/benchmark/BufferedTokenizerBenchmark.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java index b94df65eca..3512fdcc04 100644 --- a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java +++ b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java @@ -37,7 +37,7 @@ @Warmup(iterations = 3, time = 100, timeUnit = TimeUnit.MILLISECONDS) -@Measurement(iterations = 10, time = 100, timeUnit = TimeUnit.MILLISECONDS) +@Measurement(iterations = 10, time = 3000, timeUnit = TimeUnit.MILLISECONDS) @Fork(1) @BenchmarkMode(Mode.Throughput) @OutputTimeUnit(TimeUnit.NANOSECONDS) @@ -66,24 +66,29 @@ public void setUp() { @Benchmark public final void onlyOneTokenPerFragment(Blackhole blackhole) { Iterable tokens = sut.extract(singleTokenPerFragment); + tokens.forEach(blackhole::consume); blackhole.consume(tokens); } @Benchmark public final void multipleTokenPerFragment(Blackhole blackhole) { Iterable tokens = sut.extract(multipleTokensPerFragment); + tokens.forEach(blackhole::consume); blackhole.consume(tokens); } @Benchmark public final void multipleTokensCrossingMultipleFragments(Blackhole blackhole) { Iterable 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); } } From 84342a1acb3592d6ff0b789d4ce352fcd27dc701 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 6 Mar 2025 11:41:43 +0100 Subject: [PATCH 11/30] [Benchmark] JMH report in milliseconds instead of nanoseconds --- .../java/org/logstash/benchmark/BufferedTokenizerBenchmark.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java index 3512fdcc04..7eb6ea5105 100644 --- a/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java +++ b/logstash-core/benchmarks/src/main/java/org/logstash/benchmark/BufferedTokenizerBenchmark.java @@ -40,7 +40,7 @@ @Measurement(iterations = 10, time = 3000, timeUnit = TimeUnit.MILLISECONDS) @Fork(1) @BenchmarkMode(Mode.Throughput) -@OutputTimeUnit(TimeUnit.NANOSECONDS) +@OutputTimeUnit(TimeUnit.MILLISECONDS) @State(Scope.Thread) public class BufferedTokenizerBenchmark { From d5bc5435a672ffd2889b10a6b7cc306932174e16 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 12 Mar 2025 14:42:29 +0100 Subject: [PATCH 12/30] Added an isEmpty method to the iterable returned by BufferedTokenizerExt because it's expected in some use cases, like: https://github.com/logstash-plugins/logstash-input-file/blob/55a4a7099f05f29351672417036c1342850c7adc/lib/filewatch/watched_file.rb#L250 --- .../logstash/common/BufferedTokenizerExt.java | 24 +++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java index 9658ebe1bf..2863c94069 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java @@ -83,12 +83,32 @@ public IRubyObject extract(final ThreadContext context, IRubyObject data) { Iterable extractor = tokenizer.extract(data.asJavaString()); // return an iterator that does the encoding conversion - return RubyUtil.toRubyObject((Iterable) () -> new BufferedTokenizer.IteratorDecorator<>(extractor.iterator()) { + Iterator rubyStringAdpaterIterator = new BufferedTokenizer.IteratorDecorator<>(extractor.iterator()) { @Override public CharSequence next() { return toEncodedRubyString(context, iterator.next()); } - }); + }; + + return RubyUtil.toRubyObject(new IterableAdapterWithEmptyCheck(rubyStringAdpaterIterator)); + } + + // Iterator to Iterable adapter with addition of isEmpty method + public static class IterableAdapterWithEmptyCheck implements Iterable { + private final Iterator origIterator; + + public IterableAdapterWithEmptyCheck(Iterator origIterator) { + this.origIterator = origIterator; + } + + @Override + public Iterator iterator() { + return origIterator; + } + + public boolean isEmpty() { + return origIterator.hasNext(); + } } private RubyString toEncodedRubyString(ThreadContext context, String input) { From 57b6c0ccd9c125a1ace9264f1398858919ad8fef Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 26 Mar 2025 11:11:47 +0100 Subject: [PATCH 13/30] Aligned with main --- .../BufferedTokenizerWithSizeLimitTest.java | 33 ++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index d7b01686a4..864f89c5c5 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -34,11 +34,17 @@ public final class BufferedTokenizerWithSizeLimitTest { + public static final int GB = 1024 * 1024 * 1024; + private BufferedTokenizer sut; @Before public void setUp() { - sut = new BufferedTokenizer("\n", 10); + initSUTWithSizeLimit(10); + } + + private void initSUTWithSizeLimit(int sizeLimit) { + sut = new BufferedTokenizer("\n", sizeLimit); } @Test @@ -129,4 +135,29 @@ public void givenFragmentThatHasTheSecondTokenOverrunsSizeLimitThenAnErrorIsThro // third token resumes assertEquals("ccc", tokensIterator.next()); } + + @Test + public void givenTooLongInputExtractDoesntOverflow() { + assertEquals("Xmx must equals to what's defined in the Gradle's javaTests task", + 12L * GB, Runtime.getRuntime().maxMemory()); + + // re-init the tokenizer with big sizeLimit + initSUTWithSizeLimit((int) ((2L * GB) - 3)); + // Integer.MAX_VALUE is 2 * GB + String bigFirstPiece = generateString("a", Integer.MAX_VALUE - 1024); + sut.extract(bigFirstPiece); + + // add another small fragment to trigger int overflow + // sizeLimit is (2^32-1)-3 first segment length is (2^32-1) - 1024 second is 1024 +2 + // so the combined length of first and second is > sizeLimit and should throw an expection + // but because of overflow it's negative and happens to be < sizeLimit + Exception thrownException = assertThrows(IllegalStateException.class, () -> { + sut.extract(generateString("a", 1024 + 2)).iterator().next(); + }); + assertThat(thrownException.getMessage(), containsString("input buffer full")); + } + + private String generateString(String fill, int size) { + return fill.repeat(size); + } } \ No newline at end of file From c32578d9d24951b23219a5f348768c9ec0096103 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 31 Mar 2025 12:42:13 +0200 Subject: [PATCH 14/30] Removes a test that's not anymore valid. Not int math happens in code, but an OOM error is thrown from JDK libraries if an int overflow happens. --- .../BufferedTokenizerWithSizeLimitTest.java | 25 ------------------- 1 file changed, 25 deletions(-) diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index 864f89c5c5..5599f6e4f1 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -135,29 +135,4 @@ public void givenFragmentThatHasTheSecondTokenOverrunsSizeLimitThenAnErrorIsThro // third token resumes assertEquals("ccc", tokensIterator.next()); } - - @Test - public void givenTooLongInputExtractDoesntOverflow() { - assertEquals("Xmx must equals to what's defined in the Gradle's javaTests task", - 12L * GB, Runtime.getRuntime().maxMemory()); - - // re-init the tokenizer with big sizeLimit - initSUTWithSizeLimit((int) ((2L * GB) - 3)); - // Integer.MAX_VALUE is 2 * GB - String bigFirstPiece = generateString("a", Integer.MAX_VALUE - 1024); - sut.extract(bigFirstPiece); - - // add another small fragment to trigger int overflow - // sizeLimit is (2^32-1)-3 first segment length is (2^32-1) - 1024 second is 1024 +2 - // so the combined length of first and second is > sizeLimit and should throw an expection - // but because of overflow it's negative and happens to be < sizeLimit - Exception thrownException = assertThrows(IllegalStateException.class, () -> { - sut.extract(generateString("a", 1024 + 2)).iterator().next(); - }); - assertThat(thrownException.getMessage(), containsString("input buffer full")); - } - - private String generateString(String fill, int size) { - return fill.repeat(size); - } } \ No newline at end of file From 7111559a2ac522739c95ad3085379d3f5592cc45 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 10 Apr 2025 17:31:38 +0200 Subject: [PATCH 15/30] =?UTF-8?q?-=20inverted=20BufferedTokenizerExt=C2=A7?= =?UTF-8?q?IterableAdapterWithEmptyCheck#isEmpty=20-=20specs=20improved=20?= =?UTF-8?q?(yaauie@b524a67)=20with=20a=20custom=20matcher=20that=20validat?= =?UTF-8?q?es=20both=20`empty=3F`=20(which=20maps=20to=20isEmpty)=20and=20?= =?UTF-8?q?`entries`=20(which=20is=20provided=20by=20the=20jruby=20shim=20?= =?UTF-8?q?extending=20java-Iterator=20with=20RubyEnumerable)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../spec/logstash/util/buftok_spec.rb | 34 +++++++++++-------- .../logstash/common/BufferedTokenizerExt.java | 2 +- 2 files changed, 20 insertions(+), 16 deletions(-) diff --git a/logstash-core/spec/logstash/util/buftok_spec.rb b/logstash-core/spec/logstash/util/buftok_spec.rb index 4eceb20014..e5a93cce3c 100644 --- a/logstash-core/spec/logstash/util/buftok_spec.rb +++ b/logstash-core/spec/logstash/util/buftok_spec.rb @@ -20,33 +20,37 @@ describe FileWatch::BufferedTokenizer do subject { FileWatch::BufferedTokenizer.new } - def to_list(iterator) - a = [] - iterator.each { |v| a << v } - return a + + # 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(to_list(subject.extract("foo\n"))).to eq(["foo"]) + expect(subject.extract("foo\n")).to emit_exactly(["foo"]) end it "should merge multiple token" do - expect(to_list(subject.extract("foo"))).to eq([]) - expect(to_list(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(to_list(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(to_list(subject.extract(""))).to eq([]) - expect(to_list(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(to_list(subject.extract("\n"))).to eq([""]) - expect(to_list(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 @@ -89,12 +93,12 @@ def to_list(iterator) let(:delimiter) { "||" } it "should tokenize multiple token" do - expect(to_list(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(to_list(subject.extract(""))).to eq([]) - expect(to_list(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 diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java index 2863c94069..6db7a12d06 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizerExt.java @@ -107,7 +107,7 @@ public Iterator iterator() { } public boolean isEmpty() { - return origIterator.hasNext(); + return !origIterator.hasNext(); } } From 8e9d8e90902220abea5ec07ef7a602af26e6de50 Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 11 Apr 2025 15:27:58 +0200 Subject: [PATCH 16/30] Avoid to executed double token scan, in hasNext and next methods --- .../logstash/common/BufferedTokenizer.java | 42 ++++++++++++------- 1 file changed, 26 insertions(+), 16 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 9b39d97340..b97a68e825 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -61,6 +61,7 @@ public String next() { static class DataSplitter implements Iterator { private final String separator; private int currentIdx = 0; + private int nextSeparatorIdx = -1; private final StringBuilder accumulator = new StringBuilder(); DataSplitter(String separator) { @@ -69,28 +70,37 @@ static class DataSplitter implements Iterator { @Override public boolean hasNext() { - int nextIdx = accumulator.indexOf(separator, currentIdx); - if (nextIdx == -1) { - // not found next separator - cleanupAccumulator(); - return false; - } else { - return true; - } + return matchNextSeparatorIdx(); } @Override public String next() { - int nextIdx = accumulator.indexOf(separator, currentIdx); - if (nextIdx == -1) { - // not found next separator - cleanupAccumulator(); + if (!matchNextSeparatorIdx()) { throw new NoSuchElementException(); - } else { - String token = accumulator.substring(currentIdx, nextIdx); - currentIdx = nextIdx + separator.length(); - return token; } + + String token = accumulator.substring(currentIdx, nextSeparatorIdx); + currentIdx = nextSeparatorIdx + separator.length(); + nextSeparatorIdx = -1; + return token; + } + + /** + * Used to retrieve the index of the next token just one time. It saves into the nextSeparatorIdx + * and return true if a next token is present. + * Updates internal state for tracking. + * + * @return true iff a next complete token is available. + */ + private boolean matchNextSeparatorIdx() { + if (nextSeparatorIdx == -1) { + nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); + } + // clean up accumulator if no next separator found + if (nextSeparatorIdx == -1 && currentIdx > 0) { + cleanupAccumulator(); + } + return nextSeparatorIdx != -1; } private void cleanupAccumulator() { From 74e89930bf4aa35d8c567a339782955d989970a5 Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 11 Apr 2025 15:47:17 +0200 Subject: [PATCH 17/30] Emptied StringBuilder accumulator in flush and protected the state of DataSplitter with synchornized so that can be used in multithreaded contexts --- .../java/org/logstash/common/BufferedTokenizer.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index b97a68e825..42eb4cc890 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -69,12 +69,12 @@ static class DataSplitter implements Iterator { } @Override - public boolean hasNext() { + public synchronized boolean hasNext() { return matchNextSeparatorIdx(); } @Override - public String next() { + public synchronized String next() { if (!matchNextSeparatorIdx()) { throw new NoSuchElementException(); } @@ -113,11 +113,15 @@ public void append(String data) { } public String flush() { - return accumulator.substring(currentIdx); + final String flushed = accumulator.substring(currentIdx); + // empty the accumulator + accumulator.setLength(0); + currentIdx = 0; + return flushed; } @Override - public String toString() { + public synchronized String toString() { return "accumulator=" + accumulator + ", currentIdx=" + currentIdx; } } From 81fb197d8e5e12db1ef303c605b2e05d2ada9e1b Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 14 Apr 2025 09:54:35 +0200 Subject: [PATCH 18/30] [Test] Added test to verify that tokenizer condition empty is true when also unterminated token parts are present in the buffer --- .../org/logstash/common/BufferedTokenizerTest.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java index 4c2045618e..70058fc889 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerTest.java @@ -25,8 +25,7 @@ import java.util.ArrayList; import java.util.List; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; +import static org.junit.Assert.*; public final class BufferedTokenizerTest { @@ -83,4 +82,12 @@ public void shouldTokenizeEmptyPayloadWithNewline() { tokens = toList(sut.extract("\n\n\n")); assertEquals(List.of("", "", ""), tokens); } + + @Test + public void givenBufferWithTerminatedAndUnterminatedTokensWhenCheckingForEmptyThenReturnFalseIfUnterminatedTokenPartRemainInTheBuffer() { + List tokens = toList(sut.extract("foo\nbar\nbaz")); + assertEquals(List.of("foo", "bar"), tokens); + + assertFalse("Unterminated token makes the buffer to be considered non empty", sut.isEmpty()); + } } \ No newline at end of file From dc3d7b28e3e8e19fea8b7c188c219c06a1ae5d87 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 14 Apr 2025 10:17:58 +0200 Subject: [PATCH 19/30] Switched tokenizer implementation's empty to consider also the unterminated token in the buffer --- .../main/java/org/logstash/common/BufferedTokenizer.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 42eb4cc890..73c07b9443 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -120,6 +120,11 @@ public String flush() { return flushed; } + // considered empty if caught up to the accumulator + public synchronized boolean isBufferEmpty() { + return currentIdx >= accumulator.length(); + } + @Override public synchronized String toString() { return "accumulator=" + accumulator + ", currentIdx=" + currentIdx; @@ -171,6 +176,7 @@ public String toString() { } public boolean isEmpty() { - return !dataSplitter.hasNext(); +// return !dataSplitter.hasNext(); + return dataSplitter.isBufferEmpty(); } } From e2cd80f21eb2aeedcfa41a4fec0cdb73fa2d695c Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 14 Apr 2025 15:11:26 +0200 Subject: [PATCH 20/30] Minor, removed commented code --- .../src/main/java/org/logstash/common/BufferedTokenizer.java | 1 - 1 file changed, 1 deletion(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 73c07b9443..0bf4f2a8cc 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -176,7 +176,6 @@ public String toString() { } public boolean isEmpty() { -// return !dataSplitter.hasNext(); return dataSplitter.isBufferEmpty(); } } From 92c34374a744827b6759cd754a1c47dd92603d74 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 10 Mar 2025 12:10:47 +0100 Subject: [PATCH 21/30] [Test] Added test to verify the avoidance of infnite accumulation --- .../BufferedTokenizerWithSizeLimitTest.java | 31 +++++++++++++++++-- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index 5599f6e4f1..476036d4c2 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -27,9 +27,8 @@ 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.hamcrest.Matchers.*; +import static org.junit.Assert.*; import static org.logstash.common.BufferedTokenizerTest.toList; public final class BufferedTokenizerWithSizeLimitTest { @@ -135,4 +134,30 @@ public void givenFragmentThatHasTheSecondTokenOverrunsSizeLimitThenAnErrorIsThro // third token resumes assertEquals("ccc", tokensIterator.next()); } + + @Test + public void givenSequenceOfFragmentsWithoutSeparatorThenDoesntGenerateOutOfMemory() { + final String neverEndingData = generate(8, "a"); + for (int i = 0; i < 10; i++) { + // iterator has to be engaged + boolean hasNext = sut.extract(neverEndingData).iterator().hasNext(); + assertFalse(hasNext); + } + + // 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))); + + Iterable tokensIterable = sut.extract("\nbbb\n"); + Iterator tokensIterator = tokensIterable.iterator(); + // send a token delimiter and check an error is raised + Exception exception = assertThrows(IllegalStateException.class, () -> { + tokensIterator.next(); + }); + assertThat(exception.getMessage(), containsString("input buffer full")); + } + + private static String generate(int length, String fillChar) { + return fillChar.repeat(length); + } } \ No newline at end of file From 602bf9446249736c73f4d59fcf928cb60981f841 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 18 Jun 2025 11:36:26 +0200 Subject: [PATCH 22/30] Inloined PR #17293 to check for potential OOM condition during the append to the DataSplitter --- .../logstash/common/BufferedTokenizer.java | 63 +++++++++---------- 1 file changed, 30 insertions(+), 33 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 0bf4f2a8cc..03ca1d9145 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -20,12 +20,12 @@ import java.util.Iterator; import java.util.NoSuchElementException; +import java.util.function.IntPredicate; public class BufferedTokenizer { private final DataSplitter dataSplitter; private final Iterable iterable; - private Integer sizeLimit; static abstract class IteratorDecorator implements Iterator { protected final Iterator iterator; @@ -40,32 +40,28 @@ public boolean hasNext() { } } - static class ValueLimitIteratorDecorator extends IteratorDecorator { - private final int limit; - - ValueLimitIteratorDecorator(Iterator iterator, int sizeLimit) { - super(iterator); - this.limit = sizeLimit; - } - - @Override - public String next() { - String value = iterator.next(); - if (value.length() > limit) { - throw new IllegalStateException("input buffer full, consumed token which exceeded the sizeLimit " + limit); - } - return value; - } - } - static class DataSplitter implements Iterator { private final String separator; private int currentIdx = 0; private int nextSeparatorIdx = -1; private final StringBuilder accumulator = new StringBuilder(); + private boolean dropNextPartialFragments = false; + private final int sizeLimit; DataSplitter(String separator) { this.separator = separator; + this.sizeLimit = Integer.MIN_VALUE; + } + + /** + * @param separator + * is the token separator string. + * @param sizeLimit + * maximum token size length. + * */ + DataSplitter(String separator, int sizeLimit) { + this.separator = separator; + this.sizeLimit = sizeLimit; } @Override @@ -81,6 +77,9 @@ public synchronized String next() { String token = accumulator.substring(currentIdx, nextSeparatorIdx); currentIdx = nextSeparatorIdx + separator.length(); + if (sizeLimit != Integer.MIN_VALUE && accumulator.length() > sizeLimit) { + throw new IllegalStateException("input buffer full, consumed token which exceeded the sizeLimit " + sizeLimit); + } nextSeparatorIdx = -1; return token; } @@ -99,6 +98,11 @@ private boolean matchNextSeparatorIdx() { // clean up accumulator if no next separator found if (nextSeparatorIdx == -1 && currentIdx > 0) { cleanupAccumulator(); + // if it has a remaining bigger than the admitted size, then it start drop other next fragments that + // doesn't contain any separator + if (sizeLimit != Integer.MIN_VALUE && accumulator.length() > sizeLimit) { + dropNextPartialFragments = true; + } } return nextSeparatorIdx != -1; } @@ -109,6 +113,10 @@ private void cleanupAccumulator() { } public void append(String data) { + if (!data.contains(separator) && dropNextPartialFragments) { + return; + } + dropNextPartialFragments = false; accumulator.append(data); } @@ -137,7 +145,7 @@ public BufferedTokenizer() { public BufferedTokenizer(String separator) { this.dataSplitter = new DataSplitter(separator); - this.iterable = setupIterable(); + this.iterable = () -> dataSplitter; } public BufferedTokenizer(String separator, int sizeLimit) { @@ -145,9 +153,8 @@ public BufferedTokenizer(String separator, int sizeLimit) { throw new IllegalArgumentException("Size limit must be positive"); } - this.dataSplitter = new DataSplitter(separator); - this.sizeLimit = sizeLimit; - this.iterable = setupIterable(); + this.dataSplitter = new DataSplitter(separator, sizeLimit); + this.iterable = () -> dataSplitter; } public Iterable extract(String data) { @@ -156,16 +163,6 @@ public Iterable extract(String data) { return iterable; } - private Iterable setupIterable() { - return () -> { - Iterator returnedIterator = dataSplitter; - if (sizeLimit != null) { - returnedIterator = new ValueLimitIteratorDecorator(returnedIterator, sizeLimit); - } - return returnedIterator; - }; - } - public String flush() { return dataSplitter.flush(); } From 0051a13d7133763cf2e93b64381a46309e50210f Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 18 Jun 2025 16:37:46 +0200 Subject: [PATCH 23/30] Fixed matchNextSeparatorIdx to raise the flag to start dropping on write side also when there isn't yet done any read --- .../java/org/logstash/common/BufferedTokenizer.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 03ca1d9145..217b00ce5c 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -20,7 +20,6 @@ import java.util.Iterator; import java.util.NoSuchElementException; -import java.util.function.IntPredicate; public class BufferedTokenizer { @@ -77,7 +76,7 @@ public synchronized String next() { String token = accumulator.substring(currentIdx, nextSeparatorIdx); currentIdx = nextSeparatorIdx + separator.length(); - if (sizeLimit != Integer.MIN_VALUE && accumulator.length() > sizeLimit) { + if (sizeLimit != Integer.MIN_VALUE && token.length() > sizeLimit) { throw new IllegalStateException("input buffer full, consumed token which exceeded the sizeLimit " + sizeLimit); } nextSeparatorIdx = -1; @@ -103,6 +102,15 @@ private boolean matchNextSeparatorIdx() { if (sizeLimit != Integer.MIN_VALUE && accumulator.length() > sizeLimit) { dropNextPartialFragments = true; } + } else { + if (currentIdx > 0) { + nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); + } else { + // nextSeparatorIdx is -1 + if (sizeLimit != Integer.MIN_VALUE && accumulator.length() > sizeLimit) { + dropNextPartialFragments = true; + } + } } return nextSeparatorIdx != -1; } From b2c44b8f3b71114a08dd3be4b455ed993b487302 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 18 Jun 2025 16:45:40 +0200 Subject: [PATCH 24/30] [Test] fixed test, given the flush call the accumulator is emptied so no token to throw the exception --- .../common/BufferedTokenizerWithSizeLimitTest.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index 476036d4c2..dca966d481 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -150,11 +150,11 @@ public void givenSequenceOfFragmentsWithoutSeparatorThenDoesntGenerateOutOfMemor Iterable tokensIterable = sut.extract("\nbbb\n"); Iterator tokensIterator = tokensIterable.iterator(); - // send a token delimiter and check an error is raised - Exception exception = assertThrows(IllegalStateException.class, () -> { - tokensIterator.next(); - }); - assertThat(exception.getMessage(), containsString("input buffer full")); + // send a token delimiter and check is empty followed by 3 b's + String emptyToken = tokensIterator.next(); + assertTrue(emptyToken.isEmpty()); + String validToken = tokensIterator.next(); + assertEquals("bbb", validToken); } private static String generate(int length, String fillChar) { From 1f4193afe4433da8b1469e7fc2606ca1eb76f5fb Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 19 Jun 2025 11:13:53 +0200 Subject: [PATCH 25/30] Updated append method to avoid OOM accumulation without that token listener needs to be engaged --- .../logstash/common/BufferedTokenizer.java | 24 +++++++++++++++---- .../BufferedTokenizerWithSizeLimitTest.java | 7 +++--- 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 217b00ce5c..bd28152910 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -46,6 +46,7 @@ static class DataSplitter implements Iterator { private final StringBuilder accumulator = new StringBuilder(); private boolean dropNextPartialFragments = false; private final int sizeLimit; + private int lastFragmentSize = 0; DataSplitter(String separator) { this.separator = separator; @@ -76,7 +77,7 @@ public synchronized String next() { String token = accumulator.substring(currentIdx, nextSeparatorIdx); currentIdx = nextSeparatorIdx + separator.length(); - if (sizeLimit != Integer.MIN_VALUE && token.length() > sizeLimit) { + if (isSizeLimitSet() && token.length() > sizeLimit) { throw new IllegalStateException("input buffer full, consumed token which exceeded the sizeLimit " + sizeLimit); } nextSeparatorIdx = -1; @@ -121,13 +122,28 @@ private void cleanupAccumulator() { } public void append(String data) { - if (!data.contains(separator) && dropNextPartialFragments) { - return; + if (isSizeLimitSet()) { + if (!data.contains(separator) && lastFragmentSize > sizeLimit) { + // stop accumulating if last fragments already reached the sizeLimit + return; + } +// dropNextPartialFragments = false; + + // 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(); + } else { + lastFragmentSize = data.length() - (lastSeparatorIdx + separator.length()); + } } - dropNextPartialFragments = false; accumulator.append(data); } + private boolean isSizeLimitSet() { + return sizeLimit != Integer.MIN_VALUE; + } + public String flush() { final String flushed = accumulator.substring(currentIdx); // empty the accumulator diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index dca966d481..72d24d110a 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -139,9 +139,10 @@ public void givenFragmentThatHasTheSecondTokenOverrunsSizeLimitThenAnErrorIsThro public void givenSequenceOfFragmentsWithoutSeparatorThenDoesntGenerateOutOfMemory() { final String neverEndingData = generate(8, "a"); for (int i = 0; i < 10; i++) { - // iterator has to be engaged - boolean hasNext = sut.extract(neverEndingData).iterator().hasNext(); - assertFalse(hasNext); + sut.extract(neverEndingData); +// // iterator has to be engaged +// boolean hasNext = sut.extract(neverEndingData).iterator().hasNext(); +// assertFalse(hasNext); } // with the second fragment passed to extract it overrun the sizeLimit, the tokenizer From 983a5eadb6f5ff95c5cb4ea7802561ea76fd3dfc Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 19 Jun 2025 11:26:44 +0200 Subject: [PATCH 26/30] Removed unused code and expanded * imports in tests --- .../java/org/logstash/common/BufferedTokenizer.java | 12 ------------ .../common/BufferedTokenizerWithSizeLimitTest.java | 10 ++++++---- 2 files changed, 6 insertions(+), 16 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index bd28152910..2edfb7e762 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -44,7 +44,6 @@ static class DataSplitter implements Iterator { private int currentIdx = 0; private int nextSeparatorIdx = -1; private final StringBuilder accumulator = new StringBuilder(); - private boolean dropNextPartialFragments = false; private final int sizeLimit; private int lastFragmentSize = 0; @@ -98,19 +97,9 @@ private boolean matchNextSeparatorIdx() { // clean up accumulator if no next separator found if (nextSeparatorIdx == -1 && currentIdx > 0) { cleanupAccumulator(); - // if it has a remaining bigger than the admitted size, then it start drop other next fragments that - // doesn't contain any separator - if (sizeLimit != Integer.MIN_VALUE && accumulator.length() > sizeLimit) { - dropNextPartialFragments = true; - } } else { if (currentIdx > 0) { nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); - } else { - // nextSeparatorIdx is -1 - if (sizeLimit != Integer.MIN_VALUE && accumulator.length() > sizeLimit) { - dropNextPartialFragments = true; - } } } return nextSeparatorIdx != -1; @@ -127,7 +116,6 @@ public void append(String data) { // stop accumulating if last fragments already reached the sizeLimit return; } -// dropNextPartialFragments = false; // 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); diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index 72d24d110a..2cf552ff51 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -27,14 +27,16 @@ import java.util.List; import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.*; -import static org.junit.Assert.*; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.lessThan; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import static org.logstash.common.BufferedTokenizerTest.toList; public final class BufferedTokenizerWithSizeLimitTest { - public static final int GB = 1024 * 1024 * 1024; - private BufferedTokenizer sut; @Before From 1c67ff64bf52951df32246040126913a6d8479f6 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 19 Jun 2025 12:05:48 +0200 Subject: [PATCH 27/30] [Test] remove commented code in test --- .../logstash/common/BufferedTokenizerWithSizeLimitTest.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java index 2cf552ff51..09ff7f8722 100644 --- a/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java +++ b/logstash-core/src/test/java/org/logstash/common/BufferedTokenizerWithSizeLimitTest.java @@ -142,9 +142,6 @@ public void givenSequenceOfFragmentsWithoutSeparatorThenDoesntGenerateOutOfMemor final String neverEndingData = generate(8, "a"); for (int i = 0; i < 10; i++) { sut.extract(neverEndingData); -// // iterator has to be engaged -// boolean hasNext = sut.extract(neverEndingData).iterator().hasNext(); -// assertFalse(hasNext); } // with the second fragment passed to extract it overrun the sizeLimit, the tokenizer From 69764519b9cd56f95aca78b3accb3606ff3d7d35 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 19 Jun 2025 12:08:59 +0200 Subject: [PATCH 28/30] Lifted a else branch into the first if statement --- .../main/java/org/logstash/common/BufferedTokenizer.java | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 2edfb7e762..9a4a271fcd 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -93,15 +93,16 @@ public synchronized String next() { private boolean matchNextSeparatorIdx() { if (nextSeparatorIdx == -1) { nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); - } - // clean up accumulator if no next separator found - if (nextSeparatorIdx == -1 && currentIdx > 0) { - cleanupAccumulator(); } else { if (currentIdx > 0) { nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); } } + + // clean up accumulator if no next separator found + if (nextSeparatorIdx == -1 && currentIdx > 0) { + cleanupAccumulator(); + } return nextSeparatorIdx != -1; } From b3d408949ec1450b864781aa6cb05f5d8b053b69 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 19 Jun 2025 12:13:08 +0200 Subject: [PATCH 29/30] Removed not necessary condition, because if currentIdx is unset (min int) then the indexOf clamp it at least at 0. This test is useful only when the first delimiter is not yet found --- .../src/main/java/org/logstash/common/BufferedTokenizer.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index 9a4a271fcd..a676882def 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -94,9 +94,7 @@ private boolean matchNextSeparatorIdx() { if (nextSeparatorIdx == -1) { nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); } else { - if (currentIdx > 0) { - nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); - } + nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); } // clean up accumulator if no next separator found From 116288225b928916c94a00be291826559897e4ab Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 19 Jun 2025 12:14:22 +0200 Subject: [PATCH 30/30] Removed redundant if statement --- .../main/java/org/logstash/common/BufferedTokenizer.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java index a676882def..5aa513922a 100644 --- a/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java +++ b/logstash-core/src/main/java/org/logstash/common/BufferedTokenizer.java @@ -91,11 +91,7 @@ public synchronized String next() { * @return true iff a next complete token is available. */ private boolean matchNextSeparatorIdx() { - if (nextSeparatorIdx == -1) { - nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); - } else { - nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); - } + nextSeparatorIdx = accumulator.indexOf(separator, currentIdx); // clean up accumulator if no next separator found if (nextSeparatorIdx == -1 && currentIdx > 0) {