Flink watermarks not advancing in Python, stuck at -9223372036854775808

Question

I have encountered this issue with several pipelines and haven't been able to find an answer. When running a pipeline with a watermark strategy assigned for either monotonous or out of bounds timestamps with a timestamp assigner, the timestamp is extracted correctly and is advancing but the watermark is stuck at -9223372036854775808. I tried to run the event_time_timer.py example in the pyflink library as a sanity check, but upon inspection both process_element and on_timer methods have not moving watermarks of -9223372036854775808. and 9223372036854775807 respectively.

This is the code for the process function and timestamp assigners:

class Sum(KeyedProcessFunction):

    def __init__(self):
        self.state = None

    def open(self, runtime_context: RuntimeContext):
        state_descriptor = ValueStateDescriptor("state", Types.FLOAT())
        state_ttl_config = StateTtlConfig \
            .new_builder(Time.seconds(1)) \
            .set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
            .disable_cleanup_in_background() \
            .build()
        state_descriptor.enable_time_to_live(state_ttl_config)
        self.state = runtime_context.get_state(state_descriptor)

    def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):
        # retrieve the current count
        current = self.state.value()
        if current is None:
            current = 0

        # update the state's count
        current += value[2]
        self.state.update(current)

        # register an event time timer 2 seconds later
        ctx.timer_service().register_event_time_timer(ctx.timestamp() + 2000)

    def on_timer(self, timestamp: int, ctx: 'KeyedProcessFunction.OnTimerContext'):
        yield ctx.get_current_key(), self.state.value(), timestamp


class MyTimestampAssigner(TimestampAssigner):

    def extract_timestamp(self, value, record_timestamp: int) -> int:
        return int(value[0])

This is the main function:

def event_timer_timer_demo():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
    env.set_parallelism(1)
    env.get_config().set_auto_watermark_interval(1)

    ds = env.from_collection(
        collection=[
            (1000, 'Alice', 110.1),
            (4000, 'Bob', 30.2),
            (3000, 'Alice', 20.0),
            (2000, 'Bob', 53.1),
            (5000, 'Alice', 13.1),
            (3000, 'Bob', 3.1),
            (7000, 'Bob', 16.1),
            (10000, 'Alice', 20.1)
        ],
        type_info=Types.TUPLE([Types.LONG(), Types.STRING(), Types.FLOAT()]))
    ds = ds.assign_timestamps_and_watermarks(
        WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(2))
                         .with_timestamp_assigner(MyTimestampAssigner()))


    # apply the process function onto a keyed stream
    ds.key_by(lambda value: value[1]) \
      .process(Sum()) \
      .print()

    # submit for execution
    env.execute()

I have the same problem in my main pipeline, regardless of which watermark strategy I use. Shouldn't the watermarks be linked to timestamps and be seen inside ProcessFunction.Context?

Answer 1

This problem is mainly because Source is a BOUNDED Source. The execution of the entire Flink Job is over before the WatermarkStrategy is triggered.

You can refer to the following example to generate User records instead of fromCollection

/** Data-generating source function. */
public static final class Generator
        implements SourceFunction<Tuple2<Integer, Integer>>, CheckpointedFunction {

    private static final long serialVersionUID = -2819385275681175792L;

    private final int numKeys;
    private final int idlenessMs;
    private final int recordsToEmit;

    private volatile int numRecordsEmitted = 0;
    private volatile boolean canceled = false;

    private ListState<Integer> state = null;

    Generator(final int numKeys, final int idlenessMs, final int durationSeconds) {
        this.numKeys = numKeys;
        this.idlenessMs = idlenessMs;

        this.recordsToEmit = ((durationSeconds * 1000) / idlenessMs) * numKeys;
    }

    @Override
    public void run(final SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
        while (numRecordsEmitted < recordsToEmit) {
            synchronized (ctx.getCheckpointLock()) {
                for (int i = 0; i < numKeys; i++) {
                    ctx.collect(Tuple2.of(i, numRecordsEmitted));
                    numRecordsEmitted++;
                }
            }
            Thread.sleep(idlenessMs);
        }

        while (!canceled) {
            Thread.sleep(50);
        }
    }

    @Override
    public void cancel() {
        canceled = true;
    }
}

}