Paul King created FLINK-41296:
---------------------------------
Summary: fullWindowPartition().mapPartition() corrupts output: the
end-of-input watermark is emitted on the task thread while the UDF thread is
still emitting records
Key: FLINK-41296
URL: https://issues.apache.org/jira/browse/FLINK-41296
Project: Flink
Issue Type: Bug
Components: API / DataStream, Runtime / Task
Affects Versions: 1.20.5, 2.2.1, 2.3.0, 2.0.1
Reporter: Paul King
Description:
In BATCH mode, a non-keyed {{fullWindowPartition().mapPartition(...)}} whose
function emits records while it iterates over its input, followed by a network
exchange such as {{keyBy}}, regularly fails with errors from the result
partition, returns wrong results, or hangs.
Minimal reproduction (Java, no other libraries):
{code:java}
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
env.setParallelism(4);
DataStream<Integer> s = env.fromData(1, 2, 3, 4, 5, 6, 7, 8)
.fullWindowPartition()
.mapPartition(new MapPartitionFunction<Integer, Integer>() {
public void mapPartition(Iterable<Integer> in, Collector<Integer> out) {
for (Integer i : in) out.collect(i);
}
})
.keyBy(i -> i % 2)
.reduce((a, b) -> a + b);
System.out.println(s.executeAndCollect(100)); // expected [20, 16]
{code}
Run on a local MiniCluster (JDK 17 and 21):
* 1.20.5: 10 of 10 runs failed at parallelism 4; failures also seen at
parallelism 1, 2, 8 and 12, and one run returned [].
* 2.3.0: 5 of 6 runs failed. Also reproduced on 2.0.1 and 2.2.1.
Typical failures:
* {{java.lang.IllegalArgumentException: newPosition > limit: (17 > 9)}}
* {{java.nio.BufferUnderflowException}}
* {{java.lang.IllegalArgumentException: Cannot append empty data.}} thrown on
{{TaskUDFExecutor-thread-1}}, which then stops the process through
{{FatalExitExceptionHandler}}
* wrong or corrupted results, e.g. [20, 15], or values such as 16777216;
occasional hangs
Cause:
{{MapPartitionOperator}} runs the user function on a separate
{{TaskUDFExecutor}} thread, created by {{MapPartitionIterator}}. The function's
{{Collector}} is a {{TimestampedCollector}} over the operator's {{output}}, so
records are written to the {{RecordWriter}} from that thread, outside any lock.
{{MapPartitionOperator}} overrides only {{processElement}} and {{endInput}}.
Other input events use the {{AbstractStreamOperator}} defaults, which forward
them to the same output on the task thread. In BATCH mode the final
{{MAX_WATERMARK}} arrives before {{endInput}}, while the function is typically
still emitting the last records, so two threads write to the same
{{RecordWriter}} / {{SortMergeResultPartition}} at once. The failing stack is
on the task thread:
{code}
java.lang.IllegalArgumentException: newPosition > limit: (17 > 9)
at org.apache.flink.core.memory.MemorySegment.put(MemorySegment.java:1416)
at
org.apache.flink.runtime.io.network.buffer.BufferBuilder.append(BufferBuilder.java:110)
at
org.apache.flink.runtime.io.network.partition.HashBasedDataBuffer.writeRecord(HashBasedDataBuffer.java:199)
at
org.apache.flink.runtime.io.network.partition.HashBasedDataBuffer.append(HashBasedDataBuffer.java:151)
at
org.apache.flink.runtime.io.network.partition.SortMergeResultPartition.emit(SortMergeResultPartition.java:262)
at
org.apache.flink.runtime.io.network.partition.SortMergeResultPartition.emitRecord(SortMergeResultPartition.java:223)
at
org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:108)
at
org.apache.flink.streaming.runtime.io.RecordWriterOutput.emitWatermark(RecordWriterOutput.java:167)
at
org.apache.flink.streaming.api.operators.CountingOutput.emitWatermark(CountingOutput.java:44)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.emitWatermarkDirectly(AbstractStreamOperator.java:679)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.processWatermark(AbstractStreamOperator.java:671)
at
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitWatermark(OneInputStreamTask.java:244)
at
org.apache.flink.streaming.runtime.watermarkstatus.StatusWatermarkValve.findAndOutputNewMinWatermarkAcrossAlignedSubpartitions(StatusWatermarkValve.java:283)
...
at
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:638)
{code}
{{NonKeyedPartitionWindowedStream.mapPartition}} already calls
{{disableChaining()}}, with the comment that the operator "doesn't use task
main thread to write records to output, which may introduce risks to downstream
chained operators". That protects chained operators, but not the operator's own
network output, which the task thread still writes to.
Why it is easy to miss:
A function that first consumes its whole input and only then emits (for
example, aggregating a partition) is not affected. Its {{hasNext()}} only
returns false after {{endInput}} has closed the iterator, by which time the
watermark has already been forwarded. Changing the function above to collect
into a list and emit after the loop passed 10 of 10 runs on 1.20.5 and 6 of 6
on 2.3.0. Only functions that emit while iterating (map- or filter-like) race
with the watermark.
As a further check, a subclass of {{MapPartitionOperator}} that holds back
{{processWatermark}} until after {{super.endInput()}} did not fail in 10 runs,
against 2 failures in 10 for the same subclass forwarding the watermark
normally. The subclass's instrumentation changes the timing, so this is
supporting evidence only; the stack trace above is the direct evidence.
Possible fixes (for the maintainers to judge):
* Hold back watermarks (and watermark status, record attributes, and 2.x
generalized watermarks) in {{MapPartitionOperator}} until the function has
finished in {{endInput}}. This fits the operator's
{{setOutputOnlyAfterEndOfStream(true)}} attribute, which its output timing
currently contradicts, since records are emitted before end of input.
* Or have records emitted by the function handed back to the task thread (e.g.
via the mailbox), so that only the task thread writes to the output.
Not checked: the keyed variant ({{KeyedPartitionWindowedStream}}).
Related: FLINK-34543 (added the feature, PR #24398). Other open issues in this
area (FLINK-39034, FLINK-38728) are different problems.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
