BufferedTokenizer can lead json_lines to corrupt data when used in http-input with pipelining requests

[andsel]

Logstash information:

Please include the following information:

1. Logstash version 9.2.0
2. Logstash installation source (e.g. built from source, with a package manager: DEB/RPM, expanded from tar or zip archive, docker)
3. How is Logstash being run (e.g. as a service/service manager: systemd, upstart, etc. Via command line, docker/kubernetes)

Plugins installed: (bin/logstash-plugin list --verbose)

JVM (e.g. java -version):

If the affected version of Logstash is 7.9 (or earlier), or if it is NOT using the bundled JDK or using the 'no-jdk' version in 7.10 (or higher), please provide the following information:

1. JVM version (java -version)
2. JVM installation source (e.g. from the Operating System's package manager, from source, etc).
3. Value of the LS_JAVA_HOME environment variable if set.

OS version (uname -a if on a Unix-like system):

Description of the problem including expected versus actual behavior:
BufferedTokenizer is a class used to split incoming data using a separator. It accumulates data and when read provides a serie of diced strings, using the separator as cutting point.
It's used in various context, in particular inside the json_lines codec, where each instance of the codec has it's own BufferedTokenizer and should be accessed by a single thread, because the codec are cloned per connection, and usually Netty access the connection by a single thread. However, this is not true for HTTP input when used with pipelined http requests.

Each TCP connection has its own copy of the codec:

1. a messahe_handler has a codec that uses to decode incoming data https://github.com/logstash-plugins/logstash-input-http/blob/v4.1.7/lib/logstash/inputs/http/message_handler.rb#L30
2. the message handler is cloned on every net TCP connection, cloning also the codecs, and so the BufferedTokenizer:
   https://github.com/logstash-plugins/logstash-input-http/blob/v4.1.7/src/main/java/org/logstash/plugins/inputs/http/HttpInitializer.java#L51
   3 BUT the same message_handler instance is used by multiple threads from the same HTTP connection https://github.com/logstash-plugins/logstash-input-http/blob/v4.1.7/src/main/java/org/logstash/plugins/inputs/http/HttpServerHandler.java#L39

So while it's safe to use the json_lines on non pipelined request-response, where each request is received by a new TCP connection, it's not safe when the same connection is used to send multiple requests.

The original #17229 introduced the class as rewrite of a previous JRuby extension, and lately synchronized with #19345, but that's not enough for full concurrent access because multiple threads can interleave the method callsaccept and flush or next and hasNext. The concurrent section has to be considered accept+flushand next+hasNext.

Concerns about the legitimacy

HTTP 1.1 protocol has the keep-alive mode and also permit pipelining of requests (which seems discouraged).

1. keep-alive: use a persistent connection to send multiple request-response, after every request should be sent back the response, no queueing.
2. pipelining: doesn't require any special header, just a persistent connection. On the connection a sequence of requests is sent, back to back. Then a sequence of responses should be sent back, respecting the sequence order of the request, so to a req1, req2, req3, should correspond a resp1, resp2, resp3.

In case 1 HTTP input, json_lines and ultimately BufferedTokenizer works as expected because MessageProcessors are created once at a time and processed serially by https://github.com/logstash-plugins/logstash-input-http/blob/v4.1.13/src/main/java/org/logstash/plugins/inputs/http/HttpServerHandler.java#L39. This is because the client send one request at a time.

In case 2 when multiple requests are in flight, many MessageProcessors are created, one per HTTP request. The problem is that the execution order doesn't respect the arrival order. Req1, req2 at TCP connection level could be processed at MessageProcessor2 and MessageProcessor1. The plugin doesn't have control to keep sequence order, in the execution and in the responses and this is an issue in the protocol implementation. This lack of serialization on the processing of request, is what generated the problem on the tokenizer, which could led to a mess up of the data.
Suppose that json lines are split across multiple request, so that req1 carries the following payload:

{"id": "first json payload"}\r\n
{"id": "second

and req2 is:

 json payload"}\r\n
{"id": "third json payloaf"}

If request is processed in order than the tokenizer will produce the expected chunks, else if the order of processing is inverted a message is corrupted and it delivers successfully only 2 events.

Steps to reproduce:

The gist https://gist.github.com/donoghuc/c60336eed3b897a6b4d3f922eec0f70d has a full reproduction.

Provide logs (if relevant):