diff --git a/manila/api/v2/shares.py b/manila/api/v2/shares.py
index 770b22eb45..fae7602c1a 100644
--- a/manila/api/v2/shares.py
+++ b/manila/api/v2/shares.py
@@ -110,6 +110,7 @@ class ShareController(shares.ShareMixin,
             raise exc.HTTPBadRequest(explanation=msg)
 
         new_share_network = None
+        new_share_type = None
 
         preserve_metadata = params.get('preserve_metadata', True)
         try:
@@ -145,13 +146,23 @@ class ShareController(shares.ShareMixin,
             except exception.NotFound:
                 msg = _("Share network %s not "
                         "found.") % new_share_network_id
-                raise exc.HTTPNotFound(explanation=msg)
+                raise exc.HTTPBadRequest(explanation=msg)
+
+        new_share_type_id = params.get('new_share_type_id', None)
+        if new_share_type_id:
+            try:
+                new_share_type = db.share_type_get(
+                    context, new_share_type_id)
+            except exception.NotFound:
+                msg = _("Share type %s not found.") % new_share_type_id
+                raise exc.HTTPBadRequest(explanation=msg)
 
         try:
             self.share_api.migration_start(
                 context, share, host, force_host_assisted_migration,
                 preserve_metadata, writable, nondisruptive,
-                new_share_network=new_share_network)
+                new_share_network=new_share_network,
+                new_share_type=new_share_type)
         except exception.Conflict as e:
             raise exc.HTTPConflict(explanation=six.text_type(e))
 
diff --git a/manila/api/views/share_instance.py b/manila/api/views/share_instance.py
index 0259f24887..1a216f6cc5 100644
--- a/manila/api/views/share_instance.py
+++ b/manila/api/views/share_instance.py
@@ -22,6 +22,7 @@ class ViewBuilder(common.ViewBuilder):
         "remove_export_locations",
         "add_access_rules_status_field",
         "add_replication_fields",
+        "add_share_type_field",
     ]
 
     def detail_list(self, request, instances):
@@ -78,3 +79,7 @@ class ViewBuilder(common.ViewBuilder):
     @common.ViewBuilder.versioned_method("2.11")
     def add_replication_fields(self, context, instance_dict, share_instance):
         instance_dict['replica_state'] = share_instance.get('replica_state')
+
+    @common.ViewBuilder.versioned_method("2.22")
+    def add_share_type_field(self, context, instance_dict, share_instance):
+        instance_dict['share_type_id'] = share_instance.get('share_type_id')
diff --git a/manila/db/migrations/alembic/versions/48a7beae3117_move_share_type_id_to_instances.py b/manila/db/migrations/alembic/versions/48a7beae3117_move_share_type_id_to_instances.py
new file mode 100644
index 0000000000..09080d1b29
--- /dev/null
+++ b/manila/db/migrations/alembic/versions/48a7beae3117_move_share_type_id_to_instances.py
@@ -0,0 +1,83 @@
+# Copyright 2016, Hitachi Data Systems.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+"""move_share_type_id_to_instances
+
+Revision ID: 48a7beae3117
+Revises: 63809d875e32
+Create Date: 2016-07-19 13:04:50.035139
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '48a7beae3117'
+down_revision = '63809d875e32'
+
+from alembic import op
+import sqlalchemy as sa
+
+from manila.db.migrations import utils
+
+
+def upgrade():
+    """Move share_type_id from Shares to Share Instances table."""
+
+    # NOTE(ganso): Adding share_type_id as a foreign key to share_instances
+    # table. Please note that share_type_id is NOT a foreign key in shares
+    # table prior to this migration.
+    op.add_column(
+        'share_instances',
+        sa.Column('share_type_id', sa.String(36),
+                  sa.ForeignKey('share_types.id', name='si_st_id_fk'),
+                  nullable=True))
+    connection = op.get_bind()
+    shares_table = utils.load_table('shares', connection)
+    share_instances_table = utils.load_table('share_instances', connection)
+
+    for instance in connection.execute(share_instances_table.select()):
+        share = connection.execute(shares_table.select().where(
+            instance['share_id'] == shares_table.c.id)).first()
+        op.execute(share_instances_table.update().where(
+            share_instances_table.c.id == instance['id']).values(
+            {'share_type_id': share['share_type_id']}))
+
+    op.drop_column('shares', 'share_type_id')
+
+
+def downgrade():
+    """Move share_type_id from Share Instances to Shares table.
+
+    This method can lead to data loss because only the share_type_id from the
+    first share instance is moved to the shares table.
+    """
+
+    # NOTE(ganso): Adding back share_type_id to the shares table NOT as a
+    # foreign key, as it was before.
+    op.add_column(
+        'shares',
+        sa.Column('share_type_id', sa.String(36), nullable=True))
+    connection = op.get_bind()
+    shares_table = utils.load_table('shares', connection)
+    share_instances_table = utils.load_table('share_instances', connection)
+
+    for share in connection.execute(shares_table.select()):
+        instance = connection.execute(share_instances_table.select().where(
+            share['id'] == share_instances_table.c.share_id)).first()
+        op.execute(shares_table.update().where(
+            shares_table.c.id == instance['share_id']).values(
+            {'share_type_id': instance['share_type_id']}))
+
+    op.drop_constraint('si_st_id_fk', 'share_instances', type_='foreignkey')
+    op.drop_column('share_instances', 'share_type_id')
diff --git a/manila/db/sqlalchemy/api.py b/manila/db/sqlalchemy/api.py
index d506f71c33..26f4f49d74 100644
--- a/manila/db/sqlalchemy/api.py
+++ b/manila/db/sqlalchemy/api.py
@@ -1100,7 +1100,7 @@ def _extract_share_instance_values(values):
     share_instance_model_fields = [
         'status', 'host', 'scheduled_at', 'launched_at', 'terminated_at',
         'share_server_id', 'share_network_id', 'availability_zone',
-        'replica_state',
+        'replica_state', 'share_type_id', 'share_type',
     ]
     share_instance_values, share_values = (
         _extract_instance_values(values, share_instance_model_fields)
@@ -1174,6 +1174,7 @@ def share_instance_get(context, share_instance_id, session=None,
         id=share_instance_id,
     ).options(
         joinedload('export_locations'),
+        joinedload('share_type'),
     ).first()
     if result is None:
         raise exception.NotFound()
@@ -1421,8 +1422,7 @@ def _share_get_query(context, session=None):
     if session is None:
         session = get_session()
     return model_query(context, models.Share, session=session).\
-        options(joinedload('share_metadata')).\
-        options(joinedload('share_type'))
+        options(joinedload('share_metadata'))
 
 
 def _metadata_refs(metadata_dict, meta_class):
@@ -1565,7 +1565,7 @@ def _share_get_all_with_filters(context, project_id=None, share_server_id=None,
         query = query.join(
             models.ShareTypeExtraSpecs,
             models.ShareTypeExtraSpecs.share_type_id ==
-            models.Share.share_type_id)
+            models.ShareInstance.share_type_id)
         for k, v in filters['extra_specs'].items():
             query = query.filter(or_(models.ShareTypeExtraSpecs.key == k,
                                      models.ShareTypeExtraSpecs.value == v))
@@ -3222,7 +3222,7 @@ def share_type_destroy(context, id):
     session = get_session()
     with session.begin():
         _share_type_get(context, id, session)
-        results = model_query(context, models.Share, session=session,
+        results = model_query(context, models.ShareInstance, session=session,
                               read_deleted="no").\
             filter_by(share_type_id=id).count()
         cg_count = model_query(context,
diff --git a/manila/db/sqlalchemy/models.py b/manila/db/sqlalchemy/models.py
index c1c4279222..e4ae41c09e 100644
--- a/manila/db/sqlalchemy/models.py
+++ b/manila/db/sqlalchemy/models.py
@@ -189,7 +189,7 @@ class Share(BASE, ManilaBase):
     __tablename__ = 'shares'
     _extra_keys = ['name', 'export_location', 'export_locations', 'status',
                    'host', 'share_server_id', 'share_network_id',
-                   'availability_zone', 'access_rules_status']
+                   'availability_zone', 'access_rules_status', 'share_type_id']
 
     @property
     def name(self):
@@ -227,7 +227,8 @@ class Share(BASE, ManilaBase):
 
     def __getattr__(self, item):
         deprecated_properties = ('host', 'share_server_id', 'share_network_id',
-                                 'availability_zone')
+                                 'availability_zone', 'share_type_id',
+                                 'share_type')
         proxified_properties = ('status',) + deprecated_properties
 
         if item in deprecated_properties:
@@ -303,8 +304,6 @@ class Share(BASE, ManilaBase):
     snapshot_support = Column(Boolean, default=True)
     replication_type = Column(String(255), nullable=True)
     share_proto = Column(String(255))
-    share_type_id = Column(String(36), ForeignKey('share_types.id'),
-                           nullable=True)
     is_public = Column(Boolean, default=False)
     consistency_group_id = Column(String(36),
                                   ForeignKey('consistency_groups.id'),
@@ -323,13 +322,6 @@ class Share(BASE, ManilaBase):
         viewonly=True,
         join_depth=2,
     )
-    share_type = orm.relationship(
-        "ShareTypes",
-        lazy=True,
-        foreign_keys=share_type_id,
-        primaryjoin='and_('
-                    'Share.share_type_id == ShareTypes.id, '
-                    'ShareTypes.deleted == "False")')
 
 
 class ShareInstance(BASE, ManilaBase):
@@ -339,8 +331,8 @@ class ShareInstance(BASE, ManilaBase):
                    'replica_state']
     _proxified_properties = ('user_id', 'project_id', 'size',
                              'display_name', 'display_description',
-                             'snapshot_id', 'share_proto', 'share_type_id',
-                             'is_public', 'consistency_group_id',
+                             'snapshot_id', 'share_proto', 'is_public',
+                             'consistency_group_id',
                              'source_cgsnapshot_member_id')
 
     def set_share_data(self, share):
@@ -386,7 +378,8 @@ class ShareInstance(BASE, ManilaBase):
     launched_at = Column(DateTime)
     terminated_at = Column(DateTime)
     replica_state = Column(String(255), nullable=True)
-
+    share_type_id = Column(String(36), ForeignKey('share_types.id'),
+                           nullable=True)
     availability_zone_id = Column(String(36),
                                   ForeignKey('availability_zones.id'),
                                   nullable=True)
@@ -416,6 +409,13 @@ class ShareInstance(BASE, ManilaBase):
                               nullable=True)
     share_server_id = Column(String(36), ForeignKey('share_servers.id'),
                              nullable=True)
+    share_type = orm.relationship(
+        "ShareTypes",
+        lazy='immediate',
+        foreign_keys=share_type_id,
+        primaryjoin='and_('
+                    'ShareInstance.share_type_id == ShareTypes.id, '
+                    'ShareTypes.deleted == "False")')
 
 
 class ShareInstanceExportLocations(BASE, ManilaBase):
diff --git a/manila/scheduler/manager.py b/manila/scheduler/manager.py
index 89f7e13811..7318ad435e 100644
--- a/manila/scheduler/manager.py
+++ b/manila/scheduler/manager.py
@@ -143,10 +143,10 @@ class SchedulerManager(manager.Manager):
             share_rpcapi.ShareAPI().manage_share(context, share_ref,
                                                  driver_options)
 
-    def migrate_share_to_host(self, context, share_id, host,
-                              force_host_assisted_migration, preserve_metadata,
-                              writable, nondisruptive, new_share_network_id,
-                              request_spec, filter_properties=None):
+    def migrate_share_to_host(
+            self, context, share_id, host, force_host_assisted_migration,
+            preserve_metadata, writable, nondisruptive, new_share_network_id,
+            new_share_type_id, request_spec, filter_properties=None):
         """Ensure that the host exists and can accept the share."""
 
         share_ref = db.share_get(context, share_id)
@@ -177,7 +177,7 @@ class SchedulerManager(manager.Manager):
                 share_rpcapi.ShareAPI().migration_start(
                     context, share_ref, tgt_host.host,
                     force_host_assisted_migration, preserve_metadata, writable,
-                    nondisruptive, new_share_network_id)
+                    nondisruptive, new_share_network_id, new_share_type_id)
             except Exception as ex:
                 with excutils.save_and_reraise_exception():
                     _migrate_share_set_error(self, context, ex, request_spec)
diff --git a/manila/scheduler/rpcapi.py b/manila/scheduler/rpcapi.py
index 98696e584c..241a60056f 100644
--- a/manila/scheduler/rpcapi.py
+++ b/manila/scheduler/rpcapi.py
@@ -84,7 +84,7 @@ class SchedulerAPI(object):
     def migrate_share_to_host(
             self, context, share_id, host, force_host_assisted_migration,
             preserve_metadata, writable, nondisruptive, new_share_network_id,
-            request_spec=None, filter_properties=None):
+            new_share_type_id, request_spec=None, filter_properties=None):
 
         call_context = self.client.prepare(version='1.4')
         request_spec_p = jsonutils.to_primitive(request_spec)
@@ -97,6 +97,7 @@ class SchedulerAPI(object):
             writable=writable,
             nondisruptive=nondisruptive,
             new_share_network_id=new_share_network_id,
+            new_share_type_id=new_share_type_id,
             request_spec=request_spec_p,
             filter_properties=filter_properties)
 
diff --git a/manila/share/api.py b/manila/share/api.py
index c5c4a015b1..fa6369b236 100644
--- a/manila/share/api.py
+++ b/manila/share/api.py
@@ -114,11 +114,11 @@ class API(base.Base):
             if share_type is None:
                 # Grab the source share's share_type if no new share type
                 # has been provided.
-                share_type_id = source_share['share_type_id']
+                share_type_id = source_share['instance']['share_type_id']
                 share_type = share_types.get_share_type(context, share_type_id)
             else:
                 share_type_id = share_type['id']
-                if share_type_id != source_share['share_type_id']:
+                if share_type_id != source_share['instance']['share_type_id']:
                     msg = _("Invalid share type specified: the requested "
                             "share type must match the type of the source "
                             "share. If a share type is not specified when "
@@ -231,7 +231,6 @@ class API(base.Base):
                    'display_name': name,
                    'display_description': description,
                    'share_proto': share_proto,
-                   'share_type_id': share_type_id,
                    'is_public': is_public,
                    'consistency_group_id': consistency_group_id,
                    }
@@ -258,7 +257,8 @@ class API(base.Base):
         self.create_instance(context, share, share_network_id=share_network_id,
                              host=host, availability_zone=availability_zone,
                              consistency_group=consistency_group,
-                             cgsnapshot_member=cgsnapshot_member)
+                             cgsnapshot_member=cgsnapshot_member,
+                             share_type_id=share_type_id)
 
         # Retrieve the share with instance details
         share = self.db.share_get(context, share['id'])
@@ -267,14 +267,16 @@ class API(base.Base):
 
     def create_instance(self, context, share, share_network_id=None,
                         host=None, availability_zone=None,
-                        consistency_group=None, cgsnapshot_member=None):
+                        consistency_group=None, cgsnapshot_member=None,
+                        share_type_id=None):
         policy.check_policy(context, 'share', 'create')
 
         request_spec, share_instance = (
             self.create_share_instance_and_get_request_spec(
                 context, share, availability_zone=availability_zone,
                 consistency_group=consistency_group, host=host,
-                share_network_id=share_network_id))
+                share_network_id=share_network_id,
+                share_type_id=share_type_id))
 
         if cgsnapshot_member:
             # Inherit properties from the cgsnapshot_member
@@ -309,7 +311,8 @@ class API(base.Base):
 
     def create_share_instance_and_get_request_spec(
             self, context, share, availability_zone=None,
-            consistency_group=None, host=None, share_network_id=None):
+            consistency_group=None, host=None, share_network_id=None,
+            share_type_id=None):
 
         availability_zone_id = None
         if availability_zone:
@@ -327,6 +330,7 @@ class API(base.Base):
                 'scheduled_at': timeutils.utcnow(),
                 'host': host if host else '',
                 'availability_zone_id': availability_zone_id,
+                'share_type_id': share_type_id,
             }
         )
 
@@ -339,7 +343,7 @@ class API(base.Base):
             'share_server_id': share_instance['share_server_id'],
             'snapshot_support': share['snapshot_support'],
             'share_proto': share['share_proto'],
-            'share_type_id': share['share_type_id'],
+            'share_type_id': share_type_id,
             'is_public': share['is_public'],
             'consistency_group_id': share['consistency_group_id'],
             'source_cgsnapshot_member_id': share[
@@ -356,12 +360,13 @@ class API(base.Base):
             'host': share_instance['host'],
             'status': share_instance['status'],
             'replica_state': share_instance['replica_state'],
+            'share_type_id': share_instance['share_type_id'],
         }
 
         share_type = None
-        if share['share_type_id']:
+        if share_instance['share_type_id']:
             share_type = self.db.share_type_get(
-                context, share['share_type_id'])
+                context, share_instance['share_type_id'])
 
         request_spec = {
             'share_properties': share_properties,
@@ -395,7 +400,8 @@ class API(base.Base):
         request_spec, share_replica = (
             self.create_share_instance_and_get_request_spec(
                 context, share, availability_zone=availability_zone,
-                share_network_id=share_network_id))
+                share_network_id=share_network_id,
+                share_type_id=share['instance']['share_type_id']))
 
         all_replicas = self.db.share_replicas_get_all_by_share(
             context, share['id'])
@@ -873,10 +879,10 @@ class API(base.Base):
 
         return snapshot
 
-    def migration_start(self, context, share, dest_host,
-                        force_host_assisted_migration, preserve_metadata=True,
-                        writable=True, nondisruptive=False,
-                        new_share_network=None):
+    def migration_start(
+            self, context, share, dest_host, force_host_assisted_migration,
+            preserve_metadata=True, writable=True, nondisruptive=False,
+            new_share_network=None, new_share_type=None):
         """Migrates share to a new host."""
 
         share_instance = share.instance
@@ -921,13 +927,42 @@ class API(base.Base):
         service = self.db.service_get_by_args(
             context, dest_host_host, 'manila-share')
 
-        share_type = {}
-        share_type_id = share['share_type_id']
-        if share_type_id:
-            share_type = share_types.get_share_type(context, share_type_id)
+        if new_share_type:
+            share_type = new_share_type
+            new_share_type_id = new_share_type['id']
+            dhss = share_type['extra_specs']['driver_handles_share_servers']
+            dhss = strutils.bool_from_string(dhss, strict=True)
+            if (dhss and not new_share_network and
+                    not share_instance['share_network_id']):
+                msg = _(
+                    "New share network must be provided when share type of"
+                    " given share %s has extra_spec "
+                    "'driver_handles_share_servers' as True.") % share['id']
+                raise exception.InvalidInput(reason=msg)
+        else:
+            share_type = {}
+            share_type_id = share_instance['share_type_id']
+            if share_type_id:
+                share_type = share_types.get_share_type(context, share_type_id)
+            new_share_type_id = share_instance['share_type_id']
 
-        new_share_network_id = (new_share_network['id'] if new_share_network
-                                else share_instance['share_network_id'])
+        dhss = share_type['extra_specs']['driver_handles_share_servers']
+        dhss = strutils.bool_from_string(dhss, strict=True)
+
+        if dhss:
+            if new_share_network:
+                new_share_network_id = new_share_network['id']
+            else:
+                new_share_network_id = share_instance['share_network_id']
+        else:
+            if new_share_network:
+                msg = _(
+                    "New share network must not be provided when share type of"
+                    " given share %s has extra_spec "
+                    "'driver_handles_share_servers' as False.") % share['id']
+                raise exception.InvalidInput(reason=msg)
+
+            new_share_network_id = None
 
         request_spec = self._get_request_spec_dict(
             share,
@@ -945,7 +980,7 @@ class API(base.Base):
         self.scheduler_rpcapi.migrate_share_to_host(
             context, share['id'], dest_host, force_host_assisted_migration,
             preserve_metadata, writable, nondisruptive, new_share_network_id,
-            request_spec)
+            new_share_type_id, request_spec)
 
     def migration_complete(self, context, share):
 
diff --git a/manila/share/manager.py b/manila/share/manager.py
index db905bc4a5..6db9867bd3 100644
--- a/manila/share/manager.py
+++ b/manila/share/manager.py
@@ -664,9 +664,9 @@ class ShareManager(manager.SchedulerDependentManager):
                                                share_server)
 
     def _migration_start_driver(
-            self, context, share_ref, src_share_instance, dest_host,
-            writable, preserve_metadata, nondisruptive, new_share_network_id,
-            new_az_id):
+            self, context, share_ref, src_share_instance, dest_host, writable,
+            preserve_metadata, nondisruptive, new_share_network_id, new_az_id,
+            new_share_type_id):
 
         share_server = self._get_share_server(context, src_share_instance)
 
@@ -675,7 +675,7 @@ class ShareManager(manager.SchedulerDependentManager):
         request_spec, dest_share_instance = (
             share_api.create_share_instance_and_get_request_spec(
                 context, share_ref, new_az_id, None, dest_host,
-                new_share_network_id))
+                new_share_network_id, new_share_type_id))
 
         self.db.share_instance_update(
             context, dest_share_instance['id'],
@@ -852,10 +852,10 @@ class ShareManager(manager.SchedulerDependentManager):
                     LOG.exception(msg)
 
     @utils.require_driver_initialized
-    def migration_start(self, context, share_id, dest_host,
-                        force_host_assisted_migration, preserve_metadata=True,
-                        writable=True, nondisruptive=False,
-                        new_share_network_id=None):
+    def migration_start(
+            self, context, share_id, dest_host, force_host_assisted_migration,
+            preserve_metadata=True, writable=True, nondisruptive=False,
+            new_share_network_id=None, new_share_type_id=None):
         """Migrates a share from current host to another host."""
         LOG.debug("Entered migration_start method for share %s.", share_id)
 
@@ -878,7 +878,7 @@ class ShareManager(manager.SchedulerDependentManager):
                 success = self._migration_start_driver(
                     context, share_ref, share_instance, dest_host, writable,
                     preserve_metadata, nondisruptive, new_share_network_id,
-                    new_az_id)
+                    new_az_id, new_share_type_id)
 
             except Exception as e:
                 if not isinstance(e, NotImplementedError):
@@ -907,7 +907,7 @@ class ShareManager(manager.SchedulerDependentManager):
 
                 self._migration_start_host_assisted(
                     context, share_ref, share_instance, dest_host,
-                    new_share_network_id, new_az_id)
+                    new_share_network_id, new_az_id, new_share_type_id)
 
         except Exception:
             msg = _("Host-assisted migration failed for share %s.") % share_id
@@ -921,8 +921,8 @@ class ShareManager(manager.SchedulerDependentManager):
             raise exception.ShareMigrationFailed(reason=msg)
 
     def _migration_start_host_assisted(
-            self, context, share, src_share_instance,
-            dest_host, new_share_network_id, new_az_id):
+            self, context, share, src_share_instance, dest_host,
+            new_share_network_id, new_az_id, new_share_type_id):
 
         rpcapi = share_rpcapi.ShareAPI()
 
@@ -939,7 +939,8 @@ class ShareManager(manager.SchedulerDependentManager):
 
         try:
             dest_share_instance = helper.create_instance_and_wait(
-                share, dest_host, new_share_network_id, new_az_id)
+                share, dest_host, new_share_network_id, new_az_id,
+                new_share_type_id)
 
             self.db.share_instance_update(
                 context, dest_share_instance['id'],
diff --git a/manila/share/migration.py b/manila/share/migration.py
index f5104b6f88..e3b7d04ae6 100644
--- a/manila/share/migration.py
+++ b/manila/share/migration.py
@@ -84,11 +84,12 @@ class ShareMigrationHelper(object):
             else:
                 time.sleep(tries ** 2)
 
-    def create_instance_and_wait(
-            self, share, dest_host, new_share_network_id, new_az_id):
+    def create_instance_and_wait(self, share, dest_host, new_share_network_id,
+                                 new_az_id, new_share_type_id):
 
         new_share_instance = self.api.create_instance(
-            self.context, share, new_share_network_id, dest_host, new_az_id)
+            self.context, share, new_share_network_id, dest_host,
+            new_az_id, share_type_id=new_share_type_id)
 
         # Wait for new_share_instance to become ready
         starttime = time.time()
diff --git a/manila/share/rpcapi.py b/manila/share/rpcapi.py
index 2f146389ce..7bacce784c 100644
--- a/manila/share/rpcapi.py
+++ b/manila/share/rpcapi.py
@@ -126,7 +126,8 @@ class ShareAPI(object):
 
     def migration_start(self, context, share, dest_host,
                         force_host_assisted_migration, preserve_metadata,
-                        writable, nondisruptive, new_share_network_id):
+                        writable, nondisruptive, new_share_network_id,
+                        new_share_type_id):
         new_host = utils.extract_host(share['instance']['host'])
         call_context = self.client.prepare(server=new_host, version='1.12')
         call_context.cast(
@@ -138,7 +139,8 @@ class ShareAPI(object):
             preserve_metadata=preserve_metadata,
             writable=writable,
             nondisruptive=nondisruptive,
-            new_share_network_id=new_share_network_id)
+            new_share_network_id=new_share_network_id,
+            new_share_type_id=new_share_type_id)
 
     def connection_get_info(self, context, share_instance):
         new_host = utils.extract_host(share_instance['host'])
diff --git a/manila/tests/api/v2/test_shares.py b/manila/tests/api/v2/test_shares.py
index cd5d0a714f..712b893d70 100644
--- a/manila/tests/api/v2/test_shares.py
+++ b/manila/tests/api/v2/test_shares.py
@@ -287,6 +287,7 @@ class ShareAPITest(test.TestCase):
     def test_migration_start(self):
         share = db_utils.create_share()
         share_network = db_utils.create_share_network()
+        share_type = {'share_type_id': 'fake_type_id'}
         req = fakes.HTTPRequest.blank('/shares/%s/action' % share['id'],
                                       use_admin_context=True, version='2.22')
         req.method = 'POST'
@@ -296,10 +297,14 @@ class ShareAPITest(test.TestCase):
 
         self.mock_object(db, 'share_network_get', mock.Mock(
             return_value=share_network))
+        self.mock_object(db, 'share_type_get', mock.Mock(
+            return_value=share_type))
+
         body = {
             'migration_start': {
                 'host': 'fake_host',
                 'new_share_network_id': 'fake_net_id',
+                'new_share_type_id': 'fake_type_id',
             }
         }
         method = 'migration_start'
@@ -310,12 +315,15 @@ class ShareAPITest(test.TestCase):
         response = getattr(self.controller, method)(req, share['id'], body)
 
         self.assertEqual(202, response.status_int)
+
         share_api.API.get.assert_called_once_with(context, share['id'])
         share_api.API.migration_start.assert_called_once_with(
             context, share, 'fake_host', False, True, True, False,
-            new_share_network=share_network)
+            new_share_network=share_network, new_share_type=share_type)
         db.share_network_get.assert_called_once_with(
             context, 'fake_net_id')
+        db.share_type_get.assert_called_once_with(
+            context, 'fake_type_id')
 
     def test_migration_start_has_replicas(self):
         share = db_utils.create_share()
@@ -378,11 +386,30 @@ class ShareAPITest(test.TestCase):
 
         self.mock_object(db, 'share_network_get',
                          mock.Mock(side_effect=exception.NotFound()))
-        self.assertRaises(webob.exc.HTTPNotFound,
+        self.assertRaises(webob.exc.HTTPBadRequest,
                           self.controller.migration_start,
                           req, share['id'], body)
         db.share_network_get.assert_called_once_with(context, 'nonexistent')
 
+    def test_migration_start_new_share_type_not_found(self):
+        share = db_utils.create_share()
+        req = fakes.HTTPRequest.blank('/shares/%s/action' % share['id'],
+                                      use_admin_context=True, version='2.22')
+        context = req.environ['manila.context']
+        req.method = 'POST'
+        req.headers['content-type'] = 'application/json'
+        req.api_version_request.experimental = True
+
+        body = {'migration_start': {'host': 'fake_host',
+                                    'new_share_type_id': 'nonexistent'}}
+
+        self.mock_object(db, 'share_type_get',
+                         mock.Mock(side_effect=exception.NotFound()))
+        self.assertRaises(webob.exc.HTTPBadRequest,
+                          self.controller.migration_start,
+                          req, share['id'], body)
+        db.share_type_get.assert_called_once_with(context, 'nonexistent')
+
     def test_migration_start_invalid_force_host_assisted_migration(self):
         share = db_utils.create_share()
         req = fakes.HTTPRequest.blank('/shares/%s/action' % share['id'],
diff --git a/manila/tests/db/migrations/alembic/migrations_data_checks.py b/manila/tests/db/migrations/alembic/migrations_data_checks.py
index ebe17258c3..217d07587b 100644
--- a/manila/tests/db/migrations/alembic/migrations_data_checks.py
+++ b/manila/tests/db/migrations/alembic/migrations_data_checks.py
@@ -984,3 +984,100 @@ class AddAccessKeyToShareAccessMapping(BaseMigrationChecks):
         for row in rows:
             self.test_case.assertFalse(hasattr(row,
                                                self.access_key_column_name))
+
+
+@map_to_migration('48a7beae3117')
+class MoveShareTypeIdToInstancesCheck(BaseMigrationChecks):
+
+    some_shares = [
+        {
+            'id': 's1',
+            'share_type_id': 't1',
+        },
+        {
+            'id': 's2',
+            'share_type_id': 't2',
+        },
+        {
+            'id': 's3',
+            'share_type_id': 't3',
+        },
+    ]
+
+    share_ids = [x['id'] for x in some_shares]
+
+    some_instances = [
+        {
+            'id': 'i1',
+            'share_id': 's3',
+        },
+        {
+            'id': 'i2',
+            'share_id': 's2',
+        },
+        {
+            'id': 'i3',
+            'share_id': 's2',
+        },
+        {
+            'id': 'i4',
+            'share_id': 's1',
+        },
+    ]
+
+    instance_ids = [x['id'] for x in some_instances]
+
+    some_share_types = [
+        {'id': 't1'},
+        {'id': 't2'},
+        {'id': 't3'},
+    ]
+
+    def setup_upgrade_data(self, engine):
+
+        shares_table = utils.load_table('shares', engine)
+        share_instances_table = utils.load_table('share_instances', engine)
+        share_types_table = utils.load_table('share_types', engine)
+
+        for stype in self.some_share_types:
+            engine.execute(share_types_table.insert(stype))
+
+        for share in self.some_shares:
+            engine.execute(shares_table.insert(share))
+
+        for instance in self.some_instances:
+            engine.execute(share_instances_table.insert(instance))
+
+    def check_upgrade(self, engine, data):
+
+        shares_table = utils.load_table('shares', engine)
+        share_instances_table = utils.load_table('share_instances', engine)
+
+        for instance in engine.execute(share_instances_table.select().where(
+                share_instances_table.c.id in self.instance_ids)):
+            share = engine.execute(shares_table.select().where(
+                instance['share_id'] == shares_table.c.id)).first()
+            self.test_case.assertEqual(
+                next((x for x in self.some_shares if share['id'] == x['id']),
+                     None)['share_type_id'],
+                instance['share_type_id'])
+
+        for share in engine.execute(share_instances_table.select().where(
+                shares_table.c.id in self.share_ids)):
+            self.test_case.assertNotIn('share_type_id', share)
+
+    def check_downgrade(self, engine):
+
+        shares_table = utils.load_table('shares', engine)
+        share_instances_table = utils.load_table('share_instances', engine)
+
+        for instance in engine.execute(share_instances_table.select().where(
+                share_instances_table.c.id in self.instance_ids)):
+            self.test_case.assertNotIn('share_type_id', instance)
+
+        for share in engine.execute(share_instances_table.select().where(
+                shares_table.c.id in self.share_ids)):
+            self.test_case.assertEqual(
+                next((x for x in self.some_shares if share['id'] == x['id']),
+                     None)['share_type_id'],
+                share['share_type_id'])
diff --git a/manila/tests/scheduler/test_manager.py b/manila/tests/scheduler/test_manager.py
index c318d0f645..0c3f9eed88 100644
--- a/manila/tests/scheduler/test_manager.py
+++ b/manila/tests/scheduler/test_manager.py
@@ -233,15 +233,15 @@ class SchedulerManagerTestCase(test.TestCase):
 
         self.assertRaises(
             TypeError, self.manager.migrate_share_to_host,
-            self.context, share['id'], 'fake@backend#pool', False, True,
-            True, False, 'fake_net_id', {}, None)
+            self.context, share['id'], 'fake@backend#pool', False, True, True,
+            False, 'fake_net_id', 'fake_type_id', {}, None)
 
         db.share_get.assert_called_once_with(self.context, share['id'])
         base.Scheduler.host_passes_filters.assert_called_once_with(
             self.context, 'fake@backend#pool', {}, None)
         share_rpcapi.ShareAPI.migration_start.assert_called_once_with(
             self.context, share, host.host, False, True, True, False,
-            'fake_net_id')
+            'fake_net_id', 'fake_type_id')
 
     @ddt.data(exception.NoValidHost(reason='fake'), TypeError)
     def test_migrate_share_to_host_exception(self, exc):
@@ -263,7 +263,7 @@ class SchedulerManagerTestCase(test.TestCase):
         self.assertRaises(
             capture, self.manager.migrate_share_to_host,
             self.context, share['id'], host, False, True, True, False,
-            'fake_net_id', request_spec, None)
+            'fake_net_id', 'fake_type_id', request_spec, None)
 
         base.Scheduler.host_passes_filters.assert_called_once_with(
             self.context, host, request_spec, None)
diff --git a/manila/tests/scheduler/test_rpcapi.py b/manila/tests/scheduler/test_rpcapi.py
index d20f0ebf0a..0bdf74cd72 100644
--- a/manila/tests/scheduler/test_rpcapi.py
+++ b/manila/tests/scheduler/test_rpcapi.py
@@ -110,7 +110,8 @@ class SchedulerRpcAPITestCase(test.TestCase):
                                  preserve_metadata=True,
                                  writable=True,
                                  nondisruptive=False,
-                                 new_share_network_id='fake_id',
+                                 new_share_network_id='fake_net_id',
+                                 new_share_type_id='fake_type_id',
                                  request_spec='fake_request_spec',
                                  filter_properties='filter_properties',
                                  version='1.4')
diff --git a/manila/tests/share/test_api.py b/manila/tests/share/test_api.py
index 6c85009aaa..fac291114a 100644
--- a/manila/tests/share/test_api.py
+++ b/manila/tests/share/test_api.py
@@ -214,9 +214,10 @@ class ShareAPITestCase(test.TestCase):
             user_id=self.context.user_id,
             project_id=self.context.project_id,
             create_share_instance=False,
-            share_type_id=share_type_id,
         )
-        share_instance = db_utils.create_share_instance(share_id=share['id'])
+        share_instance = db_utils.create_share_instance(
+            share_id=share['id'],
+            share_type_id=share_type_id)
         share_type = {'fake': 'fake'}
         self.mock_object(db_api, 'share_instance_create',
                          mock.Mock(return_value=share_instance))
@@ -253,7 +254,6 @@ class ShareAPITestCase(test.TestCase):
 
         share, share_data = self._setup_create_mocks(
             snapshot_id=snapshot['id'], share_type_id=share_type['id'])
-
         request_spec = {
             'share_properties': share.to_dict(),
             'share_proto': share['share_proto'],
@@ -694,7 +694,8 @@ class ShareAPITestCase(test.TestCase):
         host, share, share_instance = self._setup_create_instance_mocks()
 
         self.api.create_instance(self.context, share, host=host,
-                                 availability_zone='fake')
+                                 availability_zone='fake',
+                                 share_type_id='fake_share_type')
 
         db_api.share_instance_create.assert_called_once_with(
             self.context, share['id'],
@@ -704,10 +705,11 @@ class ShareAPITestCase(test.TestCase):
                 'scheduled_at': self.dt_utc,
                 'host': host,
                 'availability_zone_id': 'fake_id',
+                'share_type_id': 'fake_share_type',
             }
         )
-        db_api.share_type_get.assert_called_once_with(self.context,
-                                                      share['share_type_id'])
+        db_api.share_type_get.assert_called_once_with(
+            self.context, share_instance['share_type_id'])
         self.api.share_rpcapi.create_share_instance.assert_called_once_with(
             self.context,
             share_instance,
@@ -1310,7 +1312,7 @@ class ShareAPITestCase(test.TestCase):
                                 db_api.share_create.call_args[0][1])
         self.api.create_instance.assert_called_once_with(
             self.context, share, share_network_id=share['share_network_id'],
-            host=valid_host,
+            host=valid_host, share_type_id=share_type['id'],
             availability_zone=snapshot['share']['availability_zone'],
             consistency_group=None, cgsnapshot_member=None)
         share_api.policy.check_policy.assert_has_calls([
@@ -2010,25 +2012,48 @@ class ShareAPITestCase(test.TestCase):
             self.context, share, new_size
         )
 
-    def test_migration_start(self):
+    @ddt.data({'share_type': True, 'share_net': True, 'dhss': True},
+              {'share_type': False, 'share_net': True, 'dhss': True},
+              {'share_type': False, 'share_net': False, 'dhss': True},
+              {'share_type': True, 'share_net': False, 'dhss': False},
+              {'share_type': False, 'share_net': False, 'dhss': False})
+    @ddt.unpack
+    def test_migration_start(self, share_type, share_net, dhss):
         host = 'fake2@backend#pool'
         service = {'availability_zone_id': 'fake_az_id'}
-        share_network = db_utils.create_share_network(id='fake_net_id')
+        share_network = None
+        share_network_id = None
+        if share_net:
+            share_network = db_utils.create_share_network(id='fake_net_id')
+            share_network_id = share_network['id']
 
         fake_type = {
             'id': 'fake_type_id',
             'extra_specs': {
                 'snapshot_support': False,
+                'driver_handles_share_servers': dhss,
             },
         }
 
+        if share_type:
+            fake_type_2 = {
+                'id': 'fake_type_2_id',
+                'extra_specs': {
+                    'snapshot_support': False,
+                    'driver_handles_share_servers': dhss,
+                },
+            }
+        else:
+            fake_type_2 = fake_type
+
         share = db_utils.create_share(
             status=constants.STATUS_AVAILABLE,
-            host='fake@backend#pool', share_type_id=fake_type['id'])
+            host='fake@backend#pool', share_type_id=fake_type['id'],
+            share_network_id=share_network_id)
 
         request_spec = self._get_request_spec_dict(
-            share, fake_type, size=0, availability_zone_id='fake_az_id',
-            share_network_id='fake_net_id')
+            share, fake_type_2, size=0, availability_zone_id='fake_az_id',
+            share_network_id=share_network_id)
 
         self.mock_object(self.scheduler_rpcapi, 'migrate_share_to_host')
         self.mock_object(share_types, 'get_share_type',
@@ -2039,14 +2064,19 @@ class ShareAPITestCase(test.TestCase):
         self.mock_object(db_api, 'service_get_by_args',
                          mock.Mock(return_value=service))
 
-        self.api.migration_start(self.context, share, host, True, True,
-                                 True, True, share_network)
+        if share_type:
+            self.api.migration_start(self.context, share, host, True, True,
+                                     True, True, share_network, fake_type_2)
+        else:
+            self.api.migration_start(self.context, share, host, True, True,
+                                     True, True, share_network, None)
 
         self.scheduler_rpcapi.migrate_share_to_host.assert_called_once_with(
             self.context, share['id'], host, True, True, True, True,
-            'fake_net_id', request_spec)
-        share_types.get_share_type.assert_called_once_with(
-            self.context, fake_type['id'])
+            share_network_id, fake_type_2['id'], request_spec)
+        if not share_type:
+            share_types.get_share_type.assert_called_once_with(
+                self.context, fake_type['id'])
         utils.validate_service_host.assert_called_once_with(
             self.context, 'fake2@backend')
         db_api.service_get_by_args.assert_called_once_with(
@@ -2058,6 +2088,47 @@ class ShareAPITestCase(test.TestCase):
             self.context, share.instance['id'],
             {'status': constants.STATUS_MIGRATING})
 
+    @ddt.data(True, False)
+    def test_migration_start_invalid_share_network_type_combo(self, dhss):
+        host = 'fake2@backend#pool'
+        service = {'availability_zone_id': 'fake_az_id'}
+        share_network = None
+        if not dhss:
+            share_network = db_utils.create_share_network(id='fake_net_id')
+
+        fake_type = {
+            'id': 'fake_type_id',
+            'extra_specs': {
+                'snapshot_support': False,
+                'driver_handles_share_servers': not dhss,
+            },
+        }
+
+        fake_type_2 = {
+            'id': 'fake_type_2_id',
+            'extra_specs': {
+                'snapshot_support': False,
+                'driver_handles_share_servers': dhss,
+            },
+        }
+
+        share = db_utils.create_share(
+            status=constants.STATUS_AVAILABLE,
+            host='fake@backend#pool', share_type_id=fake_type['id'])
+
+        self.mock_object(utils, 'validate_service_host')
+        self.mock_object(db_api, 'service_get_by_args',
+                         mock.Mock(return_value=service))
+
+        self.assertRaises(
+            exception.InvalidInput, self.api.migration_start, self.context,
+            share, host, True, True, True, True, share_network, fake_type_2)
+
+        utils.validate_service_host.assert_called_once_with(
+            self.context, 'fake2@backend')
+        db_api.service_get_by_args.assert_called_once_with(
+            self.context, 'fake2@backend', 'manila-share')
+
     def test_migration_start_status_unavailable(self):
         host = 'fake2@backend#pool'
         share = db_utils.create_share(
@@ -2186,7 +2257,7 @@ class ShareAPITestCase(test.TestCase):
     def test_create_share_replica(self, has_snapshots):
         request_spec = fakes.fake_replica_request_spec()
         replica = request_spec['share_instance_properties']
-        share = fakes.fake_share(
+        share = db_utils.create_share(
             id=replica['share_id'], replication_type='dr')
         snapshots = (
             [fakes.fake_snapshot(), fakes.fake_snapshot()]
diff --git a/manila/tests/share/test_manager.py b/manila/tests/share/test_manager.py
index bc1ca678eb..2045a851fe 100644
--- a/manila/tests/share/test_manager.py
+++ b/manila/tests/share/test_manager.py
@@ -3645,7 +3645,7 @@ class ShareManagerTestCase(test.TestCase):
         # run
         self.share_manager.migration_start(
             self.context, 'fake_id', host, False, False, False, False,
-            'fake_net_id')
+            'fake_net_id', 'fake_type_id')
 
         # asserts
         self.share_manager.db.share_get.assert_called_once_with(
@@ -3667,12 +3667,12 @@ class ShareManagerTestCase(test.TestCase):
         self.share_manager.db.share_update.assert_has_calls(share_update_calls)
         self.share_manager._migration_start_driver.assert_called_once_with(
             self.context, share, instance, host, False, False, False,
-            'fake_net_id', 'fake_az_id')
+            'fake_net_id', 'fake_az_id', 'fake_type_id')
         if not success:
             (self.share_manager._migration_start_host_assisted.
                 assert_called_once_with(
                     self.context, share, instance, host, 'fake_net_id',
-                    'fake_az_id'))
+                    'fake_az_id', 'fake_type_id'))
         self.share_manager.db.service_get_by_args.assert_called_once_with(
             self.context, 'fake2@backend', 'manila-share')
 
@@ -3743,7 +3743,7 @@ class ShareManagerTestCase(test.TestCase):
             exception.ShareMigrationFailed,
             self.share_manager.migration_start,
             self.context, 'fake_id', host, False, False, False, False,
-            'fake_net_id')
+            'fake_net_id', 'fake_type_id')
 
         # asserts
         self.share_manager.db.share_get.assert_called_once_with(
@@ -3766,7 +3766,7 @@ class ShareManagerTestCase(test.TestCase):
             {'status': constants.STATUS_AVAILABLE})
         self.share_manager._migration_start_driver.assert_called_once_with(
             self.context, share, instance, host, False, False, False,
-            'fake_net_id', 'fake_az_id')
+            'fake_net_id', 'fake_az_id', 'fake_type_id')
         self.share_manager.db.service_get_by_args.assert_called_once_with(
             self.context, 'fake2@backend', 'manila-share')
 
@@ -3815,7 +3815,7 @@ class ShareManagerTestCase(test.TestCase):
             exception.ShareMigrationFailed,
             self.share_manager._migration_start_host_assisted,
             self.context, share, instance, 'fake_host', 'fake_net_id',
-            'fake_az_id')
+            'fake_az_id', 'fake_type_id')
 
         # asserts
         self.share_manager.db.share_server_get.assert_called_once_with(
@@ -3827,7 +3827,7 @@ class ShareManagerTestCase(test.TestCase):
                                     self.share_manager.driver)
         migration_api.ShareMigrationHelper.create_instance_and_wait.\
             assert_called_once_with(share, 'fake_host', 'fake_net_id',
-                                    'fake_az_id')
+                                    'fake_az_id', 'fake_type_id')
         migration_api.ShareMigrationHelper.\
             cleanup_access_rules.assert_called_once_with(
                 instance, server, self.share_manager.driver)
@@ -3904,11 +3904,11 @@ class ShareManagerTestCase(test.TestCase):
                 exception.ShareMigrationFailed,
                 self.share_manager._migration_start_driver,
                 self.context, share, src_instance, fake_dest_host, False,
-                False, False, share_network_id, 'fake_az_id')
+                False, False, share_network_id, 'fake_az_id', 'fake_type_id')
         else:
             result = self.share_manager._migration_start_driver(
                 self.context, share, src_instance, fake_dest_host, False,
-                False, False, share_network_id, 'fake_az_id')
+                False, False, share_network_id, 'fake_az_id', 'fake_type_id')
 
         # asserts
         if not exc:
@@ -3936,7 +3936,8 @@ class ShareManagerTestCase(test.TestCase):
             self.context, 'fake_src_server_id')
         (api.API.create_share_instance_and_get_request_spec.
          assert_called_once_with(self.context, share, 'fake_az_id', None,
-                                 'fake_host', share_network_id))
+                                 'fake_host', share_network_id,
+                                 'fake_type_id'))
         (self.share_manager.driver.migration_check_compatibility.
          assert_called_once_with(self.context, src_instance,
                                  migrating_instance, src_server, dest_server))
@@ -4011,7 +4012,7 @@ class ShareManagerTestCase(test.TestCase):
             exception.ShareMigrationFailed,
             self.share_manager._migration_start_driver,
             self.context, share, src_instance, fake_dest_host, True, True,
-            True, 'fake_net_id', 'fake_az_id')
+            True, 'fake_net_id', 'fake_az_id', 'fake_new_type_id')
 
         # asserts
         self.share_manager.db.share_server_get.assert_called_once_with(
@@ -4027,7 +4028,8 @@ class ShareManagerTestCase(test.TestCase):
          assert_called_once_with('fake_dest_share_server_id'))
         (api.API.create_share_instance_and_get_request_spec.
          assert_called_once_with(self.context, share, 'fake_az_id', None,
-                                 'fake_host', 'fake_net_id'))
+                                 'fake_host', 'fake_net_id',
+                                 'fake_new_type_id'))
         self.share_manager._migration_delete_instance.assert_called_once_with(
             self.context, migrating_instance['id'])
 
diff --git a/manila/tests/share/test_migration.py b/manila/tests/share/test_migration.py
index 16a1d1a318..695e300bfa 100644
--- a/manila/tests/share/test_migration.py
+++ b/manila/tests/share/test_migration.py
@@ -132,11 +132,12 @@ class ShareMigrationHelperTestCase(test.TestCase):
 
         # run
         self.helper.create_instance_and_wait(
-            self.share, host, 'fake_net_id', 'fake_az_id')
+            self.share, host, 'fake_net_id', 'fake_az_id', 'fake_type_id')
 
         # asserts
         share_api.API.create_instance.assert_called_once_with(
-            self.context, self.share, 'fake_net_id', 'fake_host', 'fake_az_id')
+            self.context, self.share, 'fake_net_id', 'fake_host', 'fake_az_id',
+            share_type_id='fake_type_id')
 
         db.share_instance_get.assert_has_calls([
             mock.call(self.context, share_instance_creating['id'],
@@ -165,11 +166,12 @@ class ShareMigrationHelperTestCase(test.TestCase):
         self.assertRaises(
             exception.ShareMigrationFailed,
             self.helper.create_instance_and_wait, self.share,
-            host, 'fake_net_id', 'fake_az_id')
+            host, 'fake_net_id', 'fake_az_id', 'fake_type_id')
 
         # asserts
         share_api.API.create_instance.assert_called_once_with(
-            self.context, self.share, 'fake_net_id', 'fake_host', 'fake_az_id')
+            self.context, self.share, 'fake_net_id', 'fake_host', 'fake_az_id',
+            share_type_id='fake_type_id')
 
         db.share_instance_get.assert_called_once_with(
             self.context, share_instance_error['id'], with_share_data=True)
@@ -203,12 +205,13 @@ class ShareMigrationHelperTestCase(test.TestCase):
         # run
         self.assertRaises(
             exception.ShareMigrationFailed,
-            self.helper.create_instance_and_wait, self.share,
-            host, 'fake_net_id', 'fake_az_id')
+            self.helper.create_instance_and_wait,  self.share,
+            host, 'fake_net_id', 'fake_az_id', 'fake_type_id')
 
         # asserts
         share_api.API.create_instance.assert_called_once_with(
-            self.context, self.share, 'fake_net_id', 'fake_host', 'fake_az_id')
+            self.context, self.share, 'fake_net_id', 'fake_host', 'fake_az_id',
+            share_type_id='fake_type_id')
 
         db.share_instance_get.assert_called_once_with(
             self.context, share_instance_creating['id'], with_share_data=True)
diff --git a/manila/tests/share/test_rpcapi.py b/manila/tests/share/test_rpcapi.py
index cfafbeebbe..947591e713 100644
--- a/manila/tests/share/test_rpcapi.py
+++ b/manila/tests/share/test_rpcapi.py
@@ -261,7 +261,8 @@ class ShareRpcAPITestCase(test.TestCase):
                              preserve_metadata=True,
                              writable=True,
                              nondisruptive=False,
-                             new_share_network_id='fake_id')
+                             new_share_network_id='fake_net_id',
+                             new_share_type_id='fake_type_id')
 
     def test_connection_get_info(self):
         self._test_share_api('connection_get_info',
diff --git a/manila_tempest_tests/services/share/v2/json/shares_client.py b/manila_tempest_tests/services/share/v2/json/shares_client.py
index b5a745e3f8..0ccc18b712 100755
--- a/manila_tempest_tests/services/share/v2/json/shares_client.py
+++ b/manila_tempest_tests/services/share/v2/json/shares_client.py
@@ -1021,13 +1021,14 @@ class SharesV2Client(shares_client.SharesClient):
                       force_host_assisted_migration=False,
                       new_share_network_id=None, writable=False,
                       preserve_metadata=False, nondisruptive=False,
-                      version=LATEST_MICROVERSION):
+                      new_share_type_id=None, version=LATEST_MICROVERSION):
 
         body = {
             'migration_start': {
                 'host': host,
                 'force_host_assisted_migration': force_host_assisted_migration,
                 'new_share_network_id': new_share_network_id,
+                'new_share_type_id': new_share_type_id,
                 'writable': writable,
                 'preserve_metadata': preserve_metadata,
                 'nondisruptive': nondisruptive,
diff --git a/manila_tempest_tests/tests/api/admin/test_migration.py b/manila_tempest_tests/tests/api/admin/test_migration.py
index ed643ff97c..0f5559872a 100644
--- a/manila_tempest_tests/tests/api/admin/test_migration.py
+++ b/manila_tempest_tests/tests/api/admin/test_migration.py
@@ -13,8 +13,11 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import six
+
 import ddt
 from tempest import config
+from tempest.lib.common.utils import data_utils
 from tempest import test
 
 from manila_tempest_tests.common import constants
@@ -59,6 +62,18 @@ class MigrationNFSTest(base.BaseSharesAdminTest):
                 CONF.share.run_driver_assisted_migration_tests):
             raise cls.skipException("Share migration tests are disabled.")
 
+        extra_specs = {
+            'storage_protocol': CONF.share.capability_storage_protocol,
+            'driver_handles_share_servers': (
+                CONF.share.multitenancy_enabled),
+            'snapshot_support': six.text_type(
+                CONF.share.capability_snapshot_support),
+        }
+        cls.new_type = cls.create_share_type(
+            name=data_utils.rand_name('new_share_type_for_migration'),
+            cleanup_in_class=True,
+            extra_specs=extra_specs)
+
     @test.attr(type=[base.TAG_POSITIVE, base.TAG_BACKEND])
     @base.skip_if_microversion_lt("2.22")
     @ddt.data(True, False)
@@ -115,16 +130,19 @@ class MigrationNFSTest(base.BaseSharesAdminTest):
         old_share_network_id = share['share_network_id']
         new_share_network_id = self._create_secondary_share_network(
             old_share_network_id)
+        old_share_type_id = share['share_type']
+        new_share_type_id = self.new_type['share_type']['id']
 
         share = self.migrate_share(
             share['id'], dest_pool,
             force_host_assisted_migration=force_host_assisted,
-            wait_for_status=task_state,
+            wait_for_status=task_state, new_share_type_id=new_share_type_id,
             new_share_network_id=new_share_network_id)
 
         self._validate_migration_successful(
-            dest_pool, share, task_state,
-            complete=False, share_network_id=old_share_network_id)
+            dest_pool, share, task_state, complete=False,
+            share_network_id=old_share_network_id,
+            share_type_id=old_share_type_id)
 
         progress = self.shares_v2_client.migration_get_progress(share['id'])
 
@@ -135,7 +153,8 @@ class MigrationNFSTest(base.BaseSharesAdminTest):
 
         self._validate_migration_successful(
             dest_pool, share, constants.TASK_STATE_MIGRATION_SUCCESS,
-            complete=True, share_network_id=new_share_network_id)
+            complete=True, share_network_id=new_share_network_id,
+            share_type_id=new_share_type_id)
 
     def _setup_migration(self):
 
@@ -146,7 +165,7 @@ class MigrationNFSTest(base.BaseSharesAdminTest):
                                      "needed to run share migration tests.")
 
         share = self.create_share(self.protocol)
-        share = self.shares_client.get_share(share['id'])
+        share = self.shares_v2_client.get_share(share['id'])
 
         self.shares_v2_client.create_access_rule(
             share['id'], access_to="50.50.50.50", access_level="rw")
@@ -174,7 +193,8 @@ class MigrationNFSTest(base.BaseSharesAdminTest):
 
     def _validate_migration_successful(self, dest_pool, share, status_to_wait,
                                        version=CONF.share.max_api_microversion,
-                                       complete=True, share_network_id=None):
+                                       complete=True, share_network_id=None,
+                                       share_type_id=None):
 
         statuses = ((status_to_wait,)
                     if not isinstance(status_to_wait, (tuple, list, set))
@@ -190,6 +210,8 @@ class MigrationNFSTest(base.BaseSharesAdminTest):
         self.assertIn(share['task_state'], statuses)
         if share_network_id:
             self.assertEqual(share_network_id, share['share_network_id'])
+        if share_type_id:
+            self.assertEqual(share_type_id, share['share_type'])
 
         # Share migrated
         if complete:
diff --git a/manila_tempest_tests/tests/api/admin/test_migration_negative.py b/manila_tempest_tests/tests/api/admin/test_migration_negative.py
index 94450d2ea8..5d7a578ea2 100644
--- a/manila_tempest_tests/tests/api/admin/test_migration_negative.py
+++ b/manila_tempest_tests/tests/api/admin/test_migration_negative.py
@@ -13,7 +13,10 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import six
+
 from tempest import config
+from tempest.lib.common.utils import data_utils
 from tempest.lib import exceptions as lib_exc
 from tempest import test
 import testtools
@@ -65,6 +68,18 @@ class MigrationTest(base.BaseSharesAdminTest):
 
         cls.dest_pool = dest_pool['name']
 
+        extra_specs = {
+            'storage_protocol': CONF.share.capability_storage_protocol,
+            'driver_handles_share_servers': CONF.share.multitenancy_enabled,
+            'snapshot_support': six.text_type(
+                not CONF.share.capability_snapshot_support),
+        }
+        cls.new_type = cls.create_share_type(
+            name=data_utils.rand_name(
+                'new_invalid_share_type_for_migration'),
+            cleanup_in_class=True,
+            extra_specs=extra_specs)
+
     @test.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
     @base.skip_if_microversion_lt("2.22")
     def test_migration_cancel_invalid(self):
@@ -144,7 +159,18 @@ class MigrationTest(base.BaseSharesAdminTest):
             force_host_assisted_migration=True, writable=True,
             preserve_metadata=True)
         self.shares_v2_client.wait_for_migration_status(
-            self.share['id'], self.dest_pool, 'migration_error')
+            self.share['id'], self.dest_pool,
+            constants.TASK_STATE_MIGRATION_ERROR)
+
+    @test.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
+    @base.skip_if_microversion_lt("2.22")
+    def test_migrate_share_change_type_no_valid_host(self):
+        self.shares_v2_client.migrate_share(
+            self.share['id'], self.dest_pool,
+            new_share_type_id=self.new_type['share_type']['id'])
+        self.shares_v2_client.wait_for_migration_status(
+            self.share['id'], self.dest_pool,
+            constants.TASK_STATE_MIGRATION_ERROR)
 
     @test.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
     @base.skip_if_microversion_lt("2.22")
@@ -172,6 +198,14 @@ class MigrationTest(base.BaseSharesAdminTest):
     @base.skip_if_microversion_lt("2.22")
     def test_migrate_share_invalid_share_network(self):
         self.assertRaises(
-            lib_exc.NotFound, self.shares_v2_client.migrate_share,
+            lib_exc.BadRequest, self.shares_v2_client.migrate_share,
             self.share['id'], self.dest_pool,
             new_share_network_id='invalid_net_id')
+
+    @test.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
+    @base.skip_if_microversion_lt("2.22")
+    def test_migrate_share_invalid_share_type(self):
+        self.assertRaises(
+            lib_exc.BadRequest, self.shares_v2_client.migrate_share,
+            self.share['id'], self.dest_pool, True,
+            new_share_type_id='invalid_type_id')
diff --git a/manila_tempest_tests/tests/api/base.py b/manila_tempest_tests/tests/api/base.py
index 6d0eb258ef..768a115b55 100644
--- a/manila_tempest_tests/tests/api/base.py
+++ b/manila_tempest_tests/tests/api/base.py
@@ -404,14 +404,14 @@ class BaseSharesTest(test.BaseTestCase):
     def migrate_share(
             cls, share_id, dest_host, wait_for_status, client=None,
             force_host_assisted_migration=False, new_share_network_id=None,
-            **kwargs):
+            new_share_type_id=None, **kwargs):
         client = client or cls.shares_v2_client
         client.migrate_share(
             share_id, dest_host,
             force_host_assisted_migration=force_host_assisted_migration,
             new_share_network_id=new_share_network_id,
             writable=False, preserve_metadata=False, nondisruptive=False,
-            **kwargs)
+            new_share_type_id=new_share_type_id, **kwargs)
         share = client.wait_for_migration_status(
             share_id, dest_host, wait_for_status, **kwargs)
         return share
diff --git a/releasenotes/notes/migration-share-type-98e3d3c4c6f47bd9.yaml b/releasenotes/notes/migration-share-type-98e3d3c4c6f47bd9.yaml
new file mode 100644
index 0000000000..6c5746b780
--- /dev/null
+++ b/releasenotes/notes/migration-share-type-98e3d3c4c6f47bd9.yaml
@@ -0,0 +1,3 @@
+---
+features:
+  - Administrators can now change a share's type during a migration.