code: def read_from_kafka(env):
      type_info = Types.ROW([Types.INT(), Types.STRING()])
      deserialization_schema = CsvRowDeserializationSchema.Builder(
          type_info
      ).build()
  
      kafka_consumer = FlinkKafkaConsumer(
          topics="test_csv_topic",
          deserialization_schema=deserialization_schema,
          properties={
              "bootstrap.servers": "localhost:9092",
              "group.id": "test_group_1",
          },
      )
      kafka_consumer.set_start_from_earliest()
  
      env.add_source(kafka_consumer).print()
      env.execute()
  
imports: import logging
  import sys
  from pyflink.common import Types
  from pyflink.datastream import StreamExecutionEnvironment
  from pyflink.datastream.connectors.kafka import FlinkKafkaProducer, FlinkKafkaConsumer
  from pyflink.datastream.formats.csv import CsvRowSerializationSchema, CsvRowDeserializationSchema
label_desc: Use image node_id 2 for calling a global function with example usage: read_from_kafka(env) without return types