Repository: Kong/kubernetes-ingress-controller
Pull request #4989 (2023-10-26 15:35:54)

feat: override Upstream config with KongUpstreamPolicy
**What this PR does / why we need it**:

Makes `Parser` use `KongUpstreamPolicies` to override `kong.Upstreams` config. Also:
- drops `HostHeader` from the CRD as it's already covered with the `konghq.com/host-header` annotation
- adds one more `spec.hashOnFallback must not be set when spec.hashOn.cookie is set` validation

<!-- Please describe why this particular PR is necessary or why you see it as a nice addition -->

**Which issue this PR fixes**:

Closes https://github.com/Kong/kubernetes-ingress-controller/issues/4931.



codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`1351004`)](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/commit/13510041ea8fa4b9ec5701249a5737008d53859b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) 77.6% compared to head [(`1c3b24f`)](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) 77.8%.

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##            main   #4989     +/-   ##
=======================================
+ Coverage   77.6%   77.8%   +0.2%     
=======================================
  Files        167     167             
  Lines      18709   18784     +75     
=======================================
+ Hits       14527   14631    +104     
+ Misses      3345    3327     -18     
+ Partials     837     826     -11     
```


| [Files](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) | Coverage Δ | |
|---|---|---|
| [internal/annotations/annotations.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvYW5ub3RhdGlvbnMvYW5ub3RhdGlvbnMuZ28=) | `97.6% <100.0%> (+<0.1%)` | :arrow_up: |
| [internal/dataplane/kongstate/kongstate.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvZGF0YXBsYW5lL2tvbmdzdGF0ZS9rb25nc3RhdGUuZ28=) | `84.7% <100.0%> (+1.7%)` | :arrow_up: |
| [internal/dataplane/kongstate/kongupstreampolicy.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvZGF0YXBsYW5lL2tvbmdzdGF0ZS9rb25ndXBzdHJlYW1wb2xpY3kuZ28=) | `96.4% <100.0%> (ø)` | |
| [internal/dataplane/kongstate/upstream.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvZGF0YXBsYW5lL2tvbmdzdGF0ZS91cHN0cmVhbS5nbw==) | `78.2% <100.0%> (+14.7%)` | :arrow_up: |
| [internal/dataplane/kongstate/util.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvZGF0YXBsYW5lL2tvbmdzdGF0ZS91dGlsLmdv) | `90.2% <100.0%> (+1.2%)` | :arrow_up: |
| [internal/dataplane/parser/parser.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvZGF0YXBsYW5lL3BhcnNlci9wYXJzZXIuZ28=) | `92.0% <100.0%> (+0.7%)` | :arrow_up: |
| [internal/store/store.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvc3RvcmUvc3RvcmUuZ28=) | `70.4% <100.0%> (+0.1%)` | :arrow_up: |

... and [6 files with indirect coverage changes](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/4989?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong).


Patch:
diff --git a/.golangci.yaml b/.golangci.yaml
index 4abeddcb7..3f635b947 100644
--- a/.golangci.yaml
+++ b/.golangci.yaml
@@ -169,3 +169,8 @@ issues:
       - forbidigo
     text: "use of `(gatewayv1alpha2|gatewayv1beta1)"
     path: (internal/gatewayapi/aliases.go|pkg/apis/.*/.*\.go)
+
+  - linters:
+      - revive
+    text: "var-declaration: should drop = 0 from declaration of var .*; it is the zero value"
+    path: \.go
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e51e2cfbb..8c3a41bde 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -149,6 +149,15 @@ Adding a new version? You'll need three changes:
   [#4928](https://github.com/Kong/kubernetes-ingress-controller/pull/4928)
 - Fixed a panic when receiving broken configuration from Kong Gateway.
   [#5003](https://github.com/Kong/kubernetes-ingress-controller/pull/5003)
+- Use 46 bits in values of priorities of generated Kong routes when expression
+  rotuer is enabled to limit the priorities to be less than `1e14`. This
+  prevents them to be encoded into scientific notation when dumping 
+  configurations from admin API that brings precision loss and type 
+  inconsistency in decoding JSON/YAML data to `uint64`. 
+  This change will limit number of `HTTPRoute`s that can be 
+  deterministically sorted by their creation timestamps, names and internal
+  rule orders to `2^12=4096` and number of `GRPCRoutes` can be sorted to `2^8=256`.
+  [#5024](https://github.com/Kong/kubernetes-ingress-controller/pull/5024)
 
 ### Changed
 
@@ -212,6 +221,9 @@ Adding a new version? You'll need three changes:
   [#4969](https://github.com/Kong/kubernetes-ingress-controller/pull/4969)
   [#4979](https://github.com/Kong/kubernetes-ingress-controller/pull/4979)
   [#4989](https://github.com/Kong/kubernetes-ingress-controller/pull/4989)
+- KIC now specifies its `UserAgent` when communicating with kube-apiserver
+  as `kong-ingress-controller/${VERSION}` where version is the version of KIC.
+  [#5019](https://github.com/Kong/kubernetes-ingress-controller/pull/5019)
 
 [KIC Annotations reference]: https://docs.konghq.com/kubernetes-ingress-controller/latest/references/annotations/
 [KIC CRDs reference]: https://docs.konghq.com/kubernetes-ingress-controller/latest/references/custom-resources/
diff --git a/config/crd/bases/configuration.konghq.com_kongupstreampolicies.yaml b/config/crd/bases/configuration.konghq.com_kongupstreampolicies.yaml
index 3f02b1db8..d75c1882d 100644
--- a/config/crd/bases/configuration.konghq.com_kongupstreampolicies.yaml
+++ b/config/crd/bases/configuration.konghq.com_kongupstreampolicies.yaml
@@ -323,10 +323,6 @@ spec:
                       to be considered healthy.
                     type: integer
                 type: object
-              hostHeader:
-                description: HostHeader is the hostname to be used as Host header
-                  when proxying requests through Kong.
-                type: string
               slots:
                 description: Slots is the number of slots in the load balancer algorithm.
                   If not set, the default value in Kong for the algorithm is used.
@@ -341,18 +337,26 @@ spec:
           rule: 'has(self.spec.hashOn) ? [has(self.spec.hashOn.input), has(self.spec.hashOn.cookie),
             has(self.spec.hashOn.header), has(self.spec.hashOn.uriCapture), has(self.spec.hashOn.queryArg)].filter(fieldSet,
             fieldSet == true).size() <= 1 : true'
-        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
-            can be set.
-          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
-            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
-            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
-            <= 1 : true'
         - message: When spec.hashOn.cookie is set, spec.hashOn.cookiePath is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? has(self.spec.hashOn.cookiePath)
             : true'
         - message: When spec.hashOn.cookiePath is set, spec.hashOn.cookie is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookiePath) ? has(self.spec.hashOn.cookie)
             : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
+            is set.
+          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
+        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
+            can be set.
+          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
+            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
+            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
+            <= 1 : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
+            is set.
+          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
         - message: spec.hashOnFallback.cookie must not be set.
           rule: 'has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookie)
             : true'
@@ -367,14 +371,10 @@ spec:
           rule: 'has(self.spec.healthchecks) && has(self.spec.healthchecks.passive)
             && has(self.spec.healthchecks.passive.unhealthy) ? !has(self.spec.healthchecks.passive.unhealthy.interval)
             : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
-            is set.
-          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
-            is set.
-          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
+        - message: spec.hashOnFallback must not be set when spec.hashOn.cookie is
+            set.
+          rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? !has(self.spec.hashOnFallback)
+            : true'
     served: true
     storage: true
     subresources:
diff --git a/docs/api-reference.md b/docs/api-reference.md
index ac2e57766..168075eaf 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -617,7 +617,6 @@ KongUpstreamPolicySpec contains the specification for KongUpstreamPolicy.
 | `hashOn` _[KongUpstreamHash](#kongupstreamhash)_ | HashOn defines how to calculate hash for consistent-hashing load balancing algorithm. Algorithm must be set to "consistent-hashing" for this field to have effect. |
 | `hashOnFallback` _[KongUpstreamHash](#kongupstreamhash)_ | HashOnFallback defines how to calculate hash for consistent-hashing load balancing algorithm if the primary hash function fails. Algorithm must be set to "consistent-hashing" for this field to have effect. |
 | `healthchecks` _[KongUpstreamHealthcheck](#kongupstreamhealthcheck)_ | Healthchecks defines the health check configurations in Kong. |
-| `hostHeader` _string_ | HostHeader is the hostname to be used as Host header when proxying requests through Kong. |
 
 
 _Appears in:_
diff --git a/examples/kong-upstream-policy-service.yaml b/examples/kong-upstream-policy-ingress.yaml
similarity index 100%
rename from examples/kong-upstream-policy-service.yaml
rename to examples/kong-upstream-policy-ingress.yaml
diff --git a/go.mod b/go.mod
index daf8c8292..17b9713a4 100644
--- a/go.mod
+++ b/go.mod
@@ -30,7 +30,7 @@ require (
 	github.com/google/uuid v1.4.0
 	github.com/jpillora/backoff v1.0.0
 	github.com/kong/deck v1.27.1
-	github.com/kong/go-kong v0.47.0
+	github.com/kong/go-kong v0.48.0
 	github.com/kong/kubernetes-telemetry v0.1.1
 	github.com/kong/kubernetes-testing-framework v0.41.1
 	github.com/lithammer/dedent v1.1.0
@@ -111,7 +111,7 @@ require (
 	github.com/chai2010/gettext-go v1.0.2 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/docker/distribution v2.8.2+incompatible // indirect
-	github.com/docker/docker v24.0.6+incompatible // indirect
+	github.com/docker/docker v24.0.7+incompatible // indirect
 	github.com/docker/go-connections v0.4.0
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
@@ -178,7 +178,7 @@ require (
 	github.com/sourcegraph/conc v0.3.0
 	github.com/spf13/cast v1.5.1 // indirect
 	github.com/ssgelm/cookiejarparser v1.0.1 // indirect
-	github.com/tidwall/gjson v1.16.0 // indirect
+	github.com/tidwall/gjson v1.17.0 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
diff --git a/go.sum b/go.sum
index 5e4b3fa79..d1b7ec6f9 100644
--- a/go.sum
+++ b/go.sum
@@ -86,8 +86,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/docker/distribution v2.8.2+incompatible h1:T3de5rq0dB1j30rp0sA2rER+m322EBzniBPB6ZIzuh8=
 github.com/docker/distribution v2.8.2+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
-github.com/docker/docker v24.0.6+incompatible h1:hceabKCtUgDqPu+qm0NgsaXf28Ljf4/pWFL7xjWWDgE=
-github.com/docker/docker v24.0.6+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v24.0.7+incompatible h1:Wo6l37AuwP3JaMnZa226lzVXGA3F9Ig1seQen0cKYlM=
+github.com/docker/docker v24.0.7+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ=
 github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
@@ -257,8 +257,8 @@ github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGC
 github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/kong/deck v1.27.1 h1:9R/pZpwOTYz80MHz5naoN0B/k+T+wBfCtGo1nliQQ+c=
 github.com/kong/deck v1.27.1/go.mod h1:6FvrNHuCDf0I9PkUd5sPrK9EnfG4aQLhs0vPmdDRylE=
-github.com/kong/go-kong v0.47.0 h1:fuZjRnUChITogljM4mY4HGGTKTaZM50970Wwg6EvQgs=
-github.com/kong/go-kong v0.47.0/go.mod h1:QaLx1EsT+roe5V6h7aKghwLn0bXg7l/Bg8wzlcCmygc=
+github.com/kong/go-kong v0.48.0 h1:vK1OpoxO50qlKdwPfmx9ChvkTKRsoCCB3b3iHo1umLc=
+github.com/kong/go-kong v0.48.0/go.mod h1:qH4CEFqT83ywmu1TlMZX09clQH4B8/dX88CtT/jdv/E=
 github.com/kong/kubernetes-telemetry v0.1.1 h1:q8iABCuonO16p5q6QVygr7J39G5o5jK5Y3w3ZlE8JC4=
 github.com/kong/kubernetes-telemetry v0.1.1/go.mod h1:0yzRZVPwKeOQUFMWbROEJzUE1z8WWngWiujoYtdEEPA=
 github.com/kong/kubernetes-testing-framework v0.41.1 h1:rYPOUZu7961lNApLkr4QaqsKaJBuY9HClCa0r7K4P+8=
@@ -435,8 +435,8 @@ github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXl
 github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww=
 github.com/testcontainers/testcontainers-go v0.26.0 h1:uqcYdoOHBy1ca7gKODfBd9uTHVK3a7UL848z09MVZ0c=
 github.com/testcontainers/testcontainers-go v0.26.0/go.mod h1:ICriE9bLX5CLxL9OFQ2N+2N+f+803LNJ1utJb1+Inx0=
-github.com/tidwall/gjson v1.16.0 h1:SyXa+dsSPpUlcwEDuKuEBJEz5vzTvOea+9rjyYodQFg=
-github.com/tidwall/gjson v1.16.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/gjson v1.17.0 h1:/Jocvlh98kcTfpN2+JzGQWQcqrPQwDrVEMApx/M5ZwM=
+github.com/tidwall/gjson v1.17.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
diff --git a/internal/annotations/annotations.go b/internal/annotations/annotations.go
index 716543608..37795eddf 100644
--- a/internal/annotations/annotations.go
+++ b/internal/annotations/annotations.go
@@ -22,6 +22,8 @@ import (
 
 	netv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 )
 
 type ClassMatching int
@@ -381,3 +383,8 @@ func ExtractRewriteURI(anns map[string]string) (string, bool) {
 	s, ok := anns[AnnotationPrefix+RewriteURIKey]
 	return s, ok
 }
+
+func ExtractUpstreamPolicy(anns map[string]string) (string, bool) {
+	s, ok := anns[kongv1beta1.KongUpstreamPolicyAnnotationKey]
+	return s, ok
+}
diff --git a/internal/dataplane/kongstate/kongstate.go b/internal/dataplane/kongstate/kongstate.go
index db1013ae9..712e367ec 100644
--- a/internal/dataplane/kongstate/kongstate.go
+++ b/internal/dataplane/kongstate/kongstate.go
@@ -8,7 +8,10 @@ import (
 
 	"github.com/go-logr/logr"
 	"github.com/kong/go-kong/kong"
+	"github.com/samber/lo"
+	corev1 "k8s.io/api/core/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"
@@ -185,7 +188,11 @@ func (ks *KongState) FillConsumerGroups(_ logr.Logger, s store.Storer) {
 	}
 }
 
-func (ks *KongState) FillOverrides(logger logr.Logger, s store.Storer) {
+func (ks *KongState) FillOverrides(
+	logger logr.Logger,
+	s store.Storer,
+	failuresCollector *failures.ResourceFailuresCollector,
+) {
 	for i := 0; i < len(ks.Services); i++ {
 		// Services
 		ks.Services[i].override()
@@ -196,17 +203,30 @@ func (ks *KongState) FillOverrides(logger logr.Logger, s store.Storer) {
 		}
 	}
 
-	// Upstreams
+	ks.FillUpstreamOverrides(s, failuresCollector)
+}
+
+func (ks *KongState) FillUpstreamOverrides(s store.Storer, failuresCollector *failures.ResourceFailuresCollector) {
 	for i := 0; i < len(ks.Upstreams); i++ {
-		kongIngress, err := getKongIngressForServices(s, ks.Upstreams[i].Service.K8sServices)
+		servicesGroup := lo.Values(ks.Upstreams[i].Service.K8sServices)
+		servicesAsObjects := func(svc *corev1.Service, _ int) client.Object { return svc }
+
+		kongIngress, err := getKongIngressForServices(s, servicesGroup)
 		if err != nil {
-			logger.Error(err, "failed to fetch KongIngress resource for Services",
-				"names", PrettyPrintServiceList(ks.Upstreams[i].Service.K8sServices))
-			continue
+			failuresCollector.PushResourceFailure(err.Error(), lo.Map(servicesGroup, servicesAsObjects)...)
+		} else {
+			for _, svc := range ks.Upstreams[i].Service.K8sServices {
+				ks.Upstreams[i].override(kongIngress, svc)
+			}
 		}
 
-		for _, svc := range ks.Upstreams[i].Service.K8sServices {
-			ks.Upstreams[i].override(kongIngress, svc)
+		kongUpstreamPolicy, err := GetKongUpstreamPolicyForServices(s, servicesGroup)
+		if err != nil {
+			failuresCollector.PushResourceFailure(err.Error(), lo.Map(servicesGroup, servicesAsObjects)...)
+		} else {
+			if kongUpstreamPolicy != nil {
+				ks.Upstreams[i].overrideByKongUpstreamPolicy(kongUpstreamPolicy)
+			}
 		}
 	}
 }
diff --git a/internal/dataplane/parser/translators/kongupstreampolicy.go b/internal/dataplane/kongstate/kongupstreampolicy.go
similarity index 72%
rename from internal/dataplane/parser/translators/kongupstreampolicy.go
rename to internal/dataplane/kongstate/kongupstreampolicy.go
index f83313260..fb28f84d4 100644
--- a/internal/dataplane/parser/translators/kongupstreampolicy.go
+++ b/internal/dataplane/kongstate/kongupstreampolicy.go
@@ -1,9 +1,15 @@
-package translators
+package kongstate
 
 import (
+	"fmt"
+
 	"github.com/kong/go-kong/kong"
 	"github.com/samber/lo"
+	"github.com/samber/mo"
+	corev1 "k8s.io/api/core/v1"
 
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
 	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 )
 
@@ -14,6 +20,49 @@ const (
 	KongHashOnTypeURICapture string = "uri_capture"
 )
 
+// GetKongUpstreamPolicyForServices scans all Services in the group to see if their KongUpstreamPolicy is consistent
+// and returns a non-nil KongUpstreamPolicy if it is.
+//
+// We require either:
+// - all the Services to be configured with the same KongUpstreamPolicy.
+// - none of the Services to be configured with a KongUpstreamPolicy.
+//
+// If the KongUpstreamPolicy configuration is inconsistent or a configured KongUpstreamPolicy cannot be fetched from
+// the store, an error is returned.
+func GetKongUpstreamPolicyForServices(s store.Storer, servicesGroup []*corev1.Service) (*kongv1beta1.KongUpstreamPolicy, error) {
+	if len(servicesGroup) == 0 {
+		return nil, nil
+	}
+
+	servicesGroupedByUpstreamPolicy := lo.GroupBy(servicesGroup, func(svc *corev1.Service) mo.Option[string] {
+		policyName, ok := annotations.ExtractUpstreamPolicy(svc.Annotations)
+		if !ok {
+			return mo.None[string]()
+		}
+		return mo.Some(policyName)
+	})
+
+	// If there's more than one group, then there are services with different KongUpstreamPolicy configurations.
+	if len(servicesGroupedByUpstreamPolicy) > 1 {
+		return nil, fmt.Errorf("inconsistent KongUpstreamPolicy configuration for services %s",
+			prettyPrintServiceList(servicesGroup))
+	}
+
+	// If there's one group (must be at least one, since we checked len(servicesGroup) == 0 above), then
+	// there's either one KongUpstreamPolicy for all services, or none.
+	upstreamPolicyName, ok := lo.Keys(servicesGroupedByUpstreamPolicy)[0].Get()
+	if !ok {
+		return nil, nil
+	}
+
+	policy, err := s.GetKongUpstreamPolicy(servicesGroup[0].Namespace, upstreamPolicyName)
+	if err != nil {
+		return nil, fmt.Errorf("failed fetching KongUpstreamPolicy: %w", err)
+	}
+
+	return policy, nil
+}
+
 // TranslateKongUpstreamPolicy translates KongUpstreamPolicySpec to kong.Upstream. It makes assumption that
 // KongUpstreamPolicySpec has been validated on the API level.
 func TranslateKongUpstreamPolicy(policy kongv1beta1.KongUpstreamPolicySpec) *kong.Upstream {
@@ -21,7 +70,6 @@ func TranslateKongUpstreamPolicy(policy kongv1beta1.KongUpstreamPolicySpec) *kon
 		Algorithm:    policy.Algorithm,
 		Slots:        policy.Slots,
 		Healthchecks: translateHealthchecks(policy.Healthchecks),
-		HostHeader:   policy.HostHeader,
 
 		HashOn:           translateHashOn(policy.HashOn),
 		HashOnHeader:     translateHashOnHeader(policy.HashOn),
diff --git a/internal/dataplane/kongstate/upstream.go b/internal/dataplane/kongstate/upstream.go
index 48a21c541..571f216a9 100644
--- a/internal/dataplane/kongstate/upstream.go
+++ b/internal/dataplane/kongstate/upstream.go
@@ -7,6 +7,7 @@ import (
 
 	"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"
 )
 
 // Upstream is a wrapper around Upstream object in Kong.
@@ -92,6 +93,53 @@ func (u *Upstream) overrideByKongIngress(kongIngress *kongv1.KongIngress) {
 	}
 }
 
+func (u *Upstream) overrideByKongUpstreamPolicy(policy *kongv1beta1.KongUpstreamPolicy) {
+	if u == nil {
+		return
+	}
+
+	kongUpstreamOverrides := TranslateKongUpstreamPolicy(policy.Spec)
+	if kongUpstreamOverrides.Algorithm != nil {
+		u.Algorithm = kongUpstreamOverrides.Algorithm
+	}
+	if kongUpstreamOverrides.Slots != nil {
+		u.Slots = kongUpstreamOverrides.Slots
+	}
+	if kongUpstreamOverrides.Healthchecks != nil {
+		u.Healthchecks = kongUpstreamOverrides.Healthchecks
+	}
+	if kongUpstreamOverrides.HashOn != nil {
+		u.HashOn = kongUpstreamOverrides.HashOn
+	}
+	if kongUpstreamOverrides.HashFallback != nil {
+		u.HashFallback = kongUpstreamOverrides.HashFallback
+	}
+	if kongUpstreamOverrides.HashOnHeader != nil {
+		u.HashOnHeader = kongUpstreamOverrides.HashOnHeader
+	}
+	if kongUpstreamOverrides.HashFallbackHeader != nil {
+		u.HashFallbackHeader = kongUpstreamOverrides.HashFallbackHeader
+	}
+	if kongUpstreamOverrides.HashOnCookie != nil {
+		u.HashOnCookie = kongUpstreamOverrides.HashOnCookie
+	}
+	if kongUpstreamOverrides.HashOnCookiePath != nil {
+		u.HashOnCookiePath = kongUpstreamOverrides.HashOnCookiePath
+	}
+	if kongUpstreamOverrides.HashOnQueryArg != nil {
+		u.HashOnQueryArg = kongUpstreamOverrides.HashOnQueryArg
+	}
+	if kongUpstreamOverrides.HashFallbackQueryArg != nil {
+		u.HashFallbackQueryArg = kongUpstreamOverrides.HashFallbackQueryArg
+	}
+	if kongUpstreamOverrides.HashOnURICapture != nil {
+		u.HashOnURICapture = kongUpstreamOverrides.HashOnURICapture
+	}
+	if kongUpstreamOverrides.HashFallbackURICapture != nil {
+		u.HashFallbackURICapture = kongUpstreamOverrides.HashFallbackURICapture
+	}
+}
+
 // override sets Upstream fields by KongIngress first, then by k8s Service's annotations.
 func (u *Upstream) override(
 	kongIngress *kongv1.KongIngress,
diff --git a/internal/dataplane/kongstate/util.go b/internal/dataplane/kongstate/util.go
index 24176767a..a60762c01 100644
--- a/internal/dataplane/kongstate/util.go
+++ b/internal/dataplane/kongstate/util.go
@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/kong/go-kong/kong"
 	corev1 "k8s.io/api/core/v1"
@@ -18,7 +19,7 @@ import (
 
 func getKongIngressForServices(
 	s store.Storer,
-	services map[string]*corev1.Service,
+	services []*corev1.Service,
 ) (*kongv1.KongIngress, error) {
 	// loop through each service and retrieve the attached KongIngress resources.
 	// there can only be one KongIngress for a group of services: either one of
@@ -35,7 +36,7 @@ func getKongIngressForServices(
 		// retrieve the attached KongIngress for the service
 		kongIngress, err := s.GetKongIngress(svc.Namespace, confName)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to get KongIngress: %w", err)
 		}
 
 		// we found the KongIngress for these services. We don't have to check any
@@ -257,21 +258,14 @@ func SecretToConfiguration(
 	return config, nil
 }
 
-// PrettyPrintServiceList makes a clean printable list of a map of Kubernetes
-// services for the purpose of logging (errors, info, e.t.c.).
-func PrettyPrintServiceList(services map[string]*corev1.Service) string {
-	var serviceList string
-	first := true
+// prettyPrintServiceList makes a clean printable list of Kubernetes
+// services for the purpose of logging (errors, info, etc.).
+func prettyPrintServiceList(services []*corev1.Service) string {
+	serviceList := make([]string, 0, len(services))
 	for _, svc := range services {
-		if !first {
-			serviceList = serviceList + ", "
-		}
-		serviceList = serviceList + svc.Namespace + "/" + svc.Name
-		if first {
-			first = false
-		}
+		serviceList = append(serviceList, svc.Namespace+"/"+svc.Name)
 	}
-	return serviceList
+	return strings.Join(serviceList, ", ")
 }
 
 // plugin is a intermediate type to hold plugin related configuration.
diff --git a/internal/dataplane/parser/atc/route.go b/internal/dataplane/parser/atc/route.go
index 1da66655e..b045f998b 100644
--- a/internal/dataplane/parser/atc/route.go
+++ b/internal/dataplane/parser/atc/route.go
@@ -5,7 +5,7 @@ import (
 )
 
 // ApplyExpression sets a Matcher as a Kong route's expression and assigns the route the given priority.
-func ApplyExpression(r *kong.Route, m Matcher, priority int) {
+func ApplyExpression(r *kong.Route, m Matcher, priority uint64) {
 	r.Expression = kong.String(m.Expression())
-	r.Priority = kong.Int(priority)
+	r.Priority = kong.Uint64(priority)
 }
diff --git a/internal/dataplane/parser/parser.go b/internal/dataplane/parser/parser.go
index c155d1aa1..1a11bb2b9 100644
--- a/internal/dataplane/parser/parser.go
+++ b/internal/dataplane/parser/parser.go
@@ -186,7 +186,7 @@ func (p *Parser) BuildKongConfig() KongConfigBuildingResult {
 	}
 
 	// merge KongIngress with Routes, Services and Upstream
-	result.FillOverrides(p.logger, p.storer)
+	result.FillOverrides(p.logger, p.storer, p.failuresCollector)
 
 	// generate consumers and credentials
 	result.FillConsumersAndCredentials(p.logger, p.storer, p.failuresCollector)
diff --git a/internal/dataplane/parser/translators/atc_utils.go b/internal/dataplane/parser/translators/atc_utils.go
index e2f53eaf4..450ac4a87 100644
--- a/internal/dataplane/parser/translators/atc_utils.go
+++ b/internal/dataplane/parser/translators/atc_utils.go
@@ -7,12 +7,12 @@ import (
 )
 
 const (
-	// FromResourceKindPriorityShiftBits is the highest 2 bits 51-50 used in priority field of Kong route
+	// FromResourceKindPriorityShiftBits is the highest 2 bits 45-44 used in priority field of Kong route
 	// to note the kind of the resource from which the route is translated.
 	// 11 - routes from Ingress.
 	// 10 - routes from HTTPRoute.
 	// 01 - routes from GRPCRoute.
-	FromResourceKindPriorityShiftBits = 50
+	FromResourceKindPriorityShiftBits = 44
 	// ResourceKindBitsIngress is the value of highest 2 bits for routes from ingresses.
 	ResourceKindBitsIngress = 3
 	// ResourceKindBitsHTTPRoute is the value of highest 2 bits for routes from HTTPRoutes.
@@ -21,6 +21,11 @@ const (
 	ResourceKindBitsGRPCRoute = 1
 )
 
+type (
+	// RoutePriorityType is the type of priority field of Kong routes.
+	RoutePriorityType = uint64
+)
+
 const (
 	// CatchAllHTTPExpression is the expression to match all HTTP/HTTPS requests.
 	// For rules with no matches and no hostnames in its parent HTTPRoute or GRPCRoute,
diff --git a/internal/dataplane/parser/translators/grpcroute_atc.go b/internal/dataplane/parser/translators/grpcroute_atc.go
index 48545bb92..52f177455 100644
--- a/internal/dataplane/parser/translators/grpcroute_atc.go
+++ b/internal/dataplane/parser/translators/grpcroute_atc.go
@@ -285,11 +285,11 @@ func CalculateGRCPRouteMatchPriorityTraits(match SplitGRPCRouteMatch) GRPCRouteP
 
 // EncodeToPriority turns GRPCRoute priority traits into the integer expressed priority.
 //
-//					   4                   3                   2                   1
-//	 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0
-//	+-+---------------+---------------------+---------------------+---------+---------------------------+
-//	|P| host len      | GRPC service length | GRPC method length  |Header No| relative order            |
-//	+-+---------------+---------------------+---------------------+---------+---------------------------+
+//		   4                   3                   2                   1
+//	 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0
+//	+-+---------------+---------------------+---------------------+---------+----------------+
+//	|P| host len      | GRPC service length | GRPC method length  |Header No| relative order |
+//	+-+---------------+---------------------+---------------------+---------+----------------+
 //
 // Where:
 // P: set to 1 if the hostname is non-wildcard.
@@ -301,32 +301,32 @@ func CalculateGRCPRouteMatchPriorityTraits(match SplitGRPCRouteMatch) GRPCRouteP
 //
 // REVIEW: althogh not specified in official docs, do we need to assign a bit for GRPC method match type
 // to assign higher priority for method match with `Exact` match?
-func (t GRPCRoutePriorityTraits) EncodeToPriority() int {
+func (t GRPCRoutePriorityTraits) EncodeToPriority() RoutePriorityType {
 	const (
-		// PreciseHostnameShiftBits assigns bit 49 for marking if the hostname is non-wildcard.
-		PreciseHostnameShiftBits = 49
-		// HostnameLengthShiftBits assigns bits 41-48 for the length of hostname.
-		HostnameLengthShiftBits = 41
-		// ServiceLengthShiftBits assigns bits 30-40 for the length of `Service` in method match.
-		ServiceLengthShiftBits = 30
-		// MethodLengthShiftBits assigns bits 19-29 for the length of `Method` in method match.
-		MethodLengthShiftBits = 19
-		// HeaderCountShiftBits assigns bits 14-18 for the number of header matches.
-		HeaderCountShiftBits = 14
-		// bits 0-13 are used for relative order of creation timestamp, namespace/name, and internal order of rules and matches.
+		// PreciseHostnameShiftBits assigns bit 43 for marking if the hostname is non-wildcard.
+		PreciseHostnameShiftBits = 43
+		// HostnameLengthShiftBits assigns bits 35-42 for the length of hostname.
+		HostnameLengthShiftBits = 35
+		// ServiceLengthShiftBits assigns bits 24-34 for the length of `Service` in method match.
+		ServiceLengthShiftBits = 24
+		// MethodLengthShiftBits assigns bits 13-23 for the length of `Method` in method match.
+		MethodLengthShiftBits = 13
+		// HeaderCountShiftBits assigns bits 8-12 for the number of header matches.
+		HeaderCountShiftBits = 8
+		// bits 0-7 are used for relative order of creation timestamp, namespace/name, and internal order of rules and matches.
 		// the bits are calculated by sorting GRPCRoutes with the same priority calculated from the fields above
 		// and start from all 1s, then decrease by one for each GRPCRoute.
 	)
 
-	var priority int
+	var priority RoutePriorityType
 	priority += ResourceKindBitsGRPCRoute << FromResourceKindPriorityShiftBits
 	if t.PreciseHostname {
 		priority += (1 << PreciseHostnameShiftBits)
 	}
-	priority += t.HostnameLength << HostnameLengthShiftBits
-	priority += t.ServiceLength << ServiceLengthShiftBits
-	priority += t.MethodLength << MethodLengthShiftBits
-	priority += t.HeaderCount << HeaderCountShiftBits
+	priority += RoutePriorityType(t.HostnameLength) << HostnameLengthShiftBits
+	priority += RoutePriorityType(t.ServiceLength) << ServiceLengthShiftBits
+	priority += RoutePriorityType(t.MethodLength) << MethodLengthShiftBits
+	priority += RoutePriorityType(t.HeaderCount) << HeaderCountShiftBits
 
 	return priority
 }
@@ -343,7 +343,7 @@ type SplitGRPCRouteMatch struct {
 
 type SplitGRPCRouteMatchToPriority struct {
 	Match    SplitGRPCRouteMatch
-	Priority int
+	Priority uint64
 }
 
 // AssignRoutePriorityToSplitGRPCRouteMatches assigns priority to ALL split GRPCRoute matches
@@ -357,7 +357,7 @@ func AssignRoutePriorityToSplitGRPCRouteMatches(
 	logger logr.Logger,
 	splitGRPCouteMatches []SplitGRPCRouteMatch,
 ) []SplitGRPCRouteMatchToPriority {
-	priorityToSplitGRPCRouteMatches := map[int][]SplitGRPCRouteMatch{}
+	priorityToSplitGRPCRouteMatches := map[RoutePriorityType][]SplitGRPCRouteMatch{}
 	for _, match := range splitGRPCouteMatches {
 		priority := CalculateGRCPRouteMatchPriorityTraits(match).EncodeToPriority()
 		priorityToSplitGRPCRouteMatches[priority] = append(priorityToSplitGRPCRouteMatches[priority], match)
@@ -365,12 +365,12 @@ func AssignRoutePriorityToSplitGRPCRouteMatches(
 
 	splitGRPCRoutesToPriority := make([]SplitGRPCRouteMatchToPriority, 0, len(splitGRPCouteMatches))
 
-	// Bits 0-13 (14 bits) are assigned for relative order of GRPCRoutes.
+	// Bits 0-7 (8 bits) are assigned for relative order of GRPCRoutes.
 	// If multiple GRPCRoutes are assigned to the same priority in the previous step,
-	// sort them then starts with 2^14 -1 and decrease by one for each GRPCRoute;
+	// sort them then starts with 2^8 -1 and decrease by one for each GRPCRoute;
 	// If only one GRPCRoute occupies the priority, fill the relative order bits with all 1s.
-	const relativeOrderAssignedBits = 14
-	const defaultRelativeOrderPriorityBits = (1 << relativeOrderAssignedBits) - 1
+	const relativeOrderAssignedBits = 8
+	const defaultRelativeOrderPriorityBits RoutePriorityType = (1 << relativeOrderAssignedBits) - 1
 
 	for priority, matches := range priorityToSplitGRPCRouteMatches {
 		if len(matches) == 1 {
@@ -387,8 +387,8 @@ func AssignRoutePriorityToSplitGRPCRouteMatches(
 		})
 
 		for i, match := range matches {
-			relativeOrderBits := defaultRelativeOrderPriorityBits - i
-			// Although it is very unlikely that there are 2^14 = 16384 GRPCRoutes
+			relativeOrderBits := defaultRelativeOrderPriorityBits - RoutePriorityType(i)
+			// Although it is very unlikely that there are 2^8 = 256 GRPCRoutes
 			// should be given priority by their relative order, here we limit the
 			// relativeOrderBits to be at least 0.
 			if relativeOrderBits <= 0 {
@@ -400,9 +400,9 @@ func AssignRoutePriorityToSplitGRPCRouteMatches(
 			})
 		}
 
-		// Just in case, log a very unlikely scenario where we have more than 2^14 routes with the same base
+		// Just in case, log a very unlikely scenario where we have more than 2^8 routes with the same base
 		// priority and we have no bit space for them to be deterministically ordered.
-		if len(matches) > (1 << 14) {
+		if len(matches) > (1 << 8) {
 			logger.Error(nil, "Too many GRPCRoute matches to be deterministically ordered", "grpcroute_number", len(matches))
 		}
 	}
diff --git a/internal/dataplane/parser/translators/httproute_atc.go b/internal/dataplane/parser/translators/httproute_atc.go
index 5a3a5abc8..9e7343713 100644
--- a/internal/dataplane/parser/translators/httproute_atc.go
+++ b/internal/dataplane/parser/translators/httproute_atc.go
@@ -306,7 +306,7 @@ func SplitHTTPRoute(httproute *gatewayapi.HTTPRoute) []SplitHTTPRouteMatch {
 
 type SplitHTTPRouteMatchToKongRoutePriority struct {
 	Match    SplitHTTPRouteMatch
-	Priority int
+	Priority RoutePriorityType
 }
 
 type HTTPRoutePriorityTraits struct {
@@ -380,11 +380,11 @@ func CalculateHTTPRouteMatchPriorityTraits(match SplitHTTPRouteMatch) HTTPRouteP
 
 // EncodeToPriority turns HTTPRoute priority traits into the integer expressed priority.
 //
-//					   4                   3                   2                   1
-//	 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0
-//	+-+---------------+-+-+-------------------+-+---------+---------+-----------------------------------+
-//	|P| host len      |E|R|  Path length      |M|Header No|Query No.| relative order                    |
-//	+-+---------------+-+-+-------------------+-+---------+-------- +-----------------------------------+
+//		   4                   3                   2                   1
+//	 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0 9 8 7 6 5 4 3 2 1 0
+//	+-+---------------+-+-+-------------------+-+---------+---------+-----------------------+
+//	|P| host len      |E|R|  Path length      |M|Header No|Query No.| relative order        |
+//	+-+---------------+-+-+-------------------+-+---------+-------- +-----------------------+
 //
 // Where:
 // P: set to 1 if the hostname is non-wildcard.
@@ -396,34 +396,34 @@ func CalculateHTTPRouteMatchPriorityTraits(match SplitHTTPRouteMatch) HTTPRouteP
 // Header No.: number of header matches.
 // Query No.: number of query parameter matches.
 // relative order: relative order of creation timestamp, namespace and name and internal rule/match order between different (split) HTTPRoutes.
-func (t HTTPRoutePriorityTraits) EncodeToPriority() int {
+func (t HTTPRoutePriorityTraits) EncodeToPriority() RoutePriorityType {
 	const (
-		// PreciseHostnameShiftBits assigns bit 49 for marking if the hostname is non-wildcard.
-		PreciseHostnameShiftBits = 49
-		// HostnameLengthShiftBits assigns bits 41-48 for the length of hostname.
-		HostnameLengthShiftBits = 41
-		// ExactPathShiftBits assigns bit 40 to mark if the match is exact path match.
-		ExactPathShiftBits = 40
-		// RegularExpressionPathShiftBits assigns bit 39 to mark if the match is regex path match.
-		RegularExpressionPathShiftBits = 39
-		// PathLengthShiftBits assigns bits 29-38 to path length. (max length = 1024, but must start with /)
-		PathLengthShiftBits = 29
-		// MethodMatchShiftBits assigns bit 28 to mark if method is specified.
-		MethodMatchShiftBits = 28
-		// HeaderNumberShiftBits assign bits 23-27 to number of headers. (max number of headers = 16)
-		HeaderNumberShiftBits = 23
-		// QueryParamNumberShiftBits makes bits 18-22 used for number of query params (max number of query params = 16)
-		QueryParamNumberShiftBits = 18
-		// bits 0-17 are used for relative order of creation timestamp, namespace/name, and internal order of rules and matches.
+		// PreciseHostnameShiftBits assigns bit 43 for marking if the hostname is non-wildcard.
+		PreciseHostnameShiftBits = 43
+		// HostnameLengthShiftBits assigns bits 35-42 for the length of hostname.
+		HostnameLengthShiftBits = 35
+		// ExactPathShiftBits assigns bit 34 to mark if the match is exact path match.
+		ExactPathShiftBits = 34
+		// RegularExpressionPathShiftBits assigns bit 33 to mark if the match is regex path match.
+		RegularExpressionPathShiftBits = 33
+		// PathLengthShiftBits assigns bits 23-32 to path length. (max length = 1024, but must start with /)
+		PathLengthShiftBits = 23
+		// MethodMatchShiftBits assigns bit 22 to mark if method is specified.
+		MethodMatchShiftBits = 22
+		// HeaderNumberShiftBits assign bits 17-21 to number of headers. (max number of headers = 16)
+		HeaderNumberShiftBits = 17
+		// QueryParamNumberShiftBits makes bits 12-16 used for number of query params (max number of query params = 16)
+		QueryParamNumberShiftBits = 12
+		// bits 0-11 are used for relative order of creation timestamp, namespace/name, and internal order of rules and matches.
 		// the bits are calculated by sorting HTTPRoutes with the same priority calculated from the fields above
 		// and start from all 1s, then decrease by one for each HTTPRoute.
 	)
 
-	var priority int
+	var priority RoutePriorityType
 	if t.PreciseHostname {
 		priority += (1 << PreciseHostnameShiftBits)
 	}
-	priority += t.HostnameLength << HostnameLengthShiftBits
+	priority += RoutePriorityType(t.HostnameLength << HostnameLengthShiftBits)
 
 	if t.PathType == gatewayapi.PathMatchExact {
 		priority += (1 << ExactPathShiftBits)
@@ -434,15 +434,15 @@ func (t HTTPRoutePriorityTraits) EncodeToPriority() int {
 
 	// max length of path is 1024, but path must start with /, so we use PathLength-1 to fill the bits.
 	if t.PathLength > 0 {
-		priority += ((t.PathLength - 1) << PathLengthShiftBits)
+		priority += RoutePriorityType(((t.PathLength - 1) << PathLengthShiftBits))
 	}
 
-	priority += (t.HeaderCount << HeaderNumberShiftBits)
+	priority += RoutePriorityType(t.HeaderCount << HeaderNumberShiftBits)
 	if t.HasMethodMatch {
 		priority += (1 << MethodMatchShiftBits)
 	}
-	priority += (t.QueryParamCount << QueryParamNumberShiftBits)
-	priority += (ResourceKindBitsHTTPRoute << FromResourceKindPriorityShiftBits)
+	priority += RoutePriorityType(t.QueryParamCount << QueryParamNumberShiftBits)
+	priority += RoutePriorityType(ResourceKindBitsHTTPRoute << FromResourceKindPriorityShiftBits)
 
 	return priority
 }
@@ -458,7 +458,7 @@ func AssignRoutePriorityToSplitHTTPRouteMatches(
 	logger logr.Logger,
 	splitHTTPRouteMatches []SplitHTTPRouteMatch,
 ) []SplitHTTPRouteMatchToKongRoutePriority {
-	priorityToSplitHTTPRouteMatches := map[int][]SplitHTTPRouteMatch{}
+	priorityToSplitHTTPRouteMatches := map[RoutePriorityType][]SplitHTTPRouteMatch{}
 
 	for _, match := range splitHTTPRouteMatches {
 		priority := CalculateHTTPRouteMatchPriorityTraits(match).EncodeToPriority()
@@ -467,12 +467,12 @@ func AssignRoutePriorityToSplitHTTPRouteMatches(
 
 	httpRouteMatchesToPriorities := make([]SplitHTTPRouteMatchToKongRoutePriority, 0, len(splitHTTPRouteMatches))
 
-	// Bits 0-17 (18 bits) are assigned for relative order of matches.
+	// Bits 0-11 (12 bits) are assigned for relative order of matches.
 	// If multiple matches are assigned to the same priority in the previous step,
-	// sort them then starts with 2^18 -1 and decrease by one for each HTTPRoute;
+	// sort them then starts with 2^12 -1 and decrease by one for each HTTPRoute;
 	// If only one match occupies the priority, fill the relative order bits with all 1s.
-	const RelativeOrderAssignedBits = 18
-	const defaultRelativeOrderPriorityBits = (1 << RelativeOrderAssignedBits) - 1
+	const RelativeOrderAssignedBits = 12
+	const defaultRelativeOrderPriorityBits = (uint64(1) << RelativeOrderAssignedBits) - 1
 	for priority, matches := range priorityToSplitHTTPRouteMatches {
 		if len(matches) == 1 {
 			httpRouteMatchesToPriorities = append(httpRouteMatchesToPriorities, SplitHTTPRouteMatchToKongRoutePriority{
@@ -487,8 +487,8 @@ func AssignRoutePriorityToSplitHTTPRouteMatches(
 		})
 
 		for i, match := range matches {
-			relativeOrderBits := defaultRelativeOrderPriorityBits - i
-			// Although it is very unlikely that there are 2^18 = 262144 HTTPRoutes
+			relativeOrderBits := defaultRelativeOrderPriorityBits - RoutePriorityType(i)
+			// Although it is very unlikely that there are 2^12 = 4096 HTTPRoutes
 			// should be given priority by their relative order, here we limit the
 			// relativeOrderBits to be at least 0.
 			if relativeOrderBits <= 0 {
@@ -499,9 +499,9 @@ func AssignRoutePriorityToSplitHTTPRouteMatches(
 				Priority: priority + relativeOrderBits,
 			})
 		}
-		// Just in case, log a very unlikely scenario where we have more than 2^18 matches with the same base
+		// Just in case, log a very unlikely scenario where we have more than 2^12 matches with the same base
 		// priority and we have no bit space for them to be deterministically ordered.
-		if len(matches) > (1 << 18) {
+		if len(matches) > (1 << 12) {
 			logger.Error(nil, "Too many HTTPRoute matches to be deterministically ordered", "match_number", len(matches))
 		}
 	}
@@ -582,7 +582,7 @@ func KongExpressionRouteFromHTTPRouteMatchWithPriority(
 	// generate a "catch-all" route if the generated expression is empty.
 	if r.Expression == nil || len(*r.Expression) == 0 {
 		r.Expression = kong.String(CatchAllHTTPExpression)
-		r.Priority = kong.Int(httpRouteMatchWithPriority.Priority)
+		r.Priority = kong.Uint64(httpRouteMatchWithPriority.Priority)
 	}
 
 	// translate filters in the rule.
diff --git a/internal/dataplane/parser/translators/ingress_atc.go b/internal/dataplane/parser/translators/ingress_atc.go
index 09f01f370..7406a5d39 100644
--- a/internal/dataplane/parser/translators/ingress_atc.go
+++ b/internal/dataplane/parser/translators/ingress_atc.go
@@ -27,8 +27,8 @@ var (
 )
 
 var (
-	NormalIngressExpressionPriority = 1
-	IngressDefaultBackendPriority   = 0
+	NormalIngressExpressionPriority RoutePriorityType = 1
+	IngressDefaultBackendPriority   RoutePriorityType = 0
 )
 
 func (m *ingressTranslationMeta) translateIntoKongExpressionRoute() *kongstate.Route {
@@ -244,7 +244,7 @@ type IngressRoutePriorityTraits struct {
 //   - P (Plain Host): set if ALL hosts are non-wildcard.
 //   - R (Regex): if set, regex match is used.
 //   - Path Length: maximum length of the path to match.
-func (t IngressRoutePriorityTraits) EncodeToPriority() int {
+func (t IngressRoutePriorityTraits) EncodeToPriority() RoutePriorityType {
 	// route.priority in admin API could only use the lowest 52 bits
 	// because the numbers in JSON are parsed into double precision floating numbers.
 	const (
@@ -265,12 +265,12 @@ func (t IngressRoutePriorityTraits) EncodeToPriority() int {
 		pathLengthLimit   = (1 << 16) - 1
 	)
 
-	var priority int
+	var priority RoutePriorityType
 	// add max path length.
 	if t.MaxPathLength > pathLengthLimit {
 		t.MaxPathLength = pathLengthLimit
 	}
-	priority += t.MaxPathLength
+	priority += RoutePriorityType(t.MaxPathLength)
 	// add regex path mark.
 	if t.HasRegexPath {
 		priority += (1 << regexPathShiftBits)
@@ -282,9 +282,9 @@ func (t IngressRoutePriorityTraits) EncodeToPriority() int {
 	if t.HeaderCount > headerNumberLimit {
 		t.HeaderCount = headerNumberLimit
 	}
-	priority += (t.HeaderCount << headerNumberShiftBits)
-	priority += (t.MatchFields << matchFieldsShiftBits)
-	priority += (ResourceKindBitsIngress << FromResourceKindPriorityShiftBits)
+	priority += RoutePriorityType(t.HeaderCount << headerNumberShiftBits)
+	priority += RoutePriorityType(t.MatchFields << matchFieldsShiftBits)
+	priority += RoutePriorityType(ResourceKindBitsIngress << FromResourceKindPriorityShiftBits)
 
 	return priority
 }
@@ -400,7 +400,7 @@ func calculateExpressionRoutePriority(
 	regexPathPrefix string,
 	ingressHost string,
 	ingressAnnotations map[string]string,
-) int {
+) RoutePriorityType {
 	traits := calculateIngressRoutePriorityTraits(
 		paths, regexPathPrefix, ingressHost, ingressAnnotations,
 	)
diff --git a/internal/manager/config.go b/internal/manager/config.go
index de57440da..7d9e1744b 100644
--- a/internal/manager/config.go
+++ b/internal/manager/config.go
@@ -23,6 +23,7 @@ import (
 	cfgtypes "github.com/kong/kubernetes-ingress-controller/v3/internal/manager/config/types"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/manager/featuregates"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/manager/flags"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/manager/metadata"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/util/kubernetes/object/status"
 )
 
@@ -326,6 +327,8 @@ func (c *Config) GetKubeconfig() (*rest.Config, error) {
 		config.Impersonate.UserName = c.Impersonate
 	}
 
+	config.UserAgent = "kong-ingress-controller/" + metadata.Release
+
 	return config, err
 }
 
diff --git a/internal/store/store.go b/internal/store/store.go
index 90f9a2126..0243beef4 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -740,6 +740,8 @@ func mkObjFromGVK(gvk schema.GroupVersionKind) (runtime.Object, error) {
 		return &kongv1beta1.KongConsumerGroup{}, nil
 	case kongv1alpha1.SchemeGroupVersion.WithKind("IngressClassParameters"):
 		return &kongv1alpha1.IngressClassParameters{}, nil
+	case kongv1beta1.SchemeGroupVersion.WithKind("KongUpstreamPolicy"):
+		return &kongv1beta1.KongUpstreamPolicy{}, nil
 	default:
 		return nil, fmt.Errorf("%s is not a supported runtime.Object", gvk)
 	}
diff --git a/pkg/apis/configuration/v1beta1/kongupstreampolicy_types.go b/pkg/apis/configuration/v1beta1/kongupstreampolicy_types.go
index 00fcd2c78..ac5f5a768 100644
--- a/pkg/apis/configuration/v1beta1/kongupstreampolicy_types.go
+++ b/pkg/apis/configuration/v1beta1/kongupstreampolicy_types.go
@@ -1,7 +1,11 @@
 package v1beta1
 
-import (
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+const (
+	// KongUpstreamPolicyAnnotationKey is the key used to attach KongUpstreamPolicy to Services.
+	// The value of the annotation is the name of the KongUpstreamPolicy object in the same namespace as the Service.
+	KongUpstreamPolicyAnnotationKey = "konghq.com/upstream-policy"
 )
 
 func init() {
@@ -35,15 +39,16 @@ func init() {
 // +kubebuilder:storageversion
 // +kubebuilder:metadata:labels=gateway.networking.k8s.io/policy=direct
 // +kubebuilder:validation:XValidation:rule="has(self.spec.hashOn) ? [has(self.spec.hashOn.input), has(self.spec.hashOn.cookie), has(self.spec.hashOn.header), has(self.spec.hashOn.uriCapture), has(self.spec.hashOn.queryArg)].filter(fieldSet, fieldSet == true).size() <= 1 : true", message="Only one of spec.hashOn.(input|cookie|header|uriCapture|queryArg) can be set."
-// +kubebuilder:validation:XValidation:rule="has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input), has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture), has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size() <= 1 : true", message="Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg) can be set."
 // +kubebuilder:validation:XValidation:rule="has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? has(self.spec.hashOn.cookiePath) : true", message="When spec.hashOn.cookie is set, spec.hashOn.cookiePath is required."
 // +kubebuilder:validation:XValidation:rule="has(self.spec.hashOn) && has(self.spec.hashOn.cookiePath) ? has(self.spec.hashOn.cookie) : true", message="When spec.hashOn.cookiePath is set, spec.hashOn.cookie is required."
+// +kubebuilder:validation:XValidation:rule="has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm == \"consistent-hashing\" : true", message="spec.algorithm must be set to \"consistent-hashing\" when spec.hashOn is set."
+// +kubebuilder:validation:XValidation:rule="has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input), has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture), has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size() <= 1 : true", message="Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg) can be set."
+// +kubebuilder:validation:XValidation:rule="has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm == \"consistent-hashing\" : true", message="spec.algorithm must be set to \"consistent-hashing\" when spec.hashOnFallback is set."
 // +kubebuilder:validation:XValidation:rule="has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookie) : true", message="spec.hashOnFallback.cookie must not be set."
 // +kubebuilder:validation:XValidation:rule="has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookiePath) : true", message="spec.hashOnFallback.cookiePath must not be set."
 // +kubebuilder:validation:XValidation:rule="has(self.spec.healthchecks) && has(self.spec.healthchecks.passive) && has(self.spec.healthchecks.passive.healthy) ? !has(self.spec.healthchecks.passive.healthy.interval) : true", message="spec.healthchecks.passive.healthy.interval must not be set."
 // +kubebuilder:validation:XValidation:rule="has(self.spec.healthchecks) && has(self.spec.healthchecks.passive) && has(self.spec.healthchecks.passive.unhealthy) ? !has(self.spec.healthchecks.passive.unhealthy.interval) : true", message="spec.healthchecks.passive.unhealthy.interval must not be set."
-// +kubebuilder:validation:XValidation:rule="has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm == \"consistent-hashing\" : true", message="spec.algorithm must be set to \"consistent-hashing\" when spec.hashOn is set."
-// +kubebuilder:validation:XValidation:rule="has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm == \"consistent-hashing\" : true", message="spec.algorithm must be set to \"consistent-hashing\" when spec.hashOnFallback is set."
+// +kubebuilder:validation:XValidation:rule="has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? !has(self.spec.hashOnFallback) : true", message="spec.hashOnFallback must not be set when spec.hashOn.cookie is set."
 type KongUpstreamPolicy struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`
@@ -87,9 +92,6 @@ type KongUpstreamPolicySpec struct {
 
 	// Healthchecks defines the health check configurations in Kong.
 	Healthchecks *KongUpstreamHealthcheck `json:"healthchecks,omitempty"`
-
-	// HostHeader is the hostname to be used as Host header when proxying requests through Kong.
-	HostHeader *string `json:"hostHeader,omitempty"`
 }
 
 // HashInput is the input for consistent-hashing load balancing algorithm.
diff --git a/pkg/apis/configuration/v1beta1/zz_generated.deepcopy.go b/pkg/apis/configuration/v1beta1/zz_generated.deepcopy.go
index 186941dcb..bb944028b 100644
--- a/pkg/apis/configuration/v1beta1/zz_generated.deepcopy.go
+++ b/pkg/apis/configuration/v1beta1/zz_generated.deepcopy.go
@@ -488,11 +488,6 @@ func (in *KongUpstreamPolicySpec) DeepCopyInto(out *KongUpstreamPolicySpec) {
 		*out = new(KongUpstreamHealthcheck)
 		(*in).DeepCopyInto(*out)
 	}
-	if in.HostHeader != nil {
-		in, out := &in.HostHeader, &out.HostHeader
-		*out = new(string)
-		**out = **in
-	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KongUpstreamPolicySpec.


Test patch:
diff --git a/internal/dataplane/kongstate/kongstate_test.go b/internal/dataplane/kongstate/kongstate_test.go
index e37c43310..1deff122f 100644
--- a/internal/dataplane/kongstate/kongstate_test.go
+++ b/internal/dataplane/kongstate/kongstate_test.go
@@ -7,6 +7,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/go-logr/logr"
 	"github.com/go-logr/zapr"
 	"github.com/kong/go-kong/kong"
 	"github.com/samber/lo"
@@ -959,3 +960,183 @@ func TestKongState_BuildPluginsCollisions(t *testing.T) {
 		})
 	}
 }
+
+func TestKongState_FillUpstreamOverrides(t *testing.T) {
+	const (
+		kongIngressName        = "kongIngress"
+		kongUpstreamPolicyName = "policy"
+	)
+	serviceAnnotatedWithKongUpstreamPolicy := func() *corev1.Service {
+		return &corev1.Service{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "Service",
+				APIVersion: "v1",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "service",
+				Namespace: "default",
+				Annotations: map[string]string{
+					kongv1beta1.KongUpstreamPolicyAnnotationKey: kongUpstreamPolicyName,
+				},
+			},
+		}
+	}
+
+	serviceAnnotatedWithKongUpstreamPolicyAndKongIngress := func() *corev1.Service {
+		s := serviceAnnotatedWithKongUpstreamPolicy()
+		s.Annotations[annotations.AnnotationPrefix+annotations.ConfigurationKey] = kongIngressName
+		return s
+	}
+
+	testCases := []struct {
+		name                 string
+		upstream             Upstream
+		kongUpstreamPolicies []*kongv1beta1.KongUpstreamPolicy
+		kongIngresses        []*kongv1.KongIngress
+		expectedUpstream     kong.Upstream
+		expectedFailures     []failures.ResourceFailure
+	}{
+		{
+			name: "upstream with no overrides",
+			upstream: Upstream{
+				Upstream: kong.Upstream{
+					Name: kong.String("foo-upstream"),
+				},
+			},
+			expectedUpstream: kong.Upstream{
+				Name: kong.String("foo-upstream"),
+			},
+		},
+		{
+			name: "upstream backed by service annotated with KongUpstreamPolicy",
+			upstream: Upstream{
+				Upstream: kong.Upstream{
+					Name: kong.String("foo-upstream"),
+				},
+				Service: Service{
+					K8sServices: map[string]*corev1.Service{"": serviceAnnotatedWithKongUpstreamPolicy()},
+				},
+			},
+			kongUpstreamPolicies: []*kongv1beta1.KongUpstreamPolicy{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      kongUpstreamPolicyName,
+						Namespace: "default",
+					},
+					Spec: kongv1beta1.KongUpstreamPolicySpec{
+						Algorithm: lo.ToPtr("least-connections"),
+					},
+				},
+			},
+			expectedUpstream: kong.Upstream{
+				Name:      kong.String("foo-upstream"),
+				Algorithm: kong.String("least-connections"),
+			},
+		},
+		{
+			name: "upstream backed by service annotated with KongUpstreamPolicy that doesn't exist",
+			upstream: Upstream{
+				Upstream: kong.Upstream{
+					Name: kong.String("foo-upstream"),
+				},
+				Service: Service{
+					K8sServices: map[string]*corev1.Service{"": serviceAnnotatedWithKongUpstreamPolicy()},
+				},
+			},
+			expectedUpstream: kong.Upstream{
+				Name: kong.String("foo-upstream"),
+			},
+			expectedFailures: []failures.ResourceFailure{
+				lo.Must(failures.NewResourceFailure(
+					"failed fetching KongUpstreamPolicy: KongUpstreamPolicy default/policy not found",
+					serviceAnnotatedWithKongUpstreamPolicy(),
+				)),
+			},
+		},
+		{
+			name: "KongUpstreamPolicy is applied even if KongIngress is not found",
+			upstream: Upstream{
+				Upstream: kong.Upstream{
+					Name: kong.String("foo-upstream"),
+				},
+				Service: Service{
+					K8sServices: map[string]*corev1.Service{"": serviceAnnotatedWithKongUpstreamPolicyAndKongIngress()},
+				},
+			},
+			kongUpstreamPolicies: []*kongv1beta1.KongUpstreamPolicy{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      kongUpstreamPolicyName,
+						Namespace: "default",
+					},
+					Spec: kongv1beta1.KongUpstreamPolicySpec{
+						Algorithm: lo.ToPtr("least-connections"),
+					},
+				},
+			},
+			expectedUpstream: kong.Upstream{
+				Name:      kong.String("foo-upstream"),
+				Algorithm: kong.String("least-connections"),
+			},
+			expectedFailures: []failures.ResourceFailure{
+				lo.Must(failures.NewResourceFailure(
+					"failed to get KongIngress: KongIngress kongIngress not found",
+					serviceAnnotatedWithKongUpstreamPolicyAndKongIngress(),
+				)),
+			},
+		},
+		{
+			name: "KongUpstreamPolicy overwrites KongIngress",
+			upstream: Upstream{
+				Upstream: kong.Upstream{
+					Name: kong.String("foo-upstream"),
+				},
+				Service: Service{
+					K8sServices: map[string]*corev1.Service{"": serviceAnnotatedWithKongUpstreamPolicyAndKongIngress()},
+				},
+			},
+			kongUpstreamPolicies: []*kongv1beta1.KongUpstreamPolicy{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      kongUpstreamPolicyName,
+						Namespace: "default",
+					},
+					Spec: kongv1beta1.KongUpstreamPolicySpec{
+						Algorithm: lo.ToPtr("least-connections"),
+					},
+				},
+			},
+			kongIngresses: []*kongv1.KongIngress{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      kongIngressName,
+						Namespace: "default",
+					},
+					Upstream: &kongv1.KongIngressUpstream{
+						Algorithm: lo.ToPtr("round-robin"),
+					},
+				},
+			},
+			expectedUpstream: kong.Upstream{
+				Name:      kong.String("foo-upstream"),
+				Algorithm: kong.String("least-connections"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			s, err := store.NewFakeStore(store.FakeObjects{
+				KongUpstreamPolicies: tc.kongUpstreamPolicies,
+				KongIngresses:        tc.kongIngresses,
+			})
+			require.NoError(t, err)
+			failuresCollector := failures.NewResourceFailuresCollector(logr.Discard())
+
+			kongState := KongState{Upstreams: []Upstream{tc.upstream}}
+			kongState.FillUpstreamOverrides(s, failuresCollector)
+			require.Equal(t, tc.expectedUpstream, kongState.Upstreams[0].Upstream)
+			require.ElementsMatch(t, tc.expectedFailures, failuresCollector.PopResourceFailures())
+		})
+	}
+}
diff --git a/internal/dataplane/parser/translators/kongupstreampolicy_test.go b/internal/dataplane/kongstate/kongupstreampolicy_test.go
similarity index 56%
rename from internal/dataplane/parser/translators/kongupstreampolicy_test.go
rename to internal/dataplane/kongstate/kongupstreampolicy_test.go
index cecebd171..1dc632378 100644
--- a/internal/dataplane/parser/translators/kongupstreampolicy_test.go
+++ b/internal/dataplane/kongstate/kongupstreampolicy_test.go
@@ -1,4 +1,4 @@
-package translators_test
+package kongstate_test
 
 import (
 	"testing"
@@ -6,11 +6,158 @@ import (
 	"github.com/kong/go-kong/kong"
 	"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/parser/translators"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/kongstate"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
 	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 )
 
+func TestGetKongUpstreamPolicyForServices(t *testing.T) {
+	testCases := []struct {
+		name          string
+		servicesGroup []*corev1.Service
+		policies      []*kongv1beta1.KongUpstreamPolicy
+		expectPolicy  bool
+		expectError   string
+	}{
+		{
+			name:         "no services in group gives no policy",
+			expectPolicy: false,
+		},
+		{
+			name: "no KongUpstreamPolicy in store while services are configured with one gives error",
+			servicesGroup: []*corev1.Service{{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "svc",
+					Namespace: "default",
+					Annotations: map[string]string{
+						kongv1beta1.KongUpstreamPolicyAnnotationKey: "upstream-policy",
+					},
+				},
+			}},
+			expectError: "failed fetching KongUpstreamPolicy",
+		},
+		{
+			name: "all services configured with no KongUpstreamPolicy gives no policy and no error",
+			servicesGroup: []*corev1.Service{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "svc-1",
+						Namespace: "default",
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "svc-2",
+						Namespace: "default",
+					},
+				},
+			},
+			expectPolicy: false,
+		},
+		{
+			name: "services in group with different KongUpstreamPolicy configurations gives error",
+			servicesGroup: []*corev1.Service{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "svc-1",
+						Namespace: "default",
+						Annotations: map[string]string{
+							kongv1beta1.KongUpstreamPolicyAnnotationKey: "upstream-policy",
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "svc-2",
+						Namespace: "default",
+						Annotations: map[string]string{
+							kongv1beta1.KongUpstreamPolicyAnnotationKey: "other-upstream-policy",
+						},
+					},
+				},
+			},
+			expectError: "inconsistent KongUpstreamPolicy configuration for services",
+		},
+		{
+			name: "one service with and one without KongUpstreamPolicy configuration gives error",
+			servicesGroup: []*corev1.Service{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "svc-1",
+						Namespace: "default",
+						Annotations: map[string]string{
+							kongv1beta1.KongUpstreamPolicyAnnotationKey: "upstream-policy",
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "svc-2",
+						Namespace: "default",
+					},
+				},
+			},
+			expectError: "inconsistent KongUpstreamPolicy configuration for services",
+		},
+		{
+			name: "all the services configured with the same KongUpstreamPolicy gives the policy",
+			servicesGroup: []*corev1.Service{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "svc-1",
+						Namespace: "default",
+						Annotations: map[string]string{
+							kongv1beta1.KongUpstreamPolicyAnnotationKey: "upstream-policy",
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "svc-2",
+						Namespace: "default",
+						Annotations: map[string]string{
+							kongv1beta1.KongUpstreamPolicyAnnotationKey: "upstream-policy",
+						},
+					},
+				},
+			},
+			policies: []*kongv1beta1.KongUpstreamPolicy{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "upstream-policy",
+						Namespace: "default",
+					},
+				},
+			},
+			expectPolicy: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			s, err := store.NewFakeStore(store.FakeObjects{
+				Services:             tc.servicesGroup,
+				KongUpstreamPolicies: tc.policies,
+			})
+			require.NoError(t, err)
+
+			policy, err := kongstate.GetKongUpstreamPolicyForServices(s, tc.servicesGroup)
+			if tc.expectError != "" {
+				require.ErrorContains(t, err, tc.expectError)
+				return
+			}
+			if tc.expectPolicy {
+				require.NotNil(t, policy)
+			} else {
+				require.Nil(t, policy)
+			}
+		})
+	}
+}
+
 func TestTranslateKongUpstreamPolicy(t *testing.T) {
 	testCases := []struct {
 		name             string
@@ -20,14 +167,12 @@ func TestTranslateKongUpstreamPolicy(t *testing.T) {
 		{
 			name: "KongUpstreamPolicySpec with no hash-on or hash-fallback",
 			policySpec: kongv1beta1.KongUpstreamPolicySpec{
-				HostHeader: lo.ToPtr("foo"),
-				Algorithm:  lo.ToPtr("least-connections"),
-				Slots:      lo.ToPtr(10),
+				Algorithm: lo.ToPtr("least-connections"),
+				Slots:     lo.ToPtr(10),
 			},
 			expectedUpstream: &kong.Upstream{
-				HostHeader: lo.ToPtr("foo"),
-				Algorithm:  lo.ToPtr("least-connections"),
-				Slots:      lo.ToPtr(10),
+				Algorithm: lo.ToPtr("least-connections"),
+				Slots:     lo.ToPtr(10),
 			},
 		},
 		{
@@ -175,7 +320,7 @@ func TestTranslateKongUpstreamPolicy(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			actualUpstream := translators.TranslateKongUpstreamPolicy(tc.policySpec)
+			actualUpstream := kongstate.TranslateKongUpstreamPolicy(tc.policySpec)
 			require.Equal(t, tc.expectedUpstream, actualUpstream)
 		})
 	}
diff --git a/internal/dataplane/kongstate/upstream_test.go b/internal/dataplane/kongstate/upstream_test.go
index 7b59083a9..9c472e51a 100644
--- a/internal/dataplane/kongstate/upstream_test.go
+++ b/internal/dataplane/kongstate/upstream_test.go
@@ -4,16 +4,16 @@ import (
 	"testing"
 
 	"github.com/kong/go-kong/kong"
-	"github.com/stretchr/testify/assert"
+	"github.com/samber/lo"
+	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 )
 
 func TestOverrideUpstream(t *testing.T) {
-	assert := assert.New(t)
-
 	testTable := []struct {
 		inUpstream     Upstream
 		inKongIngresss *kongv1.KongIngress
@@ -83,11 +83,167 @@ func TestOverrideUpstream(t *testing.T) {
 
 	for _, testcase := range testTable {
 		testcase.inUpstream.override(testcase.inKongIngresss, testcase.svc)
-		assert.Equal(testcase.inUpstream, testcase.outUpstream)
+		require.Equal(t, testcase.inUpstream, testcase.outUpstream)
 	}
 
-	assert.NotPanics(func() {
+	require.NotPanics(t, func() {
 		var nilUpstream *Upstream
 		nilUpstream.override(nil, nil)
 	})
 }
+
+func TestUpstreamOverrideByKongUpstreamPolicy(t *testing.T) {
+	testCases := []struct {
+		name               string
+		upstream           kong.Upstream
+		kongUpstreamPolicy *kongv1beta1.KongUpstreamPolicy
+		expected           kong.Upstream
+	}{
+		{
+			name: "algorithm, slots, healthchecks, hash_on, hash_fallback, hash_fallback_header",
+			upstream: kong.Upstream{
+				Algorithm: kong.String("Algorithm"),
+				Slots:     kong.Int(42),
+				Healthchecks: &kong.Healthcheck{
+					Active: &kong.ActiveHealthcheck{
+						Concurrency: kong.Int(1),
+					},
+				},
+				HashOn:             kong.String("HashOn"),
+				HashFallback:       kong.String("HashFallback"),
+				HashFallbackHeader: kong.String("HashFallbackHeader"),
+			},
+			kongUpstreamPolicy: &kongv1beta1.KongUpstreamPolicy{
+				Spec: kongv1beta1.KongUpstreamPolicySpec{
+					Algorithm: lo.ToPtr("least-connections"),
+					Slots:     lo.ToPtr(10),
+					Healthchecks: &kongv1beta1.KongUpstreamHealthcheck{
+						Active: &kongv1beta1.KongUpstreamActiveHealthcheck{
+							Concurrency: lo.ToPtr(2),
+						},
+					},
+					HashOn: &kongv1beta1.KongUpstreamHash{
+						Input: lo.ToPtr(kongv1beta1.HashInput("consumer")),
+					},
+					HashOnFallback: &kongv1beta1.KongUpstreamHash{
+						Header: lo.ToPtr("foo"),
+					},
+				},
+			},
+			expected: kong.Upstream{
+				Algorithm: kong.String("least-connections"),
+				Slots:     kong.Int(10),
+				Healthchecks: &kong.Healthcheck{
+					Active: &kong.ActiveHealthcheck{
+						Concurrency: kong.Int(2),
+					},
+				},
+				HashOn:             kong.String("consumer"),
+				HashFallback:       kong.String("header"),
+				HashFallbackHeader: kong.String("foo"),
+			},
+		},
+		{
+			name: "hash_on_header, hash_fallback_query_arg",
+			upstream: kong.Upstream{
+				HashOn:               kong.String("HashOn"),
+				HashFallback:         kong.String("HashFallback"),
+				HashOnHeader:         kong.String("HashOnHeader"),
+				HashFallbackQueryArg: kong.String("HashOnQueryArg"),
+			},
+			kongUpstreamPolicy: &kongv1beta1.KongUpstreamPolicy{
+				Spec: kongv1beta1.KongUpstreamPolicySpec{
+					HashOn: &kongv1beta1.KongUpstreamHash{
+						Header: lo.ToPtr("foo"),
+					},
+					HashOnFallback: &kongv1beta1.KongUpstreamHash{
+						QueryArg: lo.ToPtr("foo"),
+					},
+				},
+			},
+			expected: kong.Upstream{
+				HashOn:               kong.String("header"),
+				HashFallback:         kong.String("query_arg"),
+				HashOnHeader:         kong.String("foo"),
+				HashFallbackQueryArg: kong.String("foo"),
+			},
+		},
+		{
+			name: "hash_on_cookie, hash_on_cookie_path, hash_fallback_uri_capture",
+			upstream: kong.Upstream{
+				HashOn:                 kong.String("HashOn"),
+				HashFallback:           kong.String("HashFallback"),
+				HashOnCookie:           kong.String("HashOnCookie"),
+				HashOnCookiePath:       kong.String("HashOnCookiePath"),
+				HashFallbackURICapture: kong.String("HashFallbackURICapture"),
+			},
+			kongUpstreamPolicy: &kongv1beta1.KongUpstreamPolicy{
+				Spec: kongv1beta1.KongUpstreamPolicySpec{
+					HashOn: &kongv1beta1.KongUpstreamHash{
+						Cookie:     lo.ToPtr("foo"),
+						CookiePath: lo.ToPtr("/"),
+					},
+					HashOnFallback: &kongv1beta1.KongUpstreamHash{
+						URICapture: lo.ToPtr("foo"),
+					},
+				},
+			},
+			expected: kong.Upstream{
+				HashOn:                 kong.String("cookie"),
+				HashFallback:           kong.String("uri_capture"),
+				HashOnCookie:           kong.String("foo"),
+				HashOnCookiePath:       kong.String("/"),
+				HashFallbackURICapture: kong.String("foo"),
+			},
+		},
+		{
+			name: "hash_on_uri_capture",
+			upstream: kong.Upstream{
+				HashOn:           kong.String("HashOn"),
+				HashOnURICapture: kong.String("HashOnURICapture"),
+			},
+			kongUpstreamPolicy: &kongv1beta1.KongUpstreamPolicy{
+				Spec: kongv1beta1.KongUpstreamPolicySpec{
+					HashOn: &kongv1beta1.KongUpstreamHash{
+						URICapture: lo.ToPtr("foo"),
+					},
+				},
+			},
+			expected: kong.Upstream{
+				HashOn:           kong.String("uri_capture"),
+				HashOnURICapture: kong.String("foo"),
+			},
+		},
+		{
+			name: "hash_on_query_arg",
+			upstream: kong.Upstream{
+				HashOn:         kong.String("HashOn"),
+				HashOnQueryArg: kong.String("HashOnQueryArg"),
+			},
+			kongUpstreamPolicy: &kongv1beta1.KongUpstreamPolicy{
+				Spec: kongv1beta1.KongUpstreamPolicySpec{
+					HashOn: &kongv1beta1.KongUpstreamHash{
+						QueryArg: lo.ToPtr("foo"),
+					},
+				},
+			},
+			expected: kong.Upstream{
+				HashOn:         kong.String("query_arg"),
+				HashOnQueryArg: kong.String("foo"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			upstream := Upstream{Upstream: tc.upstream}
+			upstream.overrideByKongUpstreamPolicy(tc.kongUpstreamPolicy)
+			require.Equal(t, tc.expected, upstream.Upstream)
+		})
+	}
+
+	require.NotPanics(t, func() {
+		var nilUpstream *Upstream
+		nilUpstream.overrideByKongUpstreamPolicy(nil)
+	})
+}
diff --git a/internal/dataplane/kongstate/util_test.go b/internal/dataplane/kongstate/util_test.go
index d6bb64269..49a483d9d 100644
--- a/internal/dataplane/kongstate/util_test.go
+++ b/internal/dataplane/kongstate/util_test.go
@@ -303,7 +303,7 @@ func TestKongPluginFromK8SPlugin(t *testing.T) {
 func TestGetKongIngressForServices(t *testing.T) {
 	for _, tt := range []struct {
 		name                string
-		services            map[string]*corev1.Service
+		services            []*corev1.Service
 		kongIngresses       []*kongv1.KongIngress
 		expectedKongIngress *kongv1.KongIngress
 		expectedError       error
@@ -319,14 +319,14 @@ func TestGetKongIngressForServices(t *testing.T) {
 		},
 		{
 			name: "when none of the provided services have attached KongIngress resources, no KongIngress resources will be provided",
-			services: map[string]*corev1.Service{
-				"test-service1": {
+			services: []*corev1.Service{
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service1",
 						Namespace: corev1.NamespaceDefault,
 					},
 				},
-				"test-service2": {
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service2",
 						Namespace: corev1.NamespaceDefault,
@@ -342,8 +342,8 @@ func TestGetKongIngressForServices(t *testing.T) {
 		},
 		{
 			name: "if at least one KongIngress resource is attached to a Service, it will be returned",
-			services: map[string]*corev1.Service{
-				"test-service1": {
+			services: []*corev1.Service{
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service1",
 						Namespace: corev1.NamespaceDefault,
@@ -352,7 +352,7 @@ func TestGetKongIngressForServices(t *testing.T) {
 						},
 					},
 				},
-				"test-service2": {
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service2",
 						Namespace: corev1.NamespaceDefault,
@@ -382,8 +382,8 @@ func TestGetKongIngressForServices(t *testing.T) {
 		},
 		{
 			name: "if multiple services have KongIngress resources this is accepted only if they're all attached to the same KongIngress",
-			services: map[string]*corev1.Service{
-				"test-service1": {
+			services: []*corev1.Service{
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service1",
 						Namespace: corev1.NamespaceDefault,
@@ -392,7 +392,7 @@ func TestGetKongIngressForServices(t *testing.T) {
 						},
 					},
 				},
-				"test-service2": {
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service2",
 						Namespace: corev1.NamespaceDefault,
@@ -401,7 +401,7 @@ func TestGetKongIngressForServices(t *testing.T) {
 						},
 					},
 				},
-				"test-service3": {
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service3",
 						Namespace: corev1.NamespaceDefault,
@@ -447,10 +447,10 @@ func TestGetKongIngressForServices(t *testing.T) {
 
 			kongIngress, err := getKongIngressForServices(storer, tt.services)
 			if tt.expectedError == nil {
-				assert.Equal(t, tt.expectedKongIngress, kongIngress)
+				require.Equal(t, tt.expectedKongIngress, kongIngress)
 			} else {
-				assert.Error(t, err)
-				assert.Contains(t, err.Error(), tt.expectedError.Error())
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tt.expectedError.Error())
 			}
 		})
 	}
@@ -581,7 +581,7 @@ func TestGetKongIngressFromObjectMeta(t *testing.T) {
 func TestPrettyPrintServiceList(t *testing.T) {
 	for _, tt := range []struct {
 		name     string
-		services map[string]*corev1.Service
+		services []*corev1.Service
 		expected string
 	}{
 		{
@@ -590,8 +590,8 @@ func TestPrettyPrintServiceList(t *testing.T) {
 		},
 		{
 			name: "a single service should just return the <namespace>/<name>",
-			services: map[string]*corev1.Service{
-				"test-service1": {
+			services: []*corev1.Service{
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service1",
 						Namespace: corev1.NamespaceDefault,
@@ -602,20 +602,20 @@ func TestPrettyPrintServiceList(t *testing.T) {
 		},
 		{
 			name: "multiple services should be comma deliniated",
-			services: map[string]*corev1.Service{
-				"test-service1": {
+			services: []*corev1.Service{
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service1",
 						Namespace: corev1.NamespaceDefault,
 					},
 				},
-				"test-service2": {
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service2",
 						Namespace: corev1.NamespaceDefault,
 					},
 				},
-				"test-service3": {
+				{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      "test-service3",
 						Namespace: corev1.NamespaceDefault,
@@ -627,7 +627,7 @@ func TestPrettyPrintServiceList(t *testing.T) {
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			re := regexp.MustCompile(tt.expected)
-			assert.True(t, re.MatchString(PrettyPrintServiceList(tt.services)))
+			require.True(t, re.MatchString(prettyPrintServiceList(tt.services)))
 		})
 	}
 }
diff --git a/internal/dataplane/parser/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
index 39a5caf2f..5c585b916 100644
--- a/internal/dataplane/parser/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
@@ -18,7 +18,7 @@ services:
     id: b23f135f-8d7e-54cb-96eb-f2579ef1608b
     name: grpcroute.default.grpcbin.example.com.0.0
     preserve_host: true
-    priority: 1713055227461631
+    priority: 26766487929087
     tags:
     - k8s-name:grpcbin
     - k8s-namespace:default
diff --git a/internal/dataplane/parser/testdata/golden/host-header-annotation-httproute/default_golden.yaml b/internal/dataplane/parser/testdata/golden/host-header-annotation-httproute/default_golden.yaml
new file mode 100644
index 000000000..1307534ae
--- /dev/null
+++ b/internal/dataplane/parser/testdata/golden/host-header-annotation-httproute/default_golden.yaml
@@ -0,0 +1,44 @@
+_format_version: "3.0"
+services:
+- connect_timeout: 60000
+  host: httproute.default.httpbin.0
+  id: de4fffcc-97bc-59c6-9ccd-f1fb29b93beb
+  name: httproute.default.httpbin.0
+  port: 80
+  protocol: http
+  read_timeout: 60000
+  retries: 5
+  routes:
+  - https_redirect_status_code: 426
+    id: 668cce91-3da3-590f-82c9-8296c5bb28e5
+    name: httproute.default.httpbin.0.0
+    path_handling: v0
+    paths:
+    - ~/httpbin$
+    - /httpbin/
+    preserve_host: true
+    protocols:
+    - http
+    - https
+    strip_path: true
+    tags:
+    - k8s-name:httpbin
+    - k8s-namespace:default
+    - k8s-kind:HTTPRoute
+    - k8s-group:gateway.networking.k8s.io
+    - k8s-version:v1
+  tags:
+  - k8s-name:httpbin
+  - k8s-namespace:default
+  - k8s-kind:Service
+  - k8s-version:v1
+  write_timeout: 60000
+upstreams:
+- algorithm: round-robin
+  host_header: httpbin.org
+  name: httproute.default.httpbin.0
+  tags:
+  - k8s-name:httpbin
+  - k8s-namespace:default
+  - k8s-kind:Service
+  - k8s-version:v1
diff --git a/internal/dataplane/parser/testdata/golden/host-header-annotation-httproute/in.yaml b/internal/dataplane/parser/testdata/golden/host-header-annotation-httproute/in.yaml
new file mode 100644
index 000000000..b4cf233e0
--- /dev/null
+++ b/internal/dataplane/parser/testdata/golden/host-header-annotation-httproute/in.yaml
@@ -0,0 +1,38 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    app: httpbin
+  name: httpbin
+  namespace: default
+  annotations:
+    konghq.com/host-header: "httpbin.org"
+spec:
+  ports:
+    - port: 80
+      protocol: TCP
+      targetPort: 80
+  selector:
+    app: httpbin
+  type: ClusterIP
+---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+  name: httpbin
+  namespace: default
+  annotations:
+    konghq.com/strip-path: "true"
+spec:
+  parentRefs:
+    - name: kong
+  rules:
+    - matches:
+        - path:
+            type: PathPrefix
+            value: /httpbin
+      backendRefs:
+        - name: httpbin
+          kind: Service
+          port: 80
diff --git a/internal/dataplane/parser/testdata/golden/httproute-example/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/httproute-example/expression-routes-on_golden.yaml
index 7a24a31e4..dcb6f9085 100644
--- a/internal/dataplane/parser/testdata/golden/httproute-example/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/httproute-example/expression-routes-on_golden.yaml
@@ -14,7 +14,7 @@ services:
     id: 91833860-2041-5eea-abf8-a1e85b7c64cf
     name: httproute.default.httproute-testing._.0.0
     preserve_host: true
-    priority: 2251808940752895
+    priority: 35184514699263
     strip_path: true
     tags:
     - k8s-name:httproute-testing
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-empty-path/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-empty-path/expression-routes-on_golden.yaml
index 587d489ad..73f599155 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-empty-path/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-empty-path/expression-routes-on_golden.yaml
@@ -15,7 +15,7 @@ services:
     id: 3eee2c18-8fcc-5661-8f84-5c89adfa404f
     name: foo-namespace.foo.foo-svc.example.com.80
     preserve_host: true
-    priority: 3382102062006273
+    priority: 57178899611649
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-multiple-ports-for-one-service/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-multiple-ports-for-one-service/expression-routes-on_golden.yaml
index 49dcc4f63..adadd78f1 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-multiple-ports-for-one-service/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-multiple-ports-for-one-service/expression-routes-on_golden.yaml
@@ -15,7 +15,7 @@ services:
     id: 2bd1c902-4d3c-50ed-9a78-f698cfa0cf37
     name: foo-namespace.foo.foo-svc.example.net.8000
     preserve_host: true
-    priority: 3382102062006273
+    priority: 57178899611649
     request_buffering: true
     response_buffering: true
     strip_path: false
@@ -48,7 +48,7 @@ services:
     id: 3eee2c18-8fcc-5661-8f84-5c89adfa404f
     name: foo-namespace.foo.foo-svc.example.com.80
     preserve_host: true
-    priority: 3382102062006273
+    priority: 57178899611649
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-ports-defined-by-name/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-ports-defined-by-name/expression-routes-on_golden.yaml
index 901675032..4bf47ec38 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-ports-defined-by-name/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-ports-defined-by-name/expression-routes-on_golden.yaml
@@ -15,7 +15,7 @@ services:
     id: 0673ae1f-4318-59dc-96e7-fc17c218022f
     name: foo-namespace.regex-prefix.foo-svc.example.com.http
     preserve_host: true
-    priority: 3382102062006273
+    priority: 57178899611649
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-regex-prefix-exact-rule/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-regex-prefix-exact-rule/expression-routes-on_golden.yaml
index 4c10d5d65..0aa89f9ff 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-regex-prefix-exact-rule/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-regex-prefix-exact-rule/expression-routes-on_golden.yaml
@@ -15,7 +15,7 @@ services:
     id: 3eee2c18-8fcc-5661-8f84-5c89adfa404f
     name: foo-namespace.foo.foo-svc.example.com.80
     preserve_host: true
-    priority: 3382102062071817
+    priority: 57178899677193
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-regex-prefixed-path/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-regex-prefixed-path/expression-routes-on_golden.yaml
index 1dabe3b27..64b609a8f 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-regex-prefixed-path/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-regex-prefixed-path/expression-routes-on_golden.yaml
@@ -15,7 +15,7 @@ services:
     id: 45f1e9e4-8096-5cf7-b8e0-c42f8b9b81a0
     name: foo-namespace.regex-prefix.foo-svc.example.com.80
     preserve_host: true
-    priority: 3382102062071818
+    priority: 57178899677194
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-rule-with-tls/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-rule-with-tls/expression-routes-on_golden.yaml
index 4a71ef703..395923441 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-rule-with-tls/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-rule-with-tls/expression-routes-on_golden.yaml
@@ -82,7 +82,7 @@ services:
     id: fc9a8135-0253-5631-b1e0-8712f796a4e2
     name: bar-namespace.ing-with-tls.foo-svc.example.com.80
     preserve_host: true
-    priority: 3382102062071809
+    priority: 57178899677185
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-single-service-in-multiple-ingresses/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-single-service-in-multiple-ingresses/expression-routes-on_golden.yaml
index 34fc4dfa1..364d354e4 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-single-service-in-multiple-ingresses/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-single-service-in-multiple-ingresses/expression-routes-on_golden.yaml
@@ -15,7 +15,7 @@ services:
     id: 3eee2c18-8fcc-5661-8f84-5c89adfa404f
     name: foo-namespace.foo.foo-svc.example.com.80
     preserve_host: true
-    priority: 3382102062071809
+    priority: 57178899677185
     request_buffering: true
     response_buffering: true
     strip_path: false
@@ -30,7 +30,7 @@ services:
     id: ab6b1505-ec86-5b04-9d39-a95a711564cc
     name: foo-namespace.foo-2.foo-svc.example.com.80
     preserve_host: true
-    priority: 3382102062071809
+    priority: 57178899677185
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-with-acme-like-path/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-with-acme-like-path/expression-routes-on_golden.yaml
index ef5d135da..83d338b29 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-with-acme-like-path/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-with-acme-like-path/expression-routes-on_golden.yaml
@@ -15,7 +15,7 @@ services:
     id: bba23c8e-3f3c-55dc-bfca-2fe19de118d1
     name: foo-namespace.foo.cert-manager-solver-pod.example.com.80
     preserve_host: true
-    priority: 3382102062006304
+    priority: 57178899611680
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/ingress-v1-with-default-backend/expression-routes-on_golden.yaml b/internal/dataplane/parser/testdata/golden/ingress-v1-with-default-backend/expression-routes-on_golden.yaml
index 3c9400012..e8eeda84e 100644
--- a/internal/dataplane/parser/testdata/golden/ingress-v1-with-default-backend/expression-routes-on_golden.yaml
+++ b/internal/dataplane/parser/testdata/golden/ingress-v1-with-default-backend/expression-routes-on_golden.yaml
@@ -15,7 +15,7 @@ services:
     id: 3eee2c18-8fcc-5661-8f84-5c89adfa404f
     name: foo-namespace.foo.foo-svc.example.com.80
     preserve_host: true
-    priority: 3382102062071809
+    priority: 57178899677185
     request_buffering: true
     response_buffering: true
     strip_path: false
diff --git a/internal/dataplane/parser/testdata/golden/kongupstreampolicy-httproute/default_golden.yaml b/internal/dataplane/parser/testdata/golden/kongupstreampolicy-httproute/default_golden.yaml
new file mode 100644
index 000000000..34855a460
--- /dev/null
+++ b/internal/dataplane/parser/testdata/golden/kongupstreampolicy-httproute/default_golden.yaml
@@ -0,0 +1,157 @@
+_format_version: "3.0"
+services:
+- connect_timeout: 60000
+  host: httproute.default.httpbin.1
+  id: bbf738ea-19e7-5277-8a48-36d555e5ae94
+  name: httproute.default.httpbin.1
+  port: 80
+  protocol: http
+  read_timeout: 60000
+  retries: 5
+  routes:
+  - https_redirect_status_code: 426
+    id: e148326c-f112-5b0d-9644-4219ccb43cb6
+    name: httproute.default.httpbin.1.0
+    path_handling: v0
+    paths:
+    - ~/httpbin-prod-only$
+    - /httpbin-prod-only/
+    preserve_host: true
+    protocols:
+    - http
+    - https
+    strip_path: true
+    tags:
+    - k8s-name:httpbin
+    - k8s-namespace:default
+    - k8s-kind:HTTPRoute
+    - k8s-group:gateway.networking.k8s.io
+    - k8s-version:v1
+  tags:
+  - k8s-name:httpbin-prod
+  - k8s-namespace:default
+  - k8s-kind:Service
+  - k8s-version:v1
+  write_timeout: 60000
+- connect_timeout: 60000
+  host: httproute.default.httpbin.0
+  id: de4fffcc-97bc-59c6-9ccd-f1fb29b93beb
+  name: httproute.default.httpbin.0
+  port: 80
+  protocol: http
+  read_timeout: 60000
+  retries: 5
+  routes:
+  - https_redirect_status_code: 426
+    id: 668cce91-3da3-590f-82c9-8296c5bb28e5
+    name: httproute.default.httpbin.0.0
+    path_handling: v0
+    paths:
+    - ~/httpbin-with-test$
+    - /httpbin-with-test/
+    preserve_host: true
+    protocols:
+    - http
+    - https
+    strip_path: true
+    tags:
+    - k8s-name:httpbin
+    - k8s-namespace:default
+    - k8s-kind:HTTPRoute
+    - k8s-group:gateway.networking.k8s.io
+    - k8s-version:v1
+  tags:
+  - k8s-name:httpbin
+  - k8s-namespace:default
+  - k8s-kind:HTTPRoute
+  - k8s-group:gateway.networking.k8s.io
+  - k8s-version:v1
+  write_timeout: 60000
+upstreams:
+- algorithm: consistent-hashing
+  hash_fallback: consumer
+  hash_on: header
+  hash_on_header: session-id
+  healthchecks:
+    active:
+      concurrency: 20
+      headers:
+        X-Health-Check:
+        - kong
+        - dataplane
+      healthy:
+        http_statuses:
+        - 200
+        - 302
+        interval: 5
+        successes: 5
+      http_path: /status
+      https_sni: example.com
+      https_verify_certificate: false
+      timeout: 15
+      type: http
+      unhealthy:
+        http_failures: 5
+        http_statuses:
+        - 400
+        - 500
+        interval: 10
+        timeouts: 5
+    passive:
+      healthy:
+        successes: 5
+      type: tcp
+      unhealthy:
+        tcp_failures: 5
+        timeouts: 10
+  name: httproute.default.httpbin.1
+  slots: 100
+  tags:
+  - k8s-name:httpbin-prod
+  - k8s-namespace:default
+  - k8s-kind:Service
+  - k8s-version:v1
+- algorithm: consistent-hashing
+  hash_fallback: consumer
+  hash_on: header
+  hash_on_header: session-id
+  healthchecks:
+    active:
+      concurrency: 20
+      headers:
+        X-Health-Check:
+        - kong
+        - dataplane
+      healthy:
+        http_statuses:
+        - 200
+        - 302
+        interval: 5
+        successes: 5
+      http_path: /status
+      https_sni: example.com
+      https_verify_certificate: false
+      timeout: 15
+      type: http
+      unhealthy:
+        http_failures: 5
+        http_statuses:
+        - 400
+        - 500
+        interval: 10
+        timeouts: 5
+    passive:
+      healthy:
+        successes: 5
+      type: tcp
+      unhealthy:
+        tcp_failures: 5
+        timeouts: 10
+  name: httproute.default.httpbin.0
+  slots: 100
+  tags:
+  - k8s-name:httpbin
+  - k8s-namespace:default
+  - k8s-kind:HTTPRoute
+  - k8s-group:gateway.networking.k8s.io
+  - k8s-version:v1
diff --git a/internal/dataplane/parser/testdata/golden/kongupstreampolicy-httproute/in.yaml b/internal/dataplane/parser/testdata/golden/kongupstreampolicy-httproute/in.yaml
new file mode 100644
index 000000000..ad4952f54
--- /dev/null
+++ b/internal/dataplane/parser/testdata/golden/kongupstreampolicy-httproute/in.yaml
@@ -0,0 +1,113 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    app: httpbin-prod
+  name: httpbin-prod
+  namespace: default
+  annotations:
+    konghq.com/upstream-policy: httpbin-httproute
+spec:
+  ports:
+    - port: 80
+      protocol: TCP
+      targetPort: 80
+  selector:
+    app: httpbin-prod
+  type: ClusterIP
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    app: httpbin-test
+  name: httpbin-test
+  namespace: default
+  annotations:
+    konghq.com/upstream-policy: httpbin-httproute
+spec:
+  ports:
+    - port: 80
+      protocol: TCP
+      targetPort: 80
+  selector:
+    app: httpbin
+  type: ClusterIP
+---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+  name: httpbin
+  namespace: default
+  annotations:
+    konghq.com/strip-path: "true"
+spec:
+  parentRefs:
+    - name: kong
+  rules:
+    - matches: # This rule will generate an Upstream with the name `httproute.default.httpbin.0`.
+        - path:
+            type: PathPrefix
+            value: /httpbin-with-test
+      backendRefs:
+        - name: httpbin-prod
+          kind: Service
+          port: 80
+          weight: 75
+        - name: httpbin-test
+          kind: Service
+          port: 80
+          weight: 25
+    - matches: # This rule will generate an Upstream with the name `httproute.default.httpbin.1`.
+        - path:
+            type: PathPrefix
+            value: /httpbin-prod-only
+      backendRefs:
+        - name: httpbin-prod
+          kind: Service
+          port: 80
+---
+# This policy will be applied to all Upstreams generated for the annotated Services:
+# - httproute.default.httpbin.0
+# - httproute.default.httpbin.1
+apiVersion: configuration.konghq.com/v1beta1
+kind: KongUpstreamPolicy
+metadata:
+  name: httpbin-httproute
+  namespace: default
+spec:
+  algorithm: consistent-hashing
+  slots: 100
+  hashOn:
+    header: session-id
+  hashOnFallback:
+    input: consumer
+  healthchecks:
+    active:
+      type: http
+      httpPath: /status
+      httpsSni: example.com
+      httpsVerifyCertificate: false
+      concurrency: 20
+      timeout: 15
+      headers:
+        X-Health-Check:
+          - kong
+          - dataplane
+      healthy:
+        httpStatuses: [200, 302]
+        interval: 5
+        successes: 5
+      unhealthy:
+        httpStatuses: [400, 500]
+        httpFailures: 5
+        timeouts: 5
+        interval: 10
+    passive:
+      type: tcp
+      healthy:
+        successes: 5
+      unhealthy:
+        tcpFailures: 5
+        timeouts: 10
diff --git a/internal/dataplane/parser/testdata/golden/kongupstreampolicy-ingress/default_golden.yaml b/internal/dataplane/parser/testdata/golden/kongupstreampolicy-ingress/default_golden.yaml
new file mode 100644
index 000000000..2840f892a
--- /dev/null
+++ b/internal/dataplane/parser/testdata/golden/kongupstreampolicy-ingress/default_golden.yaml
@@ -0,0 +1,162 @@
+_format_version: "3.0"
+services:
+- connect_timeout: 60000
+  host: httpbin-deployment.default.8080.svc
+  id: 47865164-d8b1-5729-86bc-bfba3880819e
+  name: default.httpbin-deployment.8080
+  path: /
+  port: 8080
+  protocol: http
+  read_timeout: 60000
+  retries: 5
+  routes:
+  - https_redirect_status_code: 426
+    id: e8dddcdf-ebe3-5a2d-af0d-865da7881126
+    name: default.httpbin-ingress-1.httpbin-deployment..8080
+    path_handling: v0
+    paths:
+    - /internal/
+    - ~/internal$
+    preserve_host: true
+    protocols:
+    - http
+    - https
+    regex_priority: 0
+    request_buffering: true
+    response_buffering: true
+    strip_path: true
+    tags:
+    - k8s-name:httpbin-ingress-1
+    - k8s-namespace:default
+    - k8s-kind:Ingress
+    - k8s-group:networking.k8s.io
+    - k8s-version:v1
+  tags:
+  - k8s-name:httpbin-deployment
+  - k8s-namespace:default
+  - k8s-kind:Service
+  - k8s-version:v1
+  write_timeout: 60000
+- connect_timeout: 60000
+  host: httpbin-deployment.default.80.svc
+  id: 41e0df75-8715-5017-b56b-431e70ccd2ff
+  name: default.httpbin-deployment.80
+  path: /
+  port: 80
+  protocol: http
+  read_timeout: 60000
+  retries: 5
+  routes:
+  - https_redirect_status_code: 426
+    id: 94a0d72e-a19f-5269-a40c-6e52a6120d34
+    name: default.httpbin-ingress-1.httpbin-deployment..80
+    path_handling: v0
+    paths:
+    - /
+    preserve_host: true
+    protocols:
+    - http
+    - https
+    regex_priority: 0
+    request_buffering: true
+    response_buffering: true
+    strip_path: true
+    tags:
+    - k8s-name:httpbin-ingress-1
+    - k8s-namespace:default
+    - k8s-kind:Ingress
+    - k8s-group:networking.k8s.io
+    - k8s-version:v1
+  tags:
+  - k8s-name:httpbin-deployment
+  - k8s-namespace:default
+  - k8s-kind:Service
+  - k8s-version:v1
+  write_timeout: 60000
+upstreams:
+- algorithm: consistent-hashing
+  hash_fallback: consumer
+  hash_on: header
+  hash_on_header: session-id
+  healthchecks:
+    active:
+      concurrency: 20
+      headers:
+        X-Health-Check:
+        - kong
+        - dataplane
+      healthy:
+        http_statuses:
+        - 200
+        - 302
+        interval: 5
+        successes: 5
+      http_path: /status
+      https_sni: example.com
+      https_verify_certificate: false
+      timeout: 15
+      type: http
+      unhealthy:
+        http_failures: 5
+        http_statuses:
+        - 400
+        - 500
+        interval: 10
+        timeouts: 5
+    passive:
+      healthy:
+        successes: 5
+      type: tcp
+      unhealthy:
+        tcp_failures: 5
+        timeouts: 10
+  name: httpbin-deployment.default.8080.svc
+  slots: 100
+  tags:
+  - k8s-name:httpbin-deployment
+  - k8s-namespace:default
+  - k8s-kind:Service
+  - k8s-version:v1
+- algorithm: consistent-hashing
+  hash_fallback: consumer
+  hash_on: header
+  hash_on_header: session-id
+  healthchecks:
+    active:
+      concurrency: 20
+      headers:
+        X-Health-Check:
+        - kong
+        - dataplane
+      healthy:
+        http_statuses:
+        - 200
+        - 302
+        interval: 5
+        successes: 5
+      http_path: /status
+      https_sni: example.com
+      https_verify_certificate: false
+      timeout: 15
+      type: http
+      unhealthy:
+        http_failures: 5
+        http_statuses:
+        - 400
+        - 500
+        interval: 10
+        timeouts: 5
+    passive:
+      healthy:
+        successes: 5
+      type: tcp
+      unhealthy:
+        tcp_failures: 5
+        timeouts: 10
+  name: httpbin-deployment.default.80.svc
+  slots: 100
+  tags:
+  - k8s-name:httpbin-deployment
+  - k8s-namespace:default
+  - k8s-kind:Service
+  - k8s-version:v1
diff --git a/internal/dataplane/parser/testdata/golden/kongupstreampolicy-ingress/in.yaml b/internal/dataplane/parser/testdata/golden/kongupstreampolicy-ingress/in.yaml
new file mode 100644
index 000000000..1b62fac40
--- /dev/null
+++ b/internal/dataplane/parser/testdata/golden/kongupstreampolicy-ingress/in.yaml
@@ -0,0 +1,93 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    app: httpbin
+  name: httpbin-deployment
+  namespace: default
+  annotations:
+    konghq.com/upstream-policy: httpbin
+spec:
+  ports:
+    - port: 80
+      protocol: TCP
+      targetPort: 80
+    - port: 8080
+      protocol: TCP
+      targetPort: 8080
+  selector:
+    app: httpbin
+  type: ClusterIP
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: httpbin-ingress-1
+  namespace: default
+  annotations:
+    konghq.com/strip-path: "true"
+spec:
+  ingressClassName: kong
+  rules:
+    - http:
+        paths:
+          - path: /
+            pathType: Prefix
+            backend:
+              service:
+                name: httpbin-deployment
+                port:
+                  number: 80
+          - path: /internal
+            pathType: Prefix
+            backend:
+              service:
+                name: httpbin-deployment
+                port:
+                  number: 8080
+---
+# This policy will be applied to every Upstream that is created for
+# the httpbin-deployment Service. In this case those would be following Kong Upstreams:
+# - default.httpbin-deployment.80
+# - default.httpbin-deployment.8080
+apiVersion: configuration.konghq.com/v1beta1
+kind: KongUpstreamPolicy
+metadata:
+  name: httpbin
+  namespace: default
+spec:
+  algorithm: consistent-hashing
+  slots: 100
+  hashOn:
+    header: session-id
+  hashOnFallback:
+    input: consumer
+  healthchecks:
+    active:
+      type: http
+      httpPath: /status
+      httpsSni: example.com
+      httpsVerifyCertificate: false
+      concurrency: 20
+      timeout: 15
+      headers:
+        X-Health-Check:
+          - kong
+          - dataplane
+      healthy:
+        httpStatuses: [200, 302]
+        interval: 5
+        successes: 5
+      unhealthy:
+        httpStatuses: [400, 500]
+        httpFailures: 5
+        timeouts: 5
+        interval: 10
+    passive:
+      type: tcp
+      healthy:
+        successes: 5
+      unhealthy:
+        tcpFailures: 5
+        timeouts: 10
diff --git a/internal/dataplane/parser/translate_httproute_test.go b/internal/dataplane/parser/translate_httproute_test.go
index 183648a6c..8cdd4177f 100644
--- a/internal/dataplane/parser/translate_httproute_test.go
+++ b/internal/dataplane/parser/translate_httproute_test.go
@@ -1887,7 +1887,7 @@ func TestIngressRulesFromSplitHTTPRouteMatchWithPriority(t *testing.T) {
 					Expression:   kong.String(`http.path == "/v1/foo"`),
 					PreserveHost: kong.Bool(true),
 					StripPath:    kong.Bool(false),
-					Priority:     kong.Int(1024),
+					Priority:     kong.Uint64(1024),
 				},
 				ExpressionRoutes: true,
 			},
@@ -1949,7 +1949,7 @@ func TestIngressRulesFromSplitHTTPRouteMatchWithPriority(t *testing.T) {
 					Expression:   kong.String(`(http.host == "foo.com") && (http.path == "/v1/foo")`),
 					PreserveHost: kong.Bool(true),
 					StripPath:    kong.Bool(false),
-					Priority:     kong.Int(1024),
+					Priority:     kong.Uint64(1024),
 				},
 				Plugins: []kong.Plugin{
 					{
@@ -2027,7 +2027,7 @@ func TestIngressRulesFromSplitHTTPRouteMatchWithPriority(t *testing.T) {
 					Expression:   kong.String(`(http.host =^ ".foo.com") && (http.path == "/v1/foo")`),
 					PreserveHost: kong.Bool(true),
 					StripPath:    kong.Bool(false),
-					Priority:     kong.Int(1024),
+					Priority:     kong.Uint64(1024),
 				},
 				ExpressionRoutes: true,
 			},
@@ -2079,7 +2079,7 @@ func TestIngressRulesFromSplitHTTPRouteMatchWithPriority(t *testing.T) {
 					Expression:   kong.String(`http.host == "a.foo.com"`),
 					PreserveHost: kong.Bool(true),
 					StripPath:    kong.Bool(false),
-					Priority:     kong.Int(1024),
+					Priority:     kong.Uint64(1024),
 				},
 				ExpressionRoutes: true,
 			},
@@ -2128,7 +2128,7 @@ func TestIngressRulesFromSplitHTTPRouteMatchWithPriority(t *testing.T) {
 					Expression:   kong.String(translators.CatchAllHTTPExpression),
 					PreserveHost: kong.Bool(true),
 					StripPath:    kong.Bool(false),
-					Priority:     kong.Int(1024),
+					Priority:     kong.Uint64(1024),
 				},
 				ExpressionRoutes: true,
 			},
diff --git a/internal/dataplane/parser/translators/grpcroute_atc_test.go b/internal/dataplane/parser/translators/grpcroute_atc_test.go
index 51f0582eb..0392ffe66 100644
--- a/internal/dataplane/parser/translators/grpcroute_atc_test.go
+++ b/internal/dataplane/parser/translators/grpcroute_atc_test.go
@@ -58,7 +58,7 @@ func TestGenerateKongExpressionRoutesFromGRPCRouteRule(t *testing.T) {
 					Route: kong.Route{
 						Name:       kong.String("grpcroute.default.single-match.0.0"),
 						Expression: kong.String(`(http.path ^= "/service0/") && (http.headers.x_foo == "Bar")`),
-						Priority:   kong.Int(1),
+						Priority:   kong.Uint64(1),
 					},
 				},
 			},
@@ -90,7 +90,7 @@ func TestGenerateKongExpressionRoutesFromGRPCRouteRule(t *testing.T) {
 					Route: kong.Route{
 						Name:       kong.String("grpcroute.default.single-match-with-hostname.0.0"),
 						Expression: kong.String(`(http.path == "/service0/method0") && ((http.host == "foo.com") || (http.host =^ ".foo.com"))`),
-						Priority:   kong.Int(1),
+						Priority:   kong.Uint64(1),
 					},
 				},
 			},
@@ -135,7 +135,7 @@ func TestGenerateKongExpressionRoutesFromGRPCRouteRule(t *testing.T) {
 					Route: kong.Route{
 						Name:       kong.String("grpcroute.default.multiple-matches.0.0"),
 						Expression: kong.String(`(http.path =^ "/method0") && ((http.headers.client == "kong-test") && (http.headers.version == "2"))`),
-						Priority:   kong.Int(1),
+						Priority:   kong.Uint64(1),
 					},
 				},
 				{
@@ -148,7 +148,7 @@ func TestGenerateKongExpressionRoutesFromGRPCRouteRule(t *testing.T) {
 					Route: kong.Route{
 						Name:       kong.String("grpcroute.default.multiple-matches.0.1"),
 						Expression: kong.String(`http.path ~ "^/v[012]/.+"`),
-						Priority:   kong.Int(1),
+						Priority:   kong.Uint64(1),
 					},
 				},
 			},
@@ -187,7 +187,7 @@ func TestGenerateKongExpressionRoutesFromGRPCRouteRule(t *testing.T) {
 					Route: kong.Route{
 						Name:       kong.String("grpcroute.default.single-match-with-annotations.0.0"),
 						Expression: kong.String(`(http.path == "/service0/method0") && (tls.sni == "kong.foo.com")`),
-						Priority:   kong.Int(1),
+						Priority:   kong.Uint64(1),
 					},
 				},
 			},
@@ -210,7 +210,7 @@ func TestGenerateKongExpressionRoutesFromGRPCRouteRule(t *testing.T) {
 					Route: kong.Route{
 						Name:       kong.String("grpcroute.default.hostname-only.0.0"),
 						Expression: kong.String(`http.host == "foo.com"`),
-						Priority:   kong.Int(1),
+						Priority:   kong.Uint64(1),
 					},
 				},
 			},
@@ -604,7 +604,7 @@ func TestGRPCRouteTraitsEncodeToPriority(t *testing.T) {
 	testCases := []struct {
 		name              string
 		traits            GRPCRoutePriorityTraits
-		exprectedPriority int
+		exprectedPriority RoutePriorityType
 	}{
 		{
 			name: "precise hostname",
@@ -613,7 +613,7 @@ func TestGRPCRouteTraitsEncodeToPriority(t *testing.T) {
 				HostnameLength:  15,
 				ServiceLength:   7,
 			},
-			exprectedPriority: (1 << 50) | (1 << 49) | (15 << 41) | (7 << 30),
+			exprectedPriority: (1 << 44) | (1 << 43) | (15 << 35) | (7 << 24),
 		},
 		{
 			name: "non precise hostname",
@@ -624,7 +624,7 @@ func TestGRPCRouteTraitsEncodeToPriority(t *testing.T) {
 				MethodLength:    7,
 				HeaderCount:     3,
 			},
-			exprectedPriority: (1 << 50) | (15 << 41) | (7 << 30) | (7 << 19) | (3 << 14),
+			exprectedPriority: (1 << 44) | (15 << 35) | (7 << 24) | (7 << 13) | (3 << 8),
 		},
 	}
 
@@ -647,13 +647,13 @@ func TestAssignRoutePriorityToSplitGRPCRouteMatches(t *testing.T) {
 		matchIndex int
 	}
 	now := time.Now()
-	const maxRelativeOrderPriorityBits = (1 << 14) - 1
+	const maxRelativeOrderPriorityBits = (1 << 8) - 1
 
 	testCases := []struct {
 		name                  string
 		splitGRPCRouteMatches []SplitGRPCRouteMatch
 		// GRPCRoute index -> priority
-		priorities map[splitGRPCRouteMatchIndex]int
+		priorities map[splitGRPCRouteMatchIndex]RoutePriorityType
 	}{
 		{
 			name: "no dupelicated fixed priority",
@@ -723,7 +723,7 @@ func TestAssignRoutePriorityToSplitGRPCRouteMatches(t *testing.T) {
 					},
 				},
 			},
-			priorities: map[splitGRPCRouteMatchIndex]int{
+			priorities: map[splitGRPCRouteMatchIndex]RoutePriorityType{
 				{
 					namespace:  "default",
 					name:       "grpcroute-1",
@@ -821,7 +821,7 @@ func TestAssignRoutePriorityToSplitGRPCRouteMatches(t *testing.T) {
 					MatchIndex: 0,
 				},
 			},
-			priorities: map[splitGRPCRouteMatchIndex]int{
+			priorities: map[splitGRPCRouteMatchIndex]RoutePriorityType{
 				{
 					namespace:  "default",
 					name:       "grpcroute-1",
@@ -918,7 +918,7 @@ func TestAssignRoutePriorityToSplitGRPCRouteMatches(t *testing.T) {
 					MatchIndex: 0,
 				},
 			},
-			priorities: map[splitGRPCRouteMatchIndex]int{
+			priorities: map[splitGRPCRouteMatchIndex]RoutePriorityType{
 				{
 					namespace:  "default",
 					name:       "grpcroute-1",
@@ -1028,7 +1028,7 @@ func TestAssignRoutePriorityToSplitGRPCRouteMatches(t *testing.T) {
 					MatchIndex: 1,
 				},
 			},
-			priorities: map[splitGRPCRouteMatchIndex]int{
+			priorities: map[splitGRPCRouteMatchIndex]RoutePriorityType{
 				{
 					namespace:  "default",
 					name:       "grpcroute-1",
@@ -1123,14 +1123,14 @@ func TestKongExpressionRouteFromSplitGRPCRouteWithPriority(t *testing.T) {
 					RuleIndex:  0,
 					MatchIndex: 0,
 				},
-				Priority: 1024,
+				Priority: 1 << 14,
 			},
 			expectedRoute: kongstate.Route{
 				Route: kong.Route{
 					Name:         kong.String("grpcroute.default.no-hostname-exact-method._.0.0"),
 					PreserveHost: kong.Bool(true),
 					Expression:   kong.String(`http.path == "/pets/list"`),
-					Priority:     kong.Int(1024),
+					Priority:     kong.Uint64(1 << 14),
 				},
 				ExpressionRoutes: true,
 			},
@@ -1175,14 +1175,14 @@ func TestKongExpressionRouteFromSplitGRPCRouteWithPriority(t *testing.T) {
 					RuleIndex:  0,
 					MatchIndex: 0,
 				},
-				Priority: 1024,
+				Priority: (1 << 31) + 1,
 			},
 			expectedRoute: kongstate.Route{
 				Route: kong.Route{
 					Name:         kong.String("grpcroute.default.precise-hostname-regex-method.foo.com.0.0"),
 					Expression:   kong.String(`(http.path ~ "^/name/[a-z0-9]+") && (http.host == "foo.com")`),
 					PreserveHost: kong.Bool(true),
-					Priority:     kong.Int(1024),
+					Priority:     kong.Uint64((1 << 31) + 1),
 				},
 				ExpressionRoutes: true,
 			},
@@ -1244,14 +1244,14 @@ func TestKongExpressionRouteFromSplitGRPCRouteWithPriority(t *testing.T) {
 					RuleIndex:  0,
 					MatchIndex: 1,
 				},
-				Priority: 1024,
+				Priority: (1 << 42) + 1,
 			},
 			expectedRoute: kongstate.Route{
 				Route: kong.Route{
 					Name:         kong.String("grpcroute.default.wildcard-hostname-header-match._.foo.com.0.1"),
 					Expression:   kong.String(`(http.path ^= "/name/") && (http.headers.foo == "bar") && (http.host =^ ".foo.com")`),
 					PreserveHost: kong.Bool(true),
-					Priority:     kong.Int(1024),
+					Priority:     kong.Uint64((1 << 42) + 1),
 				},
 				ExpressionRoutes: true,
 			},
diff --git a/internal/dataplane/parser/translators/httproute_atc_test.go b/internal/dataplane/parser/translators/httproute_atc_test.go
index 3e5b1b35e..2bd9cd545 100644
--- a/internal/dataplane/parser/translators/httproute_atc_test.go
+++ b/internal/dataplane/parser/translators/httproute_atc_test.go
@@ -57,7 +57,7 @@ func TestGenerateKongExpressionRoutesFromHTTPRouteMatches(t *testing.T) {
 						PreserveHost: kong.Bool(true),
 						StripPath:    kong.Bool(false),
 						Expression:   kong.String(`(http.host == "foo.com") || (http.host =^ ".bar.com")`),
-						Priority:     kong.Int(1),
+						Priority:     kong.Uint64(1),
 					},
 					ExpressionRoutes: true,
 				},
@@ -77,7 +77,7 @@ func TestGenerateKongExpressionRoutesFromHTTPRouteMatches(t *testing.T) {
 						PreserveHost: kong.Bool(true),
 						StripPath:    kong.Bool(false),
 						Expression:   kong.String(`(http.path == "/prefix") || (http.path ^= "/prefix/")`),
-						Priority:     kong.Int(1),
+						Priority:     kong.Uint64(1),
 					},
 					ExpressionRoutes: true,
 				},
@@ -98,7 +98,7 @@ func TestGenerateKongExpressionRoutesFromHTTPRouteMatches(t *testing.T) {
 						PreserveHost: kong.Bool(true),
 						StripPath:    kong.Bool(false),
 						Expression:   kong.String(`((http.path == "/prefix") || (http.path ^= "/prefix/")) || ((http.path == "/exact") && (http.method == "GET"))`),
-						Priority:     kong.Int(1),
+						Priority:     kong.Uint64(1),
 					},
 					ExpressionRoutes: true,
 				},
@@ -126,7 +126,7 @@ func TestGenerateKongExpressionRoutesFromHTTPRouteMatches(t *testing.T) {
 						PreserveHost: kong.Bool(true),
 						StripPath:    kong.Bool(false),
 						Expression:   kong.String(`http.path == "/exact/0"`),
-						Priority:     kong.Int(1),
+						Priority:     kong.Uint64(1),
 					},
 					Plugins: []kong.Plugin{
 						{
@@ -152,7 +152,7 @@ func TestGenerateKongExpressionRoutesFromHTTPRouteMatches(t *testing.T) {
 						PreserveHost: kong.Bool(true),
 						StripPath:    kong.Bool(false),
 						Expression:   kong.String(`http.path == "/exact/1"`),
-						Priority:     kong.Int(1),
+						Priority:     kong.Uint64(1),
 					},
 					Plugins: []kong.Plugin{
 						{
@@ -194,7 +194,7 @@ func TestGenerateKongExpressionRoutesFromHTTPRouteMatches(t *testing.T) {
 						PreserveHost: kong.Bool(true),
 						StripPath:    kong.Bool(false),
 						Expression:   kong.String(`(http.path == "/exact/0") || (http.path ~ "^/regex/[a-z]+")`),
-						Priority:     kong.Int(1),
+						Priority:     kong.Uint64(1),
 					},
 					Plugins: []kong.Plugin{
 						{
@@ -240,7 +240,7 @@ func TestGenerateKongExpressionRoutesFromHTTPRouteMatches(t *testing.T) {
 						PreserveHost: kong.Bool(true),
 						StripPath:    kong.Bool(false),
 						Expression:   kong.String(`((http.path == "/prefix/0") || (http.path ^= "/prefix/0/")) && (http.host == "a.foo.com") && (tls.sni == "a.foo.com")`),
-						Priority:     kong.Int(1),
+						Priority:     kong.Uint64(1),
 					},
 					ExpressionRoutes: true,
 				},
@@ -473,7 +473,7 @@ func TestEncodeHTTPRoutePriorityFromTraits(t *testing.T) {
 	testCases := []struct {
 		name             string
 		traits           HTTPRoutePriorityTraits
-		expectedPriority int
+		expectedPriority RoutePriorityType
 	}{
 		{
 			name: "precise hostname and exact path",
@@ -483,7 +483,7 @@ func TestEncodeHTTPRoutePriorityFromTraits(t *testing.T) {
 				PathType:        gatewayapi.PathMatchExact,
 				PathLength:      4,
 			},
-			expectedPriority: (2 << 50) | (1 << 49) | (7 << 41) | (1 << 40) | (3 << 29),
+			expectedPriority: (2 << 44) | (1 << 43) | (7 << 35) | (1 << 34) | (3 << 23),
 		},
 		{
 			name: "wildcard hostname and prefix path",
@@ -493,7 +493,7 @@ func TestEncodeHTTPRoutePriorityFromTraits(t *testing.T) {
 				PathType:        gatewayapi.PathMatchPathPrefix,
 				PathLength:      5,
 			},
-			expectedPriority: (2 << 50) | (7 << 41) | (4 << 29),
+			expectedPriority: (2 << 44) | (7 << 35) | (4 << 23),
 		},
 		{
 			name: "no hostname and regex path, with header matches",
@@ -502,7 +502,7 @@ func TestEncodeHTTPRoutePriorityFromTraits(t *testing.T) {
 				PathLength:  5,
 				HeaderCount: 2,
 			},
-			expectedPriority: (2 << 50) | (1 << 39) | (4 << 29) | (2 << 23),
+			expectedPriority: (2 << 44) | (1 << 33) | (4 << 23) | (2 << 17),
 		},
 		{
 			name: "no hostname and exact path, with method match and query parameter matches",
@@ -512,7 +512,7 @@ func TestEncodeHTTPRoutePriorityFromTraits(t *testing.T) {
 				HasMethodMatch:  true,
 				QueryParamCount: 1,
 			},
-			expectedPriority: (2 << 50) | (1 << 40) | (4 << 29) | (1 << 28) | (1 << 18),
+			expectedPriority: (2 << 44) | (1 << 34) | (4 << 23) | (1 << 22) | (1 << 12),
 		},
 	}
 
@@ -733,13 +733,13 @@ func TestAssignRoutePriorityToSplitHTTPRouteMatches(t *testing.T) {
 		matchIndex int
 	}
 	now := time.Now()
-	const maxRelativeOrderPriorityBits = (1 << 18) - 1
+	const maxRelativeOrderPriorityBits = (1 << 12) - 1
 
 	testCases := []struct {
 		name    string
 		matches []SplitHTTPRouteMatch
 		// HTTPRoute index -> priority
-		priorities map[splitHTTPRouteIndex]int
+		priorities map[splitHTTPRouteIndex]RoutePriorityType
 	}{
 		{
 			name: "no dupelicated fixed priority",
@@ -787,7 +787,7 @@ func TestAssignRoutePriorityToSplitHTTPRouteMatches(t *testing.T) {
 					MatchIndex: 0,
 				},
 			},
-			priorities: map[splitHTTPRouteIndex]int{
+			priorities: map[splitHTTPRouteIndex]RoutePriorityType{
 				{
 					namespace:  "default",
 					name:       "httproute-1",
@@ -864,7 +864,7 @@ func TestAssignRoutePriorityToSplitHTTPRouteMatches(t *testing.T) {
 					MatchIndex: 0,
 				},
 			},
-			priorities: map[splitHTTPRouteIndex]int{
+			priorities: map[splitHTTPRouteIndex]RoutePriorityType{
 				{
 					namespace:  "default",
 					name:       "httproute-1",
@@ -937,7 +937,7 @@ func TestAssignRoutePriorityToSplitHTTPRouteMatches(t *testing.T) {
 					MatchIndex: 0,
 				},
 			},
-			priorities: map[splitHTTPRouteIndex]int{
+			priorities: map[splitHTTPRouteIndex]RoutePriorityType{
 				{
 					namespace:  "default",
 					name:       "httproute-1",
@@ -1016,7 +1016,7 @@ func TestAssignRoutePriorityToSplitHTTPRouteMatches(t *testing.T) {
 					MatchIndex: 1,
 				},
 			},
-			priorities: map[splitHTTPRouteIndex]int{
+			priorities: map[splitHTTPRouteIndex]RoutePriorityType{
 				{
 					namespace:  "default",
 					name:       "httproute-1",
diff --git a/internal/dataplane/parser/translators/ingress_atc_test.go b/internal/dataplane/parser/translators/ingress_atc_test.go
index 5c0885a19..7ffa0d207 100644
--- a/internal/dataplane/parser/translators/ingress_atc_test.go
+++ b/internal/dataplane/parser/translators/ingress_atc_test.go
@@ -74,7 +74,7 @@ func TestTranslateIngressATC(t *testing.T) {
 						Route: kong.Route{
 							Name:       kong.String("default.test-ingress.test-service.konghq.com.80"),
 							Expression: kong.String(`(http.host == "konghq.com") && ((http.path == "/api") || (http.path ^= "/api/"))`),
-							Priority: kong.Int(IngressRoutePriorityTraits{
+							Priority: kong.Uint64(IngressRoutePriorityTraits{
 								MatchFields:   2,
 								PlainHostOnly: true,
 								MaxPathLength: 5,
@@ -151,7 +151,7 @@ func TestTranslateIngressATC(t *testing.T) {
 						Route: kong.Route{
 							Name:       kong.String("default.test-ingress.test-service.konghq.com.80"),
 							Expression: kong.String(`(http.host == "konghq.com") && (http.path ^= "/api/")`),
-							Priority: kong.Int(IngressRoutePriorityTraits{
+							Priority: kong.Uint64(IngressRoutePriorityTraits{
 								MatchFields:   2,
 								PlainHostOnly: true,
 								MaxPathLength: 5,
@@ -238,7 +238,7 @@ func TestTranslateIngressATC(t *testing.T) {
 						Route: kong.Route{
 							Name:       kong.String("default.test-ingress-annotations.test-service.konghq.com.80"),
 							Expression: kong.String(`(http.host == "konghq.com") && (http.path ^= "/api/") && (http.headers.foo == "bar") && (http.method == "GET")`),
-							Priority: kong.Int(IngressRoutePriorityTraits{
+							Priority: kong.Uint64(IngressRoutePriorityTraits{
 								MatchFields:   4,
 								PlainHostOnly: true,
 								MaxPathLength: 5,
@@ -420,7 +420,7 @@ func TestEncodeIngressRoutePriorityFromTraits(t *testing.T) {
 	testCases := []struct {
 		name             string
 		traits           IngressRoutePriorityTraits
-		expectedPriority int
+		expectedPriority RoutePriorityType
 	}{
 		{
 			name: "plain host true regex path false",
@@ -430,7 +430,7 @@ func TestEncodeIngressRoutePriorityFromTraits(t *testing.T) {
 				MaxPathLength: 5,
 				HasRegexPath:  false,
 			},
-			expectedPriority: (3 << 50) | (2 << 41) | (1 << 32) | 5,
+			expectedPriority: (3 << 44) | (2 << 41) | (1 << 32) | 5,
 		},
 		{
 			name: "plain host false regex path true",
@@ -441,7 +441,7 @@ func TestEncodeIngressRoutePriorityFromTraits(t *testing.T) {
 				MaxPathLength: 5,
 				HasRegexPath:  true,
 			},
-			expectedPriority: (3 << 50) | (2 << 41) | (2 << 33) | (1 << 16) | 5,
+			expectedPriority: (3 << 44) | (2 << 41) | (2 << 33) | (1 << 16) | 5,
 		},
 		{
 			name: "header number exceed limit",
@@ -452,7 +452,7 @@ func TestEncodeIngressRoutePriorityFromTraits(t *testing.T) {
 				MaxPathLength: 5,
 				HasRegexPath:  true,
 			},
-			expectedPriority: (3 << 50) | (2 << 41) | (255 << 33) | (1 << 16) | 5,
+			expectedPriority: (3 << 44) | (2 << 41) | (255 << 33) | (1 << 16) | 5,
 		},
 		{
 			name: "path length exceed limit",
@@ -461,7 +461,7 @@ func TestEncodeIngressRoutePriorityFromTraits(t *testing.T) {
 				PlainHostOnly: true,
 				MaxPathLength: 100000,
 			},
-			expectedPriority: (3 << 50) | (2 << 41) | (1 << 32) | 65535,
+			expectedPriority: (3 << 44) | (2 << 41) | (1 << 32) | 65535,
 		},
 	}
 
diff --git a/test/e2e/manifests/all-in-one-dbless-k4k8s-enterprise.yaml b/test/e2e/manifests/all-in-one-dbless-k4k8s-enterprise.yaml
index 1e3963c03..f2aaa8e8f 100644
--- a/test/e2e/manifests/all-in-one-dbless-k4k8s-enterprise.yaml
+++ b/test/e2e/manifests/all-in-one-dbless-k4k8s-enterprise.yaml
@@ -1528,10 +1528,6 @@ spec:
                       to be considered healthy.
                     type: integer
                 type: object
-              hostHeader:
-                description: HostHeader is the hostname to be used as Host header
-                  when proxying requests through Kong.
-                type: string
               slots:
                 description: Slots is the number of slots in the load balancer algorithm.
                   If not set, the default value in Kong for the algorithm is used.
@@ -1546,18 +1542,26 @@ spec:
           rule: 'has(self.spec.hashOn) ? [has(self.spec.hashOn.input), has(self.spec.hashOn.cookie),
             has(self.spec.hashOn.header), has(self.spec.hashOn.uriCapture), has(self.spec.hashOn.queryArg)].filter(fieldSet,
             fieldSet == true).size() <= 1 : true'
-        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
-            can be set.
-          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
-            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
-            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
-            <= 1 : true'
         - message: When spec.hashOn.cookie is set, spec.hashOn.cookiePath is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? has(self.spec.hashOn.cookiePath)
             : true'
         - message: When spec.hashOn.cookiePath is set, spec.hashOn.cookie is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookiePath) ? has(self.spec.hashOn.cookie)
             : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
+            is set.
+          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
+        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
+            can be set.
+          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
+            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
+            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
+            <= 1 : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
+            is set.
+          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
         - message: spec.hashOnFallback.cookie must not be set.
           rule: 'has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookie)
             : true'
@@ -1572,14 +1576,10 @@ spec:
           rule: 'has(self.spec.healthchecks) && has(self.spec.healthchecks.passive)
             && has(self.spec.healthchecks.passive.unhealthy) ? !has(self.spec.healthchecks.passive.unhealthy.interval)
             : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
-            is set.
-          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
-            is set.
-          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
+        - message: spec.hashOnFallback must not be set when spec.hashOn.cookie is
+            set.
+          rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? !has(self.spec.hashOnFallback)
+            : true'
     served: true
     storage: true
     subresources:
diff --git a/test/e2e/manifests/all-in-one-dbless-konnect-enterprise.yaml b/test/e2e/manifests/all-in-one-dbless-konnect-enterprise.yaml
index 7c14b50f4..9e1788389 100644
--- a/test/e2e/manifests/all-in-one-dbless-konnect-enterprise.yaml
+++ b/test/e2e/manifests/all-in-one-dbless-konnect-enterprise.yaml
@@ -1528,10 +1528,6 @@ spec:
                       to be considered healthy.
                     type: integer
                 type: object
-              hostHeader:
-                description: HostHeader is the hostname to be used as Host header
-                  when proxying requests through Kong.
-                type: string
               slots:
                 description: Slots is the number of slots in the load balancer algorithm.
                   If not set, the default value in Kong for the algorithm is used.
@@ -1546,18 +1542,26 @@ spec:
           rule: 'has(self.spec.hashOn) ? [has(self.spec.hashOn.input), has(self.spec.hashOn.cookie),
             has(self.spec.hashOn.header), has(self.spec.hashOn.uriCapture), has(self.spec.hashOn.queryArg)].filter(fieldSet,
             fieldSet == true).size() <= 1 : true'
-        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
-            can be set.
-          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
-            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
-            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
-            <= 1 : true'
         - message: When spec.hashOn.cookie is set, spec.hashOn.cookiePath is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? has(self.spec.hashOn.cookiePath)
             : true'
         - message: When spec.hashOn.cookiePath is set, spec.hashOn.cookie is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookiePath) ? has(self.spec.hashOn.cookie)
             : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
+            is set.
+          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
+        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
+            can be set.
+          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
+            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
+            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
+            <= 1 : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
+            is set.
+          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
         - message: spec.hashOnFallback.cookie must not be set.
           rule: 'has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookie)
             : true'
@@ -1572,14 +1576,10 @@ spec:
           rule: 'has(self.spec.healthchecks) && has(self.spec.healthchecks.passive)
             && has(self.spec.healthchecks.passive.unhealthy) ? !has(self.spec.healthchecks.passive.unhealthy.interval)
             : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
-            is set.
-          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
-            is set.
-          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
+        - message: spec.hashOnFallback must not be set when spec.hashOn.cookie is
+            set.
+          rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? !has(self.spec.hashOnFallback)
+            : true'
     served: true
     storage: true
     subresources:
diff --git a/test/e2e/manifests/all-in-one-dbless-konnect.yaml b/test/e2e/manifests/all-in-one-dbless-konnect.yaml
index 05cd374a3..8750affeb 100644
--- a/test/e2e/manifests/all-in-one-dbless-konnect.yaml
+++ b/test/e2e/manifests/all-in-one-dbless-konnect.yaml
@@ -1528,10 +1528,6 @@ spec:
                       to be considered healthy.
                     type: integer
                 type: object
-              hostHeader:
-                description: HostHeader is the hostname to be used as Host header
-                  when proxying requests through Kong.
-                type: string
               slots:
                 description: Slots is the number of slots in the load balancer algorithm.
                   If not set, the default value in Kong for the algorithm is used.
@@ -1546,18 +1542,26 @@ spec:
           rule: 'has(self.spec.hashOn) ? [has(self.spec.hashOn.input), has(self.spec.hashOn.cookie),
             has(self.spec.hashOn.header), has(self.spec.hashOn.uriCapture), has(self.spec.hashOn.queryArg)].filter(fieldSet,
             fieldSet == true).size() <= 1 : true'
-        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
-            can be set.
-          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
-            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
-            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
-            <= 1 : true'
         - message: When spec.hashOn.cookie is set, spec.hashOn.cookiePath is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? has(self.spec.hashOn.cookiePath)
             : true'
         - message: When spec.hashOn.cookiePath is set, spec.hashOn.cookie is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookiePath) ? has(self.spec.hashOn.cookie)
             : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
+            is set.
+          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
+        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
+            can be set.
+          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
+            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
+            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
+            <= 1 : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
+            is set.
+          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
         - message: spec.hashOnFallback.cookie must not be set.
           rule: 'has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookie)
             : true'
@@ -1572,14 +1576,10 @@ spec:
           rule: 'has(self.spec.healthchecks) && has(self.spec.healthchecks.passive)
             && has(self.spec.healthchecks.passive.unhealthy) ? !has(self.spec.healthchecks.passive.unhealthy.interval)
             : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
-            is set.
-          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
-            is set.
-          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
+        - message: spec.hashOnFallback must not be set when spec.hashOn.cookie is
+            set.
+          rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? !has(self.spec.hashOnFallback)
+            : true'
     served: true
     storage: true
     subresources:
diff --git a/test/e2e/manifests/all-in-one-dbless.yaml b/test/e2e/manifests/all-in-one-dbless.yaml
index 07eff47f3..254547d49 100644
--- a/test/e2e/manifests/all-in-one-dbless.yaml
+++ b/test/e2e/manifests/all-in-one-dbless.yaml
@@ -1528,10 +1528,6 @@ spec:
                       to be considered healthy.
                     type: integer
                 type: object
-              hostHeader:
-                description: HostHeader is the hostname to be used as Host header
-                  when proxying requests through Kong.
-                type: string
               slots:
                 description: Slots is the number of slots in the load balancer algorithm.
                   If not set, the default value in Kong for the algorithm is used.
@@ -1546,18 +1542,26 @@ spec:
           rule: 'has(self.spec.hashOn) ? [has(self.spec.hashOn.input), has(self.spec.hashOn.cookie),
             has(self.spec.hashOn.header), has(self.spec.hashOn.uriCapture), has(self.spec.hashOn.queryArg)].filter(fieldSet,
             fieldSet == true).size() <= 1 : true'
-        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
-            can be set.
-          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
-            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
-            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
-            <= 1 : true'
         - message: When spec.hashOn.cookie is set, spec.hashOn.cookiePath is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? has(self.spec.hashOn.cookiePath)
             : true'
         - message: When spec.hashOn.cookiePath is set, spec.hashOn.cookie is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookiePath) ? has(self.spec.hashOn.cookie)
             : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
+            is set.
+          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
+        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
+            can be set.
+          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
+            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
+            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
+            <= 1 : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
+            is set.
+          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
         - message: spec.hashOnFallback.cookie must not be set.
           rule: 'has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookie)
             : true'
@@ -1572,14 +1576,10 @@ spec:
           rule: 'has(self.spec.healthchecks) && has(self.spec.healthchecks.passive)
             && has(self.spec.healthchecks.passive.unhealthy) ? !has(self.spec.healthchecks.passive.unhealthy.interval)
             : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
-            is set.
-          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
-            is set.
-          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
+        - message: spec.hashOnFallback must not be set when spec.hashOn.cookie is
+            set.
+          rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? !has(self.spec.hashOnFallback)
+            : true'
     served: true
     storage: true
     subresources:
diff --git a/test/e2e/manifests/all-in-one-postgres-enterprise.yaml b/test/e2e/manifests/all-in-one-postgres-enterprise.yaml
index 2872440bc..8e8b97897 100644
--- a/test/e2e/manifests/all-in-one-postgres-enterprise.yaml
+++ b/test/e2e/manifests/all-in-one-postgres-enterprise.yaml
@@ -1528,10 +1528,6 @@ spec:
                       to be considered healthy.
                     type: integer
                 type: object
-              hostHeader:
-                description: HostHeader is the hostname to be used as Host header
-                  when proxying requests through Kong.
-                type: string
               slots:
                 description: Slots is the number of slots in the load balancer algorithm.
                   If not set, the default value in Kong for the algorithm is used.
@@ -1546,18 +1542,26 @@ spec:
           rule: 'has(self.spec.hashOn) ? [has(self.spec.hashOn.input), has(self.spec.hashOn.cookie),
             has(self.spec.hashOn.header), has(self.spec.hashOn.uriCapture), has(self.spec.hashOn.queryArg)].filter(fieldSet,
             fieldSet == true).size() <= 1 : true'
-        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
-            can be set.
-          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
-            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
-            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
-            <= 1 : true'
         - message: When spec.hashOn.cookie is set, spec.hashOn.cookiePath is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? has(self.spec.hashOn.cookiePath)
             : true'
         - message: When spec.hashOn.cookiePath is set, spec.hashOn.cookie is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookiePath) ? has(self.spec.hashOn.cookie)
             : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
+            is set.
+          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
+        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
+            can be set.
+          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
+            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
+            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
+            <= 1 : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
+            is set.
+          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
         - message: spec.hashOnFallback.cookie must not be set.
           rule: 'has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookie)
             : true'
@@ -1572,14 +1576,10 @@ spec:
           rule: 'has(self.spec.healthchecks) && has(self.spec.healthchecks.passive)
             && has(self.spec.healthchecks.passive.unhealthy) ? !has(self.spec.healthchecks.passive.unhealthy.interval)
             : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
-            is set.
-          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
-            is set.
-          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
+        - message: spec.hashOnFallback must not be set when spec.hashOn.cookie is
+            set.
+          rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? !has(self.spec.hashOnFallback)
+            : true'
     served: true
     storage: true
     subresources:
diff --git a/test/e2e/manifests/all-in-one-postgres.yaml b/test/e2e/manifests/all-in-one-postgres.yaml
index f94027efe..3c570281d 100644
--- a/test/e2e/manifests/all-in-one-postgres.yaml
+++ b/test/e2e/manifests/all-in-one-postgres.yaml
@@ -1528,10 +1528,6 @@ spec:
                       to be considered healthy.
                     type: integer
                 type: object
-              hostHeader:
-                description: HostHeader is the hostname to be used as Host header
-                  when proxying requests through Kong.
-                type: string
               slots:
                 description: Slots is the number of slots in the load balancer algorithm.
                   If not set, the default value in Kong for the algorithm is used.
@@ -1546,18 +1542,26 @@ spec:
           rule: 'has(self.spec.hashOn) ? [has(self.spec.hashOn.input), has(self.spec.hashOn.cookie),
             has(self.spec.hashOn.header), has(self.spec.hashOn.uriCapture), has(self.spec.hashOn.queryArg)].filter(fieldSet,
             fieldSet == true).size() <= 1 : true'
-        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
-            can be set.
-          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
-            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
-            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
-            <= 1 : true'
         - message: When spec.hashOn.cookie is set, spec.hashOn.cookiePath is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? has(self.spec.hashOn.cookiePath)
             : true'
         - message: When spec.hashOn.cookiePath is set, spec.hashOn.cookie is required.
           rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookiePath) ? has(self.spec.hashOn.cookie)
             : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
+            is set.
+          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
+        - message: Only one of spec.hashOnFallback.(input|header|uriCapture|queryArg)
+            can be set.
+          rule: 'has(self.spec.hashOnFallback) ? [has(self.spec.hashOnFallback.input),
+            has(self.spec.hashOnFallback.header), has(self.spec.hashOnFallback.uriCapture),
+            has(self.spec.hashOnFallback.queryArg)].filter(fieldSet, fieldSet == true).size()
+            <= 1 : true'
+        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
+            is set.
+          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
+            == "consistent-hashing" : true'
         - message: spec.hashOnFallback.cookie must not be set.
           rule: 'has(self.spec.hashOnFallback) ? !has(self.spec.hashOnFallback.cookie)
             : true'
@@ -1572,14 +1576,10 @@ spec:
           rule: 'has(self.spec.healthchecks) && has(self.spec.healthchecks.passive)
             && has(self.spec.healthchecks.passive.unhealthy) ? !has(self.spec.healthchecks.passive.unhealthy.interval)
             : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOn
-            is set.
-          rule: 'has(self.spec.hashOn) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
-        - message: spec.algorithm must be set to "consistent-hashing" when spec.hashOnFallback
-            is set.
-          rule: 'has(self.spec.hashOnFallback) ? has(self.spec.algorithm) && self.spec.algorithm
-            == "consistent-hashing" : true'
+        - message: spec.hashOnFallback must not be set when spec.hashOn.cookie is
+            set.
+          rule: 'has(self.spec.hashOn) && has(self.spec.hashOn.cookie) ? !has(self.spec.hashOnFallback)
+            : true'
     served: true
     storage: true
     subresources:
diff --git a/test/envtest/crds_envtest_test.go b/test/envtest/crds_envtest_test.go
index d731981e1..3cf27d44f 100644
--- a/test/envtest/crds_envtest_test.go
+++ b/test/envtest/crds_envtest_test.go
@@ -425,7 +425,6 @@ func TestCRDValidations(t *testing.T) {
 						require.NoError(t, err)
 					})
 				}
-
 				t.Run("invalid value", func(t *testing.T) {
 					err := createKongUpstreamPolicy(ctx, ctrlClient, ns, kongv1beta1.KongUpstreamPolicySpec{
 						Healthchecks: &kongv1beta1.KongUpstreamHealthcheck{
@@ -442,6 +441,22 @@ func TestCRDValidations(t *testing.T) {
 				})
 			},
 		},
+		{
+			name: "KongUpstreamPolicy - hashOnFallback must not be set when spec.hasOn.cookie is set",
+			scenario: func(ctx context.Context, t *testing.T, ns string) {
+				err := createKongUpstreamPolicy(ctx, ctrlClient, ns, kongv1beta1.KongUpstreamPolicySpec{
+					Algorithm: lo.ToPtr("consistent-hashing"),
+					HashOn: &kongv1beta1.KongUpstreamHash{
+						Cookie:     lo.ToPtr("cookie-name"),
+						CookiePath: lo.ToPtr("/"),
+					},
+					HashOnFallback: &kongv1beta1.KongUpstreamHash{
+						Header: lo.ToPtr("header-name"),
+					},
+				})
+				require.ErrorContains(t, err, `spec.hashOnFallback must not be set when spec.hashOn.cookie is set.`)
+			},
+		},
 	}
 
 	for _, tc := range testCases {
diff --git a/test/kongintegration/kongupstreampolicy_test.go b/test/kongintegration/kongupstreampolicy_test.go
index 5c0f08ca7..b5f9b71ce 100644
--- a/test/kongintegration/kongupstreampolicy_test.go
+++ b/test/kongintegration/kongupstreampolicy_test.go
@@ -13,7 +13,7 @@ import (
 	"github.com/samber/lo"
 	"github.com/stretchr/testify/require"
 
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/parser/translators"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/kongstate"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/sendconfig"
 	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 	"github.com/kong/kubernetes-ingress-controller/v3/test/kongintegration/containers"
@@ -48,14 +48,12 @@ func TestKongUpstreamPolicyTranslation(t *testing.T) {
 		{
 			name: "KongUpstreamPolicySpec with no hash-on or hash-fallback",
 			policySpec: kongv1beta1.KongUpstreamPolicySpec{
-				HostHeader: lo.ToPtr("foo"),
-				Algorithm:  lo.ToPtr("least-connections"),
-				Slots:      lo.ToPtr(20),
+				Algorithm: lo.ToPtr("least-connections"),
+				Slots:     lo.ToPtr(20),
 			},
 			expectedUpstream: &kong.Upstream{
-				HostHeader: lo.ToPtr("foo"),
-				Algorithm:  lo.ToPtr("least-connections"),
-				Slots:      lo.ToPtr(20),
+				Algorithm: lo.ToPtr("least-connections"),
+				Slots:     lo.ToPtr(20),
 			},
 		},
 		{
@@ -222,7 +220,7 @@ func TestKongUpstreamPolicyTranslation(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			translatedUpstream := translators.TranslateKongUpstreamPolicy(tc.policySpec)
+			translatedUpstream := kongstate.TranslateKongUpstreamPolicy(tc.policySpec)
 			const upstreamName = "test-upstream"
 			translatedUpstream.Name = lo.ToPtr(upstreamName)
 			tc.expectedUpstream.Name = lo.ToPtr(upstreamName)