diff --git a/glance/async_/flows/plugins/image_conversion.py b/glance/async_/flows/plugins/image_conversion.py
index 247300400c..ec92a31537 100644
--- a/glance/async_/flows/plugins/image_conversion.py
+++ b/glance/async_/flows/plugins/image_conversion.py
@@ -62,19 +62,22 @@ class _ConvertImage(task.Task):
 
     default_provides = 'file_path'
 
-    def __init__(self, context, task_id, task_type,
-                 image_repo, image_id):
+    def __init__(self, context, task_id, task_type, action_wrapper):
         self.context = context
         self.task_id = task_id
         self.task_type = task_type
-        self.image_repo = image_repo
-        self.image_id = image_id
+        self.action_wrapper = action_wrapper
+        self.image_id = action_wrapper.image_id
         self.dest_path = ""
         self.python = CONF.wsgi.python_interpreter
         super(_ConvertImage, self).__init__(
             name='%s-Convert_Image-%s' % (task_type, task_id))
 
     def execute(self, file_path, **kwargs):
+        with self.action_wrapper as action:
+            return self._execute(action, file_path, **kwargs)
+
+    def _execute(self, action, file_path, **kwargs):
 
         target_format = CONF.image_conversion.output_format
         # TODO(jokke): Once we support other schemas we need to take them into
@@ -111,13 +114,11 @@ class _ConvertImage(task.Task):
             raise RuntimeError(msg)
 
         virtual_size = metadata.get('virtual-size', 0)
-        image = self.image_repo.get(self.image_id)
-        image.virtual_size = virtual_size
+        action.set_image_attribute(virtual_size=virtual_size)
 
         if source_format == target_format:
             LOG.debug("Source is already in target format, "
                       "not doing conversion for %s", self.image_id)
-            self.image_repo.save(image)
             return file_path
 
         try:
@@ -135,9 +136,8 @@ class _ConvertImage(task.Task):
         if stderr:
             raise RuntimeError(stderr)
 
-        image.disk_format = target_format
-        image.container_format = 'bare'
-        self.image_repo.save(image)
+        action.set_image_attribute(disk_format=target_format,
+                                   container_format='bare')
 
         os.remove(src_path)
 
@@ -161,14 +161,13 @@ def get_flow(**kwargs):
     :param task_type: Type of the task.
     :param image_repo: Image repository used.
     :param image_id: Image ID
+    :param action_wrapper: An api_image_import.ActionWrapper.
     """
     context = kwargs.get('context')
     task_id = kwargs.get('task_id')
     task_type = kwargs.get('task_type')
-    image_repo = kwargs.get('image_repo')
-    image_id = kwargs.get('image_id')
+    action_wrapper = kwargs.get('action_wrapper')
 
     return lf.Flow(task_type).add(
-        _ConvertImage(context, task_id, task_type,
-                      image_repo, image_id),
+        _ConvertImage(context, task_id, task_type, action_wrapper)
     )
diff --git a/glance/tests/unit/async_/flows/plugins/test_image_conversion.py b/glance/tests/unit/async_/flows/plugins/test_image_conversion.py
index e8296b4569..9327748ddb 100644
--- a/glance/tests/unit/async_/flows/plugins/test_image_conversion.py
+++ b/glance/tests/unit/async_/flows/plugins/test_image_conversion.py
@@ -21,6 +21,7 @@ import glance_store
 from oslo_concurrency import processutils
 from oslo_config import cfg
 
+import glance.async_.flows.api_image_import as import_flow
 import glance.async_.flows.plugins.image_conversion as image_conversion
 from glance.async_ import utils as async_utils
 from glance.common import utils
@@ -83,24 +84,32 @@ class TestConvertImageTask(test_utils.BaseTestCase):
                                                task_time_to_live=task_ttl,
                                                task_input=task_input)
 
+        self.image.extra_properties = {
+            'os_glance_import_task': self.task.task_id}
+        self.wrapper = import_flow.ImportActionWrapper(self.img_repo,
+                                                       self.image_id,
+                                                       self.task.task_id)
+
     @mock.patch.object(os, 'remove')
     def test_image_convert_success(self, mock_os_remove):
         mock_os_remove.return_value = None
         image_convert = image_conversion._ConvertImage(self.context,
                                                        self.task.task_id,
                                                        self.task_type,
-                                                       self.img_repo,
-                                                       self.image_id)
+                                                       self.wrapper)
 
         self.task_repo.get.return_value = self.task
         image = mock.MagicMock(image_id=self.image_id, virtual_size=None,
+                               extra_properties={
+                                   'os_glance_import_task': self.task.task_id},
                                disk_format='qcow2')
         self.img_repo.get.return_value = image
 
         with mock.patch.object(processutils, 'execute') as exc_mock:
             exc_mock.return_value = ("", None)
             with mock.patch.object(json, 'loads') as jloads_mock:
-                jloads_mock.return_value = {'format': 'raw'}
+                jloads_mock.return_value = {'format': 'raw',
+                                            'virtual-size': 123}
                 image_convert.execute('file:///test/path.raw')
 
                 # NOTE(hemanthm): Asserting that the source format is passed
@@ -110,12 +119,15 @@ class TestConvertImageTask(test_utils.BaseTestCase):
                 self.assertIn('-f', exc_mock.call_args[0])
                 self.assertEqual("qcow2", image.disk_format)
 
+        self.assertEqual('bare', image.container_format)
+        self.assertEqual('qcow2', image.disk_format)
+        self.assertEqual(123, image.virtual_size)
+
     def _setup_image_convert_info_fail(self):
         image_convert = image_conversion._ConvertImage(self.context,
                                                        self.task.task_id,
                                                        self.task_type,
-                                                       self.img_repo,
-                                                       self.image_id)
+                                                       self.wrapper)
 
         self.task_repo.get.return_value = self.task
         image = mock.MagicMock(image_id=self.image_id, virtual_size=None,
@@ -239,8 +251,7 @@ class TestConvertImageTask(test_utils.BaseTestCase):
         image_convert = image_conversion._ConvertImage(self.context,
                                                        self.task.task_id,
                                                        self.task_type,
-                                                       self.img_repo,
-                                                       self.image_id)
+                                                       self.wrapper)
 
         self.task_repo.get.return_value = self.task