diff --git a/manila/share/drivers/netapp/dataontap/client/client_cmode.py b/manila/share/drivers/netapp/dataontap/client/client_cmode.py
index 06e691a40d..ffbf9cab6b 100644
--- a/manila/share/drivers/netapp/dataontap/client/client_cmode.py
+++ b/manila/share/drivers/netapp/dataontap/client/client_cmode.py
@@ -2827,11 +2827,46 @@ def disable_compression(self, volume_name):
         }
         self.send_request('sis-set-config', api_args)
 
+    @na_utils.trace
+    def _wait_for_efficiency_job(self, job_id, volume_name, timeout=300):
+        """Wait for efficiency async job to complete."""
+        interval = 5
+        retries = int(timeout / interval) or 1
+
+        @manila_utils.retry(retry_param=exception.ShareBackendException,
+                            interval=interval,
+                            retries=retries, backoff_rate=1)
+        def _wait():
+            job_state = self.get_job_state(job_id)
+            LOG.debug("Waiting for efficiency job %(job_id)s on volume "
+                      "%(volume)s, state: %(state)s.",
+                      {'job_id': job_id, 'volume': volume_name,
+                       'state': job_state})
+            if job_state in ('failure', 'error'):
+                msg = _("Efficiency job %(job_id)s failed for volume "
+                        "%(volume)s.")
+                raise exception.NetAppException(
+                    msg % {'job_id': job_id, 'volume': volume_name})
+            if job_state != 'success':
+                raise exception.ShareBackendException(
+                    msg=_("Efficiency job not yet complete."))
+
+        try:
+            _wait()
+        except exception.ShareBackendException:
+            msg = _("Timeout waiting for efficiency job %(job_id)s on "
+                    "volume %(volume)s.")
+            raise exception.NetAppException(
+                msg % {'job_id': job_id, 'volume': volume_name})
+
     @na_utils.trace
     def enable_dedupe_async(self, volume_name):
         """Enable deduplication on FlexVol/FlexGroup volume asynchronously."""
         api_args = {'volume-name': volume_name}
-        self.send_request('sis-enable-async', api_args)
+        result = self.send_request('sis-enable-async', api_args)
+        job_id = result.get_child_content('result-jobid')
+        if job_id:
+            self._wait_for_efficiency_job(job_id, volume_name)
 
     @na_utils.trace
     def disable_dedupe_async(self, volume_name):
diff --git a/manila/tests/share/drivers/netapp/dataontap/client/test_client_cmode.py b/manila/tests/share/drivers/netapp/dataontap/client/test_client_cmode.py
index 9d8cee7e6d..4aaf365465 100644
--- a/manila/tests/share/drivers/netapp/dataontap/client/test_client_cmode.py
+++ b/manila/tests/share/drivers/netapp/dataontap/client/test_client_cmode.py
@@ -3725,14 +3725,23 @@ def test_disable_compression(self):
         self.client.send_request.assert_called_once_with('sis-set-config',
                                                          sis_set_config_args)
 
-    def test_enable_dedupe_async(self):
-        self.mock_object(self.client, 'send_request')
+    @ddt.data(str(fake.JOB_ID), None)
+    def test_enable_dedupe_async(self, job_id):
+        mock_result = mock.Mock()
+        mock_result.get_child_content.return_value = job_id
+        self.mock_object(self.client, 'send_request',
+                         mock.Mock(return_value=mock_result))
+        mock_wait = self.mock_object(self.client, '_wait_for_efficiency_job')
 
         self.client.enable_dedupe_async(fake.SHARE_NAME)
 
         sis_enable_args = {'volume-name': fake.SHARE_NAME}
         self.client.send_request.assert_called_once_with(
             'sis-enable-async', sis_enable_args)
+        if job_id:
+            mock_wait.assert_called_once_with(job_id, fake.SHARE_NAME)
+        else:
+            mock_wait.assert_not_called()
 
     def test_disable_dedupe_async(self):
         self.mock_object(self.client, 'send_request')
@@ -9345,6 +9354,34 @@ def test_get_job_state_not_unique(self):
                           self.client.get_job_state,
                           fake.JOB_ID)
 
+    def test__wait_for_efficiency_job_success(self):
+        self.mock_object(self.client, 'get_job_state',
+                         mock.Mock(return_value='success'))
+
+        self.client._wait_for_efficiency_job(fake.JOB_ID, fake.SHARE_NAME)
+
+        self.client.get_job_state.assert_called_once_with(fake.JOB_ID)
+
+    @ddt.data('failure', 'error')
+    def test__wait_for_efficiency_job_failed(self, job_state):
+        self.mock_object(self.client, 'get_job_state',
+                         mock.Mock(return_value=job_state))
+
+        self.assertRaises(
+            exception.NetAppException,
+            self.client._wait_for_efficiency_job,
+            fake.JOB_ID, fake.SHARE_NAME)
+
+    def test__wait_for_efficiency_job_timeout(self):
+        self.mock_object(self.client, 'get_job_state',
+                         mock.Mock(return_value='running'))
+
+        self.assertRaises(
+            exception.NetAppException,
+            self.client._wait_for_efficiency_job,
+            fake.JOB_ID, fake.SHARE_NAME,
+            timeout=0)
+
     def test_check_snaprestore_license_svm_scoped_notfound(self):
         self.mock_object(self.client,
                          'restore_snapshot',