code: def test_set_value_only_deserializer(self):
      def _check(schema: DeserializationSchema, class_name: str):
          source = (
              KafkaSource.builder()
              .set_bootstrap_servers("localhost:9092")
              .set_topics("test_topic")
              .set_value_only_deserializer(schema)
              .build()
          )
          deserialization_schema_wrapper = get_field_value(
              source.get_java_function(), "deserializationSchema"
          )
          self.assertEqual(
              deserialization_schema_wrapper.getClass().getCanonicalName(),
              "org.apache.flink.connector.kafka.source.reader.deserializer"
              ".KafkaValueOnlyDeserializationSchemaWrapper",
          )
          deserialization_schema = get_field_value(
              deserialization_schema_wrapper, "deserializationSchema"
          )
          self.assertEqual(
              deserialization_schema.getClass().getCanonicalName(),
              class_name,
          )
  
      _check(
          SimpleStringSchema(),
          "org.apache.flink.api.common.serialization.SimpleStringSchema",
      )
      _check(
          JsonRowDeserializationSchema.builder()
          .type_info(Types.ROW([Types.STRING()]))
          .build(),
          "org.apache.flink.formats.json.JsonRowDeserializationSchema",
      )
      _check(
          CsvRowDeserializationSchema.Builder(
              Types.ROW([Types.STRING()])
          ).build(),
          "org.apache.flink.formats.csv.CsvRowDeserializationSchema",
      )
      avro_schema_string = """
          {
              "type": "record",
              "name": "test_record",
              "fields": []
          }
          """
      _check(
          AvroRowDeserializationSchema(
              avro_schema_string=avro_schema_string
          ),
          "org.apache.flink.formats.avro.AvroRowDeserializationSchema",
      )
  
imports: import json
  from typing import Dict
  import pyflink.datastream.data_stream
  from pyflink.common import typeinfo
  from pyflink.common.configuration import Configuration
  from pyflink.common.serialization import SimpleStringSchema, DeserializationSchema
  from pyflink.common.typeinfo import Types
  from pyflink.common.types import Row
  from pyflink.common.watermark_strategy import WatermarkStrategy
  from pyflink.datastream.connectors.base import DeliveryGuarantee
  from pyflink.datastream.connectors.kafka import KafkaSource, KafkaTopicPartition, KafkaOffsetsInitializer, KafkaOffsetResetStrategy, KafkaRecordSerializationSchema, KafkaSink, FlinkKafkaProducer, FlinkKafkaConsumer
  from pyflink.datastream.formats.avro import AvroRowDeserializationSchema, AvroRowSerializationSchema
  from pyflink.datastream.formats.csv import CsvRowDeserializationSchema, CsvRowSerializationSchema
  from pyflink.datastream.formats.json import JsonRowDeserializationSchema, JsonRowSerializationSchema
  from pyflink.java_gateway import get_gateway
  from pyflink.testing.test_case_utils import PyFlinkStreamingTestCase, PyFlinkTestCase, invoke_java_object_method, to_java_data_structure
  from pyflink.util.java_utils import to_jarray, is_instance_of, get_field_value
label_desc: Use image node_id 11 for calling the KafkaSourceTests obj's underlying member method code with example usage: obj.test_set_value_only_deserializer() without return types