code: def read_from_kafka(env):
      deserialization_schema = (
          JsonRowDeserializationSchema.Builder()
          .type_info(Types.ROW([Types.INT(), Types.STRING()]))
          .build()
      )
      kafka_consumer = FlinkKafkaConsumer(
          topics="test_json_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.json import JsonRowSerializationSchema, JsonRowDeserializationSchema
label_desc: Use image node_id 2 for calling a global function with example usage: read_from_kafka(env) without return types