Can I add the watermarks and timestamps after some steps of processing, like this:

env.from_source(
            source,
            WatermarkStrategy.no_watermarks(),
            "Kafka Source",
        )
        # Read the json-strings into dictionaries
        .map(json.loads)         .assign_timestamps_and_watermarks(WatermarkStrategy.for_monotonous_timestamps().with_timestamp_assigner(SensorTimestampAssigner()))

or must they be added to

from_source()

? I'd like to load the Kafka payload string into dictionary before I extract the timestamp.

source

here is a

KafkaSource

.

Yes, this is supported.