diff --git a/octodns/manager.py b/octodns/manager.py
index 2197d3b..f3ee56d 100644
--- a/octodns/manager.py
+++ b/octodns/manager.py
@@ -98,11 +98,21 @@ class Manager(object):
         plan = p[1]
         return len(plan.changes[0].record.zone.name) if plan.changes else 0
 
-    def __init__(self, config_file, max_workers=None, include_meta=False):
+    def __init__(
+        self,
+        config_file,
+        max_workers=None,
+        include_meta=False,
+        delay_arpa=False,
+    ):
         version = self._try_version('octodns', version=__VERSION__)
         self.log.info(
-            '__init__: config_file=%s (octoDNS %s)', config_file, version
+            '__init__: config_file=%s, delay_arpa=%s (octoDNS %s)',
+            config_file,
+            version,
+            delay_arpa,
         )
+        self.delay_arpa = delay_arpa
 
         self._configured_sub_zones = None
 
@@ -384,7 +394,6 @@ class Manager(object):
         desired=None,
         lenient=False,
     ):
-
         zone = self.get_zone(zone_name)
         self.log.debug(
             'sync:   populating, zone=%s, lenient=%s',
@@ -470,11 +479,21 @@ class Manager(object):
             getattr(plan_output_fh, 'name', plan_output_fh.__class__.__name__),
         )
 
+        if (
+            self.delay_arpa
+            and eligible_zones
+            and any(e.endswith('arpa.') for e in eligible_zones)
+        ):
+            raise ManagerException(
+                'ARPA zones cannot be synced during partial runs when delay_arpa is enabled'
+            )
+
         zones = self.config['zones']
         if eligible_zones:
             zones = IdnaDict({n: zones.get(n) for n in eligible_zones})
 
         aliased_zones = {}
+        delayed_arpa = []
         futures = []
         for zone_name, config in zones.items():
             decoded_zone_name = idna_decode(zone_name)
@@ -571,16 +590,20 @@ class Manager(object):
                     f'Zone {decoded_zone_name}, unknown ' f'target: {target}'
                 )
 
-            futures.append(
-                self._executor.submit(
-                    self._populate_and_plan,
-                    zone_name,
-                    processors,
-                    sources,
-                    targets,
-                    lenient=lenient,
+            kwargs = {
+                'zone_name': zone_name,
+                'processors': processors,
+                'sources': sources,
+                'targets': targets,
+                'lenient': lenient,
+            }
+
+            if self.delay_arpa and zone_name.endswith('arpa.'):
+                delayed_arpa.append(kwargs)
+            else:
+                futures.append(
+                    self._executor.submit(self._populate_and_plan, **kwargs)
                 )
-            )
 
         # Wait on all results and unpack/flatten the plans and store the
         # desired states in case we need them below
@@ -620,6 +643,20 @@ class Manager(object):
         # as these are aliased zones
         plans += [p for f in futures for p in f.result()[0]]
 
+        if delayed_arpa:
+            # if delaying arpa all of the non-arpa zones have been processed now
+            # so it's time to plan them
+            self.log.info(
+                'sync: processing %d delayed arpa zones', len(delayed_arpa)
+            )
+            # populate and plan them
+            futures = [
+                self._executor.submit(self._populate_and_plan, **kwargs)
+                for kwargs in delayed_arpa
+            ]
+            # wait on the results and unpack/flatten the plans
+            plans += [p for f in futures for p in f.result()[0]]
+
         # Best effort sort plans children first so that we create/update
         # children zones before parents which should allow us to more safely
         # extract things into sub-zones. Combining a child back into a parent
diff --git a/tests/config/3.2.2.in-addr.arpa.yaml b/tests/config/3.2.2.in-addr.arpa.yaml
new file mode 100644
index 0000000..a793f56
--- /dev/null
+++ b/tests/config/3.2.2.in-addr.arpa.yaml
@@ -0,0 +1,7 @@
+---
+4:
+  type: PTR
+  value: unit.tests.
+5:
+  type: PTR
+  value: unit.tests.
diff --git a/tests/config/simple-arpa.yaml b/tests/config/simple-arpa.yaml
new file mode 100644
index 0000000..b7cf38b
--- /dev/null
+++ b/tests/config/simple-arpa.yaml
@@ -0,0 +1,27 @@
+manager:
+  max_workers: 2
+  delayed_arpa: true
+
+providers:
+  in:
+    class: octodns.provider.yaml.YamlProvider
+    directory: tests/config
+    supports_root_ns: False
+    strict_supports: False
+  dump:
+    class: octodns.provider.yaml.YamlProvider
+    directory: env/YAML_TMP_DIR
+    default_ttl: 999
+    supports_root_ns: False
+    strict_supports: False
+zones:
+  unit.tests.:
+    sources:
+    - in
+    targets:
+    - dump
+  3.2.2.in-addr.arpa.:
+    sources:
+    - in
+    targets:
+    - dump
diff --git a/tests/test_octodns_manager.py b/tests/test_octodns_manager.py
index 3df325d..f3b792f 100644
--- a/tests/test_octodns_manager.py
+++ b/tests/test_octodns_manager.py
@@ -908,6 +908,33 @@ class TestManager(TestCase):
             str(ctx.exception),
         )
 
+    def test_delayed_arpa(self):
+        manager = Manager(
+            get_config_filename('simple-arpa.yaml'), delay_arpa=True
+        )
+
+        with TemporaryDirectory() as tmpdir:
+            environ['YAML_TMP_DIR'] = tmpdir.dirname
+
+            # we can sync eligible_zones so long as they're not arpa
+            tc = manager.sync(dry_run=False, eligible_zones=['unit.tests.'])
+            self.assertEqual(22, tc)
+
+            # can't do partial syncs that include arpa zones
+            with self.assertRaises(ManagerException) as ctx:
+                manager.sync(
+                    dry_run=False,
+                    eligible_zones=['unit.tests.', '3.2.2.in-addr.arpa.'],
+                )
+            self.assertEqual(
+                'ARPA zones cannot be synced during partial runs when delay_arpa is enabled',
+                str(ctx.exception),
+            )
+
+            # full sync with arpa is fine, 2 extra records from it
+            tc = manager.sync(dry_run=False)
+            self.assertEqual(24, tc)
+
 
 class TestMainThreadExecutor(TestCase):
     def test_success(self):