Repository: kubernetes-sigs/controller-runtime

Issue:
Context cancellation causes a goroutine leak (possibly intermittent)
### Problem statement

When running our integration tests suite with https://github.com/uber-go/goleak, it sometimes reports a goroutine leak in:

```
 manager_envtest_test.go:84: found unexpected goroutines:
          [Goroutine 12886 in state sync.WaitGroup.Wait, with sync.runtime_SemacquireWaitGroup on top of the stack:
          sync.runtime_SemacquireWaitGroup(0xc000cdc830?)
          	/home/runner/go/pkg/mod/golang.org/toolchain@v0.0.1-go1.24.1.linux-amd64/src/runtime/sema.go:110 +0x25
          sync.(*WaitGroup).Wait(0xc000cdc830)
          	/home/runner/go/pkg/mod/golang.org/toolchain@v0.0.1-go1.24.1.linux-amd64/src/sync/waitgroup.go:118 +0x89
          sigs.k8s.io/controller-runtime/pkg/manager.(*controllerManager).engageStopProcedure.func3.(*runnableGroup).StopAndWait.3.2()
          	/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.20.4/pkg/manager/runnable_group.go:307 +0x99
          created by sigs.k8s.io/controller-runtime/pkg/manager.(*controllerManager).engageStopProcedure.func3.(*runnableGroup).StopAndWait.3 in goroutine 12666
          	/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.20.4/pkg/manager/runnable_group.go:304 +0x213
           Goroutine 12791 in state chan send, with sigs.k8s.io/controller-runtime/pkg/manager.(*runnableGroup).reconcile.func1 on top of the stack:
          sigs.k8s.io/controller-runtime/pkg/manager.(*runnableGroup).reconcile.func1(0xc00115d940)
          	/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.20.4/pkg/manager/runnable_group.go:227 +0x1f8
          created by sigs.k8s.io/controller-runtime/pkg/manager.(*runnableGroup).reconcile in goroutine 12615
          	/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.20.4/pkg/manager/runnable_group.go:210 +0x245
           Goroutine 12838 in state chan send, with sigs.k8s.io/controller-runtime/pkg/manager.(*runnableGroup).reconcile.func1 on top of the stack:
          sigs.k8s.io/controller-runtime/pkg/manager.(*runnableGroup).reconcile.func1(0xc00115dd00)
          	/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.20.4/pkg/manager/runnable_group.go:227 +0x1f8
          created by sigs.k8s.io/controller-runtime/pkg/manager.(*runnableGroup).reconcile in goroutine 12615
          	/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.20.4/pkg/manager/runnable_group.go:210 +0x245
          ]
```

After quick glance at the code, it would seem that `controllerManager` passes its `errChan` to its `runnableGroup`s.

When runnables fail, they propagate the error through that error channel in https://github.com/kubernetes-sigs/controller-runtime/blob/dcb8a6a8d7ed700d4590f11fac171f766f54ea3c/pkg/manager/runnable_group.go#L227.

If I'm looking at this correctly, when context is done, the `errChan` is not drained in https://github.com/kubernetes-sigs/controller-runtime/blob/5af1f3ebd472b62a4a708ad3aa2d252489b91b27/pkg/manager/internal.go#L467-L469 and thus runnable groups cannot send their errors in https://github.com/kubernetes-sigs/controller-runtime/blob/dcb8a6a8d7ed700d4590f11fac171f766f54ea3c/pkg/manager/runnable_group.go#L224.

### Proposed solution

Drain the `cm.errChan` in `case <-ctx.Done()` branch at https://github.com/kubernetes-sigs/controller-runtime/blob/5af1f3ebd472b62a4a708ad3aa2d252489b91b27/pkg/manager/internal.go#L466-L473

Discussion:
@jingyih: Hi @pmalek, are you still seeing this issue? Which controller-runtime version are you using? I did some digging into the codebase. And it looks to me that the errChan is properly drained here:
https://github.com/kubernetes-sigs/controller-runtime/blob/b9a9ca01fd37a66225eb4793544349ee2ed1f196/pkg/manager/internal.go#L510@pmalek: I've moved on from this problem and added an exclusion for controller-runtime goroutines in the goroutine leak test: https://github.com/Kong/kubernetes-ingress-controller/blob/a32efd26867e255df611aa156ed9fcda291449f2/test/envtest/manager_envtest_test.go#L84-L87.

As you can see in the snippet that I've added in the description, at that time I was using v0.20.4.

Just a thought:

Would it be possible to receive errors on `cm.errChan` after `stopComplete` gets closed and returns in https://github.com/kubernetes-sigs/controller-runtime/blob/b9a9ca01fd37a66225eb4793544349ee2ed1f196/pkg/manager/internal.go#L514? @jingyih: good question

> I've moved on from this problem and added an exclusion for controller-runtime goroutines in the goroutine leak test: https://github.com/Kong/kubernetes-ingress-controller/blob/a32efd26867e255df611aa156ed9fcda291449f2/test/envtest/manager_envtest_test.go#L84-L87.
> 
> As you can see in the snippet that I've added in the description, at that time I was using v0.20.4.
> 
> Just a thought:
> 
> Would it be possible to receive errors on `cm.errChan` after `stopComplete` gets closed and returns in
> 
> [controller-runtime/pkg/manager/internal.go](https://github.com/kubernetes-sigs/controller-runtime/blob/b9a9ca01fd37a66225eb4793544349ee2ed1f196/pkg/manager/internal.go#L514)
> 
> Line 514 in [b9a9ca0](/kubernetes-sigs/controller-runtime/commit/b9a9ca01fd37a66225eb4793544349ee2ed1f196)
> 
>  case <-stopComplete: 
> ?

Good pointer. I think it is possible that `stopComplete` closes before all runnables stop. After taking a closer look, the race depends on the `GracefulShutdownTimeout`—which defaults to 30 seconds—and how long the runnable takes to return an error after being signaled to stop. Did you set a custom value for `GracefulShutdownTimeout`?

https://github.com/kubernetes-sigs/controller-runtime/blob/cacd627d0b9bc46b592cf5f96f57f1308313618e/pkg/manager/manager.go#L270@pmalek: We do set it to 0 for the tests where we observed the error: https://github.com/Kong/kubernetes-ingress-controller/blob/dd6538bf66f2e27805786fd6a21a7a93475f0323/test/envtest/run.go#L56-L59

Solution patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 5f5b2b66..47bf6eed 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -4,8 +4,10 @@ aliases:
   # active folks who can be contacted to perform admin-related
   # tasks on the repo, or otherwise approve any PRS.
   controller-runtime-admins:
-  - vincepri
+  - alvaroaleman
   - joelanford
+  - sbueringer
+  - vincepri
 
   # non-admin folks who have write-access and can approve any PRs in the repo
   controller-runtime-maintainers:
diff --git a/examples/scratch-env/go.mod b/examples/scratch-env/go.mod
index 51f9631d..25d8edae 100644
--- a/examples/scratch-env/go.mod
+++ b/examples/scratch-env/go.mod
@@ -13,7 +13,7 @@ require (
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
+	github.com/emicklei/go-restful/v3 v3.12.2 // indirect
 	github.com/evanphx/json-patch/v5 v5.9.11 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.8.0 // indirect
@@ -24,7 +24,7 @@ require (
 	github.com/go-openapi/swag v0.23.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/btree v1.1.3 // indirect
-	github.com/google/gnostic-models v0.6.9 // indirect
+	github.com/google/gnostic-models v0.7.0 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
@@ -41,6 +41,8 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
+	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/oauth2 v0.27.0 // indirect
 	golang.org/x/sync v0.12.0 // indirect
@@ -53,17 +55,17 @@ require (
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/api v0.34.0-alpha.2 // indirect
-	k8s.io/apiextensions-apiserver v0.34.0-alpha.2 // indirect
-	k8s.io/apimachinery v0.34.0-alpha.2 // indirect
-	k8s.io/client-go v0.34.0-alpha.2 // indirect
+	k8s.io/api v0.34.0-beta.0 // indirect
+	k8s.io/apiextensions-apiserver v0.34.0-beta.0 // indirect
+	k8s.io/apimachinery v0.34.0-beta.0 // indirect
+	k8s.io/client-go v0.34.0-beta.0 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
-	k8s.io/kube-openapi v0.0.0-20250610211856-8b98d1ed966a // indirect
+	k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b // indirect
 	k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 // indirect
-	sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect
+	sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect
-	sigs.k8s.io/structured-merge-diff/v4 v4.7.0 // indirect
-	sigs.k8s.io/yaml v1.4.0 // indirect
+	sigs.k8s.io/structured-merge-diff/v6 v6.2.0 // indirect
+	sigs.k8s.io/yaml v1.5.0 // indirect
 )
 
 replace sigs.k8s.io/controller-runtime => ../..
diff --git a/examples/scratch-env/go.sum b/examples/scratch-env/go.sum
index d156e98c..c6f1a8fd 100644
--- a/examples/scratch-env/go.sum
+++ b/examples/scratch-env/go.sum
@@ -9,8 +9,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
-github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
+github.com/emicklei/go-restful/v3 v3.12.2 h1:DhwDP0vY3k8ZzE0RunuJy8GhNpPL6zqLkDf9B/a0/xU=
+github.com/emicklei/go-restful/v3 v3.12.2/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/evanphx/json-patch v0.5.2 h1:xVCHIVMUu1wtM/VkR9jVZ45N3FhZfYMMYGorLCR8P3k=
 github.com/evanphx/json-patch v0.5.2/go.mod h1:ZWS5hhDbVDyob71nXKNL0+PWn6ToqBHMikGIFbs31qQ=
 github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjTM0wiaDU=
@@ -37,9 +37,8 @@ github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg=
 github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4=
-github.com/google/gnostic-models v0.6.9 h1:MU/8wDLif2qCXZmzncUQ/BOfxWfthHi63KqpoNbWqVw=
-github.com/google/gnostic-models v0.6.9/go.mod h1:CiWsm0s6BSQd1hRn8/QmxqB6BesYcbSZxsz9b0KuDBw=
-github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo=
+github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
@@ -117,6 +116,10 @@ go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
+go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
+go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
+go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -172,26 +175,25 @@ gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/api v0.34.0-alpha.2 h1:nSjp8/FHPoEn/UEy/TWBKkGA2DwS2JJypsuv1l5kidQ=
-k8s.io/api v0.34.0-alpha.2/go.mod h1:QADmt+iVmFauWh94wDpD53xDNB5cmLTTFECEJlhE5To=
-k8s.io/apiextensions-apiserver v0.34.0-alpha.2 h1:hBrCOLTT0Gavt/kviPfv6b6P+hmoCja7m9ewLzDqw6o=
-k8s.io/apiextensions-apiserver v0.34.0-alpha.2/go.mod h1:Lbus5MTRVmJ7uTzLngou3HMMTCRvnvGQmL5aJoYKOL0=
-k8s.io/apimachinery v0.34.0-alpha.2 h1:kKK+KUnbsRWJObxY3pChe0IYkEWG5L4Qsmr/iTIuFQE=
-k8s.io/apimachinery v0.34.0-alpha.2/go.mod h1:HYUvaFuBNToT/JW+lhc2D9BYgqj6xhI5eORJ7BnvKpM=
-k8s.io/client-go v0.34.0-alpha.2 h1:3TwS7HoRQZ8hQp9Sud4j+oS0O2N++3lyMVJavogALzw=
-k8s.io/client-go v0.34.0-alpha.2/go.mod h1:DWWpv2S99iGzK+cUwhQBe1ipIU9PVeREVjl/XWHlr4Y=
+k8s.io/api v0.34.0-beta.0 h1:Yok13a6+0GTJYv56+lSNlgRjF46TLUVv+gaijkLdkB4=
+k8s.io/api v0.34.0-beta.0/go.mod h1:OOkMaR1spg7vtjCCQSdxgiRD0Bbp2pcTCBxJtEEb4Pk=
+k8s.io/apiextensions-apiserver v0.34.0-beta.0 h1:zzR2yWEBkfYcjfwM42CwrBwtwLWLBJ/VwYpyP2ldNMI=
+k8s.io/apiextensions-apiserver v0.34.0-beta.0/go.mod h1:lCs/ZFdQjQWkNdEa0CS7qvLAMVDCj3fVXwSmKZ7Fn8E=
+k8s.io/apimachinery v0.34.0-beta.0 h1:C6teSJBCx6ArW7122MM9hQqeGW2w/QQ0lB4x4Z4Iftc=
+k8s.io/apimachinery v0.34.0-beta.0/go.mod h1:TP8uyOuDEOnzGpLOdffo8hPnIjVDljZCxCM/fruV+5M=
+k8s.io/client-go v0.34.0-beta.0 h1:snyuzY74xHIp7qxMYul8QRn1WxPNhs281srSl2CX9nY=
+k8s.io/client-go v0.34.0-beta.0/go.mod h1:KNxQTmxlWq7adFeedQLu+qVkIBTZqhghesFk41YWyBQ=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
 k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
-k8s.io/kube-openapi v0.0.0-20250610211856-8b98d1ed966a h1:ZV3Zr+/7s7aVbjNGICQt+ppKWsF1tehxggNfbM7XnG8=
-k8s.io/kube-openapi v0.0.0-20250610211856-8b98d1ed966a/go.mod h1:5jIi+8yX4RIb8wk3XwBo5Pq2ccx4FP10ohkbSKCZoK8=
+k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b h1:MloQ9/bdJyIu9lb1PzujOPolHyvO06MXG5TUIj2mNAA=
+k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b/go.mod h1:UZ2yyWbFTpuhSbFhv24aGNOdoRdJZgsIObGBUaYVsts=
 k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 h1:hwvWFiBzdWw1FhfY1FooPn3kzWuJ8tmbZBHi4zVsl1Y=
 k8s.io/utils v0.0.0-20250604170112-4c0f3b243397/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 h1:/Rv+M11QRah1itp8VhT6HoVx1Ray9eB4DBr+K+/sCJ8=
-sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3/go.mod h1:18nIHnGi6636UCz6m8i4DhaJ65T6EruyzmoQqI2BVDo=
-sigs.k8s.io/randfill v0.0.0-20250304075658-069ef1bbf016/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
+sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 h1:gBQPwqORJ8d8/YNZWEjoZs7npUVDpVXUUOFfW6CgAqE=
+sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
 sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU=
 sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
-sigs.k8s.io/structured-merge-diff/v4 v4.7.0 h1:qPeWmscJcXP0snki5IYF79Z8xrl8ETFxgMd7wez1XkI=
-sigs.k8s.io/structured-merge-diff/v4 v4.7.0/go.mod h1:dDy58f92j70zLsuZVuUX5Wp9vtxXpaZnkPGWeqDfCps=
-sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E=
-sigs.k8s.io/yaml v1.4.0/go.mod h1:Ejl7/uTz7PSA4eKMyQCUTnhZYNmLIl+5c2lQPGR2BPY=
+sigs.k8s.io/structured-merge-diff/v6 v6.2.0 h1:msyqjP8Nyd5sF3QSmJouFSzcBIdwq4ct8d1/7VSBHIQ=
+sigs.k8s.io/structured-merge-diff/v6 v6.2.0/go.mod h1:M3W8sfWvn2HhQDIbGWj3S099YozAsymCo/wrT5ohRUE=
+sigs.k8s.io/yaml v1.5.0 h1:M10b2U7aEUY6hRtU870n2VTPgR5RZiL/I6Lcc2F4NUQ=
+sigs.k8s.io/yaml v1.5.0/go.mod h1:wZs27Rbxoai4C0f8/9urLZtZtF3avA3gKvGyPdDqTO4=
diff --git a/go.mod b/go.mod
index f1630ad4..9c1bfba8 100644
--- a/go.mod
+++ b/go.mod
@@ -22,15 +22,15 @@ require (
 	golang.org/x/sys v0.31.0
 	gomodules.xyz/jsonpatch/v2 v2.4.0
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // Using v4 to match upstream
-	k8s.io/api v0.34.0-alpha.2
-	k8s.io/apiextensions-apiserver v0.34.0-alpha.2
-	k8s.io/apimachinery v0.34.0-alpha.2
-	k8s.io/apiserver v0.34.0-alpha.2
-	k8s.io/client-go v0.34.0-alpha.2
+	k8s.io/api v0.34.0-beta.0
+	k8s.io/apiextensions-apiserver v0.34.0-beta.0
+	k8s.io/apimachinery v0.34.0-beta.0
+	k8s.io/apiserver v0.34.0-beta.0
+	k8s.io/client-go v0.34.0-beta.0
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/utils v0.0.0-20250604170112-4c0f3b243397
-	sigs.k8s.io/structured-merge-diff/v4 v4.7.0
-	sigs.k8s.io/yaml v1.4.0
+	sigs.k8s.io/structured-merge-diff/v6 v6.2.0
+	sigs.k8s.io/yaml v1.5.0
 )
 
 require (
@@ -40,7 +40,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
+	github.com/emicklei/go-restful/v3 v3.12.2 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fxamacker/cbor/v2 v2.8.0 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
@@ -50,7 +50,7 @@ require (
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/cel-go v0.25.0 // indirect
-	github.com/google/gnostic-models v0.6.9 // indirect
+	github.com/google/gnostic-models v0.7.0 // indirect
 	github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect
@@ -80,6 +80,8 @@ require (
 	go.opentelemetry.io/otel/trace v1.35.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
+	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
 	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/oauth2 v0.27.0 // indirect
@@ -93,9 +95,9 @@ require (
 	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/component-base v0.34.0-alpha.2 // indirect
-	k8s.io/kube-openapi v0.0.0-20250610211856-8b98d1ed966a // indirect
+	k8s.io/component-base v0.34.0-beta.0 // indirect
+	k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b // indirect
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 // indirect
-	sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect
+	sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect
 )
diff --git a/go.sum b/go.sum
index 433ff6e0..b210c7d9 100644
--- a/go.sum
+++ b/go.sum
@@ -15,8 +15,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
-github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
+github.com/emicklei/go-restful/v3 v3.12.2 h1:DhwDP0vY3k8ZzE0RunuJy8GhNpPL6zqLkDf9B/a0/xU=
+github.com/emicklei/go-restful/v3 v3.12.2/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/evanphx/json-patch v0.5.2 h1:xVCHIVMUu1wtM/VkR9jVZ45N3FhZfYMMYGorLCR8P3k=
 github.com/evanphx/json-patch v0.5.2/go.mod h1:ZWS5hhDbVDyob71nXKNL0+PWn6ToqBHMikGIFbs31qQ=
 github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjTM0wiaDU=
@@ -52,9 +52,8 @@ github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg=
 github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4=
 github.com/google/cel-go v0.25.0 h1:jsFw9Fhn+3y2kBbltZR4VEz5xKkcIFRPDnuEzAGv5GY=
 github.com/google/cel-go v0.25.0/go.mod h1:hjEb6r5SuOSlhCHmFoLzu8HGCERvIsDAbxDAyNU/MmI=
-github.com/google/gnostic-models v0.6.9 h1:MU/8wDLif2qCXZmzncUQ/BOfxWfthHi63KqpoNbWqVw=
-github.com/google/gnostic-models v0.6.9/go.mod h1:CiWsm0s6BSQd1hRn8/QmxqB6BesYcbSZxsz9b0KuDBw=
-github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo=
+github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
@@ -161,6 +160,10 @@ go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
+go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
+go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
+go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -226,32 +229,31 @@ gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/api v0.34.0-alpha.2 h1:nSjp8/FHPoEn/UEy/TWBKkGA2DwS2JJypsuv1l5kidQ=
-k8s.io/api v0.34.0-alpha.2/go.mod h1:QADmt+iVmFauWh94wDpD53xDNB5cmLTTFECEJlhE5To=
-k8s.io/apiextensions-apiserver v0.34.0-alpha.2 h1:hBrCOLTT0Gavt/kviPfv6b6P+hmoCja7m9ewLzDqw6o=
-k8s.io/apiextensions-apiserver v0.34.0-alpha.2/go.mod h1:Lbus5MTRVmJ7uTzLngou3HMMTCRvnvGQmL5aJoYKOL0=
-k8s.io/apimachinery v0.34.0-alpha.2 h1:kKK+KUnbsRWJObxY3pChe0IYkEWG5L4Qsmr/iTIuFQE=
-k8s.io/apimachinery v0.34.0-alpha.2/go.mod h1:HYUvaFuBNToT/JW+lhc2D9BYgqj6xhI5eORJ7BnvKpM=
-k8s.io/apiserver v0.34.0-alpha.2 h1:8VX5Tgiaw4YqQpAa2NDbuVcceqjbrTP+fWez8s5Jub4=
-k8s.io/apiserver v0.34.0-alpha.2/go.mod h1:Jsqd/XO4Wf9SyJuD9TvYWUfyYzmaDCi0VFaVa9CoWyA=
-k8s.io/client-go v0.34.0-alpha.2 h1:3TwS7HoRQZ8hQp9Sud4j+oS0O2N++3lyMVJavogALzw=
-k8s.io/client-go v0.34.0-alpha.2/go.mod h1:DWWpv2S99iGzK+cUwhQBe1ipIU9PVeREVjl/XWHlr4Y=
-k8s.io/component-base v0.34.0-alpha.2 h1:IRXrJ1nZHmzp4sAGYiv8uN2XvOCTelLwWXr5DxJp4cQ=
-k8s.io/component-base v0.34.0-alpha.2/go.mod h1:nk7MlWTLUWnSkaR72wJEH+bD6+zBtPb3CL9EMi2aMWc=
+k8s.io/api v0.34.0-beta.0 h1:Yok13a6+0GTJYv56+lSNlgRjF46TLUVv+gaijkLdkB4=
+k8s.io/api v0.34.0-beta.0/go.mod h1:OOkMaR1spg7vtjCCQSdxgiRD0Bbp2pcTCBxJtEEb4Pk=
+k8s.io/apiextensions-apiserver v0.34.0-beta.0 h1:zzR2yWEBkfYcjfwM42CwrBwtwLWLBJ/VwYpyP2ldNMI=
+k8s.io/apiextensions-apiserver v0.34.0-beta.0/go.mod h1:lCs/ZFdQjQWkNdEa0CS7qvLAMVDCj3fVXwSmKZ7Fn8E=
+k8s.io/apimachinery v0.34.0-beta.0 h1:C6teSJBCx6ArW7122MM9hQqeGW2w/QQ0lB4x4Z4Iftc=
+k8s.io/apimachinery v0.34.0-beta.0/go.mod h1:TP8uyOuDEOnzGpLOdffo8hPnIjVDljZCxCM/fruV+5M=
+k8s.io/apiserver v0.34.0-beta.0 h1:cLhc6un0qtCb/nwGpkj8020yvs3oUxCu4vLLKEGkd4c=
+k8s.io/apiserver v0.34.0-beta.0/go.mod h1:RO2L/h9tBdqlrQGz/ghIGOWzF+aEwCkb13ckUdmLolo=
+k8s.io/client-go v0.34.0-beta.0 h1:snyuzY74xHIp7qxMYul8QRn1WxPNhs281srSl2CX9nY=
+k8s.io/client-go v0.34.0-beta.0/go.mod h1:KNxQTmxlWq7adFeedQLu+qVkIBTZqhghesFk41YWyBQ=
+k8s.io/component-base v0.34.0-beta.0 h1:w2zC7n8lXbOqsC8UQ97JqTneDsy+ThFcbxQnfQ98/TM=
+k8s.io/component-base v0.34.0-beta.0/go.mod h1:KRfhQL9KZm3sywusY38Z2An7ulFDCaL+1juNF4NLhUw=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
 k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
-k8s.io/kube-openapi v0.0.0-20250610211856-8b98d1ed966a h1:ZV3Zr+/7s7aVbjNGICQt+ppKWsF1tehxggNfbM7XnG8=
-k8s.io/kube-openapi v0.0.0-20250610211856-8b98d1ed966a/go.mod h1:5jIi+8yX4RIb8wk3XwBo5Pq2ccx4FP10ohkbSKCZoK8=
+k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b h1:MloQ9/bdJyIu9lb1PzujOPolHyvO06MXG5TUIj2mNAA=
+k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b/go.mod h1:UZ2yyWbFTpuhSbFhv24aGNOdoRdJZgsIObGBUaYVsts=
 k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 h1:hwvWFiBzdWw1FhfY1FooPn3kzWuJ8tmbZBHi4zVsl1Y=
 k8s.io/utils v0.0.0-20250604170112-4c0f3b243397/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 h1:jpcvIRr3GLoUoEKRkHKSmGjxb6lWwrBlJsXc+eUYQHM=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2/go.mod h1:Ve9uj1L+deCXFrPOk1LpFXqTg7LCFzFso6PA48q/XZw=
-sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 h1:/Rv+M11QRah1itp8VhT6HoVx1Ray9eB4DBr+K+/sCJ8=
-sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3/go.mod h1:18nIHnGi6636UCz6m8i4DhaJ65T6EruyzmoQqI2BVDo=
-sigs.k8s.io/randfill v0.0.0-20250304075658-069ef1bbf016/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
+sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 h1:gBQPwqORJ8d8/YNZWEjoZs7npUVDpVXUUOFfW6CgAqE=
+sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
 sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU=
 sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
-sigs.k8s.io/structured-merge-diff/v4 v4.7.0 h1:qPeWmscJcXP0snki5IYF79Z8xrl8ETFxgMd7wez1XkI=
-sigs.k8s.io/structured-merge-diff/v4 v4.7.0/go.mod h1:dDy58f92j70zLsuZVuUX5Wp9vtxXpaZnkPGWeqDfCps=
-sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E=
-sigs.k8s.io/yaml v1.4.0/go.mod h1:Ejl7/uTz7PSA4eKMyQCUTnhZYNmLIl+5c2lQPGR2BPY=
+sigs.k8s.io/structured-merge-diff/v6 v6.2.0 h1:msyqjP8Nyd5sF3QSmJouFSzcBIdwq4ct8d1/7VSBHIQ=
+sigs.k8s.io/structured-merge-diff/v6 v6.2.0/go.mod h1:M3W8sfWvn2HhQDIbGWj3S099YozAsymCo/wrT5ohRUE=
+sigs.k8s.io/yaml v1.5.0 h1:M10b2U7aEUY6hRtU870n2VTPgR5RZiL/I6Lcc2F4NUQ=
+sigs.k8s.io/yaml v1.5.0/go.mod h1:wZs27Rbxoai4C0f8/9urLZtZtF3avA3gKvGyPdDqTO4=
diff --git a/pkg/cache/internal/informers.go b/pkg/cache/internal/informers.go
index 4bf832b2..f216be0d 100644
--- a/pkg/cache/internal/informers.go
+++ b/pkg/cache/internal/informers.go
@@ -518,7 +518,7 @@ func (ip *Informers) makeListWatcher(gvk schema.GroupVersionKind, obj runtime.Ob
 	// Structured.
 	//
 	default:
-		client, err := apiutil.RESTClientForGVK(gvk, false, ip.config, ip.codecs, ip.httpClient)
+		client, err := apiutil.RESTClientForGVK(gvk, false, false, ip.config, ip.codecs, ip.httpClient)
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/client/apiutil/apimachinery.go b/pkg/client/apiutil/apimachinery.go
index 1d4ce264..b132cb2d 100644
--- a/pkg/client/apiutil/apimachinery.go
+++ b/pkg/client/apiutil/apimachinery.go
@@ -161,15 +161,27 @@ func GVKForObject(obj runtime.Object, scheme *runtime.Scheme) (schema.GroupVersi
 // RESTClientForGVK constructs a new rest.Interface capable of accessing the resource associated
 // with the given GroupVersionKind. The REST client will be configured to use the negotiated serializer from
 // baseConfig, if set, otherwise a default serializer will be set.
-func RESTClientForGVK(gvk schema.GroupVersionKind, isUnstructured bool, baseConfig *rest.Config, codecs serializer.CodecFactory, httpClient *http.Client) (rest.Interface, error) {
+func RESTClientForGVK(
+	gvk schema.GroupVersionKind,
+	forceDisableProtoBuf bool,
+	isUnstructured bool,
+	baseConfig *rest.Config,
+	codecs serializer.CodecFactory,
+	httpClient *http.Client,
+) (rest.Interface, error) {
 	if httpClient == nil {
 		return nil, fmt.Errorf("httpClient must not be nil, consider using rest.HTTPClientFor(c) to create a client")
 	}
-	return rest.RESTClientForConfigAndClient(createRestConfig(gvk, isUnstructured, baseConfig, codecs), httpClient)
+	return rest.RESTClientForConfigAndClient(createRestConfig(gvk, forceDisableProtoBuf, isUnstructured, baseConfig, codecs), httpClient)
 }
 
 // createRestConfig copies the base config and updates needed fields for a new rest config.
-func createRestConfig(gvk schema.GroupVersionKind, isUnstructured bool, baseConfig *rest.Config, codecs serializer.CodecFactory) *rest.Config {
+func createRestConfig(gvk schema.GroupVersionKind,
+	forceDisableProtoBuf bool,
+	isUnstructured bool,
+	baseConfig *rest.Config,
+	codecs serializer.CodecFactory,
+) *rest.Config {
 	gv := gvk.GroupVersion()
 
 	cfg := rest.CopyConfig(baseConfig)
@@ -183,7 +195,7 @@ func createRestConfig(gvk schema.GroupVersionKind, isUnstructured bool, baseConf
 		cfg.UserAgent = rest.DefaultKubernetesUserAgent()
 	}
 	// TODO(FillZpp): In the long run, we want to check discovery or something to make sure that this is actually true.
-	if cfg.ContentType == "" && !isUnstructured {
+	if cfg.ContentType == "" && !forceDisableProtoBuf {
 		protobufSchemeLock.RLock()
 		if protobufScheme.Recognizes(gvk) {
 			cfg.ContentType = runtime.ContentTypeProtobuf
diff --git a/pkg/client/applyconfigurations.go b/pkg/client/applyconfigurations.go
new file mode 100644
index 00000000..97192050
--- /dev/null
+++ b/pkg/client/applyconfigurations.go
@@ -0,0 +1,75 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/utils/ptr"
+)
+
+type unstructuredApplyConfiguration struct {
+	*unstructured.Unstructured
+}
+
+func (u *unstructuredApplyConfiguration) IsApplyConfiguration() {}
+
+// ApplyConfigurationFromUnstructured creates a runtime.ApplyConfiguration from an *unstructured.Unstructured object.
+//
+// Do not use Unstructured objects here that were generated from API objects, as its impossible to tell
+// if a zero value was explicitly set.
+func ApplyConfigurationFromUnstructured(u *unstructured.Unstructured) runtime.ApplyConfiguration {
+	return &unstructuredApplyConfiguration{Unstructured: u}
+}
+
+type applyconfigurationRuntimeObject struct {
+	runtime.ApplyConfiguration
+}
+
+func (a *applyconfigurationRuntimeObject) GetObjectKind() schema.ObjectKind {
+	return a
+}
+
+func (a *applyconfigurationRuntimeObject) GroupVersionKind() schema.GroupVersionKind {
+	return schema.GroupVersionKind{}
+}
+
+func (a *applyconfigurationRuntimeObject) SetGroupVersionKind(gvk schema.GroupVersionKind) {}
+
+func (a *applyconfigurationRuntimeObject) DeepCopyObject() runtime.Object {
+	panic("applyconfigurationRuntimeObject does not support DeepCopyObject")
+}
+
+func runtimeObjectFromApplyConfiguration(ac runtime.ApplyConfiguration) runtime.Object {
+	return &applyconfigurationRuntimeObject{ApplyConfiguration: ac}
+}
+
+func gvkFromApplyConfiguration(ac applyConfiguration) (schema.GroupVersionKind, error) {
+	var gvk schema.GroupVersionKind
+	gv, err := schema.ParseGroupVersion(ptr.Deref(ac.GetAPIVersion(), ""))
+	if err != nil {
+		return gvk, fmt.Errorf("failed to parse %q as GroupVersion: %w", ptr.Deref(ac.GetAPIVersion(), ""), err)
+	}
+	gvk.Group = gv.Group
+	gvk.Version = gv.Version
+	gvk.Kind = ptr.Deref(ac.GetKind(), "")
+
+	return gvk, nil
+}
diff --git a/pkg/client/client.go b/pkg/client/client.go
index 50b0ebf3..092deb43 100644
--- a/pkg/client/client.go
+++ b/pkg/client/client.go
@@ -329,6 +329,16 @@ func (c *client) Patch(ctx context.Context, obj Object, patch Patch, opts ...Pat
 	}
 }
 
+func (c *client) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...ApplyOption) error {
+	switch obj := obj.(type) {
+	case *unstructuredApplyConfiguration:
+		defer c.resetGroupVersionKind(obj, obj.GetObjectKind().GroupVersionKind())
+		return c.unstructuredClient.Apply(ctx, obj, opts...)
+	default:
+		return c.typedClient.Apply(ctx, obj, opts...)
+	}
+}
+
 // Get implements client.Client.
 func (c *client) Get(ctx context.Context, key ObjectKey, obj Object, opts ...GetOption) error {
 	if isUncached, err := c.shouldBypassCache(obj); err != nil {
diff --git a/pkg/client/client_rest_resources.go b/pkg/client/client_rest_resources.go
index 2d078795..acff7a46 100644
--- a/pkg/client/client_rest_resources.go
+++ b/pkg/client/client_rest_resources.go
@@ -17,16 +17,17 @@ limitations under the License.
 package client
 
 import (
+	"fmt"
 	"net/http"
 	"strings"
 	"sync"
 
 	"k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/client-go/rest"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 )
 
@@ -56,13 +57,17 @@ type clientRestResources struct {
 
 // newResource maps obj to a Kubernetes Resource and constructs a client for that Resource.
 // If the object is a list, the resource represents the item's type instead.
-func (c *clientRestResources) newResource(gvk schema.GroupVersionKind, isList, isUnstructured bool) (*resourceMeta, error) {
+func (c *clientRestResources) newResource(gvk schema.GroupVersionKind,
+	isList bool,
+	forceDisableProtoBuf bool,
+	isUnstructured bool,
+) (*resourceMeta, error) {
 	if strings.HasSuffix(gvk.Kind, "List") && isList {
 		// if this was a list, treat it as a request for the item's resource
 		gvk.Kind = gvk.Kind[:len(gvk.Kind)-4]
 	}
 
-	client, err := apiutil.RESTClientForGVK(gvk, isUnstructured, c.config, c.codecs, c.httpClient)
+	client, err := apiutil.RESTClientForGVK(gvk, forceDisableProtoBuf, isUnstructured, c.config, c.codecs, c.httpClient)
 	if err != nil {
 		return nil, err
 	}
@@ -73,15 +78,41 @@ func (c *clientRestResources) newResource(gvk schema.GroupVersionKind, isList, i
 	return &resourceMeta{Interface: client, mapping: mapping, gvk: gvk}, nil
 }
 
+type applyConfiguration interface {
+	GetName() *string
+	GetNamespace() *string
+	GetKind() *string
+	GetAPIVersion() *string
+}
+
 // getResource returns the resource meta information for the given type of object.
 // If the object is a list, the resource represents the item's type instead.
-func (c *clientRestResources) getResource(obj runtime.Object) (*resourceMeta, error) {
-	gvk, err := apiutil.GVKForObject(obj, c.scheme)
-	if err != nil {
-		return nil, err
+func (c *clientRestResources) getResource(obj any) (*resourceMeta, error) {
+	var gvk schema.GroupVersionKind
+	var err error
+	var isApplyConfiguration bool
+	switch o := obj.(type) {
+	case runtime.Object:
+		gvk, err = apiutil.GVKForObject(o, c.scheme)
+		if err != nil {
+			return nil, err
+		}
+	case runtime.ApplyConfiguration:
+		ac, ok := o.(applyConfiguration)
+		if !ok {
+			return nil, fmt.Errorf("%T is a runtime.ApplyConfiguration but not an applyConfiguration", o)
+		}
+		gvk, err = gvkFromApplyConfiguration(ac)
+		if err != nil {
+			return nil, err
+		}
+		isApplyConfiguration = true
+	default:
+		return nil, fmt.Errorf("bug: %T is neither a runtime.Object nor a runtime.ApplyConfiguration", o)
 	}
 
 	_, isUnstructured := obj.(runtime.Unstructured)
+	forceDisableProtoBuf := isUnstructured || isApplyConfiguration
 
 	// It's better to do creation work twice than to not let multiple
 	// people make requests at once
@@ -97,10 +128,15 @@ func (c *clientRestResources) getResource(obj runtime.Object) (*resourceMeta, er
 		return r, nil
 	}
 
+	var isList bool
+	if runtimeObject, ok := obj.(runtime.Object); ok && meta.IsListType(runtimeObject) {
+		isList = true
+	}
+
 	// Initialize a new Client
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	r, err = c.newResource(gvk, meta.IsListType(obj), isUnstructured)
+	r, err = c.newResource(gvk, isList, forceDisableProtoBuf, isUnstructured)
 	if err != nil {
 		return nil, err
 	}
@@ -109,16 +145,29 @@ func (c *clientRestResources) getResource(obj runtime.Object) (*resourceMeta, er
 }
 
 // getObjMeta returns objMeta containing both type and object metadata and state.
-func (c *clientRestResources) getObjMeta(obj runtime.Object) (*objMeta, error) {
+func (c *clientRestResources) getObjMeta(obj any) (*objMeta, error) {
 	r, err := c.getResource(obj)
 	if err != nil {
 		return nil, err
 	}
-	m, err := meta.Accessor(obj)
-	if err != nil {
-		return nil, err
+	objMeta := &objMeta{resourceMeta: r}
+
+	switch o := obj.(type) {
+	case runtime.Object:
+		m, err := meta.Accessor(obj)
+		if err != nil {
+			return nil, err
+		}
+		objMeta.namespace = m.GetNamespace()
+		objMeta.name = m.GetName()
+	case applyConfiguration:
+		objMeta.namespace = ptr.Deref(o.GetNamespace(), "")
+		objMeta.name = ptr.Deref(o.GetName(), "")
+	default:
+		return nil, fmt.Errorf("object %T is neither a runtime.Object nor a runtime.ApplyConfiguration", obj)
 	}
-	return &objMeta{resourceMeta: r, Object: m}, err
+
+	return objMeta, nil
 }
 
 // resourceMeta stores state for a Kubernetes type.
@@ -146,6 +195,6 @@ type objMeta struct {
 	// resourceMeta contains type information for the object
 	*resourceMeta
 
-	// Object contains meta data for the object instance
-	metav1.Object
+	namespace string
+	name      string
 }
diff --git a/pkg/client/dryrun.go b/pkg/client/dryrun.go
index bbcdd383..a185860d 100644
--- a/pkg/client/dryrun.go
+++ b/pkg/client/dryrun.go
@@ -82,6 +82,10 @@ func (c *dryRunClient) Patch(ctx context.Context, obj Object, patch Patch, opts
 	return c.client.Patch(ctx, obj, patch, append(opts, DryRunAll)...)
 }
 
+func (c *dryRunClient) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...ApplyOption) error {
+	return c.client.Apply(ctx, obj, append(opts, DryRunAll)...)
+}
+
 // Get implements client.Client.
 func (c *dryRunClient) Get(ctx context.Context, key ObjectKey, obj Object, opts ...GetOption) error {
 	return c.client.Get(ctx, key, obj, opts...)
diff --git a/pkg/client/fake/client.go b/pkg/client/fake/client.go
index 2ee891e2..380c728b 100644
--- a/pkg/client/fake/client.go
+++ b/pkg/client/fake/client.go
@@ -52,6 +52,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/validation"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -1046,17 +1047,53 @@ func (c *fakeClient) Patch(ctx context.Context, obj client.Object, patch client.
 	return c.patch(obj, patch, opts...)
 }
 
+func (c *fakeClient) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...client.ApplyOption) error {
+	applyOpts := &client.ApplyOptions{}
+	applyOpts.ApplyOptions(opts)
+
+	data, err := json.Marshal(obj)
+	if err != nil {
+		return fmt.Errorf("failed to marshal apply configuration: %w", err)
+	}
+
+	u := &unstructured.Unstructured{}
+	if err := json.Unmarshal(data, u); err != nil {
+		return fmt.Errorf("failed to unmarshal apply configuration: %w", err)
+	}
+
+	applyPatch := &fakeApplyPatch{}
+
+	patchOpts := &client.PatchOptions{}
+	patchOpts.Raw = applyOpts.AsPatchOptions()
+
+	return c.patch(u, applyPatch, patchOpts)
+}
+
+type fakeApplyPatch struct{}
+
+func (p *fakeApplyPatch) Type() types.PatchType {
+	return types.ApplyPatchType
+}
+
+func (p *fakeApplyPatch) Data(obj client.Object) ([]byte, error) {
+	return json.Marshal(obj)
+}
+
 func (c *fakeClient) patch(obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
 	if err := c.addToSchemeIfUnknownAndUnstructuredOrPartial(obj); err != nil {
 		return err
 	}
 
-	c.schemeLock.RLock()
-	defer c.schemeLock.RUnlock()
-
 	patchOptions := &client.PatchOptions{}
 	patchOptions.ApplyOptions(opts)
 
+	if errs := validation.ValidatePatchOptions(patchOptions.AsPatchOptions(), patch.Type()); len(errs) > 0 {
+		return apierrors.NewInvalid(schema.GroupKind{Group: "meta.k8s.io", Kind: "PatchOptions"}, "", errs)
+	}
+
+	c.schemeLock.RLock()
+	defer c.schemeLock.RUnlock()
+
 	for _, dryRunOpt := range patchOptions.DryRun {
 		if dryRunOpt == metav1.DryRunAll {
 			return nil
diff --git a/pkg/client/fake/typeconverter.go b/pkg/client/fake/typeconverter.go
index 5b92de73..3cb3a0dc 100644
--- a/pkg/client/fake/typeconverter.go
+++ b/pkg/client/fake/typeconverter.go
@@ -22,7 +22,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/managedfields"
-	"sigs.k8s.io/structured-merge-diff/v4/typed"
+	"sigs.k8s.io/structured-merge-diff/v6/typed"
 )
 
 type multiTypeConverter struct {
diff --git a/pkg/client/fieldowner.go b/pkg/client/fieldowner.go
index 07183cd1..93274f95 100644
--- a/pkg/client/fieldowner.go
+++ b/pkg/client/fieldowner.go
@@ -54,6 +54,10 @@ func (f *clientWithFieldManager) Patch(ctx context.Context, obj Object, patch Pa
 	return f.c.Patch(ctx, obj, patch, append([]PatchOption{FieldOwner(f.owner)}, opts...)...)
 }
 
+func (f *clientWithFieldManager) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...ApplyOption) error {
+	return f.c.Apply(ctx, obj, append([]ApplyOption{FieldOwner(f.owner)}, opts...)...)
+}
+
 func (f *clientWithFieldManager) Delete(ctx context.Context, obj Object, opts ...DeleteOption) error {
 	return f.c.Delete(ctx, obj, opts...)
 }
diff --git a/pkg/client/fieldvalidation.go b/pkg/client/fieldvalidation.go
index 659b3d44..ce8d0576 100644
--- a/pkg/client/fieldvalidation.go
+++ b/pkg/client/fieldvalidation.go
@@ -53,6 +53,10 @@ func (c *clientWithFieldValidation) Patch(ctx context.Context, obj Object, patch
 	return c.client.Patch(ctx, obj, patch, append([]PatchOption{c.validation}, opts...)...)
 }
 
+func (c *clientWithFieldValidation) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...ApplyOption) error {
+	return c.client.Apply(ctx, obj, opts...)
+}
+
 func (c *clientWithFieldValidation) Delete(ctx context.Context, obj Object, opts ...DeleteOption) error {
 	return c.client.Delete(ctx, obj, opts...)
 }
diff --git a/pkg/client/interceptor/intercept.go b/pkg/client/interceptor/intercept.go
index 3d3f3cb0..7ff73bd8 100644
--- a/pkg/client/interceptor/intercept.go
+++ b/pkg/client/interceptor/intercept.go
@@ -19,6 +19,7 @@ type Funcs struct {
 	DeleteAllOf       func(ctx context.Context, client client.WithWatch, obj client.Object, opts ...client.DeleteAllOfOption) error
 	Update            func(ctx context.Context, client client.WithWatch, obj client.Object, opts ...client.UpdateOption) error
 	Patch             func(ctx context.Context, client client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error
+	Apply             func(ctx context.Context, client client.WithWatch, obj runtime.ApplyConfiguration, opts ...client.ApplyOption) error
 	Watch             func(ctx context.Context, client client.WithWatch, obj client.ObjectList, opts ...client.ListOption) (watch.Interface, error)
 	SubResource       func(client client.WithWatch, subResource string) client.SubResourceClient
 	SubResourceGet    func(ctx context.Context, client client.Client, subResourceName string, obj client.Object, subResource client.Object, opts ...client.SubResourceGetOption) error
@@ -92,6 +93,14 @@ func (c interceptor) Patch(ctx context.Context, obj client.Object, patch client.
 	return c.client.Patch(ctx, obj, patch, opts...)
 }
 
+func (c interceptor) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...client.ApplyOption) error {
+	if c.funcs.Apply != nil {
+		return c.funcs.Apply(ctx, c.client, obj, opts...)
+	}
+
+	return c.client.Apply(ctx, obj, opts...)
+}
+
 func (c interceptor) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error {
 	if c.funcs.DeleteAllOf != nil {
 		return c.funcs.DeleteAllOf(ctx, c.client, obj, opts...)
diff --git a/pkg/client/interfaces.go b/pkg/client/interfaces.go
index 3b282fc2..61559ecb 100644
--- a/pkg/client/interfaces.go
+++ b/pkg/client/interfaces.go
@@ -61,6 +61,9 @@ type Reader interface {
 
 // Writer knows how to create, delete, and update Kubernetes objects.
 type Writer interface {
+	// Apply applies the given apply configuration to the Kubernetes cluster.
+	Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...ApplyOption) error
+
 	// Create saves the object obj in the Kubernetes cluster. obj must be a
 	// struct pointer so that obj can be updated with the content returned by the Server.
 	Create(ctx context.Context, obj Object, opts ...CreateOption) error
diff --git a/pkg/client/namespaced_client.go b/pkg/client/namespaced_client.go
index 222dc795..cacba4a9 100644
--- a/pkg/client/namespaced_client.go
+++ b/pkg/client/namespaced_client.go
@@ -19,10 +19,13 @@ package client
 import (
 	"context"
 	"fmt"
+	"reflect"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 )
 
 // NewNamespacedClient wraps an existing client enforcing the namespace value.
@@ -147,6 +150,52 @@ func (n *namespacedClient) Patch(ctx context.Context, obj Object, patch Patch, o
 	return n.client.Patch(ctx, obj, patch, opts...)
 }
 
+func (n *namespacedClient) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...ApplyOption) error {
+	var gvk schema.GroupVersionKind
+	switch o := obj.(type) {
+	case applyConfiguration:
+		var err error
+		gvk, err = gvkFromApplyConfiguration(o)
+		if err != nil {
+			return err
+		}
+	case *unstructuredApplyConfiguration:
+		gvk = o.GroupVersionKind()
+	default:
+		return fmt.Errorf("object %T is not a valid apply configuration", obj)
+	}
+	isNamespaceScoped, err := apiutil.IsGVKNamespaced(gvk, n.RESTMapper())
+	if err != nil {
+		return fmt.Errorf("error finding the scope of the object: %w", err)
+	}
+	if isNamespaceScoped {
+		switch o := obj.(type) {
+		case applyConfiguration:
+			if o.GetNamespace() != nil && *o.GetNamespace() != "" && *o.GetNamespace() != n.namespace {
+				return fmt.Errorf("namespace %s provided for the object %s does not match the namespace %s on the client",
+					*o.GetNamespace(), ptr.Deref(o.GetName(), ""), n.namespace)
+			}
+			v := reflect.ValueOf(o)
+			withNamespace := v.MethodByName("WithNamespace")
+			if !withNamespace.IsValid() {
+				return fmt.Errorf("ApplyConfiguration %T does not have a WithNamespace method", o)
+			}
+			if tp := withNamespace.Type(); tp.NumIn() != 1 || tp.In(0).Kind() != reflect.String {
+				return fmt.Errorf("WithNamespace method of ApplyConfiguration %T must take a single string argument", o)
+			}
+			withNamespace.Call([]reflect.Value{reflect.ValueOf(n.namespace)})
+		case *unstructuredApplyConfiguration:
+			if o.GetNamespace() != "" && o.GetNamespace() != n.namespace {
+				return fmt.Errorf("namespace %s provided for the object %s does not match the namespace %s on the client",
+					o.GetNamespace(), o.GetName(), n.namespace)
+			}
+			o.SetNamespace(n.namespace)
+		}
+	}
+
+	return n.client.Apply(ctx, obj, opts...)
+}
+
 // Get implements client.Client.
 func (n *namespacedClient) Get(ctx context.Context, key ObjectKey, obj Object, opts ...GetOption) error {
 	isNamespaceScoped, err := n.IsObjectNamespaced(obj)
diff --git a/pkg/client/options.go b/pkg/client/options.go
index ad4b8182..86da5ddf 100644
--- a/pkg/client/options.go
+++ b/pkg/client/options.go
@@ -62,6 +62,12 @@ type PatchOption interface {
 	ApplyToPatch(*PatchOptions)
 }
 
+// ApplyOption is some configuration that modifies options for an apply request.
+type ApplyOption interface {
+	// ApplyToApply applies this configuration to the given apply options.
+	ApplyToApply(*ApplyOptions)
+}
+
 // DeleteAllOfOption is some configuration that modifies options for a delete request.
 type DeleteAllOfOption interface {
 	// ApplyToDeleteAllOf applies this configuration to the given deletecollection options.
@@ -116,7 +122,12 @@ func (dryRunAll) ApplyToPatch(opts *PatchOptions) {
 	opts.DryRun = []string{metav1.DryRunAll}
 }
 
-// ApplyToPatch applies this configuration to the given delete options.
+// ApplyToApply applies this configuration to the given apply options.
+func (dryRunAll) ApplyToApply(opts *ApplyOptions) {
+	opts.DryRun = []string{metav1.DryRunAll}
+}
+
+// ApplyToDelete applies this configuration to the given delete options.
 func (dryRunAll) ApplyToDelete(opts *DeleteOptions) {
 	opts.DryRun = []string{metav1.DryRunAll}
 }
@@ -155,6 +166,11 @@ func (f FieldOwner) ApplyToUpdate(opts *UpdateOptions) {
 	opts.FieldManager = string(f)
 }
 
+// ApplyToApply applies this configuration to the given apply options.
+func (f FieldOwner) ApplyToApply(opts *ApplyOptions) {
+	opts.FieldManager = string(f)
+}
+
 // ApplyToSubResourcePatch applies this configuration to the given patch options.
 func (f FieldOwner) ApplyToSubResourcePatch(opts *SubResourcePatchOptions) {
 	opts.FieldManager = string(f)
@@ -872,10 +888,18 @@ func (o *PatchOptions) AsPatchOptions() *metav1.PatchOptions {
 		o.Raw = &metav1.PatchOptions{}
 	}
 
-	o.Raw.DryRun = o.DryRun
-	o.Raw.Force = o.Force
-	o.Raw.FieldManager = o.FieldManager
-	o.Raw.FieldValidation = o.FieldValidation
+	if o.DryRun != nil {
+		o.Raw.DryRun = o.DryRun
+	}
+	if o.Force != nil {
+		o.Raw.Force = o.Force
+	}
+	if o.FieldManager != "" {
+		o.Raw.FieldManager = o.FieldManager
+	}
+	if o.FieldValidation != "" {
+		o.Raw.FieldValidation = o.FieldValidation
+	}
 	return o.Raw
 }
 
@@ -908,13 +932,15 @@ var ForceOwnership = forceOwnership{}
 type forceOwnership struct{}
 
 func (forceOwnership) ApplyToPatch(opts *PatchOptions) {
-	definitelyTrue := true
-	opts.Force = &definitelyTrue
+	opts.Force = ptr.To(true)
 }
 
 func (forceOwnership) ApplyToSubResourcePatch(opts *SubResourcePatchOptions) {
-	definitelyTrue := true
-	opts.Force = &definitelyTrue
+	opts.Force = ptr.To(true)
+}
+
+func (forceOwnership) ApplyToApply(opts *ApplyOptions) {
+	opts.Force = ptr.To(true)
 }
 
 // }}}
@@ -948,3 +974,57 @@ func (o *DeleteAllOfOptions) ApplyToDeleteAllOf(do *DeleteAllOfOptions) {
 }
 
 // }}}
+
+// ApplyOptions are the options for an apply request.
+type ApplyOptions struct {
+	// When present, indicates that modifications should not be
+	// persisted. An invalid or unrecognized dryRun directive will
+	// result in an error response and no further processing of the
+	// request. Valid values are:
+	// - All: all dry run stages will be processed
+	DryRun []string
+
+	// Force is going to "force" Apply requests. It means user will
+	// re-acquire conflicting fields owned by other people.
+	Force *bool
+
+	// fieldManager is a name associated with the actor or entity
+	// that is making these changes. The value must be less than or
+	// 128 characters long, and only contain printable characters,
+	// as defined by https://golang.org/pkg/unicode/#IsPrint. This
+	// field is required.
+	//
+	// +required
+	FieldManager string
+}
+
+// ApplyOptions applies the given opts onto the ApplyOptions
+func (o *ApplyOptions) ApplyOptions(opts []ApplyOption) *ApplyOptions {
+	for _, opt := range opts {
+		opt.ApplyToApply(o)
+	}
+	return o
+}
+
+// ApplyToApply applies the given opts onto the ApplyOptions
+func (o *ApplyOptions) ApplyToApply(opts *ApplyOptions) {
+	if o.DryRun != nil {
+		opts.DryRun = o.DryRun
+	}
+	if o.Force != nil {
+		opts.Force = o.Force
+	}
+
+	if o.FieldManager != "" {
+		opts.FieldManager = o.FieldManager
+	}
+}
+
+// AsPatchOptions constructs patch options from the given ApplyOptions
+func (o *ApplyOptions) AsPatchOptions() *metav1.PatchOptions {
+	return &metav1.PatchOptions{
+		DryRun:       o.DryRun,
+		Force:        o.Force,
+		FieldManager: o.FieldManager,
+	}
+}
diff --git a/pkg/client/patch.go b/pkg/client/patch.go
index 11d60838..95cb331e 100644
--- a/pkg/client/patch.go
+++ b/pkg/client/patch.go
@@ -27,6 +27,7 @@ import (
 
 var (
 	// Apply uses server-side apply to patch the given object.
+	// Deprecated: Use client.Apply instead.
 	Apply Patch = applyPatch{}
 
 	// Merge uses the raw object as a merge patch, without modifications.
diff --git a/pkg/client/typed_client.go b/pkg/client/typed_client.go
index 92afd9a9..3bd762a6 100644
--- a/pkg/client/typed_client.go
+++ b/pkg/client/typed_client.go
@@ -18,8 +18,10 @@ package client
 
 import (
 	"context"
+	"fmt"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/util/apply"
 )
 
 var _ Reader = &typedClient{}
@@ -41,7 +43,7 @@ func (c *typedClient) Create(ctx context.Context, obj Object, opts ...CreateOpti
 	createOpts.ApplyOptions(opts)
 
 	return o.Post().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
 		Body(obj).
 		VersionedParams(createOpts.AsCreateOptions(), c.paramCodec).
@@ -60,9 +62,9 @@ func (c *typedClient) Update(ctx context.Context, obj Object, opts ...UpdateOpti
 	updateOpts.ApplyOptions(opts)
 
 	return o.Put().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		Body(obj).
 		VersionedParams(updateOpts.AsUpdateOptions(), c.paramCodec).
 		Do(ctx).
@@ -80,9 +82,9 @@ func (c *typedClient) Delete(ctx context.Context, obj Object, opts ...DeleteOpti
 	deleteOpts.ApplyOptions(opts)
 
 	return o.Delete().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		Body(deleteOpts.AsDeleteOptions()).
 		Do(ctx).
 		Error()
@@ -123,15 +125,40 @@ func (c *typedClient) Patch(ctx context.Context, obj Object, patch Patch, opts .
 	patchOpts.ApplyOptions(opts)
 
 	return o.Patch(patch.Type()).
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		VersionedParams(patchOpts.AsPatchOptions(), c.paramCodec).
 		Body(data).
 		Do(ctx).
 		Into(obj)
 }
 
+func (c *typedClient) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...ApplyOption) error {
+	o, err := c.resources.getObjMeta(obj)
+	if err != nil {
+		return err
+	}
+	req, err := apply.NewRequest(o, obj)
+	if err != nil {
+		return fmt.Errorf("failed to create apply request: %w", err)
+	}
+	applyOpts := &ApplyOptions{}
+	applyOpts.ApplyOptions(opts)
+
+	return req.
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
+		Resource(o.resource()).
+		Name(o.name).
+		VersionedParams(applyOpts.AsPatchOptions(), c.paramCodec).
+		Do(ctx).
+		// This is hacky, it is required because `Into` takes a `runtime.Object` and
+		// that is not implemented by the ApplyConfigurations. The generated clients
+		// don't have this problem because they deserialize into the api type, not the
+		// apply configuration: https://github.com/kubernetes/kubernetes/blob/22f5e01a37c0bc6a5f494dec14dd4e3688ee1d55/staging/src/k8s.io/client-go/gentype/type.go#L296-L317
+		Into(runtimeObjectFromApplyConfiguration(obj))
+}
+
 // Get implements client.Client.
 func (c *typedClient) Get(ctx context.Context, key ObjectKey, obj Object, opts ...GetOption) error {
 	r, err := c.resources.getResource(obj)
@@ -179,9 +206,9 @@ func (c *typedClient) GetSubResource(ctx context.Context, obj, subResourceObj Ob
 	getOpts.ApplyOptions(opts)
 
 	return o.Get().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		SubResource(subResource).
 		VersionedParams(getOpts.AsGetOptions(), c.paramCodec).
 		Do(ctx).
@@ -202,9 +229,9 @@ func (c *typedClient) CreateSubResource(ctx context.Context, obj Object, subReso
 	createOpts.ApplyOptions(opts)
 
 	return o.Post().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		SubResource(subResource).
 		Body(subResourceObj).
 		VersionedParams(createOpts.AsCreateOptions(), c.paramCodec).
@@ -237,9 +264,9 @@ func (c *typedClient) UpdateSubResource(ctx context.Context, obj Object, subReso
 	}
 
 	return o.Put().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		SubResource(subResource).
 		Body(body).
 		VersionedParams(updateOpts.AsUpdateOptions(), c.paramCodec).
@@ -268,9 +295,9 @@ func (c *typedClient) PatchSubResource(ctx context.Context, obj Object, subResou
 	}
 
 	return o.Patch(patch.Type()).
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		SubResource(subResource).
 		Body(data).
 		VersionedParams(patchOpts.AsPatchOptions(), c.paramCodec).
diff --git a/pkg/client/unstructured_client.go b/pkg/client/unstructured_client.go
index 0d969517..e636c3be 100644
--- a/pkg/client/unstructured_client.go
+++ b/pkg/client/unstructured_client.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/util/apply"
 )
 
 var _ Reader = &unstructuredClient{}
@@ -50,7 +51,7 @@ func (uc *unstructuredClient) Create(ctx context.Context, obj Object, opts ...Cr
 	createOpts.ApplyOptions(opts)
 
 	result := o.Post().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
 		Body(obj).
 		VersionedParams(createOpts.AsCreateOptions(), uc.paramCodec).
@@ -79,9 +80,9 @@ func (uc *unstructuredClient) Update(ctx context.Context, obj Object, opts ...Up
 	updateOpts.ApplyOptions(opts)
 
 	result := o.Put().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		Body(obj).
 		VersionedParams(updateOpts.AsUpdateOptions(), uc.paramCodec).
 		Do(ctx).
@@ -106,9 +107,9 @@ func (uc *unstructuredClient) Delete(ctx context.Context, obj Object, opts ...De
 	deleteOpts.ApplyOptions(opts)
 
 	return o.Delete().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		Body(deleteOpts.AsDeleteOptions()).
 		Do(ctx).
 		Error()
@@ -157,15 +158,41 @@ func (uc *unstructuredClient) Patch(ctx context.Context, obj Object, patch Patch
 	patchOpts.ApplyOptions(opts)
 
 	return o.Patch(patch.Type()).
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		VersionedParams(patchOpts.AsPatchOptions(), uc.paramCodec).
 		Body(data).
 		Do(ctx).
 		Into(obj)
 }
 
+func (uc *unstructuredClient) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...ApplyOption) error {
+	unstructuredApplyConfig, ok := obj.(*unstructuredApplyConfiguration)
+	if !ok {
+		return fmt.Errorf("bug: unstructured client got an applyconfiguration that was not %T but %T", &unstructuredApplyConfiguration{}, obj)
+	}
+	o, err := uc.resources.getObjMeta(unstructuredApplyConfig.Unstructured)
+	if err != nil {
+		return err
+	}
+
+	req, err := apply.NewRequest(o, obj)
+	if err != nil {
+		return fmt.Errorf("failed to create apply request: %w", err)
+	}
+	applyOpts := &ApplyOptions{}
+	applyOpts.ApplyOptions(opts)
+
+	return req.
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
+		Resource(o.resource()).
+		Name(o.name).
+		VersionedParams(applyOpts.AsPatchOptions(), uc.paramCodec).
+		Do(ctx).
+		Into(unstructuredApplyConfig.Unstructured)
+}
+
 // Get implements client.Client.
 func (uc *unstructuredClient) Get(ctx context.Context, key ObjectKey, obj Object, opts ...GetOption) error {
 	u, ok := obj.(runtime.Unstructured)
@@ -244,9 +271,9 @@ func (uc *unstructuredClient) GetSubResource(ctx context.Context, obj, subResour
 	getOpts.ApplyOptions(opts)
 
 	return o.Get().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		SubResource(subResource).
 		VersionedParams(getOpts.AsGetOptions(), uc.paramCodec).
 		Do(ctx).
@@ -275,9 +302,9 @@ func (uc *unstructuredClient) CreateSubResource(ctx context.Context, obj, subRes
 	createOpts.ApplyOptions(opts)
 
 	return o.Post().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		SubResource(subResource).
 		Body(subResourceObj).
 		VersionedParams(createOpts.AsCreateOptions(), uc.paramCodec).
@@ -310,9 +337,9 @@ func (uc *unstructuredClient) UpdateSubResource(ctx context.Context, obj Object,
 	}
 
 	return o.Put().
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		SubResource(subResource).
 		Body(body).
 		VersionedParams(updateOpts.AsUpdateOptions(), uc.paramCodec).
@@ -347,9 +374,9 @@ func (uc *unstructuredClient) PatchSubResource(ctx context.Context, obj Object,
 	}
 
 	result := o.Patch(patch.Type()).
-		NamespaceIfScoped(o.GetNamespace(), o.isNamespaced()).
+		NamespaceIfScoped(o.namespace, o.isNamespaced()).
 		Resource(o.resource()).
-		Name(o.GetName()).
+		Name(o.name).
 		SubResource(subResource).
 		Body(data).
 		VersionedParams(patchOpts.AsPatchOptions(), uc.paramCodec).
diff --git a/pkg/config/controller.go b/pkg/config/controller.go
index a5655593..89204803 100644
--- a/pkg/config/controller.go
+++ b/pkg/config/controller.go
@@ -60,6 +60,19 @@ type Controller struct {
 	// Defaults to true, which means the controller will use leader election.
 	NeedLeaderElection *bool
 
+	// EnableWarmup specifies whether the controller should start its sources when the manager is not
+	// the leader. This is useful for cases where sources take a long time to start, as it allows
+	// for the controller to warm up its caches even before it is elected as the leader. This
+	// improves leadership failover time, as the caches will be prepopulated before the controller
+	// transitions to be leader.
+	//
+	// Setting EnableWarmup to true and NeedLeaderElection to true means the controller will start its
+	// sources without waiting to become leader.
+	// Setting EnableWarmup to true and NeedLeaderElection to false is a no-op as controllers without
+	// leader election do not wait on leader election to start their sources.
+	// Defaults to false.
+	EnableWarmup *bool
+
 	// UsePriorityQueue configures the controllers queue to use the controller-runtime provided
 	// priority queue.
 	//
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 9de959b4..03e51abc 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -93,6 +93,19 @@ type TypedOptions[request comparable] struct {
 	//
 	// Note: This flag is disabled by default until a future version. It's currently in beta.
 	UsePriorityQueue *bool
+
+	// EnableWarmup specifies whether the controller should start its sources when the manager is not
+	// the leader. This is useful for cases where sources take a long time to start, as it allows
+	// for the controller to warm up its caches even before it is elected as the leader. This
+	// improves leadership failover time, as the caches will be prepopulated before the controller
+	// transitions to be leader.
+	//
+	// Setting EnableWarmup to true and NeedLeaderElection to true means the controller will start its
+	// sources without waiting to become leader.
+	// Setting EnableWarmup to true and NeedLeaderElection to false is a no-op as controllers without
+	// leader election do not wait on leader election to start their sources.
+	// Defaults to false.
+	EnableWarmup *bool
 }
 
 // DefaultFromConfig defaults the config from a config.Controller
@@ -124,6 +137,10 @@ func (options *TypedOptions[request]) DefaultFromConfig(config config.Controller
 	if options.NeedLeaderElection == nil {
 		options.NeedLeaderElection = config.NeedLeaderElection
 	}
+
+	if options.EnableWarmup == nil {
+		options.EnableWarmup = config.EnableWarmup
+	}
 }
 
 // Controller implements an API. A Controller manages a work queue fed reconcile.Requests
@@ -243,7 +260,7 @@ func NewTypedUnmanaged[request comparable](name string, options TypedOptions[req
 	}
 
 	// Create controller with dependencies set
-	return &controller.Controller[request]{
+	return controller.New[request](controller.Options[request]{
 		Do:                      options.Reconciler,
 		RateLimiter:             options.RateLimiter,
 		NewQueue:                options.NewQueue,
@@ -253,7 +270,8 @@ func NewTypedUnmanaged[request comparable](name string, options TypedOptions[req
 		LogConstructor:          options.LogConstructor,
 		RecoverPanic:            options.RecoverPanic,
 		LeaderElected:           options.NeedLeaderElection,
-	}, nil
+		EnableWarmup:            options.EnableWarmup,
+	}), nil
 }
 
 // ReconcileIDFromContext gets the reconcileID from the current context.
diff --git a/pkg/internal/controller/controller.go b/pkg/internal/controller/controller.go
index 9fa7ec71..18042e41 100644
--- a/pkg/internal/controller/controller.go
+++ b/pkg/internal/controller/controller.go
@@ -38,6 +38,51 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
+// Options are the arguments for creating a new Controller.
+type Options[request comparable] struct {
+	// Reconciler is a function that can be called at any time with the Name / Namespace of an object and
+	// ensures that the state of the system matches the state specified in the object.
+	// Defaults to the DefaultReconcileFunc.
+	Do reconcile.TypedReconciler[request]
+
+	// RateLimiter is used to limit how frequently requests may be queued into the work queue.
+	RateLimiter workqueue.TypedRateLimiter[request]
+
+	// NewQueue constructs the queue for this controller once the controller is ready to start.
+	// This is a func because the standard Kubernetes work queues start themselves immediately, which
+	// leads to goroutine leaks if something calls controller.New repeatedly.
+	NewQueue func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request]
+
+	// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
+	MaxConcurrentReconciles int
+
+	// CacheSyncTimeout refers to the time limit set on waiting for cache to sync
+	// Defaults to 2 minutes if not set.
+	CacheSyncTimeout time.Duration
+
+	// Name is used to uniquely identify a Controller in tracing, logging and monitoring.  Name is required.
+	Name string
+
+	// LogConstructor is used to construct a logger to then log messages to users during reconciliation,
+	// or for example when a watch is started.
+	// Note: LogConstructor has to be able to handle nil requests as we are also using it
+	// outside the context of a reconciliation.
+	LogConstructor func(request *request) logr.Logger
+
+	// RecoverPanic indicates whether the panic caused by reconcile should be recovered.
+	// Defaults to true.
+	RecoverPanic *bool
+
+	// LeaderElected indicates whether the controller is leader elected or always running.
+	LeaderElected *bool
+
+	// EnableWarmup specifies whether the controller should start its sources
+	// when the manager is not the leader.
+	// Defaults to false, which means that the controller will wait for leader election to start
+	// before starting sources.
+	EnableWarmup *bool
+}
+
 // Controller implements controller.Controller.
 type Controller[request comparable] struct {
 	// Name is used to uniquely identify a Controller in tracing, logging and monitoring.  Name is required.
@@ -83,6 +128,14 @@ type Controller[request comparable] struct {
 	// startWatches maintains a list of sources, handlers, and predicates to start when the controller is started.
 	startWatches []source.TypedSource[request]
 
+	// startedEventSourcesAndQueue is used to track if the event sources have been started.
+	// It ensures that we append sources to c.startWatches only until we call Start() / Warmup()
+	// It is true if startEventSourcesAndQueueLocked has been called at least once.
+	startedEventSourcesAndQueue bool
+
+	// didStartEventSourcesOnce is used to ensure that the event sources are only started once.
+	didStartEventSourcesOnce sync.Once
+
 	// LogConstructor is used to construct a logger to then log messages to users during reconciliation,
 	// or for example when a watch is started.
 	// Note: LogConstructor has to be able to handle nil requests as we are also using it
@@ -95,6 +148,35 @@ type Controller[request comparable] struct {
 
 	// LeaderElected indicates whether the controller is leader elected or always running.
 	LeaderElected *bool
+
+	// EnableWarmup specifies whether the controller should start its sources when the manager is not
+	// the leader. This is useful for cases where sources take a long time to start, as it allows
+	// for the controller to warm up its caches even before it is elected as the leader. This
+	// improves leadership failover time, as the caches will be prepopulated before the controller
+	// transitions to be leader.
+	//
+	// Setting EnableWarmup to true and NeedLeaderElection to true means the controller will start its
+	// sources without waiting to become leader.
+	// Setting EnableWarmup to true and NeedLeaderElection to false is a no-op as controllers without
+	// leader election do not wait on leader election to start their sources.
+	// Defaults to false.
+	EnableWarmup *bool
+}
+
+// New returns a new Controller configured with the given options.
+func New[request comparable](options Options[request]) *Controller[request] {
+	return &Controller[request]{
+		Do:                      options.Do,
+		RateLimiter:             options.RateLimiter,
+		NewQueue:                options.NewQueue,
+		MaxConcurrentReconciles: options.MaxConcurrentReconciles,
+		CacheSyncTimeout:        options.CacheSyncTimeout,
+		Name:                    options.Name,
+		LogConstructor:          options.LogConstructor,
+		RecoverPanic:            options.RecoverPanic,
+		LeaderElected:           options.LeaderElected,
+		EnableWarmup:            options.EnableWarmup,
+	}
 }
 
 // Reconcile implements reconcile.Reconciler.
@@ -124,10 +206,9 @@ func (c *Controller[request]) Watch(src source.TypedSource[request]) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	// Controller hasn't started yet, store the watches locally and return.
-	//
-	// These watches are going to be held on the controller struct until the manager or user calls Start(...).
-	if !c.Started {
+	// Sources weren't started yet, store the watches locally and return.
+	// These sources are going to be held until either Warmup() or Start(...) is called.
+	if !c.startedEventSourcesAndQueue {
 		c.startWatches = append(c.startWatches, src)
 		return nil
 	}
@@ -144,6 +225,21 @@ func (c *Controller[request]) NeedLeaderElection() bool {
 	return *c.LeaderElected
 }
 
+// Warmup implements the manager.WarmupRunnable interface.
+func (c *Controller[request]) Warmup(ctx context.Context) error {
+	if c.EnableWarmup == nil || !*c.EnableWarmup {
+		return nil
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	// Set the ctx so later calls to watch use this internal context
+	c.ctx = ctx
+
+	return c.startEventSourcesAndQueueLocked(ctx)
+}
+
 // Start implements controller.Controller.
 func (c *Controller[request]) Start(ctx context.Context) error {
 	// use an IIFE to get proper lock handling
@@ -158,17 +254,6 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 	// Set the internal context.
 	c.ctx = ctx
 
-	queue := c.NewQueue(c.Name, c.RateLimiter)
-	if priorityQueue, isPriorityQueue := queue.(priorityqueue.PriorityQueue[request]); isPriorityQueue {
-		c.Queue = priorityQueue
-	} else {
-		c.Queue = &priorityQueueWrapper[request]{TypedRateLimitingInterface: queue}
-	}
-	go func() {
-		<-ctx.Done()
-		c.Queue.ShutDown()
-	}()
-
 	wg := &sync.WaitGroup{}
 	err := func() error {
 		defer c.mu.Unlock()
@@ -179,18 +264,12 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 		// NB(directxman12): launch the sources *before* trying to wait for the
 		// caches to sync so that they have a chance to register their intended
 		// caches.
-		if err := c.startEventSources(ctx); err != nil {
+		if err := c.startEventSourcesAndQueueLocked(ctx); err != nil {
 			return err
 		}
 
 		c.LogConstructor(nil).Info("Starting Controller")
 
-		// All the watches have been started, we can reset the local slice.
-		//
-		// We should never hold watches more than necessary, each watch source can hold a backing cache,
-		// which won't be garbage collected if we hold a reference to it.
-		c.startWatches = nil
-
 		// Launch workers to process resources
 		c.LogConstructor(nil).Info("Starting workers", "worker count", c.MaxConcurrentReconciles)
 		wg.Add(c.MaxConcurrentReconciles)
@@ -218,63 +297,90 @@ func (c *Controller[request]) Start(ctx context.Context) error {
 	return nil
 }
 
-// startEventSources launches all the sources registered with this controller and waits
+// startEventSourcesAndQueueLocked launches all the sources registered with this controller and waits
 // for them to sync. It returns an error if any of the sources fail to start or sync.
-func (c *Controller[request]) startEventSources(ctx context.Context) error {
-	errGroup := &errgroup.Group{}
-	for _, watch := range c.startWatches {
-		log := c.LogConstructor(nil)
-		_, ok := watch.(interface {
-			String() string
-		})
-
-		if !ok {
-			log = log.WithValues("source", fmt.Sprintf("%T", watch))
+func (c *Controller[request]) startEventSourcesAndQueueLocked(ctx context.Context) error {
+	var retErr error
+
+	c.didStartEventSourcesOnce.Do(func() {
+		queue := c.NewQueue(c.Name, c.RateLimiter)
+		if priorityQueue, isPriorityQueue := queue.(priorityqueue.PriorityQueue[request]); isPriorityQueue {
+			c.Queue = priorityQueue
 		} else {
-			log = log.WithValues("source", fmt.Sprintf("%s", watch))
+			c.Queue = &priorityQueueWrapper[request]{TypedRateLimitingInterface: queue}
 		}
-		didStartSyncingSource := &atomic.Bool{}
-		errGroup.Go(func() error {
-			// Use a timeout for starting and syncing the source to avoid silently
-			// blocking startup indefinitely if it doesn't come up.
-			sourceStartCtx, cancel := context.WithTimeout(ctx, c.CacheSyncTimeout)
-			defer cancel()
-
-			sourceStartErrChan := make(chan error, 1) // Buffer chan to not leak goroutine if we time out
-			go func() {
-				defer close(sourceStartErrChan)
-				log.Info("Starting EventSource")
-				if err := watch.Start(ctx, c.Queue); err != nil {
-					sourceStartErrChan <- err
-					return
-				}
-				syncingSource, ok := watch.(source.TypedSyncingSource[request])
-				if !ok {
-					return
-				}
-				didStartSyncingSource.Store(true)
-				if err := syncingSource.WaitForSync(sourceStartCtx); err != nil {
-					err := fmt.Errorf("failed to wait for %s caches to sync %v: %w", c.Name, syncingSource, err)
-					log.Error(err, "Could not wait for Cache to sync")
-					sourceStartErrChan <- err
+		go func() {
+			<-ctx.Done()
+			c.Queue.ShutDown()
+		}()
+
+		errGroup := &errgroup.Group{}
+		for _, watch := range c.startWatches {
+			log := c.LogConstructor(nil)
+			_, ok := watch.(interface {
+				String() string
+			})
+			if !ok {
+				log = log.WithValues("source", fmt.Sprintf("%T", watch))
+			} else {
+				log = log.WithValues("source", fmt.Sprintf("%s", watch))
+			}
+			didStartSyncingSource := &atomic.Bool{}
+			errGroup.Go(func() error {
+				// Use a timeout for starting and syncing the source to avoid silently
+				// blocking startup indefinitely if it doesn't come up.
+				sourceStartCtx, cancel := context.WithTimeout(ctx, c.CacheSyncTimeout)
+				defer cancel()
+
+				sourceStartErrChan := make(chan error, 1) // Buffer chan to not leak goroutine if we time out
+				go func() {
+					defer close(sourceStartErrChan)
+					log.Info("Starting EventSource")
+
+					if err := watch.Start(ctx, c.Queue); err != nil {
+						sourceStartErrChan <- err
+						return
+					}
+					syncingSource, ok := watch.(source.TypedSyncingSource[request])
+					if !ok {
+						return
+					}
+					didStartSyncingSource.Store(true)
+					if err := syncingSource.WaitForSync(sourceStartCtx); err != nil {
+						err := fmt.Errorf("failed to wait for %s caches to sync %v: %w", c.Name, syncingSource, err)
+						log.Error(err, "Could not wait for Cache to sync")
+						sourceStartErrChan <- err
+					}
+				}()
+
+				select {
+				case err := <-sourceStartErrChan:
+					return err
+				case <-sourceStartCtx.Done():
+					if didStartSyncingSource.Load() { // We are racing with WaitForSync, wait for it to let it tell us what happened
+						return <-sourceStartErrChan
+					}
+					if ctx.Err() != nil { // Don't return an error if the root context got cancelled
+						return nil
+					}
+					return fmt.Errorf("timed out waiting for source %s to Start. Please ensure that its Start() method is non-blocking", watch)
 				}
-			}()
+			})
+		}
+		retErr = errGroup.Wait()
 
-			select {
-			case err := <-sourceStartErrChan:
-				return err
-			case <-sourceStartCtx.Done():
-				if didStartSyncingSource.Load() { // We are racing with WaitForSync, wait for it to let it tell us what happened
-					return <-sourceStartErrChan
-				}
-				if ctx.Err() != nil { // Don't return an error if the root context got cancelled
-					return nil
-				}
-				return fmt.Errorf("timed out waiting for source %s to Start. Please ensure that its Start() method is non-blocking", watch)
-			}
-		})
-	}
-	return errGroup.Wait()
+		// All the watches have been started, we can reset the local slice.
+		//
+		// We should never hold watches more than necessary, each watch source can hold a backing cache,
+		// which won't be garbage collected if we hold a reference to it.
+		c.startWatches = nil
+
+		// Mark event sources as started after resetting the startWatches slice so that watches from
+		// a new Watch() call are immediately started.
+		c.startedEventSourcesAndQueue = true
+	})
+
+	return retErr
 }
 
 // processNextWorkItem will read a single work item off the workqueue and
diff --git a/pkg/leaderelection/fake/leader_election.go b/pkg/leaderelection/fake/leader_election.go
index 5a82cf43..ab816a19 100644
--- a/pkg/leaderelection/fake/leader_election.go
+++ b/pkg/leaderelection/fake/leader_election.go
@@ -19,7 +19,9 @@ package fake
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"os"
+	"sync/atomic"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -30,6 +32,19 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
 )
 
+// ControllableResourceLockInterface is an interface that extends resourcelock.Interface to be
+// controllable.
+type ControllableResourceLockInterface interface {
+	resourcelock.Interface
+
+	// BlockLeaderElection blocks the leader election process when called. It will not be unblocked
+	// until UnblockLeaderElection is called.
+	BlockLeaderElection()
+
+	// UnblockLeaderElection unblocks the leader election.
+	UnblockLeaderElection()
+}
+
 // NewResourceLock creates a new ResourceLock for use in testing
 // leader election.
 func NewResourceLock(config *rest.Config, recorderProvider recorder.Provider, options leaderelection.Options) (resourcelock.Interface, error) {
@@ -40,27 +55,31 @@ func NewResourceLock(config *rest.Config, recorderProvider recorder.Provider, op
 	}
 	id = id + "_" + string(uuid.NewUUID())
 
-	return &ResourceLock{
+	return &resourceLock{
 		id: id,
 		record: resourcelock.LeaderElectionRecord{
 			HolderIdentity:       id,
-			LeaseDurationSeconds: 15,
+			LeaseDurationSeconds: 1,
 			AcquireTime:          metav1.NewTime(time.Now()),
-			RenewTime:            metav1.NewTime(time.Now().Add(15 * time.Second)),
+			RenewTime:            metav1.NewTime(time.Now().Add(1 * time.Second)),
 			LeaderTransitions:    1,
 		},
 	}, nil
 }
 
-// ResourceLock implements the ResourceLockInterface.
+var _ ControllableResourceLockInterface = &resourceLock{}
+
+// resourceLock implements the ResourceLockInterface.
 // By default returns that the current identity holds the lock.
-type ResourceLock struct {
+type resourceLock struct {
 	id     string
 	record resourcelock.LeaderElectionRecord
+
+	blockedLeaderElection atomic.Bool
 }
 
 // Get implements the ResourceLockInterface.
-func (f *ResourceLock) Get(ctx context.Context) (*resourcelock.LeaderElectionRecord, []byte, error) {
+func (f *resourceLock) Get(ctx context.Context) (*resourcelock.LeaderElectionRecord, []byte, error) {
 	recordBytes, err := json.Marshal(f.record)
 	if err != nil {
 		return nil, nil, err
@@ -69,28 +88,49 @@ func (f *ResourceLock) Get(ctx context.Context) (*resourcelock.LeaderElectionRec
 }
 
 // Create implements the ResourceLockInterface.
-func (f *ResourceLock) Create(ctx context.Context, ler resourcelock.LeaderElectionRecord) error {
+func (f *resourceLock) Create(ctx context.Context, ler resourcelock.LeaderElectionRecord) error {
+	if f.blockedLeaderElection.Load() {
+		// If leader election is blocked, we do not allow creating a new record.
+		return errors.New("leader election is blocked, cannot create new record")
+	}
+
 	f.record = ler
 	return nil
 }
 
 // Update implements the ResourceLockInterface.
-func (f *ResourceLock) Update(ctx context.Context, ler resourcelock.LeaderElectionRecord) error {
+func (f *resourceLock) Update(ctx context.Context, ler resourcelock.LeaderElectionRecord) error {
+	if f.blockedLeaderElection.Load() {
+		// If leader election is blocked, we do not allow updating records
+		return errors.New("leader election is blocked, cannot update record")
+	}
+
 	f.record = ler
+
 	return nil
 }
 
 // RecordEvent implements the ResourceLockInterface.
-func (f *ResourceLock) RecordEvent(s string) {
+func (f *resourceLock) RecordEvent(s string) {
 
 }
 
 // Identity implements the ResourceLockInterface.
-func (f *ResourceLock) Identity() string {
+func (f *resourceLock) Identity() string {
 	return f.id
 }
 
 // Describe implements the ResourceLockInterface.
-func (f *ResourceLock) Describe() string {
+func (f *resourceLock) Describe() string {
 	return f.id
 }
+
+// BlockLeaderElection blocks the leader election process when called.
+func (f *resourceLock) BlockLeaderElection() {
+	f.blockedLeaderElection.Store(true)
+}
+
+// UnblockLeaderElection blocks the leader election process when called.
+func (f *resourceLock) UnblockLeaderElection() {
+	f.blockedLeaderElection.Store(false)
+}
diff --git a/pkg/manager/internal.go b/pkg/manager/internal.go
index e5204a75..a9f91cbd 100644
--- a/pkg/manager/internal.go
+++ b/pkg/manager/internal.go
@@ -439,6 +439,11 @@ func (cm *controllerManager) Start(ctx context.Context) (err error) {
 		return fmt.Errorf("failed to start other runnables: %w", err)
 	}
 
+	// Start WarmupRunnables and wait for warmup to complete.
+	if err := cm.runnables.Warmup.Start(cm.internalCtx); err != nil {
+		return fmt.Errorf("failed to start warmup runnables: %w", err)
+	}
+
 	// Start the leader election and all required runnables.
 	{
 		ctx, cancel := context.WithCancel(context.Background())
@@ -534,6 +539,18 @@ func (cm *controllerManager) engageStopProcedure(stopComplete <-chan struct{}) e
 	}()
 
 	go func() {
+		go func() {
+			// Stop the warmup runnables in a separate goroutine to avoid blocking.
+			// It is important to stop the warmup runnables in parallel with the other runnables
+			// since we cannot assume ordering of whether or not one of the warmup runnables or one
+			// of the other runnables is holding a lock.
+			// Cancelling the wrong runnable (one that is not holding the lock) will cause the
+			// shutdown sequence to block indefinitely as it will wait for the runnable that is
+			// holding the lock to finish.
+			cm.logger.Info("Stopping and waiting for warmup runnables")
+			cm.runnables.Warmup.StopAndWait(cm.shutdownCtx)
+		}()
+
 		// First stop the non-leader election runnables.
 		cm.logger.Info("Stopping and waiting for non leader election runnables")
 		cm.runnables.Others.StopAndWait(cm.shutdownCtx)
diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go
index a10e0d51..e0e94245 100644
--- a/pkg/manager/manager.go
+++ b/pkg/manager/manager.go
@@ -319,6 +319,15 @@ type LeaderElectionRunnable interface {
 	NeedLeaderElection() bool
 }
 
+// warmupRunnable knows if a Runnable requires warmup. A warmup runnable is a runnable
+// that should be run when the manager is started but before it becomes leader.
+// Note: Implementing this interface is only useful when LeaderElection can be enabled, as the
+// behavior when leaderelection is not enabled is to run LeaderElectionRunnables immediately.
+type warmupRunnable interface {
+	// Warmup will be called when the manager is started but before it becomes leader.
+	Warmup(context.Context) error
+}
+
 // New returns a new Manager for creating Controllers.
 // Note that if ContentType in the given config is not set, "application/vnd.kubernetes.protobuf"
 // will be used for all built-in resources of Kubernetes, and "application/json" is for other types
@@ -423,7 +432,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 	}
 
 	errChan := make(chan error, 1)
-	runnables := newRunnables(options.BaseContext, errChan)
+	runnables := newRunnables(options.BaseContext, errChan).withLogger(options.Logger)
 	return &controllerManager{
 		stopProcedureEngaged:          ptr.To(int64(0)),
 		cluster:                       cluster,
diff --git a/pkg/manager/runnable_group.go b/pkg/manager/runnable_group.go
index db5cda7c..53e29fc5 100644
--- a/pkg/manager/runnable_group.go
+++ b/pkg/manager/runnable_group.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"sync"
 
+	"github.com/go-logr/logr"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 )
 
@@ -32,6 +33,7 @@ type runnables struct {
 	Webhooks       *runnableGroup
 	Caches         *runnableGroup
 	LeaderElection *runnableGroup
+	Warmup         *runnableGroup
 	Others         *runnableGroup
 }
 
@@ -42,10 +44,21 @@ func newRunnables(baseContext BaseContextFunc, errChan chan error) *runnables {
 		Webhooks:       newRunnableGroup(baseContext, errChan),
 		Caches:         newRunnableGroup(baseContext, errChan),
 		LeaderElection: newRunnableGroup(baseContext, errChan),
+		Warmup:         newRunnableGroup(baseContext, errChan),
 		Others:         newRunnableGroup(baseContext, errChan),
 	}
 }
 
+// withLogger returns the runnables with the logger set for all runnable groups.
+func (r *runnables) withLogger(logger logr.Logger) *runnables {
+	r.HTTPServers.withLogger(logger)
+	r.Webhooks.withLogger(logger)
+	r.Caches.withLogger(logger)
+	r.LeaderElection.withLogger(logger)
+	r.Others.withLogger(logger)
+	return r
+}
+
 // Add adds a runnable to closest group of runnable that they belong to.
 //
 // Add should be able to be called before and after Start, but not after StopAndWait.
@@ -65,8 +78,20 @@ func (r *runnables) Add(fn Runnable) error {
 		})
 	case webhook.Server:
 		return r.Webhooks.Add(fn, nil)
-	case LeaderElectionRunnable:
-		if !runnable.NeedLeaderElection() {
+	case warmupRunnable, LeaderElectionRunnable:
+		if warmupRunnable, ok := fn.(warmupRunnable); ok {
+			if err := r.Warmup.Add(RunnableFunc(warmupRunnable.Warmup), nil); err != nil {
+				return err
+			}
+		}
+
+		leaderElectionRunnable, ok := fn.(LeaderElectionRunnable)
+		if !ok {
+			// If the runnable is not a LeaderElectionRunnable, add it to the leader election group for backwards compatibility
+			return r.LeaderElection.Add(fn, nil)
+		}
+
+		if !leaderElectionRunnable.NeedLeaderElection() {
 			return r.Others.Add(fn, nil)
 		}
 		return r.LeaderElection.Add(fn, nil)
@@ -105,6 +130,9 @@ type runnableGroup struct {
 	// wg is an internal sync.WaitGroup that allows us to properly stop
 	// and wait for all the runnables to finish before returning.
 	wg *sync.WaitGroup
+
+	// logger is used for logging when errors are dropped during shutdown
+	logger logr.Logger
 }
 
 func newRunnableGroup(baseContext BaseContextFunc, errChan chan error) *runnableGroup {
@@ -113,12 +141,18 @@ func newRunnableGroup(baseContext BaseContextFunc, errChan chan error) *runnable
 		errChan:      errChan,
 		ch:           make(chan *readyRunnable),
 		wg:           new(sync.WaitGroup),
+		logger:       logr.Discard(), // Default to no-op logger
 	}
 
 	r.ctx, r.cancel = context.WithCancel(baseContext())
 	return r
 }
 
+// withLogger sets the logger for this runnable group.
+func (r *runnableGroup) withLogger(logger logr.Logger) {
+	r.logger = logger
+}
+
 // Started returns true if the group has started.
 func (r *runnableGroup) Started() bool {
 	r.start.Lock()
@@ -224,7 +258,27 @@ func (r *runnableGroup) reconcile() {
 
 			// Start the runnable.
 			if err := rn.Start(r.ctx); err != nil {
-				r.errChan <- err
+				// Check if we're during the shutdown process.
+				r.stop.RLock()
+				isStopped := r.stopped
+				r.stop.RUnlock()
+
+				if isStopped {
+					// During shutdown, try to send error first (error drain goroutine might still be running)
+					// but drop if it would block to prevent goroutine leaks
+					select {
+					case r.errChan <- err:
+						// Error sent successfully (error drain goroutine is still running)
+					default:
+						// Error drain goroutine has exited, drop error to prevent goroutine leak
+						if !errors.Is(err, context.Canceled) { // don't log context.Canceled errors as they are expected during shutdown
+							r.logger.Info("error dropped during shutdown to prevent goroutine leak", "error", err)
+						}
+					}
+				} else {
+					// During normal operation, always try to send errors (may block briefly)
+					r.errChan <- err
+				}
 			}
 		}(runnable)
 	}



Test patch:
diff --git a/pkg/client/client_test.go b/pkg/client/client_test.go
index 01075f12..4086845c 100644
--- a/pkg/client/client_test.go
+++ b/pkg/client/client_test.go
@@ -43,6 +43,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	corev1applyconfigurations "k8s.io/client-go/applyconfigurations/core/v1"
 	kscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/utils/ptr"
@@ -859,6 +860,101 @@ U5wwSivyi7vmegHKmblOzNVKA5qPO8zWzqBC
 		})
 	})
 
+	Describe("Apply", func() {
+		Context("Unstructured Client", func() {
+			It("should create and update a configMap using SSA", func() {
+				cl, err := client.New(cfg, client.Options{})
+				Expect(err).NotTo(HaveOccurred())
+				Expect(cl).NotTo(BeNil())
+
+				data := map[string]any{
+					"some-key": "some-value",
+				}
+				obj := &unstructured.Unstructured{Object: map[string]any{
+					"apiVersion": "v1",
+					"kind":       "ConfigMap",
+					"metadata": map[string]any{
+						"name":      "test-configmap",
+						"namespace": "default",
+					},
+					"data": data,
+				}}
+
+				err = cl.Apply(context.Background(), client.ApplyConfigurationFromUnstructured(obj), &client.ApplyOptions{FieldManager: "test-manager"})
+				Expect(err).NotTo(HaveOccurred())
+
+				cm, err := clientset.CoreV1().ConfigMaps(obj.GetNamespace()).Get(context.Background(), obj.GetName(), metav1.GetOptions{})
+				Expect(err).NotTo(HaveOccurred())
+
+				actualData := map[string]any{}
+				for k, v := range cm.Data {
+					actualData[k] = v
+				}
+
+				Expect(actualData).To(BeComparableTo(data))
+				Expect(actualData).To(BeComparableTo(obj.Object["data"]))
+
+				data = map[string]any{
+					"a-new-key": "a-new-value",
+				}
+				obj.Object["data"] = data
+				unstructured.RemoveNestedField(obj.Object, "metadata", "managedFields")
+
+				err = cl.Apply(context.Background(), client.ApplyConfigurationFromUnstructured(obj), &client.ApplyOptions{FieldManager: "test-manager"})
+				Expect(err).NotTo(HaveOccurred())
+
+				cm, err = clientset.CoreV1().ConfigMaps(obj.GetNamespace()).Get(context.Background(), obj.GetName(), metav1.GetOptions{})
+				Expect(err).NotTo(HaveOccurred())
+
+				actualData = map[string]any{}
+				for k, v := range cm.Data {
+					actualData[k] = v
+				}
+
+				Expect(actualData).To(BeComparableTo(data))
+				Expect(actualData).To(BeComparableTo(obj.Object["data"]))
+			})
+		})
+
+		Context("Structured Client", func() {
+			It("should create and update a configMap using SSA", func() {
+				cl, err := client.New(cfg, client.Options{})
+				Expect(err).NotTo(HaveOccurred())
+				Expect(cl).NotTo(BeNil())
+
+				data := map[string]string{
+					"some-key": "some-value",
+				}
+				obj := corev1applyconfigurations.
+					ConfigMap("test-configmap", "default").
+					WithData(data)
+
+				err = cl.Apply(context.Background(), obj, &client.ApplyOptions{FieldManager: "test-manager"})
+				Expect(err).NotTo(HaveOccurred())
+
+				cm, err := clientset.CoreV1().ConfigMaps(ptr.Deref(obj.GetNamespace(), "")).Get(context.Background(), ptr.Deref(obj.GetName(), ""), metav1.GetOptions{})
+				Expect(err).NotTo(HaveOccurred())
+
+				Expect(cm.Data).To(BeComparableTo(data))
+				Expect(cm.Data).To(BeComparableTo(obj.Data))
+
+				data = map[string]string{
+					"a-new-key": "a-new-value",
+				}
+				obj.Data = data
+
+				err = cl.Apply(context.Background(), obj, &client.ApplyOptions{FieldManager: "test-manager"})
+				Expect(err).NotTo(HaveOccurred())
+
+				cm, err = clientset.CoreV1().ConfigMaps(ptr.Deref(obj.GetNamespace(), "")).Get(context.Background(), ptr.Deref(obj.GetName(), ""), metav1.GetOptions{})
+				Expect(err).NotTo(HaveOccurred())
+
+				Expect(cm.Data).To(BeComparableTo(data))
+				Expect(cm.Data).To(BeComparableTo(obj.Data))
+			})
+		})
+	})
+
 	Describe("SubResourceClient", func() {
 		Context("with structured objects", func() {
 			It("should be able to read the Scale subresource", func() {
diff --git a/pkg/client/fake/client_test.go b/pkg/client/fake/client_test.go
index 707d74ad..c20581d9 100644
--- a/pkg/client/fake/client_test.go
+++ b/pkg/client/fake/client_test.go
@@ -45,6 +45,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 	clientgoapplyconfigurations "k8s.io/client-go/applyconfigurations"
+	corev1applyconfigurations "k8s.io/client-go/applyconfigurations/core/v1"
 	"k8s.io/client-go/kubernetes/fake"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/utils/ptr"
@@ -2548,7 +2549,7 @@ var _ = Describe("Fake client", func() {
 		obj.SetName("foo")
 		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"some": "data"}, "data")).To(Succeed())
 
-		Expect(cl.Patch(context.Background(), obj, client.Apply)).To(Succeed())
+		Expect(cl.Patch(context.Background(), obj, client.Apply, client.FieldOwner("foo"))).To(Succeed())
 
 		cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "foo"}}
 
@@ -2556,7 +2557,7 @@ var _ = Describe("Fake client", func() {
 		Expect(cm.Data).To(Equal(map[string]string{"some": "data"}))
 
 		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"other": "data"}, "data")).To(Succeed())
-		Expect(cl.Patch(context.Background(), obj, client.Apply)).To(Succeed())
+		Expect(cl.Patch(context.Background(), obj, client.Apply, client.FieldOwner("foo"))).To(Succeed())
 
 		Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(cm), cm)).To(Succeed())
 		Expect(cm.Data).To(Equal(map[string]string{"other": "data"}))
@@ -2572,13 +2573,13 @@ var _ = Describe("Fake client", func() {
 
 		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"some": "data"}, "spec")).To(Succeed())
 
-		Expect(cl.Patch(context.Background(), obj, client.Apply)).To(Succeed())
+		Expect(cl.Patch(context.Background(), obj, client.Apply, client.FieldOwner("foo"))).To(Succeed())
 
 		Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(result), result)).To(Succeed())
 		Expect(result.Object["spec"]).To(Equal(map[string]any{"some": "data"}))
 
 		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"other": "data"}, "spec")).To(Succeed())
-		Expect(cl.Patch(context.Background(), obj, client.Apply)).To(Succeed())
+		Expect(cl.Patch(context.Background(), obj, client.Apply, client.FieldOwner("foo"))).To(Succeed())
 
 		Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(result), result)).To(Succeed())
 		Expect(result.Object["spec"]).To(Equal(map[string]any{"other": "data"}))
@@ -2592,9 +2593,9 @@ var _ = Describe("Fake client", func() {
 		obj.SetName("foo")
 		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"some": "data"}, "data")).To(Succeed())
 
-		Expect(cl.Patch(context.Background(), obj, client.Apply)).To(Succeed())
+		Expect(cl.Patch(context.Background(), obj, client.Apply, client.FieldOwner("foo"))).To(Succeed())
 
-		err := cl.Patch(context.Background(), obj, client.Apply)
+		err := cl.Patch(context.Background(), obj, client.Apply, client.FieldOwner("foo"))
 		Expect(err).To(HaveOccurred())
 		Expect(err.Error()).To(ContainSubstring("metadata.managedFields must be nil"))
 	})
@@ -2610,7 +2611,7 @@ var _ = Describe("Fake client", func() {
 
 		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"some": "data"}, "data")).To(Succeed())
 
-		Expect(cl.Patch(context.Background(), obj, client.Apply)).To(Succeed())
+		Expect(cl.Patch(context.Background(), obj, client.Apply, client.FieldOwner("foo"))).To(Succeed())
 
 		cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "foo"}}
 
@@ -2618,7 +2619,7 @@ var _ = Describe("Fake client", func() {
 		Expect(cm.Data).To(Equal(map[string]string{"some": "data"}))
 
 		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"other": "data"}, "data")).To(Succeed())
-		Expect(cl.Patch(context.Background(), obj, client.Apply)).To(Succeed())
+		Expect(cl.Patch(context.Background(), obj, client.Apply, client.FieldOwner("foo"))).To(Succeed())
 
 		Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(cm), cm)).To(Succeed())
 		Expect(cm.Data).To(Equal(map[string]string{"other": "data"}))
@@ -2664,12 +2665,78 @@ var _ = Describe("Fake client", func() {
 				"ssa": "value",
 			},
 		}}
-		Expect(cl.Patch(context.Background(), u, client.Apply)).NotTo(HaveOccurred())
+		Expect(cl.Patch(context.Background(), u, client.Apply, client.FieldOwner("foo"))).NotTo(HaveOccurred())
 		_, exists, err := unstructured.NestedFieldNoCopy(u.Object, "metadata", "managedFields")
 		Expect(err).NotTo(HaveOccurred())
 		Expect(exists).To(BeTrue())
 	})
 
+	It("supports server-side apply of a client-go resource via Apply method", func() {
+		cl := NewClientBuilder().Build()
+		obj := corev1applyconfigurations.
+			ConfigMap("foo", "default").
+			WithData(map[string]string{"some": "data"})
+
+		Expect(cl.Apply(context.Background(), obj, &client.ApplyOptions{FieldManager: "test-manager"})).To(Succeed())
+
+		cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "default"}}
+
+		Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(cm), cm)).To(Succeed())
+		Expect(cm.Data).To(BeComparableTo(map[string]string{"some": "data"}))
+
+		obj.Data = map[string]string{"other": "data"}
+		Expect(cl.Apply(context.Background(), obj, &client.ApplyOptions{FieldManager: "test-manager"})).To(Succeed())
+
+		Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(cm), cm)).To(Succeed())
+		Expect(cm.Data).To(BeComparableTo(map[string]string{"other": "data"}))
+	})
+
+	It("errors when trying to server-side apply an object without configuring a FieldManager", func() {
+		cl := NewClientBuilder().Build()
+		obj := corev1applyconfigurations.
+			ConfigMap("foo", "default").
+			WithData(map[string]string{"some": "data"})
+
+		err := cl.Apply(context.Background(), obj)
+		Expect(err).To(HaveOccurred())
+		Expect(apierrors.IsInvalid(err)).To(BeTrue(), "Expected error to be an invalid error")
+	})
+
+	It("errors when trying to server-side apply an object with an invalid FieldManager", func() {
+		cl := NewClientBuilder().Build()
+		obj := corev1applyconfigurations.
+			ConfigMap("foo", "default").
+			WithData(map[string]string{"some": "data"})
+
+		err := cl.Apply(context.Background(), obj, client.FieldOwner("\x00"))
+		Expect(err).To(HaveOccurred())
+		Expect(apierrors.IsInvalid(err)).To(BeTrue(), "Expected error to be an invalid error")
+	})
+
+	It("supports server-side apply of a custom resource via Apply method", func() {
+		cl := NewClientBuilder().Build()
+		obj := &unstructured.Unstructured{}
+		obj.SetAPIVersion("custom/v1")
+		obj.SetKind("FakeResource")
+		obj.SetName("foo")
+		result := obj.DeepCopy()
+
+		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"some": "data"}, "spec")).To(Succeed())
+
+		applyConfig := client.ApplyConfigurationFromUnstructured(obj)
+		Expect(cl.Apply(context.Background(), applyConfig, &client.ApplyOptions{FieldManager: "test-manager"})).To(Succeed())
+
+		Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(result), result)).To(Succeed())
+		Expect(result.Object["spec"]).To(Equal(map[string]any{"some": "data"}))
+
+		Expect(unstructured.SetNestedField(obj.Object, map[string]any{"other": "data"}, "spec")).To(Succeed())
+		applyConfig2 := client.ApplyConfigurationFromUnstructured(obj)
+		Expect(cl.Apply(context.Background(), applyConfig2, &client.ApplyOptions{FieldManager: "test-manager"})).To(Succeed())
+
+		Expect(cl.Get(context.Background(), client.ObjectKeyFromObject(result), result)).To(Succeed())
+		Expect(result.Object["spec"]).To(Equal(map[string]any{"other": "data"}))
+	})
+
 	scalableObjs := []client.Object{
 		&appsv1.Deployment{
 			ObjectMeta: metav1.ObjectMeta{
diff --git a/pkg/client/fake/typeconverter_test.go b/pkg/client/fake/typeconverter_test.go
index 3bc56baa..8acba79f 100644
--- a/pkg/client/fake/typeconverter_test.go
+++ b/pkg/client/fake/typeconverter_test.go
@@ -24,7 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/managedfields"
-	"sigs.k8s.io/structured-merge-diff/v4/typed"
+	"sigs.k8s.io/structured-merge-diff/v6/typed"
 )
 
 var _ = Describe("multiTypeConverter", func() {
diff --git a/pkg/client/fieldvalidation_test.go b/pkg/client/fieldvalidation_test.go
index 4d06e5d9..17395790 100644
--- a/pkg/client/fieldvalidation_test.go
+++ b/pkg/client/fieldvalidation_test.go
@@ -25,7 +25,9 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	corev1applyconfigurations "k8s.io/client-go/applyconfigurations/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
@@ -104,6 +106,7 @@ func TestWithStrictFieldValidation(t *testing.T) {
 
 	_ = wrappedClient.Create(ctx, dummyObj)
 	_ = wrappedClient.Update(ctx, dummyObj)
+	_ = wrappedClient.Apply(ctx, corev1applyconfigurations.ConfigMap("foo", "bar"))
 	_ = wrappedClient.Patch(ctx, dummyObj, nil)
 	_ = wrappedClient.Status().Create(ctx, dummyObj, dummyObj)
 	_ = wrappedClient.Status().Update(ctx, dummyObj)
@@ -112,7 +115,7 @@ func TestWithStrictFieldValidation(t *testing.T) {
 	_ = wrappedClient.SubResource("some-subresource").Update(ctx, dummyObj)
 	_ = wrappedClient.SubResource("some-subresource").Patch(ctx, dummyObj, nil)
 
-	if expectedCalls := 9; calls != expectedCalls {
+	if expectedCalls := 10; calls != expectedCalls {
 		t.Fatalf("wrong number of calls to assertions: expected=%d; got=%d", expectedCalls, calls)
 	}
 }
@@ -188,6 +191,10 @@ func testFieldValidationClient(t *testing.T, expectedFieldValidation string, cal
 			}
 			return nil
 		},
+		Apply: func(ctx context.Context, client client.WithWatch, obj runtime.ApplyConfiguration, opts ...client.ApplyOption) error {
+			callback()
+			return nil
+		},
 		Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
 			callback()
 			out := &client.PatchOptions{}
diff --git a/pkg/client/interceptor/intercept_test.go b/pkg/client/interceptor/intercept_test.go
index a0536789..c549b4fc 100644
--- a/pkg/client/interceptor/intercept_test.go
+++ b/pkg/client/interceptor/intercept_test.go
@@ -63,6 +63,29 @@ var _ = Describe("NewClient", func() {
 		_ = client2.List(ctx, nil)
 		Expect(called).To(BeTrue())
 	})
+	It("should call the provided Apply function", func() {
+		var called bool
+		client := NewClient(wrappedClient, Funcs{
+			Apply: func(ctx context.Context, client client.WithWatch, obj runtime.ApplyConfiguration, opts ...client.ApplyOption) error {
+				called = true
+				return nil
+			},
+		})
+		_ = client.Apply(ctx, nil)
+		Expect(called).To(BeTrue())
+	})
+	It("should call the underlying client if the provided Apply function is nil", func() {
+		var called bool
+		client1 := NewClient(wrappedClient, Funcs{
+			Apply: func(ctx context.Context, client client.WithWatch, obj runtime.ApplyConfiguration, opts ...client.ApplyOption) error {
+				called = true
+				return nil
+			},
+		})
+		client2 := NewClient(client1, Funcs{})
+		_ = client2.Apply(ctx, nil)
+		Expect(called).To(BeTrue())
+	})
 	It("should call the provided Create function", func() {
 		var called bool
 		client := NewClient(wrappedClient, Funcs{
@@ -360,6 +383,10 @@ func (d dummyClient) Patch(ctx context.Context, obj client.Object, patch client.
 	return nil
 }
 
+func (d dummyClient) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...client.ApplyOption) error {
+	return nil
+}
+
 func (d dummyClient) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error {
 	return nil
 }
diff --git a/pkg/client/namespaced_client_test.go b/pkg/client/namespaced_client_test.go
index 6e1c2641..d7e18d0f 100644
--- a/pkg/client/namespaced_client_test.go
+++ b/pkg/client/namespaced_client_test.go
@@ -25,19 +25,25 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
-	rbacv1 "k8s.io/api/rbac/v1"
-
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	appsv1applyconfigurations "k8s.io/client-go/applyconfigurations/apps/v1"
+	corev1applyconfigurations "k8s.io/client-go/applyconfigurations/core/v1"
+	metav1applyconfigurations "k8s.io/client-go/applyconfigurations/meta/v1"
+	rbacv1applyconfigurations "k8s.io/client-go/applyconfigurations/rbac/v1"
+
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 var _ = Describe("NamespacedClient", func() {
 	var dep *appsv1.Deployment
+	var acDep *appsv1applyconfigurations.DeploymentApplyConfiguration
 	var ns = "default"
 	ctx := context.Background()
 	var count uint64 = 0
@@ -75,10 +81,25 @@ var _ = Describe("NamespacedClient", func() {
 				},
 			},
 		}
+		acDep = appsv1applyconfigurations.Deployment(dep.Name, "").
+			WithLabels(dep.Labels).
+			WithSpec(appsv1applyconfigurations.DeploymentSpec().
+				WithReplicas(*dep.Spec.Replicas).
+				WithSelector(metav1applyconfigurations.LabelSelector().WithMatchLabels(dep.Spec.Selector.MatchLabels)).
+				WithTemplate(corev1applyconfigurations.PodTemplateSpec().
+					WithLabels(dep.Spec.Template.Labels).
+					WithSpec(corev1applyconfigurations.PodSpec().
+						WithContainers(corev1applyconfigurations.Container().
+							WithName(dep.Spec.Template.Spec.Containers[0].Name).
+							WithImage(dep.Spec.Template.Spec.Containers[0].Image),
+						),
+					),
+				),
+			)
+
 	})
 
 	Describe("Get", func() {
-
 		BeforeEach(func() {
 			var err error
 			dep, err = clientset.AppsV1().Deployments(ns).Create(ctx, dep, metav1.CreateOptions{})
@@ -88,6 +109,7 @@ var _ = Describe("NamespacedClient", func() {
 		AfterEach(func() {
 			deleteDeployment(ctx, dep, ns)
 		})
+
 		It("should successfully Get a namespace-scoped object", func() {
 			name := types.NamespacedName{Name: dep.Name}
 			result := &appsv1.Deployment{}
@@ -135,6 +157,66 @@ var _ = Describe("NamespacedClient", func() {
 		})
 	})
 
+	Describe("Apply", func() {
+		AfterEach(func() {
+			deleteDeployment(ctx, dep, ns)
+		})
+
+		It("should successfully apply an object in the right namespace", func() {
+			err := getClient().Apply(ctx, acDep, client.FieldOwner("test"))
+			Expect(err).NotTo(HaveOccurred())
+
+			res, err := clientset.AppsV1().Deployments(ns).Get(ctx, dep.Name, metav1.GetOptions{})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(res.GetNamespace()).To(BeEquivalentTo(ns))
+		})
+
+		It("should successfully apply an object in the right namespace through unstructured", func() {
+			serialized, err := json.Marshal(acDep)
+			Expect(err).NotTo(HaveOccurred())
+			u := &unstructured.Unstructured{}
+			Expect(json.Unmarshal(serialized, &u.Object)).To(Succeed())
+			err = getClient().Apply(ctx, client.ApplyConfigurationFromUnstructured(u), client.FieldOwner("test"))
+			Expect(err).NotTo(HaveOccurred())
+
+			res, err := clientset.AppsV1().Deployments(ns).Get(ctx, dep.Name, metav1.GetOptions{})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(res.GetNamespace()).To(BeEquivalentTo(ns))
+		})
+
+		It("should not create an object if the namespace of the object is different", func() {
+			acDep.WithNamespace("non-default")
+			err := getClient().Apply(ctx, acDep, client.FieldOwner("test"))
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("does not match the namespace"))
+		})
+
+		It("should not create an object through unstructured if the namespace of the object is different", func() {
+			acDep.WithNamespace("non-default")
+			serialized, err := json.Marshal(acDep)
+			Expect(err).NotTo(HaveOccurred())
+			u := &unstructured.Unstructured{}
+			Expect(json.Unmarshal(serialized, &u.Object)).To(Succeed())
+			err = getClient().Apply(ctx, client.ApplyConfigurationFromUnstructured(u), client.FieldOwner("test"))
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("does not match the namespace"))
+		})
+
+		It("should create a cluster scoped object", func() {
+			cr := rbacv1applyconfigurations.ClusterRole(fmt.Sprintf("clusterRole-%v", count))
+
+			err := getClient().Apply(ctx, cr, client.FieldOwner("test"))
+			Expect(err).NotTo(HaveOccurred())
+
+			By("checking if the object was created")
+			res, err := clientset.RbacV1().ClusterRoles().Get(ctx, *cr.Name, metav1.GetOptions{})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(res).NotTo(BeNil())
+
+			deleteClusterRole(ctx, &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{Name: *cr.Name}})
+		})
+	})
+
 	Describe("Create", func() {
 		AfterEach(func() {
 			deleteDeployment(ctx, dep, ns)
diff --git a/pkg/client/options_test.go b/pkg/client/options_test.go
index 9ad092a2..e50669c6 100644
--- a/pkg/client/options_test.go
+++ b/pkg/client/options_test.go
@@ -109,6 +109,27 @@ var _ = Describe("GetOptions", func() {
 	})
 })
 
+var _ = Describe("ApplyOptions", func() {
+	It("Should set DryRun", func() {
+		o := &client.ApplyOptions{DryRun: []string{"Hello", "Theodore"}}
+		newApplyOpts := &client.ApplyOptions{}
+		o.ApplyToApply(newApplyOpts)
+		Expect(newApplyOpts).To(Equal(o))
+	})
+	It("Should set Force", func() {
+		o := &client.ApplyOptions{Force: ptr.To(true)}
+		newApplyOpts := &client.ApplyOptions{}
+		o.ApplyToApply(newApplyOpts)
+		Expect(newApplyOpts).To(Equal(o))
+	})
+	It("Should set FieldManager", func() {
+		o := &client.ApplyOptions{FieldManager: "field-manager"}
+		newApplyOpts := &client.ApplyOptions{}
+		o.ApplyToApply(newApplyOpts)
+		Expect(newApplyOpts).To(Equal(o))
+	})
+})
+
 var _ = Describe("CreateOptions", func() {
 	It("Should set DryRun", func() {
 		o := &client.CreateOptions{DryRun: []string{"Hello", "Theodore"}}
@@ -278,6 +299,57 @@ var _ = Describe("MatchingLabels", func() {
 	})
 })
 
+var _ = Describe("DryRunAll", func() {
+	It("Should apply to ApplyOptions", func() {
+		o := &client.ApplyOptions{DryRun: []string{"server"}}
+		t := client.DryRunAll
+		t.ApplyToApply(o)
+		Expect(o.DryRun).To(Equal([]string{metav1.DryRunAll}))
+	})
+	It("Should apply to CreateOptions", func() {
+		o := &client.CreateOptions{DryRun: []string{"server"}}
+		t := client.DryRunAll
+		t.ApplyToCreate(o)
+		Expect(o.DryRun).To(Equal([]string{metav1.DryRunAll}))
+	})
+	It("Should apply to UpdateOptions", func() {
+		o := &client.UpdateOptions{DryRun: []string{"server"}}
+		t := client.DryRunAll
+		t.ApplyToUpdate(o)
+		Expect(o.DryRun).To(Equal([]string{metav1.DryRunAll}))
+	})
+	It("Should apply to PatchOptions", func() {
+		o := &client.PatchOptions{DryRun: []string{"server"}}
+		t := client.DryRunAll
+		t.ApplyToPatch(o)
+		Expect(o.DryRun).To(Equal([]string{metav1.DryRunAll}))
+	})
+	It("Should apply to DeleteOptions", func() {
+		o := &client.DeleteOptions{DryRun: []string{"server"}}
+		t := client.DryRunAll
+		t.ApplyToDelete(o)
+		Expect(o.DryRun).To(Equal([]string{metav1.DryRunAll}))
+	})
+	It("Should apply to SubResourcePatchOptions", func() {
+		o := &client.SubResourcePatchOptions{PatchOptions: client.PatchOptions{DryRun: []string{"server"}}}
+		t := client.DryRunAll
+		t.ApplyToSubResourcePatch(o)
+		Expect(o.DryRun).To(Equal([]string{metav1.DryRunAll}))
+	})
+	It("Should apply to SubResourceCreateOptions", func() {
+		o := &client.SubResourceCreateOptions{CreateOptions: client.CreateOptions{DryRun: []string{"server"}}}
+		t := client.DryRunAll
+		t.ApplyToSubResourceCreate(o)
+		Expect(o.DryRun).To(Equal([]string{metav1.DryRunAll}))
+	})
+	It("Should apply to SubResourceUpdateOptions", func() {
+		o := &client.SubResourceUpdateOptions{UpdateOptions: client.UpdateOptions{DryRun: []string{"server"}}}
+		t := client.DryRunAll
+		t.ApplyToSubResourceUpdate(o)
+		Expect(o.DryRun).To(Equal([]string{metav1.DryRunAll}))
+	})
+})
+
 var _ = Describe("FieldOwner", func() {
 	It("Should apply to PatchOptions", func() {
 		o := &client.PatchOptions{FieldManager: "bar"}
@@ -285,6 +357,12 @@ var _ = Describe("FieldOwner", func() {
 		t.ApplyToPatch(o)
 		Expect(o.FieldManager).To(Equal("foo"))
 	})
+	It("Should apply to ApplyOptions", func() {
+		o := &client.ApplyOptions{FieldManager: "bar"}
+		t := client.FieldOwner("foo")
+		t.ApplyToApply(o)
+		Expect(o.FieldManager).To(Equal("foo"))
+	})
 	It("Should apply to CreateOptions", func() {
 		o := &client.CreateOptions{FieldManager: "bar"}
 		t := client.FieldOwner("foo")
@@ -330,6 +408,12 @@ var _ = Describe("ForceOwnership", func() {
 		t.ApplyToSubResourcePatch(o)
 		Expect(*o.Force).To(BeTrue())
 	})
+	It("Should apply to ApplyOptions", func() {
+		o := &client.ApplyOptions{}
+		t := client.ForceOwnership
+		t.ApplyToApply(o)
+		Expect(*o.Force).To(BeTrue())
+	})
 })
 
 var _ = Describe("HasLabels", func() {
diff --git a/pkg/controller/controller_integration_test.go b/pkg/controller/controller_integration_test.go
index c8e8a790..994a9a2c 100644
--- a/pkg/controller/controller_integration_test.go
+++ b/pkg/controller/controller_integration_test.go
@@ -18,16 +18,21 @@ package controller_test
 
 import (
 	"context"
+	"fmt"
+	"strconv"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllertest"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
@@ -48,30 +53,48 @@ var _ = Describe("controller", func() {
 	Describe("controller", func() {
 		// TODO(directxman12): write a whole suite of controller-client interaction tests
 
-		It("should reconcile", func() {
+		// The watches in this test are setup with a namespace predicate to avoid each table entry
+		// from interfering with the others. We cannot add a delete call for the pods created in the
+		// test, as it causes flakes with the api-server termination timing out.
+		// See https://github.com/kubernetes-sigs/controller-runtime/issues/1571 for a description
+		// of the issue, and a discussion here: https://github.com/kubernetes-sigs/controller-runtime/pull/3192#discussion_r2186967799
+		DescribeTable("should reconcile", func(enableWarmup bool) {
 			By("Creating the Manager")
 			cm, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
 
 			By("Creating the Controller")
-			instance, err := controller.New("foo-controller", cm, controller.Options{
-				Reconciler: reconcile.Func(
-					func(_ context.Context, request reconcile.Request) (reconcile.Result, error) {
-						reconciled <- request
-						return reconcile.Result{}, nil
-					}),
-			})
+			instance, err := controller.New(
+				fmt.Sprintf("foo-controller-%t", enableWarmup),
+				cm,
+				controller.Options{
+					Reconciler: reconcile.Func(
+						func(_ context.Context, request reconcile.Request) (reconcile.Result, error) {
+							reconciled <- request
+							return reconcile.Result{}, nil
+						}),
+					EnableWarmup: ptr.To(enableWarmup),
+				},
+			)
 			Expect(err).NotTo(HaveOccurred())
 
+			testNamespace := strconv.FormatBool(enableWarmup)
+
 			By("Watching Resources")
 			err = instance.Watch(
 				source.Kind(cm.GetCache(), &appsv1.ReplicaSet{},
 					handler.TypedEnqueueRequestForOwner[*appsv1.ReplicaSet](cm.GetScheme(), cm.GetRESTMapper(), &appsv1.Deployment{}),
+					makeNamespacePredicate[*appsv1.ReplicaSet](testNamespace),
 				),
 			)
 			Expect(err).NotTo(HaveOccurred())
 
-			err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}, &handler.TypedEnqueueRequestForObject[*appsv1.Deployment]{}))
+			err = instance.Watch(
+				source.Kind(cm.GetCache(), &appsv1.Deployment{},
+					&handler.TypedEnqueueRequestForObject[*appsv1.Deployment]{},
+					makeNamespacePredicate[*appsv1.Deployment](testNamespace),
+				),
+			)
 			Expect(err).NotTo(HaveOccurred())
 
 			err = cm.GetClient().Get(ctx, types.NamespacedName{Name: "foo"}, &corev1.Namespace{})
@@ -110,19 +133,25 @@ var _ = Describe("controller", func() {
 				},
 			}
 			expectedReconcileRequest := reconcile.Request{NamespacedName: types.NamespacedName{
-				Namespace: "default",
+				Namespace: testNamespace,
 				Name:      "deployment-name",
 			}}
 
+			By("Creating the test namespace")
+			_, err = clientset.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{Name: testNamespace},
+			}, metav1.CreateOptions{})
+			Expect(err).NotTo(HaveOccurred())
+
 			By("Invoking Reconciling for Create")
-			deployment, err = clientset.AppsV1().Deployments("default").Create(ctx, deployment, metav1.CreateOptions{})
+			deployment, err = clientset.AppsV1().Deployments(testNamespace).Create(ctx, deployment, metav1.CreateOptions{})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(<-reconciled).To(Equal(expectedReconcileRequest))
 
 			By("Invoking Reconciling for Update")
 			newDeployment := deployment.DeepCopy()
 			newDeployment.Labels = map[string]string{"foo": "bar"}
-			_, err = clientset.AppsV1().Deployments("default").Update(ctx, newDeployment, metav1.UpdateOptions{})
+			_, err = clientset.AppsV1().Deployments(testNamespace).Update(ctx, newDeployment, metav1.UpdateOptions{})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(<-reconciled).To(Equal(expectedReconcileRequest))
 
@@ -145,24 +174,24 @@ var _ = Describe("controller", func() {
 					Template: deployment.Spec.Template,
 				},
 			}
-			replicaset, err = clientset.AppsV1().ReplicaSets("default").Create(ctx, replicaset, metav1.CreateOptions{})
+			replicaset, err = clientset.AppsV1().ReplicaSets(testNamespace).Create(ctx, replicaset, metav1.CreateOptions{})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(<-reconciled).To(Equal(expectedReconcileRequest))
 
 			By("Invoking Reconciling for an OwnedObject when it is updated")
 			newReplicaset := replicaset.DeepCopy()
 			newReplicaset.Labels = map[string]string{"foo": "bar"}
-			_, err = clientset.AppsV1().ReplicaSets("default").Update(ctx, newReplicaset, metav1.UpdateOptions{})
+			_, err = clientset.AppsV1().ReplicaSets(testNamespace).Update(ctx, newReplicaset, metav1.UpdateOptions{})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(<-reconciled).To(Equal(expectedReconcileRequest))
 
 			By("Invoking Reconciling for an OwnedObject when it is deleted")
-			err = clientset.AppsV1().ReplicaSets("default").Delete(ctx, replicaset.Name, metav1.DeleteOptions{})
+			err = clientset.AppsV1().ReplicaSets(testNamespace).Delete(ctx, replicaset.Name, metav1.DeleteOptions{})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(<-reconciled).To(Equal(expectedReconcileRequest))
 
 			By("Invoking Reconciling for Delete")
-			err = clientset.AppsV1().Deployments("default").
+			err = clientset.AppsV1().Deployments(testNamespace).
 				Delete(ctx, "deployment-name", metav1.DeleteOptions{})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(<-reconciled).To(Equal(expectedReconcileRequest))
@@ -174,7 +203,12 @@ var _ = Describe("controller", func() {
 
 			By("Invoking Reconciling for a pod when it is created when adding watcher dynamically")
 			// Add new watcher dynamically
-			err = instance.Watch(source.Kind(cm.GetCache(), &corev1.Pod{}, &handler.TypedEnqueueRequestForObject[*corev1.Pod]{}))
+			err = instance.Watch(
+				source.Kind(cm.GetCache(), &corev1.Pod{},
+					&handler.TypedEnqueueRequestForObject[*corev1.Pod]{},
+					makeNamespacePredicate[*corev1.Pod](testNamespace),
+				),
+			)
 			Expect(err).NotTo(HaveOccurred())
 
 			pod := &corev1.Pod{
@@ -194,16 +228,27 @@ var _ = Describe("controller", func() {
 				},
 			}
 			expectedReconcileRequest = reconcile.Request{NamespacedName: types.NamespacedName{
-				Namespace: "default",
+				Namespace: testNamespace,
 				Name:      "pod-name",
 			}}
-			_, err = clientset.CoreV1().Pods("default").Create(ctx, pod, metav1.CreateOptions{})
+			_, err = clientset.CoreV1().Pods(testNamespace).Create(ctx, pod, metav1.CreateOptions{})
 			Expect(err).NotTo(HaveOccurred())
 			Expect(<-reconciled).To(Equal(expectedReconcileRequest))
-		})
+		},
+			Entry("with controller warmup enabled", true),
+			Entry("with controller warmup not enabled", false),
+		)
 	})
 })
 
+// makeNamespacePredicate returns a predicate that filters out all objects not in the passed in
+// namespace.
+func makeNamespacePredicate[object client.Object](namespace string) predicate.TypedPredicate[object] {
+	return predicate.NewTypedPredicateFuncs[object](func(obj object) bool {
+		return obj.GetNamespace() == namespace
+	})
+}
+
 func truePtr() *bool {
 	t := true
 	return &t
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index 1c5b11d7..244a9f17 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -474,5 +474,71 @@ var _ = Describe("controller.Controller", func() {
 			_, ok = q.(priorityqueue.PriorityQueue[reconcile.Request])
 			Expect(ok).To(BeFalse())
 		})
+
+		It("should set EnableWarmup correctly", func() {
+			m, err := manager.New(cfg, manager.Options{})
+			Expect(err).NotTo(HaveOccurred())
+
+			// Test with EnableWarmup set to true
+			ctrlWithWarmup, err := controller.New("warmup-enabled-ctrl", m, controller.Options{
+				Reconciler:   reconcile.Func(nil),
+				EnableWarmup: ptr.To(true),
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			internalCtrlWithWarmup, ok := ctrlWithWarmup.(*internalcontroller.Controller[reconcile.Request])
+			Expect(ok).To(BeTrue())
+			Expect(internalCtrlWithWarmup.EnableWarmup).To(HaveValue(BeTrue()))
+
+			// Test with EnableWarmup set to false
+			ctrlWithoutWarmup, err := controller.New("warmup-disabled-ctrl", m, controller.Options{
+				Reconciler:   reconcile.Func(nil),
+				EnableWarmup: ptr.To(false),
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			internalCtrlWithoutWarmup, ok := ctrlWithoutWarmup.(*internalcontroller.Controller[reconcile.Request])
+			Expect(ok).To(BeTrue())
+			Expect(internalCtrlWithoutWarmup.EnableWarmup).To(HaveValue(BeFalse()))
+
+			// Test with EnableWarmup not set (should default to nil)
+			ctrlWithDefaultWarmup, err := controller.New("warmup-default-ctrl", m, controller.Options{
+				Reconciler: reconcile.Func(nil),
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			internalCtrlWithDefaultWarmup, ok := ctrlWithDefaultWarmup.(*internalcontroller.Controller[reconcile.Request])
+			Expect(ok).To(BeTrue())
+			Expect(internalCtrlWithDefaultWarmup.EnableWarmup).To(BeNil())
+		})
+
+		It("should inherit EnableWarmup from manager config", func() {
+			// Test with manager default setting EnableWarmup to true
+			managerWithWarmup, err := manager.New(cfg, manager.Options{
+				Controller: config.Controller{
+					EnableWarmup: ptr.To(true),
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
+			ctrlInheritingWarmup, err := controller.New("inherit-warmup-enabled", managerWithWarmup, controller.Options{
+				Reconciler: reconcile.Func(nil),
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			internalCtrlInheritingWarmup, ok := ctrlInheritingWarmup.(*internalcontroller.Controller[reconcile.Request])
+			Expect(ok).To(BeTrue())
+			Expect(internalCtrlInheritingWarmup.EnableWarmup).To(HaveValue(BeTrue()))
+
+			// Test that explicit controller setting overrides manager setting
+			ctrlOverridingWarmup, err := controller.New("override-warmup-disabled", managerWithWarmup, controller.Options{
+				Reconciler:   reconcile.Func(nil),
+				EnableWarmup: ptr.To(false),
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			internalCtrlOverridingWarmup, ok := ctrlOverridingWarmup.(*internalcontroller.Controller[reconcile.Request])
+			Expect(ok).To(BeTrue())
+			Expect(internalCtrlOverridingWarmup.EnableWarmup).To(HaveValue(BeFalse()))
+		})
 	})
 })
diff --git a/pkg/internal/controller/controller_test.go b/pkg/internal/controller/controller_test.go
index 3fde5da9..8d7d978c 100644
--- a/pkg/internal/controller/controller_test.go
+++ b/pkg/internal/controller/controller_test.go
@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -28,6 +29,7 @@ import (
 	. "github.com/onsi/gomega"
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
+	"go.uber.org/goleak"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -39,10 +41,14 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllertest"
 	"sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/internal/log"
+	"sigs.k8s.io/controller-runtime/pkg/leaderelection"
+	fakeleaderelection "sigs.k8s.io/controller-runtime/pkg/leaderelection/fake"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
@@ -51,6 +57,8 @@ type TestRequest struct {
 	Key string
 }
 
+const testControllerName = "testcontroller"
+
 var _ = Describe("controller", func() {
 	var fakeReconcile *fakeReconciler
 	var ctrl *Controller[reconcile.Request]
@@ -69,7 +77,7 @@ var _ = Describe("controller", func() {
 		queue = &controllertest.Queue{
 			TypedInterface: workqueue.NewTyped[reconcile.Request](),
 		}
-		ctrl = &Controller[reconcile.Request]{
+		ctrl = New[reconcile.Request](Options[reconcile.Request]{
 			MaxConcurrentReconciles: 1,
 			Do:                      fakeReconcile,
 			NewQueue: func(string, workqueue.TypedRateLimiter[reconcile.Request]) workqueue.TypedRateLimitingInterface[reconcile.Request] {
@@ -78,7 +86,7 @@ var _ = Describe("controller", func() {
 			LogConstructor: func(_ *reconcile.Request) logr.Logger {
 				return log.RuntimeLog.WithName("controller").WithName("test")
 			},
-		}
+		})
 	})
 
 	Describe("Reconciler", func() {
@@ -172,7 +180,7 @@ var _ = Describe("controller", func() {
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
 				source.Kind(c, &appsv1.Deployment{}, &handler.TypedEnqueueRequestForObject[*appsv1.Deployment]{}),
 			}
-			ctrl.Name = "testcontroller"
+			ctrl.Name = testControllerName
 
 			err = ctrl.Start(context.TODO())
 			Expect(err).To(HaveOccurred())
@@ -192,7 +200,7 @@ var _ = Describe("controller", func() {
 					cacheSyncDone: sourceSynced,
 				},
 			}
-			ctrl.Name = "testcontroller"
+			ctrl.Name = testControllerName
 
 			ctx, cancel := context.WithCancel(context.TODO())
 			go func() {
@@ -350,14 +358,14 @@ var _ = Describe("controller", func() {
 				TypedRateLimitingInterface: &controllertest.TypedQueue[TestRequest]{
 					TypedInterface: workqueue.NewTyped[TestRequest](),
 				}}
-			ctrl := &Controller[TestRequest]{
+			ctrl := New[TestRequest](Options[TestRequest]{
 				NewQueue: func(string, workqueue.TypedRateLimiter[TestRequest]) workqueue.TypedRateLimitingInterface[TestRequest] {
 					return queue
 				},
 				LogConstructor: func(*TestRequest) logr.Logger {
 					return log.RuntimeLog.WithName("controller").WithName("test")
 				},
-			}
+			})
 			ctrl.CacheSyncTimeout = time.Second
 			src := &bisignallingSource[TestRequest]{
 				startCall: make(chan workqueue.TypedRateLimitingInterface[TestRequest]),
@@ -383,16 +391,26 @@ var _ = Describe("controller", func() {
 		})
 	})
 
-	Describe("startEventSources", func() {
+	Describe("startEventSourcesAndQueueLocked", func() {
 		It("should return nil when no sources are provided", func() {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{}
-			err := ctrl.startEventSources(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).NotTo(HaveOccurred())
 		})
 
+		It("should initialize controller queue when called", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{}
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ctrl.Queue).NotTo(BeNil())
+		})
+
 		It("should return an error if a source fails to start", func() {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -403,10 +421,10 @@ var _ = Describe("controller", func() {
 				return expectedErr
 			})
 
-			// // Set a sufficiently long timeout to avoid timeouts interfering with the error being returned
+			// Set a sufficiently long timeout to avoid timeouts interfering with the error being returned
 			ctrl.CacheSyncTimeout = 5 * time.Second
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{src}
-			err := ctrl.startEventSources(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).To(Equal(expectedErr))
 		})
 
@@ -420,7 +438,7 @@ var _ = Describe("controller", func() {
 			ctrl.Name = "test-controller"
 			ctrl.CacheSyncTimeout = 5 * time.Second
 
-			err := ctrl.startEventSources(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("failed to wait for test-controller caches to sync"))
 		})
@@ -436,7 +454,7 @@ var _ = Describe("controller", func() {
 			ctrl.Name = "test-controller"
 			ctrl.CacheSyncTimeout = 5 * time.Second
 
-			err := ctrl.startEventSources(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).NotTo(HaveOccurred())
 		})
 
@@ -459,7 +477,8 @@ var _ = Describe("controller", func() {
 			// Start the sources in a goroutine
 			startErrCh := make(chan error)
 			go func() {
-				startErrCh <- ctrl.startEventSources(sourceCtx)
+				defer GinkgoRecover()
+				startErrCh <- ctrl.startEventSourcesAndQueueLocked(sourceCtx)
 			}()
 
 			// Allow source to start successfully
@@ -494,10 +513,104 @@ var _ = Describe("controller", func() {
 
 			ctrl.startWatches = []source.TypedSource[reconcile.Request]{blockingSrc}
 
-			err := ctrl.startEventSources(ctx)
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("timed out waiting for source"))
 		})
+
+		It("should only start sources once when called multiple times concurrently", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.CacheSyncTimeout = 1 * time.Millisecond
+
+			var startCount atomic.Int32
+			src := source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+				startCount.Add(1)
+				return nil
+			})
+
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{src}
+
+			By("Calling startEventSourcesAndQueueLocked multiple times in parallel")
+			var wg sync.WaitGroup
+			for i := 1; i <= 5; i++ {
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					err := ctrl.startEventSourcesAndQueueLocked(ctx)
+					// All calls should return the same nil error
+					Expect(err).NotTo(HaveOccurred())
+				}()
+			}
+
+			wg.Wait()
+			Expect(startCount.Load()).To(Equal(int32(1)), "Source should only be started once even when called multiple times")
+		})
+
+		It("should block subsequent calls from returning until the first call to startEventSourcesAndQueueLocked has returned", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			ctrl.CacheSyncTimeout = 5 * time.Second
+
+			// finishSourceChan is closed to unblock startEventSourcesAndQueueLocked from returning
+			finishSourceChan := make(chan struct{})
+
+			src := source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+				<-finishSourceChan
+				return nil
+			})
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{src}
+
+			By("Calling startEventSourcesAndQueueLocked asynchronously")
+			wg := sync.WaitGroup{}
+			go func() {
+				defer GinkgoRecover()
+				defer wg.Done()
+
+				wg.Add(1)
+				Expect(ctrl.startEventSourcesAndQueueLocked(ctx)).To(Succeed())
+			}()
+
+			By("Calling startEventSourcesAndQueueLocked again")
+			var didSubsequentCallComplete atomic.Bool
+			go func() {
+				defer GinkgoRecover()
+				defer wg.Done()
+
+				wg.Add(1)
+				Expect(ctrl.startEventSourcesAndQueueLocked(ctx)).To(Succeed())
+				didSubsequentCallComplete.Store(true)
+			}()
+
+			// Assert that second call to startEventSourcesAndQueueLocked is blocked while source has not finished
+			Consistently(didSubsequentCallComplete.Load).Should(BeFalse())
+
+			By("Finishing source start + sync")
+			finishSourceChan <- struct{}{}
+
+			// Assert that second call to startEventSourcesAndQueueLocked is now complete
+			Eventually(didSubsequentCallComplete.Load).Should(BeTrue(), "startEventSourcesAndQueueLocked should complete after source is started and synced")
+			wg.Wait()
+		})
+
+		It("should reset c.startWatches to nil after returning and startedEventSourcesAndQueue", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.CacheSyncTimeout = 1 * time.Millisecond
+
+			src := source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+				return nil
+			})
+
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{src}
+
+			err := ctrl.startEventSourcesAndQueueLocked(ctx)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ctrl.startWatches).To(BeNil(), "startWatches should be reset to nil after returning")
+			Expect(ctrl.startedEventSourcesAndQueue).To(BeTrue(), "startedEventSourcesAndQueue should be set to true after startEventSourcesAndQueueLocked returns without error")
+		})
 	})
 
 	Describe("Processing queue items from a Controller", func() {
@@ -1014,6 +1127,529 @@ var _ = Describe("controller", func() {
 			})
 		})
 	})
+
+	Describe("Warmup", func() {
+		JustBeforeEach(func() {
+			ctrl.EnableWarmup = ptr.To(true)
+		})
+
+		It("should track warmup status correctly with successful sync", func() {
+			// Setup controller with sources that complete successfully
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.CacheSyncTimeout = time.Second
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					return nil
+				}),
+			}
+
+			Expect(ctrl.Warmup(ctx)).To(Succeed())
+		})
+
+		It("should return an error if there is an error waiting for the informers", func() {
+			ctrl.CacheSyncTimeout = time.Second
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Kind(&informertest.FakeInformers{Synced: ptr.To(false)}, &corev1.Pod{}, &handler.TypedEnqueueRequestForObject[*corev1.Pod]{}),
+			}
+			ctrl.Name = testControllerName
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			err := ctrl.Warmup(ctx)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("failed to wait for testcontroller caches to sync"))
+		})
+
+		It("should error when cache sync timeout occurs", func() {
+			c, err := cache.New(cfg, cache.Options{})
+			Expect(err).NotTo(HaveOccurred())
+			c = &cacheWithIndefinitelyBlockingGetInformer{c}
+
+			ctrl.CacheSyncTimeout = time.Second
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Kind(c, &appsv1.Deployment{}, &handler.TypedEnqueueRequestForObject[*appsv1.Deployment]{}),
+			}
+			ctrl.Name = testControllerName
+
+			err = ctrl.Warmup(context.TODO())
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("failed to wait for testcontroller caches to sync kind source: *v1.Deployment: timed out waiting for cache to be synced"))
+		})
+
+		It("should not error when controller Warmup context is cancelled during Sources WaitForSync", func() {
+			ctrl.CacheSyncTimeout = 1 * time.Second
+
+			sourceSynced := make(chan struct{})
+			c, err := cache.New(cfg, cache.Options{})
+			Expect(err).NotTo(HaveOccurred())
+			c = &cacheWithIndefinitelyBlockingGetInformer{c}
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				&singnallingSourceWrapper{
+					SyncingSource: source.Kind[client.Object](c, &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}),
+					cacheSyncDone: sourceSynced,
+				},
+			}
+			ctrl.Name = testControllerName
+
+			ctx, cancel := context.WithCancel(context.TODO())
+			go func() {
+				defer GinkgoRecover()
+				err = ctrl.Warmup(ctx)
+				Expect(err).To(Succeed())
+			}()
+
+			cancel()
+			<-sourceSynced
+		})
+
+		It("should error when Warmup() is blocking forever", func() {
+			ctrl.CacheSyncTimeout = time.Second
+
+			controllerDone := make(chan struct{})
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					<-controllerDone
+					return ctx.Err()
+				})}
+
+			ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
+			defer cancel()
+
+			err := ctrl.Warmup(ctx)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("Please ensure that its Start() method is non-blocking"))
+
+			close(controllerDone)
+		})
+
+		It("should not error when cache sync timeout is of sufficiently high", func() {
+			ctrl.CacheSyncTimeout = 10 * time.Second
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			sourceSynced := make(chan struct{})
+			c := &informertest.FakeInformers{}
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				&singnallingSourceWrapper{
+					SyncingSource: source.Kind[client.Object](c, &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}),
+					cacheSyncDone: sourceSynced,
+				},
+			}
+
+			go func() {
+				defer GinkgoRecover()
+				Expect(ctrl.Warmup(ctx)).To(Succeed())
+			}()
+
+			<-sourceSynced
+		})
+
+		It("should process events from source.Channel", func() {
+			ctrl.CacheSyncTimeout = 10 * time.Second
+			// channel to be closed when event is processed
+			processed := make(chan struct{})
+			// source channel
+			ch := make(chan event.GenericEvent, 1)
+
+			ctx, cancel := context.WithCancel(context.TODO())
+			defer cancel()
+
+			// event to be sent to the channel
+			p := &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "bar"},
+			}
+			evt := event.GenericEvent{
+				Object: p,
+			}
+
+			ins := source.Channel(
+				ch,
+				handler.Funcs{
+					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+						defer GinkgoRecover()
+						close(processed)
+					},
+				},
+			)
+
+			// send the event to the channel
+			ch <- evt
+
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{ins}
+
+			go func() {
+				defer GinkgoRecover()
+				Expect(ctrl.Warmup(ctx)).To(Succeed())
+			}()
+			<-processed
+		})
+
+		It("should error when channel source is not specified", func() {
+			ctrl.CacheSyncTimeout = 10 * time.Second
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ins := source.Channel[string](nil, nil)
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{ins}
+
+			e := ctrl.Warmup(ctx)
+			Expect(e).To(HaveOccurred())
+			Expect(e.Error()).To(ContainSubstring("must specify Channel.Source"))
+		})
+
+		It("should call Start on sources with the appropriate EventHandler, Queue, and Predicates", func() {
+			ctrl.CacheSyncTimeout = 10 * time.Second
+			started := false
+			ctx, cancel := context.WithCancel(context.Background())
+			src := source.Func(func(ctx context.Context, q workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+				defer GinkgoRecover()
+				Expect(q).To(Equal(ctrl.Queue))
+
+				started = true
+				cancel() // Cancel the context so ctrl.Warmup() doesn't block forever
+				return nil
+			})
+			Expect(ctrl.Watch(src)).NotTo(HaveOccurred())
+
+			err := ctrl.Warmup(ctx)
+			Expect(err).To(Succeed())
+			Expect(started).To(BeTrue())
+		})
+
+		It("should return an error if there is an error starting sources", func() {
+			ctrl.CacheSyncTimeout = 10 * time.Second
+			err := fmt.Errorf("Expected Error: could not start source")
+			src := source.Func(func(context.Context,
+				workqueue.TypedRateLimitingInterface[reconcile.Request],
+			) error {
+				defer GinkgoRecover()
+				return err
+			})
+			Expect(ctrl.Watch(src)).To(Succeed())
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			Expect(ctrl.Warmup(ctx)).To(Equal(err))
+		})
+
+		It("should track warmup status correctly with unsuccessful sync", func() {
+			// Setup controller with sources that complete with error
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.CacheSyncTimeout = time.Second
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					return errors.New("sync error")
+				}),
+			}
+
+			err := ctrl.Warmup(ctx)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("sync error"))
+		})
+
+		It("should call Start on sources with the appropriate non-nil queue", func() {
+			ctrl.CacheSyncTimeout = 10 * time.Second
+			started := false
+			ctx, cancel := context.WithCancel(context.Background())
+			src := source.Func(func(ctx context.Context, q workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+				defer GinkgoRecover()
+				Expect(q).ToNot(BeNil())
+				Expect(q).To(Equal(ctrl.Queue))
+
+				started = true
+				cancel() // Cancel the context so ctrl.Start() doesn't block forever
+				return nil
+			})
+			Expect(ctrl.Watch(src)).To(Succeed())
+			Expect(ctrl.Warmup(ctx)).To(Succeed())
+			Expect(ctrl.Queue).ToNot(BeNil())
+			Expect(started).To(BeTrue())
+		})
+
+		It("should return true if context is cancelled while waiting for source to start", func() {
+			// Setup controller with sources that complete with error
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.CacheSyncTimeout = time.Second
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					<-ctx.Done()
+					return nil
+				}),
+			}
+
+			// channel to prevent the goroutine from outliving the It test
+			waitChan := make(chan struct{})
+
+			// Invoked in a goroutine because Warmup will block
+			go func() {
+				defer GinkgoRecover()
+				defer close(waitChan)
+				Expect(ctrl.Warmup(ctx)).To(Succeed())
+			}()
+
+			cancel()
+			<-waitChan
+		})
+
+		It("should be called before leader election runnables if warmup is enabled", func() {
+			// This unit test exists to ensure that a warmup enabled controller will actually be
+			// called in the warmup phase before the leader election runnables are started. It
+			// catches regressions in the controller that would not implement warmupRunnable from
+			// pkg/manager.
+			ctx, cancel := context.WithCancel(context.Background())
+
+			By("Creating a channel to track execution order")
+			runnableExecutionOrderChan := make(chan string, 2)
+			const nonWarmupRunnableName = "nonWarmupRunnable"
+			const warmupRunnableName = "warmupRunnable"
+
+			ctrl.CacheSyncTimeout = time.Second
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					runnableExecutionOrderChan <- warmupRunnableName
+					return nil
+				}),
+			}
+
+			nonWarmupCtrl := New[reconcile.Request](Options[reconcile.Request]{
+				MaxConcurrentReconciles: 1,
+				Do:                      fakeReconcile,
+				NewQueue: func(string, workqueue.TypedRateLimiter[reconcile.Request]) workqueue.TypedRateLimitingInterface[reconcile.Request] {
+					return queue
+				},
+				LogConstructor: func(_ *reconcile.Request) logr.Logger {
+					return log.RuntimeLog.WithName("controller").WithName("test")
+				},
+				CacheSyncTimeout: time.Second,
+				EnableWarmup:     ptr.To(false),
+				LeaderElected:    ptr.To(true),
+			})
+			nonWarmupCtrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					runnableExecutionOrderChan <- nonWarmupRunnableName
+					return nil
+				}),
+			}
+
+			By("Creating a test resource lock with hooks")
+			resourceLock, err := fakeleaderelection.NewResourceLock(nil, nil, leaderelection.Options{})
+			Expect(err).ToNot(HaveOccurred())
+
+			By("Creating a manager")
+			testenv = &envtest.Environment{}
+			cfg, err := testenv.Start()
+			Expect(err).NotTo(HaveOccurred())
+			m, err := manager.New(cfg, manager.Options{
+				LeaderElection:                      true,
+				LeaderElectionID:                    "some-leader-election-id",
+				LeaderElectionNamespace:             "default",
+				LeaderElectionResourceLockInterface: resourceLock,
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Adding warmup and non-warmup controllers to the manager")
+			Expect(m.Add(ctrl)).To(Succeed())
+			Expect(m.Add(nonWarmupCtrl)).To(Succeed())
+
+			By("Blocking leader election")
+			resourceLockWithHooks, ok := resourceLock.(fakeleaderelection.ControllableResourceLockInterface)
+			Expect(ok).To(BeTrue(), "resource lock should implement ResourceLockInterfaceWithHooks")
+			resourceLockWithHooks.BlockLeaderElection()
+
+			By("Starting the manager")
+			waitChan := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				defer close(waitChan)
+				Expect(m.Start(ctx)).To(Succeed())
+			}()
+			Expect(<-runnableExecutionOrderChan).To(Equal(warmupRunnableName))
+
+			By("Unblocking leader election")
+			resourceLockWithHooks.UnblockLeaderElection()
+			<-m.Elected()
+			Expect(<-runnableExecutionOrderChan).To(Equal(nonWarmupRunnableName))
+
+			cancel()
+			<-waitChan
+		})
+
+		It("should not cause a data race when called concurrently", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.CacheSyncTimeout = time.Second
+
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					return nil
+				}),
+			}
+
+			var wg sync.WaitGroup
+			for i := 0; i < 5; i++ {
+				wg.Add(1)
+				go func() {
+					defer GinkgoRecover()
+					defer wg.Done()
+					Expect(ctrl.Warmup(ctx)).To(Succeed())
+				}()
+			}
+
+			wg.Wait()
+		})
+
+		It("should not cause a data race when called concurrently with Start and only start sources once", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			ctrl.CacheSyncTimeout = time.Second
+			numWatches := 10
+
+			var watchStartedCount atomic.Int32
+			for range numWatches {
+				ctrl.startWatches = append(ctrl.startWatches, source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					watchStartedCount.Add(1)
+					return nil
+				}))
+			}
+
+			By("calling Warmup and Start concurrently")
+			go func() {
+				defer GinkgoRecover()
+				Expect(ctrl.Start(ctx)).To(Succeed())
+			}()
+
+			blockOnWarmupChan := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				Expect(ctrl.Warmup(ctx)).To(Succeed())
+				close(blockOnWarmupChan)
+			}()
+
+			<-blockOnWarmupChan
+
+			Expect(watchStartedCount.Load()).To(Equal(int32(numWatches)), "source should only be started once")
+			Expect(ctrl.startWatches).To(BeNil(), "startWatches should be reset to nil after they are started")
+		})
+
+		It("should start sources added after Warmup is called", func() {
+			ctx, cancel := context.WithCancel(context.Background())
+
+			ctrl.CacheSyncTimeout = time.Second
+
+			Expect(ctrl.Warmup(ctx)).To(Succeed())
+
+			By("starting a watch after warmup is added")
+			var didWatchStart atomic.Bool
+			Expect(ctrl.Watch(source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+				didWatchStart.Store(true)
+				return nil
+			}))).To(Succeed())
+
+			waitChan := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				Expect(ctrl.Start(ctx)).To(Succeed())
+				close(waitChan)
+			}()
+
+			Eventually(didWatchStart.Load).Should(BeTrue(), "watch should be started if it is added after Warmup")
+
+			cancel()
+			<-waitChan
+		})
+
+		DescribeTable("should not leak goroutines when manager is stopped with warmup runnable",
+			func(leaderElection bool) {
+				ctx, cancel := context.WithCancel(context.Background())
+				defer cancel()
+
+				ctrl.CacheSyncTimeout = time.Second
+
+				By("Creating a manager")
+				testenv = &envtest.Environment{}
+				cfg, err := testenv.Start()
+				Expect(err).NotTo(HaveOccurred())
+				m, err := manager.New(cfg, manager.Options{
+					LeaderElection:          leaderElection,
+					LeaderElectionID:        "some-leader-election-id",
+					LeaderElectionNamespace: "default",
+				})
+				Expect(err).NotTo(HaveOccurred())
+
+				ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+					source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+						<-ctx.Done()
+						return nil
+					}),
+				}
+				Expect(m.Add(ctrl)).To(Succeed())
+
+				// ignore needs to go after the testenv.Start() call to ignore the apiserver
+				// process
+				currentGRs := goleak.IgnoreCurrent()
+				waitChan := make(chan struct{})
+				go func() {
+					defer GinkgoRecover()
+					Expect(m.Start(ctx)).To(Succeed())
+					close(waitChan)
+				}()
+
+				<-m.Elected()
+				By("stopping the manager via context")
+				cancel()
+
+				Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
+				<-waitChan
+			},
+			Entry("and with leader election enabled", true),
+			Entry("and without leader election enabled", false),
+		)
+	})
+
+	Describe("Warmup with warmup disabled", func() {
+		JustBeforeEach(func() {
+			ctrl.EnableWarmup = ptr.To(false)
+		})
+
+		It("should not start sources when Warmup is called if warmup is disabled but start it when Start is called.", func() {
+			// Setup controller with sources that complete successfully
+			ctx, cancel := context.WithCancel(context.Background())
+
+			ctrl.CacheSyncTimeout = time.Second
+			var isSourceStarted atomic.Bool
+			isSourceStarted.Store(false)
+			ctrl.startWatches = []source.TypedSource[reconcile.Request]{
+				source.Func(func(ctx context.Context, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
+					isSourceStarted.Store(true)
+					return nil
+				}),
+			}
+
+			By("Calling Warmup when EnableWarmup is false")
+			err := ctrl.Warmup(ctx)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(isSourceStarted.Load()).To(BeFalse())
+
+			By("Calling Start when EnableWarmup is false")
+			waitChan := make(chan struct{})
+
+			go func() {
+				defer GinkgoRecover()
+				Expect(ctrl.Start(ctx)).To(Succeed())
+				close(waitChan)
+			}()
+			Eventually(isSourceStarted.Load).Should(BeTrue())
+			cancel()
+			<-waitChan
+		})
+	})
 })
 
 var _ = Describe("ReconcileIDFromContext function", func() {
diff --git a/pkg/manager/internal/integration/manager_test.go b/pkg/manager/internal/integration/manager_test.go
index 346daa1e..c83eead3 100644
--- a/pkg/manager/internal/integration/manager_test.go
+++ b/pkg/manager/internal/integration/manager_test.go
@@ -34,10 +34,12 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/utils/ptr"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
@@ -90,6 +92,8 @@ var (
 			},
 		},
 	}
+
+	ctx = ctrl.SetupSignalHandler()
 )
 
 var _ = Describe("manger.Manager Start", func() {
@@ -107,9 +111,7 @@ var _ = Describe("manger.Manager Start", func() {
 	//   * Add an index on v2 Driver to ensure we start and wait for an informer during cache.Start (as part of manager.Start)
 	//     * Note: cache.Start would fail if the conversion webhook doesn't work (which in turn depends on the readiness probe)
 	//     * Note: Adding the index for v2 ensures the Driver list call during Informer sync goes through conversion.
-	It("should start all components without deadlock", func() {
-		ctx := ctrl.SetupSignalHandler()
-
+	DescribeTable("should start all components without deadlock", func(warmupEnabled bool) {
 		// Set up schema.
 		Expect(clientgoscheme.AddToScheme(scheme)).To(Succeed())
 		Expect(apiextensionsv1.AddToScheme(scheme)).To(Succeed())
@@ -163,7 +165,13 @@ var _ = Describe("manger.Manager Start", func() {
 		driverReconciler := &DriverReconciler{
 			Client: mgr.GetClient(),
 		}
-		Expect(ctrl.NewControllerManagedBy(mgr).For(&crewv2.Driver{}).Complete(driverReconciler)).To(Succeed())
+		Expect(
+			ctrl.NewControllerManagedBy(mgr).
+				For(&crewv2.Driver{}).
+				Named(fmt.Sprintf("driver_warmup_%t", warmupEnabled)).
+				WithOptions(controller.Options{EnableWarmup: ptr.To(warmupEnabled)}).
+				Complete(driverReconciler),
+		).To(Succeed())
 
 		// Set up a conversion webhook.
 		conversionWebhook := createConversionWebhook(mgr)
@@ -211,7 +219,10 @@ var _ = Describe("manger.Manager Start", func() {
 
 		// Shutdown the server
 		cancel()
-	})
+	},
+		Entry("controller warmup enabled", true),
+		Entry("controller warmup not enabled", false),
+	)
 })
 
 type DriverReconciler struct {
diff --git a/pkg/manager/manager_test.go b/pkg/manager/manager_test.go
index db161952..3796c6da 100644
--- a/pkg/manager/manager_test.go
+++ b/pkg/manager/manager_test.go
@@ -1905,6 +1905,54 @@ var _ = Describe("manger.Manager", func() {
 		Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
 	})
 
+	It("should not leak goroutines when a runnable returns error slowly after being signaled to stop", func() {
+		// This test reproduces the race condition where the manager's Start method
+		// exits due to context cancellation, leaving no one to drain errChan
+
+		currentGRs := goleak.IgnoreCurrent()
+
+		// Create manager with a very short graceful shutdown timeout to reliablytrigger the race condition
+		shortGracefulShutdownTimeout := 10 * time.Millisecond
+		m, err := New(cfg, Options{
+			GracefulShutdownTimeout: &shortGracefulShutdownTimeout,
+		})
+		Expect(err).NotTo(HaveOccurred())
+
+		// Add the slow runnable that will return an error after some delay
+		for i := 0; i < 3; i++ {
+			slowRunnable := RunnableFunc(func(c context.Context) error {
+				<-c.Done()
+
+				// Simulate some work that delays the error from being returned
+				// Choosing a large delay to reliably trigger the race condition
+				time.Sleep(100 * time.Millisecond)
+
+				// This simulates the race condition where runnables try to send
+				// errors after the manager has stopped reading from errChan
+				return errors.New("slow runnable error")
+			})
+
+			Expect(m.Add(slowRunnable)).To(Succeed())
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+		go func() {
+			defer GinkgoRecover()
+			Expect(m.Start(ctx)).To(HaveOccurred()) // We expect error here because the slow runnables will return errors
+		}()
+
+		// Wait for context to be cancelled
+		<-ctx.Done()
+
+		// Give time for any leaks to become apparent. This makes sure that we don't false alarm on go routine leaks because runnables are still running.
+		time.Sleep(300 * time.Millisecond)
+
+		// force-close keep-alive connections
+		clientTransport.CloseIdleConnections()
+		Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
+	})
+
 	It("should provide a function to get the Config", func() {
 		m, err := New(cfg, Options{})
 		Expect(err).NotTo(HaveOccurred())
@@ -1947,6 +1995,79 @@ var _ = Describe("manger.Manager", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(m.GetAPIReader()).NotTo(BeNil())
 	})
+
+	It("should run warmup runnables before leader election is won", func() {
+		By("Creating a channel to track execution order")
+		runnableExecutionOrderChan := make(chan string, 2)
+		const leaderElectionRunnableName = "leaderElectionRunnable"
+		const warmupRunnableName = "warmupRunnable"
+
+		By("Creating a manager with leader election enabled")
+		m, err := New(cfg, Options{
+			LeaderElection:          true,
+			LeaderElectionNamespace: "default",
+			LeaderElectionID:        "test-leader-election-warmup",
+			newResourceLock:         fakeleaderelection.NewResourceLock,
+			HealthProbeBindAddress:  "0",
+			Metrics:                 metricsserver.Options{BindAddress: "0"},
+			PprofBindAddress:        "0",
+		})
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Creating a runnable that implements WarmupRunnable interface")
+		// Create a warmup runnable
+		warmupRunnable := newWarmupRunnableFunc(
+			func(ctx context.Context) error {
+				// This is the leader election runnable that will be executed after leader election
+				// It will block until context is done/cancelled
+				<-ctx.Done()
+				return nil
+			},
+			func(ctx context.Context) error {
+				// This should be called during startup before leader election
+				runnableExecutionOrderChan <- warmupRunnableName
+				return nil
+			},
+		)
+		Expect(m.Add(warmupRunnable)).To(Succeed())
+
+		By("Creating a runnable that requires leader election")
+		leaderElectionRunnable := RunnableFunc(
+			func(ctx context.Context) error {
+				runnableExecutionOrderChan <- leaderElectionRunnableName
+				<-ctx.Done()
+				return nil
+			},
+		)
+		Expect(m.Add(leaderElectionRunnable)).To(Succeed())
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		cm, ok := m.(*controllerManager)
+		Expect(ok).To(BeTrue())
+		resourceLockWithHooks, ok := cm.resourceLock.(fakeleaderelection.ControllableResourceLockInterface)
+		Expect(ok).To(BeTrue())
+
+		By("Blocking leader election")
+		resourceLockWithHooks.BlockLeaderElection()
+
+		By("Starting the manager")
+		go func() {
+			defer GinkgoRecover()
+			Expect(m.Start(ctx)).To(Succeed())
+		}()
+
+		By("Waiting for the warmup runnable to be executed without leader election being won")
+		Expect(<-runnableExecutionOrderChan).To(Equal(warmupRunnableName))
+
+		By("Unblocking leader election")
+		resourceLockWithHooks.UnblockLeaderElection()
+
+		By("Waiting for the leader election runnable to be executed after leader election was won")
+		<-m.Elected()
+		Expect(<-runnableExecutionOrderChan).To(Equal(leaderElectionRunnableName))
+	})
 })
 
 type runnableError struct {
diff --git a/pkg/manager/runnable_group_test.go b/pkg/manager/runnable_group_test.go
index f2f4119b..52086047 100644
--- a/pkg/manager/runnable_group_test.go
+++ b/pkg/manager/runnable_group_test.go
@@ -27,6 +27,7 @@ var _ = Describe("runnables", func() {
 		r := newRunnables(defaultBaseContext, errCh)
 		Expect(r.Add(server)).To(Succeed())
 		Expect(r.HTTPServers.startQueue).To(HaveLen(1))
+		Expect(r.Others.startQueue).To(BeEmpty())
 	})
 
 	It("should add caches to the appropriate group", func() {
@@ -34,6 +35,7 @@ var _ = Describe("runnables", func() {
 		r := newRunnables(defaultBaseContext, errCh)
 		Expect(r.Add(cache)).To(Succeed())
 		Expect(r.Caches.startQueue).To(HaveLen(1))
+		Expect(r.Others.startQueue).To(BeEmpty())
 	})
 
 	It("should add webhooks to the appropriate group", func() {
@@ -41,6 +43,7 @@ var _ = Describe("runnables", func() {
 		r := newRunnables(defaultBaseContext, errCh)
 		Expect(r.Add(webhook)).To(Succeed())
 		Expect(r.Webhooks.startQueue).To(HaveLen(1))
+		Expect(r.Others.startQueue).To(BeEmpty())
 	})
 
 	It("should add any runnable to the leader election group", func() {
@@ -52,6 +55,121 @@ var _ = Describe("runnables", func() {
 		r := newRunnables(defaultBaseContext, errCh)
 		Expect(r.Add(runnable)).To(Succeed())
 		Expect(r.LeaderElection.startQueue).To(HaveLen(1))
+		Expect(r.Others.startQueue).To(BeEmpty())
+	})
+
+	It("should add WarmupRunnable to the Warmup and LeaderElection group", func() {
+		warmupRunnable := newWarmupRunnableFunc(
+			func(c context.Context) error {
+				<-c.Done()
+				return nil
+			},
+			func(c context.Context) error { return nil },
+		)
+
+		r := newRunnables(defaultBaseContext, errCh)
+		Expect(r.Add(warmupRunnable)).To(Succeed())
+		Expect(r.Warmup.startQueue).To(HaveLen(1))
+		Expect(r.LeaderElection.startQueue).To(HaveLen(1))
+		Expect(r.Others.startQueue).To(BeEmpty())
+	})
+
+	It("should add WarmupRunnable that doesn't needs leader election to warmup group only", func() {
+		warmupRunnable := newLeaderElectionAndWarmupRunnable(
+			func(c context.Context) error {
+				<-c.Done()
+				return nil
+			},
+			func(c context.Context) error { return nil },
+			false,
+		)
+
+		r := newRunnables(defaultBaseContext, errCh)
+		Expect(r.Add(warmupRunnable)).To(Succeed())
+
+		Expect(r.Warmup.startQueue).To(HaveLen(1))
+		Expect(r.LeaderElection.startQueue).To(BeEmpty())
+		Expect(r.Others.startQueue).To(HaveLen(1))
+	})
+
+	It("should add WarmupRunnable that needs leader election to Warmup and LeaderElection group, not Others", func() {
+		warmupRunnable := newLeaderElectionAndWarmupRunnable(
+			func(c context.Context) error {
+				<-c.Done()
+				return nil
+			},
+			func(c context.Context) error { return nil },
+			true,
+		)
+
+		r := newRunnables(defaultBaseContext, errCh)
+		Expect(r.Add(warmupRunnable)).To(Succeed())
+
+		Expect(r.Warmup.startQueue).To(HaveLen(1))
+		Expect(r.LeaderElection.startQueue).To(HaveLen(1))
+		Expect(r.Others.startQueue).To(BeEmpty())
+	})
+
+	It("should execute the Warmup function when Warmup group is started", func() {
+		var warmupExecuted atomic.Bool
+
+		warmupRunnable := newWarmupRunnableFunc(
+			func(c context.Context) error {
+				<-c.Done()
+				return nil
+			},
+			func(c context.Context) error {
+				warmupExecuted.Store(true)
+				return nil
+			},
+		)
+
+		r := newRunnables(defaultBaseContext, errCh)
+		Expect(r.Add(warmupRunnable)).To(Succeed())
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		// Start the Warmup group
+		Expect(r.Warmup.Start(ctx)).To(Succeed())
+
+		// Verify warmup function was called
+		Expect(warmupExecuted.Load()).To(BeTrue())
+	})
+
+	It("should propagate errors from Warmup function to error channel", func() {
+		expectedErr := fmt.Errorf("expected warmup error")
+
+		warmupRunnable := newWarmupRunnableFunc(
+			func(c context.Context) error {
+				<-c.Done()
+				return nil
+			},
+			func(c context.Context) error { return expectedErr },
+		)
+
+		testErrChan := make(chan error, 1)
+		r := newRunnables(defaultBaseContext, testErrChan)
+		Expect(r.Add(warmupRunnable)).To(Succeed())
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		// Start the Warmup group in a goroutine
+		go func() {
+			Expect(r.Warmup.Start(ctx)).To(Succeed())
+		}()
+
+		// Error from Warmup should be sent to error channel
+		var receivedErr error
+		Eventually(func() error {
+			select {
+			case receivedErr = <-testErrChan:
+				return receivedErr
+			default:
+				return nil
+			}
+		}).Should(Equal(expectedErr))
 	})
 })
 
@@ -224,3 +342,57 @@ var _ = Describe("runnableGroup", func() {
 		}
 	})
 })
+
+var _ warmupRunnable = &warmupRunnableFunc{}
+
+func newWarmupRunnableFunc(
+	startFunc func(context.Context) error,
+	warmupFunc func(context.Context) error,
+) *warmupRunnableFunc {
+	return &warmupRunnableFunc{
+		startFunc:  startFunc,
+		warmupFunc: warmupFunc,
+	}
+}
+
+// warmupRunnableFunc is a helper struct that implements WarmupRunnable
+// for testing purposes.
+type warmupRunnableFunc struct {
+	startFunc  func(context.Context) error
+	warmupFunc func(context.Context) error
+}
+
+func (r *warmupRunnableFunc) Start(ctx context.Context) error {
+	return r.startFunc(ctx)
+}
+
+func (r *warmupRunnableFunc) Warmup(ctx context.Context) error {
+	return r.warmupFunc(ctx)
+}
+
+var _ LeaderElectionRunnable = &leaderElectionAndWarmupRunnable{}
+var _ warmupRunnable = &leaderElectionAndWarmupRunnable{}
+
+// leaderElectionAndWarmupRunnable implements both WarmupRunnable and LeaderElectionRunnable
+type leaderElectionAndWarmupRunnable struct {
+	*warmupRunnableFunc
+	needLeaderElection bool
+}
+
+func newLeaderElectionAndWarmupRunnable(
+	startFunc func(context.Context) error,
+	warmupFunc func(context.Context) error,
+	needLeaderElection bool,
+) *leaderElectionAndWarmupRunnable {
+	return &leaderElectionAndWarmupRunnable{
+		warmupRunnableFunc: &warmupRunnableFunc{
+			startFunc:  startFunc,
+			warmupFunc: warmupFunc,
+		},
+		needLeaderElection: needLeaderElection,
+	}
+}
+
+func (r leaderElectionAndWarmupRunnable) NeedLeaderElection() bool {
+	return r.needLeaderElection
+}
diff --git a/tools/setup-envtest/go.mod b/tools/setup-envtest/go.mod
index 73c8dc62..08698ed6 100644
--- a/tools/setup-envtest/go.mod
+++ b/tools/setup-envtest/go.mod
@@ -10,8 +10,8 @@ require (
 	github.com/spf13/afero v1.12.0
 	github.com/spf13/pflag v1.0.6
 	go.uber.org/zap v1.27.0
-	k8s.io/apimachinery v0.34.0-alpha.2
-	sigs.k8s.io/yaml v1.4.0
+	k8s.io/apimachinery v0.34.0-beta.0
+	sigs.k8s.io/yaml v1.5.0
 )
 
 require (
@@ -19,6 +19,7 @@ require (
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
 	go.uber.org/multierr v1.10.0 // indirect
+	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/sys v0.31.0 // indirect
 	golang.org/x/text v0.23.0 // indirect
diff --git a/tools/setup-envtest/go.sum b/tools/setup-envtest/go.sum
index c3b796a9..e785ee09 100644
--- a/tools/setup-envtest/go.sum
+++ b/tools/setup-envtest/go.sum
@@ -6,7 +6,6 @@ github.com/go-logr/zapr v1.3.0 h1:XGdV8XW8zdwFiwOA2Dryh1gj2KRQyOOoNmBy4EplIcQ=
 github.com/go-logr/zapr v1.3.0/go.mod h1:YKepepNBd1u/oyhd/yQmtjVXmm9uML4IXUgMOwR8/Gg=
 github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
 github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
-github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad h1:a6HEuzUHeKH6hwfN/ZoQgRgVIWFJljSWa/zetS2WTvg=
@@ -29,6 +28,10 @@ go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
 go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
+go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
+go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
+go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
 golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8=
 golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
 golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
@@ -43,7 +46,7 @@ gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/apimachinery v0.34.0-alpha.2 h1:kKK+KUnbsRWJObxY3pChe0IYkEWG5L4Qsmr/iTIuFQE=
-k8s.io/apimachinery v0.34.0-alpha.2/go.mod h1:HYUvaFuBNToT/JW+lhc2D9BYgqj6xhI5eORJ7BnvKpM=
-sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E=
-sigs.k8s.io/yaml v1.4.0/go.mod h1:Ejl7/uTz7PSA4eKMyQCUTnhZYNmLIl+5c2lQPGR2BPY=
+k8s.io/apimachinery v0.34.0-beta.0 h1:C6teSJBCx6ArW7122MM9hQqeGW2w/QQ0lB4x4Z4Iftc=
+k8s.io/apimachinery v0.34.0-beta.0/go.mod h1:TP8uyOuDEOnzGpLOdffo8hPnIjVDljZCxCM/fruV+5M=
+sigs.k8s.io/yaml v1.5.0 h1:M10b2U7aEUY6hRtU870n2VTPgR5RZiL/I6Lcc2F4NUQ=
+sigs.k8s.io/yaml v1.5.0/go.mod h1:wZs27Rbxoai4C0f8/9urLZtZtF3avA3gKvGyPdDqTO4=