type: source
    schema:
      - name: payload
        type: ROW(before ROW(available INT, booked INT, blocked INT, modifiedon TIMESTAMP), after ROW(available INT, booked INT, blocked INT, modifiedon TIMESTAMP))
      - name: rowTime
        type: TIMESTAMP
        rowtime:
          timestamps:
            type: "from-field"
            from: "?????payload?????"
          watermarks:
            type: "periodic-ascending"
    connector:
      property-version: 1
      type: kafka
      version: 0.11
      topic: topic123466788910
      startup-mode: latest-offset
      properties:
        - key: zookeeper.connect
          value: zookeeper01.prod.xxxxxx.com:2181
        - key: bootstrap.servers
          value: kafka01.prod.xxxxxxx.com:9092
        - key: group.id
          value: test_flink_sql_04
    format:
      property-version: 1
      type: json
      schema: "ROW(before ROW(available INT, booked INT, blocked INT, modifiedon TIMESTAMP), after ROW(available INT, booked INT, blocked INT, modifiedon TIMESTAMP))"