@Override
  public void initialize(Configuration configuration, Properties tableProperties, Properties partitionProperties)
      throws SerDeException {
    // This method is called before {@link
    // org.apache.hadoop.hive.kafka.KafkaStorageHandler.preCreateTable}
    // Thus we need to default to
    // org.apache.hadoop.hive.kafka.KafkaUtils.DEFAULT_PROPERTIES if any
    // property is needed
    super.initialize(configuration, tableProperties, partitionProperties);

    final String className = properties.getProperty(KafkaTableProperties.SERDE_CLASS_NAME.getName(),
        KafkaTableProperties.SERDE_CLASS_NAME.getDefaultValue());
    delegateSerDe = KafkaUtils.createDelegate(className);
    // noinspection deprecation
    delegateSerDe.initialize(configuration, tableProperties, partitionProperties);

    if (!(delegateSerDe.getObjectInspector() instanceof StructObjectInspector)) {
      throw new SerDeException(
          "Was expecting Struct Object Inspector but have " + delegateSerDe.getObjectInspector().getClass().getName());
    }
    delegateDeserializerOI = (StructObjectInspector) delegateSerDe.getObjectInspector();

    // Build column names Order matters here
    columnNames.addAll(delegateDeserializerOI.getAllStructFieldRefs().stream().map(StructField::getFieldName)
        .collect(Collectors.toList()));
    columnNames.addAll(MetadataColumn.KAFKA_METADATA_COLUMN_NAMES);

    final List<ObjectInspector> inspectors = new ArrayList<>(columnNames.size());
    inspectors.addAll(delegateDeserializerOI.getAllStructFieldRefs().stream().map(StructField::getFieldObjectInspector)
        .collect(Collectors.toList()));
    inspectors.addAll(MetadataColumn.KAFKA_METADATA_INSPECTORS);
    objectInspector = ObjectInspectorFactory.getStandardStructObjectInspector(columnNames, inspectors);
    metadataStartIndex = columnNames.size() - MetadataColumn.values().length;
    // Setup Read and Write Path From/To Kafka
    if (delegateSerDe.getSerializedClass() == Text.class) {
      bytesConverter = new TextBytesConverter();
    } else if (delegateSerDe.getSerializedClass() == AvroGenericRecordWritable.class) {
      String schemaFromProperty = properties.getProperty(AvroSerdeUtils.AvroTableProperties.SCHEMA_LITERAL.getPropName(), "");
      Preconditions.checkArgument(!schemaFromProperty.isEmpty(), "Avro Schema is empty Can not go further");
      Schema schema = AvroSerdeUtils.getSchemaFor(schemaFromProperty);
      LOG.debug("Building Avro Reader with schema {}", schemaFromProperty);
      bytesConverter = getByteConverterForAvroDelegate(schema, properties);
    } else {
      bytesConverter = new BytesWritableConverter();
    }
  }
Object Inspector of original row plus metadata.
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaSerDe.java