diff --git a/cinder/tests/unit/api/contrib/test_snapshot_manage.py b/cinder/tests/unit/api/contrib/test_snapshot_manage.py
index 152db592606..933e3da62e5 100644
--- a/cinder/tests/unit/api/contrib/test_snapshot_manage.py
+++ b/cinder/tests/unit/api/contrib/test_snapshot_manage.py
@@ -135,8 +135,7 @@ class SnapshotManageTest(test.TestCase):
         args = mock_create_snapshot.call_args[0]
         named_args = mock_create_snapshot.call_args[1]
         self.assertEqual(fake.VOLUME_ID, args[1].get('id'))
-        # We should commit quota in cinder-volume layer for this operation.
-        self.assertFalse(named_args['commit_quota'])
+        self.assertTrue(named_args['commit_quota'])
 
         # Check the volume_rpcapi.manage_existing_snapshot was called with
         # correct arguments.
diff --git a/cinder/tests/unit/volume/flows/test_manage_snapshot_flow.py b/cinder/tests/unit/volume/flows/test_manage_snapshot_flow.py
index a2a1af110f1..102f4795a1a 100644
--- a/cinder/tests/unit/volume/flows/test_manage_snapshot_flow.py
+++ b/cinder/tests/unit/volume/flows/test_manage_snapshot_flow.py
@@ -16,6 +16,7 @@
 
 # TODO(mdovgal): add tests for other TaskFlow cases
 
+import ddt
 import mock
 
 from cinder import context
@@ -26,6 +27,7 @@ from cinder.tests.unit import fake_volume
 from cinder.volume.flows.manager import manage_existing_snapshot as manager
 
 
+@ddt.ddt
 class ManageSnapshotFlowTestCase(test.TestCase):
     def setUp(self):
         super(ManageSnapshotFlowTestCase, self).setUp()
@@ -58,9 +60,10 @@ class ManageSnapshotFlowTestCase(test.TestCase):
     @mock.patch('cinder.objects.volume.Volume.get_by_id')
     def test_quota_reservation_task(self, mock_get_vol_by_id, mock_type_get,
                                     mock_quota_reserve):
-        fake_size = 1
+        volume_size = 1
+        fake_size = '2'
         fake_snap = fake_snapshot.fake_snapshot_obj(self.ctxt,
-                                                    volume_size=fake_size)
+                                                    volume_size=volume_size)
         fake_snap.save = mock.MagicMock()
         fake_vol = fake_volume.fake_volume_obj(
             self.ctxt, id=fake.VOLUME_ID, volume_type_id=fake.VOLUME_TYPE_ID)
@@ -74,3 +77,51 @@ class ManageSnapshotFlowTestCase(test.TestCase):
                         'gigabytes_fake_type_name': 1,
                         'snapshots_fake_type_name': 1}
         mock_quota_reserve.assert_called_once_with(self.ctxt, **reserve_opts)
+
+    @ddt.data(True, False)
+    @mock.patch('cinder.quota.QuotaEngine.reserve')
+    @mock.patch('cinder.db.sqlalchemy.api.volume_type_get')
+    @mock.patch('cinder.objects.volume.Volume.get_by_id')
+    def test_quota_reservation_task_with_update_flag(
+            self, need_update, mock_get_vol_by_id,
+            mock_type_get, mock_quota_reserve):
+        volume_size = 1
+        fake_size = '2'
+        fake_snap = fake_snapshot.fake_snapshot_obj(self.ctxt,
+                                                    volume_size=volume_size)
+        fake_snap.save = mock.MagicMock()
+        fake_vol = fake_volume.fake_volume_obj(
+            self.ctxt, id=fake.VOLUME_ID, volume_type_id=fake.VOLUME_TYPE_ID)
+        mock_get_vol_by_id.return_value = fake_vol
+        mock_type_get.return_value = {'name': 'fake_type_name'}
+
+        task = manager.QuotaReserveTask()
+        task.execute(self.ctxt, fake_size, fake_snap,
+                     {'update_size': need_update})
+
+        reserve_opts = {'gigabytes': 1, 'gigabytes_fake_type_name': 1}
+
+        if not need_update:
+            reserve_opts.update({'snapshots': 1,
+                                 'snapshots_fake_type_name': 1})
+        mock_quota_reserve.assert_called_once_with(self.ctxt, **reserve_opts)
+
+    def test_prepare_for_quota_reserveration_task_execute(self):
+        mock_db = mock.MagicMock()
+        mock_driver = mock.MagicMock()
+        mock_manage_existing_ref = mock.MagicMock()
+        mock_get_snapshot_size = self.mock_object(
+            mock_driver, 'manage_existing_snapshot_get_size')
+        mock_get_snapshot_size.return_value = '5'
+
+        fake_snap = fake_snapshot.fake_snapshot_obj(self.ctxt,
+                                                    volume_size=1)
+        task = manager.PrepareForQuotaReservationTask(mock_db, mock_driver)
+
+        result = task.execute(self.ctxt, fake_snap, mock_manage_existing_ref)
+
+        self.assertEqual(fake_snap, result['snapshot_properties'])
+        self.assertEqual('5', result['size'])
+        mock_get_snapshot_size.assert_called_once_with(
+            snapshot=fake_snap,
+            existing_ref=mock_manage_existing_ref)
diff --git a/cinder/tests/unit/volume/flows/test_manage_volume_flow.py b/cinder/tests/unit/volume/flows/test_manage_volume_flow.py
index ed5b6f6ab19..bcbe0d109f3 100644
--- a/cinder/tests/unit/volume/flows/test_manage_volume_flow.py
+++ b/cinder/tests/unit/volume/flows/test_manage_volume_flow.py
@@ -110,7 +110,8 @@ class ManageVolumeFlowTestCase(test.TestCase):
             'volume': mock.sentinel.volume,
             'manage_existing_ref': mock.sentinel.ref,
             'group_snapshot': None,
-            'optional_args': {'is_quota_committed': False},
+            'optional_args': {'is_quota_committed': False,
+                              'update_size': True}
         }
 
         manager.get_flow(
diff --git a/cinder/volume/api.py b/cinder/volume/api.py
index 3e3320942f9..e0ba5760f3e 100644
--- a/cinder/volume/api.py
+++ b/cinder/volume/api.py
@@ -1709,6 +1709,10 @@ class API(base.Base):
             'metadata': metadata,
             'availability_zone': availability_zone,
             'bootable': bootable,
+            'size': 0,
+            'group_snapshot': None,
+            'optional_args': {'is_quota_committed': False},
+            'volume_type_id': None if not volume_type else volume_type['id'],
         }
 
         try:
@@ -1749,7 +1753,7 @@ class API(base.Base):
         snapshot_object = self.create_snapshot_in_db(context, volume, name,
                                                      description, True,
                                                      metadata, None,
-                                                     commit_quota=False)
+                                                     commit_quota=True)
         self.volume_rpcapi.manage_existing_snapshot(
             context, snapshot_object, ref, service.service_topic_queue)
         return snapshot_object
diff --git a/cinder/volume/flows/api/create_volume.py b/cinder/volume/flows/api/create_volume.py
index 0460a0a9e69..7b261180118 100644
--- a/cinder/volume/flows/api/create_volume.py
+++ b/cinder/volume/flows/api/create_volume.py
@@ -631,6 +631,9 @@ class QuotaReserveTask(flow_utils.CinderTask):
                 reserve_opts = {'volumes': 1}
             else:
                 reserve_opts = {'volumes': 1, 'gigabytes': size}
+            if ('update_size' in optional_args
+                    and optional_args['update_size']):
+                reserve_opts.pop('volumes', None)
             QUOTAS.add_volume_type_opts(context, reserve_opts, volume_type_id)
             reservations = QUOTAS.reserve(context, **reserve_opts)
             return {
diff --git a/cinder/volume/flows/api/manage_existing.py b/cinder/volume/flows/api/manage_existing.py
index 9c7cb543770..e180b2ec5cf 100644
--- a/cinder/volume/flows/api/manage_existing.py
+++ b/cinder/volume/flows/api/manage_existing.py
@@ -20,6 +20,7 @@ from cinder import exception
 from cinder import flow_utils
 from cinder import objects
 from cinder.objects import fields
+from cinder.volume.flows.api import create_volume as create_api
 from cinder.volume.flows import common
 
 LOG = logging.getLogger(__name__)
@@ -138,7 +139,9 @@ def get_flow(scheduler_rpcapi, db_api, create_what):
 
     # This will cast it out to either the scheduler or volume manager via
     # the rpc apis provided.
-    api_flow.add(EntryCreateTask(db_api),
+    api_flow.add(create_api.QuotaReserveTask(),
+                 EntryCreateTask(db_api),
+                 create_api.QuotaCommitTask(),
                  ManageCastTask(scheduler_rpcapi, db_api))
 
     # Now load (but do not run) the flow using the provided initial data.
diff --git a/cinder/volume/flows/manager/manage_existing.py b/cinder/volume/flows/manager/manage_existing.py
index e6f33940fc2..2ec5330b9e1 100644
--- a/cinder/volume/flows/manager/manage_existing.py
+++ b/cinder/volume/flows/manager/manage_existing.py
@@ -118,7 +118,7 @@ def get_flow(context, db, driver, host, volume, ref):
         'volume': volume,
         'manage_existing_ref': ref,
         'group_snapshot': None,
-        'optional_args': {'is_quota_committed': False},
+        'optional_args': {'is_quota_committed': False, 'update_size': True}
     }
 
     volume_flow.add(create_mgr.NotifyVolumeActionTask(db,
diff --git a/cinder/volume/flows/manager/manage_existing_snapshot.py b/cinder/volume/flows/manager/manage_existing_snapshot.py
index 574fa5037b5..bc951086970 100644
--- a/cinder/volume/flows/manager/manage_existing_snapshot.py
+++ b/cinder/volume/flows/manager/manage_existing_snapshot.py
@@ -147,7 +147,15 @@ class QuotaReserveTask(flow_utils.CinderTask):
             if CONF.no_snapshot_gb_quota:
                 reserve_opts = {'snapshots': 1}
             else:
-                reserve_opts = {'snapshots': 1, 'gigabytes': size}
+                # NOTE(tommylikehu): We only use the difference of size here
+                # as we already committed the original size at the API
+                # service before and this reservation task is only used for
+                # managing snapshots now.
+                reserve_opts = {'snapshots': 1,
+                                'gigabytes':
+                                    int(size) - snapshot_ref.volume_size}
+            if 'update_size' in optional_args and optional_args['update_size']:
+                reserve_opts.pop('snapshots', None)
             volume = objects.Volume.get_by_id(context, snapshot_ref.volume_id)
             QUOTAS.add_volume_type_opts(context,
                                         reserve_opts,
@@ -322,7 +330,7 @@ def get_flow(context, db, driver, host, snapshot_id, ref):
         'context': context,
         'snapshot_id': snapshot_id,
         'manage_existing_ref': ref,
-        'optional_args': {'is_quota_committed': False}
+        'optional_args': {'is_quota_committed': False, 'update_size': True}
     }
 
     notify_start_msg = "manage_existing_snapshot.start"
diff --git a/releasenotes/notes/bug-1587376-fix-manage-resource-quota-issue-78f59f39b9fa4762.yaml b/releasenotes/notes/bug-1587376-fix-manage-resource-quota-issue-78f59f39b9fa4762.yaml
new file mode 100644
index 00000000000..beacc152792
--- /dev/null
+++ b/releasenotes/notes/bug-1587376-fix-manage-resource-quota-issue-78f59f39b9fa4762.yaml
@@ -0,0 +1,4 @@
+---
+fixes:
+  - Fix the bug that Cinder would commit quota twice in a clean environment
+    when managing volume and snapshot resource (Bug #1587376).