I'm encountering similar issue to [Flink EventTime Processing Watermark is always coming as -9223372036854725808](https://stackoverflow.com/questions/60487571/flink-eventtime-processing-watermark-is-always-coming-as-9223372036854725808) However, the suggested solutions (set parallelism and disable checkpointing) do not have any effect. In this example, I'm simply streaming 1000 events 1 second apart, and then comparing the event timestamp to ctx.timerService().currentWatermark()

```
>>> v=(61538659200000,0), watermark=-9223372036854775808
>>> v=(61538659201000,1), watermark=-9223372036854775808
>>> v=(61538660198000,998), watermark=-9223372036854775808
>>> v=(61538660199000,999), watermark=-9223372036854775808

public void watermarks()
    throws Exception
{
    final var env = StreamExecutionEnvironment.createLocalEnvironment();
    env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    env.setMaxParallelism(1);

    final long startMs = new Date(2020, 1, 1).getTime();
    final var events = new ArrayList<Tuple2<Long, Integer>>();
    for (var ii = 0; ii < 1000; ++ii ) {
        events.add(new Tuple2<Long, Integer>(startMs + ii * 1000, ii));
    }

    env.fromCollection(events)
        .assignTimestampsAndWatermarks(
            WatermarkStrategy.<Tuple2<Long, Integer>>forMonotonousTimestamps()
                .withTimestampAssigner((event, ts) -> event.f0))
        .setParallelism(1)
        .keyBy(row -> row.f1 % 2)
        .process(new ProcessFunction<Tuple2<Long, Integer>, String>()
        {
            @Override
            public void processElement(
                final Tuple2<Long, Integer> value,
                final Context ctx,
                final Collector<String> out)
                throws Exception
            {
                out.collect("v=" + value + ", watermark=" + ctx.timerService().currentWatermark());
            }
        })
        .setParallelism(1)
        .print()
        .setParallelism(1);
    final var result = env.execute();
    System.out.println(result);
}

```