Mate Czagany created FLINK-41268:
------------------------------------
Summary: Source idleness lets a watermark overtake on-time records
after two shuffles
Key: FLINK-41268
URL: https://issues.apache.org/jira/browse/FLINK-41268
Project: Flink
Issue Type: Bug
Components: Runtime / Task
Affects Versions: 2.3.0
Reporter: Mate Czagany
With source idleness enabled and parallelism above 1, a record that is on time
at its source can reach a downstream event-time operator after a watermark
equal to its own timestamp. The operator then treats it as late and, depending
on the operator, drops or delays it.
h3. How it happens
Take a source subtask S, two subtasks A and B after a first exchange, and an
event-time operator X after a second exchange:
{noformat}
S -> A (gets record r) -> X
S -> B (never sees r) -> X
{noformat}
# S has been idle, so A and B announced IDLE, and X leaves both of their
channels out of its minimum watermark ({{{}StatusWatermarkValve{}}}).
# S reads r with timestamp T. It sends r to A only, but ACTIVE and the
watermark W(T) to both A and B.
# A has to process r before it gets to ACTIVE and W(T). B has nothing to
process and forwards ACTIVE and W(T) right away.
# If B's W(T) reaches X while X still treats A's channel as idle, X advances
to T, and r arrives with ts = T <= watermark.
Each channel keeps its order, but nothing orders the two paths against each
other. Which message wins depends on output buffer flushes, processing time in
A, GC pauses or backpressure.
Conditions:
* Idleness is enabled ({{{}WatermarkStrategy#withIdleness{}}},
{{table.exec.source.idle-timeout}} or a connector option such as
{{{}scan.watermark.idle-timeout{}}}).
* There are two all-to-all exchanges between the source and the event-time
operator, with parallelism above 1 between them. With a single exchange, the
record and its watermark share a channel.
* The watermark that covers r arrives with r. That's the case for a zero-delay
watermark such as {{WATERMARK FOR ts AS ts}} on a sparse stream. With an
out-of-orderness bound D, a burst after idleness that spans more than D should
be enough as well: a later record of the burst goes through B, and its
watermark covers r.
* It doesn't depend on the connector. The reproducer uses the datagen source,
and the original report used Kafka.
h3. Impact
When the race hits:
* unbounded OVER aggregations drop the record, on all versions
* the event-time temporal join delays it until the next watermark up to 2.3,
and drops it from 2.4.0 on (FLINK-39719)
* process table functions with ORDER BY drop it
* windows drop it only if its timestamp is the window's last millisecond
h3. Reproducer
The test makes the ordering deterministic:
* The source emits record 0 at startup and record 1 about 2 s later, and is
idle in between. Each record's timestamp is its value, and its watermark equals
its timestamp, like {{{}WATERMARK FOR ts AS ts{}}}.
* {{keyBy}} sends record 1 to one of two subtasks (A). Its watermark goes to
both.
* A's map is slow and takes 200 ms per record. B forwards the watermark within
one output flush, which is at most 100 ms by default.
On master (227b2c4a990) it always fails with {{{}Expecting empty but was:
[1L]{}}}: B's watermark reached X while A was still processing record 1, and X
advanced. Record 0 is sent before anything is idle and always arrives on time.
The test belongs in {{{}flink-connector-datagen-test{}}}, which has both
{{DataGeneratorSource}} and {{{}MiniClusterExtension{}}}.
{code:java}
package org.apache.flink.connector.datagen.source;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.test.junit5.MiniClusterExtension;
import org.apache.flink.util.Collector;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import static org.assertj.core.api.Assertions.assertThat;
/**
* A source emits two records 2 s apart and is idle in between. Its watermark
equals the record's
* timestamp. The first exchange sends record 1 to one of two subtasks, A, and
its watermark to
* both. A's map is slow, so the other subtask forwards the watermark to the
event-time operator X
* first. X still treats A as idle, so it advances, and record 1 arrives late.
*/
class IdleSourceWatermarkRaceITCase {
@RegisterExtension
private static final MiniClusterExtension MINI_CLUSTER =
new MiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberSlotsPerTaskManager(2)
.build());
private static final List<Long> LATE_AT_X = new CopyOnWriteArrayList<>();
@Test
void testWatermarkDoesNotOvertakeRecordAfterIdleness() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.fromSource(
// One record every 2 s, so that every task is idle
before record 1.
new DataGeneratorSource<>(
i -> i, 2, RateLimiterStrategy.perSecond(0.5),
Types.LONG),
WatermarkStrategy.<Long>forGenerator(ctx -> new
ZeroDelayWatermarks())
.withTimestampAssigner((i, ignored) -> i)
.withIdleness(Duration.ofMillis(300)),
"source")
.setParallelism(1)
.keyBy(i -> i)
.map(
i -> {
Thread.sleep(200);
return i;
})
.setParallelism(2)
.keyBy(i -> i)
.process(new LateRecordProbe())
.setParallelism(1);
env.execute();
assertThat(LATE_AT_X).as("records that reached X at or below its
watermark").isEmpty();
}
/** Watermark equal to the record's timestamp, like {@code WATERMARK FOR ts
AS ts} in SQL. */
private static class ZeroDelayWatermarks implements
WatermarkGenerator<Long> {
@Override
public void onEvent(Long event, long timestamp, WatermarkOutput output)
{
output.emitWatermark(new Watermark(timestamp));
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {}
}
/** The event-time operator X. */
private static class LateRecordProbe extends KeyedProcessFunction<Long,
Long, Long> {
@Override
public void processElement(Long value, Context ctx, Collector<Long>
out) {
if (ctx.timestamp() <= ctx.timerService().currentWatermark()) {
LATE_AT_X.add(value);
}
}
}
}
{code}
h3. Possible fixes
Discussed on mailing list:
[https://lists.apache.org/thread/n2p0zt28k90n4bbnjdnwbdb0hx7n0ns5]
* Flushing the network output when a task announces ACTIVE narrows the window.
It doesn't help when A is slow: with A stalled on the record, records are still
late.
* A complete fix likely needs a FLIP. Two options are being discussed:
** resume markers, which let X keep A's channel in its minimum until it
catches up
** borrowed watermarks, where idleness stays inside the source and an idle
reader emits a watermark it gets from the source coordinator
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
