diff --git a/cinder/db/api.py b/cinder/db/api.py
index 4e378a6206e..56eb69b9ebc 100644
--- a/cinder/db/api.py
+++ b/cinder/db/api.py
@@ -281,10 +281,10 @@ def volume_get_all(context, marker=None, limit=None, sort_keys=None,
                                offset=offset)
 
 
-def get_host_by_volume_metadata(key, value, filters=None):
+def get_hosts_by_volume_metadata(key, value, filters=None):
     """Returns the host with the most volumes matching volume metadata."""
-    return IMPL.get_host_by_volume_metadata(key, value,
-                                            filters=filters)
+    return IMPL.get_hosts_by_volume_metadata(key, value,
+                                             filters=filters)
 
 
 def calculate_resource_count(context, resource_type, filters):
diff --git a/cinder/db/sqlalchemy/api.py b/cinder/db/sqlalchemy/api.py
index a88d7ee88be..7ecee7e4e3a 100644
--- a/cinder/db/sqlalchemy/api.py
+++ b/cinder/db/sqlalchemy/api.py
@@ -2176,7 +2176,7 @@ def volume_get_all(context, marker=None, limit=None, sort_keys=None,
         return query.all()
 
 
-def get_host_by_volume_metadata(meta_key, meta_value, filters=None):
+def get_hosts_by_volume_metadata(meta_key, meta_value, filters=None):
     session = get_session()
     count_label = func.count().label("n")
     query = session.query(
@@ -2199,13 +2199,10 @@ def get_host_by_volume_metadata(meta_key, meta_value, filters=None):
                 models.Volume.availability_zone == az)
 
     query = query.group_by("h")\
-        .order_by(desc(count_label)).limit(1)
+        .order_by(desc(count_label))
 
     with session.begin():
-        result = query.first()
-        if result:
-            return result[0]
-    return None
+        return query.all()
 
 
 @require_context
diff --git a/cinder/scheduler/filters/shard_filter.py b/cinder/scheduler/filters/shard_filter.py
index 61903bbf47e..87bc0810da7 100644
--- a/cinder/scheduler/filters/shard_filter.py
+++ b/cinder/scheduler/filters/shard_filter.py
@@ -175,24 +175,39 @@ def _filter_by_k8s_cluster(self, backends, filter_properties):
         if not cluster_name:
             return backends
 
-        props = spec.get('resource_properties', {})
-        availability_zone = props.get('availability_zone')
+        availability_zone = filter_properties.get('availability_zone')
         query_filters = None
         if availability_zone:
             query_filters = {'availability_zone': availability_zone}
 
-        k8s_host = db.get_host_by_volume_metadata(
+        results = db.get_hosts_by_volume_metadata(
             key=CSI_CLUSTER_METADATA_KEY,
             value=cluster_name,
             filters=query_filters)
 
-        if not k8s_host:
+        if not results:
             return backends
 
+        # Allowing new volumes to be created only in the dominant shard
+        if spec.get('operation') == 'create_volume':
+            results = results[:1]
+
+        k8s_hosts = dict(results)
+
+        def _is_k8s_host(b):
+            host = volume_utils.extract_host(b.host, 'host')
+            if host in k8s_hosts:
+                return True
+            else:
+                LOG.debug('%(backend)s not in the allowed '
+                          'K8S hosts %(k8s_hosts)s.',
+                          {'backend': b,
+                           'k8s_hosts': k8s_hosts})
+                return False
+
         return [
             b for b in backends if
-            (not self._is_vmware(b) or
-             volume_utils.extract_host(b.host, 'host') == k8s_host)
+            (not self._is_vmware(b) or _is_k8s_host(b))
         ]
 
     def _backend_passes(self, backend_state, filter_properties):
diff --git a/cinder/tests/unit/scheduler/test_shard_filter.py b/cinder/tests/unit/scheduler/test_shard_filter.py
index 34cc79c3bc4..f2a9ed584ea 100644
--- a/cinder/tests/unit/scheduler/test_shard_filter.py
+++ b/cinder/tests/unit/scheduler/test_shard_filter.py
@@ -228,7 +228,7 @@ def test_noop_for_find_backend_by_connector_without_hint(self):
         self.backend_passes(host, self.props)
 
     @mock.patch('cinder.context.get_admin_context')
-    @mock.patch('cinder.db.get_host_by_volume_metadata')
+    @mock.patch('cinder.db.get_hosts_by_volume_metadata')
     def test_same_shard_for_k8s_volumes(self, mock_get_hosts,
                                         mock_get_context):
         CSI_KEY = 'cinder.csi.openstack.org/cluster'
@@ -246,7 +246,8 @@ def test_same_shard_for_k8s_volumes(self, mock_get_hosts,
         fake_meta = {
             CSI_KEY: 'cluster-1',
         }
-        mock_get_hosts.return_value = 'volume-vc-a-1'
+        mock_get_hosts.return_value = [('volume-vc-x-1', 2),
+                                       ('volume-vc-a-1', 1)]
         self.filt_cls._PROJECT_SHARD_CACHE['baz'] = ['sharding_enabled',
                                                      'vc-a-1']
         filter_props = dict(self.props)
@@ -254,9 +255,7 @@ def test_same_shard_for_k8s_volumes(self, mock_get_hosts,
             'project_id': 'baz',
             'metadata': fake_meta
         })
-        filter_props['request_spec']['resource_properties'] = {
-            'availability_zone': 'az-1'
-        }
+        filter_props['availability_zone'] = 'az-1'
 
         filtered = self.filt_cls.filter_all(all_backends, filter_props)