code: def python_data_stream_example():
      env = StreamExecutionEnvironment.get_execution_environment()
      # Set the parallelism to be one to make sure that all data including fired timer and normal data
      # are processed by the same worker and the collected result would be in order which is good for
      # assertion.
      env.set_parallelism(1)
      env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
  
      type_info = Types.ROW_NAMED(
          [
              "createTime",
              "orderId",
              "payAmount",
              "payPlatform",
              "provinceId",
          ],
          [
              Types.LONG(),
              Types.LONG(),
              Types.DOUBLE(),
              Types.INT(),
              Types.INT(),
          ],
      )
      json_row_schema = (
          JsonRowDeserializationSchema.builder()
          .type_info(type_info)
          .build()
      )
      kafka_props = {
          "bootstrap.servers": "localhost:9092",
          "group.id": "pyflink-e2e-source",
      }
  
      kafka_consumer = FlinkKafkaConsumer(
          "timer-stream-source", json_row_schema, kafka_props
      )
      kafka_producer = FlinkKafkaProducer(
          "timer-stream-sink", SimpleStringSchema(), kafka_props
      )
  
      watermark_strategy = (
          WatermarkStrategy.for_bounded_out_of_orderness(
              Duration.of_seconds(5)
          ).with_timestamp_assigner(KafkaRowTimestampAssigner())
      )
  
      kafka_consumer.set_start_from_earliest()
      ds = env.add_source(
          kafka_consumer
      ).assign_timestamps_and_watermarks(watermark_strategy)
      ds.key_by(MyKeySelector(), key_type=Types.LONG()).process(
          MyProcessFunction(), output_type=Types.STRING()
      ).add_sink(kafka_producer)
      env.execute_async("test data stream timer")
  
file_path: flink/flink-end-to-end-tests/flink-python-test/python/datastream/data_stream_job.py
imports: from typing import Any
  from pyflink.common import Duration
  from pyflink.common.serialization import SimpleStringSchema
  from pyflink.common.typeinfo import Types
  from pyflink.common.watermark_strategy import TimestampAssigner, WatermarkStrategy
  from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
  from pyflink.datastream.connectors import FlinkKafkaProducer, FlinkKafkaConsumer
  from pyflink.datastream.formats.json import JsonRowDeserializationSchema
  from pyflink.datastream.functions import KeyedProcessFunction
  from functions import MyKeySelector
label_desc: Use image node_id 1 for calling a global function with example usage: python_data_stream_example() without return types