diff --git a/cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_base.py b/cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_base.py
index 6b1bb5e4c4c..597c25e8625 100644
--- a/cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_base.py
+++ b/cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_base.py
@@ -104,6 +104,8 @@ class NetAppBaseClientTestCase(test.TestCase):
             client_base.Client, 'do_direct_resize')
         mock_set_space_reservation = self.mock_object(
             client_base.Client, 'set_lun_space_reservation')
+        mock_validate_qos_policy_group = self.mock_object(
+            client_base.Client, '_validate_qos_policy_group')
         initial_size = self.fake_size
 
         if ontap_version < '9.5':
@@ -118,6 +120,7 @@ class NetAppBaseClientTestCase(test.TestCase):
                                    self.fake_size,
                                    self.fake_metadata)
 
+            mock_validate_qos_policy_group.assert_called_once()
             mock_create_node.assert_called_with(
                 'lun-create-by-size',
                 **{'path': expected_path,
@@ -154,6 +157,8 @@ class NetAppBaseClientTestCase(test.TestCase):
             client_base.Client, 'do_direct_resize')
         mock_set_space_reservation = self.mock_object(
             client_base.Client, 'set_lun_space_reservation')
+        mock_validate_qos_policy_group = self.mock_object(
+            client_base.Client, '_validate_qos_policy_group')
         initial_size = self.fake_size
 
         if ontap_version < '9.5':
@@ -168,6 +173,7 @@ class NetAppBaseClientTestCase(test.TestCase):
                                    self.fake_size,
                                    self.fake_metadata)
 
+            mock_validate_qos_policy_group.assert_called_once()
             mock_create_node.assert_called_with(
                 'lun-create-by-size',
                 **{'path': expected_path,
@@ -208,6 +214,8 @@ class NetAppBaseClientTestCase(test.TestCase):
             client_base.Client, 'do_direct_resize')
         mock_set_space_reservation = self.mock_object(
             client_base.Client, 'set_lun_space_reservation')
+        mock_validate_qos_policy_group = self.mock_object(
+            client_base.Client, '_validate_qos_policy_group')
         initial_size = self.fake_size
 
         if ontap_version < '9.5':
@@ -225,6 +233,7 @@ class NetAppBaseClientTestCase(test.TestCase):
                 self.fake_metadata,
                 qos_policy_group_name=expected_qos_group_name)
 
+            mock_validate_qos_policy_group.assert_called_once()
             mock_create_node.assert_called_with(
                 'lun-create-by-size',
                 **{'path': expected_path, 'size': initial_size,
@@ -284,6 +293,8 @@ class NetAppBaseClientTestCase(test.TestCase):
             side_effect=netapp_api.NaApiError)
         self.mock_object(self.client, 'get_ontap_version',
                          return_value=ontap_version)
+        mock_validate_qos_policy_group = self.mock_object(
+            client_base.Client, '_validate_qos_policy_group')
 
         self.assertRaises(netapp_api.NaApiError,
                           self.client.create_lun,
@@ -291,6 +302,7 @@ class NetAppBaseClientTestCase(test.TestCase):
                           self.fake_lun,
                           self.fake_size,
                           self.fake_metadata)
+        mock_validate_qos_policy_group.assert_called_once()
 
     def test_destroy_lun(self):
         path = '/vol/%s/%s' % (self.fake_volume, self.fake_lun)
diff --git a/cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_cmode.py b/cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_cmode.py
index 76ba450ab96..21903866200 100644
--- a/cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_cmode.py
+++ b/cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_cmode.py
@@ -729,6 +729,67 @@ class NetAppCmodeClientTestCase(test.TestCase):
 
         self.assertSetEqual(igroups, expected)
 
+    @ddt.data(True, False)
+    def test__validate_qos_policy_group_none_adaptive(self, is_adaptive):
+        self.client.features.add_feature('ADAPTIVE_QOS', supported=True)
+        self.client._validate_qos_policy_group(
+            is_adaptive=is_adaptive, spec=None)
+
+    def test__validate_qos_policy_group_none_adaptive_no_support(self):
+        self.client.features.add_feature('ADAPTIVE_QOS', supported=False)
+        self.assertRaises(
+            netapp_utils.NetAppDriverException,
+            self.client._validate_qos_policy_group,
+            is_adaptive=True,
+            spec=None)
+
+    @ddt.data(True, False)
+    def test__validate_qos_policy_group_no_qos_min_support(self, is_adaptive):
+        spec = {'min_throughput': '10'}
+        self.assertRaises(
+            netapp_utils.NetAppDriverException,
+            self.client._validate_qos_policy_group,
+            is_adaptive=is_adaptive,
+            spec=spec,
+            qos_min_support=False)
+
+    def test__validate_qos_policy_group_no_block_size_support(self):
+        self.client.features.add_feature(
+            'ADAPTIVE_QOS_BLOCK_SIZE', supported=False)
+        spec = {'block_size': '4K'}
+        self.assertRaises(
+            netapp_utils.NetAppDriverException,
+            self.client._validate_qos_policy_group,
+            is_adaptive=True,
+            spec=spec)
+
+    def test__validate_qos_policy_group_no_expected_iops_allocation_support(
+            self):
+        self.client.features.add_feature(
+            'ADAPTIVE_QOS_EXPECTED_IOPS_ALLOCATION', supported=False)
+        spec = {'expected_iops_allocation': 'used-space'}
+        self.assertRaises(
+            netapp_utils.NetAppDriverException,
+            self.client._validate_qos_policy_group,
+            is_adaptive=True,
+            spec=spec)
+
+    def test__validate_qos_policy_group_adaptive_qos_spec(self):
+        self.client.features.add_feature('ADAPTIVE_QOS', supported=True)
+        self.client.features.add_feature(
+            'ADAPTIVE_QOS_BLOCK_SIZE', supported=True)
+        self.client.features.add_feature(
+            'ADAPTIVE_QOS_EXPECTED_IOPS_ALLOCATION', supported=True)
+        spec = {
+            'expected_iops': '128IOPS/GB',
+            'peak_iops': '512IOPS/GB',
+            'expected_iops_allocation': 'used-space',
+            'peak_iops_allocation': 'used-space',
+            'absolute_min_iops': '64IOPS',
+            'block_size': '4K',
+        }
+        self.client._validate_qos_policy_group(is_adaptive=True, spec=spec)
+
     def test_clone_lun(self):
         self.client.clone_lun(
             'volume', 'fakeLUN', 'newFakeLUN',
@@ -853,19 +914,38 @@ class NetAppCmodeClientTestCase(test.TestCase):
             mock.call('lun-set-qos-policy-group', api_args)])
 
     def test_provision_qos_policy_group_no_qos_policy_group_info(self):
+        mock_qos_policy_group_create = self.mock_object(
+            self.client, 'qos_policy_group_create')
 
         self.client.provision_qos_policy_group(qos_policy_group_info=None,
                                                qos_min_support=True)
 
-        self.assertEqual(0, self.connection.qos_policy_group_create.call_count)
+        mock_qos_policy_group_create.assert_not_called()
+
+    def test_provision_qos_policy_group_no_legacy_no_spec(self):
+        mock_qos_policy_group_exists = self.mock_object(
+            self.client, 'qos_policy_group_exists')
+        mock_qos_policy_group_create = self.mock_object(
+            self.client, 'qos_policy_group_create')
+        mock_qos_policy_group_modify = self.mock_object(
+            self.client, 'qos_policy_group_modify')
+
+        self.client.provision_qos_policy_group(qos_policy_group_info={},
+                                               qos_min_support=False)
+
+        mock_qos_policy_group_exists.assert_not_called()
+        mock_qos_policy_group_create.assert_not_called()
+        mock_qos_policy_group_modify.assert_not_called()
 
     def test_provision_qos_policy_group_legacy_qos_policy_group_info(self):
+        mock_qos_policy_group_create = self.mock_object(
+            self.client, 'qos_policy_group_create')
 
         self.client.provision_qos_policy_group(
             qos_policy_group_info=fake.QOS_POLICY_GROUP_INFO_LEGACY,
             qos_min_support=True)
 
-        self.assertEqual(0, self.connection.qos_policy_group_create.call_count)
+        mock_qos_policy_group_create.assert_not_called()
 
     def test_provision_qos_policy_group_with_qos_spec_create_with_min(self):
 
@@ -880,15 +960,43 @@ class NetAppCmodeClientTestCase(test.TestCase):
         self.client.provision_qos_policy_group(fake.QOS_POLICY_GROUP_INFO,
                                                True)
 
-        mock_qos_policy_group_create.assert_has_calls([
-            mock.call({
-                'policy_name': fake.QOS_POLICY_GROUP_NAME,
-                'min_throughput': fake.MIN_IOPS,
-                'max_throughput': fake.MAX_IOPS,
-            })])
+        mock_qos_policy_group_create.assert_called_once_with({
+            'policy_name': fake.QOS_POLICY_GROUP_NAME,
+            'min_throughput': fake.MIN_IOPS,
+            'max_throughput': fake.MAX_IOPS,
+        })
+        mock_qos_policy_group_modify.assert_not_called()
+
+    def test_provision_qos_policy_group_with_qos_spec_create_with_aqos(self):
+        self.client.features.add_feature('ADAPTIVE_QOS', supported=True)
+        self.client.features.add_feature(
+            'ADAPTIVE_QOS_BLOCK_SIZE', supported=True)
+        self.client.features.add_feature(
+            'ADAPTIVE_QOS_EXPECTED_IOPS_ALLOCATION', supported=True)
+        self.mock_object(self.client,
+                         'qos_policy_group_exists',
+                         return_value=False)
+        mock_qos_policy_group_create = self.mock_object(
+            self.client, 'qos_policy_group_create')
+        mock_qos_policy_group_modify = self.mock_object(
+            self.client, 'qos_policy_group_modify')
+        mock_qos_adaptive_policy_group_create = self.mock_object(
+            self.client, 'qos_adaptive_policy_group_create')
+        mock_qos_adaptive_policy_group_modify = self.mock_object(
+            self.client, 'qos_adaptive_policy_group_modify')
+
+        self.client.provision_qos_policy_group(
+            fake.ADAPTIVE_QOS_POLICY_GROUP_INFO, False)
+
+        mock_qos_adaptive_policy_group_create.assert_called_once_with(
+            fake.ADAPTIVE_QOS_SPEC)
+        mock_qos_adaptive_policy_group_modify.assert_not_called()
+        mock_qos_policy_group_create.assert_not_called()
         mock_qos_policy_group_modify.assert_not_called()
 
     def test_provision_qos_policy_group_with_qos_spec_create_unsupported(self):
+        mock_qos_policy_group_exists = self.mock_object(
+            self.client, 'qos_policy_group_exists')
         mock_qos_policy_group_create = self.mock_object(
             self.client, 'qos_policy_group_create')
         mock_qos_policy_group_modify = self.mock_object(
@@ -897,8 +1005,34 @@ class NetAppCmodeClientTestCase(test.TestCase):
         self.assertRaises(
             netapp_utils.NetAppDriverException,
             self.client.provision_qos_policy_group,
-            fake.QOS_POLICY_GROUP_INFO, False)
+            fake.QOS_POLICY_GROUP_INFO,
+            False)
 
+        mock_qos_policy_group_exists.assert_not_called()
+        mock_qos_policy_group_create.assert_not_called()
+        mock_qos_policy_group_modify.assert_not_called()
+
+    def test_provision_qos_policy_group_with_invalid_qos_spec(self):
+        self.mock_object(self.client, '_validate_qos_policy_group',
+                         side_effect=netapp_utils.NetAppDriverException)
+        mock_policy_group_spec_is_adaptive = self.mock_object(
+            netapp_utils, 'is_qos_policy_group_spec_adaptive')
+        mock_qos_policy_group_exists = self.mock_object(
+            self.client, 'qos_policy_group_exists')
+        mock_qos_policy_group_create = self.mock_object(
+            self.client, 'qos_policy_group_create')
+        mock_qos_policy_group_modify = self.mock_object(
+            self.client, 'qos_policy_group_modify')
+
+        self.assertRaises(
+            netapp_utils.NetAppDriverException,
+            self.client.provision_qos_policy_group,
+            fake.QOS_POLICY_GROUP_INFO,
+            False)
+
+        mock_policy_group_spec_is_adaptive.assert_called_once_with(
+            fake.QOS_POLICY_GROUP_INFO)
+        mock_qos_policy_group_exists.assert_not_called()
         mock_qos_policy_group_create.assert_not_called()
         mock_qos_policy_group_modify.assert_not_called()
 
@@ -963,6 +1097,33 @@ class NetAppCmodeClientTestCase(test.TestCase):
                 'max_throughput': fake.MAX_THROUGHPUT,
             })])
 
+    def test_provision_qos_policy_group_with_qos_spec_modify_with_aqos(self):
+        self.client.features.add_feature('ADAPTIVE_QOS', supported=True)
+        self.client.features.add_feature(
+            'ADAPTIVE_QOS_BLOCK_SIZE', supported=True)
+        self.client.features.add_feature(
+            'ADAPTIVE_QOS_EXPECTED_IOPS_ALLOCATION', supported=True)
+        self.mock_object(self.client,
+                         'qos_policy_group_exists',
+                         return_value=True)
+        mock_qos_policy_group_create = self.mock_object(
+            self.client, 'qos_policy_group_create')
+        mock_qos_policy_group_modify = self.mock_object(
+            self.client, 'qos_policy_group_modify')
+        mock_qos_adaptive_policy_group_create = self.mock_object(
+            self.client, 'qos_adaptive_policy_group_create')
+        mock_qos_adaptive_policy_group_modify = self.mock_object(
+            self.client, 'qos_adaptive_policy_group_modify')
+
+        self.client.provision_qos_policy_group(
+            fake.ADAPTIVE_QOS_POLICY_GROUP_INFO, False)
+
+        mock_qos_adaptive_policy_group_modify.assert_called_once_with(
+            fake.ADAPTIVE_QOS_SPEC)
+        mock_qos_adaptive_policy_group_create.assert_not_called()
+        mock_qos_policy_group_create.assert_not_called()
+        mock_qos_policy_group_modify.assert_not_called()
+
     def test_qos_policy_group_exists(self):
 
         self.mock_send_request.return_value = netapp_api.NaElement(
@@ -1015,6 +1176,30 @@ class NetAppCmodeClientTestCase(test.TestCase):
         self.mock_send_request.assert_has_calls([
             mock.call('qos-policy-group-create', api_args, False)])
 
+    def test_qos_adaptive_policy_group_create(self):
+
+        api_args = {
+            'policy-group': fake.QOS_POLICY_GROUP_NAME,
+            'expected-iops': '%sIOPS/GB' % fake.EXPECTED_IOPS_PER_GB,
+            'peak-iops': '%sIOPS/GB' % fake.PEAK_IOPS_PER_GB,
+            'expected-iops-allocation': fake.EXPECTED_IOPS_ALLOCATION,
+            'peak-iops-allocation': fake.PEAK_IOPS_ALLOCATION,
+            'block-size': fake.BLOCK_SIZE,
+            'vserver': self.vserver,
+        }
+
+        self.client.qos_adaptive_policy_group_create({
+            'policy_name': fake.QOS_POLICY_GROUP_NAME,
+            'expected_iops': '%sIOPS/GB' % fake.EXPECTED_IOPS_PER_GB,
+            'peak_iops': '%sIOPS/GB' % fake.PEAK_IOPS_PER_GB,
+            'expected_iops_allocation': fake.EXPECTED_IOPS_ALLOCATION,
+            'peak_iops_allocation': fake.PEAK_IOPS_ALLOCATION,
+            'block_size': fake.BLOCK_SIZE,
+        })
+
+        self.mock_send_request.assert_has_calls([
+            mock.call('qos-adaptive-policy-group-create', api_args, False)])
+
     def test_qos_policy_group_modify(self):
 
         api_args = {
@@ -1032,17 +1217,28 @@ class NetAppCmodeClientTestCase(test.TestCase):
         self.mock_send_request.assert_has_calls([
             mock.call('qos-policy-group-modify', api_args, False)])
 
-    def test_qos_policy_group_delete(self):
+    def test_qos_adaptive_policy_group_modify(self):
 
         api_args = {
-            'policy-group': fake.QOS_POLICY_GROUP_NAME
+            'policy-group': fake.QOS_POLICY_GROUP_NAME,
+            'expected-iops': '%sIOPS/GB' % fake.EXPECTED_IOPS_PER_GB,
+            'peak-iops': '%sIOPS/GB' % fake.PEAK_IOPS_PER_GB,
+            'expected-iops-allocation': fake.EXPECTED_IOPS_ALLOCATION,
+            'peak-iops-allocation': fake.PEAK_IOPS_ALLOCATION,
+            'block-size': fake.BLOCK_SIZE,
         }
 
-        self.client.qos_policy_group_delete(
-            fake.QOS_POLICY_GROUP_NAME)
+        self.client.qos_adaptive_policy_group_modify({
+            'policy_name': fake.QOS_POLICY_GROUP_NAME,
+            'expected_iops': '%sIOPS/GB' % fake.EXPECTED_IOPS_PER_GB,
+            'peak_iops': '%sIOPS/GB' % fake.PEAK_IOPS_PER_GB,
+            'expected_iops_allocation': fake.EXPECTED_IOPS_ALLOCATION,
+            'peak_iops_allocation': fake.PEAK_IOPS_ALLOCATION,
+            'block_size': fake.BLOCK_SIZE,
+        })
 
         self.mock_send_request.assert_has_calls([
-            mock.call('qos-policy-group-delete', api_args, False)])
+            mock.call('qos-adaptive-policy-group-modify', api_args, False)])
 
     def test_qos_policy_group_rename(self):
 
@@ -1082,7 +1278,9 @@ class NetAppCmodeClientTestCase(test.TestCase):
         self.assertEqual(0, mock_rename.call_count)
         self.assertEqual(1, mock_remove.call_count)
 
-    def test_mark_qos_policy_group_for_deletion_w_qos_spec(self):
+    @ddt.data(True, False)
+    def test_mark_qos_policy_group_for_deletion_w_qos_spec(self,
+                                                           is_adaptive):
 
         mock_rename = self.mock_object(self.client, 'qos_policy_group_rename')
         mock_remove = self.mock_object(self.client,
@@ -1091,14 +1289,17 @@ class NetAppCmodeClientTestCase(test.TestCase):
         new_name = 'deleted_cinder_%s' % fake.QOS_POLICY_GROUP_NAME
 
         self.client.mark_qos_policy_group_for_deletion(
-            qos_policy_group_info=fake.QOS_POLICY_GROUP_INFO_MAX)
+            qos_policy_group_info=fake.QOS_POLICY_GROUP_INFO_MAX,
+            is_adaptive=is_adaptive)
 
         mock_rename.assert_has_calls([
-            mock.call(fake.QOS_POLICY_GROUP_NAME, new_name)])
+            mock.call(fake.QOS_POLICY_GROUP_NAME, new_name, is_adaptive)])
         self.assertEqual(0, mock_log.call_count)
         self.assertEqual(1, mock_remove.call_count)
 
-    def test_mark_qos_policy_group_for_deletion_exception_path(self):
+    @ddt.data(True, False)
+    def test_mark_qos_policy_group_for_deletion_exception_path(self,
+                                                               is_adaptive):
 
         mock_rename = self.mock_object(self.client, 'qos_policy_group_rename')
         mock_rename.side_effect = netapp_api.NaApiError
@@ -1108,10 +1309,11 @@ class NetAppCmodeClientTestCase(test.TestCase):
         new_name = 'deleted_cinder_%s' % fake.QOS_POLICY_GROUP_NAME
 
         self.client.mark_qos_policy_group_for_deletion(
-            qos_policy_group_info=fake.QOS_POLICY_GROUP_INFO_MAX)
+            qos_policy_group_info=fake.QOS_POLICY_GROUP_INFO_MAX,
+            is_adaptive=is_adaptive)
 
         mock_rename.assert_has_calls([
-            mock.call(fake.QOS_POLICY_GROUP_NAME, new_name)])
+            mock.call(fake.QOS_POLICY_GROUP_NAME, new_name, is_adaptive)])
         self.assertEqual(1, mock_log.call_count)
         self.assertEqual(1, mock_remove.call_count)
 
@@ -1138,15 +1340,29 @@ class NetAppCmodeClientTestCase(test.TestCase):
         self.assertEqual(0, mock_log.call_count)
 
     def test_remove_unused_qos_policy_groups_api_error(self):
-
+        self.client.features.add_feature('ADAPTIVE_QOS', supported=True)
         mock_log = self.mock_object(client_cmode.LOG, 'debug')
-        api_args = {
-            'query': {
-                'qos-policy-group-info': {
-                    'policy-group': 'deleted_cinder_*',
-                    'vserver': self.vserver,
-                }
-            },
+        qos_query = {
+            'qos-policy-group-info': {
+                'policy-group': 'deleted_cinder_*',
+                'vserver': self.vserver,
+            }
+        }
+        adaptive_qos_query = {
+            'qos-adaptive-policy-group-info': {
+                'policy-group': 'deleted_cinder_*',
+                'vserver': self.vserver,
+            }
+        }
+        qos_api_args = {
+            'query': qos_query,
+            'max-records': 3500,
+            'continue-on-failure': 'true',
+            'return-success-list': 'false',
+            'return-failure-list': 'false',
+        }
+        adaptive_qos_api_args = {
+            'query': adaptive_qos_query,
             'max-records': 3500,
             'continue-on-failure': 'true',
             'return-success-list': 'false',
@@ -1157,8 +1373,12 @@ class NetAppCmodeClientTestCase(test.TestCase):
         self.client.remove_unused_qos_policy_groups()
 
         self.mock_send_request.assert_has_calls([
-            mock.call('qos-policy-group-delete-iter', api_args, False)])
-        self.assertEqual(1, mock_log.call_count)
+            mock.call('qos-policy-group-delete-iter',
+                      qos_api_args, False),
+            mock.call('qos-adaptive-policy-group-delete-iter',
+                      adaptive_qos_api_args, False),
+        ])
+        self.assertEqual(2, mock_log.call_count)
 
     @mock.patch('cinder.volume.volume_utils.resolve_hostname',
                 return_value='192.168.1.101')
diff --git a/cinder/tests/unit/volume/drivers/netapp/dataontap/fakes.py b/cinder/tests/unit/volume/drivers/netapp/dataontap/fakes.py
index e5af4553178..a3df5d3dba7 100644
--- a/cinder/tests/unit/volume/drivers/netapp/dataontap/fakes.py
+++ b/cinder/tests/unit/volume/drivers/netapp/dataontap/fakes.py
@@ -280,8 +280,28 @@ QOS_POLICY_GROUP_SPEC_MAX = {
     'policy_name': QOS_POLICY_GROUP_NAME,
 }
 
+EXPECTED_IOPS_PER_GB = '128'
+PEAK_IOPS_PER_GB = '512'
+EXPECTED_IOPS_ALLOCATION = 'used-space'
+PEAK_IOPS_ALLOCATION = 'used-space'
+ABSOLUTE_MIN_IOPS = '75'
+BLOCK_SIZE = 'ANY'
+ADAPTIVE_QOS_SPEC = {
+    'policy_name': QOS_POLICY_GROUP_NAME,
+    'expected_iops': EXPECTED_IOPS_PER_GB,
+    'peak_iops': PEAK_IOPS_PER_GB,
+    'expected_iops_allocation': EXPECTED_IOPS_ALLOCATION,
+    'peak_iops_allocation': PEAK_IOPS_ALLOCATION,
+    'absolute_min_iops': ABSOLUTE_MIN_IOPS,
+    'block_size': BLOCK_SIZE,
+}
+
 QOS_POLICY_GROUP_INFO = {'legacy': None, 'spec': QOS_POLICY_GROUP_SPEC}
 QOS_POLICY_GROUP_INFO_MAX = {'legacy': None, 'spec': QOS_POLICY_GROUP_SPEC_MAX}
+ADAPTIVE_QOS_POLICY_GROUP_INFO = {
+    'legacy': None,
+    'spec': ADAPTIVE_QOS_SPEC,
+}
 
 CLONE_SOURCE_NAME = 'fake_clone_source_name'
 CLONE_SOURCE_ID = 'fake_clone_source_id'
diff --git a/cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_base.py b/cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_base.py
index b8bf4be28d5..5983bbf14df 100644
--- a/cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_base.py
+++ b/cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_base.py
@@ -21,6 +21,7 @@
 """Mock unit tests for the NetApp block storage library"""
 
 import copy
+import itertools
 from unittest import mock
 import uuid
 
@@ -124,7 +125,7 @@ class NetAppBlockStorageLibraryTestCase(test.TestCase):
         self.mock_object(volume_utils, 'extract_host',
                          return_value=fake.POOL_NAME)
         self.mock_object(self.library, '_setup_qos_for_volume',
-                         return_value=None)
+                         return_value=fake.QOS_POLICY_GROUP_INFO)
         self.mock_object(self.library, '_create_lun')
         self.mock_object(self.library, '_create_lun_handle')
         self.mock_object(self.library, '_add_lun_to_table')
@@ -135,7 +136,7 @@ class NetAppBlockStorageLibraryTestCase(test.TestCase):
 
         self.library._create_lun.assert_called_once_with(
             fake.POOL_NAME, fake.LUN_NAME, volume_size_in_bytes,
-            fake.LUN_METADATA, None, False)
+            fake.LUN_METADATA, fake.QOS_POLICY_GROUP_NAME, False)
         self.library._get_volume_model_update.assert_called_once_with(
             fake.VOLUME)
         self.assertEqual(
@@ -152,7 +153,7 @@ class NetAppBlockStorageLibraryTestCase(test.TestCase):
         self.mock_object(block_base, 'LOG')
         self.mock_object(na_utils, 'get_volume_extra_specs')
         self.mock_object(self.library, '_setup_qos_for_volume',
-                         return_value=None)
+                         return_value=fake.QOS_POLICY_GROUP_INFO)
         self.mock_object(self.library, '_create_lun', side_effect=Exception)
         self.mock_object(self.library, '_mark_qos_policy_group_for_deletion')
 
@@ -692,8 +693,11 @@ class NetAppBlockStorageLibraryTestCase(test.TestCase):
         self.assertEqual(2, mock_info_log.call_count)
         self.library._add_lun_to_table.assert_called_once_with(mock_lun)
 
-    @ddt.data(None, 'fake_qos_policy_group_name')
-    def test_manage_existing_rename_lun(self, qos_policy_group_name):
+    @ddt.data(*itertools.product((None, 'fake_qos_policy_group_name'),
+                                 (True, False)))
+    @ddt.unpack
+    def test_manage_existing_rename_lun(self, qos_policy_group_name,
+                                        is_qos_policy_group_spec_adaptive):
         expected_update = (
             {'replication_status': fields.ReplicationStatus.ENABLED})
         volume = fake_volume.fake_volume_obj(self.ctxt)
@@ -710,6 +714,8 @@ class NetAppBlockStorageLibraryTestCase(test.TestCase):
         self.mock_object(self.library, '_setup_qos_for_volume')
         self.mock_object(na_utils, 'get_qos_policy_group_name_from_info',
                          return_value=qos_policy_group_name)
+        self.mock_object(na_utils, 'is_qos_policy_group_spec_adaptive',
+                         return_value=is_qos_policy_group_spec_adaptive)
         self.mock_object(self.library, '_add_lun_to_table')
         self.mock_object(self.library, '_get_volume_model_update',
                          return_value=expected_update)
@@ -724,7 +730,8 @@ class NetAppBlockStorageLibraryTestCase(test.TestCase):
         self.library._add_lun_to_table.assert_called_once_with(mock_lun)
         if qos_policy_group_name:
             (self.zapi_client.set_lun_qos_policy_group.
-             assert_called_once_with(expected_new_path, qos_policy_group_name))
+             assert_called_once_with(expected_new_path, qos_policy_group_name,
+                                     is_qos_policy_group_spec_adaptive))
         else:
             self.assertFalse(
                 self.zapi_client.set_lun_qos_policy_group.called)
diff --git a/cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_cmode.py b/cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_cmode.py
index 61c011f682e..f6fa0af3e30 100644
--- a/cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_cmode.py
+++ b/cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_cmode.py
@@ -599,15 +599,20 @@ class NetAppBlockStorageCmodeLibraryTestCase(test.TestCase):
                          self.zapi_client.
                          provision_qos_policy_group.call_count)
 
-    def test_mark_qos_policy_group_for_deletion(self):
+    @ddt.data(True, False)
+    def test_mark_qos_policy_group_for_deletion(self, is_adaptive):
         self.mock_object(self.zapi_client,
                          'mark_qos_policy_group_for_deletion')
+        self.mock_object(na_utils,
+                         'is_qos_policy_group_spec_adaptive',
+                         return_value=is_adaptive)
 
         self.library._mark_qos_policy_group_for_deletion(
             fake.QOS_POLICY_GROUP_INFO)
 
         self.zapi_client.mark_qos_policy_group_for_deletion\
-            .assert_called_once_with(fake.QOS_POLICY_GROUP_INFO)
+            .assert_called_once_with(fake.QOS_POLICY_GROUP_INFO,
+                                     is_adaptive)
 
     def test_unmanage(self):
         self.mock_object(na_utils, 'get_valid_qos_policy_group_info',
@@ -639,7 +644,8 @@ class NetAppBlockStorageCmodeLibraryTestCase(test.TestCase):
         block_base.NetAppBlockStorageLibrary.unmanage.assert_called_once_with(
             fake.VOLUME)
 
-    def test_manage_existing_lun_same_name(self):
+    @ddt.data(True, False)
+    def test_manage_existing_lun_same_name(self, is_adaptive):
         mock_lun = block_base.NetAppLun('handle', 'name', '1',
                                         {'Path': '/vol/FAKE_CMODE_VOL1/name'})
         self.library._get_existing_vol_with_manage_ref = mock.Mock(
@@ -650,6 +656,8 @@ class NetAppBlockStorageCmodeLibraryTestCase(test.TestCase):
         self.library._setup_qos_for_volume = mock.Mock()
         self.mock_object(na_utils, 'get_qos_policy_group_name_from_info',
                          return_value=fake.QOS_POLICY_GROUP_NAME)
+        self.mock_object(na_utils, 'is_qos_policy_group_spec_adaptive',
+                         return_value=is_adaptive)
         self.library._add_lun_to_table = mock.Mock()
         self.zapi_client.move_lun = mock.Mock()
         mock_set_lun_qos_policy_group = self.mock_object(
@@ -675,6 +683,8 @@ class NetAppBlockStorageCmodeLibraryTestCase(test.TestCase):
         self.library._setup_qos_for_volume = mock.Mock()
         self.mock_object(na_utils, 'get_qos_policy_group_name_from_info',
                          return_value=None)
+        self.mock_object(na_utils, 'is_qos_policy_group_spec_adaptive',
+                         return_value=False)
         self.library._add_lun_to_table = mock.Mock()
         self.zapi_client.move_lun = mock.Mock()
 
diff --git a/cinder/tests/unit/volume/drivers/netapp/dataontap/test_nfs_cmode.py b/cinder/tests/unit/volume/drivers/netapp/dataontap/test_nfs_cmode.py
index ae1982434ec..640cbb453ec 100644
--- a/cinder/tests/unit/volume/drivers/netapp/dataontap/test_nfs_cmode.py
+++ b/cinder/tests/unit/volume/drivers/netapp/dataontap/test_nfs_cmode.py
@@ -494,7 +494,9 @@ class NetAppCmodeNfsDriverTestCase(test.TestCase):
         self.mock_object(self.driver, '_delete_backing_file_for_volume')
         self.mock_object(na_utils,
                          'get_valid_qos_policy_group_info',
-                         return_value='fake_qos_policy_group_info')
+                         return_value=fake.QOS_POLICY_GROUP_INFO)
+        self.mock_object(na_utils, 'is_qos_policy_group_spec_adaptive',
+                         return_value=False)
 
         self.driver.delete_volume(fake_volume)
 
@@ -502,8 +504,10 @@ class NetAppCmodeNfsDriverTestCase(test.TestCase):
             fake_volume)
         na_utils.get_valid_qos_policy_group_info.assert_called_once_with(
             fake_volume)
+        na_utils.is_qos_policy_group_spec_adaptive.assert_called_once_with(
+            fake.QOS_POLICY_GROUP_INFO)
         (self.driver.zapi_client.mark_qos_policy_group_for_deletion.
-         assert_called_once_with('fake_qos_policy_group_info'))
+         assert_called_once_with(fake.QOS_POLICY_GROUP_INFO, False))
 
     def test_delete_volume_exception_path(self):
         fake_provider_location = 'fake_provider_location'
@@ -511,7 +515,9 @@ class NetAppCmodeNfsDriverTestCase(test.TestCase):
         self.mock_object(self.driver, '_delete_backing_file_for_volume')
         self.mock_object(na_utils,
                          'get_valid_qos_policy_group_info',
-                         return_value='fake_qos_policy_group_info')
+                         return_value=fake.QOS_POLICY_GROUP_INFO)
+        self.mock_object(na_utils, 'is_qos_policy_group_spec_adaptive',
+                         return_value=False)
         self.mock_object(
             self.driver.zapi_client,
             'mark_qos_policy_group_for_deletion',
@@ -523,8 +529,10 @@ class NetAppCmodeNfsDriverTestCase(test.TestCase):
             fake_volume)
         na_utils.get_valid_qos_policy_group_info.assert_called_once_with(
             fake_volume)
+        na_utils.is_qos_policy_group_spec_adaptive.assert_called_once_with(
+            fake.QOS_POLICY_GROUP_INFO)
         (self.driver.zapi_client.mark_qos_policy_group_for_deletion.
-         assert_called_once_with('fake_qos_policy_group_info'))
+         assert_called_once_with(fake.QOS_POLICY_GROUP_INFO, False))
 
     def test_delete_backing_file_for_volume(self):
         mock_filer_delete = self.mock_object(self.driver, '_delete_file')
diff --git a/cinder/tests/unit/volume/drivers/netapp/fakes.py b/cinder/tests/unit/volume/drivers/netapp/fakes.py
index d9ddc403836..1ee5b288966 100644
--- a/cinder/tests/unit/volume/drivers/netapp/fakes.py
+++ b/cinder/tests/unit/volume/drivers/netapp/fakes.py
@@ -98,6 +98,21 @@ MAX_THROUGHPUT_BPS = '21734278B/s'
 QOS_POLICY_GROUP_NAME = 'fake_qos_policy_group_name'
 LEGACY_EXTRA_SPECS = {'netapp:qos_policy_group': QOS_POLICY_GROUP_NAME}
 
+EXPECTED_IOPS_PER_GB = '128'
+PEAK_IOPS_PER_GB = '512'
+EXPECTED_IOPS_ALLOCATION = 'used-space'
+PEAK_IOPS_ALLOCATION = 'used-space'
+ABSOLUTE_MIN_IOPS = '75'
+BLOCK_SIZE = 'ANY'
+ADAPTIVE_QOS_SPEC = {
+    'expectedIOPSperGiB': EXPECTED_IOPS_PER_GB,
+    'peakIOPSperGiB': PEAK_IOPS_PER_GB,
+    'expectedIOPSAllocation': EXPECTED_IOPS_ALLOCATION,
+    'peakIOPSAllocation': PEAK_IOPS_ALLOCATION,
+    'absoluteMinIOPS': ABSOLUTE_MIN_IOPS,
+    'blockSize': BLOCK_SIZE,
+}
+
 LEGACY_QOS = {
     'policy_name': QOS_POLICY_GROUP_NAME,
 }
@@ -111,16 +126,31 @@ QOS_POLICY_GROUP_INFO_NONE = {'legacy': None, 'spec': None}
 
 QOS_POLICY_GROUP_INFO = {'legacy': None, 'spec': QOS_POLICY_GROUP_SPEC}
 
+ADAPTIVE_QOS_POLICY_GROUP_SPEC = {
+    'expected_iops': '128IOPS/GB',
+    'peak_iops': '512IOPS/GB',
+    'expected_iops_allocation': 'used-space',
+    'peak_iops_allocation': 'used-space',
+    'absolute_min_iops': '75IOPS',
+    'block_size': 'ANY',
+    'policy_name': 'openstack-%s' % VOLUME_ID,
+}
+
 LEGACY_QOS_POLICY_GROUP_INFO = {
     'legacy': LEGACY_QOS,
     'spec': None,
 }
 
-INVALID_QOS_POLICY_GROUP_INFO = {
+INVALID_QOS_POLICY_GROUP_INFO_LEGACY_AND_SPEC = {
     'legacy': LEGACY_QOS,
     'spec': QOS_POLICY_GROUP_SPEC,
 }
 
+INVALID_QOS_POLICY_GROUP_INFO_STANDARD_AND_ADAPTIVE = {
+    'legacy': None,
+    'spec': {**QOS_POLICY_GROUP_SPEC, **ADAPTIVE_QOS_SPEC},
+}
+
 QOS_SPECS_ID = 'fake_qos_specs_id'
 QOS_SPEC = {'maxBPS': 21734278}
 OUTER_BACKEND_QOS_SPEC = {
diff --git a/cinder/tests/unit/volume/drivers/netapp/test_utils.py b/cinder/tests/unit/volume/drivers/netapp/test_utils.py
index 39ffbd4a27c..cf075d45020 100644
--- a/cinder/tests/unit/volume/drivers/netapp/test_utils.py
+++ b/cinder/tests/unit/volume/drivers/netapp/test_utils.py
@@ -45,13 +45,13 @@ class NetAppDriverUtilsTestCase(test.TestCase):
     def test_validate_instantiation_proxy(self):
         kwargs = {'netapp_mode': 'proxy'}
         na_utils.validate_instantiation(**kwargs)
-        self.assertEqual(0, na_utils.LOG.warning.call_count)
+        na_utils.LOG.warning.assert_not_called()
 
     @mock.patch.object(na_utils, 'LOG', mock.Mock())
     def test_validate_instantiation_no_proxy(self):
         kwargs = {'netapp_mode': 'asdf'}
         na_utils.validate_instantiation(**kwargs)
-        self.assertEqual(1, na_utils.LOG.warning.call_count)
+        na_utils.LOG.warning.assert_called_once()
 
     def test_check_flags(self):
 
@@ -228,7 +228,7 @@ class NetAppDriverUtilsTestCase(test.TestCase):
 
         na_utils.log_extra_spec_warnings({'netapp:raid_type': 'raid4'})
 
-        self.assertEqual(1, mock_log.call_count)
+        mock_log.assert_called_once()
 
     def test_log_extra_spec_warnings_deprecated_specs(self):
 
@@ -236,7 +236,13 @@ class NetAppDriverUtilsTestCase(test.TestCase):
 
         na_utils.log_extra_spec_warnings({'netapp_thick_provisioned': 'true'})
 
-        self.assertEqual(1, mock_log.call_count)
+        mock_log.assert_called_once()
+
+    def test_validate_qos_spec(self):
+        qos_spec = fake.QOS_SPEC
+
+        # Just return without raising an exception.
+        na_utils.validate_qos_spec(qos_spec)
 
     def test_validate_qos_spec_none(self):
         qos_spec = None
@@ -244,6 +250,10 @@ class NetAppDriverUtilsTestCase(test.TestCase):
         # Just return without raising an exception.
         na_utils.validate_qos_spec(qos_spec)
 
+    def test_validate_qos_spec_adaptive(self):
+        # Just return without raising an exception.
+        na_utils.validate_qos_spec(fake.ADAPTIVE_QOS_SPEC)
+
     def test_validate_qos_spec_keys_weirdly_cased(self):
         qos_spec = {'mAxIopS': 33000, 'mInIopS': 0}
 
@@ -280,6 +290,34 @@ class NetAppDriverUtilsTestCase(test.TestCase):
 
     def test_validate_qos_spec_bad_key_combination_miniops_miniopspergib(self):
         qos_spec = {'minIOPS': 33000, 'minIOPSperGiB': 10000000}
+        self.assertRaises(exception.Invalid,
+                          na_utils.validate_qos_spec,
+                          qos_spec)
+
+    def test_validate_qos_spec_bad_key_combination_aqos_qos_max(self):
+        qos_spec = {'peakIOPSperGiB': 33000, 'maxIOPS': 33000}
+        self.assertRaises(exception.Invalid,
+                          na_utils.validate_qos_spec,
+                          qos_spec)
+
+    def test_validate_qos_spec_bad_key_combination_aqos_qos_min(self):
+        qos_spec = {'absoluteMinIOPS': 33000, 'minIOPS': 33000}
+        self.assertRaises(exception.Invalid,
+                          na_utils.validate_qos_spec,
+                          qos_spec)
+
+    def test_validate_qos_spec_bad_key_combination_aqos_qos_min_max(self):
+        qos_spec = {
+            'expectedIOPSperGiB': 33000,
+            'minIOPS': 33000,
+            'maxIOPS': 33000,
+        }
+        self.assertRaises(exception.Invalid,
+                          na_utils.validate_qos_spec,
+                          qos_spec)
+
+    def test_validate_qos_spec_adaptive_and_non_adaptive(self):
+        qos_spec = fake.INVALID_QOS_POLICY_GROUP_INFO_STANDARD_AND_ADAPTIVE
 
         self.assertRaises(exception.Invalid,
                           na_utils.validate_qos_spec,
@@ -408,6 +446,101 @@ class NetAppDriverUtilsTestCase(test.TestCase):
 
         self.assertEqual(expected, result)
 
+    def test_map_aqos_spec(self):
+        qos_spec = {
+            'expectedIOPSperGiB': '128',
+            'peakIOPSperGiB': '512',
+            'expectedIOPSAllocation': 'used-space',
+            'peakIOPSAllocation': 'used-space',
+            'absoluteMinIOPS': '75',
+            'blockSize': 'ANY',
+        }
+        mock_get_name = self.mock_object(na_utils, 'get_qos_policy_group_name')
+        mock_get_name.return_value = 'fake_qos_policy'
+        expected = {
+            'expected_iops': '128IOPS/GB',
+            'peak_iops': '512IOPS/GB',
+            'expected_iops_allocation': 'used-space',
+            'peak_iops_allocation': 'used-space',
+            'absolute_min_iops': '75IOPS',
+            'block_size': 'ANY',
+            'policy_name': 'fake_qos_policy',
+        }
+
+        result = na_utils.map_aqos_spec(qos_spec, fake.VOLUME)
+
+        self.assertEqual(expected, result)
+
+    @ddt.data({'expectedIOPSperGiB': '528', 'peakIOPSperGiB': '128'},
+              {'expectedIOPSperGiB': '528'})
+    def test_map_aqos_spec_error(self, qos_spec):
+        mock_get_name = self.mock_object(na_utils, 'get_qos_policy_group_name')
+        mock_get_name.return_value = 'fake_qos_policy'
+
+        self.assertRaises(exception.Invalid, na_utils.map_aqos_spec, qos_spec,
+                          fake.VOLUME)
+
+    def test_is_qos_adaptive_adaptive_spec(self):
+        aqos_spec = fake.ADAPTIVE_QOS_SPEC
+
+        self.assertTrue(na_utils.is_qos_adaptive(aqos_spec))
+
+    def test_is_qos_adaptive_weirdly_cased_adaptive_spec(self):
+        aqos_spec = {'expecTEDiopsPERgib': '128IOPS/GB'}
+
+        self.assertTrue(na_utils.is_qos_adaptive(aqos_spec))
+
+    def test_is_qos_adaptive_non_adaptive_spec(self):
+        qos_spec = fake.QOS_SPEC
+
+        self.assertFalse(na_utils.is_qos_adaptive(qos_spec))
+
+    def test_is_qos_policy_group_spec_adaptive_adaptive_spec(self):
+        aqos_spec = {
+            'spec': {
+                'expected_iops': '128IOPS/GB',
+                'peak_iops': '512IOPS/GB',
+                'expected_iops_allocation': 'used-space',
+                'absolute_min_iops': '75IOPS',
+                'block_size': 'ANY',
+                'policy_name': 'fake_policy_name',
+            }
+        }
+
+        self.assertTrue(na_utils.is_qos_policy_group_spec_adaptive(aqos_spec))
+
+    def test_is_qos_policy_group_spec_adaptive_none(self):
+        qos_spec = None
+
+        self.assertFalse(na_utils.is_qos_policy_group_spec_adaptive(qos_spec))
+
+    def test_is_qos_policy_group_spec_adaptive_legacy(self):
+        qos_spec = {
+            'legacy': fake.LEGACY_QOS,
+        }
+
+        self.assertFalse(na_utils.is_qos_policy_group_spec_adaptive(qos_spec))
+
+    def test_is_qos_policy_group_spec_adaptive_non_adaptive_spec(self):
+        qos_spec = {
+            'spec': {
+                'max_throughput': '21834289B/s',
+                'policy_name': 'fake_policy_name',
+            }
+        }
+
+        self.assertFalse(na_utils.is_qos_policy_group_spec_adaptive(qos_spec))
+
+    def test_policy_group_qos_spec_is_adaptive_invalid_spec(self):
+        qos_spec = {
+            'spec': {
+                'max_flops': '512',
+                'policy_name': 'fake_policy_name',
+            }
+        }
+
+        self.assertFalse(na_utils.is_qos_policy_group_spec_adaptive(qos_spec))
+
     def test_map_dict_to_lower(self):
         original = {'UPperKey': 'Value'}
         expected = {'upperkey': 'Value'}
@@ -498,7 +631,6 @@ class NetAppDriverUtilsTestCase(test.TestCase):
         mock_get_valid_qos_spec_from_volume_type = self.mock_object(
             na_utils, 'get_valid_backend_qos_spec_from_volume_type')
         mock_get_valid_qos_spec_from_volume_type.return_value = None
-        self.mock_object(na_utils, 'check_for_invalid_qos_spec_combination')
         expected = fake.QOS_POLICY_GROUP_INFO_NONE
 
         result = na_utils.get_valid_qos_policy_group_info(fake.VOLUME)
@@ -516,7 +648,6 @@ class NetAppDriverUtilsTestCase(test.TestCase):
         mock_get_valid_qos_spec_from_volume_type = self.mock_object(
             na_utils, 'get_valid_backend_qos_spec_from_volume_type')
         mock_get_valid_qos_spec_from_volume_type.return_value = None
-        self.mock_object(na_utils, 'check_for_invalid_qos_spec_combination')
 
         result = na_utils.get_valid_qos_policy_group_info(fake.VOLUME)
 
@@ -533,7 +664,6 @@ class NetAppDriverUtilsTestCase(test.TestCase):
             na_utils, 'get_valid_backend_qos_spec_from_volume_type')
         mock_get_valid_qos_spec_from_volume_type.return_value =\
             fake.QOS_POLICY_GROUP_SPEC
-        self.mock_object(na_utils, 'check_for_invalid_qos_spec_combination')
 
         result = na_utils.get_valid_qos_policy_group_info(fake.VOLUME)
 
@@ -543,25 +673,43 @@ class NetAppDriverUtilsTestCase(test.TestCase):
         mock_get_spec = self.mock_object(
             na_utils, 'get_backend_qos_spec_from_volume_type')
         mock_get_spec.return_value = None
-        mock_validate = self.mock_object(na_utils, 'validate_qos_spec')
+        mock_map_qos_spec = self.mock_object(
+            na_utils, 'map_qos_spec')
+        mock_map_aqos_spec = self.mock_object(
+            na_utils, 'map_aqos_spec')
 
         result = na_utils.get_valid_backend_qos_spec_from_volume_type(
             fake.VOLUME, fake.VOLUME_TYPE)
 
         self.assertIsNone(result)
-        self.assertEqual(0, mock_validate.call_count)
+        mock_map_qos_spec.assert_not_called()
+        mock_map_aqos_spec.assert_not_called()
 
     def test_get_valid_backend_qos_spec_from_volume_type(self):
         mock_get_spec = self.mock_object(
             na_utils, 'get_backend_qos_spec_from_volume_type')
         mock_get_spec.return_value = fake.QOS_SPEC
-        mock_validate = self.mock_object(na_utils, 'validate_qos_spec')
+        mock_map_aqos_spec = self.mock_object(
+            na_utils, 'map_aqos_spec')
 
         result = na_utils.get_valid_backend_qos_spec_from_volume_type(
             fake.VOLUME, fake.VOLUME_TYPE)
 
         self.assertEqual(fake.QOS_POLICY_GROUP_SPEC, result)
-        self.assertEqual(1, mock_validate.call_count)
+        mock_map_aqos_spec.assert_not_called()
+
+    def test_get_valid_backend_qos_spec_from_volume_type_adaptive(self):
+        mock_get_spec = self.mock_object(
+            na_utils, 'get_backend_qos_spec_from_volume_type')
+        mock_get_spec.return_value = fake.ADAPTIVE_QOS_SPEC
+        mock_map_qos_spec = self.mock_object(
+            na_utils, 'map_qos_spec')
+
+        result = na_utils.get_valid_backend_qos_spec_from_volume_type(
+            fake.VOLUME, fake.VOLUME_TYPE)
+
+        self.assertEqual(fake.ADAPTIVE_QOS_POLICY_GROUP_SPEC, result)
+        mock_map_qos_spec.assert_not_called()
 
     def test_get_backend_qos_spec_from_volume_type_no_qos_specs_id(self):
         volume_type = copy.deepcopy(fake.VOLUME_TYPE)
@@ -571,7 +719,7 @@ class NetAppDriverUtilsTestCase(test.TestCase):
         result = na_utils.get_backend_qos_spec_from_volume_type(volume_type)
 
         self.assertIsNone(result)
-        self.assertEqual(0, mock_get_context.call_count)
+        mock_get_context.assert_not_called()
 
     def test_get_backend_qos_spec_from_volume_type_no_qos_spec(self):
         volume_type = fake.VOLUME_TYPE
@@ -613,11 +761,20 @@ class NetAppDriverUtilsTestCase(test.TestCase):
 
         self.assertEqual(fake.QOS_SPEC, result)
 
-    def test_check_for_invalid_qos_spec_combination(self):
+    def test_check_for_invalid_qos_spec_combination_legacy(self):
+        na_utils.check_for_invalid_qos_spec_combination(
+            fake.LEGACY_QOS_POLICY_GROUP_INFO,
+            fake.VOLUME_TYPE)
 
+    def test_check_for_invalid_qos_spec_combination_spec(self):
+        na_utils.check_for_invalid_qos_spec_combination(
+            fake.QOS_POLICY_GROUP_INFO,
+            fake.VOLUME_TYPE)
+
+    def test_check_for_invalid_qos_spec_combination_legacy_and_spec(self):
         self.assertRaises(exception.Invalid,
                           na_utils.check_for_invalid_qos_spec_combination,
-                          fake.INVALID_QOS_POLICY_GROUP_INFO,
+                          fake.INVALID_QOS_POLICY_GROUP_INFO_LEGACY_AND_SPEC,
                           fake.VOLUME_TYPE)
 
     def test_get_legacy_qos_policy(self):
diff --git a/cinder/volume/drivers/netapp/dataontap/block_base.py b/cinder/volume/drivers/netapp/dataontap/block_base.py
index d65bfbca970..e95f163a186 100644
--- a/cinder/volume/drivers/netapp/dataontap/block_base.py
+++ b/cinder/volume/drivers/netapp/dataontap/block_base.py
@@ -236,8 +236,10 @@ class NetAppBlockStorageLibrary(object):
         qos_policy_group_name = (
             na_utils.get_qos_policy_group_name_from_info(
                 qos_policy_group_info))
-        qos_policy_group_is_adaptive = volume_utils.is_boolean_str(
-            extra_specs.get('netapp:qos_policy_group_is_adaptive'))
+        qos_policy_group_is_adaptive = (volume_utils.is_boolean_str(
+            extra_specs.get('netapp:qos_policy_group_is_adaptive')) or
+            na_utils.is_qos_policy_group_spec_adaptive(
+                qos_policy_group_info))
 
         try:
             self._create_lun(pool_name, lun_name, size, metadata,
@@ -359,8 +361,10 @@ class NetAppBlockStorageLibrary(object):
         qos_policy_group_name = (
             na_utils.get_qos_policy_group_name_from_info(
                 qos_policy_group_info))
-        qos_policy_group_is_adaptive = volume_utils.is_boolean_str(
-            extra_specs.get('netapp:qos_policy_group_is_adaptive'))
+        qos_policy_group_is_adaptive = (volume_utils.is_boolean_str(
+            extra_specs.get('netapp:qos_policy_group_is_adaptive')) or
+            na_utils.is_qos_policy_group_spec_adaptive(
+                qos_policy_group_info))
 
         try:
             self._clone_lun(
@@ -741,6 +745,8 @@ class NetAppBlockStorageLibrary(object):
         qos_policy_group_name = (
             na_utils.get_qos_policy_group_name_from_info(
                 qos_policy_group_info))
+        is_adaptive = na_utils.is_qos_policy_group_spec_adaptive(
+            qos_policy_group_info)
 
         path = lun.get_metadata_property('Path')
         if lun.name == volume['name']:
@@ -756,7 +762,8 @@ class NetAppBlockStorageLibrary(object):
 
         if qos_policy_group_name is not None:
             self.zapi_client.set_lun_qos_policy_group(new_path,
-                                                      qos_policy_group_name)
+                                                      qos_policy_group_name,
+                                                      is_adaptive)
         self._add_lun_to_table(lun)
         LOG.info("Manage operation completed for LUN with new path"
                  " %(path)s and uuid %(uuid)s.",
diff --git a/cinder/volume/drivers/netapp/dataontap/block_cmode.py b/cinder/volume/drivers/netapp/dataontap/block_cmode.py
index 35e4d378065..9839296137a 100644
--- a/cinder/volume/drivers/netapp/dataontap/block_cmode.py
+++ b/cinder/volume/drivers/netapp/dataontap/block_cmode.py
@@ -414,8 +414,10 @@ class NetAppBlockStorageCmodeLibrary(block_base.NetAppBlockStorageLibrary,
             return {'replication_status': fields.ReplicationStatus.ENABLED}
 
     def _mark_qos_policy_group_for_deletion(self, qos_policy_group_info):
-        self.zapi_client.mark_qos_policy_group_for_deletion(
+        is_adaptive = na_utils.is_qos_policy_group_spec_adaptive(
             qos_policy_group_info)
+        self.zapi_client.mark_qos_policy_group_for_deletion(
+            qos_policy_group_info, is_adaptive)
 
     def unmanage(self, volume):
         """Removes the specified volume from Cinder management.
diff --git a/cinder/volume/drivers/netapp/dataontap/client/client_base.py b/cinder/volume/drivers/netapp/dataontap/client/client_base.py
index cfb970fb5a4..9fdbabc0908 100644
--- a/cinder/volume/drivers/netapp/dataontap/client/client_base.py
+++ b/cinder/volume/drivers/netapp/dataontap/client/client_base.py
@@ -109,6 +109,7 @@ class Client(object):
                    qos_policy_group_name=None,
                    qos_policy_group_is_adaptive=False):
         """Issues API request for creating LUN on volume."""
+        self._validate_qos_policy_group(qos_policy_group_is_adaptive)
 
         path = '/vol/%s/%s' % (volume_name, lun_name)
         space_reservation = metadata['SpaceReserved']
@@ -314,6 +315,10 @@ class Client(object):
         """Get igroups exactly matching a set of initiators."""
         raise NotImplementedError()
 
+    def _validate_qos_policy_group(self, is_adaptive, spec=None, is_nfs=False):
+        """Raises an exception if the backend doesn't support the QoS spec."""
+        raise NotImplementedError()
+
     def _has_luns_mapped_to_initiator(self, initiator):
         """Checks whether any LUNs are mapped to the given initiator."""
         lun_list_api = netapp_api.NaElement('lun-initiator-list-map-info')
diff --git a/cinder/volume/drivers/netapp/dataontap/client/client_cmode.py b/cinder/volume/drivers/netapp/dataontap/client/client_cmode.py
index 9af979ad268..f17aae9150a 100644
--- a/cinder/volume/drivers/netapp/dataontap/client/client_cmode.py
+++ b/cinder/volume/drivers/netapp/dataontap/client/client_cmode.py
@@ -50,9 +50,9 @@ class Client(client_base.Client):
         self.connection.set_api_version(1, 15)
         (major, minor) = self.get_ontapi_version(cached=False)
         self.connection.set_api_version(major, minor)
-        self._init_features()
         ontap_version = self.get_ontap_version(cached=False)
         self.connection.set_ontap_version(ontap_version)
+        self._init_features()
 
     def _init_features(self):
         super(Client, self)._init_features()
@@ -65,6 +65,8 @@ class Client(client_base.Client):
         ontapi_1_100 = ontapi_version >= (1, 100)
         ontapi_1_1xx = (1, 100) <= ontapi_version < (1, 200)
         ontapi_1_60 = ontapi_version >= (1, 160)
+        ontapi_1_40 = ontapi_version >= (1, 140)
+        ontapi_1_50 = ontapi_version >= (1, 150)
 
         nodes_info = self._get_cluster_nodes_info()
         for node in nodes_info:
@@ -99,6 +101,12 @@ class Client(client_base.Client):
         self.features.add_feature('CLUSTER_PEER_POLICY', supported=ontapi_1_30)
         self.features.add_feature('FLEXVOL_ENCRYPTION', supported=ontapi_1_1xx)
 
+        self.features.add_feature('ADAPTIVE_QOS', supported=ontapi_1_40)
+        self.features.add_feature('ADAPTIVE_QOS_BLOCK_SIZE',
+                                  supported=ontapi_1_50)
+        self.features.add_feature('ADAPTIVE_QOS_EXPECTED_IOPS_ALLOCATION',
+                                  supported=ontapi_1_50)
+
         LOG.info('Reported ONTAPI Version: %(major)s.%(minor)s',
                  {'major': ontapi_version[0], 'minor': ontapi_version[1]})
 
@@ -511,10 +519,39 @@ class Client(client_base.Client):
 
         return igroup_list
 
+    def _validate_qos_policy_group(self, is_adaptive, spec=None,
+                                   qos_min_support=False):
+        if is_adaptive and not self.features.ADAPTIVE_QOS:
+            msg = _("Adaptive QoS feature requires ONTAP 9.4 or later.")
+            raise na_utils.NetAppDriverException(msg)
+
+        if not spec:
+            return
+
+        qos_spec_support = [
+            {'key': 'min_throughput',
+             'support': qos_min_support,
+             'reason': _('is not supported by this back end.')},
+            {'key': 'block_size',
+             'support': self.features.ADAPTIVE_QOS_BLOCK_SIZE,
+             'reason': _('requires ONTAP >= 9.5.')},
+            {'key': 'expected_iops_allocation',
+             'support': self.features.ADAPTIVE_QOS_EXPECTED_IOPS_ALLOCATION,
+             'reason': _('requires ONTAP >= 9.5.')},
+        ]
+        for feature in qos_spec_support:
+            if feature['key'] in spec and not feature['support']:
+                msg = '%(key)s %(reason)s'
+                raise na_utils.NetAppDriverException(msg % {
+                    'key': feature['key'],
+                    'reason': feature['reason']})
+
     def clone_lun(self, volume, name, new_name, space_reserved='true',
                   qos_policy_group_name=None, src_block=0, dest_block=0,
                   block_count=0, source_snapshot=None, is_snapshot=False,
                   qos_policy_group_is_adaptive=False):
+        self._validate_qos_policy_group(qos_policy_group_is_adaptive)
+
         # ONTAP handles only 128 MB per call as of v9.1
         bc_limit = 2 ** 18  # 2^18 blocks * 512 bytes/block = 128 MB
         z_calls = int(math.ceil(block_count / float(bc_limit)))
@@ -541,13 +578,9 @@ class Client(client_base.Client):
             clone_create = netapp_api.NaElement.create_node_with_children(
                 'clone-create', **zapi_args)
             if qos_policy_group_name is not None:
-                if qos_policy_group_is_adaptive:
-                    clone_create.add_new_child(
-                        'qos-adaptive-policy-group-name',
-                        qos_policy_group_name)
-                else:
-                    clone_create.add_new_child('qos-policy-group-name',
-                                               qos_policy_group_name)
+                child_name = 'qos-%spolicy-group-name' % (
+                    'adaptive-' if qos_policy_group_is_adaptive else '')
+                clone_create.add_new_child(child_name, qos_policy_group_name)
             if block_count > 0:
                 block_ranges = netapp_api.NaElement("block-ranges")
                 segments = int(math.ceil(block_count / float(bc_limit)))
@@ -589,6 +622,8 @@ class Client(client_base.Client):
     def file_assign_qos(self, flex_vol, qos_policy_group_name,
                         qos_policy_group_is_adaptive, file_path):
         """Assigns the named QoS policy-group to a file."""
+        self._validate_qos_policy_group(qos_policy_group_is_adaptive)
+
         qos_arg_name = "qos-%spolicy-group-name" % (
             "adaptive-" if qos_policy_group_is_adaptive else "")
         api_args = {
@@ -612,33 +647,46 @@ class Client(client_base.Client):
             return
 
         spec = qos_policy_group_info.get('spec')
-        if spec:
-            if spec.get('min_throughput') and not qos_min_support:
-                msg = _('QoS min_throughput is not supported by this back '
-                        'end.')
-                raise na_utils.NetAppDriverException(msg)
+
+        if not spec:
+            return
+
+        is_adaptive = na_utils.is_qos_policy_group_spec_adaptive(
+            qos_policy_group_info)
+        self._validate_qos_policy_group(is_adaptive, spec=spec,
+                                        qos_min_support=qos_min_support)
+        if is_adaptive:
+            if not self.qos_policy_group_exists(spec['policy_name'],
+                                                is_adaptive=True):
+                self.qos_adaptive_policy_group_create(spec)
+            else:
+                self.qos_adaptive_policy_group_modify(spec)
+        else:
             if not self.qos_policy_group_exists(spec['policy_name']):
                 self.qos_policy_group_create(spec)
             else:
                 self.qos_policy_group_modify(spec)
 
-    def qos_policy_group_exists(self, qos_policy_group_name):
+    def qos_policy_group_exists(self, qos_policy_group_name,
+                                is_adaptive=False):
         """Checks if a QOS policy group exists."""
+        query_name = 'qos-%spolicy-group-info' % (
+            'adaptive-' if is_adaptive else '')
+        request_name = 'qos-%spolicy-group-get-iter' % (
+            'adaptive-' if is_adaptive else '')
         api_args = {
             'query': {
-                'qos-policy-group-info': {
+                query_name: {
                     'policy-group': qos_policy_group_name,
                 },
             },
             'desired-attributes': {
-                'qos-policy-group-info': {
+                query_name: {
                     'policy-group': None,
                 },
             },
         }
-        result = self.connection.send_request('qos-policy-group-get-iter',
-                                              api_args,
-                                              False)
+        result = self.connection.send_request(request_name, api_args, False)
         return self._has_records(result)
 
     def _qos_spec_to_api_args(self, spec, **kwargs):
@@ -656,29 +704,39 @@ class Client(client_base.Client):
         return self.connection.send_request(
             'qos-policy-group-create', api_args, False)
 
+    def qos_adaptive_policy_group_create(self, spec):
+        """Creates a QOS adaptive policy group."""
+        api_args = self._qos_spec_to_api_args(
+            spec, vserver=self.vserver)
+        return self.connection.send_request(
+            'qos-adaptive-policy-group-create', api_args, False)
+
     def qos_policy_group_modify(self, spec):
         """Modifies a QOS policy group."""
         api_args = self._qos_spec_to_api_args(spec)
         return self.connection.send_request(
             'qos-policy-group-modify', api_args, False)
 
-    def qos_policy_group_delete(self, qos_policy_group_name):
-        """Attempts to delete a QOS policy group."""
-        api_args = {'policy-group': qos_policy_group_name}
+    def qos_adaptive_policy_group_modify(self, spec):
+        """Modifies a QOS adaptive policy group."""
+        api_args = self._qos_spec_to_api_args(spec)
         return self.connection.send_request(
-            'qos-policy-group-delete', api_args, False)
+            'qos-adaptive-policy-group-modify', api_args, False)
 
-    def qos_policy_group_rename(self, qos_policy_group_name, new_name):
+    def qos_policy_group_rename(self, qos_policy_group_name, new_name,
+                                is_adaptive=False):
         """Renames a QOS policy group."""
+        request_name = 'qos-%spolicy-group-rename' % (
+            'adaptive-' if is_adaptive else '')
         api_args = {
             'policy-group-name': qos_policy_group_name,
             'new-name': new_name,
         }
-        return self.connection.send_request(
-            'qos-policy-group-rename', api_args, False)
+        return self.connection.send_request(request_name, api_args, False)
 
-    def mark_qos_policy_group_for_deletion(self, qos_policy_group_info):
-        """Do (soft) delete of backing QOS policy group for a cinder volume."""
+    def mark_qos_policy_group_for_deletion(self, qos_policy_group_info,
+                                           is_adaptive=False):
+        """Soft delete a QOS policy group backing a cinder volume."""
         if qos_policy_group_info is None:
             return
 
@@ -690,11 +748,12 @@ class Client(client_base.Client):
         # we instead rename the QoS policy group using a specific pattern and
         # later attempt on a best effort basis to delete any QoS policy groups
         # matching that pattern.
-        if spec is not None:
+        if spec:
             current_name = spec['policy_name']
             new_name = client_base.DELETED_PREFIX + current_name
             try:
-                self.qos_policy_group_rename(current_name, new_name)
+                self.qos_policy_group_rename(current_name, new_name,
+                                             is_adaptive)
             except netapp_api.NaApiError as ex:
                 LOG.warning('Rename failure in cleanup of cDOT QOS policy '
                             'group %(name)s: %(ex)s',
@@ -703,11 +762,15 @@ class Client(client_base.Client):
         # Attempt to delete any QoS policies named "delete-openstack-*".
         self.remove_unused_qos_policy_groups()
 
-    def remove_unused_qos_policy_groups(self):
-        """Deletes all QOS policy groups that are marked for deletion."""
+    def _send_qos_policy_group_delete_iter_request(self, is_adaptive=False):
+        request_name = 'qos-%spolicy-group-delete-iter' % (
+            'adaptive-' if is_adaptive else '')
+        query_name = 'qos-%spolicy-group-info' % (
+            'adaptive-' if is_adaptive else '')
+
         api_args = {
             'query': {
-                'qos-policy-group-info': {
+                query_name: {
                     'policy-group': '%s*' % client_base.DELETED_PREFIX,
                     'vserver': self.vserver,
                 }
@@ -719,18 +782,32 @@ class Client(client_base.Client):
         }
 
         try:
-            self.connection.send_request(
-                'qos-policy-group-delete-iter', api_args, False)
+            self.connection.send_request(request_name, api_args, False)
         except netapp_api.NaApiError as ex:
-            msg = 'Could not delete QOS policy groups. Details: %(ex)s'
-            msg_args = {'ex': ex}
+            msg = ('Could not delete QOS %(prefix)spolicy groups. '
+                   'Details: %(ex)s')
+            msg_args = {
+                'prefix': 'adaptive ' if is_adaptive else '',
+                'ex': ex,
+            }
             LOG.debug(msg, msg_args)
 
-    def set_lun_qos_policy_group(self, path, qos_policy_group):
+    def remove_unused_qos_policy_groups(self):
+        """Deletes all QOS policy groups that are marked for deletion."""
+        self._send_qos_policy_group_delete_iter_request()
+        if self.features.ADAPTIVE_QOS:
+            self._send_qos_policy_group_delete_iter_request(is_adaptive=True)
+
+    def set_lun_qos_policy_group(self, path, qos_policy_group,
+                                 is_adaptive=False):
         """Sets qos_policy_group on a LUN."""
+        self._validate_qos_policy_group(is_adaptive)
+
+        policy_group_key = 'qos-%spolicy-group' % (
+            'adaptive-' if is_adaptive else '')
         api_args = {
             'path': path,
-            'qos-policy-group': qos_policy_group,
+            policy_group_key: qos_policy_group,
         }
         return self.connection.send_request(
             'lun-set-qos-policy-group', api_args)
diff --git a/cinder/volume/drivers/netapp/dataontap/nfs_cmode.py b/cinder/volume/drivers/netapp/dataontap/nfs_cmode.py
index df95aa8523a..1dab7a4b782 100644
--- a/cinder/volume/drivers/netapp/dataontap/nfs_cmode.py
+++ b/cinder/volume/drivers/netapp/dataontap/nfs_cmode.py
@@ -160,13 +160,15 @@ class NetAppCmodeNfsDriver(nfs_base.NetAppNfsDriver,
                 self.ssc_library.get_ssc_flexvol_names())
 
     def _do_qos_for_volume(self, volume, extra_specs, cleanup=True):
-        qos_policy_group_is_adaptive = volume_utils.is_boolean_str(
-            extra_specs.get('netapp:qos_policy_group_is_adaptive'))
         try:
             qos_policy_group_info = na_utils.get_valid_qos_policy_group_info(
                 volume, extra_specs)
             pool = volume_utils.extract_host(volume['host'], level='pool')
             qos_min_support = self.ssc_library.is_qos_min_supported(pool)
+            qos_policy_group_is_adaptive = (volume_utils.is_boolean_str(
+                extra_specs.get('netapp:qos_policy_group_is_adaptive')) or
+                na_utils.is_qos_policy_group_spec_adaptive(
+                    qos_policy_group_info))
             self.zapi_client.provision_qos_policy_group(qos_policy_group_info,
                                                         qos_min_support)
             self._set_qos_policy_group_on_volume(volume, qos_policy_group_info,
@@ -416,8 +418,10 @@ class NetAppCmodeNfsDriver(nfs_base.NetAppNfsDriver,
         try:
             qos_policy_group_info = na_utils.get_valid_qos_policy_group_info(
                 volume)
-            self.zapi_client.mark_qos_policy_group_for_deletion(
+            is_adaptive = na_utils.is_qos_policy_group_spec_adaptive(
                 qos_policy_group_info)
+            self.zapi_client.mark_qos_policy_group_for_deletion(
+                qos_policy_group_info, is_adaptive)
         except Exception:
             # Don't blow up here if something went wrong de-provisioning the
             # QoS policy for the volume.
diff --git a/cinder/volume/drivers/netapp/utils.py b/cinder/volume/drivers/netapp/utils.py
index a8bf11901b1..695a8697c0d 100644
--- a/cinder/volume/drivers/netapp/utils.py
+++ b/cinder/volume/drivers/netapp/utils.py
@@ -60,6 +60,23 @@ MAX_QOS_KEYS = frozenset([
     'maxBPS',
     'maxBPSperGiB',
 ])
+ADAPTIVE_QOS_KEYS = frozenset([
+    'expectedIOPSperGiB',
+    'peakIOPSperGiB',
+    'expectedIOPSAllocation',
+    'peakIOPSAllocation',
+    'absoluteMinIOPS',
+    'blockSize',
+])
+QOS_ADAPTIVE_POLICY_GROUP_SPEC_KEYS = frozenset([
+    'expected_iops',
+    'peak_iops',
+    'expected_iops_allocation',
+    'peak_iops_allocation',
+    'absolute_min_iops',
+    'block_size',
+    'policy_name',
+])
 BACKEND_QOS_CONSUMERS = frozenset(['back-end', 'both'])
 
 # Secret length cannot be less than 96 bits. http://tools.ietf.org/html/rfc3723
@@ -219,10 +236,12 @@ def validate_qos_spec(qos_spec):
 
     normalized_min_keys = [key.lower() for key in MIN_QOS_KEYS]
     normalized_max_keys = [key.lower() for key in MAX_QOS_KEYS]
+    normalized_aqos_keys = [key.lower() for key in ADAPTIVE_QOS_KEYS]
 
     unrecognized_keys = [
         k for k in qos_spec.keys()
-        if k.lower() not in normalized_max_keys + normalized_min_keys]
+        if k.lower() not in
+        normalized_max_keys + normalized_min_keys + normalized_aqos_keys]
 
     if unrecognized_keys:
         msg = _('Unrecognized QOS keywords: "%s"') % unrecognized_keys
@@ -240,6 +259,13 @@ def validate_qos_spec(qos_spec):
         msg = _('Only one maximum limit can be set in a QoS spec.')
         raise exception.Invalid(msg)
 
+    aqos_dict = {k: v for k, v in qos_spec.items()
+                 if k.lower() in normalized_aqos_keys}
+    if aqos_dict and (min_dict or max_dict):
+        msg = _('Adaptive QoS specs and non-adaptive QoS specs '
+                'cannot be used together.')
+        raise exception.Invalid(msg)
+
 
 def get_volume_type_from_volume(volume):
     """Provides volume type associated with volume."""
@@ -313,6 +339,42 @@ def map_qos_spec(qos_spec, volume):
     return policy
 
 
+def map_aqos_spec(qos_spec, volume):
+    """Map Cinder QOS spec to Adaptive QoS values."""
+    if qos_spec is None:
+        return None
+
+    qos_spec = map_dict_to_lower(qos_spec)
+    spec = dict(policy_name=get_qos_policy_group_name(volume))
+
+    # Adaptive QoS specs
+    if 'expectediopspergib' in qos_spec:
+        spec['expected_iops'] = (
+            '%sIOPS/GB' % qos_spec['expectediopspergib'])
+    if 'peakiopspergib' in qos_spec:
+        spec['peak_iops'] = '%sIOPS/GB' % qos_spec['peakiopspergib']
+    if 'expectediopsallocation' in qos_spec:
+        spec['expected_iops_allocation'] = qos_spec['expectediopsallocation']
+    if 'peakiopsallocation' in qos_spec:
+        spec['peak_iops_allocation'] = qos_spec['peakiopsallocation']
+    if 'absoluteminiops' in qos_spec:
+        spec['absolute_min_iops'] = '%sIOPS' % qos_spec['absoluteminiops']
+    if 'blocksize' in qos_spec:
+        spec['block_size'] = qos_spec['blocksize']
+
+    if 'peak_iops' not in spec or 'expected_iops' not in spec:
+        msg = _('Adaptive QoS requires the expected property and '
+                'the peak property set together.')
+        raise exception.Invalid(msg)
+
+    if spec['peak_iops'] < spec['expected_iops']:
+        msg = _('Adaptive maximum limit should be greater than or equal to '
+                'the adaptive minimum limit.')
+        raise exception.Invalid(msg)
+
+    return spec
+
+
 def map_dict_to_lower(input_dict):
     """Return an equivalent to the input dictionary with lower-case keys."""
     lower_case_dict = {}
@@ -389,11 +451,35 @@ def get_valid_qos_policy_group_info(volume, extra_specs=None):
 
 def get_valid_backend_qos_spec_from_volume_type(volume, volume_type):
     """Given a volume type, return the associated Cinder QoS spec."""
-    spec_key_values = get_backend_qos_spec_from_volume_type(volume_type)
-    if spec_key_values is None:
+    spec_dict = get_backend_qos_spec_from_volume_type(volume_type)
+    if spec_dict is None:
         return None
-    validate_qos_spec(spec_key_values)
-    return map_qos_spec(spec_key_values, volume)
+    validate_qos_spec(spec_dict)
+    map_spec = (map_aqos_spec
+                if is_qos_adaptive(spec_dict)
+                else map_qos_spec)
+    return map_spec(spec_dict, volume)
+
+
+def is_qos_adaptive(spec_dict):
+    if not spec_dict:
+        return False
+
+    normalized_aqos_keys = [key.lower() for key in ADAPTIVE_QOS_KEYS]
+    return all(key in normalized_aqos_keys
+               for key in map_dict_to_lower(spec_dict).keys())
+
+
+def is_qos_policy_group_spec_adaptive(policy):
+    if not policy:
+        return False
+
+    spec = policy.get('spec')
+    if not spec:
+        return False
+
+    return all(key in QOS_ADAPTIVE_POLICY_GROUP_SPEC_KEYS
+               for key in map_dict_to_lower(spec).keys())
 
 
 def get_backend_qos_spec_from_volume_type(volume_type):
@@ -408,8 +494,7 @@ def get_backend_qos_spec_from_volume_type(volume_type):
     # Front end QoS specs are handled by libvirt and we ignore them here.
     if consumer not in BACKEND_QOS_CONSUMERS:
         return None
-    spec_key_values = qos_spec['specs']
-    return spec_key_values
+    return qos_spec['specs']
 
 
 def check_for_invalid_qos_spec_combination(info, volume_type):
diff --git a/releasenotes/notes/bp-netapp-ontap-adaptive-qos-45891585a91eab75.yaml b/releasenotes/notes/bp-netapp-ontap-adaptive-qos-45891585a91eab75.yaml
new file mode 100644
index 00000000000..969bdd8b651
--- /dev/null
+++ b/releasenotes/notes/bp-netapp-ontap-adaptive-qos-45891585a91eab75.yaml
@@ -0,0 +1,11 @@
+---
+features:
+  - |
+    NetApp ONTAP driver: Added support for Adaptive QoS specs. The driver now
+    accepts ``expectedIOPSperGiB``, ``peakIOPSperGiB``, ``expectedIOPSAllocation``,
+    ``peakIOPSAllocation``, ``absoluteMinIOPS`` and ``blockSize``. The field
+    ``peakIOPSperGiB`` and the field ``expectedIOPSperGiB`` are required together.
+    The ``expectedIOPSperGiB`` and ``absoluteMinIOPS`` specs are only guaranteed
+    by ONTAP AFF systems. All specs can only be used with ONTAP version equal
+    or greater than 9.4, excepting the ``expectedIOPSAllocation`` and
+    ``blockSize`` specs which require at least 9.5.