diff --git a/awx/main/tasks.py b/awx/main/tasks.py
index 32a953e968..0f12c4af12 100644
--- a/awx/main/tasks.py
+++ b/awx/main/tasks.py
@@ -229,8 +229,11 @@ def apply_cluster_membership_policies():
         actual_instances = [Node(obj=i, groups=[]) for i in all_instances if i.managed_by_policy]
         logger.debug("Total instances: {}, available for policy: {}".format(total_instances, len(actual_instances)))
         for g in sorted(actual_groups, key=lambda x: len(x.instances)):
+            exclude_type = 'execution' if g.obj.name == settings.DEFAULT_CONTROL_PLANE_QUEUE_NAME else 'control'
             policy_min_added = []
             for i in sorted(actual_instances, key=lambda x: len(x.groups)):
+                if i.obj.node_type == exclude_type:
+                    continue  # never place execution instances in controlplane group or control instances in other groups
                 if len(g.instances) >= g.obj.policy_instance_minimum:
                     break
                 if i.obj.id in g.instances:
@@ -245,13 +248,19 @@ def apply_cluster_membership_policies():
 
         # Finally, process instance policy percentages
         for g in sorted(actual_groups, key=lambda x: len(x.instances)):
+            exclude_type = 'execution' if g.obj.name == settings.DEFAULT_CONTROL_PLANE_QUEUE_NAME else 'control'
+            candidate_pool_ct = len([i for i in actual_instances if i.obj.node_type != exclude_type])
+            if not candidate_pool_ct:
+                continue
             policy_per_added = []
             for i in sorted(actual_instances, key=lambda x: len(x.groups)):
+                if i.obj.node_type == exclude_type:
+                    continue
                 if i.obj.id in g.instances:
                     # If the instance is already _in_ the group, it was
                     # applied earlier via a minimum policy or policy list
                     continue
-                if 100 * float(len(g.instances)) / len(actual_instances) >= g.obj.policy_instance_percentage:
+                if 100 * float(len(g.instances)) / candidate_pool_ct >= g.obj.policy_instance_percentage:
                     break
                 g.instances.append(i.obj.id)
                 i.groups.append(g.obj.id)
diff --git a/awx/main/tests/functional/test_instances.py b/awx/main/tests/functional/test_instances.py
index 0c4d73a296..efb9904d2d 100644
--- a/awx/main/tests/functional/test_instances.py
+++ b/awx/main/tests/functional/test_instances.py
@@ -243,6 +243,41 @@ def test_policy_instance_list_explicitly_pinned(instance_factory, instance_group
     assert set(ig_2.instances.all()) == set([i2])
 
 
+@pytest.mark.django_db
+def test_control_plane_policy_exception(controlplane_instance_group):
+    controlplane_instance_group.policy_instance_percentage = 100
+    controlplane_instance_group.policy_instance_minimum = 2
+    controlplane_instance_group.save()
+    Instance.objects.create(hostname='foo-1', node_type='execution')
+    apply_cluster_membership_policies()
+    assert 'foo-1' not in [inst.hostname for inst in controlplane_instance_group.instances.all()]
+
+
+@pytest.mark.django_db
+def test_normal_instance_group_policy_exception():
+    ig = InstanceGroup.objects.create(name='bar', policy_instance_percentage=100, policy_instance_minimum=2)
+    Instance.objects.create(hostname='foo-1', node_type='control')
+    apply_cluster_membership_policies()
+    assert 'foo-1' not in [inst.hostname for inst in ig.instances.all()]
+
+
+@pytest.mark.django_db
+def test_percentage_as_fraction_of_execution_nodes():
+    """
+    If an instance requests 50 percent of instances, then those should be 50 percent
+    of available execution nodes (1 out of 2), as opposed to 50 percent
+    of all available nodes (2 out of 4) which include unusable control nodes
+    """
+    ig = InstanceGroup.objects.create(name='bar', policy_instance_percentage=50)
+    for i in range(2):
+        Instance.objects.create(hostname=f'foo-{i}', node_type='control')
+    for i in range(2):
+        Instance.objects.create(hostname=f'bar-{i}', node_type='execution')
+    apply_cluster_membership_policies()
+    assert ig.instances.count() == 1
+    assert ig.instances.first().hostname.startswith('bar-')
+
+
 @pytest.mark.django_db
 def test_basic_instance_group_membership(instance_group_factory, default_instance_group, job_factory):
     j = job_factory()