mapStateful / mapStatefulConcat initialise state once per construction, not per run

[Mendrejk]

Problem

Flow.mapStateful and Flow.mapStatefulConcat call initializeState when the operator is constructed, then reuse that state for every subsequent execution. Repeated or partially consumed runs can therefore corrupt results.

Impact

Any operator built on these APIs, including linesUtf8, can leak framing or aggregation state across runs/cancellations.

Expected

Initialize state once per execution of the returned flow.

Suggested tests

• Repeated full runs of a stateful flow produce identical results.
• A cancelled/partial first run does not affect a second run.