diff --git a/octodns/manager.py b/octodns/manager.py
index 4305451..6ad44c1 100644
--- a/octodns/manager.py
+++ b/octodns/manager.py
@@ -703,24 +703,35 @@ class Manager(object):
         return sources
 
     def sync(
-        self, dry_run=True, force=False, plan_output_fh=stdout, checksum=None
+        self,
+        dry_run=True,
+        force=False,
+        plan_output_fh=stdout,
+        checksum=None,
+        eligible_zones=None,
     ):
         self.log.info(
-            'sync: dry_run=%s, force=%s, plan_output_fh=%s, checksum=%s',
+            'sync: dry_run=%s, force=%s, plan_output_fh=%s, checksum=%s, eligible_zones=%s',
             dry_run,
             force,
             getattr(plan_output_fh, 'name', plan_output_fh.__class__.__name__),
             checksum,
+            eligible_zones,
         )
 
         zones = self.zones
 
+        if eligible_zones:
+            zones = IdnaDict({n: zones.get(n) for n in eligible_zones})
+
         includes_arpa = any(e.endswith('arpa.') for e in zones.keys())
         if self.auto_arpa and includes_arpa:
             # it's not safe to mess with auto_arpa when we don't have a complete
             # picture of records, so if any filtering is happening while arpa
-            # zones are in play we need to abort
-            if any(e.endswith('arpa.') for e in (self.active_zones or [])):
+            # zones are in play we need to abort. If eligible or active are present
+            # we are filtering.
+            target_zones = eligible_zones or self.active_zones or []
+            if any(e.endswith('arpa.') for e in target_zones):
                 raise ManagerException(
                     'ARPA zones cannot be synced during partial runs when auto_arpa is enabled'
                 )
diff --git a/tests/test_octodns_manager.py b/tests/test_octodns_manager.py
index 0fe7248..15cee12 100644
--- a/tests/test_octodns_manager.py
+++ b/tests/test_octodns_manager.py
@@ -311,6 +311,15 @@ class TestManager(TestCase):
             ).sync()
             self.assertEqual(0, tc)
 
+    def test_eligible_zones(self):
+        with TemporaryDirectory() as tmpdir:
+            environ['YAML_TMP_DIR'] = tmpdir.dirname
+            environ['YAML_TMP_DIR2'] = tmpdir.dirname
+            # Test eligible_zones parameter for runtime filtering
+            manager = Manager(get_config_filename('simple.yaml'))
+            tc = manager.sync(eligible_zones=['unit.tests.'], dry_run=False)
+            self.assertEqual(22, tc)
+
     def test_aliases(self):
         with TemporaryDirectory() as tmpdir:
             environ['YAML_TMP_DIR'] = tmpdir.dirname
@@ -1094,6 +1103,25 @@ class TestManager(TestCase):
                 str(ctx.exception),
             )
 
+            # same for eligible_zones parameter
+            reset(tmpdir.dirname)
+            manager.active_zones = None
+            manager.active_sources = None
+            manager.active_targets = None
+            manager._zones = None
+            tc = manager.sync(eligible_zones=['unit.tests.'], dry_run=False)
+            self.assertEqual(22, tc)
+            # can't do partial syncs that include arpa zones via eligible_zones
+            with self.assertRaises(ManagerException) as ctx:
+                manager.sync(
+                    eligible_zones=['unit.tests.', '3.2.2.in-addr.arpa.'],
+                    dry_run=False,
+                )
+            self.assertEqual(
+                'ARPA zones cannot be synced during partial runs when auto_arpa is enabled',
+                str(ctx.exception),
+            )
+
             # full sync with arpa is fine, 2 extra records from it
             reset(tmpdir.dirname)
             manager.active_zones = None