[jira] [Created] (FLINK-40829) operators hidden by `assign_timestamps_and_watermarks` get no uid, breaking `pipeline.auto-generate-uids: false`

["Florian Vazelle (Jira)" <ji...@apache.org>]

Florian Vazelle created FLINK-40829:
---------------------------------------

             Summary: operators hidden by `assign_timestamps_and_watermarks` get no uid, breaking `pipeline.auto-generate-uids: false`
                 Key: FLINK-40829
                 URL: https://issues.apache.org/jira/browse/FLINK-40829
             Project: Flink
          Issue Type: Bug
          Components: API / Python
            Reporter: Florian Vazelle


{{DataStream.assign_timestamps_and_watermarks()}} expands a *Python* {{TimestampAssigner}} into three transformations, but returns a {{DataStream}} which represents only the last one:
{noformat}
Extract-Timestamp (python)  ->  Timestamps/Watermarks (java)  ->  Remove-Timestamp (java map){noformat}
A uid set on the returned stream reaches {{Remove-Timestamp}} only. The two others are not reachable from the Python API, and with {{pipeline.auto-generate-uids: false}} (or {{{}ExecutionConfig.disable_auto_generated_uids(){}}}) the job cannot be submitted at all:
{noformat}
java.lang.IllegalStateException: Auto generated UIDs have been disabled but no UID or hash has been assigned to operator Timestamps/Watermarks
{noformat}
h2. How to reproduce
{code:python}
from pyflink.common import WatermarkStrategy
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import StreamExecutionEnvironment


class MyAssigner(TimestampAssigner):
    def extract_timestamp(self, value, record_timestamp):
        return value[0]


env = StreamExecutionEnvironment.get_execution_environment()
env.get_config().disable_auto_generated_uids()

stream = env.from_collection([(1000, "a"), (2000, "b")]).uid("source")
timed = stream.assign_timestamps_and_watermarks(
    WatermarkStrategy.for_monotonous_timestamps().with_timestamp_assigner(MyAssigner())
).uid("watermarks")
timed.print().uid("sink")

env.execute()
{code}
*Expected:* the job is submitted, because the uid the user sets covers every operator the call created.

*Actual:* submission fails on {{{}Extract-Timestamp{}}}, and on {{Timestamps/Watermarks}} once that one has a uid.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)