Repository: Kong/kubernetes-ingress-controller
Pull request #6044 (2024-05-16 14:16:12)

chore(deps): bump k8s.io/kubectl from 0.29.4 to 0.30.1
Bumps [k8s.io/kubectl](https://github.com/kubernetes/kubectl) from 0.29.4 to 0.30.1.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/kubernetes/kubectl/commit/60fb00bf74046682d819589497ee3fbb982ac568"><code>60fb00b</code></a> Update dependencies to v0.30.1 tag</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/0fe334ad70b454f330f8c80851c6e2e0ec8c290c"><code>0fe334a</code></a> Merge remote-tracking branch 'origin/master' into release-1.30</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/20502511128118df14119b3c144493243b0a221e"><code>2050251</code></a> Update x/net for CVE-2023-45288</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/f1fd500c1919b4fcb5e80d41426bed847c75a21b"><code>f1fd500</code></a> Bump github.com/golang/protobuf v1.5.4, google.golang.org/protobuf v1.33.0</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/d982873ba4ee524b7de2d31102732c792e8c5f99"><code>d982873</code></a> Merge pull request <a href="https://redirect.github.com/kubernetes/kubectl/issues/122653">#122653</a> from ardaguclu/interactive-delete-e2e-test</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/90ec7ffac46e52e4b72bd45b6cb4740b19966bd0"><code>90ec7ff</code></a> Merge pull request <a href="https://redirect.github.com/kubernetes/kubectl/issues/120346">#120346</a> from ardaguclu/custom-debug-profile</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/49f3ec973c3f8c7e4cda966bcfd2ea00d971f28a"><code>49f3ec9</code></a> Add e2e test for kubectl interactive delete</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/14d23cc87990b87b53563b6e4f15d304a624e190"><code>14d23cc</code></a> Add custom debug profiles on top of static profiles</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/2855ed117e2c8bf02238339e8034f3bb99350990"><code>2855ed1</code></a> Merge pull request <a href="https://redirect.github.com/kubernetes/kubectl/issues/123413">#123413</a> from seans3/tunneling-spdy-websockets</li>
<li><a href="https://github.com/kubernetes/kubectl/commit/18888dae3e71e4b11789ca162f79415d213c666a"><code>18888da</code></a> adds comments to tunnelingResponseWriter</li>
<li>Additional commits viewable in <a href="https://github.com/kubernetes/kubectl/compare/v0.29.4...v0.30.1">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=k8s.io/kubectl&package-manager=go_modules&previous-version=0.29.4&new-version=0.30.1)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/.github/workflows/checks.yaml b/.github/workflows/checks.yaml
index 138dafcc2..5c8578cf7 100644
--- a/.github/workflows/checks.yaml
+++ b/.github/workflows/checks.yaml
@@ -33,7 +33,7 @@ jobs:
       - name: Check if PR is up to date, if it is skip workflows for this ref
         id: 'up-to-date'
         if: github.event_name == 'push' && startsWith(github.ref, 'refs/heads/')
-        uses: Kong/public-shared-actions/pr-previews/up-to-date@v2.2.3
+        uses: Kong/public-shared-actions/pr-previews/up-to-date@v2.3.0
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
 
diff --git a/.golangci.yaml b/.golangci.yaml
index 85efd4395..82d98e060 100644
--- a/.golangci.yaml
+++ b/.golangci.yaml
@@ -159,6 +159,9 @@ issues:
    - (Expect directory permissions to be 0750 or less|Expect file permissions to be 0600 or less)
    # EXC0010 gosec: False positive is triggered by 'src, err := ioutil.ReadFile(filename)'
    - Potential file inclusion via variable
+   # G601 Implicit memory aliasing in for loop
+   # Irrelevant as of Go 1.22
+   - (G601)
 
   exclude-rules:
   # Ignore insecure TLS in tests and hardcoded credentials
diff --git a/.tools_versions.yaml b/.tools_versions.yaml
index 93d67392d..ec0378e39 100644
--- a/.tools_versions.yaml
+++ b/.tools_versions.yaml
@@ -5,7 +5,7 @@ controller-tools: "0.15.0"
 # renovate: datasource=github-releases depName=kubernetes-sigs/kustomize
 kustomize: "5.3.0"
 # renovate: datasource=github-releases depName=golangci/golangci-lint
-golangci-lint: "1.58.1"
+golangci-lint: "1.58.2"
 # renovate: datasource=github-releases depName=GoogleContainerTools/skaffold
 skaffold: "2.12.0"
 # renovate: datasource=github-releases depName=kubernetes-sigs/controller-runtime
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 84479a42a..98b6f2eb9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -137,13 +137,27 @@ Adding a new version? You'll need three changes:
   [#5982](https://github.com/Kong/kubernetes-ingress-controller/pull/5982)
   [#6006](https://github.com/Kong/kubernetes-ingress-controller/pull/6006)
 - Added `FallbackConfiguration` feature gate to enable the controller to generate a fallback configuration
-  for Kong when it fails to apply the configuration. The feature gate is disabled by default.
+  for Kong when it fails to apply the original one. The feature gate is disabled by default.
   [#5993](https://github.com/Kong/kubernetes-ingress-controller/pull/5993)
+  [#6010](https://github.com/Kong/kubernetes-ingress-controller/pull/6010)
+  [#6047](https://github.com/Kong/kubernetes-ingress-controller/pull/6047)
+  
 - Add support for Kubernetes Gateway API v1.1:
   - add a flag `--enable-controller-gwapi-grpcroute` to control whether enable or disable GRPCRoute controller.
   - add support for `GRPCRoute` v1, which requires users to upgrade the Gateway API's CRD to v1.1.
   [#5918](https://github.com/Kong/kubernetes-ingress-controller/pull/5918)
-
+- The `konghq.com/plugins` annotation supports a new `<namespace>:<name>`
+  format. This format requests a KongPlugin from a remote namespace. Binding
+  plugins across namespaces requires a ReferenceGrant from the requesting
+  resource to KongPlugins in the target namespace. This approach is useful for
+  some plugins bound to different types of entities, such as a set of
+  rate-limiting plugins applied to a service and various consumers. The
+  cross-namespace grant allows the service manager to define different limits
+  for consumers managed by other users without requiring those users to create
+  consumers in the Service's namespace.
+  [#5965](https://github.com/Kong/kubernetes-ingress-controller/pull/5965)
+- Fallback configuration no longer omits licenses and vaults.
+  [#6048](https://github.com/Kong/kubernetes-ingress-controller/pull/6048)
 
 ### Fixed
 
diff --git a/Makefile b/Makefile
index 2e3dc904c..b80a59117 100644
--- a/Makefile
+++ b/Makefile
@@ -62,7 +62,7 @@ mise:
 	@mise -V >/dev/null || (echo "mise - https://github.com/jdx/mise - not found. Please install it." && exit 1)
 
 .PHONY: tools
-tools: controller-gen kustomize client-gen golangci-lint.download gotestsum crd-ref-docs skaffold looppointer.download staticcheck.download
+tools: controller-gen kustomize client-gen golangci-lint.download gotestsum crd-ref-docs skaffold staticcheck.download
 
 export MISE_DATA_DIR = $(PROJECT_DIR)/bin/
 
@@ -161,11 +161,6 @@ go-junit-report: ## Download go-junit-report locally if necessary.
 	@$(MAKE) mise-plugin-install DEP=go-junit-report URL=https://github.com/pmalek/asdf-go-junit-report.git
 	@$(MISE) install go-junit-report@$(GOJUNIT_REPORT_VERSION)
 
-LOOPPOINTER= $(PROJECT_DIR)/bin/looppointer
-.PHONY: looppointer.download
-looppointer.download: ## Download looppointer locally if necessary.
-	@$(MAKE) _download_tool TOOL=looppointer
-
 # ------------------------------------------------------------------------------
 # Build
 # ------------------------------------------------------------------------------
@@ -216,7 +211,7 @@ fmt:
 	go fmt ./...
 
 .PHONY: lint
-lint: verify.tidy golangci-lint staticcheck looppointer
+lint: verify.tidy golangci-lint staticcheck 
 
 .PHONY: golangci-lint
 golangci-lint: golangci-lint.download
@@ -229,9 +224,6 @@ staticcheck: staticcheck.download
 		grep -F -e internal/konnect/controlplanes -v | \
 		xargs $(STATICCHECK) -tags envtest,e2e_tests,integration_tests,istio_tests,conformance_tests -f stylish
 
-looppointer: looppointer.download
-	$(LOOPPOINTER) -v ./internal/... ./test/...
-
 .PHONY: verify.tidy
 verify.tidy:
 	./scripts/verify-tidy.sh
diff --git a/examples/cross-namespace-plugin-grant.yaml b/examples/cross-namespace-plugin-grant.yaml
new file mode 100644
index 000000000..956c532e7
--- /dev/null
+++ b/examples/cross-namespace-plugin-grant.yaml
@@ -0,0 +1,166 @@
+# Service A, Consumer 1 in Namespace Foo, Service B in Namespace Bar (KongPlugin)
+# With the original behavior, KIC provided no means for the Service in Bar to use the KongPlugin in Foo.
+# With the modified behavior, the Service in Bar can use namespace:plugin_name to request another
+# namespace's plugin. Functionally, the Service in Bar behaves the same as a Service in Foo for the
+# purpose of plugin binding, i.e. the Service in Bar can create a multi-entity plugin with the KongConsumer
+# in Foo same as the Service in Foo can, if it has permission. Permissions are ReferenceGrants to
+# the KongPlugin:
+# - In namespace Foo, grant from Bar:Services to KongPlugins.
+# These grants may optionally use the standard ReferenceGrant specific target syntax to limit which
+# plugins external resources can bind to. Because this KongPlugin inherently has access to only
+# resources in its namespace, permissions to it are a clearer indication of access to those resources
+# than the ambiguous KongClusterPlugin situation, and we do not need bidirectional permissions.
+
+# Service A in Foo, Service B in Bar, Consumer in Baz
+# This doesn't differ much for KongPlugins either, in that you simply need an additional grant from
+# both other resource namespaces. You do need to choose whether the plugin resides in the KongConsumer
+# namespace or the Service namespaces. Functionally this doesn't change much other than the permission
+# configuration, though it may make sense to choose one or the other based other resources in the from
+# namespace. Note that you can still vary configuration if you choose the consumer namespace, you just
+# need multiple KongPlugins to do so.
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: one
+  namespace: pwuh
+  labels:
+    app: httpbin
+  annotations:
+    konghq.com/plugins: test
+spec:
+  ports:
+  - name: http
+    port: 80
+    targetPort: 80
+  selector:
+    app: httpbin
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: two
+  namespace: fwuh
+  labels:
+    app: httpbin
+  annotations:
+    konghq.com/plugins: pwuh:test
+spec:
+  ports:
+  - name: http
+    port: 80
+    targetPort: 80
+  selector:
+    app: httpbin
+---
+apiVersion: configuration.konghq.com/v1
+kind: KongConsumer
+metadata:
+  name: admin
+  namespace: pwuh
+  annotations:
+    kubernetes.io/ingress.class: kong
+    konghq.com/plugins: test
+username: admin
+---
+apiVersion: configuration.konghq.com/v1
+kind: KongPlugin
+metadata:
+  name: test
+  namespace: pwuh
+plugin: udp-log
+config:
+  host: example.com
+  port: 80
+---
+kind: ReferenceGrant
+metadata:
+  name: example
+  namespace: pwuh
+spec:
+  from:
+  - group: ""
+    kind: Service
+    namespace: fwuh
+  to:
+  - group: "configuration.konghq.com"
+    kind: KongPlugin
+    name: test # Optional, to limit external resources to a specific KongPlugin only.
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: httpbin
+  namespace: pwuh
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: httpbin
+  template:
+    metadata:
+      labels:
+        app: httpbin
+    spec:
+      containers:
+      - image: docker.io/kennethreitz/httpbin
+        name: httpbin
+        ports:
+        - containerPort: 80
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: httpbin
+  namespace: pwuh
+spec:
+  ingressClassName: kong
+  rules:
+  - http:
+      paths:
+      - path: /one
+        pathType: ImplementationSpecific
+        backend:
+          service:
+            name: one
+            port:
+              number: 80
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: httpbin
+  namespace: fwuh
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: httpbin
+  template:
+    metadata:
+      labels:
+        app: httpbin
+    spec:
+      containers:
+      - image: docker.io/kennethreitz/httpbin
+        name: httpbin
+        ports:
+        - containerPort: 80
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: httpbin
+  namespace: fwuh
+spec:
+  ingressClassName: kong
+  rules:
+  - http:
+      paths:
+      - path: /two
+        pathType: ImplementationSpecific
+        backend:
+          service:
+            name: two
+            port:
+              number: 80
diff --git a/go.mod b/go.mod
index 3e22329c7..27272143e 100644
--- a/go.mod
+++ b/go.mod
@@ -26,9 +26,9 @@ require (
 	github.com/avast/retry-go/v4 v4.6.0
 	github.com/blang/semver/v4 v4.0.0
 	github.com/dominikbraun/graph v0.23.0
-	github.com/go-logr/logr v1.4.1
+	github.com/go-logr/logr v1.4.2
 	github.com/go-logr/zapr v1.3.0
-	github.com/goccy/go-json v0.10.2
+	github.com/goccy/go-json v0.10.3
 	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.6.0
 	github.com/jpillora/backoff v1.0.0
@@ -95,7 +95,7 @@ require (
 	github.com/cpuguy83/dockercfg v0.3.1 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/distribution/reference v0.5.0 // indirect
-	github.com/docker/docker v26.1.2+incompatible
+	github.com/docker/docker v26.1.3+incompatible
 	github.com/docker/go-connections v0.5.0
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/emicklei/go-restful/v3 v3.12.0 // indirect
@@ -216,16 +216,16 @@ require (
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240506185236-b8a5c65736ae // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240506185236-b8a5c65736ae // indirect
-	google.golang.org/grpc v1.63.2
+	google.golang.org/grpc v1.64.0
 	google.golang.org/protobuf v1.34.1 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/cli-runtime v0.29.4
+	k8s.io/cli-runtime v0.30.1
 	k8s.io/klog/v2 v2.120.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20240430033511-f0e62f92d13f // indirect
-	k8s.io/kubectl v0.29.4
+	k8s.io/kubectl v0.30.1
 	k8s.io/utils v0.0.0-20240502163921-fe8a2dddb1d0 // indirect
 	sigs.k8s.io/e2e-framework v0.3.1-0.20231113122213-262cac32d35e
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
diff --git a/go.sum b/go.sum
index bbbe5eef2..2be2b0d86 100644
--- a/go.sum
+++ b/go.sum
@@ -78,8 +78,8 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/distribution/reference v0.5.0 h1:/FUIFXtfc/x2gpa5/VGfiGLuOIdYa1t65IKK2OFGvA0=
 github.com/distribution/reference v0.5.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
-github.com/docker/docker v26.1.2+incompatible h1:UVX5ZOrrfTGZZYEP+ZDq3Xn9PdHNXaSYMFPDumMqG2k=
-github.com/docker/docker v26.1.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v26.1.3+incompatible h1:lLCzRbrVZrljpVNobJu1J2FHk8V0s4BawoZippkc+xo=
+github.com/docker/docker v26.1.3+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
 github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@@ -120,8 +120,8 @@ github.com/gammazero/workerpool v1.1.3/go.mod h1:wPjyBLDbyKnUn2XwwyD3EEwo9dHutia
 github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA=
 github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
-github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
-github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
+github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-logr/zapr v1.3.0 h1:XGdV8XW8zdwFiwOA2Dryh1gj2KRQyOOoNmBy4EplIcQ=
@@ -138,8 +138,8 @@ github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg78
 github.com/go-task/slim-sprig v2.20.0+incompatible h1:4Xh3bDzO29j4TWNOI+24ubc0vbVFMg2PMnXKxK54/CA=
 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/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
-github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
+github.com/goccy/go-json v0.10.3 h1:KZ5WoDbxAIgm2HNbYckL0se1fHD6rz5j4ywS6ebzDqA=
+github.com/goccy/go-json v0.10.3/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
@@ -599,8 +599,8 @@ google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyac
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.63.2 h1:MUeiw1B2maTVZthpU5xvASfTh3LDbxHd6IJ6QQVU+xM=
-google.golang.org/grpc v1.63.2/go.mod h1:WAX/8DgncnokcFUldAxq7GeB5DXHDbMF+lLvDomNkRA=
+google.golang.org/grpc v1.64.0 h1:KH3VH9y/MgNQg1dE7b3XfVK0GsPSIzJwdF617gUSbvY=
+google.golang.org/grpc v1.64.0/go.mod h1:oxjF8E3FBnjp+/gVFYdWacaLDx9na1aqy9oovLpxQYg=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
@@ -643,8 +643,8 @@ k8s.io/apiextensions-apiserver v0.30.1 h1:4fAJZ9985BmpJG6PkoxVRpXv9vmPUOVzl614xa
 k8s.io/apiextensions-apiserver v0.30.1/go.mod h1:R4GuSrlhgq43oRY9sF2IToFh7PVlF1JjfWdoG3pixk4=
 k8s.io/apimachinery v0.30.1 h1:ZQStsEfo4n65yAdlGTfP/uSHMQSoYzU/oeEbkmF7P2U=
 k8s.io/apimachinery v0.30.1/go.mod h1:iexa2somDaxdnj7bha06bhb43Zpa6eWH8N8dbqVjTUc=
-k8s.io/cli-runtime v0.29.4 h1:QvUrddBxVX6XFJ6z64cGpEk7e4bQduKweqbqq+qBd9g=
-k8s.io/cli-runtime v0.29.4/go.mod h1:NmklYuZ4DLfOr2XEIT8Nzl883KMZUCv7KMj3wMHayCA=
+k8s.io/cli-runtime v0.30.1 h1:kSBBpfrJGS6lllc24KeniI9JN7ckOOJKnmFYH1RpTOw=
+k8s.io/cli-runtime v0.30.1/go.mod h1:zhHgbqI4J00pxb6gM3gJPVf2ysDjhQmQtnTxnMScab8=
 k8s.io/client-go v0.30.1 h1:uC/Ir6A3R46wdkgCV3vbLyNOYyCJ8oZnjtJGKfytl/Q=
 k8s.io/client-go v0.30.1/go.mod h1:wrAqLNs2trwiCH/wxxmT/x3hKVH9PuV0GGW0oDoHVqc=
 k8s.io/component-base v0.30.1 h1:bvAtlPh1UrdaZL20D9+sWxsJljMi0QZ3Lmw+kmZAaxQ=
@@ -653,8 +653,8 @@ k8s.io/klog/v2 v2.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw=
 k8s.io/klog/v2 v2.120.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-openapi v0.0.0-20240430033511-f0e62f92d13f h1:0LQagt0gDpKqvIkAMPaRGcXawNMouPECM1+F9BVxEaM=
 k8s.io/kube-openapi v0.0.0-20240430033511-f0e62f92d13f/go.mod h1:S9tOR0FxgyusSNR+MboCuiDpVWkAifZvaYI1Q2ubgro=
-k8s.io/kubectl v0.29.4 h1:2LFrAznoDZjN8JFMSUcuhER5o+yjTLzWWbOiDzVjmd8=
-k8s.io/kubectl v0.29.4/go.mod h1:YTKRF9y1/ccqZ2bnpOWaJD8V7johKqZR/qOMq+0pfxU=
+k8s.io/kubectl v0.30.1 h1:sHFIRI3oP0FFZmBAVEE8ErjnTyXDPkBcvO88mH9RjuY=
+k8s.io/kubectl v0.30.1/go.mod h1:7j+L0Cc38RYEcx+WH3y44jRBe1Q1jxdGPKkX0h4iDq0=
 k8s.io/utils v0.0.0-20240502163921-fe8a2dddb1d0 h1:jgGTlFYnhF1PM1Ax/lAlxUPE+KfCIXHaathvJg1C3ak=
 k8s.io/utils v0.0.0-20240502163921-fe8a2dddb1d0/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 sigs.k8s.io/controller-runtime v0.18.2 h1:RqVW6Kpeaji67CY5nPEfRz6ZfFMk0lWQlNrLqlNpx+Q=
diff --git a/internal/admission/validation/gateway/httproute.go b/internal/admission/validation/gateway/httproute.go
index 5c897b64d..70c676caa 100644
--- a/internal/admission/validation/gateway/httproute.go
+++ b/internal/admission/validation/gateway/httproute.go
@@ -146,21 +146,11 @@ func validateHTTPRouteFeatures(httproute *gatewayapi.HTTPRoute, translatorFeatur
 	for ruleIndex, rule := range httproute.Spec.Rules {
 		// Filter RequestMirror is not supported.
 
-		// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/3686
-		// For URLRewrite, only FullPathHTTPPathModifier is supported.
 		for filterIndex, filter := range rule.Filters {
 			if _, unsupported := unsupportedFilterMap[filter.Type]; unsupported {
 				return fmt.Errorf("rules[%d].filters[%d]: filter type %s is unsupported",
 					ruleIndex, filterIndex, filter.Type)
 			}
-
-			if filter.Type == gatewayapi.HTTPRouteFilterURLRewrite && filter.URLRewrite != nil {
-				// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/3685
-				if filter.URLRewrite.Hostname != nil {
-					return fmt.Errorf("rules[%d].filters[%d]: filter type %s (with hostname replace) is unsupported",
-						ruleIndex, filterIndex, filter.Type)
-				}
-			}
 		}
 
 		for refIndex, ref := range rule.BackendRefs {
diff --git a/internal/annotations/annotations.go b/internal/annotations/annotations.go
index c5db8a2da..f76f0e13c 100644
--- a/internal/annotations/annotations.go
+++ b/internal/annotations/annotations.go
@@ -23,6 +23,7 @@ import (
 	"github.com/samber/lo"
 	netv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8stypes "k8s.io/apimachinery/pkg/types"
 
 	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 )
@@ -155,6 +156,34 @@ func ExtractKongPluginsFromAnnotations(anns map[string]string) []string {
 	return kongPluginCRs
 }
 
+type NamespacedKongPlugin k8stypes.NamespacedName
+
+// ExtractNamespacedKongPluginsFromAnnotations extracts a KongPlugin name and optional namespace from an annotation
+// value. Plugins are delimited by ",". Values are either colon-delimited "namespace:name" strings or name-only
+// strings.
+func ExtractNamespacedKongPluginsFromAnnotations(anns map[string]string) []NamespacedKongPlugin {
+	v := pluginsFromAnnotations(anns)
+	if v == "" {
+		return nil
+	}
+	split := strings.Split(v, ",")
+	plugins := make([]NamespacedKongPlugin, 0, len(split))
+	for _, s := range split {
+		if s != "" {
+			plugin := NamespacedKongPlugin{}
+			if strings.Contains(s, ":") {
+				split := strings.Split(s, ":")
+				plugin.Namespace = strings.TrimSpace(split[0])
+				plugin.Name = strings.TrimSpace(split[1])
+			} else {
+				plugin.Name = strings.TrimSpace(s)
+			}
+			plugins = append(plugins, plugin)
+		}
+	}
+	return plugins
+}
+
 // ExtractConfigurationName extracts the name of the KongIngress object that holds
 // information about the configuration to use in Routes, Services and Upstreams.
 func ExtractConfigurationName(anns map[string]string) string {
@@ -392,6 +421,7 @@ func ExtractRewriteURI(anns map[string]string) (string, bool) {
 	return s, ok
 }
 
+// ExtractUpstreamPolicy extracts the upstream policy annotation value.
 func ExtractUpstreamPolicy(anns map[string]string) (string, bool) {
 	s, ok := anns[kongv1beta1.KongUpstreamPolicyAnnotationKey]
 	return s, ok
diff --git a/internal/dataplane/configfetcher/config_fetcher.go b/internal/dataplane/configfetcher/config_fetcher.go
index 9b1537af4..d36411fb7 100644
--- a/internal/dataplane/configfetcher/config_fetcher.go
+++ b/internal/dataplane/configfetcher/config_fetcher.go
@@ -13,6 +13,7 @@ import (
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/adminapi"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/kongstate"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/sendconfig"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/license"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
 )
 
@@ -36,6 +37,8 @@ type DefaultKongLastGoodConfigFetcher struct {
 	fillIDs bool
 	// workspace is the workspace name used in generating deterministic IDs. Only used when fillIDs = true.
 	workspace string
+	// licenseGetter is an optional license provider.
+	licenseGetter license.Getter
 }
 
 func NewDefaultKongLastGoodConfigFetcher(fillIDs bool, workspace string) *DefaultKongLastGoodConfigFetcher {
@@ -46,8 +49,52 @@ func NewDefaultKongLastGoodConfigFetcher(fillIDs bool, workspace string) *Defaul
 	}
 }
 
+// InjectLicenseGetter adds a license getter to the config fetcher.
+func (cf *DefaultKongLastGoodConfigFetcher) InjectLicenseGetter(licenseGetter license.Getter) {
+	cf.licenseGetter = licenseGetter
+}
+
 func (cf *DefaultKongLastGoodConfigFetcher) LastValidConfig() (*kongstate.KongState, bool) {
 	if cf.lastValidState != nil {
+		// TODO the translator version of this also has a condition on
+		// `t.featureFlags.EnterpriseEdition == true`. This isn't a typical feature
+		// flag, and is instead set based on whether we see we're talking to a
+		// kong-gateway image. It's not used for anything other than the license.
+		// Do we actually need this? While the OSS image does not recognize license
+		// entities, you'll only have license entities if you created a KongLicense
+		// or set up Konnect credentials that can pull one. In either of those
+		// cases you arguably _should_ be using the kong-gateway image, so we
+		// shouldn't really exclude it if you're not :thinking:
+		if cf.licenseGetter != nil {
+			optionalLicense := cf.licenseGetter.GetLicense()
+			if l, ok := optionalLicense.Get(); ok {
+				// TODO cf.lastValidState is a pointer, and a pointer to a rather large
+				// struct we probably don't want to copy. this assignment without
+				// copying it does actually _modify_ the last good config, and could
+				// technically transform it from valid to invalid in a contrived
+				// scenario where the config had a valid license and the environment
+				// provided an invalid one.
+				//
+				// The environment arguably shouldn't provide an invalid one, but we
+				// have seen Konnect provide invalid licenses (though we think this was
+				// a fluke due to a change in their license management code), and you
+				// could potentially overwrite your previously valid KongLicense with
+				// garbage.
+				//
+				// On the other hand, any license in the last valid config would simply
+				// be whatever license you had when the config was last valid. That
+				// license could have since expired, and attempting to use it would
+				// fail. Using the latest available license is the better option in
+				// that case.
+				//
+				// We can reasonably expect Konnect to provide valid licenses (Konnect
+				// needs to fix something if they aren't) and a broken KongLicense is
+				// something you should be able to fix easily: you should only have one
+				// of them managed by the superuser, so there shouldn't be any
+				// confusion around where it's coming from.
+				cf.lastValidState.Licenses = []kongstate.License{{License: l}}
+			}
+		}
 		return cf.lastValidState, true
 	}
 	return nil, false
diff --git a/internal/dataplane/configfetcher/kongrawstate.go b/internal/dataplane/configfetcher/kongrawstate.go
index 6e8090e9c..5872df2b1 100644
--- a/internal/dataplane/configfetcher/kongrawstate.go
+++ b/internal/dataplane/configfetcher/kongrawstate.go
@@ -75,6 +75,8 @@ func KongRawStateToKongState(rawstate *utils.KongRawState) *kongstate.KongState
 		kongState.Upstreams = append(kongState.Upstreams, sanitizeUpstream(newUpstream))
 	}
 
+	kongState.Vaults = rawVaultsToVaults(rawstate.Vaults)
+
 	kongState.CACertificates = rawCACertificatesToCACertificates(rawstate.CACertificates)
 	kongState.Certificates = rawCertificatesToCertificates(rawstate.Certificates)
 
@@ -221,6 +223,20 @@ func rawCACertificatesToCACertificates(caCertificates []*kong.CACertificate) []k
 	return certs
 }
 
+func rawVaultsToVaults(rawVaults []*kong.Vault) []kongstate.Vault {
+	if len(rawVaults) == 0 {
+		return nil
+	}
+	vaults := []kongstate.Vault{}
+
+	for _, v := range rawVaults {
+		vaults = append(vaults, kongstate.Vault{
+			Vault: sanitizeVault(*v),
+		})
+	}
+	return vaults
+}
+
 // -----------------------------------------------------------------------------
 // Sanitization functions
 // -----------------------------------------------------------------------------
@@ -271,6 +287,12 @@ func sanitizeCACertificate(caCertificate kong.CACertificate) kong.CACertificate
 	return caCertificate
 }
 
+func sanitizeVault(v kong.Vault) kong.Vault {
+	v.ID = nil
+	v.CreatedAt = nil
+	return v
+}
+
 func sanitizeConsumer(consumer kong.Consumer) kong.Consumer {
 	consumer.ID = nil
 	consumer.CreatedAt = nil
diff --git a/internal/dataplane/fallback/cache_to_graph.go b/internal/dataplane/fallback/cache_to_graph.go
new file mode 100644
index 000000000..e3de15a3e
--- /dev/null
+++ b/internal/dataplane/fallback/cache_to_graph.go
@@ -0,0 +1,62 @@
+package fallback
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/dominikbraun/graph"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+)
+
+// DefaultCacheGraphProvider is a default implementation of the CacheGraphProvider interface.
+type DefaultCacheGraphProvider struct{}
+
+func NewDefaultCacheGraphProvider() *DefaultCacheGraphProvider {
+	return &DefaultCacheGraphProvider{}
+}
+
+// CacheToGraph creates a new ConfigGraph from the given cache stores. It adds all objects
+// from the cache stores to the graph as vertices as well as edges between objects and their dependencies
+// resolved by the ResolveDependencies function.
+func (p DefaultCacheGraphProvider) CacheToGraph(c store.CacheStores) (*ConfigGraph, error) {
+	g := NewConfigGraph()
+
+	for _, s := range c.ListAllStores() {
+		for _, o := range s.List() {
+			obj, ok := o.(client.Object)
+			if !ok {
+				// Should not happen since all objects in the cache are client.Objects, but better safe than sorry.
+				return nil, fmt.Errorf("expected client.Object, got %T", o)
+			}
+			// Add the object to the graph. It can happen that the object is already in the graph (i.e. was already added
+			// as a dependency of another object), in which case we ignore the error.
+			if err := g.AddVertex(obj); err != nil && !errors.Is(err, graph.ErrVertexAlreadyExists) {
+				return nil, fmt.Errorf("failed to add %s to the graph: %w", GetObjectHash(obj), err)
+			}
+
+			deps, err := ResolveDependencies(c, obj)
+			if err != nil {
+				return nil, fmt.Errorf("failed to resolve dependencies for %s: %w", GetObjectHash(obj), err)
+			}
+			// Add the object's dependencies to the graph.
+			for _, dep := range deps {
+				// Add the dependency to the graph in case it wasn't added before. If it was added before, we ignore the
+				// error.
+				if err := g.AddVertex(dep); err != nil && !errors.Is(err, graph.ErrVertexAlreadyExists) {
+					return nil, fmt.Errorf("failed to add %s to the graph: %w", GetObjectHash(obj), err)
+				}
+
+				// Add an edge from a dependency to the object. If the edge was already added before, we ignore the error.
+				// It's on purpose that we add the edge from the dependency to the object, as it makes it easier to traverse
+				// the graph from the object to its dependants once it is broken.
+				if err := g.AddEdge(GetObjectHash(dep), GetObjectHash(obj)); err != nil && !errors.Is(err, graph.ErrEdgeAlreadyExists) {
+					return nil, fmt.Errorf("failed to add edge from %s to %s: %w", GetObjectHash(obj), GetObjectHash(dep), err)
+				}
+			}
+		}
+	}
+
+	return g, nil
+}
diff --git a/internal/dataplane/fallback/fallback.go b/internal/dataplane/fallback/fallback.go
new file mode 100644
index 000000000..35f31b16b
--- /dev/null
+++ b/internal/dataplane/fallback/fallback.go
@@ -0,0 +1,53 @@
+package fallback
+
+import (
+	"fmt"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+)
+
+type CacheGraphProvider interface {
+	// CacheToGraph returns a new ConfigGraph instance built from the given cache snapshot.
+	CacheToGraph(cache store.CacheStores) (*ConfigGraph, error)
+}
+
+// Generator is responsible for generating fallback cache snapshots.
+type Generator struct {
+	cacheGraphProvider CacheGraphProvider
+}
+
+func NewGenerator(cacheGraphProvider CacheGraphProvider) *Generator {
+	return &Generator{
+		cacheGraphProvider: cacheGraphProvider,
+	}
+}
+
+// GenerateExcludingAffected generates a new cache snapshot that excludes all objects that depend on the broken objects.
+func (g *Generator) GenerateExcludingAffected(
+	cache store.CacheStores,
+	brokenObjects []ObjectHash,
+) (store.CacheStores, error) {
+	graph, err := g.cacheGraphProvider.CacheToGraph(cache)
+	if err != nil {
+		return store.CacheStores{}, fmt.Errorf("failed to build cache graph: %w", err)
+	}
+
+	fallbackCache, err := cache.TakeSnapshot()
+	if err != nil {
+		return store.CacheStores{}, fmt.Errorf("failed to take cache snapshot: %w", err)
+	}
+
+	for _, brokenObject := range brokenObjects {
+		subgraphObjects, err := graph.SubgraphObjects(brokenObject)
+		if err != nil {
+			return store.CacheStores{}, fmt.Errorf("failed to find dependants for %s: %w", brokenObject, err)
+		}
+		for _, obj := range subgraphObjects {
+			if err := fallbackCache.Delete(obj); err != nil {
+				return store.CacheStores{}, fmt.Errorf("failed to delete %s from the cache: %w", GetObjectHash(obj), err)
+			}
+		}
+	}
+
+	return fallbackCache, nil
+}
diff --git a/internal/dataplane/fallback/graph.go b/internal/dataplane/fallback/graph.go
index 595eaae67..a6dbe50ed 100644
--- a/internal/dataplane/fallback/graph.go
+++ b/internal/dataplane/fallback/graph.go
@@ -1,15 +1,12 @@
 package fallback
 
 import (
-	"errors"
 	"fmt"
 
 	"github.com/dominikbraun/graph"
 	"github.com/samber/lo"
 	k8stypes "k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
 )
 
 // ConfigGraph is a graph representation of the Kubernetes resources kept in the cache.
@@ -23,6 +20,12 @@ type ConfigGraph struct {
 	graph graph.Graph[ObjectHash, client.Object]
 }
 
+func NewConfigGraph() *ConfigGraph {
+	return &ConfigGraph{
+		graph: graph.New[ObjectHash, client.Object](GetObjectHash, graph.Directed()),
+	}
+}
+
 // ObjectHash is a unique identifier for a given object that is used as a vertex key in the graph.
 // It could consist of the object's UID only, but we also include the object's kind, namespace and name
 // to make it easier to debug and understand the graph.
@@ -56,46 +59,14 @@ func GetObjectHash(obj client.Object) ObjectHash {
 	}
 }
 
-// NewConfigGraphFromCacheStores creates a new ConfigGraph from the given cache stores. It adds all objects
-// from the cache stores to the graph as vertices as well as edges between objects and their dependencies
-// resolved by the ResolveDependencies function.
-func NewConfigGraphFromCacheStores(c store.CacheStores) (*ConfigGraph, error) {
-	g := graph.New[ObjectHash, client.Object](GetObjectHash, graph.Directed())
-
-	for _, s := range c.ListAllStores() {
-		for _, o := range s.List() {
-			obj, ok := o.(client.Object)
-			if !ok {
-				// Should not happen since all objects in the cache are client.Objects, but better safe than sorry.
-				return nil, fmt.Errorf("expected client.Object, got %T", o)
-			}
-			// Add the object to the graph. It can happen that the object is already in the graph (i.e. was already added
-			// as a dependency of another object), in which case we ignore the error.
-			if err := g.AddVertex(obj); err != nil && !errors.Is(err, graph.ErrVertexAlreadyExists) {
-				return nil, fmt.Errorf("failed to add %s to the graph: %w", GetObjectHash(obj), err)
-			}
-
-			deps, err := ResolveDependencies(c, obj)
-			if err != nil {
-				return nil, fmt.Errorf("failed to resolve dependencies for %s: %w", GetObjectHash(obj), err)
-			}
-			// Add the object's dependencies to the graph.
-			for _, dep := range deps {
-				// Add the dependency to the graph in case it wasn't added before. If it was added before, we ignore the
-				// error.
-				if err := g.AddVertex(dep); err != nil && !errors.Is(err, graph.ErrVertexAlreadyExists) {
-					return nil, fmt.Errorf("failed to add %s to the graph: %w", GetObjectHash(obj), err)
-				}
-
-				// Add an edge from a dependency to the object. If the edge was already added before, we ignore the error.
-				if err := g.AddEdge(GetObjectHash(dep), GetObjectHash(obj)); err != nil && !errors.Is(err, graph.ErrEdgeAlreadyExists) {
-					return nil, fmt.Errorf("failed to add edge from %s to %s: %w", GetObjectHash(obj), GetObjectHash(dep), err)
-				}
-			}
-		}
-	}
+// AddVertex adds a vertex to the graph.
+func (g *ConfigGraph) AddVertex(obj client.Object) error {
+	return g.graph.AddVertex(obj)
+}
 
-	return &ConfigGraph{graph: g}, nil
+// AddEdge adds an edge between two vertices in the graph.
+func (g *ConfigGraph) AddEdge(from, to ObjectHash) error {
+	return g.graph.AddEdge(from, to)
 }
 
 // AdjacencyMap is a map of object hashes to their neighbours' hashes.
@@ -114,3 +85,20 @@ func (g *ConfigGraph) AdjacencyMap() (AdjacencyMap, error) {
 	}
 	return m, nil
 }
+
+// SubgraphObjects returns all objects in the graph reachable from the source object, including the source object.
+// It uses a depth-first search to traverse the graph.
+func (g *ConfigGraph) SubgraphObjects(sourceHash ObjectHash) ([]client.Object, error) {
+	var objects []client.Object
+	if err := graph.DFS(g.graph, sourceHash, func(hash ObjectHash) bool {
+		obj, err := g.graph.Vertex(hash)
+		if err != nil {
+			return false
+		}
+		objects = append(objects, obj)
+		return false
+	}); err != nil {
+		return nil, fmt.Errorf("failed to traverse the graph: %w", err)
+	}
+	return objects, nil
+}
diff --git a/internal/dataplane/fallback/graph_dependencies.go b/internal/dataplane/fallback/graph_dependencies.go
index 55476c537..77e285fcf 100644
--- a/internal/dataplane/fallback/graph_dependencies.go
+++ b/internal/dataplane/fallback/graph_dependencies.go
@@ -21,8 +21,12 @@ import (
 // Every supported object type should explicitly have a case in this function.
 func ResolveDependencies(cache store.CacheStores, obj client.Object) ([]client.Object, error) {
 	switch obj := obj.(type) {
+	// Standard Kubernetes objects.
+	case *corev1.Service:
+		return resolveServiceDependencies(cache, obj), nil
 	case *netv1.Ingress:
 		return resolveIngressDependencies(cache, obj), nil
+	// Gateway API objects.
 	case *gatewayapi.HTTPRoute:
 		return resolveHTTPRouteDependencies(cache, obj), nil
 	case *gatewayapi.TLSRoute:
@@ -33,24 +37,27 @@ func ResolveDependencies(cache store.CacheStores, obj client.Object) ([]client.O
 		return resolveUDPRouteDependencies(cache, obj), nil
 	case *gatewayapi.GRPCRoute:
 		return resolveGRPCRouteDependencies(cache, obj), nil
+	// Kong specific objects.
 	case *kongv1.KongPlugin:
 		return resolveKongPluginDependencies(cache, obj), nil
 	case *kongv1.KongClusterPlugin:
 		return resolveKongClusterPluginDependencies(cache, obj), nil
+	case *kongv1.KongConsumer:
+		return resolveKongConsumerDependencies(cache, obj), nil
+	case *kongv1beta1.KongConsumerGroup:
+		return resolveKongConsumerGroupDependencies(cache, obj), nil
 	case *kongv1beta1.UDPIngress:
 		return resolveUDPIngressDependencies(cache, obj), nil
 	case *kongv1beta1.TCPIngress:
 		return resolveTCPIngressDependencies(cache, obj), nil
 	case *incubatorv1alpha1.KongServiceFacade:
 		return resolveKongServiceFacadeDependencies(cache, obj), nil
-	case *netv1.IngressClass, // Object types that have no dependencies.
-		*corev1.Service,
+	// Object types that have no dependencies.
+	case *netv1.IngressClass,
 		*corev1.Secret,
 		*discoveryv1.EndpointSlice,
 		*gatewayapi.ReferenceGrant,
 		*gatewayapi.Gateway,
-		*kongv1.KongConsumer,
-		*kongv1beta1.KongConsumerGroup,
 		*kongv1.KongIngress,
 		*kongv1beta1.KongUpstreamPolicy,
 		*kongv1alpha1.IngressClassParameters,
diff --git a/internal/dataplane/fallback/graph_dependencies_common.go b/internal/dataplane/fallback/graph_dependencies_common.go
new file mode 100644
index 000000000..276140ff7
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_common.go
@@ -0,0 +1,58 @@
+package fallback
+
+import (
+	"fmt"
+
+	k8stypes "k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+)
+
+// resolveObjectDependenciesPlugin resolves KongPlugin and KongClusterPlugin dependencies for an arbitrary object
+// that refers them in its annotations.
+func resolveObjectDependenciesPlugin(cache store.CacheStores, obj client.Object) []client.Object {
+	var dependencies []client.Object
+	for _, pluginName := range annotations.ExtractKongPluginsFromAnnotations(obj.GetAnnotations()) {
+		// KongPlugin is tied to a namespace.
+		if plugin, exists, err := cache.Plugin.GetByKey(
+			fmt.Sprintf("%s/%s", obj.GetNamespace(), pluginName),
+		); err == nil && exists {
+			dependencies = append(dependencies, plugin.(client.Object))
+			// A namespaced KongPlugin resource takes priority over a KongClusterPlugin with the same name
+			// source: https://docs.konghq.com/kubernetes-ingress-controller/latest/concepts/custom-resources/#kongclusterplugin
+			// so it's desired to skip the KongClusterPlugin lookup if a KongPlugin is found.
+			continue
+		}
+		// KongClusterPlugin is global.
+		if plugin, exists, err := cache.ClusterPlugin.GetByKey(pluginName); err == nil && exists {
+			dependencies = append(dependencies, plugin.(client.Object))
+		}
+	}
+	return dependencies
+}
+
+// resolveDependenciesForServiceLikeObj resolves potential dependencies for a Service-like objects use for Service or KongServiceFacade.
+// Potential dependencies are:
+// - KongPlugin
+// - KongClusterPlugin
+// - KongUpstreamPolicy.
+func resolveDependenciesForServiceLikeObj(cache store.CacheStores, obj client.Object) []client.Object {
+	dependencies := resolveObjectDependenciesPlugin(cache, obj)
+	if kupName, ok := annotations.ExtractUpstreamPolicy(obj.GetAnnotations()); ok {
+		if kup, exists, err := cache.KongUpstreamPolicy.GetByKey(fmt.Sprintf("%s/%s", obj.GetNamespace(), kupName)); err == nil && exists {
+			dependencies = append(dependencies, kup.(client.Object))
+		}
+	}
+	return dependencies
+}
+
+// fetchSecret retrieves a Secret object as client.Object from the cache.
+func fetchSecret(cache store.CacheStores, nn k8stypes.NamespacedName) (client.Object, bool) {
+	secret, exists, err := cache.Secret.GetByKey(nn.String())
+	if err != nil || !exists {
+		return nil, false
+	}
+	return secret.(client.Object), true
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_gatewayapi.go b/internal/dataplane/fallback/graph_dependencies_gatewayapi.go
index 94324e3ad..1620b9a63 100644
--- a/internal/dataplane/fallback/graph_dependencies_gatewayapi.go
+++ b/internal/dataplane/fallback/graph_dependencies_gatewayapi.go
@@ -1,33 +1,137 @@
 package fallback
 
 import (
+	"fmt"
+	"slices"
+
+	"github.com/samber/lo"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
 )
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
-func resolveHTTPRouteDependencies(_ store.CacheStores, _ *gatewayapi.HTTPRoute) []client.Object {
-	return nil
+// resolveHTTPRouteDependencies resolves potential dependencies for a given HTTPRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveHTTPRouteDependencies(cache store.CacheStores, route *gatewayapi.HTTPRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getHTTPRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// resolveTCPRouteDependencies resolves potential dependencies for a given TCPRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveTCPRouteDependencies(cache store.CacheStores, route *gatewayapi.TCPRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getTCPRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// resolveUDPRouteDependencies resolves potential dependencies for a given UDPRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveUDPRouteDependencies(cache store.CacheStores, route *gatewayapi.UDPRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getUDPRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// resolveTLSRouteDependencies resolves potential dependencies for a given TLSRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveTLSRouteDependencies(cache store.CacheStores, route *gatewayapi.TLSRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getTLSRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// resolveGRPCRouteDependencies resolves potential dependencies for a given GRPCRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveGRPCRouteDependencies(cache store.CacheStores, route *gatewayapi.GRPCRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getGRPCRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// gatewayAPIRoute is an interface that represents a GatewayAPI Route object.
+type gatewayAPIRoute interface {
+	client.Object
+	*gatewayapi.HTTPRoute | *gatewayapi.TCPRoute | *gatewayapi.UDPRoute | *gatewayapi.TLSRoute | *gatewayapi.GRPCRoute
+}
+
+// resolveGatewayAPIRouteDependenciesBackendRefs resolves backend references for a given gatewayAPIRoute object.
+func resolveGatewayAPIRouteDependenciesBackendRefs[T gatewayAPIRoute](cache store.CacheStores, route T, backendRefs []gatewayapi.BackendRef) []client.Object {
+	var dependencies []client.Object
+	for _, backendRef := range backendRefs {
+		if !util.IsBackendRefGroupKindSupported(backendRef.Group, backendRef.Kind) {
+			continue
+		}
+		ns := route.GetNamespace()
+		if backendRef.Namespace != nil {
+			ns = string(*backendRef.Namespace)
+		}
+		ingressClass, exists, err := cache.Service.GetByKey(fmt.Sprintf("%s/%s", ns, backendRef.Name))
+		if err == nil && exists {
+			dependencies = append(dependencies, ingressClass.(client.Object))
+		}
+	}
+	return dependencies
+}
+
+func getHTTPRouteBackendRefs(route *gatewayapi.HTTPRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, lo.Map(rule.BackendRefs, func(b gatewayapi.HTTPBackendRef, _ int) gatewayapi.BackendRef {
+			return b.BackendRef
+		})...)
+	}
+	return backendRefs
 }
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
-func resolveTLSRouteDependencies(_ store.CacheStores, _ *gatewayapi.TLSRoute) []client.Object {
-	return nil
+func getTCPRouteBackendRefs(route *gatewayapi.TCPRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, rule.BackendRefs...)
+	}
+	return backendRefs
 }
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
-func resolveTCPRouteDependencies(_ store.CacheStores, _ *gatewayapi.TCPRoute) []client.Object {
-	return nil
+func getUDPRouteBackendRefs(route *gatewayapi.UDPRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, rule.BackendRefs...)
+	}
+	return backendRefs
 }
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
-func resolveUDPRouteDependencies(_ store.CacheStores, _ *gatewayapi.UDPRoute) []client.Object {
-	return nil
+func getTLSRouteBackendRefs(route *gatewayapi.TLSRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, rule.BackendRefs...)
+	}
+	return backendRefs
 }
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
-func resolveGRPCRouteDependencies(_ store.CacheStores, _ *gatewayapi.GRPCRoute) []client.Object {
-	return nil
+func getGRPCRouteBackendRefs(route *gatewayapi.GRPCRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, lo.Map(rule.BackendRefs, func(b gatewayapi.GRPCBackendRef, _ int) gatewayapi.BackendRef {
+			return b.BackendRef
+		})...)
+	}
+	return backendRefs
 }
diff --git a/internal/dataplane/fallback/graph_dependencies_ingress.go b/internal/dataplane/fallback/graph_dependencies_ingress.go
index d97c6541e..21b8f1f7f 100644
--- a/internal/dataplane/fallback/graph_dependencies_ingress.go
+++ b/internal/dataplane/fallback/graph_dependencies_ingress.go
@@ -24,7 +24,7 @@ func resolveIngressDependencies(cache store.CacheStores, ingress *netv1.Ingress)
 		resolveIngressDependenciesIngressClass(cache, ingress),
 		resolveIngressDependenciesService(cache, ingress),
 		resolveIngressDependenciesKongUpstreamPolicy(cache, ingress),
-		resolveIngressDependenciesPlugin(cache, ingress),
+		resolveObjectDependenciesPlugin(cache, ingress),
 	)
 }
 
@@ -90,20 +90,3 @@ func resolveIngressDependenciesKongUpstreamPolicy(cache store.CacheStores, ingre
 	}
 	return nil
 }
-
-// resolveIngressDependenciesPlugin resolves KongPlugin and KongClusterPlugin dependencies for an Ingress object.
-func resolveIngressDependenciesPlugin(cache store.CacheStores, ingress *netv1.Ingress) []client.Object {
-	var dependencies []client.Object
-	for _, pluginName := range annotations.ExtractKongPluginsFromAnnotations(ingress.Annotations) {
-		if plugin, exists, err := cache.Plugin.GetByKey(
-			fmt.Sprintf("%s/%s", ingress.GetNamespace(), pluginName),
-		); err == nil && exists {
-			dependencies = append(dependencies, plugin.(client.Object))
-		}
-
-		if plugin, exists, err := cache.ClusterPlugin.GetByKey(pluginName); err == nil && exists {
-			dependencies = append(dependencies, plugin.(client.Object))
-		}
-	}
-	return dependencies
-}
diff --git a/internal/dataplane/fallback/graph_dependencies_kong.go b/internal/dataplane/fallback/graph_dependencies_kong.go
index ddcf3f7f5..225ee8022 100644
--- a/internal/dataplane/fallback/graph_dependencies_kong.go
+++ b/internal/dataplane/fallback/graph_dependencies_kong.go
@@ -1,6 +1,7 @@
 package fallback
 
 import (
+	k8stypes "k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
@@ -9,27 +10,92 @@ import (
 	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
 )
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
-func resolveKongPluginDependencies(_ store.CacheStores, _ *kongv1.KongPlugin) []client.Object {
-	return nil
+// resolveKongPluginDependencies resolves potential dependencies for a KongPlugin object:
+// - Secret.
+func resolveKongPluginDependencies(cache store.CacheStores, kongPlugin *kongv1.KongPlugin) []client.Object {
+	var dependencies []client.Object
+	if cf := kongPlugin.ConfigFrom; cf != nil {
+		if s, ok := fetchSecret(
+			cache,
+			k8stypes.NamespacedName{
+				Namespace: kongPlugin.Namespace,
+				Name:      cf.SecretValue.Secret,
+			},
+		); ok {
+			dependencies = append(dependencies, s)
+		}
+	}
+	for _, cp := range kongPlugin.ConfigPatches {
+		if s, ok := fetchSecret(
+			cache,
+			k8stypes.NamespacedName{
+				Namespace: kongPlugin.Namespace,
+				Name:      cp.ValueFrom.SecretValue.Secret,
+			},
+		); ok {
+			dependencies = append(dependencies, s)
+		}
+	}
+	return dependencies
 }
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
-func resolveKongClusterPluginDependencies(_ store.CacheStores, _ *kongv1.KongClusterPlugin) []client.Object {
-	return nil
+// resolveKongClusterPluginDependencies resolves potential dependencies for a KongClusterPlugin object:
+// - Secret.
+func resolveKongClusterPluginDependencies(cache store.CacheStores, kongClusterPlugin *kongv1.KongClusterPlugin) []client.Object {
+	var dependencies []client.Object
+	if cf := kongClusterPlugin.ConfigFrom; cf != nil {
+		if s, ok := fetchSecret(
+			cache,
+			k8stypes.NamespacedName{
+				Namespace: cf.SecretValue.Namespace,
+				Name:      cf.SecretValue.Secret,
+			},
+		); ok {
+			dependencies = append(dependencies, s)
+		}
+	}
+	for _, cp := range kongClusterPlugin.ConfigPatches {
+		if s, ok := fetchSecret(
+			cache,
+			k8stypes.NamespacedName{
+				Namespace: cp.ValueFrom.SecretValue.Namespace,
+				Name:      cp.ValueFrom.SecretValue.Secret,
+			},
+		); ok {
+			dependencies = append(dependencies, s)
+		}
+	}
+	return dependencies
+}
+
+// resolveKongConsumerDependencies resolves potential dependencies for a KongConsumer object:
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveKongConsumerDependencies(cache store.CacheStores, kongConsumer *kongv1.KongConsumer) []client.Object {
+	return resolveObjectDependenciesPlugin(cache, kongConsumer)
 }
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
+// resolveKongConsumerGroupDependencies resolves potential dependencies for a KongConsumerGroup object:
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveKongConsumerGroupDependencies(cache store.CacheStores, kongConsumerGroup *kongv1beta1.KongConsumerGroup) []client.Object {
+	return resolveObjectDependenciesPlugin(cache, kongConsumerGroup)
+}
+
+// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/5929
 func resolveUDPIngressDependencies(_ store.CacheStores, _ *kongv1beta1.UDPIngress) []client.Object {
 	return nil
 }
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
+// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/5929
 func resolveTCPIngressDependencies(_ store.CacheStores, _ *kongv1beta1.TCPIngress) []client.Object {
 	return nil
 }
 
-// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
-func resolveKongServiceFacadeDependencies(_ store.CacheStores, _ *incubatorv1alpha1.KongServiceFacade) []client.Object {
-	return nil
+// resolveKongServiceFacadeDependencies resolves potential dependencies for a KongServiceFacade object:
+// - KongPlugin
+// - KongClusterPlugin
+// - KongUpstreamPolicy.
+func resolveKongServiceFacadeDependencies(cache store.CacheStores, kongServiceFacade *incubatorv1alpha1.KongServiceFacade) []client.Object {
+	return resolveDependenciesForServiceLikeObj(cache, kongServiceFacade)
 }
diff --git a/internal/dataplane/fallback/graph_dependencies_service.go b/internal/dataplane/fallback/graph_dependencies_service.go
new file mode 100644
index 000000000..11dc74f2c
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_service.go
@@ -0,0 +1,16 @@
+package fallback
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+)
+
+// resolveServiceDependencies resolves potential dependencies for a Service object:
+// - KongPlugin
+// - KongClusterPlugin
+// - KongUpstreamPolicy.
+func resolveServiceDependencies(cache store.CacheStores, service *corev1.Service) []client.Object {
+	return resolveDependenciesForServiceLikeObj(cache, service)
+}
diff --git a/internal/dataplane/kong_client.go b/internal/dataplane/kong_client.go
index 54fc5d823..e9f415673 100644
--- a/internal/dataplane/kong_client.go
+++ b/internal/dataplane/kong_client.go
@@ -598,22 +598,23 @@ func (c *KongClient) sendToClient(
 		c.recordApplyConfigurationEvents(err, client.BaseRootURL())
 	}
 	if err != nil {
-		var updateErr sendconfig.UpdateError
+		var (
+			rawResponseBody    []byte
+			updateErr          sendconfig.UpdateError
+			responseParsingErr sendconfig.ResponseParsingError
+		)
 		if errors.As(err, &updateErr) {
-			c.recordResourceFailureEvents(updateErr.ResourceFailures, KongConfigurationApplyFailedEventReason)
-
-			sendDiagnostic(updateErr.Err != nil, updateErr.RawBody)
+			c.recordResourceFailureEvents(updateErr.ResourceFailures(), KongConfigurationApplyFailedEventReason)
+		}
+		if errors.As(err, &responseParsingErr) {
+			rawResponseBody = responseParsingErr.ResponseBody()
+		}
+		sendDiagnostic(true, rawResponseBody)
 
-			if updateErr.Err != nil {
-				if err := ctx.Err(); err != nil {
-					logger.Error(err, "Exceeded Kong API timeout, consider increasing --proxy-timeout-seconds")
-				}
-				return "", fmt.Errorf("performing update for %s failed: %w", client.BaseRootURL(), updateErr)
-			}
-		} else {
-			// It should never happen.
-			return "", fmt.Errorf("performing update for %s failed with unexpected type of error: %w", client.BaseRootURL(), err)
+		if err := ctx.Err(); err != nil {
+			logger.Error(err, "Exceeded Kong API timeout, consider increasing --proxy-timeout-seconds")
 		}
+		return "", fmt.Errorf("performing update for %s failed: %w", client.BaseRootURL(), updateErr)
 	}
 	sendDiagnostic(false, nil) // No error occurred.
 	// update the lastConfigSHA with the new updated checksum
@@ -664,7 +665,7 @@ func prepareSendDiagnosticFn(
 		config = targetContent
 	}
 
-	return func(failed bool, raw []byte) {
+	return func(failed bool, rawResponseBody []byte) {
 		// Given that we can send multiple configs to this channel and
 		// the fact that the API that exposes that can only expose 1 config
 		// at a time it means that users utilizing the diagnostics API
@@ -672,7 +673,7 @@ func prepareSendDiagnosticFn(
 		// or successfully send configs might be covered by those send
 		// later on but we're OK with this limitation of said API.
 		select {
-		case diagnosticConfig.Configs <- util.ConfigDump{Failed: failed, Config: *config, RawResponseBody: raw}:
+		case diagnosticConfig.Configs <- util.ConfigDump{Failed: failed, Config: *config, RawResponseBody: rawResponseBody}:
 			logger.V(util.DebugLevel).Info("Shipping config to diagnostic server")
 		default:
 			logger.Error(nil, "Config diagnostic buffer full, dropping diagnostic config")
diff --git a/internal/dataplane/kongstate/kongstate.go b/internal/dataplane/kongstate/kongstate.go
index a2b7d5be6..75169adce 100644
--- a/internal/dataplane/kongstate/kongstate.go
+++ b/internal/dataplane/kongstate/kongstate.go
@@ -10,14 +10,18 @@ import (
 	"github.com/kong/go-kong/kong"
 	"github.com/samber/lo"
 	corev1 "k8s.io/api/core/v1"
+	netv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/admission/validation/consumers/credentials"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
 	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
 	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 )
@@ -330,81 +334,147 @@ func (ks *KongState) FillVaults(
 	}
 }
 
-func (ks *KongState) getPluginRelations() map[string]util.ForeignRelations {
+type NamespacedKongPlugin struct {
+	Namespace string
+	Name      string
+}
+
+func (ks *KongState) getPluginRelations(cacheStore store.Storer, log logr.Logger) map[string]util.ForeignRelations {
 	// KongPlugin key (KongPlugin's name:namespace) to corresponding associations
 	pluginRels := map[string]util.ForeignRelations{}
-	addConsumerRelation := func(namespace, pluginName, identifier string) {
-		pluginKey := namespace + ":" + pluginName
-		relations, ok := pluginRels[pluginKey]
-		if !ok {
-			relations = util.ForeignRelations{}
+
+	type entityRelationType int
+	const (
+		ConsumerRelation      entityRelationType = iota
+		ConsumerGroupRelation entityRelationType = iota
+		RouteRelation         entityRelationType = iota
+		ServiceRelation       entityRelationType = iota
+	)
+	addRelation := func(referer client.Object, plugin annotations.NamespacedKongPlugin, identifier string, t entityRelationType) {
+		// There are 2 types of KongPlugin references: local and remote.
+		// A local reference is one where the KongPlugin is in the same namespace as the referer.
+		// A remote reference is one where the KongPlugin is in a different namespace.
+		// By default a KongPlugin is considered local.
+		// If the plugin has a namespace specified, it is considered remote.
+		//
+		// The referer is the entity that the KongPlugin is associated with.
+		//
+		// Code in buildPlugins() will combine plugin associations into
+		// multi-entity plugins within the local namespace
+		namespace := referer.GetNamespace()
+		if plugin.Namespace != "" {
+			// remote KongPlugin, permitted if ReferenceGrant allows
+			if err := isRemotePluginReferenceAllowed(
+				cacheStore,
+				pluginReference{
+					Referer:   referer,
+					Namespace: plugin.Namespace,
+					Name:      plugin.Name,
+				},
+			); err != nil {
+				log.Error(err, "could not bind requested plugin", "plugin", plugin.Name, "namespace", plugin.Namespace)
+				return
+			}
+
+			namespace = plugin.Namespace
 		}
-		relations.Consumer = append(relations.Consumer, identifier)
-		pluginRels[pluginKey] = relations
-	}
-	addConsumerGroupRelation := func(namespace, pluginName, identifier string) {
-		pluginKey := namespace + ":" + pluginName
+
+		pluginKey := namespace + ":" + plugin.Name
 		relations, ok := pluginRels[pluginKey]
 		if !ok {
 			relations = util.ForeignRelations{}
 		}
-		relations.ConsumerGroup = append(relations.ConsumerGroup, identifier)
-		pluginRels[pluginKey] = relations
-	}
-	addRouteRelation := func(namespace, pluginName, identifier string) {
-		pluginKey := namespace + ":" + pluginName
-		relations, ok := pluginRels[pluginKey]
-		if !ok {
-			relations = util.ForeignRelations{}
+		switch t {
+		case ConsumerRelation:
+			relations.Consumer = append(relations.Consumer, identifier)
+		case ConsumerGroupRelation:
+			relations.ConsumerGroup = append(relations.ConsumerGroup, identifier)
+		case RouteRelation:
+			relations.Route = append(relations.Route, identifier)
+		case ServiceRelation:
+			relations.Service = append(relations.Service, identifier)
 		}
-		relations.Route = append(relations.Route, identifier)
-		pluginRels[pluginKey] = relations
-	}
-	addServiceRelation := func(namespace, pluginName, identifier string) {
-		pluginKey := namespace + ":" + pluginName
-		relations, ok := pluginRels[pluginKey]
-		if !ok {
-			relations = util.ForeignRelations{}
-		}
-		relations.Service = append(relations.Service, identifier)
 		pluginRels[pluginKey] = relations
 	}
 
 	for i := range ks.Services {
-		// service
 		for _, svc := range ks.Services[i].K8sServices {
-			pluginList := annotations.ExtractKongPluginsFromAnnotations(svc.GetAnnotations())
-			for _, pluginName := range pluginList {
-				addServiceRelation(svc.Namespace, pluginName, *ks.Services[i].Name)
+			pluginList := annotations.ExtractNamespacedKongPluginsFromAnnotations(svc.GetAnnotations())
+			for _, plugin := range pluginList {
+				addRelation(svc, plugin, *ks.Services[i].Name, ServiceRelation)
 			}
 		}
-		// route
+
 		for j := range ks.Services[i].Routes {
 			ingress := ks.Services[i].Routes[j].Ingress
-			pluginList := annotations.ExtractKongPluginsFromAnnotations(ingress.Annotations)
-			for _, pluginName := range pluginList {
-				addRouteRelation(ingress.Namespace, pluginName, *ks.Services[i].Routes[j].Name)
+			pluginList := annotations.ExtractNamespacedKongPluginsFromAnnotations(ingress.Annotations)
+			for _, plugin := range pluginList {
+				// pretend we have a full Ingress struct for reference checks
+				virtualIngress := netv1.Ingress{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: ingress.Namespace,
+						Name:      ingress.Name,
+					},
+				}
+				addRelation(&virtualIngress, plugin, *ks.Services[i].Routes[j].Name, RouteRelation)
 			}
 		}
 	}
-	// consumer
+
 	for _, c := range ks.Consumers {
-		pluginList := annotations.ExtractKongPluginsFromAnnotations(c.K8sKongConsumer.GetAnnotations())
-		for _, pluginName := range pluginList {
-			addConsumerRelation(c.K8sKongConsumer.Namespace, pluginName, *c.Username)
+		pluginList := annotations.ExtractNamespacedKongPluginsFromAnnotations(c.K8sKongConsumer.GetAnnotations())
+		for _, plugin := range pluginList {
+			addRelation(&c.K8sKongConsumer, plugin, *c.Username, ConsumerRelation)
 		}
 	}
-	// consumer group
+
 	for _, cg := range ks.ConsumerGroups {
-		pluginList := annotations.ExtractKongPluginsFromAnnotations(cg.K8sKongConsumerGroup.GetAnnotations())
-		for _, pluginName := range pluginList {
-			addConsumerGroupRelation(cg.K8sKongConsumerGroup.Namespace, pluginName, *cg.Name)
+		pluginList := annotations.ExtractNamespacedKongPluginsFromAnnotations(cg.K8sKongConsumerGroup.GetAnnotations())
+		for _, plugin := range pluginList {
+			addRelation(&cg.K8sKongConsumerGroup, plugin, *cg.Name, ConsumerGroupRelation)
 		}
 	}
 
 	return pluginRels
 }
 
+type pluginReference struct {
+	Referer   client.Object
+	Namespace string
+	Name      string
+}
+
+func isRemotePluginReferenceAllowed(s store.Storer, r pluginReference) error {
+	// remote plugin. considered part of this namespace if a suitable ReferenceGrant exists
+	grants, err := s.ListReferenceGrants()
+	if err != nil {
+		return fmt.Errorf("could not retrieve ReferenceGrants from store when building plugin relations map: %w", err)
+	}
+	allowed := gatewayapi.GetPermittedForReferenceGrantFrom(gatewayapi.ReferenceGrantFrom{
+		Group:     gatewayapi.Group(r.Referer.GetObjectKind().GroupVersionKind().Group),
+		Kind:      gatewayapi.Kind(r.Referer.GetObjectKind().GroupVersionKind().Kind),
+		Namespace: gatewayapi.Namespace(r.Referer.GetNamespace()),
+	}, grants)
+
+	// we don't have a full plugin resource here for the grant checker, so we build a fake one with the correct
+	// name and namespace
+	virtualReference := gatewayapi.PluginLabelReference{
+		Namespace: lo.ToPtr(r.Referer.GetNamespace()),
+		Name:      r.Name,
+	}
+	virtualPlugin := &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: r.Namespace,
+			Name:      r.Name,
+		},
+	}
+	if !gatewayapi.NewRefCheckerForKongPlugin(virtualPlugin, virtualReference).IsRefAllowedByGrant(allowed) {
+		return fmt.Errorf("no grant found for %s in %s to plugin %s in %s requested remote KongPlugin bind",
+			r.Referer.GetObjectKind().GroupVersionKind().Kind, r.Referer.GetNamespace(), r.Name, r.Namespace)
+	}
+	return nil
+}
+
 func buildPlugins(
 	logger logr.Logger,
 	s store.Storer,
@@ -539,7 +609,7 @@ func (ks *KongState) FillPlugins(
 	s store.Storer,
 	failuresCollector *failures.ResourceFailuresCollector,
 ) {
-	ks.Plugins = buildPlugins(log, s, failuresCollector, ks.getPluginRelations())
+	ks.Plugins = buildPlugins(log, s, failuresCollector, ks.getPluginRelations(s, log))
 }
 
 // FillIDs iterates over the KongState and fills in the ID field for each entity
diff --git a/internal/dataplane/sendconfig/backoff_strategy.go b/internal/dataplane/sendconfig/backoff_strategy.go
index 5a7482ccb..96dfc0121 100644
--- a/internal/dataplane/sendconfig/backoff_strategy.go
+++ b/internal/dataplane/sendconfig/backoff_strategy.go
@@ -47,25 +47,20 @@ func NewUpdateStrategyWithBackoff(
 // In case it's not, it will return a predefined ErrUpdateSkippedDueToBackoffStrategy.
 // In case it is, apart from calling UpdateStrategy.Update, it will also register a success or a failure of an update
 // attempt so that the UpdateBackoffStrategy can keep track of it.
-func (s UpdateStrategyWithBackoff) Update(ctx context.Context, targetContent ContentWithHash) (
-	err error,
-	resourceErrors []ResourceError,
-	rawErrBody []byte,
-	resourceErrorsParseErr error,
-) {
+func (s UpdateStrategyWithBackoff) Update(ctx context.Context, targetContent ContentWithHash) (err error) {
 	if canUpdate, whyNot := s.backoffStrategy.CanUpdate(targetContent.Hash); !canUpdate {
-		return NewUpdateSkippedDueToBackoffStrategyError(whyNot), nil, nil, nil
+		return NewUpdateSkippedDueToBackoffStrategyError(whyNot)
 	}
 
-	err, resourceErrors, rawErrBody, resourceErrorsParseErr = s.decorated.Update(ctx, targetContent)
+	err = s.decorated.Update(ctx, targetContent)
 	if err != nil {
 		s.logger.V(util.DebugLevel).Info("Update failed, registering it for backoff strategy", "reason", err.Error())
 		s.backoffStrategy.RegisterUpdateFailure(err, targetContent.Hash)
-	} else {
-		s.backoffStrategy.RegisterUpdateSuccess()
+		return err
 	}
 
-	return err, resourceErrors, rawErrBody, resourceErrorsParseErr
+	s.backoffStrategy.RegisterUpdateSuccess()
+	return nil
 }
 
 func (s UpdateStrategyWithBackoff) MetricsProtocol() metrics.Protocol {
diff --git a/internal/dataplane/sendconfig/dbmode.go b/internal/dataplane/sendconfig/dbmode.go
index 9ccce2e50..5d0e35bac 100644
--- a/internal/dataplane/sendconfig/dbmode.go
+++ b/internal/dataplane/sendconfig/dbmode.go
@@ -2,6 +2,7 @@ package sendconfig
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"reflect"
 	"sync"
@@ -63,22 +64,15 @@ func NewUpdateStrategyDBModeKonnect(
 	return s
 }
 
-func (s UpdateStrategyDBMode) Update(ctx context.Context, targetContent ContentWithHash) (
-	err error,
-	resourceErrors []ResourceError,
-	// this is always empty for DB mode, as it does not have a single config error, and will instead return individual
-	// errors as a deckutils.ErrArray. we need it for the interface signature, however.
-	rawErrBody []byte,
-	resourceErrorsParseErr error,
-) {
+func (s UpdateStrategyDBMode) Update(ctx context.Context, targetContent ContentWithHash) error {
 	cs, err := s.currentState(ctx)
 	if err != nil {
-		return fmt.Errorf("failed getting current state for %s: %w", s.client.BaseRootURL(), err), nil, nil, nil
+		return fmt.Errorf("failed getting current state for %s: %w", s.client.BaseRootURL(), err)
 	}
 
 	ts, err := s.targetState(ctx, cs, targetContent.Content)
 	if err != nil {
-		return deckerrors.ConfigConflictError{Err: err}, nil, nil, nil
+		return deckerrors.ConfigConflictError{Err: err}
 	}
 
 	syncer, err := diff.NewSyncer(diff.SyncerOpts{
@@ -91,7 +85,7 @@ func (s UpdateStrategyDBMode) Update(ctx context.Context, targetContent ContentW
 		EnableEntityActions: true,
 	})
 	if err != nil {
-		return fmt.Errorf("creating a new syncer for %s: %w", s.client.BaseRootURL(), err), nil, nil, nil
+		return fmt.Errorf("creating a new syncer for %s: %w", s.client.BaseRootURL(), err)
 	}
 
 	ctx, cancel := context.WithCancel(ctx)
@@ -101,13 +95,24 @@ func (s UpdateStrategyDBMode) Update(ctx context.Context, targetContent ContentW
 	cancel()
 	s.resourceErrorLock.Lock()
 	defer s.resourceErrorLock.Unlock()
+	resourceFailures := resourceErrorsToResourceFailures(s.resourceErrors, s.logger)
 	if errs != nil {
-		return deckutils.ErrArray{Errors: errs}, s.resourceErrors, nil, nil
+		return NewUpdateError(
+			resourceFailures,
+			deckutils.ErrArray{Errors: errs},
+		)
 	}
 
 	// as of GDR 1.8 we should always get a plain error set in addition to resourceErrors, so returning resourceErrors
 	// here should not be necessary. Return it anyway as a future-proof because why not.
-	return nil, s.resourceErrors, nil, nil
+	if len(resourceFailures) > 0 {
+		return NewUpdateError(
+			resourceFailures,
+			errors.New("go-database-reconciler found resource errors"),
+		)
+	}
+
+	return nil
 }
 
 // HandleEvents handles logging and error reporting for individual entity change events generated during a sync by
diff --git a/internal/dataplane/sendconfig/errors.go b/internal/dataplane/sendconfig/errors.go
new file mode 100644
index 000000000..d64bb359c
--- /dev/null
+++ b/internal/dataplane/sendconfig/errors.go
@@ -0,0 +1,63 @@
+package sendconfig
+
+import "github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
+
+// UpdateError wraps several pieces of error information relevant to a failed Kong update attempt.
+type UpdateError struct {
+	rawResponseBody  []byte
+	resourceFailures []failures.ResourceFailure
+	err              error
+}
+
+func NewUpdateError(resourceFailures []failures.ResourceFailure, err error) UpdateError {
+	return UpdateError{
+		resourceFailures: resourceFailures,
+		err:              err,
+	}
+}
+
+func NewUpdateErrorWithResponseBody(rawResponseBody []byte, resourceFailures []failures.ResourceFailure, err error) UpdateError {
+	return UpdateError{
+		rawResponseBody:  rawResponseBody,
+		resourceFailures: resourceFailures,
+		err:              err,
+	}
+}
+
+// Error implements the Error interface. It returns the string value of the err field.
+func (e UpdateError) Error() string {
+	return e.err.Error()
+}
+
+// RawResponseBody returns the raw HTTP response body from Kong for the failed update if it was captured.
+func (e UpdateError) RawResponseBody() []byte {
+	return e.rawResponseBody
+}
+
+// ResourceFailures returns per-resource failures from a Kong configuration update attempt.
+func (e UpdateError) ResourceFailures() []failures.ResourceFailure {
+	return e.resourceFailures
+}
+
+func (e UpdateError) Unwrap() error {
+	return e.err
+}
+
+// ResponseParsingError is an error type that is returned when the response from Kong is not in the expected format.
+type ResponseParsingError struct {
+	responseBody []byte
+}
+
+func NewResponseParsingError(responseBody []byte) ResponseParsingError {
+	return ResponseParsingError{responseBody: responseBody}
+}
+
+// Error implements the Error interface.
+func (e ResponseParsingError) Error() string {
+	return "failed to parse Kong error response"
+}
+
+// ResponseBody returns the raw HTTP response body from Kong that could not be parsed.
+func (e ResponseParsingError) ResponseBody() []byte {
+	return e.responseBody
+}
diff --git a/internal/dataplane/sendconfig/inmemory.go b/internal/dataplane/sendconfig/inmemory.go
index b6a004e23..2da7f8a7e 100644
--- a/internal/dataplane/sendconfig/inmemory.go
+++ b/internal/dataplane/sendconfig/inmemory.go
@@ -48,24 +48,25 @@ func NewUpdateStrategyInMemory(
 	}
 }
 
-func (s UpdateStrategyInMemory) Update(ctx context.Context, targetState ContentWithHash) (
-	err error,
-	resourceErrors []ResourceError,
-	rawErrBody []byte,
-	resourceErrorsParseErr error,
-) {
+func (s UpdateStrategyInMemory) Update(ctx context.Context, targetState ContentWithHash) error {
 	dblessConfig := s.configConverter.Convert(targetState.Content)
 	config, err := json.Marshal(dblessConfig)
 	if err != nil {
-		return fmt.Errorf("constructing kong configuration: %w", err), nil, nil, nil
+		return fmt.Errorf("constructing kong configuration: %w", err)
 	}
-
-	if errBody, err := s.configService.ReloadDeclarativeRawConfig(ctx, bytes.NewReader(config), true, true); err != nil {
+	if errBody, reloadConfigErr := s.configService.ReloadDeclarativeRawConfig(ctx, bytes.NewReader(config), true, true); reloadConfigErr != nil {
 		resourceErrors, parseErr := parseFlatEntityErrors(errBody, s.logger)
-		return err, resourceErrors, errBody, parseErr
-	}
+		if parseErr != nil {
+			return fmt.Errorf("failed to parse flat entity errors from error response: %w", parseErr)
+		}
 
-	return nil, nil, nil, nil
+		return NewUpdateErrorWithResponseBody(
+			errBody,
+			resourceErrorsToResourceFailures(resourceErrors, s.logger),
+			reloadConfigErr,
+		)
+	}
+	return nil
 }
 
 func (s UpdateStrategyInMemory) MetricsProtocol() metrics.Protocol {
diff --git a/internal/dataplane/sendconfig/inmemory_error_handling.go b/internal/dataplane/sendconfig/inmemory_error_handling.go
index 9602a868a..f190aed0c 100644
--- a/internal/dataplane/sendconfig/inmemory_error_handling.go
+++ b/internal/dataplane/sendconfig/inmemory_error_handling.go
@@ -6,8 +6,11 @@ import (
 	"strings"
 
 	"github.com/go-logr/logr"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	k8stypes "k8s.io/apimachinery/pkg/types"
 
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
 	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
 )
@@ -100,7 +103,7 @@ func parseFlatEntityErrors(body []byte, logger logr.Logger) ([]ResourceError, er
 				logger.Error(nil, "Could not fully parse config error", "message", message)
 			}
 		}
-		return resourceErrors, fmt.Errorf("could not unmarshal config error: %w", err)
+		return resourceErrors, NewResponseParsingError(body)
 	}
 	if len(configError.Flattened) == 0 {
 		if len(configError.Message) > 0 {
@@ -198,3 +201,35 @@ func gvkIsClusterScoped(gvk schema.GroupVersionKind) bool {
 	}
 	return false
 }
+
+// resourceErrorsToResourceFailures translates a slice of ResourceError to a slice of failures.ResourceFailure.
+func resourceErrorsToResourceFailures(resourceErrors []ResourceError, logger logr.Logger) []failures.ResourceFailure {
+	var out []failures.ResourceFailure
+	for _, ee := range resourceErrors {
+		obj := metav1.PartialObjectMetadata{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       ee.Kind,
+				APIVersion: ee.APIVersion,
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: ee.Namespace,
+				Name:      ee.Name,
+				UID:       k8stypes.UID(ee.UID),
+			},
+		}
+		for problemSource, problem := range ee.Problems {
+			logger.V(util.DebugLevel).Info("Adding failure", "resource_name", ee.Name, "source", problemSource, "problem", problem)
+			resourceFailure, failureCreateErr := failures.NewResourceFailure(
+				fmt.Sprintf("invalid %s: %s", problemSource, problem),
+				&obj,
+			)
+			if failureCreateErr != nil {
+				logger.Error(failureCreateErr, "Could not create resource failure event")
+			} else {
+				out = append(out, resourceFailure)
+			}
+		}
+	}
+
+	return out
+}
diff --git a/internal/dataplane/sendconfig/sendconfig.go b/internal/dataplane/sendconfig/sendconfig.go
index 3a360046c..bae6e6643 100644
--- a/internal/dataplane/sendconfig/sendconfig.go
+++ b/internal/dataplane/sendconfig/sendconfig.go
@@ -9,11 +9,8 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/kong/go-database-reconciler/pkg/file"
 	"github.com/kong/go-kong/kong"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	k8stypes "k8s.io/apimachinery/pkg/types"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/deckgen"
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/metrics"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
 )
@@ -37,26 +34,6 @@ type AdminAPIClient interface {
 	KonnectControlPlane() string
 }
 
-// Note that UpdateError is used for both DB-less and DB-backed update strategies, but does not have parity between the
-// two. Pending a refactor of database reconciler (https://github.com/Kong/go-database-reconciler/issues/22), KIC does
-// not have access to per-resource errors or any original response bodies in database mode. Future DB mode work would
-// need to find some way to reconcile the single /config raw body and many per-resource responses from DB endpoints.
-
-// UpdateError wraps several pieces of error information relevant to a failed Kong update attempt.
-type UpdateError struct {
-	// RawBody is the original Kong HTTP error response body from a failed update.
-	RawBody []byte
-	// ResourceFailures are per-resource failures from a Kong configuration update attempt.
-	ResourceFailures []failures.ResourceFailure
-	// Err is an overall description of the update failure.
-	Err error
-}
-
-// Error implements the Error interface. It returns the string value of the Err field.
-func (e UpdateError) Error() string {
-	return fmt.Sprintf("%s", e.Err)
-}
-
 // PerformUpdate writes `targetContent` to Kong Admin API specified by `kongConfig`.
 func PerformUpdate(
 	ctx context.Context,
@@ -71,14 +48,14 @@ func PerformUpdate(
 	oldSHA := client.LastConfigSHA()
 	newSHA, err := deckgen.GenerateSHA(targetContent)
 	if err != nil {
-		return oldSHA, UpdateError{ResourceFailures: []failures.ResourceFailure{}, Err: err}
+		return oldSHA, fmt.Errorf("failed to generate SHA for target content: %w", err)
 	}
 
 	// disable optimization if reverse sync is enabled
 	if !config.EnableReverseSync {
 		configurationChanged, err := configChangeDetector.HasConfigurationChanged(ctx, oldSHA, newSHA, targetContent, client, client.AdminAPIClient())
 		if err != nil {
-			return nil, UpdateError{Err: err}
+			return nil, fmt.Errorf("failed to detect configuration change: %w", err)
 		}
 		if !configurationChanged {
 			if client.IsKonnect() {
@@ -93,7 +70,7 @@ func PerformUpdate(
 	updateStrategy := updateStrategyResolver.ResolveUpdateStrategy(client)
 	logger = logger.WithValues("update_strategy", updateStrategy.Type())
 	timeStart := time.Now()
-	err, resourceErrors, rawErrBody, resourceErrorsParseErr := updateStrategy.Update(ctx, ContentWithHash{
+	err = updateStrategy.Update(ctx, ContentWithHash{
 		Content: targetContent,
 		Hash:    newSHA,
 	})
@@ -101,14 +78,15 @@ func PerformUpdate(
 
 	metricsProtocol := updateStrategy.MetricsProtocol()
 	if err != nil {
-		// Not pushing metrics in case it's an update skip due to a backoff.
-		if errors.As(err, &UpdateSkippedDueToBackoffStrategyError{}) {
-			return nil, UpdateError{RawBody: rawErrBody, Err: err}
+		// For UpdateError, record the failure and return the error.
+		var updateError UpdateError
+		if errors.As(err, &updateError) {
+			promMetrics.RecordPushFailure(metricsProtocol, duration, client.BaseRootURL(), len(updateError.ResourceFailures()), updateError.err)
+			return nil, updateError
 		}
 
-		resourceFailures := resourceErrorsToResourceFailures(resourceErrors, resourceErrorsParseErr, logger)
-		promMetrics.RecordPushFailure(metricsProtocol, duration, client.BaseRootURL(), len(resourceFailures), err)
-		return nil, UpdateError{ResourceFailures: resourceFailures, RawBody: rawErrBody, Err: err}
+		// Any other error, simply return it and skip metrics recording - we have no details to record.
+		return nil, fmt.Errorf("config update failed: %w", err)
 	}
 
 	promMetrics.RecordPushSuccess(metricsProtocol, duration, client.BaseRootURL())
@@ -125,41 +103,3 @@ func PerformUpdate(
 // -----------------------------------------------------------------------------
 // Sendconfig - Private Functions
 // -----------------------------------------------------------------------------
-
-// resourceErrorsToResourceFailures translates a slice of ResourceError to a slice of failures.ResourceFailure.
-// In case of parseErr being not nil, it just returns a nil slice.
-func resourceErrorsToResourceFailures(resourceErrors []ResourceError, parseErr error, logger logr.Logger) []failures.ResourceFailure {
-	if parseErr != nil {
-		logger.Error(parseErr, "Failed parsing resource errors")
-		return nil
-	}
-
-	var out []failures.ResourceFailure
-	for _, ee := range resourceErrors {
-		obj := metav1.PartialObjectMetadata{
-			TypeMeta: metav1.TypeMeta{
-				Kind:       ee.Kind,
-				APIVersion: ee.APIVersion,
-			},
-			ObjectMeta: metav1.ObjectMeta{
-				Namespace: ee.Namespace,
-				Name:      ee.Name,
-				UID:       k8stypes.UID(ee.UID),
-			},
-		}
-		for problemSource, problem := range ee.Problems {
-			logger.V(util.DebugLevel).Info("Adding failure", "resource_name", ee.Name, "source", problemSource, "problem", problem)
-			resourceFailure, failureCreateErr := failures.NewResourceFailure(
-				fmt.Sprintf("invalid %s: %s", problemSource, problem),
-				&obj,
-			)
-			if failureCreateErr != nil {
-				logger.Error(failureCreateErr, "Could not create resource failure event")
-			} else {
-				out = append(out, resourceFailure)
-			}
-		}
-	}
-
-	return out
-}
diff --git a/internal/dataplane/sendconfig/strategy.go b/internal/dataplane/sendconfig/strategy.go
index a460be194..c81dd5f60 100644
--- a/internal/dataplane/sendconfig/strategy.go
+++ b/internal/dataplane/sendconfig/strategy.go
@@ -21,12 +21,7 @@ type ContentWithHash struct {
 // UpdateStrategy is the way we approach updating data-plane's configuration, depending on its type.
 type UpdateStrategy interface {
 	// Update applies targetConfig to the data-plane.
-	Update(ctx context.Context, targetContent ContentWithHash) (
-		err error,
-		resourceErrors []ResourceError,
-		rawErrorBody []byte,
-		resourceErrorsParseErr error,
-	)
+	Update(ctx context.Context, targetContent ContentWithHash) error
 
 	// MetricsProtocol returns a string describing the update strategy type to be used in metrics.
 	MetricsProtocol() metrics.Protocol
diff --git a/internal/dataplane/translator/backendref.go b/internal/dataplane/translator/backendref.go
index ef26c24d3..781fb9c2a 100644
--- a/internal/dataplane/translator/backendref.go
+++ b/internal/dataplane/translator/backendref.go
@@ -65,7 +65,7 @@ func backendRefsToKongStateBackends(
 		}
 
 		if !util.IsBackendRefGroupKindSupported(backendRef.Group, backendRef.Kind) ||
-			!newRefCheckerForRoute(route, backendRef).IsRefAllowedByGrant(allowed) {
+			!gatewayapi.NewRefCheckerForRoute(route, backendRef).IsRefAllowedByGrant(allowed) {
 			// we log impermissible refs rather than failing the entire rule. while we cannot actually route to
 			// these, we do not want a single impermissible ref to take the entire rule offline. in the case of edits,
 			// failing the entire rule could potentially delete routes that were previously online and in use, and
diff --git a/internal/dataplane/translator/translate_utils.go b/internal/dataplane/translator/translate_utils.go
index c5d8b5f7d..6a5449885 100644
--- a/internal/dataplane/translator/translate_utils.go
+++ b/internal/dataplane/translator/translate_utils.go
@@ -42,10 +42,10 @@ func convertGatewayMatchHeadersToKongRouteMatchHeaders(headers []gatewayapi.HTTP
 	return convertedHeaders, nil
 }
 
-// getPermittedForReferenceGrantFrom takes a ReferenceGrant From (a namespace, group, and kind) and returns a map
+// GetPermittedForReferenceGrantFrom takes a ReferenceGrant From (a namespace, group, and kind) and returns a map
 // from a namespace to a slice of ReferenceGrant Tos. When a To is included in the slice, the key namespace has a
 // ReferenceGrant with those Tos and the input From.
-func getPermittedForReferenceGrantFrom(
+func GetPermittedForReferenceGrantFrom(
 	from gatewayapi.ReferenceGrantFrom,
 	grants []*gatewayapi.ReferenceGrant,
 ) map[gatewayapi.Namespace][]gatewayapi.ReferenceGrantTo {
@@ -82,7 +82,7 @@ func generateKongServiceFromBackendRefWithName(
 	if err != nil {
 		return kongstate.Service{}, fmt.Errorf("could not retrieve ReferenceGrants for %s: %w", objName, err)
 	}
-	allowed := getPermittedForReferenceGrantFrom(gatewayapi.ReferenceGrantFrom{
+	allowed := GetPermittedForReferenceGrantFrom(gatewayapi.ReferenceGrantFrom{
 		Group:     gatewayapi.Group(route.GetObjectKind().GroupVersionKind().Group),
 		Kind:      gatewayapi.Kind(route.GetObjectKind().GroupVersionKind().Kind),
 		Namespace: gatewayapi.Namespace(route.GetNamespace()),
diff --git a/internal/dataplane/translator/translator.go b/internal/dataplane/translator/translator.go
index d1abad4a8..4806ea0d4 100644
--- a/internal/dataplane/translator/translator.go
+++ b/internal/dataplane/translator/translator.go
@@ -2,14 +2,13 @@ package translator
 
 import (
 	"github.com/go-logr/logr"
-	"github.com/kong/go-kong/kong"
-	"github.com/samber/mo"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	dpconf "github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/config"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/kongstate"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/license"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/manager/featuregates"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
 )
@@ -64,12 +63,6 @@ func NewFeatureFlags(
 	}
 }
 
-// LicenseGetter is an interface for getting the Kong Enterprise license.
-type LicenseGetter interface {
-	// GetLicense returns an optional license.
-	GetLicense() mo.Option[kong.License]
-}
-
 // Translator translates Kubernetes objects and configurations into their
 // equivalent Kong objects and configurations, producing a complete
 // state configuration for the Kong Admin API.
@@ -77,7 +70,7 @@ type Translator struct {
 	logger        logr.Logger
 	storer        store.Storer
 	workspace     string
-	licenseGetter LicenseGetter
+	licenseGetter license.Getter
 	featureFlags  FeatureFlags
 
 	failuresCollector          *failures.ResourceFailuresCollector
@@ -228,7 +221,7 @@ func (t *Translator) BuildKongConfig() KongConfigBuildingResult {
 // -----------------------------------------------------------------------------
 
 // InjectLicenseGetter sets a license getter to be used by the translator.
-func (t *Translator) InjectLicenseGetter(licenseGetter LicenseGetter) {
+func (t *Translator) InjectLicenseGetter(licenseGetter license.Getter) {
 	t.licenseGetter = licenseGetter
 }
 
diff --git a/internal/dataplane/translator/wrappers_refchecker.go b/internal/dataplane/translator/wrappers_refchecker.go
deleted file mode 100644
index 3492d4aff..000000000
--- a/internal/dataplane/translator/wrappers_refchecker.go
+++ /dev/null
@@ -1,95 +0,0 @@
-package translator
-
-import (
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
-)
-
-// refChecker is a wrapper type that facilitates checking whether a backenRef is allowed
-// by a referenceGrantTo set.
-type refChecker[T gatewayapi.BackendRefT] struct {
-	route      client.Object
-	backendRef T
-}
-
-// newRefCheckerForRoute returns a refChecker for the provided route and backendRef.
-func newRefCheckerForRoute[T gatewayapi.BackendRefT](route client.Object, ref T) refChecker[T] {
-	return refChecker[T]{
-		route:      route,
-		backendRef: ref,
-	}
-}
-
-// IsRefAllowedByGrant is a wrapper on top of isRefAllowedByGrant checks if backendRef (that RefChecker
-// holds) is permitted by the provided namespace-indexed ReferenceGrantTo set: allowedRefs.
-// allowedRefs is assumed to contain Tos that only match the backendRef's parent's From, as returned by
-// getPermittedForReferenceGrantFrom.
-func (rc refChecker[T]) IsRefAllowedByGrant(
-	allowedRefs map[gatewayapi.Namespace][]gatewayapi.ReferenceGrantTo,
-) bool {
-	switch br := (interface{})(rc.backendRef).(type) {
-	case gatewayapi.BackendRef:
-		if br.Namespace == nil {
-			return true
-		}
-
-		// If the namespace is specified but is the same as the route's namespace, then the ref is allowed.
-		if rc.route.GetNamespace() == string(*br.Namespace) {
-			return true
-		}
-
-		return isRefAllowedByGrant(
-			(*string)(br.Namespace),
-			(string)(br.Name),
-			(string)(*br.Group),
-			(string)(*br.Kind),
-			allowedRefs,
-		)
-
-	case gatewayapi.SecretObjectReference:
-		if br.Namespace == nil {
-			return true
-		}
-
-		return isRefAllowedByGrant(
-			(*string)(br.Namespace),
-			(string)(br.Name),
-			(string)(*br.Group),
-			(string)(*br.Kind),
-			allowedRefs,
-		)
-	}
-
-	return false
-}
-
-// isRefAllowedByGrant checks if backendRef is permitted by the provided namespace-indexed ReferenceGrantTo set: allowed.
-// allowed is assumed to contain Tos that only match the backendRef's parent's From, as returned by
-// getPermittedForReferenceGrantFrom.
-func isRefAllowedByGrant(
-	namespace *string,
-	name string,
-	group string,
-	kind string,
-	allowed map[gatewayapi.Namespace][]gatewayapi.ReferenceGrantTo,
-) bool {
-	if namespace == nil {
-		// local references are always fine
-		return true
-	}
-	for _, to := range allowed[gatewayapi.Namespace(*namespace)] {
-		if string(to.Group) == group && string(to.Kind) == kind {
-			if to.Name != nil {
-				if string(*to.Name) == name {
-					return true
-				}
-			} else {
-				// if no referent name specified, matching group/kind is sufficient
-				return true
-			}
-		}
-	}
-
-	return false
-}
diff --git a/internal/gatewayapi/contraints.go b/internal/gatewayapi/contraints.go
index 0e11b3d6a..c52e0f3ed 100644
--- a/internal/gatewayapi/contraints.go
+++ b/internal/gatewayapi/contraints.go
@@ -16,7 +16,20 @@ type RouteT interface {
 		*GRPCRoute
 }
 
+// TODO https://github.com/Kong/kubernetes-ingress-controller/issues/6000
+// This is currently a "BackendRefT" but it's used as a generic target object.
+// Needs to be renamed as such. It should be able to handle anything that satisfies
+// https://pkg.go.dev/sigs.k8s.io/controller-runtime/pkg/client#Object
+// ReferenceGrants deal with anything that has a Group, Kind, Namespace, and sometimes a Name.
+// client.Object wraps two interfaces that have methods to surface all of those.
+
 type BackendRefT interface {
 	BackendRef |
-		SecretObjectReference
+		SecretObjectReference |
+		PluginLabelReference
+}
+
+type PluginLabelReference struct {
+	Namespace *string
+	Name      string
 }
diff --git a/internal/gatewayapi/references.go b/internal/gatewayapi/references.go
new file mode 100644
index 000000000..c97112739
--- /dev/null
+++ b/internal/gatewayapi/references.go
@@ -0,0 +1,147 @@
+package gatewayapi
+
+import (
+	"reflect"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// RefChecker is a wrapper type that facilitates checking whether a backendRef is allowed
+// by a referenceGrantTo set.
+type RefChecker[T BackendRefT] struct {
+	target     client.Object
+	backendRef T
+}
+
+// NewRefCheckerForRoute returns a RefChecker for the provided route and backendRef.
+func NewRefCheckerForRoute[T BackendRefT](route client.Object, ref T) RefChecker[T] {
+	return RefChecker[T]{
+		target:     route,
+		backendRef: ref,
+	}
+}
+
+func NewRefCheckerForKongPlugin[T BackendRefT](target client.Object, requester T) RefChecker[T] {
+	return RefChecker[T]{
+		target:     target,
+		backendRef: requester,
+	}
+}
+
+// TODO https://github.com/Kong/kubernetes-ingress-controller/issues/6000 this has separate cases for different types,
+// but doesn't do anything meaningfully different for them (it only fills in some default info that should be available
+// from the involved objects' methods). We want a generic utility that handles relationship checks for any
+// client.Object. Any behavior particular to specific combinations of GVK->GVK relationships should be handled in the
+// code that implements those relationships.
+
+// IsRefAllowedByGrant is a wrapper on top of isRefAllowedByGrant checks if backendRef (that RefChecker
+// holds) is permitted by the provided namespace-indexed ReferenceGrantTo set: allowedRefs.
+// allowedRefs is assumed to contain Tos that only match the backendRef's parent's From, as returned by
+// GetPermittedForReferenceGrantFrom.
+func (rc RefChecker[T]) IsRefAllowedByGrant(
+	allowedRefs map[Namespace][]ReferenceGrantTo,
+) bool {
+	switch br := (interface{})(rc.backendRef).(type) {
+	case BackendRef:
+		// NOTE https://github.com/Kong/kubernetes-ingress-controller/issues/6000
+		// This is a catch-all that the plugins technically won't need as-is because they have their own
+		// inherent namespace check: if no namespace specified the ref is assumed allowed because it must be local.
+		if br.Namespace == nil {
+			return true
+		}
+
+		// If the namespace is specified but is the same as the route's namespace, then the ref is allowed.
+		if rc.target.GetNamespace() == string(*br.Namespace) {
+			return true
+		}
+
+		return isRefAllowedByGrant(
+			(*string)(br.Namespace),
+			(string)(br.Name),
+			(string)(*br.Group),
+			(string)(*br.Kind),
+			allowedRefs,
+		)
+
+	case SecretObjectReference:
+		if br.Namespace == nil {
+			return true
+		}
+
+		return isRefAllowedByGrant(
+			(*string)(br.Namespace),
+			(string)(br.Name),
+			(string)(*br.Group),
+			(string)(*br.Kind),
+			allowedRefs,
+		)
+
+	case PluginLabelReference:
+		if br.Namespace == nil {
+			return true
+		}
+
+		return isRefAllowedByGrant(
+			(br.Namespace),
+			(br.Name),
+			"configuration.konghq.com", // TODO https://github.com/Kong/kubernetes-ingress-controller/issues/6000
+			"KongPlugin",               // TODO These magic strings should become unnecessary once we work with client.Object
+			allowedRefs,
+		)
+	}
+
+	return false
+}
+
+// isRefAllowedByGrant checks if backendRef is permitted by the provided namespace-indexed ReferenceGrantTo set: allowed.
+// allowed is assumed to contain Tos that only match the backendRef's parent's From, as returned by
+// GetPermittedForReferenceGrantFrom.
+func isRefAllowedByGrant(
+	namespace *string,
+	name string,
+	group string,
+	kind string,
+	allowed map[Namespace][]ReferenceGrantTo,
+) bool {
+	if namespace == nil {
+		// local references are always fine
+		return true
+	}
+	for _, to := range allowed[Namespace(*namespace)] {
+		if string(to.Group) == group && string(to.Kind) == kind {
+			if to.Name != nil {
+				if string(*to.Name) == name {
+					return true
+				}
+			} else {
+				// if no referent name specified, matching group/kind is sufficient
+				return true
+			}
+		}
+	}
+
+	return false
+}
+
+// GetPermittedForReferenceGrantFrom takes a ReferenceGrant From (a namespace, group, and kind) and returns a map
+// from a namespace to a slice of ReferenceGrant Tos. When a To is included in the slice, the key namespace has a
+// ReferenceGrant with those Tos and the input From.
+func GetPermittedForReferenceGrantFrom(
+	from ReferenceGrantFrom,
+	grants []*ReferenceGrant,
+) map[Namespace][]ReferenceGrantTo {
+	allowed := make(map[Namespace][]ReferenceGrantTo)
+	// loop over all From values in all grants. if we find a match, add all Tos to the list of Tos allowed for the
+	// grant namespace. this technically could add duplicate copies of the Tos if there are duplicate Froms (it makes
+	// no sense to add them, but it's allowed), but duplicate Tos are harmless (we only care about having at least one
+	// matching To when checking if a ReferenceGrant allows a reference)
+	for _, grant := range grants {
+		for _, otherFrom := range grant.Spec.From {
+			if reflect.DeepEqual(from, otherFrom) {
+				allowed[Namespace(grant.ObjectMeta.Namespace)] = append(allowed[Namespace(grant.ObjectMeta.Namespace)], grant.Spec.To...)
+			}
+		}
+	}
+
+	return allowed
+}
diff --git a/internal/license/agent.go b/internal/license/agent.go
index 4ea9e02a5..6a46e026c 100644
--- a/internal/license/agent.go
+++ b/internal/license/agent.go
@@ -26,6 +26,12 @@ const (
 	PollingTimeout = time.Minute * 5
 )
 
+// Getter is an interface for getting a Kong Enterprise license.
+type Getter interface {
+	// GetLicense returns an optional license.
+	GetLicense() mo.Option[kong.License]
+}
+
 // KonnectLicense is a license retrieved from Konnect.
 type KonnectLicense struct {
 	ID        string
diff --git a/internal/manager/run.go b/internal/manager/run.go
index ccc8a59d8..741e3c15a 100644
--- a/internal/manager/run.go
+++ b/internal/manager/run.go
@@ -300,6 +300,7 @@ func Run(
 		setupLog.Info("Inject license getter to config translator",
 			"license_getter_type", fmt.Sprintf("%T", licenseGetter))
 		configTranslator.InjectLicenseGetter(licenseGetter)
+		kongConfigFetcher.InjectLicenseGetter(licenseGetter)
 	}
 
 	if c.AnonymousReports {
diff --git a/internal/manager/setup.go b/internal/manager/setup.go
index 439a796a0..ec61cbae0 100644
--- a/internal/manager/setup.go
+++ b/internal/manager/setup.go
@@ -424,7 +424,7 @@ func setupLicenseGetter(
 	setupLog logr.Logger,
 	mgr manager.Manager,
 	statusQueue *status.Queue,
-) (translator.LicenseGetter, error) {
+) (license.Getter, error) {
 	// TODO https://github.com/Kong/kubernetes-ingress-controller/issues/3922
 	// This requires the Konnect client, which currently requires c.Konnect.ConfigSynchronizationEnabled also.
 	// We need to figure out exactly how that config surface works. Initial direction says add a separate toggle, but
diff --git a/internal/store/store.go b/internal/store/store.go
index 62574c4fe..86972f481 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -727,7 +727,7 @@ func mkObjFromGVK(gvk schema.GroupVersionKind) (runtime.Object, error) {
 	// ----------------------------------------------------------------------------
 	case gatewayv1.SchemeGroupVersion.WithKind("HTTPRoute"):
 		return &gatewayapi.HTTPRoute{}, nil
-	case gatewayv1alpha2.SchemeGroupVersion.WithKind("GRPCRoute"):
+	case gatewayv1.SchemeGroupVersion.WithKind("GRPCRoute"):
 		return &gatewayapi.GRPCRoute{}, nil
 	case gatewayv1alpha2.SchemeGroupVersion.WithKind("TCPRoute"):
 		return &gatewayapi.TCPRoute{}, nil


Test patch:
diff --git a/.github/test_dependencies.yaml b/.github/test_dependencies.yaml
index 1465228a6..a38d8314c 100644
--- a/.github/test_dependencies.yaml
+++ b/.github/test_dependencies.yaml
@@ -35,7 +35,7 @@ e2e:
       istio: '1.17.8'
 
   # renovate: datasource=helm depName=kuma registryUrl=https://kumahq.github.io/charts versioning=helm
-  kuma: '2.7.2'
+  kuma: '2.7.3'
 
 integration:
   helm:
diff --git a/internal/admission/validation/gateway/httproute_test.go b/internal/admission/validation/gateway/httproute_test.go
index a52342c4b..86c17f034 100644
--- a/internal/admission/validation/gateway/httproute_test.go
+++ b/internal/admission/validation/gateway/httproute_test.go
@@ -895,8 +895,7 @@ func TestValidateHTTPRoute(t *testing.T) {
 					},
 				},
 			},
-			valid:         false,
-			validationMsg: "HTTPRoute spec did not pass validation: rules[0].filters[0]: filter type URLRewrite (with hostname replace) is unsupported",
+			valid: true,
 		},
 		{
 			msg: "HTTPRoute URLRewrite ReplacePrefixMatch",
diff --git a/internal/annotations/annotations_test.go b/internal/annotations/annotations_test.go
index f283c7a48..bb7fdbec2 100644
--- a/internal/annotations/annotations_test.go
+++ b/internal/annotations/annotations_test.go
@@ -157,6 +157,52 @@ func TestExtractKongPluginsFromAnnotations(t *testing.T) {
 	}
 }
 
+func TestExtractNamespacedKongPluginsFromAnnotations(t *testing.T) {
+	type args struct {
+		anns map[string]string
+	}
+	tests := []struct {
+		name string
+		args args
+		want []NamespacedKongPlugin
+	}{
+		{
+			name: "empty",
+			args: args{
+				anns: map[string]string{
+					"konghq.com/nothing": "whatever",
+				},
+			},
+			want: nil,
+		},
+		{
+			name: "basic",
+			args: args{
+				anns: map[string]string{
+					"konghq.com/plugins": "kp-rl, kp-cors",
+				},
+			},
+			want: []NamespacedKongPlugin{{Name: "kp-rl"}, {Name: "kp-cors"}},
+		},
+		{
+			name: "mixed",
+			args: args{
+				anns: map[string]string{
+					"konghq.com/plugins": "default:kp-rl, kp-cors",
+				},
+			},
+			want: []NamespacedKongPlugin{{Name: "kp-rl", Namespace: "default"}, {Name: "kp-cors"}},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ExtractNamespacedKongPluginsFromAnnotations(tt.args.anns); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ExtractNamespacedKongPluginsFromAnnotations() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestExtractConfigurationName(t *testing.T) {
 	type args struct {
 		anns map[string]string
diff --git a/internal/dataplane/configfetcher/kongrawstate_test.go b/internal/dataplane/configfetcher/kongrawstate_test.go
index c128f83ea..338a162b2 100644
--- a/internal/dataplane/configfetcher/kongrawstate_test.go
+++ b/internal/dataplane/configfetcher/kongrawstate_test.go
@@ -60,6 +60,11 @@ func TestKongRawStateToKongState(t *testing.T) {
 						},
 					},
 				},
+				Vaults: []*kong.Vault{
+					{
+						Name: kong.String("test-vault"), Prefix: kong.String("test-vault"),
+					},
+				},
 				Plugins: []*kong.Plugin{
 					{
 						Name: kong.String("plugin1"),
@@ -205,6 +210,13 @@ func TestKongRawStateToKongState(t *testing.T) {
 						},
 					},
 				},
+				Vaults: []kongstate.Vault{
+					{
+						Vault: kong.Vault{
+							Name: kong.String("test-vault"), Prefix: kong.String("test-vault"),
+						},
+					},
+				},
 				Certificates: []kongstate.Certificate{
 					{
 						Certificate: kong.Certificate{
@@ -292,7 +304,7 @@ func TestKongRawStateToKongState(t *testing.T) {
 
 			// Collect all fields that are tested in this test case.
 			if tt.kongRawState != nil {
-				testedKongRawStateFields.Insert(extractNotEmptyFieldNames(*tt.kongRawState)...)
+				testedKongRawStateFields.Insert(extractNotEmptyFieldNames(*tt.expectedKongState)...)
 			}
 
 			var state *kongstate.KongState
@@ -305,12 +317,12 @@ func TestKongRawStateToKongState(t *testing.T) {
 		})
 	}
 
-	ensureAllKongRawStateFieldsAreTested(t, testedKongRawStateFields.UnsortedList())
+	ensureAllKongStateFieldsAreTested(t, testedKongRawStateFields.UnsortedList())
 }
 
-// extractNotEmptyFieldNames returns the names of all non-empty fields in the given KongRawState.
+// extractNotEmptyFieldNames returns the names of all non-empty fields in the given KongState.
 // This is to programmatically find out what fields are used in a test case.
-func extractNotEmptyFieldNames(s utils.KongRawState) []string {
+func extractNotEmptyFieldNames(s kongstate.KongState) []string {
 	var fields []string
 	typ := reflect.ValueOf(s).Type()
 	for i := 0; i < typ.NumField(); i++ {
@@ -323,34 +335,30 @@ func extractNotEmptyFieldNames(s utils.KongRawState) []string {
 	return fields
 }
 
-// ensureAllKongRawStateFieldsAreTested verifies that all fields in KongRawState are tested.
+// ensureAllKongStateFieldsAreTested verifies that all fields in KongState are tested.
 // It uses the testedFields slice to determine what fields were actually tested and compares
-// it to the list of all fields in KongRawState, excluding fields that KIC doesn't support.
-func ensureAllKongRawStateFieldsAreTested(t *testing.T, testedFields []string) {
-	kongRawStateFieldsKICDoesntSupport := []string{
-		// These are fields that KIC explicitly doesn't support.
-		"SNIs",
-		"CustomEntities",
-		"Vaults",
-		"RBACRoles",
-		"RBACEndpointPermissions",
+// it to the list of all fields in KongState, excluding fields that KIC doesn't support.
+func ensureAllKongStateFieldsAreTested(t *testing.T, testedFields []string) {
+	exempt := []string{
+		// Plugins live under their attached objects and are not populated independently at the top level.
+		"Plugins",
+		// Licenses are injected from the license getter rather than extracted from the last state.
 		"Licenses",
 	}
-	allKongRawStateFields := func() []string {
+	allKongStateFields := func() []string {
 		var fields []string
-		typ := reflect.ValueOf(utils.KongRawState{}).Type()
+		typ := reflect.ValueOf(kongstate.KongState{}).Type()
 		for i := 0; i < typ.NumField(); i++ {
-			fields = append(fields, typ.Field(i).Name)
+			name := typ.Field(i).Name
+			if !lo.Contains(exempt, name) {
+				fields = append(fields, name)
+			}
 		}
 		return fields
 	}()
 
 	// Meta test - ensure we have testcases covering all fields in KongRawState.
-	for _, field := range allKongRawStateFields {
-		if lo.Contains(kongRawStateFieldsKICDoesntSupport, field) {
-			t.Logf("skipping field %s - explicitly unsupported", field)
-			continue
-		}
+	for _, field := range allKongStateFields {
 		assert.True(t, lo.Contains(testedFields, field), "field %s not tested", field)
 	}
 }
diff --git a/internal/dataplane/fallback/cache_to_graph_test.go b/internal/dataplane/fallback/cache_to_graph_test.go
new file mode 100644
index 000000000..03f01e7ef
--- /dev/null
+++ b/internal/dataplane/fallback/cache_to_graph_test.go
@@ -0,0 +1,345 @@
+package fallback_test
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/samber/lo"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	netv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/fallback"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+func TestDefaultCacheGraphProvider_CacheToGraph(t *testing.T) {
+	// adjacencyGraphStrings returns a map of stringified vertices and their neighbours
+	// in the given graph for easy comparison.
+	adjacencyGraphStrings := func(t *testing.T, g *fallback.ConfigGraph) map[string][]string {
+		am, err := g.AdjacencyMap()
+		require.NoError(t, err)
+		adjacencyMapStrings := make(map[string][]string, len(am))
+		for v, neighbours := range am {
+			neighboursStrings := lo.Map(neighbours, func(n fallback.ObjectHash, _ int) string {
+				return n.String()
+			})
+			sort.Strings(neighboursStrings) // Sort for deterministic output.
+			adjacencyMapStrings[v.String()] = neighboursStrings
+		}
+		return adjacencyMapStrings
+	}
+
+	testCases := []struct {
+		name                 string
+		cache                store.CacheStores
+		expectedAdjacencyMap map[string][]string
+	}{
+		{
+			name:                 "empty cache",
+			cache:                store.NewCacheStores(),
+			expectedAdjacencyMap: map[string][]string{},
+		},
+		{
+			name: "cache with Ingress and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&netv1.Ingress{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-ingress",
+						Namespace: "test-namespace",
+					},
+					Spec: netv1.IngressSpec{
+						IngressClassName: lo.ToPtr("test-ingress-class"),
+						Rules: []netv1.IngressRule{
+							{
+								IngressRuleValue: netv1.IngressRuleValue{
+									HTTP: &netv1.HTTPIngressRuleValue{
+										Paths: []netv1.HTTPIngressPath{
+											{
+												Backend: netv1.IngressBackend{
+													Service: &netv1.IngressServiceBackend{
+														Name: "test-service",
+													},
+												},
+											},
+											{
+												Backend: netv1.IngressBackend{
+													Resource: &corev1.TypedLocalObjectReference{
+														Name:     "test-kong-service-facade",
+														Kind:     "KongServiceFacade",
+														APIGroup: lo.ToPtr(incubatorv1alpha1.GroupVersion.Group),
+													},
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				&netv1.IngressClass{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test-ingress-class",
+					},
+				},
+				&corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-service",
+						Namespace: "test-namespace",
+					},
+				},
+				&incubatorv1alpha1.KongServiceFacade{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-kong-service-facade",
+						Namespace: "test-namespace",
+					},
+				},
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"Ingress:test-namespace/test-ingress": {},
+				"IngressClass:test-ingress-class": {
+					"Ingress:test-namespace/test-ingress",
+				},
+				"Service:test-namespace/test-service": {
+					"Ingress:test-namespace/test-ingress",
+				},
+				"KongServiceFacade:test-namespace/test-kong-service-facade": {
+					"Ingress:test-namespace/test-ingress",
+				},
+			},
+		},
+		{
+			name: "cache with HTTPRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.HTTPRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.HTTPRouteSpec{
+						Rules: []gatewayapi.HTTPRouteRule{
+							{
+								BackendRefs: []gatewayapi.HTTPBackendRef{
+									{
+										BackendRef: gatewayapi.BackendRef{
+											BackendObjectReference: gatewayapi.BackendObjectReference{
+												Name: "1",
+												Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"HTTPRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"HTTPRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"HTTPRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"HTTPRoute:test-namespace/test-route",
+				},
+			},
+		},
+		{
+			name: "cache with TLSRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.TLSRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.TLSRouteSpec{
+						Rules: []gatewayapi.TLSRouteRule{
+							{
+								BackendRefs: []gatewayapi.BackendRef{
+									{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"TLSRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"TLSRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"TLSRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"TLSRoute:test-namespace/test-route",
+				},
+			},
+		},
+		{
+			name: "cache with TCPRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.TCPRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.TCPRouteSpec{
+						Rules: []gatewayapi.TCPRouteRule{
+							{
+								BackendRefs: []gatewayapi.BackendRef{
+									{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"TCPRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"TCPRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"TCPRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"TCPRoute:test-namespace/test-route",
+				},
+			},
+		},
+		{
+			name: "cache with UDPRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.UDPRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.UDPRouteSpec{
+						Rules: []gatewayapi.UDPRouteRule{
+							{
+								BackendRefs: []gatewayapi.BackendRef{
+									{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"UDPRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"UDPRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"UDPRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"UDPRoute:test-namespace/test-route",
+				},
+			},
+		},
+		{
+			name: "cache with GRPCRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.GRPCRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.GRPCRouteSpec{
+						Rules: []gatewayapi.GRPCRouteRule{
+							{
+								BackendRefs: []gatewayapi.GRPCBackendRef{
+									{
+										BackendRef: gatewayapi.BackendRef{
+											BackendObjectReference: gatewayapi.BackendObjectReference{
+												Name: "1",
+												Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"GRPCRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"GRPCRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"GRPCRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"GRPCRoute:test-namespace/test-route",
+				},
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			p := fallback.NewDefaultCacheGraphProvider()
+			g, err := p.CacheToGraph(tc.cache)
+			require.NoError(t, err)
+			require.NotNil(t, g)
+			require.Equal(t, tc.expectedAdjacencyMap, adjacencyGraphStrings(t, g))
+		})
+	}
+}
diff --git a/internal/dataplane/fallback/fallback_test.go b/internal/dataplane/fallback/fallback_test.go
new file mode 100644
index 000000000..1fc05c58c
--- /dev/null
+++ b/internal/dataplane/fallback/fallback_test.go
@@ -0,0 +1,112 @@
+package fallback_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/fallback"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+)
+
+// mockGraphProvider is a mock implementation of the CacheGraphProvider interface.
+type mockGraphProvider struct {
+	graph               *fallback.ConfigGraph
+	lastCalledWithStore store.CacheStores
+}
+
+// CacheToGraph returns the graph that was set on the mockGraphProvider. It also records the last store that was passed to it.
+func (m *mockGraphProvider) CacheToGraph(s store.CacheStores) (*fallback.ConfigGraph, error) {
+	m.lastCalledWithStore = s
+	return m.graph, nil
+}
+
+func TestGenerator_GenerateExcludingAffected(t *testing.T) {
+	// We have to use real-world object types here as we're testing integration with store.CacheStores.
+	ingressClass := testIngressClass(t, "ingressClass")
+	service := testService(t, "service")
+	serviceFacade := testKongServiceFacade(t, "serviceFacade")
+	plugin := testKongPlugin(t, "kongPlugin")
+	inputCacheStores := cacheStoresFromObjs(t, ingressClass, service, serviceFacade, plugin)
+
+	// This graph doesn't reflect real dependencies between the objects - it's only used for testing purposes.
+	// It will be injected into the Generator via the mockGraphProvider.
+	// Dependency resolving between the objects is tested in TestResolveDependencies_* tests.
+	//
+	// Graph structure (edges define dependency -> dependant relationship):
+	//  ┌────────────┐  ┌──────┐
+	//  │ingressClass│  │plugin│
+	//  └──────┬─────┘  └──────┘
+	//         │
+	//     ┌───▼───┐
+	//     │service│
+	//     └───┬───┘
+	//         │
+	//  ┌──────▼──────┐
+	//  │serviceFacade│
+	//  └─────────────┘
+	graph, err := NewGraphBuilder().
+		WithVertices(ingressClass, service, serviceFacade, plugin).
+		WithEdge(ingressClass, service).
+		WithEdge(service, serviceFacade).
+		Build()
+	require.NoError(t, err)
+
+	graphProvider := &mockGraphProvider{graph: graph}
+	g := fallback.NewGenerator(graphProvider)
+
+	t.Run("ingressClass is broken", func(t *testing.T) {
+		fallbackCache, err := g.GenerateExcludingAffected(inputCacheStores, []fallback.ObjectHash{fallback.GetObjectHash(ingressClass)})
+		require.NoError(t, err)
+		require.Equal(t, inputCacheStores, graphProvider.lastCalledWithStore, "expected the generator to call CacheToGraph with the input cache stores")
+		require.NotSame(t, inputCacheStores, fallbackCache)
+		require.Empty(t, fallbackCache.IngressClassV1.List(), "ingressClass should be excluded as it's broken")
+		require.Empty(t, fallbackCache.Service.List(), "service should be excluded as it depends on ingressClass")
+		require.Empty(t, fallbackCache.KongServiceFacade.List(), "serviceFacade should be excluded as it depends on service")
+		require.ElementsMatch(t, fallbackCache.Plugin.List(), []any{plugin}, "plugin shouldn't be excluded as it doesn't depend on ingressClass")
+	})
+
+	t.Run("service is broken", func(t *testing.T) {
+		fallbackCache, err := g.GenerateExcludingAffected(inputCacheStores, []fallback.ObjectHash{fallback.GetObjectHash(service)})
+		require.NoError(t, err)
+		require.Equal(t, inputCacheStores, graphProvider.lastCalledWithStore, "expected the generator to call CacheToGraph with the input cache stores")
+		require.NotSame(t, inputCacheStores, fallbackCache)
+		require.Empty(t, fallbackCache.Service.List(), "service should be excluded as it's broken")
+		require.Empty(t, fallbackCache.KongServiceFacade.List(), "serviceFacade should be excluded as it depends on service")
+		require.ElementsMatch(t, fallbackCache.IngressClassV1.List(), []any{ingressClass}, "ingressClass shouldn't be excluded as it doesn't depend on service")
+		require.ElementsMatch(t, fallbackCache.Plugin.List(), []any{plugin}, "plugin shouldn't be excluded as it doesn't depend on service")
+	})
+
+	t.Run("serviceFacade is broken", func(t *testing.T) {
+		fallbackCache, err := g.GenerateExcludingAffected(inputCacheStores, []fallback.ObjectHash{fallback.GetObjectHash(serviceFacade)})
+		require.NoError(t, err)
+		require.Equal(t, inputCacheStores, graphProvider.lastCalledWithStore, "expected the generator to call CacheToGraph with the input cache stores")
+		require.NotSame(t, inputCacheStores, fallbackCache)
+		require.Empty(t, fallbackCache.KongServiceFacade.List(), "serviceFacade should be excluded as it's broken")
+		require.ElementsMatch(t, fallbackCache.IngressClassV1.List(), []any{ingressClass}, "ingressClass shouldn't be excluded as it doesn't depend on service")
+		require.ElementsMatch(t, fallbackCache.Service.List(), []any{service}, "service shouldn't be excluded as it doesn't depend on serviceFacade")
+		require.ElementsMatch(t, fallbackCache.Plugin.List(), []any{plugin}, "plugin shouldn't be excluded as it doesn't depend on serviceFacade")
+	})
+
+	t.Run("plugin is broken", func(t *testing.T) {
+		fallbackCache, err := g.GenerateExcludingAffected(inputCacheStores, []fallback.ObjectHash{fallback.GetObjectHash(plugin)})
+		require.NoError(t, err)
+		require.Equal(t, inputCacheStores, graphProvider.lastCalledWithStore, "expected the generator to call CacheToGraph with the input cache stores")
+		require.NotSame(t, inputCacheStores, fallbackCache)
+		require.Empty(t, fallbackCache.Plugin.List(), "plugin should be excluded as it's broken")
+		require.ElementsMatch(t, fallbackCache.IngressClassV1.List(), []any{ingressClass}, "ingressClass shouldn't be excluded as it doesn't depend on plugin")
+		require.ElementsMatch(t, fallbackCache.Service.List(), []any{service}, "service shouldn't be excluded as it doesn't depend on plugin")
+		require.ElementsMatch(t, fallbackCache.KongServiceFacade.List(), []any{serviceFacade}, "serviceFacade shouldn't be excluded as it doesn't depend on plugin")
+	})
+
+	t.Run("multiple objects are broken", func(t *testing.T) {
+		fallbackCache, err := g.GenerateExcludingAffected(inputCacheStores, []fallback.ObjectHash{fallback.GetObjectHash(ingressClass), fallback.GetObjectHash(service)})
+		require.NoError(t, err)
+		require.Equal(t, inputCacheStores, graphProvider.lastCalledWithStore, "expected the generator to call CacheToGraph with the input cache stores")
+		require.NotSame(t, inputCacheStores, fallbackCache)
+		require.Empty(t, fallbackCache.IngressClassV1.List(), "ingressClass should be excluded as it's broken")
+		require.Empty(t, fallbackCache.Service.List(), "service should be excluded as it's broken")
+		require.Empty(t, fallbackCache.KongServiceFacade.List(), "serviceFacade should be excluded as it depends on service")
+		require.ElementsMatch(t, fallbackCache.Plugin.List(), []any{plugin}, "plugin shouldn't be excluded as it doesn't depend on either ingressClass or service")
+	})
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_gatewayapi_test.go b/internal/dataplane/fallback/graph_dependencies_gatewayapi_test.go
new file mode 100644
index 000000000..9b1a6facc
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_gatewayapi_test.go
@@ -0,0 +1,460 @@
+package fallback_test
+
+import (
+	"testing"
+
+	"github.com/samber/lo"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+)
+
+func TestResolveDependencies_HTTPRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.HTTPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "HTTPRoute -> Service",
+			object: &gatewayapi.HTTPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.HTTPRouteSpec{
+					Rules: []gatewayapi.HTTPRouteRule{
+						{
+							BackendRefs: []gatewayapi.HTTPBackendRef{
+								{
+									BackendRef: gatewayapi.BackendRef{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+								{
+									BackendRef: gatewayapi.BackendRef{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "2",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "HTTPRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.HTTPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_TLSRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.TLSRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "TLSRoute -> Service",
+			object: &gatewayapi.TLSRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.TLSRouteSpec{
+					Rules: []gatewayapi.TLSRouteRule{
+						{
+							BackendRefs: []gatewayapi.BackendRef{
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "1",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "2",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "TLSRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.TLSRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_TCPRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.TCPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "TCPRoute -> Service",
+			object: &gatewayapi.TCPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.TCPRouteSpec{
+					Rules: []gatewayapi.TCPRouteRule{
+						{
+							BackendRefs: []gatewayapi.BackendRef{
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "1",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "2",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "TCPRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.TCPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_UDPRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.UDPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "UDPRoute -> Service",
+			object: &gatewayapi.UDPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.UDPRouteSpec{
+					Rules: []gatewayapi.UDPRouteRule{
+						{
+							BackendRefs: []gatewayapi.BackendRef{
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "1",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "2",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "UDPRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.UDPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_GRPCRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.GRPCRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "GRPCRoute -> Service",
+			object: &gatewayapi.GRPCRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.GRPCRouteSpec{
+					Rules: []gatewayapi.GRPCRouteRule{
+						{
+							BackendRefs: []gatewayapi.GRPCBackendRef{
+								{
+									BackendRef: gatewayapi.BackendRef{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+								{
+									BackendRef: gatewayapi.BackendRef{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "2",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "GRPCRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.GRPCRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_kong_test.go b/internal/dataplane/fallback/graph_dependencies_kong_test.go
new file mode 100644
index 000000000..bdc3ca024
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_kong_test.go
@@ -0,0 +1,470 @@
+package fallback_test
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+func TestResolveDependencies_KongPlugin(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &kongv1.KongPlugin{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongPlugin",
+					Namespace: testNamespace,
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testSecret(t, "1"),
+				testSecret(t, "2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "KongPlugin -> Secret referenced by ConfigFrom (secret with the same name exists in multiple namespaces)",
+			object: &kongv1.KongPlugin{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongPlugin",
+					Namespace: testNamespace,
+				},
+				ConfigFrom: &kongv1.ConfigSource{
+					SecretValue: kongv1.SecretValueFromSource{
+						Secret: "1",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testSecret(t, "1"),
+				testSecret(t, "1", func(s *corev1.Secret) {
+					s.Namespace = "another-namespace"
+				}),
+			),
+			expected: []client.Object{testSecret(t, "1")},
+		},
+		{
+			name: "KongPlugin -> Secret referenced by ConfigFrom does not exist in the same namespace as the KongPlugin",
+			object: &kongv1.KongPlugin{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongPlugin",
+					Namespace: testNamespace,
+				},
+				ConfigFrom: &kongv1.ConfigSource{
+					SecretValue: kongv1.SecretValueFromSource{
+						Secret: "2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testSecret(t, "1"),
+				testSecret(t, "2", func(s *corev1.Secret) {
+					s.Namespace = "another-namespace"
+				}),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "KongPlugin -> two Secrets referenced by ConfigPatches (Secret with the same name exists in multiple namespaces)",
+			object: &kongv1.KongPlugin{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongPlugin",
+					Namespace: testNamespace,
+				},
+				ConfigPatches: []kongv1.ConfigPatch{
+					{
+						ValueFrom: kongv1.ConfigSource{
+							SecretValue: kongv1.SecretValueFromSource{
+								Secret: "1",
+							},
+						},
+					},
+					{
+						ValueFrom: kongv1.ConfigSource{
+							SecretValue: kongv1.SecretValueFromSource{
+								Secret: "2",
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testSecret(t, "1"),
+				testSecret(t, "1", func(s *corev1.Secret) {
+					s.Namespace = "another-namespace"
+				}),
+				testSecret(t, "2"),
+			),
+			expected: []client.Object{testSecret(t, "1"), testSecret(t, "2")},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_KongClusterPlugin(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &kongv1.KongClusterPlugin{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-KongClusterPlugin",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testSecret(t, "1"),
+				testSecret(t, "2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "KongClusterPlugin -> Secret referenced by ConfigFrom (Secret with the same name exists in multiple namespaces)",
+			object: &kongv1.KongClusterPlugin{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-KongClusterPlugin",
+				},
+				ConfigFrom: &kongv1.NamespacedConfigSource{
+					SecretValue: kongv1.NamespacedSecretValueFromSource{
+						Namespace: testNamespace,
+						Secret:    "1",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testSecret(t, "1"),
+				testSecret(t, "1", func(s *corev1.Secret) {
+					s.Namespace = "another-namespace"
+				}),
+			),
+			expected: []client.Object{testSecret(t, "1")},
+		},
+		{
+			name: "KongClusterPlugin -> Secret referenced by ConfigFrom does not exists",
+			object: &kongv1.KongClusterPlugin{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-KongClusterPlugin",
+				},
+				ConfigFrom: &kongv1.NamespacedConfigSource{
+					SecretValue: kongv1.NamespacedSecretValueFromSource{
+						Namespace: testNamespace,
+						Secret:    "1",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testSecret(t, "1", func(s *corev1.Secret) {
+					s.Namespace = "another-namespace"
+				}),
+				testSecret(t, "2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "KongClusterPlugin -> two Secrets referenced by ConfigPatches (Secret with the same name exists in multiple namespaces)",
+			object: &kongv1.KongClusterPlugin{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-KongClusterPlugin",
+				},
+				ConfigPatches: []kongv1.NamespacedConfigPatch{
+					{
+						ValueFrom: kongv1.NamespacedConfigSource{
+							SecretValue: kongv1.NamespacedSecretValueFromSource{
+								Namespace: "another-namespace",
+								Secret:    "1",
+							},
+						},
+					},
+					{
+						ValueFrom: kongv1.NamespacedConfigSource{
+							SecretValue: kongv1.NamespacedSecretValueFromSource{
+								Namespace: testNamespace,
+								Secret:    "2",
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testSecret(t, "1"),
+				testSecret(t, "1", func(s *corev1.Secret) {
+					s.Namespace = "another-namespace"
+				}),
+				testSecret(t, "2"),
+			),
+			expected: []client.Object{
+				testSecret(t, "1", func(s *corev1.Secret) {
+					s.Namespace = "another-namespace"
+				}),
+				testSecret(t, "2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_KongConsumer(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &kongv1.KongConsumer{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongConsumer",
+					Namespace: testNamespace,
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "KongConsumer -> plugins - annotation (KongPlugin and KongClusterPlugin with the same name)",
+			object: &kongv1.KongConsumer{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-kongconsumer",
+					Namespace: testNamespace,
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1, 2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{testKongPlugin(t, "1"), testKongPlugin(t, "2")},
+		},
+		{
+			name: "KongConsumer -> plugins - annotation (KongPlugin and KongClusterPlugin with different names)",
+			object: &kongv1.KongConsumer{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-kongconsumer",
+					Namespace: testNamespace,
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1, 3",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "3"),
+			),
+			expected: []client.Object{testKongPlugin(t, "1"), testKongClusterPlugin(t, "3")},
+		},
+		{
+			name: "KongConsumer -> plugins - annotation (KongClusterPlugin)",
+			object: &kongv1.KongConsumer{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-kongconsumer",
+					Namespace: testNamespace,
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "3",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "3"),
+			),
+			expected: []client.Object{testKongClusterPlugin(t, "3")},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_KongConsumerGroup(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &kongv1beta1.KongConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongConsumerGroup",
+					Namespace: testNamespace,
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "KongConsumerGroup -> plugins - annotation (KongPlugin and KongClusterPlugin with the same name)",
+			object: &kongv1beta1.KongConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongConsumerGroup",
+					Namespace: testNamespace,
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1, 2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{testKongPlugin(t, "1"), testKongPlugin(t, "2")},
+		},
+		{
+			name: "KongConsumerGroup -> plugins - annotation (KongPlugin and KongClusterPlugin with different names)",
+			object: &kongv1beta1.KongConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongConsumerGroup",
+					Namespace: testNamespace,
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1, 3",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "3"),
+			),
+			expected: []client.Object{testKongPlugin(t, "1"), testKongClusterPlugin(t, "3")},
+		},
+		{
+			name: "KongConsumerGroup -> plugins - annotation (KongClusterPlugin)",
+			object: &kongv1beta1.KongConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongConsumerGroup",
+					Namespace: testNamespace,
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "3",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "3"),
+			),
+			expected: []client.Object{testKongClusterPlugin(t, "3")},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_KongServiceFacade(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &incubatorv1alpha1.KongServiceFacade{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongServiceFacade",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "KongServiceFacade -> plugins - annotation (KongPlugin and KongClusterPlugin with the same name) and KongUpstreamPolicy",
+			object: &incubatorv1alpha1.KongServiceFacade{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongServiceFacade",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1, 2",
+						kongv1beta1.KongUpstreamPolicyAnnotationKey:           "1",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongUpstreamPolicy(t, "1"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{testKongPlugin(t, "1"), testKongPlugin(t, "2"), testKongUpstreamPolicy(t, "1")},
+		},
+		{
+			name: "KongServiceFacade -> plugins - annotation (KongPlugin and KongClusterPlugin with different names)",
+			object: &incubatorv1alpha1.KongServiceFacade{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongServiceFacade",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1, 3",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "3"),
+				testKongUpstreamPolicy(t, "1"),
+			),
+			expected: []client.Object{testKongPlugin(t, "1"), testKongClusterPlugin(t, "3")},
+		},
+		{
+			name: "KongServiceFacade -> plugins - annotation (KongClusterPlugin) and KongUpstreamPolicy",
+			object: &incubatorv1alpha1.KongServiceFacade{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongServiceFacade",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "3",
+						kongv1beta1.KongUpstreamPolicyAnnotationKey:           "3",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "3"),
+				testKongUpstreamPolicy(t, "3"),
+			),
+			expected: []client.Object{testKongClusterPlugin(t, "3"), testKongUpstreamPolicy(t, "3")},
+		},
+		{
+			name: "KongServiceFacade -> KongUpstreamPolicy - the same name in different namespaces",
+			object: &incubatorv1alpha1.KongServiceFacade{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-KongServiceFacade",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						kongv1beta1.KongUpstreamPolicyAnnotationKey: "1",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongUpstreamPolicy(t, "1"),
+				testKongUpstreamPolicy(t, "1", func(kup *kongv1beta1.KongUpstreamPolicy) {
+					kup.Namespace = "other-namespace"
+				}),
+			),
+			expected: []client.Object{testKongUpstreamPolicy(t, "1")},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_service_test.go b/internal/dataplane/fallback/graph_dependencies_service_test.go
new file mode 100644
index 000000000..c47f9c7ff
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_service_test.go
@@ -0,0 +1,113 @@
+package fallback_test
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
+)
+
+func TestResolveDependencies_Service(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-service",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "Service -> plugins - annotation (KongPlugin and KongClusterPlugin with the same name) and KongUpstreamPolicy",
+			object: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-service",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1, 2",
+						kongv1beta1.KongUpstreamPolicyAnnotationKey:           "1",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongUpstreamPolicy(t, "1"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{testKongPlugin(t, "1"), testKongPlugin(t, "2"), testKongUpstreamPolicy(t, "1")},
+		},
+		{
+			name: "Service -> plugins - annotation (KongPlugin and KongClusterPlugin with different names)",
+			object: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-service",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1, 3",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "3"),
+				testKongUpstreamPolicy(t, "1"),
+			),
+			expected: []client.Object{testKongPlugin(t, "1"), testKongClusterPlugin(t, "3")},
+		},
+		{
+			name: "Service -> plugins - annotation (KongClusterPlugin) and KongUpstreamPolicy",
+			object: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-service",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "3",
+						kongv1beta1.KongUpstreamPolicyAnnotationKey:           "3",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "3"),
+				testKongUpstreamPolicy(t, "3"),
+			),
+			expected: []client.Object{testKongClusterPlugin(t, "3"), testKongUpstreamPolicy(t, "3")},
+		},
+		{
+			name: "Service -> KongUpstreamPolicy - the same name in different namespaces",
+			object: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-service",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						kongv1beta1.KongUpstreamPolicyAnnotationKey: "1",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongUpstreamPolicy(t, "1"),
+				testKongUpstreamPolicy(t, "1", func(kup *kongv1beta1.KongUpstreamPolicy) {
+					kup.Namespace = "other-namespace"
+				}),
+			),
+			expected: []client.Object{testKongUpstreamPolicy(t, "1")},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
diff --git a/internal/dataplane/fallback/graph_test.go b/internal/dataplane/fallback/graph_test.go
index 4a7de5736..74338766f 100644
--- a/internal/dataplane/fallback/graph_test.go
+++ b/internal/dataplane/fallback/graph_test.go
@@ -1,123 +1,83 @@
 package fallback_test
 
 import (
-	"sort"
 	"testing"
 
-	"github.com/samber/lo"
 	"github.com/stretchr/testify/require"
-	corev1 "k8s.io/api/core/v1"
-	netv1 "k8s.io/api/networking/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/fallback"
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
-	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
 )
 
-func TestNewConfigGraphFromCacheStores(t *testing.T) {
-	// adjacencyGraphStrings returns a map of stringified vertices and their neighbours
-	// in the given graph for easy comparison.
-	adjacencyGraphStrings := func(t *testing.T, g *fallback.ConfigGraph) map[string][]string {
-		am, err := g.AdjacencyMap()
-		require.NoError(t, err)
-		adjacencyMapStrings := make(map[string][]string, len(am))
-		for v, neighbours := range am {
-			neighboursStrings := lo.Map(neighbours, func(n fallback.ObjectHash, _ int) string {
-				return n.String()
-			})
-			sort.Strings(neighboursStrings) // Sort for deterministic output.
-			adjacencyMapStrings[v.String()] = neighboursStrings
-		}
-		return adjacencyMapStrings
-	}
+func TestConfigGraph_SubgraphObjects(t *testing.T) {
+	var (
+		A = NewMockObject("A")
+		B = NewMockObject("B")
+		C = NewMockObject("C")
+		D = NewMockObject("D")
+		E = NewMockObject("E")
+		F = NewMockObject("F")
+		G = NewMockObject("G")
+		H = NewMockObject("H")
+	)
 
-	testCases := []struct {
-		name                 string
-		cache                store.CacheStores
-		expectedAdjacencyMap map[string][]string
-	}{
-		{
-			name:                 "empty cache",
-			cache:                store.NewCacheStores(),
-			expectedAdjacencyMap: map[string][]string{},
-		},
-		{
-			name: "cache with Ingress and its dependencies",
-			cache: cacheStoresFromObjs(t,
-				&netv1.Ingress{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      "test-ingress",
-						Namespace: "test-namespace",
-					},
-					Spec: netv1.IngressSpec{
-						IngressClassName: lo.ToPtr("test-ingress-class"),
-						Rules: []netv1.IngressRule{
-							{
-								IngressRuleValue: netv1.IngressRuleValue{
-									HTTP: &netv1.HTTPIngressRuleValue{
-										Paths: []netv1.HTTPIngressPath{
-											{
-												Backend: netv1.IngressBackend{
-													Service: &netv1.IngressServiceBackend{
-														Name: "test-service",
-													},
-												},
-											},
-											{
-												Backend: netv1.IngressBackend{
-													Resource: &corev1.TypedLocalObjectReference{
-														Name:     "test-kong-service-facade",
-														Kind:     "KongServiceFacade",
-														APIGroup: lo.ToPtr(incubatorv1alpha1.GroupVersion.Group),
-													},
-												},
-											},
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-				&netv1.IngressClass{
-					ObjectMeta: metav1.ObjectMeta{
-						Name: "test-ingress-class",
-					},
-				},
-				&corev1.Service{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      "test-service",
-						Namespace: "test-namespace",
-					},
-				},
-				&incubatorv1alpha1.KongServiceFacade{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      "test-kong-service-facade",
-						Namespace: "test-namespace",
-					},
-				},
-			),
-			expectedAdjacencyMap: map[string][]string{
-				"Ingress:test-namespace/test-ingress": {},
-				"IngressClass:test-ingress-class": {
-					"Ingress:test-namespace/test-ingress",
-				},
-				"Service:test-namespace/test-service": {
-					"Ingress:test-namespace/test-ingress",
-				},
-				"KongServiceFacade:test-namespace/test-kong-service-facade": {
-					"Ingress:test-namespace/test-ingress",
-				},
-			},
-		},
-	}
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			g, err := fallback.NewConfigGraphFromCacheStores(tc.cache)
-			require.NoError(t, err)
-			require.NotNil(t, g)
-			require.Equal(t, tc.expectedAdjacencyMap, adjacencyGraphStrings(t, g))
-		})
-	}
+	// Graph structure (edges define dependency -> dependant relationship):
+	//     ┌───┐     ┌───┐   ┌───┐   ┌───┐
+	//     │ A │     │ E │   │ F │   │ G │
+	//     └─┬─┘     └─┬─┘   └─┬─┘   └─┬─┘
+	//       │         │       │       │
+	//   ┌───┴───┐     │       └───┬───┘
+	//   │       │     │           │
+	// ┌─▼─┐   ┌─▼─┐   │         ┌─▼─┐
+	// │ B │   │ C │   │         │ H │
+	// └───┘   └─┬─┘   │         └───┘
+	//           │     │
+	//           ├─────┘
+	//           │
+	//         ┌─▼─┐
+	//         │ D │
+	//         └───┘
+	g, err := NewGraphBuilder().
+		WithVertices(A, B, C, D, E).
+		WithEdge(A, B).
+		WithEdge(A, C).
+		WithEdge(C, D).
+		WithEdge(E, D).
+		WithVertices(F, G, H).
+		WithEdge(F, H).
+		WithEdge(G, H).
+		Build()
+	require.NoError(t, err)
+
+	objects, err := g.SubgraphObjects(fallback.GetObjectHash(A))
+	require.NoError(t, err)
+	require.ElementsMatch(t, []client.Object{A, B, C, D}, objects)
+
+	objects, err = g.SubgraphObjects(fallback.GetObjectHash(B))
+	require.NoError(t, err)
+	require.ElementsMatch(t, []client.Object{B}, objects)
+
+	objects, err = g.SubgraphObjects(fallback.GetObjectHash(C))
+	require.NoError(t, err)
+	require.ElementsMatch(t, []client.Object{C, D}, objects)
+
+	objects, err = g.SubgraphObjects(fallback.GetObjectHash(D))
+	require.NoError(t, err)
+	require.ElementsMatch(t, []client.Object{D}, objects)
+
+	objects, err = g.SubgraphObjects(fallback.GetObjectHash(E))
+	require.NoError(t, err)
+	require.ElementsMatch(t, []client.Object{E, D}, objects)
+
+	objects, err = g.SubgraphObjects(fallback.GetObjectHash(F))
+	require.NoError(t, err)
+	require.ElementsMatch(t, []client.Object{F, H}, objects)
+
+	objects, err = g.SubgraphObjects(fallback.GetObjectHash(G))
+	require.NoError(t, err)
+	require.ElementsMatch(t, []client.Object{G, H}, objects)
+
+	objects, err = g.SubgraphObjects(fallback.GetObjectHash(H))
+	require.NoError(t, err)
+	require.ElementsMatch(t, []client.Object{H}, objects)
 }
diff --git a/internal/dataplane/fallback/helpers_test.go b/internal/dataplane/fallback/helpers_test.go
index 1e39d28d6..24845f6c1 100644
--- a/internal/dataplane/fallback/helpers_test.go
+++ b/internal/dataplane/fallback/helpers_test.go
@@ -1,13 +1,18 @@
 package fallback_test
 
 import (
+	"fmt"
 	"testing"
 
 	corev1 "k8s.io/api/core/v1"
 	netv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/fallback"
 	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
 	"github.com/kong/kubernetes-ingress-controller/v3/test/helpers"
 )
@@ -33,6 +38,19 @@ func testService(t *testing.T, name string) *corev1.Service {
 	})
 }
 
+func testSecret(t *testing.T, name string, modifiers ...func(s *corev1.Secret)) *corev1.Secret {
+	s := helpers.WithTypeMeta(t, &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: testNamespace,
+		},
+	})
+	for _, mod := range modifiers {
+		mod(s)
+	}
+	return s
+}
+
 func testKongServiceFacade(t *testing.T, name string) *incubatorv1alpha1.KongServiceFacade {
 	return helpers.WithTypeMeta(t, &incubatorv1alpha1.KongServiceFacade{
 		ObjectMeta: metav1.ObjectMeta{
@@ -59,3 +77,85 @@ func testKongClusterPlugin(t *testing.T, name string) *kongv1.KongClusterPlugin
 		},
 	})
 }
+
+func testKongUpstreamPolicy(t *testing.T, name string, modifiers ...func(kup *kongv1beta1.KongUpstreamPolicy)) *kongv1beta1.KongUpstreamPolicy {
+	kup := helpers.WithTypeMeta(t, &kongv1beta1.KongUpstreamPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: testNamespace,
+		},
+	})
+	for _, mod := range modifiers {
+		mod(kup)
+	}
+	return kup
+}
+
+// GraphBuilder is a helper to build a graph for testing.
+type GraphBuilder struct {
+	vertices []client.Object
+	edges    map[fallback.ObjectHash][]fallback.ObjectHash
+}
+
+func NewGraphBuilder() *GraphBuilder {
+	return &GraphBuilder{
+		edges: make(map[fallback.ObjectHash][]fallback.ObjectHash),
+	}
+}
+
+// WithVertices adds vertices to the graph.
+func (b *GraphBuilder) WithVertices(objs ...client.Object) *GraphBuilder {
+	b.vertices = append(b.vertices, objs...)
+	return b
+}
+
+// WithEdge adds an edge between two vertices in the graph.
+func (b *GraphBuilder) WithEdge(from, to client.Object) *GraphBuilder {
+	fromHash := fallback.GetObjectHash(from)
+	toHash := fallback.GetObjectHash(to)
+	b.edges[fromHash] = append(b.edges[fromHash], toHash)
+	return b
+}
+
+// Build builds the graph.
+func (b *GraphBuilder) Build() (*fallback.ConfigGraph, error) {
+	g := fallback.NewConfigGraph()
+
+	for _, v := range b.vertices {
+		if err := g.AddVertex(v); err != nil {
+			return nil, fmt.Errorf("failed to add vertex %s to the graph: %w", fallback.GetObjectHash(v), err)
+		}
+	}
+
+	for from, tos := range b.edges {
+		for _, to := range tos {
+			if err := g.AddEdge(from, to); err != nil {
+				return nil, fmt.Errorf("failed to add edge from %s to %s: %w", from, to, err)
+			}
+		}
+	}
+
+	return g, nil
+}
+
+var _ client.Object = &MockObject{}
+
+// MockObject is a mock object that implements the client.Object interface.
+type MockObject struct {
+	metav1.ObjectMeta
+	metav1.TypeMeta
+}
+
+// NewMockObject creates a new mock object with the given name.
+func NewMockObject(name string) *MockObject {
+	return &MockObject{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+}
+
+// DeepCopyObject is required for runtime.Object interface.
+func (m *MockObject) DeepCopyObject() runtime.Object {
+	return m
+}
diff --git a/internal/dataplane/kong_client_test.go b/internal/dataplane/kong_client_test.go
index de1fda47f..1a1c07403 100644
--- a/internal/dataplane/kong_client_test.go
+++ b/internal/dataplane/kong_client_test.go
@@ -250,14 +250,8 @@ type mockUpdateStrategy struct {
 	onUpdate func(content sendconfig.ContentWithHash) error
 }
 
-func (m *mockUpdateStrategy) Update(_ context.Context, content sendconfig.ContentWithHash) (
-	err error,
-	resourceErrors []sendconfig.ResourceError,
-	rawErrBody []byte,
-	resourceErrorsParseErr error,
-) {
-	err = m.onUpdate(content)
-	return err, nil, nil, nil
+func (m *mockUpdateStrategy) Update(_ context.Context, targetContent sendconfig.ContentWithHash) (err error) {
+	return m.onUpdate(targetContent)
 }
 
 func (m *mockUpdateStrategy) MetricsProtocol() metrics.Protocol {
diff --git a/internal/dataplane/kongstate/kongstate_test.go b/internal/dataplane/kongstate/kongstate_test.go
index e87399b3b..b9bbcf21b 100644
--- a/internal/dataplane/kongstate/kongstate_test.go
+++ b/internal/dataplane/kongstate/kongstate_test.go
@@ -458,7 +458,8 @@ func TestGetPluginRelations(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := tt.args.state.getPluginRelations(); !reflect.DeepEqual(got, tt.want) {
+			store, _ := store.NewFakeStore(store.FakeObjects{})
+			if got := tt.args.state.getPluginRelations(store, logr.Discard()); !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("getPluginRelations() = %v, want %v", got, tt.want)
 			}
 		})
diff --git a/internal/dataplane/sendconfig/backoff_strategy_test.go b/internal/dataplane/sendconfig/backoff_strategy_test.go
index 8b45a2212..38b06c6ba 100644
--- a/internal/dataplane/sendconfig/backoff_strategy_test.go
+++ b/internal/dataplane/sendconfig/backoff_strategy_test.go
@@ -24,19 +24,14 @@ func newMockUpdateStrategy(shouldSucceed bool) *mockUpdateStrategy {
 	return &mockUpdateStrategy{shouldSucceed: shouldSucceed}
 }
 
-func (m *mockUpdateStrategy) Update(context.Context, sendconfig.ContentWithHash) (
-	err error,
-	resourceErrors []sendconfig.ResourceError,
-	rawErrBody []byte,
-	resourceErrorsParseErr error,
-) {
+func (m *mockUpdateStrategy) Update(context.Context, sendconfig.ContentWithHash) (err error) {
 	m.wasUpdateCalled = true
 
 	if !m.shouldSucceed {
-		return errors.New("update failure occurred"), nil, nil, nil
+		return errors.New("update failure occurred")
 	}
 
-	return nil, nil, nil, nil
+	return nil
 }
 
 func (m *mockUpdateStrategy) MetricsProtocol() metrics.Protocol {
@@ -120,7 +115,7 @@ func TestUpdateStrategyWithBackoff(t *testing.T) {
 			backoffStrategy := newMockBackoffStrategy(tc.updateShouldBeAllowed)
 
 			decoratedStrategy := sendconfig.NewUpdateStrategyWithBackoff(updateStrategy, backoffStrategy, logger)
-			err, _, _, _ := decoratedStrategy.Update(ctx, sendconfig.ContentWithHash{})
+			err := decoratedStrategy.Update(ctx, sendconfig.ContentWithHash{})
 			if tc.expectError != nil {
 				require.Equal(t, tc.expectError, err)
 			} else {
diff --git a/internal/dataplane/sendconfig/errors_test.go b/internal/dataplane/sendconfig/errors_test.go
new file mode 100644
index 000000000..4b9e7f162
--- /dev/null
+++ b/internal/dataplane/sendconfig/errors_test.go
@@ -0,0 +1,39 @@
+package sendconfig_test
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/samber/lo"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/sendconfig"
+)
+
+type testError struct{}
+
+func (t testError) Error() string {
+	return "test error"
+}
+
+func TestUpdateError(t *testing.T) {
+	someResourceFailure := lo.Must(failures.NewResourceFailure("some reason", &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "name",
+			Namespace: "namespace",
+		},
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+	}))
+
+	updateErr := sendconfig.NewUpdateError([]failures.ResourceFailure{someResourceFailure}, testError{})
+	require.Equal(t, "test error", updateErr.Error())
+	require.Len(t, updateErr.ResourceFailures(), 1)
+	unwraps := errors.As(updateErr, &testError{})
+	require.True(t, unwraps, "UpdateError should unwrap to inner error")
+}
diff --git a/internal/dataplane/translator/testdata/golden/grpcroute-example/default_golden.yaml b/internal/dataplane/translator/testdata/golden/grpcroute-example/default_golden.yaml
index 4f3152ba6..87bc8c088 100644
--- a/internal/dataplane/translator/testdata/golden/grpcroute-example/default_golden.yaml
+++ b/internal/dataplane/translator/testdata/golden/grpcroute-example/default_golden.yaml
@@ -29,7 +29,7 @@ services:
     - k8s-namespace:default
     - k8s-kind:GRPCRoute
     - k8s-group:gateway.networking.k8s.io
-    - k8s-version:v1alpha2
+    - k8s-version:v1
   tags:
   - k8s-name:UNKNOWN
   - k8s-namespace:UNKNOWN
diff --git a/internal/dataplane/translator/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml b/internal/dataplane/translator/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
index 5c585b916..8c9197377 100644
--- a/internal/dataplane/translator/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
+++ b/internal/dataplane/translator/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
@@ -24,7 +24,7 @@ services:
     - k8s-namespace:default
     - k8s-kind:GRPCRoute
     - k8s-group:gateway.networking.k8s.io
-    - k8s-version:v1alpha2
+    - k8s-version:v1
   tags:
   - k8s-name:UNKNOWN
   - k8s-namespace:UNKNOWN
diff --git a/internal/dataplane/translator/testdata/golden/grpcroute-example/in.yaml b/internal/dataplane/translator/testdata/golden/grpcroute-example/in.yaml
index 45461deb9..be2c54ee9 100644
--- a/internal/dataplane/translator/testdata/golden/grpcroute-example/in.yaml
+++ b/internal/dataplane/translator/testdata/golden/grpcroute-example/in.yaml
@@ -1,4 +1,4 @@
-apiVersion: gateway.networking.k8s.io/v1alpha2
+apiVersion: gateway.networking.k8s.io/v1
 kind: GRPCRoute
 metadata:
   name: grpcbin
@@ -30,4 +30,4 @@ spec:
     port: 443
     targetPort: 9001
   selector:
-    app: grpcbin
\ No newline at end of file
+    app: grpcbin
diff --git a/internal/dataplane/translator/translate_utils_test.go b/internal/dataplane/translator/translate_utils_test.go
index 48d576e4a..9bfb62425 100644
--- a/internal/dataplane/translator/translate_utils_test.go
+++ b/internal/dataplane/translator/translate_utils_test.go
@@ -250,7 +250,7 @@ func TestGetPermittedForReferenceGrantFrom(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.msg, func(t *testing.T) {
-			result := getPermittedForReferenceGrantFrom(tt.from, grants)
+			result := GetPermittedForReferenceGrantFrom(tt.from, grants)
 			assert.Equal(t, tt.result, result)
 		})
 	}
diff --git a/internal/dataplane/translator/wrappers_refchecker_test.go b/internal/dataplane/translator/wrappers_refchecker_test.go
index 233a73b78..9a18525c5 100644
--- a/internal/dataplane/translator/wrappers_refchecker_test.go
+++ b/internal/dataplane/translator/wrappers_refchecker_test.go
@@ -111,7 +111,7 @@ func TestRefChecker_IsRefAllowedByGrant(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			rc := newRefCheckerForRoute(tc.route, tc.backendRef)
+			rc := gatewayapi.NewRefCheckerForRoute(tc.route, tc.backendRef)
 			result := rc.IsRefAllowedByGrant(tc.allowedRefs)
 			require.Equal(t, tc.expected, result)
 		})
diff --git a/test/integration/plugin_test.go b/test/integration/plugin_test.go
index cbfd9da7e..dc21fd0bf 100644
--- a/test/integration/plugin_test.go
+++ b/test/integration/plugin_test.go
@@ -11,6 +11,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/kong/go-kong/kong"
 	"github.com/kong/kubernetes-testing-framework/pkg/clusters"
 	"github.com/kong/kubernetes-testing-framework/pkg/utils/kubernetes/generators"
@@ -19,8 +20,11 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	gatewayclient "sigs.k8s.io/gateway-api/pkg/client/clientset/versioned"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/labels"
 	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
 	"github.com/kong/kubernetes-ingress-controller/v3/pkg/clientset"
 	"github.com/kong/kubernetes-ingress-controller/v3/test"
@@ -472,3 +476,193 @@ func TestPluginOrdering(t *testing.T) {
 	require.NoError(t, clusters.DeleteIngress(ctx, env.Cluster(), ns.Name, ingress))
 	helpers.EventuallyExpectHTTP404WithNoRoute(t, proxyHTTPURL, proxyHTTPURL.Host, "/test_plugin_ordering", ingressWait, waitTick, nil)
 }
+
+// TestPluginCrossNamespaceReference creates an Ingress in one namespace and a KongConsumer in another. It tests that
+// the controller can generate a plugin attached to the generated route and consumer if and only if a ReferenceGrant
+// allows access to the KongPlugin from KongConsumers in a remote namespace.
+func TestPluginCrossNamespaceReference(t *testing.T) {
+	ctx := context.Background()
+
+	t.Parallel()
+	ns, cleaner := helpers.Setup(ctx, t, env)
+	cluster := env.Cluster()
+	remote, err := clusters.GenerateNamespace(ctx, cluster, helpers.LabelValueForTest(t))
+	require.NoError(t, err)
+	cleaner.AddNamespace(remote)
+
+	gatewayClient, err := gatewayclient.NewForConfig(env.Cluster().Config())
+	require.NoError(t, err)
+
+	t.Log("deploying a minimal HTTP container deployment to test Ingress routes")
+	container := generators.NewContainer("httpbin", test.HTTPBinImage, test.HTTPBinPort)
+	deployment := generators.NewDeploymentForContainer(container)
+	deployment, err = env.Cluster().Client().AppsV1().Deployments(ns.Name).Create(ctx, deployment, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(deployment)
+
+	t.Logf("exposing deployment %s via service", deployment.Name)
+	service := generators.NewServiceForDeployment(deployment, corev1.ServiceTypeLoadBalancer)
+	_, err = env.Cluster().Client().CoreV1().Services(ns.Name).Create(ctx, service, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(service)
+
+	t.Logf("creating an ingress for service %s with ingress.class %s", service.Name, consts.IngressClass)
+	ingress := generators.NewIngressForService("/test_plugin_reference", map[string]string{
+		"konghq.com/strip-path": "true",
+	}, service)
+	ingress.Spec.IngressClassName = kong.String(consts.IngressClass)
+	require.NoError(t, clusters.DeployIngress(ctx, env.Cluster(), ns.Name, ingress))
+	cleaner.Add(ingress)
+
+	t.Log("waiting for routes from Ingress to be operational")
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		resp, err := helpers.DefaultHTTPClient().Get(fmt.Sprintf("%s/test_plugin_reference", proxyHTTPURL))
+		if !assert.NoError(c, err) {
+			return
+		}
+		defer resp.Body.Close()
+		if resp.StatusCode == http.StatusOK {
+			b := new(bytes.Buffer)
+			_, err := b.ReadFrom(resp.Body)
+			assert.NoError(c, err)
+			assert.True(c, strings.Contains(b.String(), "<title>httpbin.org</title>"))
+		}
+	}, ingressWait, waitTick)
+
+	t.Log("creating plugins")
+	kongplugin := &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ns.Name,
+			Name:      "teapot",
+		},
+		InstanceName: "example",
+		PluginName:   "request-termination",
+		Config: apiextensionsv1.JSON{
+			Raw: []byte(`{"status_code": 418}`),
+		},
+	}
+	c, err := clientset.NewForConfig(env.Cluster().Config())
+	require.NoError(t, err)
+	kongplugin, err = c.ConfigurationV1().KongPlugins(ns.Name).Create(ctx, kongplugin, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(kongplugin)
+
+	// this isn't strictly necessary, but it's convenient for diagnosing issues if the test breaks, as it confirms
+	// that a request was indeed recognized as being from the consumer. unlike the other consumer plugin, it's associated
+	// with the consumer alone and works regardless of the grant state or any cross-namespace shenanigans
+	kongslugin := &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ns.Name,
+			Name:      "snurch",
+		},
+		InstanceName: "snurch",
+		PluginName:   "correlation-id",
+		Config: apiextensionsv1.JSON{
+			Raw: []byte(`{"header_name": "snurch", "echo_downstream": true}`),
+		},
+	}
+	kongslugin, err = c.ConfigurationV1().KongPlugins(ns.Name).Create(ctx, kongslugin, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(kongslugin)
+
+	t.Log("creating consumer and credential")
+	credential := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: uuid.NewString(),
+			Labels: map[string]string{
+				labels.LabelPrefix + labels.CredentialKey: "key-auth",
+			},
+		},
+		StringData: map[string]string{
+			"key": "thirtytangas",
+		},
+	}
+	_, err = env.Cluster().Client().CoreV1().Secrets(remote.Name).Create(ctx, credential, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(credential)
+
+	consumer := &kongv1.KongConsumer{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: uuid.NewString(),
+			Annotations: map[string]string{
+				annotations.IngressClassKey:                           consts.IngressClass,
+				annotations.AnnotationPrefix + annotations.PluginsKey: strings.Join([]string{kongslugin.Name, fmt.Sprintf("%s:%s", kongplugin.Namespace, kongplugin.Name)}, ","),
+			},
+		},
+		Username:    uuid.NewString(),
+		Credentials: []string{credential.Name},
+	}
+	_, err = c.ConfigurationV1().KongConsumers(remote.Name).Create(ctx, consumer, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(consumer)
+
+	t.Log("creating auth plugin to identify consumer accessing route")
+	authplugin := &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ns.Name,
+			Name:      "keykey",
+		},
+		PluginName: "key-auth",
+	}
+	authplugin, err = c.ConfigurationV1().KongPlugins(ns.Name).Create(ctx, authplugin, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(authplugin)
+
+	cleaner.Add(kongplugin)
+	t.Logf("updating Ingress to use auth plugin %s", kongplugin.Name)
+	require.Eventually(t, func() bool {
+		ingress, err := env.Cluster().Client().NetworkingV1().Ingresses(ns.Name).Get(ctx, ingress.Name, metav1.GetOptions{})
+		if err != nil {
+			return false
+		}
+		ingress.ObjectMeta.Annotations[annotations.AnnotationPrefix+annotations.PluginsKey] = strings.Join([]string{kongplugin.Name, authplugin.Name}, ",")
+		_, err = env.Cluster().Client().NetworkingV1().Ingresses(ns.Name).Update(ctx, ingress, metav1.UpdateOptions{})
+		return err == nil
+	}, ingressWait, waitTick)
+
+	t.Logf("validating that plugin %s is not configured without a grant", kongplugin.Name)
+	assert.Never(t, func() bool {
+		req := helpers.MustHTTPRequest(t, "GET", proxyHTTPURL.String(), "/test_plugin_reference?key=thirtytangas", nil)
+		resp, err := helpers.DefaultHTTPClientWithProxy(proxyHTTPURL).Do(req)
+		if err != nil {
+			return false
+		}
+		defer resp.Body.Close()
+		return resp.StatusCode == http.StatusTeapot
+	}, ingressWait, waitTick)
+
+	t.Logf("creating a ReferenceGrant that permits kongconsumer access from %s to kongplugins in %s", remote.Name, ns.Name)
+	grant := &gatewayapi.ReferenceGrant{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: uuid.NewString(),
+		},
+		Spec: gatewayapi.ReferenceGrantSpec{
+			From: []gatewayapi.ReferenceGrantFrom{
+				{
+					Group:     gatewayapi.Group("configuration.konghq.com"),
+					Kind:      gatewayapi.Kind("KongConsumer"),
+					Namespace: gatewayapi.Namespace(remote.Name),
+				},
+			},
+			To: []gatewayapi.ReferenceGrantTo{
+				{
+					Group: gatewayapi.Group("configuration.konghq.com"),
+					Kind:  gatewayapi.Kind("KongPlugin"),
+				},
+			},
+		},
+	}
+	_, err = gatewayClient.GatewayV1beta1().ReferenceGrants(ns.Name).Create(ctx, grant, metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	t.Logf("validating that plugin %s was successfully configured", kongplugin.Name)
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		req := helpers.MustHTTPRequest(t, "GET", proxyHTTPURL.String(), "/test_plugin_reference?apikey=thirtytangas", nil)
+		resp, err := helpers.DefaultHTTPClientWithProxy(proxyHTTPURL).Do(req)
+		if !assert.NoError(c, err) {
+			return
+		}
+		defer resp.Body.Close()
+		assert.True(c, resp.StatusCode == http.StatusTeapot)
+	}, ingressWait, waitTick)
+}
diff --git a/test/kongintegration/inmemory_update_strategy_test.go b/test/kongintegration/inmemory_update_strategy_test.go
index e64d427cb..e58f5c9a4 100644
--- a/test/kongintegration/inmemory_update_strategy_test.go
+++ b/test/kongintegration/inmemory_update_strategy_test.go
@@ -3,6 +3,7 @@ package kongintegration
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"testing"
 	"time"
@@ -14,8 +15,11 @@ import (
 	"github.com/samber/lo"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/adminapi"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/sendconfig"
 	"github.com/kong/kubernetes-ingress-controller/v3/test/kongintegration/containers"
 )
@@ -73,53 +77,62 @@ func TestUpdateStrategyInMemory_PropagatesResourcesErrors(t *testing.T) {
 			},
 		},
 	}
-	expectedResourceError := sendconfig.ResourceError{
-		Name:       "test-service",
-		Namespace:  "default",
-		Kind:       "Service",
-		UID:        "a3b8afcc-9f19-42e4-aa8f-5866168c2ad3",
-		APIVersion: "v1",
-		Problems: map[string]string{
-			"service:test-service": "failed conditional validation given value of field 'protocol'",
-			"path":                 "value must be null",
+	expectedCausingObjects := []client.Object{
+		&metav1.PartialObjectMetadata{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "Service",
+				APIVersion: "v1",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "default",
+				Name:      "test-service",
+				UID:       "a3b8afcc-9f19-42e4-aa8f-5866168c2ad3",
+			},
 		},
 	}
+	expectedMessage := "invalid service:test-service: failed conditional validation given value of field 'protocol'"
 	expectedRawErrBody := []byte(`{"code":14,"name":"invalid declarative configuration","fields":{},"message":"declarative config is invalid: {}","flattened_errors":[{"entity_type":"service","entity_name":"test-service","entity_tags":["k8s-name:test-service","k8s-namespace:default","k8s-kind:Service","k8s-uid:a3b8afcc-9f19-42e4-aa8f-5866168c2ad3","k8s-group:","k8s-version:v1"],"errors":[{"type":"field","message":"value must be null","field":"path"},{"type":"entity","message":"failed conditional validation given value of field 'protocol'"}],"entity":{"path":"/test","name":"test-service","protocol":"grpc","tags":["k8s-name:test-service","k8s-namespace:default","k8s-kind:Service","k8s-uid:a3b8afcc-9f19-42e4-aa8f-5866168c2ad3","k8s-group:","k8s-version:v1"],"host":"konghq.com","port":80}}]}`)
 	expectedBody := map[string]interface{}{}
 	require.NoError(t, json.Unmarshal(expectedRawErrBody, &expectedBody))
 
 	require.Eventually(t, func() bool {
-		err, resourceErrors, rawErrBody, parseErr := sut.Update(ctx, faultyConfig)
+		err := sut.Update(ctx, faultyConfig)
 		if err == nil {
 			t.Logf("expected error: %v", err)
 			return false
 		}
-		if len(resourceErrors) == 0 {
-			t.Log("expected resource errors")
+		var updateError sendconfig.UpdateError
+		if !errors.As(err, &updateError) {
+			t.Logf("expected UpdateError, got: %T", err)
 			return false
 		}
-		if len(rawErrBody) == 0 {
-			t.Log("expected error response body")
+		if len(updateError.ResourceFailures()) == 0 {
+			t.Log("expected resource errors")
 			return false
 		}
-		if parseErr != nil {
-			t.Logf("expected no parse error: %v", parseErr)
+		if len(updateError.RawResponseBody()) == 0 {
+			t.Log("expected error response body")
 			return false
 		}
-
-		resourceErr, found := lo.Find(resourceErrors, func(r sendconfig.ResourceError) bool {
-			return r.Name == "test-service"
+		resourceErr, found := lo.Find(updateError.ResourceFailures(), func(r failures.ResourceFailure) bool {
+			return lo.ContainsBy(r.CausingObjects(), func(obj client.Object) bool {
+				return obj.GetName() == "test-service"
+			})
 		})
 		if !found {
-			t.Logf("expected resource error for test-service, got: %+v", resourceErrors)
+			t.Logf("expected resource error for test-service, got: %+v", updateError.ResourceFailures())
+			return false
+		}
+		if resourceErr.Message() != expectedMessage {
+			t.Logf("expected resource error message to be %q, got %q", expectedMessage, resourceErr.Message())
 			return false
 		}
-		if diff := cmp.Diff(expectedResourceError, resourceErr); diff != "" {
+		if diff := cmp.Diff(expectedCausingObjects, resourceErr.CausingObjects()); diff != "" {
 			t.Logf("expected resource error to match, got diff: %s", diff)
 			return false
 		}
 		actualBody := map[string]interface{}{}
-		if err := json.Unmarshal(rawErrBody, &actualBody); err != nil {
+		if err := json.Unmarshal(updateError.RawResponseBody(), &actualBody); err != nil {
 			t.Logf("could not unmarshal error body: %s", err)
 			return false
 		}
diff --git a/test/kongintegration/kongupstreampolicy_test.go b/test/kongintegration/kongupstreampolicy_test.go
index d5a7863e7..153c732cf 100644
--- a/test/kongintegration/kongupstreampolicy_test.go
+++ b/test/kongintegration/kongupstreampolicy_test.go
@@ -239,8 +239,7 @@ func TestKongUpstreamPolicyTranslation(t *testing.T) {
 
 			// Update Kong with the Upstream.
 			require.Eventually(t, func() bool {
-				err, _, _, _ = updateStrategy.Update(ctx, content)
-				if err != nil {
+				if err := updateStrategy.Update(ctx, content); err != nil {
 					t.Logf("error updating Kong configuration: %v", err)
 					return false
 				}
diff --git a/test/kongintegration/translator_golden_tests_outputs_test.go b/test/kongintegration/translator_golden_tests_outputs_test.go
index d56064279..27331724d 100644
--- a/test/kongintegration/translator_golden_tests_outputs_test.go
+++ b/test/kongintegration/translator_golden_tests_outputs_test.go
@@ -127,20 +127,10 @@ func ensureGoldenTestOutputIsAccepted(
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
-		err, resourceErrors, rawErrBody, parseErr := sut.Update(ctx, sendconfig.ContentWithHash{Content: content})
-		if err != nil {
+		if err := sut.Update(ctx, sendconfig.ContentWithHash{Content: content}); err != nil {
 			t.Logf("error: %v", err)
 			return false
 		}
-		if len(resourceErrors) > 0 {
-			t.Logf("resource errors: %v", resourceErrors)
-			return false
-		}
-		if parseErr != nil {
-			t.Logf("parse error: %v", parseErr)
-			t.Logf("raw error: %v", rawErrBody)
-			return false
-		}
 		return true
 	}, timeout, tick)
 }