diff --git a/cinder/scheduler/filters/capacity_filter.py b/cinder/scheduler/filters/capacity_filter.py
index 3349fe397c..984a5bd81f 100644
--- a/cinder/scheduler/filters/capacity_filter.py
+++ b/cinder/scheduler/filters/capacity_filter.py
@@ -25,6 +25,11 @@
 
 LOG = logging.getLogger(__name__)
 
+_SAME_AGGREGATE_MIGRATION_OPERATIONS = (
+    'migrate_volume',
+    'find_backend_for_connector',
+)
+
 
 class CapacityFilter(filters.BaseBackendFilter):
     """Capacity filters based on volume backend's capacity utilization."""
@@ -45,7 +50,8 @@ def backend_passes(self, backend_state, filter_properties):
         source_agg_id = filter_properties.get('source_aggregate_id')
         if source_agg_id:
             spec = filter_properties.get('request_spec', {})
-            if spec.get('operation') == 'migrate_volume':
+            if (spec.get('operation') in
+                    _SAME_AGGREGATE_MIGRATION_OPERATIONS):
                 dest_agg_id = None
                 if backend_state.capabilities:
                     dest_agg_id = backend_state.capabilities.get(
diff --git a/cinder/scheduler/manager.py b/cinder/scheduler/manager.py
index fd12fc87fd..4613a62485 100644
--- a/cinder/scheduler/manager.py
+++ b/cinder/scheduler/manager.py
@@ -447,6 +447,9 @@ def _retype_volume_set_error(self, context, ex, request_spec,
     def find_backend_for_connector(self, context, connector, request_spec,
                                    volume_size, filter_properties=None):
         self._wait_for_scheduler()
+        volume = request_spec.get('volume_properties')
+        if volume:
+            self._set_source_aggregate_id(volume, filter_properties)
         backend = self.driver.find_backend_for_connector(context,
                                                          connector,
                                                          request_spec,
diff --git a/cinder/tests/unit/scheduler/test_host_filters.py b/cinder/tests/unit/scheduler/test_host_filters.py
index 3f2881d921..41ad8c01db 100644
--- a/cinder/tests/unit/scheduler/test_host_filters.py
+++ b/cinder/tests/unit/scheduler/test_host_filters.py
@@ -594,7 +594,9 @@ def test_filter_provisioning_type(self, _mock_serv_is_up, volume_type):
                                        'service': service})
         self.assertTrue(filt_cls.backend_passes(host, filter_properties))
 
-    def test_filter_passes_same_aggregate_migration(self, _mock_serv_is_up):
+    @ddt.data('migrate_volume', 'find_backend_for_connector')
+    def test_filter_passes_same_aggregate_migration(
+            self, operation, _mock_serv_is_up):
         """Cross-vcenter migration with same aggregate_id should pass.
 
         When migrating a volume between vcenters where both pools share the
@@ -609,7 +611,7 @@ def test_filter_passes_same_aggregate_migration(self, _mock_serv_is_up):
             'size': 2048,
             'request_spec': {
                 'volume_id': fake.VOLUME_ID,
-                'operation': 'migrate_volume',
+                'operation': operation,
                 'volume_properties': {
                     'host': 'vc-a-0@vmware_fcd#pool_A',
                 },
@@ -657,6 +659,28 @@ def test_filter_fails_different_aggregate_migration(
                                            'aggregate_id': 'agg_456'}})
         self.assertFalse(filt_cls.backend_passes(host, filter_properties))
 
+    def test_filter_fails_same_aggregate_non_migration_operation(
+            self, _mock_serv_is_up):
+        _mock_serv_is_up.return_value = True
+        filt_cls = self.class_map['CapacityFilter']()
+        filter_properties = {
+            'size': 2048,
+            'request_spec': {
+                'volume_id': fake.VOLUME_ID,
+                'operation': 'create_volume',
+            },
+            'source_aggregate_id': 'agg_123',
+        }
+        service = {'disabled': False}
+        host = fakes.FakeBackendState('vc-b-0@vmware_fcd#pool_B',
+                                      {'total_capacity_gb': 5000,
+                                       'free_capacity_gb': 100,
+                                       'updated_at': None,
+                                       'service': service,
+                                       'capabilities': {
+                                           'aggregate_id': 'agg_123'}})
+        self.assertFalse(filt_cls.backend_passes(host, filter_properties))
+
     def test_filter_passes_migration_no_aggregate_id(self, _mock_serv_is_up):
         """Migration without aggregate_id should use normal capacity checks.
 
diff --git a/cinder/tests/unit/scheduler/test_scheduler.py b/cinder/tests/unit/scheduler/test_scheduler.py
index 0854dcf4e8..5ae2dfb327 100644
--- a/cinder/tests/unit/scheduler/test_scheduler.py
+++ b/cinder/tests/unit/scheduler/test_scheduler.py
@@ -756,7 +756,7 @@ def test_get_az(self):
     @mock.patch('cinder.scheduler.driver.Scheduler.find_backend_for_connector')
     def test_find_backend_for_connector(self, _mock_find_backend_for_conector):
         connector = mock.Mock()
-        request_spec = mock.Mock()
+        request_spec = {}
         volume_size = mock.Mock()
         backend_ret = mock.Mock(host='fake-host',
                                 cluster_name='fake-cluster', capabilities=[])
@@ -772,6 +772,30 @@ def test_find_backend_for_connector(self, _mock_find_backend_for_conector):
             'capabilities': backend_ret.capabilities
         })
 
+    @mock.patch('cinder.scheduler.driver.Scheduler.find_backend_for_connector')
+    def test_find_backend_for_connector_sets_source_aggregate_id(
+            self, _mock_find_backend_for_connector):
+        volume = fake_volume.fake_volume_obj(
+            self.context, host='vc-a-0@vmware_fcd#pool_A')
+        pool_state = mock.Mock(capabilities={'aggregate_id': 'agg_123'})
+        backend_state = mock.Mock(pools={'pool_A': pool_state})
+        self.manager.driver.host_manager.backend_state_map = {
+            'vc-a-0@vmware_fcd': backend_state}
+        backend = mock.Mock(host='fake-host', cluster_name=None,
+                            capabilities={})
+        _mock_find_backend_for_connector.return_value = backend
+        connector = mock.Mock()
+        request_spec = {'volume_properties': volume}
+        filter_properties = {}
+
+        self.manager.find_backend_for_connector(
+            self.context, connector, request_spec, 1, filter_properties)
+
+        self.assertEqual('agg_123',
+                         filter_properties.get('source_aggregate_id'))
+        _mock_find_backend_for_connector.assert_called_once_with(
+            self.context, connector, request_spec, filter_properties)
+
 
 class SchedulerTestCase(test.TestCase):
     """Test case for base scheduler driver class."""