Repository: envoyproxy/gateway

Issue:
Support PDB for Ratelimit service
*Description*:
>Describe the desired behavior, what scenario it enables and how it
would be used.

Similar to PDB support for Envoy Proxy fleet https://github.com/envoyproxy/gateway/blame/b97f1a14f877f00317bb8467de7ea161cfff4781/internal/infrastructure/kubernetes/proxy/resource_provider.go#L530

also add support for PDB in RateLimit Service 

[optional *Relevant Links*:]
>Any extra documentation required to understand the issue.


Discussion:

Solution patch:
diff --git a/api/v1alpha1/envoygateway_types.go b/api/v1alpha1/envoygateway_types.go
index 7f2d6723c..27c1ded9d 100644
--- a/api/v1alpha1/envoygateway_types.go
+++ b/api/v1alpha1/envoygateway_types.go
@@ -269,6 +269,11 @@ type EnvoyGatewayKubernetesProvider struct {
 	// +optional
 	RateLimitHpa *KubernetesHorizontalPodAutoscalerSpec `json:"rateLimitHpa,omitempty"`
 
+	// RateLimitPDB allows to control the pod disruption budget of rate limit service.
+	//
+	// +optional
+	RateLimitPDB *KubernetesPodDisruptionBudgetSpec `json:"rateLimitPdb,omitempty"`
+
 	// Watch holds configuration of which input resources should be watched and reconciled.
 	// +optional
 	Watch *KubernetesWatchMode `json:"watch,omitempty"`
diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go
index abef1052a..59e1c985c 100644
--- a/api/v1alpha1/zz_generated.deepcopy.go
+++ b/api/v1alpha1/zz_generated.deepcopy.go
@@ -2022,6 +2022,11 @@ func (in *EnvoyGatewayKubernetesProvider) DeepCopyInto(out *EnvoyGatewayKubernet
 		*out = new(KubernetesHorizontalPodAutoscalerSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.RateLimitPDB != nil {
+		in, out := &in.RateLimitPDB, &out.RateLimitPDB
+		*out = new(KubernetesPodDisruptionBudgetSpec)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Watch != nil {
 		in, out := &in.Watch, &out.Watch
 		*out = new(KubernetesWatchMode)
diff --git a/go.mod b/go.mod
index b9e21b1ea..1d1c080ff 100644
--- a/go.mod
+++ b/go.mod
@@ -15,9 +15,9 @@ require (
 	github.com/docker/cli v28.5.1+incompatible
 	github.com/docker/docker v28.3.3+incompatible
 	github.com/dominikbraun/graph v0.23.0
-	github.com/envoyproxy/go-control-plane v0.13.5-0.20250929230642-07d3df27ff4f
+	github.com/envoyproxy/go-control-plane v0.13.5-0.20251022160057-de4316c523b7
 	github.com/envoyproxy/go-control-plane/contrib v1.32.5-0.20250430092421-68a532e11403
-	github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20250929230642-07d3df27ff4f
+	github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251022160057-de4316c523b7
 	github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20250805143705-d51f8590a549
 	github.com/envoyproxy/ratelimit v1.4.1-0.20230427142404-e2a87f41d3a7
 	github.com/evanphx/json-patch v5.9.11+incompatible
diff --git a/go.sum b/go.sum
index e6b8f4b58..f998ef795 100644
--- a/go.sum
+++ b/go.sum
@@ -173,12 +173,12 @@ github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRr
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
-github.com/envoyproxy/go-control-plane v0.13.5-0.20250929230642-07d3df27ff4f h1:36vvJBe/wXWfD7qrTb1WnbPVPMxNFDfEygztH8wgebw=
-github.com/envoyproxy/go-control-plane v0.13.5-0.20250929230642-07d3df27ff4f/go.mod h1:PTY7yDlLxB4bW7rEOO7e79uTDr9yXzpuI1QGIDfxEzc=
+github.com/envoyproxy/go-control-plane v0.13.5-0.20251022160057-de4316c523b7 h1:JsOVlgacLOjgYWGb4V2E/d/c/o4+HEvYjr0+KnL4j4o=
+github.com/envoyproxy/go-control-plane v0.13.5-0.20251022160057-de4316c523b7/go.mod h1:Alz8LEClvR7xKsrq3qzoc4N0guvVNSS8KmSChGYr9hs=
 github.com/envoyproxy/go-control-plane/contrib v1.32.5-0.20250430092421-68a532e11403 h1:5wPocL1bGYhA4TtKZwcdVI5fsXo1JatkbcxPBcFQswc=
 github.com/envoyproxy/go-control-plane/contrib v1.32.5-0.20250430092421-68a532e11403/go.mod h1:Xkwx/TGvEKRCL2mitdiuQWOD1ECvfM5krWWVo2vI2Zk=
-github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20250929230642-07d3df27ff4f h1:4efYrIQgVRwCmwCveby6ck+VpxqzibdOL1Out1rJqqc=
-github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20250929230642-07d3df27ff4f/go.mod h1:2LcmvJoXsDSrsGZIxGM0Gah9ykiwTn/kgjyQdnNH8Jc=
+github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251022160057-de4316c523b7 h1:Q4zISHdb9brRNvzmQl8Bwvap7GAtGrQjHyyw9OYmkFk=
+github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251022160057-de4316c523b7/go.mod h1:ty89S1YCCVruQAm9OtKeEkQLTb+Lkz0k8v9W0Oxsv98=
 github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20250805143705-d51f8590a549 h1:5K0vH5H4dtCIO8+w/yq6vDaMcGn9RoPrHfmPAFAztwU=
 github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20250805143705-d51f8590a549/go.mod h1:KxtyvDAPIEkqUUvF9ooo5gSGVOtQ08wUTnQe5LsJC6c=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
diff --git a/internal/infrastructure/kubernetes/common/resource.go b/internal/infrastructure/kubernetes/common/resource.go
new file mode 100644
index 000000000..3c70d78db
--- /dev/null
+++ b/internal/infrastructure/kubernetes/common/resource.go
@@ -0,0 +1,65 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package common
+
+import (
+	policyv1 "k8s.io/api/policy/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
+
+	egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
+)
+
+// ResourceKind indicates the main resources of envoy-ratelimit,
+// but also the key for the uid of their ownerReference.
+const (
+	ResourceKindDeployment = "Deployment"
+)
+
+func GetPodDisruptionBudget(pdb *egv1a1.KubernetesPodDisruptionBudgetSpec,
+	selector *metav1.LabelSelector, nn *types.NamespacedName, ownerReferences []metav1.OwnerReference,
+) (*policyv1.PodDisruptionBudget, error) {
+	// If podDisruptionBudget config is nil, ignore PodDisruptionBudget.
+	if pdb == nil {
+		return nil, nil
+	}
+
+	pdbSpec := policyv1.PodDisruptionBudgetSpec{
+		Selector: selector,
+	}
+
+	switch {
+	case pdb.MinAvailable != nil:
+		pdbSpec.MinAvailable = pdb.MinAvailable
+	case pdb.MaxUnavailable != nil:
+		pdbSpec.MaxUnavailable = pdb.MaxUnavailable
+	default:
+		pdbSpec.MinAvailable = &intstr.IntOrString{Type: intstr.Int, IntVal: 0}
+	}
+
+	podDisruptionBudget := &policyv1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      nn.Name,
+			Namespace: nn.Namespace,
+			Labels:    selector.MatchLabels,
+		},
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "policy/v1",
+			Kind:       "PodDisruptionBudget",
+		},
+		Spec: pdbSpec,
+	}
+
+	podDisruptionBudget.OwnerReferences = append(podDisruptionBudget.OwnerReferences, ownerReferences...)
+
+	// apply merge patch to PodDisruptionBudget
+	podDisruptionBudget, err := pdb.ApplyMergePatch(podDisruptionBudget)
+	if err != nil {
+		return nil, err
+	}
+	return podDisruptionBudget, nil
+}
diff --git a/internal/infrastructure/kubernetes/proxy/resource_provider.go b/internal/infrastructure/kubernetes/proxy/resource_provider.go
index 29955fb57..51fd2d4fb 100644
--- a/internal/infrastructure/kubernetes/proxy/resource_provider.go
+++ b/internal/infrastructure/kubernetes/proxy/resource_provider.go
@@ -27,6 +27,7 @@ import (
 	"github.com/envoyproxy/gateway/internal/gatewayapi"
 	gwapiresource "github.com/envoyproxy/gateway/internal/gatewayapi/resource"
 	"github.com/envoyproxy/gateway/internal/infrastructure/common"
+	infracommon "github.com/envoyproxy/gateway/internal/infrastructure/kubernetes/common"
 	"github.com/envoyproxy/gateway/internal/infrastructure/kubernetes/resource"
 	"github.com/envoyproxy/gateway/internal/ir"
 	"github.com/envoyproxy/gateway/internal/utils"
@@ -134,7 +135,7 @@ func (r *ResourceRender) LabelSelector() labels.Selector {
 	return labels.SelectorFromSet(r.stableSelector().MatchLabels)
 }
 
-func (r *ResourceRender) OwnerReferences() []metav1.OwnerReference {
+func (r *ResourceRender) ownerReferences() []metav1.OwnerReference {
 	var ownerReferences []metav1.OwnerReference
 	if r.ownerReferenceUID != nil {
 		key := gwapiresource.KindGatewayClass
@@ -172,7 +173,7 @@ func (r *ResourceRender) ServiceAccount() (*corev1.ServiceAccount, error) {
 			Name:            r.serviceAccountName(),
 			Labels:          saLabels,
 			Annotations:     r.infra.GetProxyMetadata().Annotations,
-			OwnerReferences: r.OwnerReferences(),
+			OwnerReferences: r.ownerReferences(),
 		},
 	}, nil
 }
@@ -290,7 +291,7 @@ func (r *ResourceRender) Service() (*corev1.Service, error) {
 			Namespace:       r.Namespace(),
 			Labels:          svcLabels,
 			Annotations:     annotations,
-			OwnerReferences: r.OwnerReferences(),
+			OwnerReferences: r.ownerReferences(),
 		},
 		Spec: serviceSpec,
 	}
@@ -342,7 +343,7 @@ func (r *ResourceRender) ConfigMap(cert string) (*corev1.ConfigMap, error) {
 			Name:            r.Name(),
 			Labels:          cmLabels,
 			Annotations:     r.infra.GetProxyMetadata().Annotations,
-			OwnerReferences: r.OwnerReferences(),
+			OwnerReferences: r.ownerReferences(),
 		},
 		Data: data,
 	}, nil
@@ -401,7 +402,7 @@ func (r *ResourceRender) Deployment() (*appsv1.Deployment, error) {
 			Namespace:       r.Namespace(),
 			Labels:          dpLabels,
 			Annotations:     dpAnnotations,
-			OwnerReferences: r.OwnerReferences(),
+			OwnerReferences: r.ownerReferences(),
 		},
 		Spec: appsv1.DeploymentSpec{
 			Replicas: deploymentConfig.Replicas,
@@ -491,7 +492,7 @@ func (r *ResourceRender) DaemonSet() (*appsv1.DaemonSet, error) {
 			Namespace:       r.Namespace(),
 			Labels:          dsLabels,
 			Annotations:     dsAnnotations,
-			OwnerReferences: r.OwnerReferences(),
+			OwnerReferences: r.ownerReferences(),
 		},
 		Spec: appsv1.DaemonSetSpec{
 			// Daemonset's selector is immutable.
@@ -543,45 +544,13 @@ func (r *ResourceRender) PodDisruptionBudget() (*policyv1.PodDisruptionBudget, e
 		return nil, err
 	}
 
-	pdbSpec := policyv1.PodDisruptionBudgetSpec{
-		Selector: r.stableSelector(),
-	}
-	switch {
-	case pdb.MinAvailable != nil:
-		pdbSpec.MinAvailable = pdb.MinAvailable
-	case pdb.MaxUnavailable != nil:
-		pdbSpec.MaxUnavailable = pdb.MaxUnavailable
-	default:
-		pdbSpec.MinAvailable = &intstr.IntOrString{Type: intstr.Int, IntVal: 0}
-	}
-
-	podDisruptionBudget := &policyv1.PodDisruptionBudget{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace:       r.Namespace(),
-			OwnerReferences: r.OwnerReferences(),
-			Annotations:     r.infra.GetProxyMetadata().Annotations,
-			Labels:          r.stableSelector().MatchLabels,
-		},
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "policy/v1",
-			Kind:       "PodDisruptionBudget",
-		},
-		Spec: pdbSpec,
-	}
-
 	// set name
+	resourceName := r.Name()
 	if pdb.Name != nil {
-		podDisruptionBudget.Name = *pdb.Name
-	} else {
-		podDisruptionBudget.Name = r.Name()
+		resourceName = *pdb.Name
 	}
 
-	// apply merge patch to PodDisruptionBudget
-	if podDisruptionBudget, err = pdb.ApplyMergePatch(podDisruptionBudget); err != nil {
-		return nil, err
-	}
-
-	return podDisruptionBudget, nil
+	return infracommon.GetPodDisruptionBudget(pdb, r.stableSelector(), &types.NamespacedName{Name: resourceName, Namespace: r.Namespace()}, r.ownerReferences())
 }
 
 func (r *ResourceRender) HorizontalPodAutoscaler() (*autoscalingv2.HorizontalPodAutoscaler, error) {
@@ -604,7 +573,7 @@ func (r *ResourceRender) HorizontalPodAutoscaler() (*autoscalingv2.HorizontalPod
 			Namespace:       r.Namespace(),
 			Annotations:     r.infra.GetProxyMetadata().Annotations,
 			Labels:          r.stableSelector().MatchLabels,
-			OwnerReferences: r.OwnerReferences(),
+			OwnerReferences: r.ownerReferences(),
 		},
 		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
 			ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
diff --git a/internal/infrastructure/kubernetes/ratelimit/resource_provider.go b/internal/infrastructure/kubernetes/ratelimit/resource_provider.go
index 31654b4cd..61339c761 100644
--- a/internal/infrastructure/kubernetes/ratelimit/resource_provider.go
+++ b/internal/infrastructure/kubernetes/ratelimit/resource_provider.go
@@ -21,6 +21,7 @@ import (
 	"k8s.io/utils/ptr"
 
 	egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
+	"github.com/envoyproxy/gateway/internal/infrastructure/kubernetes/common"
 	"github.com/envoyproxy/gateway/internal/infrastructure/kubernetes/resource"
 	"github.com/envoyproxy/gateway/internal/utils"
 )
@@ -44,6 +45,7 @@ type ResourceRender struct {
 	rateLimit           *egv1a1.RateLimit
 	rateLimitDeployment *egv1a1.KubernetesDeploymentSpec
 	rateLimitHpa        *egv1a1.KubernetesHorizontalPodAutoscalerSpec
+	rateLimitPdb        *egv1a1.KubernetesPodDisruptionBudgetSpec
 
 	// ownerReferenceUID store the uid of its owner reference.
 	ownerReferenceUID map[string]types.UID
@@ -51,13 +53,18 @@ type ResourceRender struct {
 
 // NewResourceRender returns a new ResourceRender.
 func NewResourceRender(ns string, gateway *egv1a1.EnvoyGateway, ownerReferenceUID map[string]types.UID) *ResourceRender {
-	return &ResourceRender{
+	prov := gateway.GetEnvoyGatewayProvider().GetEnvoyGatewayKubeProvider()
+
+	render := &ResourceRender{
 		namespace:           ns,
 		rateLimit:           gateway.RateLimit,
-		rateLimitDeployment: gateway.GetEnvoyGatewayProvider().GetEnvoyGatewayKubeProvider().RateLimitDeployment,
-		rateLimitHpa:        gateway.GetEnvoyGatewayProvider().GetEnvoyGatewayKubeProvider().RateLimitHpa,
+		rateLimitDeployment: prov.RateLimitDeployment,
+		rateLimitHpa:        prov.RateLimitHpa,
+		rateLimitPdb:        prov.RateLimitPDB,
 		ownerReferenceUID:   ownerReferenceUID,
 	}
+
+	return render
 }
 
 func (r *ResourceRender) Name() string {
@@ -84,7 +91,7 @@ func enablePrometheus(rl *egv1a1.RateLimit) bool {
 }
 
 // ConfigMap returns the expected rate limit ConfigMap based on the provided infra.
-func (r *ResourceRender) ConfigMap(cert string) (*corev1.ConfigMap, error) {
+func (r *ResourceRender) ConfigMap(_ string) (*corev1.ConfigMap, error) {
 	if !enablePrometheus(r.rateLimit) {
 		return nil, nil
 	}
@@ -357,5 +364,36 @@ func (r *ResourceRender) HorizontalPodAutoscaler() (*autoscalingv2.HorizontalPod
 }
 
 func (r *ResourceRender) PodDisruptionBudget() (*policyv1.PodDisruptionBudget, error) {
-	return nil, nil
+	pdb := r.rateLimitPdb
+	if pdb == nil {
+		return nil, nil
+	}
+
+	// set name
+	resourceName := r.Name()
+	if pdb.Name != nil {
+		resourceName = *pdb.Name
+	}
+
+	return common.GetPodDisruptionBudget(r.rateLimitPdb, resource.GetSelector(rateLimitLabels()), &types.NamespacedName{
+		Name:      resourceName,
+		Namespace: r.Namespace(),
+	}, r.ownerReferences())
+}
+
+func (r *ResourceRender) ownerReferences() []metav1.OwnerReference {
+	var ownerReferences []metav1.OwnerReference
+	if r.ownerReferenceUID != nil {
+		if uid, ok := r.ownerReferenceUID[ResourceKindDeployment]; ok {
+			ownerReferences = []metav1.OwnerReference{
+				{
+					Kind:       ResourceKindDeployment,
+					APIVersion: appsAPIVersion,
+					Name:       "envoy-gateway",
+					UID:        uid,
+				},
+			}
+		}
+	}
+	return ownerReferences
 }
diff --git a/internal/message/watchutil.go b/internal/message/watchutil.go
index 04f13b1d6..83ef831f6 100644
--- a/internal/message/watchutil.go
+++ b/internal/message/watchutil.go
@@ -99,8 +99,41 @@ func HandleSubscription[K comparable, V any](
 	}
 	for snapshot := range subscription {
 		watchableDepth.With(meta.LabelValues()...).Record(float64(len(subscription)))
-		for _, update := range snapshot.Updates {
+
+		for _, update := range coalesceUpdates(meta.Runner, snapshot.Updates) {
 			handleWithCrashRecovery(handle, Update[K, V](update), meta, errChans)
 		}
 	}
 }
+
+// coalesceUpdates merges multiple updates for the same key into a single update,
+// preserving the latest state for each key.
+// This helps reduce redundant processing and ensures that only the most recent update per key is handled.
+func coalesceUpdates[K comparable, V any](runner string, updates []watchable.Update[K, V]) []watchable.Update[K, V] {
+	if len(updates) <= 1 {
+		return updates
+	}
+
+	seen := make(map[K]struct{}, len(updates))
+	write := len(updates) - 1
+
+	for read := len(updates) - 1; read >= 0; read-- {
+		update := updates[read]
+		if _, ok := seen[update.Key]; ok {
+			continue
+		}
+		seen[update.Key] = struct{}{}
+		updates[write] = update
+		write--
+	}
+
+	result := updates[write+1:]
+	if len(result) != len(updates) {
+		logger.WithValues("runner", runner).Info(
+			"coalesced updates",
+			"count", len(result),
+			"before", len(updates),
+		)
+	}
+	return result
+}
diff --git a/internal/xds/extensions/extensions.gen.go b/internal/xds/extensions/extensions.gen.go
index 0ef2ad09f..cf91ea812 100644
--- a/internal/xds/extensions/extensions.gen.go
+++ b/internal/xds/extensions/extensions.gen.go
@@ -76,6 +76,7 @@ import (
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/dns/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/dynamic_forward_proxy/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/redis/v3"
+	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/reverse_connection/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/common/async_files/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/common/aws/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/common/dynamic_forward_proxy/v3"
@@ -140,6 +141,7 @@ import (
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/kill_request/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/local_ratelimit/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/lua/v3"
+	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/mcp/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/oauth2/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/on_demand/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/original_src/v3"
@@ -224,6 +226,7 @@ import (
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/custom_response/local_response_policy/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/custom_response/redirect_policy/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/early_header_mutation/header_mutation/v3"
+	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/ext_proc/processing_request_modifiers/mapped_attribute_builder/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/ext_proc/response_processors/save_processing_response/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/header_formatters/preserve_case/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/header_validators/envoy_default/v3"
@@ -253,6 +256,7 @@ import (
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/common_inputs/environment_variable/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/common_inputs/network/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/common_inputs/ssl/v3"
+	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/common_inputs/stats/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/input_matchers/consistent_hashing/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/input_matchers/ip/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/input_matchers/metadata/v3"
diff --git a/internal/xds/translator/extauth.go b/internal/xds/translator/extauth.go
index 3fb3ce4cc..00f3ec871 100644
--- a/internal/xds/translator/extauth.go
+++ b/internal/xds/translator/extauth.go
@@ -134,11 +134,23 @@ func extAuthConfig(extAuth *ir.ExtAuth) (*extauthv3.ExtAuthz, error) {
 		timeout = durationpb.New(extAuth.Timeout.Duration)
 	}
 
+	var rp *corev3.RetryPolicy
+	// Set the retry policy if it exists.
+	if extAuth.Traffic != nil && extAuth.Traffic.Retry != nil {
+		var err error
+		rp, err = buildNonRouteRetryPolicy(extAuth.Traffic.Retry)
+		if err != nil {
+			return nil, fmt.Errorf("build retry policy for http service: %w", err)
+		}
+	}
+
 	if extAuth.HTTP != nil {
+		hs := httpService(extAuth.HTTP, timeout)
+		hs.RetryPolicy = rp
+
 		config.Services = &extauthv3.ExtAuthz_HttpService{
-			HttpService: httpService(extAuth.HTTP, timeout),
+			HttpService: hs,
 		}
-		// Retry policy is not supported for HTTP service.
 	} else if extAuth.GRPC != nil {
 		service := &corev3.GrpcService{
 			TargetSpecifier: &corev3.GrpcService_EnvoyGrpc_{
@@ -146,14 +158,8 @@ func extAuthConfig(extAuth *ir.ExtAuth) (*extauthv3.ExtAuthz, error) {
 			},
 			Timeout: timeout,
 		}
-		// Set the retry policy if it exists.
-		if extAuth.Traffic != nil && extAuth.Traffic.Retry != nil {
-			rp, err := buildNonRouteRetryPolicy(extAuth.Traffic.Retry)
-			if err != nil {
-				return nil, fmt.Errorf("build retry policy for gRPC service: %w", err)
-			}
-			service.RetryPolicy = rp
-		}
+		service.RetryPolicy = rp
+
 		config.Services = &extauthv3.ExtAuthz_GrpcService{
 			GrpcService: service,
 		}
diff --git a/release-notes/current.yaml b/release-notes/current.yaml
index 4f006ab5f..9524ec001 100644
--- a/release-notes/current.yaml
+++ b/release-notes/current.yaml
@@ -28,6 +28,7 @@ new features: |
   Added support for http2 stream timeout
   Added support for Envoy PreconnectPolicy in BackendTrafficPolicy.
   Added support for binaryData in ConfigMap referenced by HTTPRouteFilter for direct response.
+  Added support PDB for Ratelimit service.
 
 bug fixes: |
   Fixed %ROUTE_KIND% operator to be lower-cased when used by clusterStatName in EnvoyProxy API.



Test patch:
diff --git a/internal/infrastructure/kubernetes/ratelimit/resource_provider_test.go b/internal/infrastructure/kubernetes/ratelimit/resource_provider_test.go
index b281a346b..a881b2c4e 100644
--- a/internal/infrastructure/kubernetes/ratelimit/resource_provider_test.go
+++ b/internal/infrastructure/kubernetes/ratelimit/resource_provider_test.go
@@ -11,15 +11,16 @@ import (
 	"strconv"
 	"testing"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	autoscalingv2 "k8s.io/api/autoscaling/v2"
 	corev1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/ptr"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 	"sigs.k8s.io/yaml"
@@ -103,20 +104,7 @@ func TestServiceAccount(t *testing.T) {
 	sa, err := r.ServiceAccount()
 	require.NoError(t, err)
 
-	expected, err := loadServiceAccount()
-	require.NoError(t, err)
-
-	assert.Equal(t, expected, sa)
-}
-
-func loadServiceAccount() (*corev1.ServiceAccount, error) {
-	saYAML, err := os.ReadFile("testdata/envoy-ratelimit-serviceaccount.yaml")
-	if err != nil {
-		return nil, err
-	}
-	sa := &corev1.ServiceAccount{}
-	_ = yaml.Unmarshal(saYAML, sa)
-	return sa, nil
+	requireObject[corev1.ServiceAccount](t, sa, "testdata/envoy-ratelimit-serviceaccount.yaml")
 }
 
 func TestService(t *testing.T) {
@@ -164,22 +152,8 @@ func TestService(t *testing.T) {
 
 		svc, err := r.Service()
 		require.NoError(t, err)
-
-		expected, err := loadService(tc.caseName)
-		require.NoError(t, err)
-
-		assert.Equal(t, expected, svc)
-	}
-}
-
-func loadService(caseName string) (*corev1.Service, error) {
-	serviceYAML, err := os.ReadFile(fmt.Sprintf("testdata/services/envoy-ratelimit-service-%s.yaml", caseName))
-	if err != nil {
-		return nil, err
+		requireObject[corev1.Service](t, svc, fmt.Sprintf("testdata/services/envoy-ratelimit-service-%s.yaml", tc.caseName))
 	}
-	svc := &corev1.Service{}
-	_ = yaml.Unmarshal(serviceYAML, svc)
-	return svc, nil
 }
 
 func TestConfigmap(t *testing.T) {
@@ -197,30 +171,81 @@ func TestConfigmap(t *testing.T) {
 	r := NewResourceRender(cfg.ControllerNamespace, cfg.EnvoyGateway, ownerReferenceUID)
 	cm, err := r.ConfigMap("")
 	require.NoError(t, err)
+	requireObject[corev1.ConfigMap](t, cm, "testdata/envoy-ratelimit-configmap.yaml")
+}
 
-	if test.OverrideTestData() {
-		cmYAML, err := yaml.Marshal(cm)
-		require.NoError(t, err)
-		// nolint:gosec
-		err = os.WriteFile("testdata/envoy-ratelimit-configmap.yaml", cmYAML, 0o644)
-		require.NoError(t, err)
-		return
-	}
-
-	expected, err := loadConfigmap()
+func TestPDB(t *testing.T) {
+	cfg, err := config.New(os.Stdout, os.Stderr)
 	require.NoError(t, err)
+	cases := []struct {
+		caseName  string
+		rateLimit *egv1a1.RateLimit
+		pdb       *egv1a1.KubernetesPodDisruptionBudgetSpec
+	}{
+		{
+			caseName: "default",
+			rateLimit: &egv1a1.RateLimit{
+				Backend: egv1a1.RateLimitDatabaseBackend{
+					Type: egv1a1.RedisBackendType,
+					Redis: &egv1a1.RateLimitRedisSettings{
+						URL: "redis.redis.svc:6379",
+					},
+				},
+			},
+			pdb: &egv1a1.KubernetesPodDisruptionBudgetSpec{},
+		},
+		{
+			caseName: "custom",
+			rateLimit: &egv1a1.RateLimit{
+				Backend: egv1a1.RateLimitDatabaseBackend{
+					Type: egv1a1.RedisBackendType,
+					Redis: &egv1a1.RateLimitRedisSettings{
+						URL: "redis.redis.svc:6379",
+					},
+				},
+			},
+			pdb: &egv1a1.KubernetesPodDisruptionBudgetSpec{
+				MinAvailable: &intstr.IntOrString{Type: intstr.Int, IntVal: 10},
+				Patch: &egv1a1.KubernetesPatchSpec{
+					Type: ptr.To(egv1a1.StrategicMerge),
+					Value: apiextensionsv1.JSON{
+						Raw: []byte("{\"spec\":{\"maxUnavailable\": 11}}"),
+					},
+				},
+			},
+		},
+		{
+			caseName: "with-name",
+			rateLimit: &egv1a1.RateLimit{
+				Backend: egv1a1.RateLimitDatabaseBackend{
+					Type: egv1a1.RedisBackendType,
+					Redis: &egv1a1.RateLimitRedisSettings{
+						URL: "redis.redis.svc:6379",
+					},
+				},
+			},
+			pdb: &egv1a1.KubernetesPodDisruptionBudgetSpec{
+				Name: ptr.To("custom-pdb-name"),
+			},
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.caseName, func(t *testing.T) {
+			cfg.EnvoyGateway.RateLimit = tc.rateLimit
 
-	assert.Equal(t, expected, cm)
-}
+			cfg.EnvoyGateway.Provider = &egv1a1.EnvoyGatewayProvider{
+				Type: egv1a1.ProviderTypeKubernetes,
+				Kubernetes: &egv1a1.EnvoyGatewayKubernetesProvider{
+					RateLimitPDB: tc.pdb,
+				},
+			}
+			r := NewResourceRender(cfg.ControllerNamespace, cfg.EnvoyGateway, ownerReferenceUID)
+			pdb, err := r.PodDisruptionBudget()
+			require.NoError(t, err)
 
-func loadConfigmap() (*corev1.ConfigMap, error) {
-	configmapYAML, err := os.ReadFile("testdata/envoy-ratelimit-configmap.yaml")
-	if err != nil {
-		return nil, err
+			requireObject[policyv1.PodDisruptionBudget](t, pdb, fmt.Sprintf("testdata/pdb/%s.yaml", tc.caseName))
+		})
 	}
-	cm := &corev1.ConfigMap{}
-	_ = yaml.Unmarshal(configmapYAML, cm)
-	return cm, nil
 }
 
 func TestDeployment(t *testing.T) {
@@ -767,34 +792,11 @@ func TestDeployment(t *testing.T) {
 			r := NewResourceRender(cfg.ControllerNamespace, cfg.EnvoyGateway, ownerReferenceUID)
 			dp, err := r.Deployment()
 			require.NoError(t, err)
-
-			if test.OverrideTestData() {
-				deploymentYAML, err := yaml.Marshal(dp)
-				require.NoError(t, err)
-				// nolint:gosec
-				err = os.WriteFile(fmt.Sprintf("testdata/deployments/%s.yaml", tc.caseName), deploymentYAML, 0o644)
-				require.NoError(t, err)
-				return
-			}
-
-			expected, err := loadDeployment(tc.caseName)
-			require.NoError(t, err)
-
-			assert.Equal(t, expected, dp)
+			requireObject[appsv1.Deployment](t, dp, fmt.Sprintf("testdata/deployments/%s.yaml", tc.caseName))
 		})
 	}
 }
 
-func loadDeployment(caseName string) (*appsv1.Deployment, error) {
-	deploymentYAML, err := os.ReadFile(fmt.Sprintf("testdata/deployments/%s.yaml", caseName))
-	if err != nil {
-		return nil, err
-	}
-	deployment := &appsv1.Deployment{}
-	_ = yaml.Unmarshal(deploymentYAML, deployment)
-	return deployment, nil
-}
-
 func TestHorizontalPodAutoscaler(t *testing.T) {
 	cfg, err := config.New(os.Stdout, os.Stderr)
 	require.NoError(t, err)
@@ -898,35 +900,38 @@ func TestHorizontalPodAutoscaler(t *testing.T) {
 			hpa, err := r.HorizontalPodAutoscaler()
 			require.NoError(t, err)
 
-			if test.OverrideTestData() {
-				hpaYAML, err := yaml.Marshal(hpa)
-				require.NoError(t, err)
-				// nolint:gosec
-				err = os.WriteFile(fmt.Sprintf("testdata/hpa/%s.yaml", tc.caseName), hpaYAML, 0o644)
-				require.NoError(t, err)
-				return
-			}
-
-			expected, err := loadHpa(tc.caseName)
-			require.NoError(t, err)
-
-			assert.Equal(t, expected, hpa)
+			requireObject[autoscalingv2.HorizontalPodAutoscaler](t, hpa, fmt.Sprintf("testdata/hpa/%s.yaml", tc.caseName))
 		})
 	}
 }
 
-func loadHpa(caseName string) (*autoscalingv2.HorizontalPodAutoscaler, error) {
-	hpaYAML, err := os.ReadFile(fmt.Sprintf("testdata/hpa/%s.yaml", caseName))
+func loadObject[T any](filename string) (*T, error) {
+	obj := new(T)
+	data, err := os.ReadFile(filename)
 	if err != nil {
-		return nil, err
+		return obj, err
+	}
+
+	_ = yaml.Unmarshal(data, obj)
+	return obj, nil
+}
+
+func requireObject[ObjectT any](t *testing.T, got *ObjectT, filename string) {
+	if test.OverrideTestData() {
+		cmYAML, err := yaml.Marshal(got)
+		require.NoError(t, err)
+		// nolint:gosec
+		err = os.WriteFile(filename, cmYAML, 0o644)
+		require.NoError(t, err)
+		return
 	}
 
-	hpa := &autoscalingv2.HorizontalPodAutoscaler{}
-	_ = yaml.Unmarshal(hpaYAML, hpa)
-	return hpa, nil
+	expected, err := loadObject[ObjectT](filename)
+	require.NoError(t, err)
+	require.Equal(t, expected, got)
 }
 
 func TestGetServiceURL(t *testing.T) {
 	got := GetServiceURL("envoy-gateway-system", "example-cluster.local")
-	assert.Equal(t, "grpc://envoy-ratelimit.envoy-gateway-system.svc.example-cluster.local:8081", got)
+	require.Equal(t, "grpc://envoy-ratelimit.envoy-gateway-system.svc.example-cluster.local:8081", got)
 }
diff --git a/internal/infrastructure/kubernetes/ratelimit/testdata/envoy-ratelimit-serviceaccount.yaml b/internal/infrastructure/kubernetes/ratelimit/testdata/envoy-ratelimit-serviceaccount.yaml
index c11aaac1e..86b4cfb20 100644
--- a/internal/infrastructure/kubernetes/ratelimit/testdata/envoy-ratelimit-serviceaccount.yaml
+++ b/internal/infrastructure/kubernetes/ratelimit/testdata/envoy-ratelimit-serviceaccount.yaml
@@ -5,7 +5,7 @@ metadata:
   name: envoy-ratelimit
   namespace: envoy-gateway-system
   ownerReferences:
-    - apiVersion: v1
-      kind: ServiceAccount
-      name: envoy-gateway
-      uid: test-owner-reference-uid-for-service-account
+  - apiVersion: v1
+    kind: ServiceAccount
+    name: envoy-gateway
+    uid: test-owner-reference-uid-for-service-account
diff --git a/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/custom.yaml b/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/custom.yaml
new file mode 100644
index 000000000..2899038da
--- /dev/null
+++ b/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/custom.yaml
@@ -0,0 +1,27 @@
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  labels:
+    app.kubernetes.io/component: ratelimit
+    app.kubernetes.io/managed-by: envoy-gateway
+    app.kubernetes.io/name: envoy-ratelimit
+  name: envoy-ratelimit
+  namespace: envoy-gateway-system
+  ownerReferences:
+  - apiVersion: apps/v1
+    kind: Deployment
+    name: envoy-gateway
+    uid: test-owner-reference-uid-for-deployment
+spec:
+  maxUnavailable: 11
+  minAvailable: 10
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: ratelimit
+      app.kubernetes.io/managed-by: envoy-gateway
+      app.kubernetes.io/name: envoy-ratelimit
+status:
+  currentHealthy: 0
+  desiredHealthy: 0
+  disruptionsAllowed: 0
+  expectedPods: 0
diff --git a/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/default.yaml b/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/default.yaml
new file mode 100644
index 000000000..4d7db02e7
--- /dev/null
+++ b/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/default.yaml
@@ -0,0 +1,26 @@
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  labels:
+    app.kubernetes.io/component: ratelimit
+    app.kubernetes.io/managed-by: envoy-gateway
+    app.kubernetes.io/name: envoy-ratelimit
+  name: envoy-ratelimit
+  namespace: envoy-gateway-system
+  ownerReferences:
+  - apiVersion: apps/v1
+    kind: Deployment
+    name: envoy-gateway
+    uid: test-owner-reference-uid-for-deployment
+spec:
+  minAvailable: 0
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: ratelimit
+      app.kubernetes.io/managed-by: envoy-gateway
+      app.kubernetes.io/name: envoy-ratelimit
+status:
+  currentHealthy: 0
+  desiredHealthy: 0
+  disruptionsAllowed: 0
+  expectedPods: 0
diff --git a/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/with-name.yaml b/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/with-name.yaml
new file mode 100644
index 000000000..b1aa798ae
--- /dev/null
+++ b/internal/infrastructure/kubernetes/ratelimit/testdata/pdb/with-name.yaml
@@ -0,0 +1,26 @@
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  labels:
+    app.kubernetes.io/component: ratelimit
+    app.kubernetes.io/managed-by: envoy-gateway
+    app.kubernetes.io/name: envoy-ratelimit
+  name: custom-pdb-name
+  namespace: envoy-gateway-system
+  ownerReferences:
+  - apiVersion: apps/v1
+    kind: Deployment
+    name: envoy-gateway
+    uid: test-owner-reference-uid-for-deployment
+spec:
+  minAvailable: 0
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: ratelimit
+      app.kubernetes.io/managed-by: envoy-gateway
+      app.kubernetes.io/name: envoy-ratelimit
+status:
+  currentHealthy: 0
+  desiredHealthy: 0
+  disruptionsAllowed: 0
+  expectedPods: 0
diff --git a/internal/infrastructure/kubernetes/ratelimit/testdata/services/envoy-ratelimit-service-default.yaml b/internal/infrastructure/kubernetes/ratelimit/testdata/services/envoy-ratelimit-service-default.yaml
index be6bddc9a..a57933809 100644
--- a/internal/infrastructure/kubernetes/ratelimit/testdata/services/envoy-ratelimit-service-default.yaml
+++ b/internal/infrastructure/kubernetes/ratelimit/testdata/services/envoy-ratelimit-service-default.yaml
@@ -2,29 +2,31 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    app.kubernetes.io/name: envoy-ratelimit
     app.kubernetes.io/component: ratelimit
     app.kubernetes.io/managed-by: envoy-gateway
+    app.kubernetes.io/name: envoy-ratelimit
   name: envoy-ratelimit
   namespace: envoy-gateway-system
   ownerReferences:
-    - apiVersion: v1
-      kind: Service
-      name: envoy-gateway
-      uid: test-owner-reference-uid-for-service
+  - apiVersion: v1
+    kind: Service
+    name: envoy-gateway
+    uid: test-owner-reference-uid-for-service
 spec:
-  type: ClusterIP
-  sessionAffinity: None
   ports:
-    - name: http
-      port: 8081
-      protocol: TCP
-      targetPort: 8081
-    - name: metrics
-      port: 19001
-      protocol: TCP
-      targetPort: 19001
+  - name: http
+    port: 8081
+    protocol: TCP
+    targetPort: 8081
+  - name: metrics
+    port: 19001
+    protocol: TCP
+    targetPort: 19001
   selector:
-    app.kubernetes.io/name: envoy-ratelimit
     app.kubernetes.io/component: ratelimit
     app.kubernetes.io/managed-by: envoy-gateway
+    app.kubernetes.io/name: envoy-ratelimit
+  sessionAffinity: None
+  type: ClusterIP
+status:
+  loadBalancer: {}
diff --git a/internal/infrastructure/kubernetes/ratelimit/testdata/services/envoy-ratelimit-service-no-prometheus.yaml b/internal/infrastructure/kubernetes/ratelimit/testdata/services/envoy-ratelimit-service-no-prometheus.yaml
index 562d013ef..fd91b68cd 100644
--- a/internal/infrastructure/kubernetes/ratelimit/testdata/services/envoy-ratelimit-service-no-prometheus.yaml
+++ b/internal/infrastructure/kubernetes/ratelimit/testdata/services/envoy-ratelimit-service-no-prometheus.yaml
@@ -2,25 +2,27 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    app.kubernetes.io/name: envoy-ratelimit
     app.kubernetes.io/component: ratelimit
     app.kubernetes.io/managed-by: envoy-gateway
+    app.kubernetes.io/name: envoy-ratelimit
   name: envoy-ratelimit
   namespace: envoy-gateway-system
   ownerReferences:
-    - apiVersion: v1
-      kind: Service
-      name: envoy-gateway
-      uid: test-owner-reference-uid-for-service
+  - apiVersion: v1
+    kind: Service
+    name: envoy-gateway
+    uid: test-owner-reference-uid-for-service
 spec:
-  type: ClusterIP
-  sessionAffinity: None
   ports:
-    - name: http
-      port: 8081
-      protocol: TCP
-      targetPort: 8081
+  - name: http
+    port: 8081
+    protocol: TCP
+    targetPort: 8081
   selector:
-    app.kubernetes.io/name: envoy-ratelimit
     app.kubernetes.io/component: ratelimit
     app.kubernetes.io/managed-by: envoy-gateway
+    app.kubernetes.io/name: envoy-ratelimit
+  sessionAffinity: None
+  type: ClusterIP
+status:
+  loadBalancer: {}
diff --git a/internal/message/watchutil_internal_test.go b/internal/message/watchutil_internal_test.go
new file mode 100644
index 000000000..8bf56ab2e
--- /dev/null
+++ b/internal/message/watchutil_internal_test.go
@@ -0,0 +1,68 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package message
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/telepresenceio/watchable"
+)
+
+func TestCoalesceUpdates(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name     string
+		input    []watchable.Update[string, int]
+		expected []watchable.Update[string, int]
+	}{
+		{
+			name:     "empty input returns nil",
+			input:    []watchable.Update[string, int]{},
+			expected: []watchable.Update[string, int]{},
+		},
+		{
+			name: "simple updates without repeats",
+			input: []watchable.Update[string, int]{
+				{Key: "foo", Value: 1},
+				{Key: "bar", Value: 2},
+				{Key: "baz", Value: 3},
+			},
+			expected: []watchable.Update[string, int]{
+				{Key: "foo", Value: 1},
+				{Key: "bar", Value: 2},
+				{Key: "baz", Value: 3},
+			},
+		},
+		{
+			name: "latest update per key wins",
+			input: []watchable.Update[string, int]{
+				{Key: "foo", Value: 1},
+				{Key: "bar", Delete: true, Value: 2},
+				{Key: "baz", Value: 3},
+				{Key: "bar", Value: 4},
+				{Key: "foo", Value: 5},
+				{Key: "baz", Delete: true, Value: 6},
+				{Key: "bar", Value: 7},
+			},
+			expected: []watchable.Update[string, int]{
+				{Key: "foo", Value: 5},
+				{Key: "baz", Delete: true, Value: 6},
+				{Key: "bar", Value: 7},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			actual := coalesceUpdates("test-runner", tc.input)
+			require.Equal(t, tc.expected, actual)
+		})
+	}
+}
diff --git a/internal/message/watchutil_test.go b/internal/message/watchutil_test.go
index bc6b10c9c..21411b3f6 100644
--- a/internal/message/watchutil_test.go
+++ b/internal/message/watchutil_test.go
@@ -89,7 +89,7 @@ func TestHandleSubscriptionAlreadyInitialized(t *testing.T) {
 			}
 		},
 	)
-	assert.Equal(t, 2, storeCalls)
+	assert.LessOrEqual(t, storeCalls, 2) // updates can be coalesced
 	assert.Equal(t, 1, deleteCalls)
 }
 
@@ -268,7 +268,11 @@ func TestControllerResourceUpdate(t *testing.T) {
 					m.GatewayAPIResources.Close()
 				}
 			})
-			assert.Equal(t, tc.updates, updates)
+			if tc.updates > 1 {
+				assert.LessOrEqual(t, updates, tc.updates) // Updates can be coalesced
+			} else {
+				assert.Equal(t, 1, updates)
+			}
 		})
 	}
 }
diff --git a/internal/xds/translator/testdata/in/xds-ir/ext-auth-http-retry.yaml b/internal/xds/translator/testdata/in/xds-ir/ext-auth-http-retry.yaml
new file mode 100644
index 000000000..7c1ae56de
--- /dev/null
+++ b/internal/xds/translator/testdata/in/xds-ir/ext-auth-http-retry.yaml
@@ -0,0 +1,178 @@
+http:
+  - address: 0.0.0.0
+    hostnames:
+      - '*'
+    isHTTP2: false
+    name: default/gateway-1/http
+    path:
+      escapedSlashesAction: UnescapeAndRedirect
+      mergeSlashes: true
+    port: 10080
+    routes:
+      - name: httproute/default/httproute-1/rule/0/match/0/www_foo_com
+        hostname: www.foo.com
+        isHTTP2: false
+        pathMatch:
+          distinct: false
+          name: ""
+          prefix: /foo1
+        backendWeights:
+          invalid: 0
+          valid: 0
+        destination:
+          name: httproute/default/httproute-1/rule/0
+          settings:
+            - addressType: IP
+              endpoints:
+                - host: 7.7.7.7
+                  port: 8080
+              protocol: HTTP
+              weight: 1
+              name: httproute/default/httproute-1/rule/0/backend/0
+        security:
+          extAuth:
+            name: securitypolicy/default/policy-for-http-route-1
+            traffic:
+              retry:
+                numRetries: 2
+                perRetry:
+                  backOff:
+                    baseInterval: 200ms
+                    maxInterval: 3s
+                  timeout: 3s
+                retryOn:
+                  triggers:
+                    - 5xx
+                    - deadline-exceeded
+                    - internal
+                    - unavailable
+            failOpen: false
+            http:
+              authority: http-backend.default:9000
+              destination:
+                name: securitypolicy/default/policy-for-http-route-1/default/http-backend
+                settings:
+                  - addressType: IP
+                    endpoints:
+                      - host: 8.8.4.4
+                        port: 9001
+                    protocol: HTTP
+                    weight: 1
+                    name: securitypolicy/default/policy-for-http-route-1/default/http-backend/backend/0
+                  - addressType: IP
+                    endpoints:
+                      - host: 8.8.8.8
+                        port: 9000
+                    protocol: HTTP
+                    weight: 1
+                    name: securitypolicy/default/policy-for-http-route-1/default/http-backend/backend/1
+            headersToExtAuth:
+              - header1
+              - header2
+      - name: httproute/default/httproute-1/rule/1/match/0/www_foo_com
+        hostname: www.foo.com
+        isHTTP2: false
+        pathMatch:
+          distinct: false
+          name: ""
+          prefix: /foo2
+        backendWeights:
+          invalid: 0
+          valid: 0
+        destination:
+          name: httproute/default/httproute-1/rule/1
+          settings:
+            - addressType: IP
+              endpoints:
+                - host: 7.7.7.7
+                  port: 8080
+              protocol: HTTP
+              weight: 1
+              name: httproute/default/httproute-1/rule/1/backend/0
+        security:
+          extAuth:
+            name: securitypolicy/default/policy-for-http-route-1
+            traffic:
+              retry:
+                numRetries: 2
+                perRetry:
+                  backOff:
+                    baseInterval: 200ms
+                    maxInterval: 3s
+                  timeout: 3s
+                retryOn:
+                  triggers:
+                    - 5xx
+                    - deadline-exceeded
+                    - internal
+                    - unavailable
+            failOpen: false
+            http:
+              authority: http-backend.default:9000
+              destination:
+                name: securitypolicy/default/policy-for-http-route-1/default/http-backend
+                settings:
+                  - addressType: IP
+                    endpoints:
+                      - host: 8.8.8.8
+                        port: 9000
+                    protocol: HTTP
+                    weight: 1
+                    name: securitypolicy/default/policy-for-http-route-1/default/http-backend/backend/0
+            headersToExtAuth:
+              - header1
+              - header2
+      - name: httproute/default/httproute-2/rule/0/match/0/www_bar_com
+        hostname: www.bar.com
+        isHTTP2: false
+        pathMatch:
+          distinct: false
+          name: ""
+          prefix: /bar
+        backendWeights:
+          invalid: 0
+          valid: 0
+        destination:
+          name: httproute/default/httproute-2/rule/0
+          settings:
+            - addressType: IP
+              endpoints:
+                - host: 7.7.7.7
+                  port: 8080
+              protocol: HTTP
+              weight: 1
+              name: httproute/default/httproute-2/rule/0/backend/0
+        security:
+          extAuth:
+            name: securitypolicy/default/policy-for-gateway-1
+            traffic:
+              retry:
+                numRetries: 2
+                perRetry:
+                  backOff:
+                    baseInterval: 200ms
+                    maxInterval: 3s
+                  timeout: 3s
+                retryOn:
+                  triggers:
+                    - 5xx
+                    - deadline-exceeded
+                    - internal
+                    - unavailable
+            failOpen: true
+            http:
+              authority: http-backend.envoy-gateway:80
+              destination:
+                name: securitypolicy/default/policy-for-gateway-1/envoy-gateway/http-backend
+                settings:
+                  - addressType: IP
+                    endpoints:
+                      - host: 7.7.7.7
+                        port: 80
+                    protocol: HTTP
+                    weight: 1
+                    name: securitypolicy/default/policy-for-gateway-1/envoy-gateway/http-backend/backend/0
+              headersToBackend:
+                - header1
+                - header2
+              path: /auth
diff --git a/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.clusters.yaml b/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.clusters.yaml
new file mode 100644
index 000000000..dcaf5cc85
--- /dev/null
+++ b/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.clusters.yaml
@@ -0,0 +1,120 @@
+- circuitBreakers:
+    thresholds:
+    - maxRetries: 1024
+  commonLbConfig: {}
+  connectTimeout: 10s
+  dnsLookupFamily: V4_PREFERRED
+  edsClusterConfig:
+    edsConfig:
+      ads: {}
+      resourceApiVersion: V3
+    serviceName: httproute/default/httproute-1/rule/0
+  ignoreHealthOnHostRemoval: true
+  lbPolicy: LEAST_REQUEST
+  loadBalancingPolicy:
+    policies:
+    - typedExtensionConfig:
+        name: envoy.load_balancing_policies.least_request
+        typedConfig:
+          '@type': type.googleapis.com/envoy.extensions.load_balancing_policies.least_request.v3.LeastRequest
+          localityLbConfig:
+            localityWeightedLbConfig: {}
+  name: httproute/default/httproute-1/rule/0
+  perConnectionBufferLimitBytes: 32768
+  type: EDS
+- circuitBreakers:
+    thresholds:
+    - maxRetries: 1024
+  commonLbConfig: {}
+  connectTimeout: 10s
+  dnsLookupFamily: V4_PREFERRED
+  edsClusterConfig:
+    edsConfig:
+      ads: {}
+      resourceApiVersion: V3
+    serviceName: httproute/default/httproute-1/rule/1
+  ignoreHealthOnHostRemoval: true
+  lbPolicy: LEAST_REQUEST
+  loadBalancingPolicy:
+    policies:
+    - typedExtensionConfig:
+        name: envoy.load_balancing_policies.least_request
+        typedConfig:
+          '@type': type.googleapis.com/envoy.extensions.load_balancing_policies.least_request.v3.LeastRequest
+          localityLbConfig:
+            localityWeightedLbConfig: {}
+  name: httproute/default/httproute-1/rule/1
+  perConnectionBufferLimitBytes: 32768
+  type: EDS
+- circuitBreakers:
+    thresholds:
+    - maxRetries: 1024
+  commonLbConfig: {}
+  connectTimeout: 10s
+  dnsLookupFamily: V4_PREFERRED
+  edsClusterConfig:
+    edsConfig:
+      ads: {}
+      resourceApiVersion: V3
+    serviceName: httproute/default/httproute-2/rule/0
+  ignoreHealthOnHostRemoval: true
+  lbPolicy: LEAST_REQUEST
+  loadBalancingPolicy:
+    policies:
+    - typedExtensionConfig:
+        name: envoy.load_balancing_policies.least_request
+        typedConfig:
+          '@type': type.googleapis.com/envoy.extensions.load_balancing_policies.least_request.v3.LeastRequest
+          localityLbConfig:
+            localityWeightedLbConfig: {}
+  name: httproute/default/httproute-2/rule/0
+  perConnectionBufferLimitBytes: 32768
+  type: EDS
+- circuitBreakers:
+    thresholds:
+    - maxRetries: 1024
+  commonLbConfig: {}
+  connectTimeout: 10s
+  dnsLookupFamily: V4_PREFERRED
+  edsClusterConfig:
+    edsConfig:
+      ads: {}
+      resourceApiVersion: V3
+    serviceName: securitypolicy/default/policy-for-http-route-1/default/http-backend
+  ignoreHealthOnHostRemoval: true
+  lbPolicy: LEAST_REQUEST
+  loadBalancingPolicy:
+    policies:
+    - typedExtensionConfig:
+        name: envoy.load_balancing_policies.least_request
+        typedConfig:
+          '@type': type.googleapis.com/envoy.extensions.load_balancing_policies.least_request.v3.LeastRequest
+          localityLbConfig:
+            localityWeightedLbConfig: {}
+  name: securitypolicy/default/policy-for-http-route-1/default/http-backend
+  perConnectionBufferLimitBytes: 32768
+  type: EDS
+- circuitBreakers:
+    thresholds:
+    - maxRetries: 1024
+  commonLbConfig: {}
+  connectTimeout: 10s
+  dnsLookupFamily: V4_PREFERRED
+  edsClusterConfig:
+    edsConfig:
+      ads: {}
+      resourceApiVersion: V3
+    serviceName: securitypolicy/default/policy-for-gateway-1/envoy-gateway/http-backend
+  ignoreHealthOnHostRemoval: true
+  lbPolicy: LEAST_REQUEST
+  loadBalancingPolicy:
+    policies:
+    - typedExtensionConfig:
+        name: envoy.load_balancing_policies.least_request
+        typedConfig:
+          '@type': type.googleapis.com/envoy.extensions.load_balancing_policies.least_request.v3.LeastRequest
+          localityLbConfig:
+            localityWeightedLbConfig: {}
+  name: securitypolicy/default/policy-for-gateway-1/envoy-gateway/http-backend
+  perConnectionBufferLimitBytes: 32768
+  type: EDS
diff --git a/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.endpoints.yaml b/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.endpoints.yaml
new file mode 100644
index 000000000..ec38cf183
--- /dev/null
+++ b/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.endpoints.yaml
@@ -0,0 +1,70 @@
+- clusterName: httproute/default/httproute-1/rule/0
+  endpoints:
+  - lbEndpoints:
+    - endpoint:
+        address:
+          socketAddress:
+            address: 7.7.7.7
+            portValue: 8080
+      loadBalancingWeight: 1
+    loadBalancingWeight: 1
+    locality:
+      region: httproute/default/httproute-1/rule/0/backend/0
+- clusterName: httproute/default/httproute-1/rule/1
+  endpoints:
+  - lbEndpoints:
+    - endpoint:
+        address:
+          socketAddress:
+            address: 7.7.7.7
+            portValue: 8080
+      loadBalancingWeight: 1
+    loadBalancingWeight: 1
+    locality:
+      region: httproute/default/httproute-1/rule/1/backend/0
+- clusterName: httproute/default/httproute-2/rule/0
+  endpoints:
+  - lbEndpoints:
+    - endpoint:
+        address:
+          socketAddress:
+            address: 7.7.7.7
+            portValue: 8080
+      loadBalancingWeight: 1
+    loadBalancingWeight: 1
+    locality:
+      region: httproute/default/httproute-2/rule/0/backend/0
+- clusterName: securitypolicy/default/policy-for-http-route-1/default/http-backend
+  endpoints:
+  - lbEndpoints:
+    - endpoint:
+        address:
+          socketAddress:
+            address: 8.8.4.4
+            portValue: 9001
+      loadBalancingWeight: 1
+    loadBalancingWeight: 1
+    locality:
+      region: securitypolicy/default/policy-for-http-route-1/default/http-backend/backend/0
+  - lbEndpoints:
+    - endpoint:
+        address:
+          socketAddress:
+            address: 8.8.8.8
+            portValue: 9000
+      loadBalancingWeight: 1
+    loadBalancingWeight: 1
+    locality:
+      region: securitypolicy/default/policy-for-http-route-1/default/http-backend/backend/1
+- clusterName: securitypolicy/default/policy-for-gateway-1/envoy-gateway/http-backend
+  endpoints:
+  - lbEndpoints:
+    - endpoint:
+        address:
+          socketAddress:
+            address: 7.7.7.7
+            portValue: 80
+      loadBalancingWeight: 1
+    loadBalancingWeight: 1
+    locality:
+      region: securitypolicy/default/policy-for-gateway-1/envoy-gateway/http-backend/backend/0
diff --git a/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.listeners.yaml b/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.listeners.yaml
new file mode 100644
index 000000000..6780d9706
--- /dev/null
+++ b/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.listeners.yaml
@@ -0,0 +1,82 @@
+- address:
+    socketAddress:
+      address: 0.0.0.0
+      portValue: 10080
+  defaultFilterChain:
+    filters:
+    - name: envoy.filters.network.http_connection_manager
+      typedConfig:
+        '@type': type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
+        commonHttpProtocolOptions:
+          headersWithUnderscoresAction: REJECT_REQUEST
+        http2ProtocolOptions:
+          initialConnectionWindowSize: 1048576
+          initialStreamWindowSize: 65536
+          maxConcurrentStreams: 100
+        httpFilters:
+        - disabled: true
+          name: envoy.filters.http.ext_authz/securitypolicy/default/policy-for-gateway-1
+          typedConfig:
+            '@type': type.googleapis.com/envoy.extensions.filters.http.ext_authz.v3.ExtAuthz
+            failureModeAllow: true
+            httpService:
+              authorizationResponse:
+                allowedUpstreamHeaders:
+                  patterns:
+                  - exact: header1
+                    ignoreCase: true
+                  - exact: header2
+                    ignoreCase: true
+              pathPrefix: /auth
+              retryPolicy:
+                numRetries: 2
+                retryBackOff:
+                  baseInterval: 0.200s
+                  maxInterval: 3s
+                retryOn: 5xx,deadline-exceeded,internal,unavailable
+              serverUri:
+                cluster: securitypolicy/default/policy-for-gateway-1/envoy-gateway/http-backend
+                timeout: 10s
+                uri: http://http-backend.envoy-gateway:80/auth
+            transportApiVersion: V3
+        - disabled: true
+          name: envoy.filters.http.ext_authz/securitypolicy/default/policy-for-http-route-1
+          typedConfig:
+            '@type': type.googleapis.com/envoy.extensions.filters.http.ext_authz.v3.ExtAuthz
+            allowedHeaders:
+              patterns:
+              - exact: header1
+                ignoreCase: true
+              - exact: header2
+                ignoreCase: true
+            httpService:
+              retryPolicy:
+                numRetries: 2
+                retryBackOff:
+                  baseInterval: 0.200s
+                  maxInterval: 3s
+                retryOn: 5xx,deadline-exceeded,internal,unavailable
+              serverUri:
+                cluster: securitypolicy/default/policy-for-http-route-1/default/http-backend
+                timeout: 10s
+                uri: http://http-backend.default:9000
+            transportApiVersion: V3
+        - name: envoy.filters.http.router
+          typedConfig:
+            '@type': type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
+            suppressEnvoyHeaders: true
+        mergeSlashes: true
+        normalizePath: true
+        pathWithEscapedSlashesAction: UNESCAPE_AND_REDIRECT
+        rds:
+          configSource:
+            ads: {}
+            resourceApiVersion: V3
+          routeConfigName: default/gateway-1/http
+        serverHeaderTransformation: PASS_THROUGH
+        statPrefix: http-10080
+        useRemoteAddress: true
+    name: default/gateway-1/http
+  maxConnectionsToAcceptPerSocketEvent: 1
+  name: default/gateway-1/http
+  perConnectionBufferLimitBytes: 32768
diff --git a/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.routes.yaml b/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.routes.yaml
new file mode 100644
index 000000000..08edfc3c4
--- /dev/null
+++ b/internal/xds/translator/testdata/out/xds-ir/ext-auth-http-retry.routes.yaml
@@ -0,0 +1,44 @@
+- ignorePortInHostMatching: true
+  name: default/gateway-1/http
+  virtualHosts:
+  - domains:
+    - www.foo.com
+    name: default/gateway-1/http/www_foo_com
+    routes:
+    - match:
+        pathSeparatedPrefix: /foo1
+      name: httproute/default/httproute-1/rule/0/match/0/www_foo_com
+      route:
+        cluster: httproute/default/httproute-1/rule/0
+        upgradeConfigs:
+        - upgradeType: websocket
+      typedPerFilterConfig:
+        envoy.filters.http.ext_authz/securitypolicy/default/policy-for-http-route-1:
+          '@type': type.googleapis.com/envoy.config.route.v3.FilterConfig
+          config: {}
+    - match:
+        pathSeparatedPrefix: /foo2
+      name: httproute/default/httproute-1/rule/1/match/0/www_foo_com
+      route:
+        cluster: httproute/default/httproute-1/rule/1
+        upgradeConfigs:
+        - upgradeType: websocket
+      typedPerFilterConfig:
+        envoy.filters.http.ext_authz/securitypolicy/default/policy-for-http-route-1:
+          '@type': type.googleapis.com/envoy.config.route.v3.FilterConfig
+          config: {}
+  - domains:
+    - www.bar.com
+    name: default/gateway-1/http/www_bar_com
+    routes:
+    - match:
+        pathSeparatedPrefix: /bar
+      name: httproute/default/httproute-2/rule/0/match/0/www_bar_com
+      route:
+        cluster: httproute/default/httproute-2/rule/0
+        upgradeConfigs:
+        - upgradeType: websocket
+      typedPerFilterConfig:
+        envoy.filters.http.ext_authz/securitypolicy/default/policy-for-gateway-1:
+          '@type': type.googleapis.com/envoy.config.route.v3.FilterConfig
+          config: {}
diff --git a/internal/xds/translator/testdata/out/xds-ir/ext-auth-retry.listeners.yaml b/internal/xds/translator/testdata/out/xds-ir/ext-auth-retry.listeners.yaml
index cd1dfed29..ad501f621 100644
--- a/internal/xds/translator/testdata/out/xds-ir/ext-auth-retry.listeners.yaml
+++ b/internal/xds/translator/testdata/out/xds-ir/ext-auth-retry.listeners.yaml
@@ -28,6 +28,12 @@
                   - exact: header2
                     ignoreCase: true
               pathPrefix: /auth
+              retryPolicy:
+                numRetries: 2
+                retryBackOff:
+                  baseInterval: 0.200s
+                  maxInterval: 3s
+                retryOn: 5xx,deadline-exceeded,internal,unavailable
               serverUri:
                 cluster: securitypolicy/default/policy-for-gateway-1/envoy-gateway/http-backend
                 timeout: 10s
diff --git a/site/content/en/latest/api/extension_types.md b/site/content/en/latest/api/extension_types.md
index 10ca4e6c5..c1ea50f53 100644
--- a/site/content/en/latest/api/extension_types.md
+++ b/site/content/en/latest/api/extension_types.md
@@ -1391,6 +1391,7 @@ _Appears in:_
 | ---   | ---  | ---      | ---     | ---         |
 | `rateLimitDeployment` | _[KubernetesDeploymentSpec](#kubernetesdeploymentspec)_ |  false  |  | RateLimitDeployment defines the desired state of the Envoy ratelimit deployment resource.<br />If unspecified, default settings for the managed Envoy ratelimit deployment resource<br />are applied. |
 | `rateLimitHpa` | _[KubernetesHorizontalPodAutoscalerSpec](#kuberneteshorizontalpodautoscalerspec)_ |  false  |  | RateLimitHpa defines the Horizontal Pod Autoscaler settings for Envoy ratelimit Deployment.<br />If the HPA is set, Replicas field from RateLimitDeployment will be ignored. |
+| `rateLimitPdb` | _[KubernetesPodDisruptionBudgetSpec](#kubernetespoddisruptionbudgetspec)_ |  false  |  | RateLimitPDB allows to control the pod disruption budget of rate limit service. |
 | `watch` | _[KubernetesWatchMode](#kuberneteswatchmode)_ |  false  |  | Watch holds configuration of which input resources should be watched and reconciled. |
 | `leaderElection` | _[LeaderElection](#leaderelection)_ |  false  |  | LeaderElection specifies the configuration for leader election.<br />If it's not set up, leader election will be active by default, using Kubernetes' standard settings. |
 | `shutdownManager` | _[ShutdownManager](#shutdownmanager)_ |  false  |  | ShutdownManager defines the configuration for the shutdown manager. |
@@ -3059,6 +3060,7 @@ _Appears in:_
 KubernetesPodDisruptionBudgetSpec defines Kubernetes PodDisruptionBudget settings of Envoy Proxy Deployment.
 
 _Appears in:_
+- [EnvoyGatewayKubernetesProvider](#envoygatewaykubernetesprovider)
 - [EnvoyProxyKubernetesProvider](#envoyproxykubernetesprovider)
 
 | Field | Type | Required | Default | Description |