void deserializeKWritable(KafkaWritable kafkaWritable, Object[] rowBoat) throws SerDeException {

    final Object row = delegateSerDe.deserialize(bytesConverter.getWritable(kafkaWritable.getValue()));
    //first add the value payload elements
    for (int i = 0; i < metadataStartIndex; i++) {
      rowBoat[i] =
          delegateDeserializerOI.getStructFieldData(row, delegateDeserializerOI.getStructFieldRef(columnNames.get(i)));
    }
    //add the metadata columns
    for (int i = metadataStartIndex; i < columnNames.size(); i++) {
      final MetadataColumn metadataColumn = MetadataColumn.forName(columnNames.get(i));
      rowBoat[i] = (kafkaWritable.getHiveWritable(metadataColumn));
    }
  }
@param kafkaWritable Kafka writable object containing the row plus kafka metadata
@param rowBoat Boat sized to width of the kafka row plus metadata to carry the row to operator upstream.

@throws SerDeException in case of any serde issue.
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaSerDe.java