Repository: kubernetes/kubernetes (language: go)

Issue:
Implement a kube-proxy conntrack reconciler
### What would you like to be added?

We'd like to have conntrack reconciler for Services in kube-proxy, for each Service or Endpoint change it should reconcile the Service/Endpoint table with the existing conntrack table and remove stale UDP entries (we have concluded in other issues that UDP is the only protocol that requires this)

### Why is this needed?

Despite a lot of work and testing was added to solve the problem with stale UDP entries that can blackhole traffic, we still have reports of issues related to UDP stale entries

https://github.com/kubernetes/kubernetes/issues/125467

Also, the conntrack logic is event based and, as expected, if something fails it never reconciles

https://github.com/kubernetes/kubernetes/issues/112604

The reconciler will only delete the entries that are known to be stale, other entries will time out.

### References

Some WIP I had https://github.com/kubernetes/kubernetes/compare/master...aojea:kubernetes:conntrack_done_right?expand=1

### Related issues that will be fixed

- https://github.com/kubernetes/kubernetes/issues/125979
- https://github.com/kubernetes/kubernetes/issues/125467
- https://github.com/kubernetes/kubernetes/issues/122740



Discussion / hints:
/sig network
/assign @aroradaman 
/cc @danwinship @thockin 
It is important to understand the performance implications, as listing all the conntrack entries is expensive, I remember talking with Florian Westphal (Kernel developer) about some way of doing server side filtering in the kernel, and avoid having to dump the entire conntrack table and filter flow by flow, but I don't know what is the state of this
The netlink interface to conntrack has the ability to get specific flows and dump all flows that match a given mark and mask.
/triage accepted
/kind bug

Despite I added the feature label, is basically rewriting the existing logic to fix the limitations of current implementation
Hi @aroradaman! Just checking in, as we talked about this one on the SIG network sync today: Have any questions or need any support with this one?
> Hi @aroradaman! Just checking in, as we talked about this one on the SIG network sync today: Have any questions or need any support with this one?

Was waiting for https://github.com/vishvananda/netlink/pull/989 to be merged, will resume the work on this now.
Prior to Kubernetes 1.28 we were bitten by a conntrack issue on UDP services with iTP set to Local.
When the destination Pod was recreated the conntrack entry persisted.

We fixed this by switching iTP to Cluster. 
Since Kubernetes 1.29 (on GKE) we've started getting this problem again, but now when the Service's iTP is set to Cluster.
But only when a node-upgrade happens (I can't reproduce it any other way).

So I appreciate you working on the reconciler. I'm looking forward to this problem going away.
I've been trying to get my head around this problem some more. In AWS (with the AWS CNI) on Kubernetes v1.29.6 (installed by kOps) I seem to be seeing a behaviour where a conntrack entry is being left behind by the CNI.

Our situation is that we multiple pods, many clients and a single server (the server is a DaemonSet).
The clients are streaming UDP over 8125 to the server.
The clients are using a ClusterIP service, with the iTP set to Local.

When everything is working, I see a conntrack entry:
```
udp      17 30 src=172.24.37.89 dst=100.69.132.196 sport=39931 dport=8125 [UNREPLIED] src=172.24.57.127 dst=172.24.37.89 sport=8125 dport=39931 mark=0 use=1
```

When the server pod is deleted, I see these kube-proxy logs:
```
I0815 11:40:56.724267      11 cleanup.go:63] "Deleting conntrack stale entries for services" IPs=[]
I0815 11:40:56.724290      11 cleanup.go:69] "Deleting conntrack stale entries for services" nodePorts=[]
I0815 11:40:56.724374      11 conntrack.go:66] "Clearing conntrack entries" parameters=["-D","--orig-dst","100.69.132.196","--dst-nat","172.24.57.127","-p","udp"]
I0815 11:40:56.781317      11 conntrack.go:71] "Conntrack entries deleted" output=<
	conntrack v1.4.7 (conntrack-tools): 7 flow entries have been deleted.
	udp      17 28 src=172.24.37.89 dst=100.69.132.196 sport=59909 dport=8125 [UNREPLIED] src=172.24.57.127 dst=172.24.37.89 sport=8125 dport=59909 mark=0 use=1
	udp      17 28 src=172.24.37.89 dst=100.69.132.196 sport=60287 dport=8125 [UNREPLIED] src=172.24.57.127 dst=172.24.37.89 sport=8125 dport=60287 mark=0 use=1
	udp      17 28 src=172.24.37.89 dst=100.69.132.196 sport=55964 dport=8125 [UNREPLIED] src=172.24.57.127 dst=172.24.37.89 sport=8125 dport=55964 mark=0 use=1
	udp      17 30 src=172.24.37.89 dst=100.69.132.196 sport=41937 dport=8125 [UNREPLIED] src=172.24.57.127 dst=172.24.37.89 sport=8125 dport=41937 mark=0 use=52
	udp      17 28 src=172.24.37.89 dst=100.69.132.196 sport=50068 dport=8125 [UNREPLIED] src=172.24.57.127 dst=172.24.37.89 sport=8125 dport=50068 mark=0 use=1
	udp      17 28 src=172.24.37.89 dst=100.69.132.196 sport=39931 dport=8125 [UNREPLIED] src=172.24.57.127 dst=172.24.37.89 sport=8125 dport=39931 mark=0 use=1
	udp      17 28 src=172.24.37.89 dst=100.69.132.196 sport=42561 dport=8125 [UNREPLIED] src=172.24.57.127 dst=172.24.37.89 sport=8125 dport=42561 mark=0 use=1
 >
```

After this time, I have the following conntrack entry:
```
udp      17 30 src=172.24.37.89 dst=100.69.132.196 sport=41937 dport=8125 [UNREPLIED] src=100.69.132.196 dst=172.24.33.81 sport=8125 dport=1267 mark=128 use=3
```

(notice the mark=128)

Later on, when the new server Pod has started, I see the following:
```
I0815 11:41:26.590933      11 cleanup.go:63] "Deleting conntrack stale entries for services" IPs=[]
I0815 11:41:26.590965      11 cleanup.go:69] "Deleting conntrack stale entries for services" nodePorts=[]
```

I'm not too familiar with this entire process, but it seems like the [AWS-CNI](https://github.com/aws/amazon-vpc-cni-k8s/blob/v1.18.1/pkg/networkutils/network.go#L113-L116) is adding a conntrack entry, which isn't being cleaned up by kube-proxy.

I'm not sure if the new reconciler will take this situation into account, so I thought I'd mention it.
Same setup as before, but in GCP with GKE v1.29.6-gke.1326000:

The clients are streaming UDP over 8125 to the server.
The clients are using a ClusterIP service, with the iTP set to Local.

When everything is working, I see a conntrack entry:
```
udp      17 30 src=10.11.4.5 dst=10.12.201.0 sport=48931 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.5 sport=8125 dport=48931 mark=0 use=1
```

When the server pod is deleted, I see these kube-proxy logs:
```
I0815 12:15:14.624545       1 cleanup.go:63] "Deleting conntrack stale entries for services" IPs=[]
I0815 12:15:14.624580       1 cleanup.go:69] "Deleting conntrack stale entries for services" nodePorts=[]
I0815 12:15:14.624646       1 conntrack.go:66] "Clearing conntrack entries" parameters=["-D","--orig-dst","10.12.201.0","--dst-nat","10.11.4.22","-p","udp"]
I0815 12:15:14.637712       1 conntrack.go:71] "Conntrack entries deleted" output=<
	udp      17 23 src=10.11.4.30 dst=10.12.201.0 sport=34152 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.30 sport=8125 dport=34152 mark=0 use=1
	udp      17 26 src=10.11.4.38 dst=10.12.201.0 sport=50751 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.38 sport=8125 dport=50751 mark=0 use=1
	udp      17 28 src=10.11.4.5 dst=10.12.201.0 sport=48931 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.5 sport=8125 dport=48931 mark=0 use=1
	udp      17 28 src=10.11.4.5 dst=10.12.201.0 sport=42608 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.5 sport=8125 dport=42608 mark=0 use=1
	udp      17 21 src=10.11.4.56 dst=10.12.201.0 sport=50648 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.56 sport=8125 dport=50648 mark=0 use=1
	udp      17 21 src=10.11.4.64 dst=10.12.201.0 sport=60746 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.64 sport=8125 dport=60746 mark=0 use=1
	udp      17 18 src=10.11.4.20 dst=10.12.201.0 sport=40877 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.20 sport=8125 dport=40877 mark=0 use=1
	udp      17 20 src=10.11.4.55 dst=10.12.201.0 sport=56370 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.55 sport=8125 dport=56370 mark=0 use=1
	udp      17 12 src=10.11.4.64 dst=10.12.201.0 sport=37004 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.64 sport=8125 dport=37004 mark=0 use=1
	udp      17 28 src=10.11.4.50 dst=10.12.201.0 sport=57899 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.50 sport=8125 dport=57899 mark=0 use=1
	udp      17 29 src=10.11.4.113 dst=10.12.201.0 sport=58278 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.113 sport=8125 dport=58278 mark=0 use=1
	udp      17 26 src=10.11.4.8 dst=10.12.201.0 sport=40499 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.8 sport=8125 dport=40499 mark=0 use=1
	udp      17 20 src=10.11.4.67 dst=10.12.201.0 sport=48386 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.67 sport=8125 dport=48386 mark=0 use=1
	udp      17 29 src=10.11.4.50 dst=10.12.201.0 sport=55565 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.50 sport=8125 dport=55565 mark=0 use=1
	udp      17 29 src=10.11.4.61 dst=10.12.201.0 sport=36647 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.61 sport=8125 dport=36647 mark=0 use=1
	udp      17 29 src=10.11.4.59 dst=10.12.201.0 sport=52344 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.59 sport=8125 dport=52344 mark=0 use=1
	udp      17 28 src=10.11.4.5 dst=10.12.201.0 sport=35652 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.5 sport=8125 dport=35652 mark=0 use=1
	udp      17 22 src=10.11.4.145 dst=10.12.201.0 sport=46692 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.145 sport=8125 dport=46692 mark=0 use=1
	udp      17 28 src=10.11.4.5 dst=10.12.201.0 sport=43098 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.5 sport=8125 dport=43098 mark=0 use=1
	udp      17 29 src=10.11.4.59 dst=10.12.201.0 sport=51314 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.59 sport=8125 dport=51314 mark=0 use=1
	udp      17 29 src=10.11.4.70 dst=10.12.201.0 sport=48682 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.70 sport=8125 dport=48682 mark=0 use=1
	udp      17 20 src=10.11.4.65 dst=10.12.201.0 sport=50380 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.65 sport=8125 dport=50380 mark=0 use=1
	udp      17 16 src=10.11.4.217 dst=10.12.201.0 sport=44527 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.217 sport=8125 dport=44527 mark=0 use=1
	udp      17 21 src=10.11.4.66 dst=10.12.201.0 sport=37202 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.66 sport=8125 dport=37202 mark=0 use=1
	udp      17 24 src=10.11.4.31 dst=10.12.201.0 sport=38039 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.31 sport=8125 dport=38039 mark=0 use=1
	udp      17 22 src=10.11.4.69 dst=10.12.201.0 sport=39822 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.69 sport=8125 dport=39822 mark=0 use=1
	udp      17 29 src=10.11.4.143 dst=10.12.201.0 sport=52499 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.143 sport=8125 dport=52499 mark=0 use=1
	udp      17 21 src=10.11.4.64 dst=10.12.201.0 sport=46370 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.64 sport=8125 dport=4conntrack v1.4.7 (conntrack-tools): 37 flow entries have been deleted.
	6370 mark=0 use=1
	udp      17 1 src=10.11.4.27 dst=10.12.201.0 sport=41082 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.27 sport=8125 dport=41082 mark=0 use=1
	udp      17 28 src=10.11.4.5 dst=10.12.201.0 sport=40601 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.5 sport=8125 dport=40601 mark=0 use=1
	udp      17 24 src=10.11.4.64 dst=10.12.201.0 sport=57893 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.64 sport=8125 dport=57893 mark=0 use=1
	udp      17 27 src=10.11.4.2 dst=10.12.201.0 sport=32932 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.2 sport=8125 dport=32932 mark=0 use=1
	udp      17 16 src=10.11.4.64 dst=10.12.201.0 sport=47022 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.64 sport=8125 dport=47022 mark=0 use=1
	udp      17 28 src=10.11.4.5 dst=10.12.201.0 sport=40394 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.5 sport=8125 dport=40394 mark=0 use=1
	udp      17 29 src=10.11.4.61 dst=10.12.201.0 sport=47876 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.61 sport=8125 dport=47876 mark=0 use=1
	udp      17 29 src=10.11.4.24 dst=10.12.201.0 sport=40378 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.24 sport=8125 dport=40378 mark=0 use=1
	udp      17 30 src=10.11.4.5 dst=10.12.201.0 sport=46634 dport=8125 [UNREPLIED] src=10.11.4.22 dst=10.11.4.5 sport=8125 dport=46634 mark=0 use=5
```

Once the Pod starts up again, I have the incorrect conntrack entry:
```
udp      17 30 src=10.11.4.5 dst=10.12.201.0 sport=46634 dport=8125 [UNREPLIED] src=10.12.201.0 dst=10.11.4.5 sport=8125 dport=46634 mark=0 use=5
```
(notice the `src=10.12.201.0` on the right)

If I delete it, a new one appears:
```
udp      17 30 src=10.11.4.5 dst=10.12.201.0 sport=46634 dport=8125 [UNREPLIED] src=10.11.4.23 dst=10.11.4.5 sport=8125 dport=46634 mark=0 use=1
```

I'm unsure what it causing that incorrect conntrack entry, I assume this is the blackhole route problem that has been plaguing kube-proxy.

 in 1.28 there was a refactor of the conntrack logic https://github.com/kubernetes/kubernetes/pull/116104 but I think is correct, and I couldn't find anything suspicious 

> (notice the src=10.12.201.0 on the right)

no nat is performed so it keeps the original IP 

>If I delete it, a new one appears:

that is the client `10.11.4.5` connection to the service IP `10.12.201.0` on port `8125` is DNATed to `10.11.4.23`  port `8125`

We have a test that exercise this scenario, at the moment that the new pod is created, in your case the one with  `10.11.4.23` , kube-proxy should delete that entry 

> I'm unsure what it causing that incorrect conntrack entry, I assume this is the blackhole route problem that has been plaguing kube-proxy.

@adrianmoisey is this a reproducer or a hypothesis? if you have a repro in gke let me know so I can take a look

> in 1.28 there was a refactor of the conntrack logic #116104 but I think is correct, and I couldn't find anything suspicious

We noticed a behaviour change between 1.28 and 1.29, which I suspect may be related to https://github.com/kubernetes/kubernetes/pull/119394

> @adrianmoisey is this a reproducer or a hypothesis? if you have a repro in gke let me know so I can take a look

I can reproduce in GKE v1.29.6-gke.1326000
Linking the PR to remove dependency from conntrack binary here - https://github.com/kubernetes/kubernetes/pull/126847
With https://github.com/vishvananda/netlink/pull/989 we can delete all the conntrack entries in single GET/DUMP conntrack table command, reducing the number of netlink calls. That will require a change in the conntrack.Interface to have single generic method to clear all entries, should I include that in this PR or a new one after this gets in?
```
// Interface for dealing with conntrack
type Interface interface {
	// ClearEntriesForIP deletes conntrack entries for the give flow filters.
	ClearEntriesForFlow(filters ...conntrackFilter) error

```
> should I include that in this PR or a new one after this gets in?

My opinion is to include it in a new PR.
I don't think we handle the conntrack cleanup for services that transition from eTP to iTP. Ideally after the transition, all the entries DNATted to cluster endpoints should be removed, will also add an e2e for the same.
> I don't think we handle the conntrack cleanup for services that transition from eTP to iTP. Ideally after the transition, all the entries DNATted to cluster endpoints should be removed, will also add an e2e for the same.

Currently, the service changes(traffic policy and topology) only apply to the new connections, with the reconciler we have the opportunity to apply the changes to the existing connection as well.
@danwinship @aojea 
> Currently, the service changes(traffic policy and topology) only apply to the new connections, with the reconciler we have the opportunity to apply the changes to the existing connection as well. @danwinship @aojea

with the reconciler we can fix all of that as we no longer will depend on the events, so we don't care about transitions anymore.

Reconciler will check desired state by parsing all the services and its characteristics, will dump the conntrack table , and delete the conntrack entries we know will blackhole traffic ... it is UDP so we don't need to react immediately 
> I don't think we handle the conntrack cleanup for services that transition from eTP to iTP. Ideally after the transition, all the entries DNATted to cluster endpoints should be removed

I'm not sure there is really a good use case for switching a live service between eTP and iTP, but if we do care about that, it seems more consistent to me to preserve existing connections rather than killing them.

EDIT: I think by "transition from eTP to iTP", Daman meant "transition either eTP or iTP between Local and Cluster", and that's certainly what _I_ meant...
> I'm not sure there is really a good use case for switching a live service between eTP and iTP, but if we do care about that, it seems more consistent to me to preserve existing connections rather than killing them.
> 
> EDIT: I think by "transition from eTP to iTP", Daman meant "transition either eTP or iTP between Local and Cluster", and that's certainly what _I_ meant...

Speaking about changing iTP from Cluster to Local here:
I had a use case where a vendor's Helm chart changed their Service iTP from Cluster to Local for a DaemonSet that receives statsd metrics over UDP. Since iTP of Local is a "fairly" recent addition I understand why they did that, since I assume many of their customers had clusters created prior to Kubernetes 1.26 (when they didn't have the option of Local)

In this particular use case, I'd expect the blackhole contract entries to be flushed, at some point after the ITP changes, in order to send traffic to the local Pod.

I can't say for certain if my expectation matches other use cases though.
> with the reconciler we can fix all of that as we no longer will depend on the events

Currently, when a service is deleted, we delete all existing DNAT records for its ClusterIP (`DeletedUDPClusterIPs`). You can't figure that from reconciling, because you can't distinguish "a DNAT record for a ClusterIP that isn't used by any Service" from "a DNAT record for someone else's IP that has nothing to do with us". (Well, OK, if you have ServiceCIDRs you can figure it out.)

Then again, I'm not actually sure why we delete those entries... the original UDP conntrack problem bug report (#19029) only described the problem that is fixed by the `DeletedUDPEndpoints` case, but the fix (#22573) included the `DeletedUDPClusterIPs` part too. (`NewlyActiveUDPServices` came later.) Also, it's conspicuous that we delete records for ClusterIPs, but _not_ for externalIPs, LB IPs, and NodePorts, and _no one has ever reported this_. It's possible this cleanup is actually pointless.

(I guess it has the effect of immediately killing UDP Service connections when you delete the Service, even if the endpoints still exist, but that behavior is the opposite of how TCP Services behave, so that doesn't seem like a good argument for keeping it...)
I'm sorry for being pedantic but there is no connections in UDP, we delete a conntrack entry, and that does not have any problem, we are not breaking traffic .... an application that is using UDP can not make the assumption that has a connection and is always going to hit the same backend 
I'm not talking about "same backend", I'm talking about "at all".

Deleting a TCP Service while its endpoints are still running results in graceful termination of its clients; you can't create new connections, but existing connections continue to function for as long as the endpoints are there.

Deleting a UDP Service while its endpoints are still running results in immediate termination of all clients; further packets using previously-established 5-tuples will be dropped.

And this is something the code does deliberately; it _could_ implement immediate termination in the TCP case (by deleting conntrack records for deleted TCP ClusterIPs too), or graceful termination in the UDP case (by _not_ deleting conntrack records for deleted UDP ClusterIPs), but it doesn't. And this doesn't seem to be have been an intentional decision. (And it only applies to ClusterIPs! ExternalIPs, LB IPs, and NodePorts get graceful termination for both TCP and UDP.)

Akshually...

I recall having this discussion in the sig a couple years ago. I'm on mobile now but I can try to find the issue later. We talked about whether it is correct to leave TCP connections open when the service has been deleted, or whether we should break it from the middle. Does NOT break TCP sessions the way you think it might. I was pretty surprised.  I tried but I did not find an obvious way to break a TCP session from the middle, short of a black hole.

IIRC we didn't really come to a conclusion about whether it's right or wrong, but it didn't matter because we didn't have a mechanism. 

Maybe ebpf has a better way of doing this, or maybe using the netlink library, rather than the conntrack tool?
> Deleting a UDP Service while its endpoints are still running results in immediate termination of all clients; further packets using previously-established 5-tuples will be dropped.

doh, I see it now, I was not getting the point ,  sorry ... but we never need to delete entries if there are no backends, we only need to delete entries when there is at least one backend, because the other entries may blackhole the legit traffic, if there are no backends we let the packets flow

@thockin I think you mean this https://github.com/kubernetes/kubernetes/issues/108523

> Maybe ebpf has a better way of doing this

yeah, it does 
> I tried but I did not find an obvious way to break a TCP session from the middle, short of a black hole.

@thockin in this case we don't want to leave the connection hanging, since TCP has long timers, even if we achieve to delete the entry I think we want to signal both ends sending a RST so they don't leave sockets open
> Maybe ebpf has a better way of doing this

ebpf programs can read arbitrary conntrack entries, while netfilter rules can only see the results of applying the current conntrack entries to the current connection... but it doesn't seem like reading arbitrary entries would really be helpful.

For nftables kube-proxy, "Drop any packets to IPs in the ServiceCIDR range that aren't currently-active ClusterIPs" is just a recombination of the two behaviors in #122692, and should be doable with one or two new chains with the right hook+priority, with a single rule.

> or maybe using the netlink library, rather than the conntrack tool?

The big thing the netlink library gives us is that we can delete arbitrary sets of entries (in particular, now we can delete blackhole entries for a given IP without also deleting DNAT entries for that IP, which we couldn't do before, #122741), so if the problem before had been "there's no way to delete _just_ the conntrack entries we want to delete", then the library should help. But if the problem was "I deleted all the conntrack entries related to the cluster IP and open connections still stayed open" then the netlink library probably won't help.

FTR: https://github.com/kubernetes/kubernetes/issues/108523#issuecomment-1079290227 and down was an interesting refresher on the stat of things.


> But if the problem was "I deleted all the conntrack entries related to the cluster IP and open connections still stayed open" then the netlink library probably won't help.

I repeated the experiment in https://github.com/kubernetes/kubernetes/issues/108523#issuecomment-1072603233 and got the same results, whether client and server were on the same node or different (through a clusterIP).  Perhaps `conntrack -F` is not aggressive enough?  Or more likely, the conntrack listing is a reflection of other state (the established TCP session) and thus can "helpfully" be rebuilt.

So again, if the intention is to break a TCP connection "from the middle", I don't think conntrack is the mechanism, or at least I am not using it right.
Can we set custom marks on conntrack entries that have been DNATed by the proxy?

Currently, we use marking at the packet level, but this is limited to masquerade hairpin traffic, and we also remove the mark in post-routing. By using `CONNMARK` / `ct mark` we can set a mark on the flow entry itself, which will help distinguish flows that are created or DNATed by the proxy. From a reconciliation perspective, this marking can be used as a filter to prevent accidental cleanup of non-kube-proxy owned flows.
> Can we set custom marks on conntrack entries that have been DNATed by the proxy?
> 
> Currently, we use marking at the packet level, but this is limited to masquerade hairpin traffic, and we also remove the mark in post-routing. By using `CONNMARK` / `ct mark` we can set a mark on the flow entry itself, which will help distinguish flows that are created or DNATed by the proxy. From a reconciliation perspective, this marking can be used as a filter to prevent accidental cleanup of non-kube-proxy owned flows.

ClusterIPs can be filtered out using ServiceCIDRs, CONNMARK will help to distinguish DNAT flows for LoadBalancerIP, ExternalIP and NodePorts that are owned by proxy.

The current approach which I'm experimenting is clearing flow entries which have ClusterIP/ExternalIP/LoadBalancerIP [--orig-dst] or NodePort [--orig-port-dst] as origin and DNATed destination [--reply-src] doesn't belong to any of the serving endpoints.
> ONNMARK will help to distinguish DNAT flows for LoadBalancerIP, ExternalIP and NodePorts that are owned by proxy.

I have two concerns with this approach:
- mark becomes APIs and a point of conflicts with other elements setting marks, Dan had a nice write up on the iptables case in https://github.com/kubernetes/enhancements/tree/master/keps/sig-network/3178-iptables-cleanup#eliminating-kube-mark-drop
- I think that independently who creates the entries, if something external create a conntrack entry that blackhole the ClusterIP/ExternalIP/ ... of. a kubernetes Service, kube-proxy should delete it anyway, as it should be authoritative on the system to do that

> The current approach which I'm experimenting is clearing flow entries which have ClusterIP/ExternalIP/LoadBalancerIP [--orig-dst] or NodePort [--orig-port-dst] as origin and DNATed destination [--reply-src] doesn't belong to any of the serving endpoints.

I think is correct, it is simple and solve all the problems: And UDP entry that is stale on an active Service, by active it has at least one endpoint, should be deleted any sync period

Fix patch:
diff --git a/pkg/proxy/conntrack/cleanup.go b/pkg/proxy/conntrack/cleanup.go
index c39794c919b04..e23cada043a6f 100644
--- a/pkg/proxy/conntrack/cleanup.go
+++ b/pkg/proxy/conntrack/cleanup.go
@@ -20,6 +20,8 @@ limitations under the License.
 package conntrack
 
 import (
+	"time"
+
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
 
@@ -27,84 +29,96 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/proxy"
-	proxyutil "k8s.io/kubernetes/pkg/proxy/util"
 	netutils "k8s.io/utils/net"
 )
 
-// CleanStaleEntries takes care of flushing stale conntrack entries for services and endpoints.
-func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily, svcPortMap proxy.ServicePortMap,
-	serviceUpdateResult proxy.UpdateServiceMapResult, endpointsUpdateResult proxy.UpdateEndpointsMapResult) {
-	deleteStaleServiceConntrackEntries(ct, ipFamily, svcPortMap, serviceUpdateResult, endpointsUpdateResult)
-	deleteStaleEndpointConntrackEntries(ct, ipFamily, svcPortMap, endpointsUpdateResult)
-}
+// CleanStaleEntries scans conntrack table and removes any entries
+// for a service that do not correspond to a serving endpoint.
+func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
+	svcPortMap proxy.ServicePortMap, endpointsMap proxy.EndpointsMap) {
 
-// deleteStaleServiceConntrackEntries takes care of flushing stale conntrack entries related
-// to UDP Service IPs. When a service has no endpoints and we drop traffic to it, conntrack
-// may create "black hole" entries for that IP+port. When the service gets endpoints we
-// need to delete those entries so further traffic doesn't get dropped.
-func deleteStaleServiceConntrackEntries(ct Interface, ipFamily v1.IPFamily, svcPortMap proxy.ServicePortMap, serviceUpdateResult proxy.UpdateServiceMapResult, endpointsUpdateResult proxy.UpdateEndpointsMapResult) {
-	var filters []netlink.CustomConntrackFilter
-	conntrackCleanupServiceIPs := serviceUpdateResult.DeletedUDPClusterIPs
-	conntrackCleanupServiceNodePorts := sets.New[int]()
-
-	// merge newly active services gathered from endpointsUpdateResult
-	// a UDP service that changes from 0 to non-0 endpoints is newly active.
-	for _, svcPortName := range endpointsUpdateResult.NewlyActiveUDPServices {
-		if svcInfo, ok := svcPortMap[svcPortName]; ok {
-			klog.V(4).InfoS("Newly-active UDP service may have stale conntrack entries", "servicePortName", svcPortName)
-			conntrackCleanupServiceIPs.Insert(svcInfo.ClusterIP().String())
-			for _, extIP := range svcInfo.ExternalIPs() {
-				conntrackCleanupServiceIPs.Insert(extIP.String())
-			}
-			for _, lbIP := range svcInfo.LoadBalancerVIPs() {
-				conntrackCleanupServiceIPs.Insert(lbIP.String())
-			}
-			nodePort := svcInfo.NodePort()
-			if svcInfo.Protocol() == v1.ProtocolUDP && nodePort != 0 {
-				conntrackCleanupServiceNodePorts.Insert(nodePort)
-			}
-		}
-	}
+	start := time.Now()
+	klog.V(4).InfoS("Started to reconcile conntrack entries", "ipFamily", ipFamily)
 
-	klog.V(4).InfoS("Deleting conntrack stale entries for services", "IPs", conntrackCleanupServiceIPs.UnsortedList())
-	for _, svcIP := range conntrackCleanupServiceIPs.UnsortedList() {
-		filters = append(filters, filterForIP(svcIP, v1.ProtocolUDP))
-	}
-	klog.V(4).InfoS("Deleting conntrack stale entries for services", "nodePorts", conntrackCleanupServiceNodePorts.UnsortedList())
-	for _, nodePort := range conntrackCleanupServiceNodePorts.UnsortedList() {
-		filters = append(filters, filterForPort(nodePort, v1.ProtocolUDP))
+	entries, err := ct.ListEntries(ipFamilyMap[ipFamily])
+	if err != nil {
+		klog.ErrorS(err, "Failed to list conntrack entries")
+		return
 	}
 
-	if err := ct.ClearEntries(ipFamilyMap[ipFamily], filters...); err != nil {
-		klog.ErrorS(err, "Failed to delete stale service connections")
+	// serviceIPEndpointIPs maps service IPs (ClusterIP, LoadBalancerIPs and ExternalIPs)
+	// to the set of serving endpoint IPs.
+	serviceIPEndpointIPs := make(map[string]sets.Set[string])
+	// serviceNodePortEndpointIPs maps service NodePort to the set of serving endpoint IPs.
+	serviceNodePortEndpointIPs := make(map[int]sets.Set[string])
+
+	for svcName, svc := range svcPortMap {
+		// we are only interested in UDP services
+		if svc.Protocol() != v1.ProtocolUDP {
+			continue
+		}
+
+		endpointIPs := sets.New[string]()
+		for _, endpoint := range endpointsMap[svcName] {
+			// We need to remove all the conntrack entries for a Service (IP or NodePort)
+			// that are not pointing to a serving endpoint.
+			// We map all the serving endpoint IPs to the service and clear all the conntrack
+			// entries which are destined for the service and are not DNATed to these endpoints.
+			// Changes to the service should not affect existing flows, so we do not take
+			// traffic policies, topology, or terminating status of the service into account.
+			// This ensures that the behavior of UDP services remains consistent with TCP
+			// services.
+			if endpoint.IsServing() {
+				endpointIPs.Insert(endpoint.IP())
+			}
+		}
+
+		serviceIPEndpointIPs[svc.ClusterIP().String()] = endpointIPs
+		for _, loadBalancerIP := range svc.LoadBalancerVIPs() {
+			serviceIPEndpointIPs[loadBalancerIP.String()] = endpointIPs
+		}
+		for _, externalIP := range svc.ExternalIPs() {
+			serviceIPEndpointIPs[externalIP.String()] = endpointIPs
+		}
+		if svc.NodePort() != 0 {
+			serviceNodePortEndpointIPs[svc.NodePort()] = endpointIPs
+		}
 	}
-}
 
-// deleteStaleEndpointConntrackEntries takes care of flushing stale conntrack entries related
-// to UDP endpoints. After a UDP endpoint is removed we must flush any conntrack entries
-// for it so that if the same client keeps sending, the packets will get routed to a new endpoint.
-func deleteStaleEndpointConntrackEntries(ct Interface, ipFamily v1.IPFamily, svcPortMap proxy.ServicePortMap, endpointsUpdateResult proxy.UpdateEndpointsMapResult) {
 	var filters []netlink.CustomConntrackFilter
-	for _, epSvcPair := range endpointsUpdateResult.DeletedUDPEndpoints {
-		if svcInfo, ok := svcPortMap[epSvcPair.ServicePortName]; ok {
-			endpointIP := proxyutil.IPPart(epSvcPair.Endpoint)
-			nodePort := svcInfo.NodePort()
-			if nodePort != 0 {
-				filters = append(filters, filterForPortNAT(endpointIP, nodePort, v1.ProtocolUDP))
+	for _, entry := range entries {
+		// we only deal with UDP protocol entries
+		if entry.Forward.Protocol != unix.IPPROTO_UDP {
+			continue
+		}
 
+		origDst := entry.Forward.DstIP.String()
+		origPortDst := int(entry.Forward.DstPort)
+		replySrc := entry.Reverse.SrcIP.String()
+
+		// if the original destination (--orig-dst) of the entry is service IP (ClusterIP,
+		// LoadBalancerIPs or ExternalIPs) and the reply source (--reply-src) is not IP of
+		// any serving endpoint, we clear the entry.
+		if _, ok := serviceIPEndpointIPs[origDst]; ok {
+			if !serviceIPEndpointIPs[origDst].Has(replySrc) {
+				filters = append(filters, filterForNAT(origDst, replySrc, v1.ProtocolUDP))
 			}
-			filters = append(filters, filterForNAT(svcInfo.ClusterIP().String(), endpointIP, v1.ProtocolUDP))
-			for _, extIP := range svcInfo.ExternalIPs() {
-				filters = append(filters, filterForNAT(extIP.String(), endpointIP, v1.ProtocolUDP))
-			}
-			for _, lbIP := range svcInfo.LoadBalancerVIPs() {
-				filters = append(filters, filterForNAT(lbIP.String(), endpointIP, v1.ProtocolUDP))
+		}
+
+		// if the original port destination (--orig-port-dst) of the flow is service
+		// NodePort and the reply source (--reply-src) is not IP of any serving endpoint,
+		// we clear the entry.
+		if _, ok := serviceNodePortEndpointIPs[origPortDst]; ok {
+			if !serviceNodePortEndpointIPs[origPortDst].Has(replySrc) {
+				filters = append(filters, filterForPortNAT(replySrc, origPortDst, v1.ProtocolUDP))
 			}
 		}
 	}
 
-	if err := ct.ClearEntries(ipFamilyMap[ipFamily], filters...); err != nil {
-		klog.ErrorS(err, "Failed to delete stale endpoint connections")
+	if n, err := ct.ClearEntries(ipFamilyMap[ipFamily], filters...); err != nil {
+		klog.ErrorS(err, "Failed to clear all conntrack entries", "ipFamily", ipFamily, "entriesDeleted", n, "took", time.Since(start))
+	} else {
+		klog.V(4).InfoS("Finished reconciling conntrack entries", "ipFamily", ipFamily, "entriesDeleted", n, "took", time.Since(start))
 	}
 }
 
@@ -122,30 +136,6 @@ var protocolMap = map[v1.Protocol]uint8{
 	v1.ProtocolSCTP: unix.IPPROTO_SCTP,
 }
 
-// filterForIP returns *conntrackFilter to delete the conntrack entries for connections
-// specified by the destination IP (original direction).
-func filterForIP(ip string, protocol v1.Protocol) *conntrackFilter {
-	klog.V(4).InfoS("Adding conntrack filter for cleanup", "org-dst", ip, "protocol", protocol)
-	return &conntrackFilter{
-		protocol: protocolMap[protocol],
-		original: &connectionTuple{
-			dstIP: netutils.ParseIPSloppy(ip),
-		},
-	}
-}
-
-// filterForPort returns *conntrackFilter to delete the conntrack entries for connections
-// specified by the destination Port (original direction).
-func filterForPort(port int, protocol v1.Protocol) *conntrackFilter {
-	klog.V(4).InfoS("Adding conntrack filter for cleanup", "org-port-dst", port, "protocol", protocol)
-	return &conntrackFilter{
-		protocol: protocolMap[protocol],
-		original: &connectionTuple{
-			dstPort: uint16(port),
-		},
-	}
-}
-
 // filterForNAT returns *conntrackFilter to delete the conntrack entries for connections
 // specified by the destination IP (original direction) and source IP (reply direction).
 func filterForNAT(origin, dest string, protocol v1.Protocol) *conntrackFilter {
diff --git a/pkg/proxy/conntrack/conntrack.go b/pkg/proxy/conntrack/conntrack.go
index c5ec762f3cb08..1e01d6546809f 100644
--- a/pkg/proxy/conntrack/conntrack.go
+++ b/pkg/proxy/conntrack/conntrack.go
@@ -29,13 +29,15 @@ import (
 
 // Interface for dealing with conntrack
 type Interface interface {
+	ListEntries(ipFamily uint8) ([]*netlink.ConntrackFlow, error)
 	// ClearEntries deletes conntrack entries for connections of the given IP family,
 	// filtered by the given filters.
-	ClearEntries(ipFamily uint8, filters ...netlink.CustomConntrackFilter) error
+	ClearEntries(ipFamily uint8, filters ...netlink.CustomConntrackFilter) (int, error)
 }
 
 // netlinkHandler allows consuming real and mockable implementation for testing.
 type netlinkHandler interface {
+	ConntrackTableList(netlink.ConntrackTableType, netlink.InetFamily) ([]*netlink.ConntrackFlow, error)
 	ConntrackDeleteFilters(netlink.ConntrackTableType, netlink.InetFamily, ...netlink.CustomConntrackFilter) (uint, error)
 }
 
@@ -54,18 +56,22 @@ func newConntracker(handler netlinkHandler) Interface {
 	return &conntracker{handler: handler}
 }
 
+// ListEntries list all conntrack entries for connections of the given IP family.
+func (ct *conntracker) ListEntries(ipFamily uint8) ([]*netlink.ConntrackFlow, error) {
+	return ct.handler.ConntrackTableList(netlink.ConntrackTable, netlink.InetFamily(ipFamily))
+}
+
 // ClearEntries deletes conntrack entries for connections of the given IP family,
 // filtered by the given filters.
-func (ct *conntracker) ClearEntries(ipFamily uint8, filters ...netlink.CustomConntrackFilter) error {
+func (ct *conntracker) ClearEntries(ipFamily uint8, filters ...netlink.CustomConntrackFilter) (int, error) {
 	if len(filters) == 0 {
 		klog.V(7).InfoS("no conntrack filters provided")
-		return nil
+		return 0, nil
 	}
 
 	n, err := ct.handler.ConntrackDeleteFilters(netlink.ConntrackTable, netlink.InetFamily(ipFamily), filters...)
 	if err != nil {
-		return fmt.Errorf("error deleting conntrack entries, error: %w", err)
+		return int(n), fmt.Errorf("error deleting conntrack entries, error: %w", err)
 	}
-	klog.V(4).InfoS("Cleared conntrack entries", "count", n)
-	return nil
+	return int(n), nil
 }
diff --git a/pkg/proxy/conntrack/fake.go b/pkg/proxy/conntrack/fake.go
index a90aa3f2cfdf1..bedbb99b6e9ac 100644
--- a/pkg/proxy/conntrack/fake.go
+++ b/pkg/proxy/conntrack/fake.go
@@ -20,77 +20,42 @@ limitations under the License.
 package conntrack
 
 import (
-	"fmt"
-
 	"github.com/vishvananda/netlink"
-
-	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 // FakeInterface implements Interface by just recording entries that have been cleared.
 type FakeInterface struct {
-	ClearedIPs      sets.Set[string]
-	ClearedPorts    sets.Set[int]
-	ClearedNATs     map[string]string // origin -> dest
-	ClearedPortNATs map[int]string    // port -> dest
+	entries []*netlink.ConntrackFlow
 }
 
 var _ Interface = &FakeInterface{}
 
 // NewFake creates a new FakeInterface
 func NewFake() *FakeInterface {
-	fake := &FakeInterface{}
-	fake.Reset()
-	return fake
+	return &FakeInterface{entries: make([]*netlink.ConntrackFlow, 0)}
 }
 
-// Reset clears fake's sets/maps
-func (fake *FakeInterface) Reset() {
-	fake.ClearedIPs = sets.New[string]()
-	fake.ClearedPorts = sets.New[int]()
-	fake.ClearedNATs = make(map[string]string)
-	fake.ClearedPortNATs = make(map[int]string)
+// ListEntries is part of Interface
+func (fake *FakeInterface) ListEntries(_ uint8) ([]*netlink.ConntrackFlow, error) {
+	return fake.entries, nil
 }
 
 // ClearEntries is part of Interface
-func (fake *FakeInterface) ClearEntries(_ uint8, filters ...netlink.CustomConntrackFilter) error {
-	for _, anyFilter := range filters {
-		filter := anyFilter.(*conntrackFilter)
-		if filter.protocol != protocolMap[v1.ProtocolUDP] {
-			return fmt.Errorf("FakeInterface currently only supports UDP")
-		}
-
-		// record IP and Port entries
-		if filter.original != nil && filter.reply == nil {
-			if filter.original.dstIP != nil {
-				fake.ClearedIPs.Insert(filter.original.dstIP.String())
-			}
-			if filter.original.dstPort != 0 {
-				fake.ClearedPorts.Insert(int(filter.original.dstPort))
+func (fake *FakeInterface) ClearEntries(_ uint8, filters ...netlink.CustomConntrackFilter) (int, error) {
+	var flows []*netlink.ConntrackFlow
+	before := len(fake.entries)
+	for _, flow := range fake.entries {
+		var matched bool
+		for _, filter := range filters {
+			matched = filter.MatchConntrackFlow(flow)
+			if matched {
+				break
 			}
 		}
-
-		// record NAT and NATPort entries
-		if filter.original != nil && filter.reply != nil {
-			if filter.original.dstIP != nil && filter.reply.srcIP != nil {
-				origin := filter.original.dstIP.String()
-				dest := filter.reply.srcIP.String()
-				if previous, exists := fake.ClearedNATs[origin]; exists && previous != dest {
-					return fmt.Errorf("filter for NAT passed with same origin (%s), different destination (%s / %s)", origin, previous, dest)
-				}
-				fake.ClearedNATs[filter.original.dstIP.String()] = filter.reply.srcIP.String()
-			}
-
-			if filter.original.dstPort != 0 && filter.reply.srcIP != nil {
-				dest := filter.reply.srcIP.String()
-				port := int(filter.original.dstPort)
-				if previous, exists := fake.ClearedPortNATs[port]; exists && previous != dest {
-					return fmt.Errorf("filter for PortNAT passed with same port (%d), different destination (%s / %s)", port, previous, dest)
-				}
-				fake.ClearedPortNATs[port] = dest
-			}
+		if !matched {
+			flows = append(flows, flow)
 		}
 	}
-	return nil
+	fake.entries = flows
+	return before - len(fake.entries), nil
 }
diff --git a/pkg/proxy/endpointschangetracker.go b/pkg/proxy/endpointschangetracker.go
index 1ce3172561a03..3918aae24c8f1 100644
--- a/pkg/proxy/endpointschangetracker.go
+++ b/pkg/proxy/endpointschangetracker.go
@@ -178,18 +178,9 @@ type UpdateEndpointsMapResult struct {
 	// UpdatedServices lists the names of all services with added/updated/deleted
 	// endpoints since the last Update.
 	UpdatedServices sets.Set[types.NamespacedName]
-
-	// DeletedUDPEndpoints identifies UDP endpoints that have just been deleted.
-	// Existing conntrack NAT entries pointing to these endpoints must be deleted to
-	// ensure that no further traffic for the Service gets delivered to them.
-	DeletedUDPEndpoints []ServiceEndpoint
-
-	// NewlyActiveUDPServices identifies UDP Services that have just gone from 0 to
-	// non-0 endpoints. Existing conntrack entries caching the fact that these
-	// services are black holes must be deleted to ensure that traffic can immediately
-	// begin flowing to the new endpoints.
-	NewlyActiveUDPServices []ServicePortName
-
+	// ConntrackCleanupRequired will be true if any UDP ServicePort changed endpoints, false otherwise.
+	// It's used to minimise conntrack cleanup calls.
+	ConntrackCleanupRequired bool
 	// List of the trigger times for all endpoints objects that changed. It's used to export the
 	// network programming latency.
 	// NOTE(oxddr): this can be simplified to []time.Time if memory consumption becomes an issue.
@@ -205,8 +196,6 @@ type EndpointsMap map[ServicePortName][]Endpoint
 func (em EndpointsMap) Update(ect *EndpointsChangeTracker) UpdateEndpointsMapResult {
 	result := UpdateEndpointsMapResult{
 		UpdatedServices:        sets.New[types.NamespacedName](),
-		DeletedUDPEndpoints:    make([]ServiceEndpoint, 0),
-		NewlyActiveUDPServices: make([]ServicePortName, 0),
 		LastChangeTriggerTimes: make(map[types.NamespacedName][]time.Time),
 	}
 	if ect == nil {
@@ -222,7 +211,26 @@ func (em EndpointsMap) Update(ect *EndpointsChangeTracker) UpdateEndpointsMapRes
 
 		em.unmerge(change.previous)
 		em.merge(change.current)
-		detectStaleConntrackEntries(change.previous, change.current, &result.DeletedUDPEndpoints, &result.NewlyActiveUDPServices)
+
+		// result.ConntrackCleanupRequired should be true if any one of the UDP
+		// ServicePort changed endpoint. Once true, we don't update the value.
+		if result.ConntrackCleanupRequired {
+			continue
+		}
+		// Check if the changed service had any UDP ServicePort
+		for svcPort := range change.previous {
+			if svcPort.NamespacedName == nn && svcPort.Protocol == v1.ProtocolUDP {
+				result.ConntrackCleanupRequired = true
+				break
+			}
+		}
+		// Check if the changed service has any UDP ServicePort
+		for svcPort := range change.current {
+			if svcPort.NamespacedName == nn && svcPort.Protocol == v1.ProtocolUDP {
+				result.ConntrackCleanupRequired = true
+				break
+			}
+		}
 	}
 	ect.checkoutTriggerTimes(&result.LastChangeTriggerTimes)
 
@@ -284,68 +292,3 @@ func (em EndpointsMap) LocalReadyEndpoints() map[types.NamespacedName]int {
 	}
 	return eps
 }
-
-// detectStaleConntrackEntries detects services that may be associated with stale conntrack entries.
-// (See UpdateEndpointsMapResult.DeletedUDPEndpoints and .NewlyActiveUDPServices.)
-func detectStaleConntrackEntries(oldEndpointsMap, newEndpointsMap EndpointsMap, deletedUDPEndpoints *[]ServiceEndpoint, newlyActiveUDPServices *[]ServicePortName) {
-	// Find the UDP endpoints that we were sending traffic to in oldEndpointsMap, but
-	// are no longer sending to newEndpointsMap. The proxier should make sure that
-	// conntrack does not accidentally route any new connections to them.
-	for svcPortName, epList := range oldEndpointsMap {
-		if svcPortName.Protocol != v1.ProtocolUDP {
-			continue
-		}
-
-		for _, ep := range epList {
-			// If the old endpoint wasn't Serving then there can't be stale
-			// conntrack entries since there was no traffic sent to it.
-			if !ep.IsServing() {
-				continue
-			}
-
-			deleted := true
-			// Check if the endpoint has changed, including if it went from
-			// serving to not serving. If it did change stale entries for the old
-			// endpoint have to be cleared.
-			for i := range newEndpointsMap[svcPortName] {
-				if newEndpointsMap[svcPortName][i].String() == ep.String() &&
-					newEndpointsMap[svcPortName][i].IsServing() == ep.IsServing() {
-					deleted = false
-					break
-				}
-			}
-			if deleted {
-				klog.V(4).InfoS("Deleted endpoint may have stale conntrack entries", "portName", svcPortName, "endpoint", ep)
-				*deletedUDPEndpoints = append(*deletedUDPEndpoints, ServiceEndpoint{Endpoint: ep.String(), ServicePortName: svcPortName})
-			}
-		}
-	}
-
-	// Detect services that have gone from 0 to non-0 ready endpoints. If there were
-	// previously 0 endpoints, but someone tried to connect to it, then a conntrack
-	// entry may have been created blackholing traffic to that IP, which should be
-	// deleted now.
-	for svcPortName, epList := range newEndpointsMap {
-		if svcPortName.Protocol != v1.ProtocolUDP {
-			continue
-		}
-
-		epServing := 0
-		for _, ep := range epList {
-			if ep.IsServing() {
-				epServing++
-			}
-		}
-
-		oldEpServing := 0
-		for _, ep := range oldEndpointsMap[svcPortName] {
-			if ep.IsServing() {
-				oldEpServing++
-			}
-		}
-
-		if epServing > 0 && oldEpServing == 0 {
-			*newlyActiveUDPServices = append(*newlyActiveUDPServices, svcPortName)
-		}
-	}
-}
diff --git a/pkg/proxy/iptables/proxier.go b/pkg/proxy/iptables/proxier.go
index 9ff20521f5a0b..086c4f30bb409 100644
--- a/pkg/proxy/iptables/proxier.go
+++ b/pkg/proxy/iptables/proxier.go
@@ -1594,8 +1594,10 @@ func (proxier *Proxier) syncProxyRules() {
 		proxier.logger.Error(err, "Error syncing healthcheck endpoints")
 	}
 
-	// Finish housekeeping, clear stale conntrack entries for UDP Services
-	conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, serviceUpdateResult, endpointUpdateResult)
+	if endpointUpdateResult.ConntrackCleanupRequired {
+		// Finish housekeeping, clear stale conntrack entries for UDP Services
+		conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, proxier.endpointsMap)
+	}
 }
 
 func (proxier *Proxier) writeServiceToEndpointRules(natRules proxyutil.LineBuffer, svcPortNameString string, svcInfo proxy.ServicePort, svcChain utiliptables.Chain, endpoints []proxy.Endpoint, args []string) {
diff --git a/pkg/proxy/ipvs/proxier.go b/pkg/proxy/ipvs/proxier.go
index 2378c075ee07d..f7c220e2defbd 100644
--- a/pkg/proxy/ipvs/proxier.go
+++ b/pkg/proxy/ipvs/proxier.go
@@ -932,7 +932,7 @@ func (proxier *Proxier) syncProxyRules() {
 	// We assume that if this was called, we really want to sync them,
 	// even if nothing changed in the meantime. In other words, callers are
 	// responsible for detecting no-op changes and not calling this function.
-	serviceUpdateResult := proxier.svcPortMap.Update(proxier.serviceChanges)
+	_ = proxier.svcPortMap.Update(proxier.serviceChanges)
 	endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)
 
 	proxier.logger.V(3).Info("Syncing ipvs proxier rules")
@@ -1497,8 +1497,10 @@ func (proxier *Proxier) syncProxyRules() {
 	metrics.SyncProxyRulesNoLocalEndpointsTotal.WithLabelValues("internal").Set(float64(proxier.serviceNoLocalEndpointsInternal.Len()))
 	metrics.SyncProxyRulesNoLocalEndpointsTotal.WithLabelValues("external").Set(float64(proxier.serviceNoLocalEndpointsExternal.Len()))
 
-	// Finish housekeeping, clear stale conntrack entries for UDP Services
-	conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, serviceUpdateResult, endpointUpdateResult)
+	if endpointUpdateResult.ConntrackCleanupRequired {
+		// Finish housekeeping, clear stale conntrack entries for UDP Services
+		conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, proxier.endpointsMap)
+	}
 }
 
 // writeIptablesRules write all iptables rules to proxier.natRules or proxier.FilterRules that ipvs proxier needed
diff --git a/pkg/proxy/nftables/proxier.go b/pkg/proxy/nftables/proxier.go
index 7c419b9c54e96..538869cb061d9 100644
--- a/pkg/proxy/nftables/proxier.go
+++ b/pkg/proxy/nftables/proxier.go
@@ -1838,8 +1838,10 @@ func (proxier *Proxier) syncProxyRules() {
 		proxier.logger.Error(err, "Error syncing healthcheck endpoints")
 	}
 
-	// Finish housekeeping, clear stale conntrack entries for UDP Services
-	conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, serviceUpdateResult, endpointUpdateResult)
+	if endpointUpdateResult.ConntrackCleanupRequired {
+		// Finish housekeeping, clear stale conntrack entries for UDP Services
+		conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, proxier.endpointsMap)
+	}
 }
 
 func (proxier *Proxier) writeServiceToEndpointRules(tx *knftables.Transaction, svcInfo *servicePortInfo, svcChain string, endpoints []proxy.Endpoint) {
diff --git a/pkg/proxy/servicechangetracker.go b/pkg/proxy/servicechangetracker.go
index fb5fd4de406ee..8c42917e44ab9 100644
--- a/pkg/proxy/servicechangetracker.go
+++ b/pkg/proxy/servicechangetracker.go
@@ -117,11 +117,6 @@ type UpdateServiceMapResult struct {
 	// UpdatedServices lists the names of all services added/updated/deleted since the
 	// last Update.
 	UpdatedServices sets.Set[types.NamespacedName]
-
-	// DeletedUDPClusterIPs holds stale (no longer assigned to a Service) Service IPs
-	// that had UDP ports. Callers can use this to abort timeout-waits or clear
-	// connection-tracking information.
-	DeletedUDPClusterIPs sets.Set[string]
 }
 
 // HealthCheckNodePorts returns a map of Service names to HealthCheckNodePort values
@@ -178,8 +173,7 @@ func (sm ServicePortMap) Update(sct *ServiceChangeTracker) UpdateServiceMapResul
 	defer sct.lock.Unlock()
 
 	result := UpdateServiceMapResult{
-		UpdatedServices:      sets.New[types.NamespacedName](),
-		DeletedUDPClusterIPs: sets.New[string](),
+		UpdatedServices: sets.New[types.NamespacedName](),
 	}
 
 	for nn, change := range sct.items {
@@ -192,7 +186,7 @@ func (sm ServicePortMap) Update(sct *ServiceChangeTracker) UpdateServiceMapResul
 		// filter out the Update event of current changes from previous changes
 		// before calling unmerge() so that can skip deleting the Update events.
 		change.previous.filter(change.current)
-		sm.unmerge(change.previous, result.DeletedUDPClusterIPs)
+		sm.unmerge(change.previous)
 	}
 	// clear changes after applying them to ServicePortMap.
 	sct.items = make(map[types.NamespacedName]*serviceChange)
@@ -226,16 +220,12 @@ func (sm *ServicePortMap) filter(other ServicePortMap) {
 	}
 }
 
-// unmerge deletes all other ServicePortMap's elements from current ServicePortMap and
-// updates deletedUDPClusterIPs with all of the newly-deleted UDP cluster IPs.
-func (sm *ServicePortMap) unmerge(other ServicePortMap, deletedUDPClusterIPs sets.Set[string]) {
+// unmerge deletes all other ServicePortMap's elements from current ServicePortMap.
+func (sm *ServicePortMap) unmerge(other ServicePortMap) {
 	for svcPortName := range other {
-		info, exists := (*sm)[svcPortName]
+		_, exists := (*sm)[svcPortName]
 		if exists {
 			klog.V(4).InfoS("Removing service port", "portName", svcPortName)
-			if info.Protocol() == v1.ProtocolUDP {
-				deletedUDPClusterIPs.Insert(info.ClusterIP().String())
-			}
 			delete(*sm, svcPortName)
 		} else {
 			klog.ErrorS(nil, "Service port does not exists", "portName", svcPortName)
diff --git a/pkg/proxy/winkernel/proxier.go b/pkg/proxy/winkernel/proxier.go
index aa456ab00cdec..92a8969c98fdf 100644
--- a/pkg/proxy/winkernel/proxier.go
+++ b/pkg/proxy/winkernel/proxier.go
@@ -1166,17 +1166,9 @@ func (proxier *Proxier) syncProxyRules() {
 	// We assume that if this was called, we really want to sync them,
 	// even if nothing changed in the meantime. In other words, callers are
 	// responsible for detecting no-op changes and not calling this function.
-	serviceUpdateResult := proxier.svcPortMap.Update(proxier.serviceChanges)
-	endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)
-
-	deletedUDPClusterIPs := serviceUpdateResult.DeletedUDPClusterIPs
-	// merge stale services gathered from EndpointsMap.Update
-	for _, svcPortName := range endpointUpdateResult.NewlyActiveUDPServices {
-		if svcInfo, ok := proxier.svcPortMap[svcPortName]; ok && svcInfo != nil && svcInfo.Protocol() == v1.ProtocolUDP {
-			klog.V(2).InfoS("Newly-active UDP service may have stale conntrack entries", "servicePortName", svcPortName)
-			deletedUDPClusterIPs.Insert(svcInfo.ClusterIP().String())
-		}
-	}
+	_ = proxier.svcPortMap.Update(proxier.serviceChanges)
+	_ = proxier.endpointsMap.Update(proxier.endpointsChanges)
+
 	// Query HNS for endpoints and load balancers
 	queriedEndpoints, err := hns.getAllEndpointsByNetwork(hnsNetworkName)
 	if err != nil {
@@ -1715,13 +1707,6 @@ func (proxier *Proxier) syncProxyRules() {
 		klog.ErrorS(err, "Error syncing healthcheck endpoints")
 	}
 
-	// Finish housekeeping.
-	// TODO: these could be made more consistent.
-	for _, svcIP := range deletedUDPClusterIPs.UnsortedList() {
-		// TODO : Check if this is required to cleanup stale services here
-		klog.V(5).InfoS("Pending delete stale service IP connections", "IP", svcIP)
-	}
-
 	// remove stale endpoint refcount entries
 	for epIP := range proxier.terminatedEndpoints {
 		if epToDelete := queriedEndpoints[epIP]; epToDelete != nil && epToDelete.hnsID != "" {


Test patch:
diff --git a/pkg/proxy/conntrack/cleanup_test.go b/pkg/proxy/conntrack/cleanup_test.go
index 134f4d724d116..d6232a02b3a7a 100644
--- a/pkg/proxy/conntrack/cleanup_test.go
+++ b/pkg/proxy/conntrack/cleanup_test.go
@@ -20,62 +20,75 @@ limitations under the License.
 package conntrack
 
 import (
-	"net"
-	"reflect"
+	"fmt"
+	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 	"github.com/vishvananda/netlink"
+	"golang.org/x/sys/unix"
 
 	v1 "k8s.io/api/core/v1"
+	discovery "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/proxy"
 	netutils "k8s.io/utils/net"
+	"k8s.io/utils/ptr"
 )
 
 const (
+	testServiceName      = "cleanup-test"
+	testServiceNamespace = "test"
+
 	testIPFamily       = v1.IPv4Protocol
 	testClusterIP      = "172.30.1.1"
 	testExternalIP     = "192.168.99.100"
 	testLoadBalancerIP = "1.2.3.4"
 
-	testEndpointIP = "10.240.0.4"
+	testServingEndpointIP    = "10.240.0.4"
+	testNonServingEndpointIP = "10.240.1.5"
+	testDeletedEndpointIP    = "10.240.2.6"
 
-	testPort         = 53
-	testNodePort     = 5353
-	testEndpointPort = "5300"
+	testPort     = 8000
+	testNodePort = 32000
 )
 
 func TestCleanStaleEntries(t *testing.T) {
-	// We need to construct a proxy.ServicePortMap to pass to CleanStaleEntries.
-	// ServicePortMap is just map[string]proxy.ServicePort, but there are no public
-	// constructors for any implementation of proxy.ServicePort, so we have to either
-	// provide our own implementation of that interface, or else use a
-	// proxy.ServiceChangeTracker to construct them and fill in the map for us.
+	// We need to construct proxy.ServicePortMap and proxy.EndpointsMap to pass to
+	// CleanStaleEntries. ServicePortMap and EndpointsMap are just maps, but there are
+	// no public constructors for any implementation of proxy.ServicePort and
+	// proxy.EndpointsMap, so we have to either provide our own implementation of that
+	// interface, or else use a proxy.ServiceChangeTracker and proxy.NewEndpointsChangeTracker
+	// to construct them and fill in the maps for us.
 
 	sct := proxy.NewServiceChangeTracker(nil, v1.IPv4Protocol, nil, nil)
 	svc := &v1.Service{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      "cleanup-test",
-			Namespace: "test",
+			Name:      testServiceName,
+			Namespace: testServiceNamespace,
 		},
 		Spec: v1.ServiceSpec{
 			ClusterIP:   testClusterIP,
 			ExternalIPs: []string{testExternalIP},
 			Ports: []v1.ServicePort{
 				{
-					Name:     "dns-tcp",
+					Name:     "test-tcp",
 					Port:     testPort,
 					Protocol: v1.ProtocolTCP,
 				},
 				{
-					Name:     "dns-udp",
+					Name:     "test-udp",
 					Port:     testPort,
 					NodePort: testNodePort,
 					Protocol: v1.ProtocolUDP,
 				},
+				{
+					Name:     "test-sctp",
+					Port:     testPort,
+					NodePort: testNodePort,
+					Protocol: v1.ProtocolSCTP,
+				},
 			},
 		},
 		Status: v1.ServiceStatus{
@@ -86,14 +99,52 @@ func TestCleanStaleEntries(t *testing.T) {
 			},
 		},
 	}
-	sct.Update(nil, svc)
 
+	sct.Update(nil, svc)
 	svcPortMap := make(proxy.ServicePortMap)
 	_ = svcPortMap.Update(sct)
 
-	// (At this point we are done with sct, and in particular, we don't use sct to
-	// construct UpdateServiceMapResults, because pkg/proxy already has its own tests
-	// for that. Also, svcPortMap is read-only from this point on.)
+	ect := proxy.NewEndpointsChangeTracker("test-worker", nil, v1.IPv4Protocol, nil, nil)
+	eps := &discovery.EndpointSlice{
+		TypeMeta:    metav1.TypeMeta{},
+		AddressType: discovery.AddressTypeIPv4,
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-0", testServiceName),
+			Namespace: testServiceNamespace,
+			Labels:    map[string]string{discovery.LabelServiceName: testServiceName},
+		},
+		Endpoints: []discovery.Endpoint{
+			{
+				Addresses:  []string{testServingEndpointIP},
+				Conditions: discovery.EndpointConditions{Serving: ptr.To(true)},
+			},
+			{
+				Addresses:  []string{testNonServingEndpointIP},
+				Conditions: discovery.EndpointConditions{Serving: ptr.To(false)},
+			},
+		},
+		Ports: []discovery.EndpointPort{
+			{
+				Name:     ptr.To("test-tcp"),
+				Port:     ptr.To(int32(testPort)),
+				Protocol: ptr.To(v1.ProtocolTCP),
+			},
+			{
+				Name:     ptr.To("test-udp"),
+				Port:     ptr.To(int32(testPort)),
+				Protocol: ptr.To(v1.ProtocolUDP),
+			},
+			{
+				Name:     ptr.To("test-sctp"),
+				Port:     ptr.To(int32(testPort)),
+				Protocol: ptr.To(v1.ProtocolSCTP),
+			},
+		},
+	}
+
+	ect.EndpointSliceUpdate(eps, false)
+	endpointsMap := make(proxy.EndpointsMap)
+	_ = endpointsMap.Update(ect)
 
 	tcpPortName := proxy.ServicePortName{
 		NamespacedName: types.NamespacedName{
@@ -113,227 +164,134 @@ func TestCleanStaleEntries(t *testing.T) {
 		Protocol: svc.Spec.Ports[1].Protocol,
 	}
 
-	unknownPortName := udpPortName
-	unknownPortName.Namespace = "unknown"
+	sctpPortName := proxy.ServicePortName{
+		NamespacedName: types.NamespacedName{
+			Namespace: svc.Namespace,
+			Name:      svc.Name,
+		},
+		Port:     svc.Spec.Ports[2].Name,
+		Protocol: svc.Spec.Ports[2].Protocol,
+	}
 
-	// Sanity-check to make sure we constructed the map correctly
-	if len(svcPortMap) != 2 {
+	// Sanity-check to make sure we constructed the ServicePortMap correctly
+	if len(svcPortMap) != 3 {
 		t.Fatalf("expected svcPortMap to have 2 entries, got %+v", svcPortMap)
 	}
 	servicePort := svcPortMap[tcpPortName]
-	if servicePort == nil || servicePort.String() != "172.30.1.1:53/TCP" {
-		t.Fatalf("expected svcPortMap[%q] to be \"172.30.1.1:53/TCP\", got %q", tcpPortName.String(), servicePort.String())
+	if servicePort == nil || servicePort.String() != "172.30.1.1:8000/TCP" {
+		t.Fatalf("expected svcPortMap[%q] to be \"172.30.1.1:8000/TCP\", got %q", tcpPortName.String(), servicePort.String())
 	}
 	servicePort = svcPortMap[udpPortName]
-	if servicePort == nil || servicePort.String() != "172.30.1.1:53/UDP" {
-		t.Fatalf("expected svcPortMap[%q] to be \"172.30.1.1:53/UDP\", got %q", udpPortName.String(), servicePort.String())
+	if servicePort == nil || servicePort.String() != "172.30.1.1:8000/UDP" {
+		t.Fatalf("expected svcPortMap[%q] to be \"172.30.1.1:8000/UDP\", got %q", udpPortName.String(), servicePort.String())
+	}
+	servicePort = svcPortMap[sctpPortName]
+	if servicePort == nil || servicePort.String() != "172.30.1.1:8000/SCTP" {
+		t.Fatalf("expected svcPortMap[%q] to be \"172.30.1.1:8000/SCTP\", got %q", sctpPortName.String(), servicePort.String())
 	}
 
-	testCases := []struct {
-		description string
-
-		serviceUpdates   proxy.UpdateServiceMapResult
-		endpointsUpdates proxy.UpdateEndpointsMapResult
-
-		result FakeInterface
-	}{
-		{
-			description: "DeletedUDPClusterIPs clears entries for given clusterIPs (only)",
-
-			serviceUpdates: proxy.UpdateServiceMapResult{
-				// Note: this isn't testClusterIP; it's the IP of some
-				// unknown (because deleted) service.
-				DeletedUDPClusterIPs: sets.New("172.30.99.99"),
-			},
-			endpointsUpdates: proxy.UpdateEndpointsMapResult{},
-
-			result: FakeInterface{
-				ClearedIPs: sets.New("172.30.99.99"),
-
-				ClearedPorts:    sets.New[int](),
-				ClearedNATs:     map[string]string{},
-				ClearedPortNATs: map[int]string{},
-			},
-		},
-		{
-			description: "DeletedUDPEndpoints clears NAT entries for all IPs and NodePorts",
-
-			serviceUpdates: proxy.UpdateServiceMapResult{
-				DeletedUDPClusterIPs: sets.New[string](),
-			},
-			endpointsUpdates: proxy.UpdateEndpointsMapResult{
-				DeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-					Endpoint:        net.JoinHostPort(testEndpointIP, testEndpointPort),
-					ServicePortName: udpPortName,
-				}},
-			},
-
-			result: FakeInterface{
-				ClearedIPs:   sets.New[string](),
-				ClearedPorts: sets.New[int](),
-
-				ClearedNATs: map[string]string{
-					testClusterIP:      testEndpointIP,
-					testExternalIP:     testEndpointIP,
-					testLoadBalancerIP: testEndpointIP,
-				},
-				ClearedPortNATs: map[int]string{
-					testNodePort: testEndpointIP,
-				},
-			},
-		},
-		{
-			description: "NewlyActiveUDPServices clears entries for all IPs and NodePorts",
-
-			serviceUpdates: proxy.UpdateServiceMapResult{
-				DeletedUDPClusterIPs: sets.New[string](),
-			},
-			endpointsUpdates: proxy.UpdateEndpointsMapResult{
-				DeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-				NewlyActiveUDPServices: []proxy.ServicePortName{
-					udpPortName,
-				},
-			},
-
-			result: FakeInterface{
-				ClearedIPs:   sets.New(testClusterIP, testExternalIP, testLoadBalancerIP),
-				ClearedPorts: sets.New(testNodePort),
-
-				ClearedNATs:     map[string]string{},
-				ClearedPortNATs: map[int]string{},
-			},
-		},
-
-		{
-			description: "DeletedUDPEndpoints for unknown Service has no effect",
-
-			serviceUpdates: proxy.UpdateServiceMapResult{
-				DeletedUDPClusterIPs: sets.New[string](),
-			},
-			endpointsUpdates: proxy.UpdateEndpointsMapResult{
-				DeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-					Endpoint:        "10.240.0.4:80",
-					ServicePortName: unknownPortName,
-				}},
-				NewlyActiveUDPServices: []proxy.ServicePortName{},
-			},
-
-			result: FakeInterface{
-				ClearedIPs:      sets.New[string](),
-				ClearedPorts:    sets.New[int](),
-				ClearedNATs:     map[string]string{},
-				ClearedPortNATs: map[int]string{},
-			},
-		},
-		{
-			description: "NewlyActiveUDPServices for unknown Service has no effect",
-
-			serviceUpdates: proxy.UpdateServiceMapResult{
-				DeletedUDPClusterIPs: sets.New[string](),
-			},
-			endpointsUpdates: proxy.UpdateEndpointsMapResult{
-				DeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-				NewlyActiveUDPServices: []proxy.ServicePortName{
-					unknownPortName,
-				},
-			},
-
-			result: FakeInterface{
-				ClearedIPs:      sets.New[string](),
-				ClearedPorts:    sets.New[int](),
-				ClearedNATs:     map[string]string{},
-				ClearedPortNATs: map[int]string{},
-			},
-		},
+	// Sanity-check to make sure we constructed the EndpointsMap map correctly
+	if len(endpointsMap) != 3 {
+		t.Fatalf("expected endpointsMap to have 3 entries, got %+v", endpointsMap)
+	}
+	for _, svcPortName := range []proxy.ServicePortName{tcpPortName, udpPortName, sctpPortName} {
+		if len(endpointsMap[svcPortName]) != 2 {
+			t.Fatalf("expected endpointsMap[%q] to have 2 entries, got %+v", svcPortName.String(), endpointsMap[svcPortName])
+		}
+		if endpointsMap[svcPortName][0].IP() != "10.240.0.4" {
+			t.Fatalf("expected endpointsMap[%q][0] IP to be \"10.240.0.4\", got \"%s\"", svcPortName.String(), endpointsMap[svcPortName][0].IP())
+		}
+		if endpointsMap[svcPortName][1].IP() != "10.240.1.5" {
+			t.Fatalf("expected endpointsMap[%q][1] IP to be \"10.240.1.5\", got \"%s\"", svcPortName.String(), endpointsMap[svcPortName][1].IP())
+		}
+		if !endpointsMap[svcPortName][0].IsServing() {
+			t.Fatalf("expected endpointsMap[%q][0] to be serving", svcPortName.String())
+		}
+		if endpointsMap[svcPortName][1].IsServing() {
+			t.Fatalf("expected endpointsMap[%q][1] to be not serving", svcPortName.String())
+		}
 	}
 
-	for _, tc := range testCases {
-		t.Run(tc.description, func(t *testing.T) {
-			fake := NewFake()
-			CleanStaleEntries(fake, testIPFamily, svcPortMap, tc.serviceUpdates, tc.endpointsUpdates)
-			if !fake.ClearedIPs.Equal(tc.result.ClearedIPs) {
-				t.Errorf("Expected ClearedIPs=%v, got %v", tc.result.ClearedIPs, fake.ClearedIPs)
+	// mock existing entries before cleanup
+	// we create 36 fake flow entries ( 3 Endpoints * 3 Protocols * ( 3 (ServiceIPs) + 1 (NodePort))
+	var mockEntries []*netlink.ConntrackFlow
+	// expectedEntries are the entries on which we will assert the cleanup logic
+	var expectedEntries []*netlink.ConntrackFlow
+	for _, dnatDest := range []string{testServingEndpointIP, testNonServingEndpointIP, testDeletedEndpointIP} {
+		for _, proto := range []uint8{unix.IPPROTO_TCP, unix.IPPROTO_UDP, unix.IPPROTO_SCTP} {
+			for _, origDest := range []string{testClusterIP, testLoadBalancerIP, testExternalIP} {
+				entry := &netlink.ConntrackFlow{
+					FamilyType: unix.AF_INET,
+					Forward: netlink.IPTuple{
+						DstIP:    netutils.ParseIPSloppy(origDest),
+						Protocol: proto,
+					},
+					Reverse: netlink.IPTuple{
+						Protocol: proto,
+						SrcIP:    netutils.ParseIPSloppy(dnatDest),
+					},
+				}
+				mockEntries = append(mockEntries, entry)
+				// we do not expect deleted or non-serving UDP endpoints flows to be present after cleanup
+				if !(proto == unix.IPPROTO_UDP && (dnatDest == testNonServingEndpointIP || dnatDest == testDeletedEndpointIP)) {
+					expectedEntries = append(expectedEntries, entry)
+				}
 			}
-			if !fake.ClearedPorts.Equal(tc.result.ClearedPorts) {
-				t.Errorf("Expected ClearedPorts=%v, got %v", tc.result.ClearedPorts, fake.ClearedPorts)
-			}
-			if !reflect.DeepEqual(fake.ClearedNATs, tc.result.ClearedNATs) {
-				t.Errorf("Expected ClearedNATs=%v, got %v", tc.result.ClearedNATs, fake.ClearedNATs)
+			entry := &netlink.ConntrackFlow{
+				FamilyType: unix.AF_INET,
+				Forward: netlink.IPTuple{
+					DstPort:  testNodePort,
+					Protocol: proto,
+				},
+				Reverse: netlink.IPTuple{
+					Protocol: proto,
+					SrcIP:    netutils.ParseIPSloppy(dnatDest),
+				},
 			}
-			if !reflect.DeepEqual(fake.ClearedPortNATs, tc.result.ClearedPortNATs) {
-				t.Errorf("Expected ClearedPortNATs=%v, got %v", tc.result.ClearedPortNATs, fake.ClearedPortNATs)
+			mockEntries = append(mockEntries, entry)
+			// we do not expect deleted or non-serving UDP endpoints entries to be present after cleanup
+			if !(proto == unix.IPPROTO_UDP && (dnatDest == testNonServingEndpointIP || dnatDest == testDeletedEndpointIP)) {
+				expectedEntries = append(expectedEntries, entry)
 			}
-		})
+		}
 	}
-}
 
-func TestFilterForIP(t *testing.T) {
-	testCases := []struct {
-		name           string
-		ip             string
-		protocol       v1.Protocol
-		expectedFamily netlink.InetFamily
-		expectedFilter *conntrackFilter
-	}{
-		{
-			name:     "ipv4 + UDP",
-			ip:       "10.96.0.10",
-			protocol: v1.ProtocolUDP,
-			expectedFilter: &conntrackFilter{
-				protocol: 17,
-				original: &connectionTuple{dstIP: netutils.ParseIPSloppy("10.96.0.10")},
+	// add some non-DNATed mock entries which should be cleared up by reconciler
+	// These will exist if the proxy don't have DROP/REJECT rule for service with
+	// no endpoints, --orig-dst and --reply-src will be same for these entries.
+	for _, ip := range []string{testClusterIP, testLoadBalancerIP, testExternalIP} {
+		entry := &netlink.ConntrackFlow{
+			FamilyType: unix.AF_INET,
+			Forward: netlink.IPTuple{
+				DstIP:    netutils.ParseIPSloppy(ip),
+				Protocol: unix.IPPROTO_UDP,
 			},
-		},
-		{
-			name:     "ipv6 + TCP",
-			ip:       "2001:db8:1::2",
-			protocol: v1.ProtocolTCP,
-			expectedFilter: &conntrackFilter{
-				protocol: 6,
-				original: &connectionTuple{dstIP: netutils.ParseIPSloppy("2001:db8:1::2")},
+			Reverse: netlink.IPTuple{
+				Protocol: unix.IPPROTO_UDP,
+				SrcIP:    netutils.ParseIPSloppy(ip),
 			},
-		},
+		}
+		mockEntries = append(mockEntries, entry)
 	}
 
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			require.Equal(t, tc.expectedFilter, filterForIP(tc.ip, tc.protocol))
-		})
-	}
-}
+	fake := NewFake()
+	fake.entries = mockEntries
+	CleanStaleEntries(fake, testIPFamily, svcPortMap, endpointsMap)
 
-func TestFilterForPort(t *testing.T) {
-	testCases := []struct {
-		name           string
-		port           int
-		protocol       v1.Protocol
-		expectedFilter *conntrackFilter
-	}{
-		{
-			name:     "UDP",
-			port:     5000,
-			protocol: v1.ProtocolUDP,
+	actualEntries, _ := fake.ListEntries(ipFamilyMap[testIPFamily])
+	require.Equal(t, len(expectedEntries), len(actualEntries))
 
-			expectedFilter: &conntrackFilter{
-				protocol: 17,
-				original: &connectionTuple{dstPort: 5000},
-			},
-		},
-		{
-			name:     "SCTP",
-			port:     3000,
-			protocol: v1.ProtocolSCTP,
-			expectedFilter: &conntrackFilter{
-				protocol: 132,
-				original: &connectionTuple{dstPort: 3000},
-			},
-		},
-	}
+	// sort the actual flows before comparison
+	sort.Slice(actualEntries, func(i, j int) bool {
+		return actualEntries[i].String() < actualEntries[j].String()
+	})
+	// sort the expected flows before comparison
+	sort.Slice(expectedEntries, func(i, j int) bool {
+		return expectedEntries[i].String() < expectedEntries[j].String()
+	})
 
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			require.Equal(t, tc.expectedFilter, filterForPort(tc.port, tc.protocol))
-		})
+	for i := 0; i < len(expectedEntries); i++ {
+		require.Equal(t, expectedEntries[i], actualEntries[i])
 	}
 }
 
diff --git a/pkg/proxy/conntrack/conntrack_test.go b/pkg/proxy/conntrack/conntrack_test.go
index 729ac39b443eb..98f005a186cc1 100644
--- a/pkg/proxy/conntrack/conntrack_test.go
+++ b/pkg/proxy/conntrack/conntrack_test.go
@@ -35,6 +35,10 @@ type fakeHandler struct {
 	filters   []*conntrackFilter
 }
 
+func (f *fakeHandler) ConntrackTableList(_ netlink.ConntrackTableType, _ netlink.InetFamily) ([]*netlink.ConntrackFlow, error) {
+	return nil, nil
+}
+
 func (f *fakeHandler) ConntrackDeleteFilters(tableType netlink.ConntrackTableType, family netlink.InetFamily, netlinkFilters ...netlink.CustomConntrackFilter) (uint, error) {
 	f.tableType = tableType
 	f.ipFamily = family
@@ -48,7 +52,6 @@ func (f *fakeHandler) ConntrackDeleteFilters(tableType netlink.ConntrackTableTyp
 var _ netlinkHandler = (*fakeHandler)(nil)
 
 func TestConntracker_ClearEntries(t *testing.T) {
-
 	testCases := []struct {
 		name     string
 		ipFamily uint8
@@ -91,7 +94,8 @@ func TestConntracker_ClearEntries(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			handler := &fakeHandler{}
 			ct := newConntracker(handler)
-			require.NoError(t, ct.ClearEntries(tc.ipFamily, tc.filters...))
+			_, err := ct.ClearEntries(tc.ipFamily, tc.filters...)
+			require.NoError(t, err)
 			require.Equal(t, netlink.ConntrackTableType(netlink.ConntrackTable), handler.tableType)
 			require.Equal(t, netlink.InetFamily(tc.ipFamily), handler.ipFamily)
 			require.Equal(t, len(tc.filters), len(handler.filters))
diff --git a/pkg/proxy/endpointschangetracker_test.go b/pkg/proxy/endpointschangetracker_test.go
index dd3ddb0fd6837..89940b2912d89 100644
--- a/pkg/proxy/endpointschangetracker_test.go
+++ b/pkg/proxy/endpointschangetracker_test.go
@@ -525,23 +525,21 @@ func TestUpdateEndpointsMap(t *testing.T) {
 		// previousEndpointSlices and currentEndpointSlices are used to call appropriate
 		// handlers OnEndpointSlice* (based on whether corresponding values are nil
 		// or non-nil) and must be of equal length.
-		name                           string
-		previousEndpointSlices         []*discovery.EndpointSlice
-		currentEndpointSlices          []*discovery.EndpointSlice
-		previousEndpointsMap           map[ServicePortName][]*BaseEndpointInfo
-		expectedResult                 map[ServicePortName][]*BaseEndpointInfo
-		expectedDeletedUDPEndpoints    []ServiceEndpoint
-		expectedNewlyActiveUDPServices map[ServicePortName]bool
-		expectedLocalEndpoints         map[types.NamespacedName]int
-		expectedChangedEndpoints       sets.Set[types.NamespacedName]
+		name                             string
+		previousEndpointSlices           []*discovery.EndpointSlice
+		currentEndpointSlices            []*discovery.EndpointSlice
+		previousEndpointsMap             map[ServicePortName][]*BaseEndpointInfo
+		expectedResult                   map[ServicePortName][]*BaseEndpointInfo
+		expectedConntrackCleanupRequired bool
+		expectedLocalEndpoints           map[types.NamespacedName]int
+		expectedChangedEndpoints         sets.Set[types.NamespacedName]
 	}{{
-		name:                           "empty",
-		previousEndpointsMap:           map[ServicePortName][]*BaseEndpointInfo{},
-		expectedResult:                 map[ServicePortName][]*BaseEndpointInfo{},
-		expectedDeletedUDPEndpoints:    []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New[types.NamespacedName](),
+		name:                             "empty",
+		previousEndpointsMap:             map[ServicePortName][]*BaseEndpointInfo{},
+		expectedResult:                   map[ServicePortName][]*BaseEndpointInfo{},
+		expectedConntrackCleanupRequired: false,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New[types.NamespacedName](),
 	}, {
 		name: "no change, unnamed port",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -560,10 +558,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: false, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New[types.NamespacedName](),
+		expectedConntrackCleanupRequired: false,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New[types.NamespacedName](),
 	}, {
 		name: "no change, named port, local",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -582,8 +579,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: true, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -614,10 +610,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.2", port: 12, endpoint: "1.1.1.2:12", isLocal: false, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New[types.NamespacedName](),
+		expectedConntrackCleanupRequired: false,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New[types.NamespacedName](),
 	}, {
 		name: "no change, multiple slices, multiple ports, local",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -650,8 +645,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.3", port: 13, endpoint: "1.1.1.3:13", isLocal: false, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -720,8 +714,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "2.2.2.2", port: 22, endpoint: "2.2.2.2:22", isLocal: true, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 2,
 			makeNSN("ns2", "ep2"): 1,
@@ -741,10 +734,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: true, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -762,14 +752,10 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: true, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedResult: map[ServicePortName][]*BaseEndpointInfo{},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{{
-			Endpoint:        "1.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New(makeNSN("ns1", "ep1")),
+		expectedConntrackCleanupRequired: true,
+		expectedResult:                   map[ServicePortName][]*BaseEndpointInfo{},
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New(makeNSN("ns1", "ep1")),
 	}, {
 		name: "add an IP and port",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -793,10 +779,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.2", port: 12, endpoint: "1.1.1.2:12", isLocal: true, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -824,19 +807,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: false, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{{
-			Endpoint:        "1.1.1.2:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "1.1.1.1:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "1.1.1.2:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New(makeNSN("ns1", "ep1")),
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New(makeNSN("ns1", "ep1")),
 	}, {
 		name: "add a slice to an endpoint",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -860,10 +833,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.2", port: 12, endpoint: "1.1.1.2:12", isLocal: true, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -891,13 +861,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: false, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{{
-			Endpoint:        "1.1.1.2:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New(makeNSN("ns1", "ep1")),
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New(makeNSN("ns1", "ep1")),
 	}, {
 		name: "rename a port",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -916,15 +882,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: false, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{{
-			Endpoint:        "1.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11-2", v1.ProtocolUDP): true,
-		},
-		expectedLocalEndpoints:   map[types.NamespacedName]int{},
-		expectedChangedEndpoints: sets.New(makeNSN("ns1", "ep1")),
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New(makeNSN("ns1", "ep1")),
 	}, {
 		name: "renumber a port",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -943,13 +903,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 22, endpoint: "1.1.1.1:22", isLocal: false, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{{
-			Endpoint:        "1.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New(makeNSN("ns1", "ep1")),
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New(makeNSN("ns1", "ep1")),
 	}, {
 		name: "complex add and remove",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -1015,27 +971,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "4.4.4.4", port: 44, endpoint: "4.4.4.4:44", isLocal: true, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{{
-			Endpoint:        "2.2.2.2:22",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "2.2.2.22:22",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "2.2.2.3:23",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p23", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "4.4.4.5:44",
-			ServicePortName: makeServicePortName("ns4", "ep4", "p44", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "4.4.4.6:45",
-			ServicePortName: makeServicePortName("ns4", "ep4", "p45", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP):  true,
-			makeServicePortName("ns1", "ep1", "p122", v1.ProtocolUDP): true,
-			makeServicePortName("ns3", "ep3", "p33", v1.ProtocolUDP):  true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns4", "ep4"): 1,
 		},
@@ -1054,12 +990,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: false, ready: true, serving: true, terminating: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP): true,
-		},
-		expectedLocalEndpoints:   map[types.NamespacedName]int{},
-		expectedChangedEndpoints: sets.New(makeNSN("ns1", "ep1")),
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New(makeNSN("ns1", "ep1")),
 	}, {
 		name: "change from ready to terminating pod",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -1078,10 +1011,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: false, ready: false, serving: true, terminating: true},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New(makeNSN("ns1", "ep1")),
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New(makeNSN("ns1", "ep1")),
 	}, {
 		name: "change from terminating to empty pod",
 		previousEndpointSlices: []*discovery.EndpointSlice{
@@ -1095,14 +1027,10 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{ip: "1.1.1.1", port: 11, endpoint: "1.1.1.1:11", isLocal: false, ready: false, serving: true, terminating: true},
 			},
 		},
-		expectedResult: map[ServicePortName][]*BaseEndpointInfo{},
-		expectedDeletedUDPEndpoints: []ServiceEndpoint{{
-			Endpoint:        "1.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
-		expectedChangedEndpoints:       sets.New(makeNSN("ns1", "ep1")),
+		expectedConntrackCleanupRequired: true,
+		expectedResult:                   map[ServicePortName][]*BaseEndpointInfo{},
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
+		expectedChangedEndpoints:         sets.New(makeNSN("ns1", "ep1")),
 	},
 	}
 
@@ -1147,36 +1075,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 			if !result.UpdatedServices.Equal(tc.expectedChangedEndpoints) {
 				t.Errorf("[%d] expected changed endpoints %q, got %q", tci, tc.expectedChangedEndpoints.UnsortedList(), result.UpdatedServices.UnsortedList())
 			}
-			if len(result.DeletedUDPEndpoints) != len(tc.expectedDeletedUDPEndpoints) {
-				t.Errorf("[%d] expected %d staleEndpoints, got %d: %v", tci, len(tc.expectedDeletedUDPEndpoints), len(result.DeletedUDPEndpoints), result.DeletedUDPEndpoints)
-			}
-			for _, x := range tc.expectedDeletedUDPEndpoints {
-				found := false
-				for _, stale := range result.DeletedUDPEndpoints {
-					if stale == x {
-						found = true
-						break
-					}
-				}
-				if !found {
-					t.Errorf("[%d] expected staleEndpoints[%v], but didn't find it: %v", tci, x, result.DeletedUDPEndpoints)
-				}
+			if result.ConntrackCleanupRequired != tc.expectedConntrackCleanupRequired {
+				t.Errorf("[%d] expected conntrackCleanupRequired to be %t, got %t", tci, tc.expectedConntrackCleanupRequired, result.ConntrackCleanupRequired)
 			}
-			if len(result.NewlyActiveUDPServices) != len(tc.expectedNewlyActiveUDPServices) {
-				t.Errorf("[%d] expected %d newlyActiveUDPServices, got %d: %v", tci, len(tc.expectedNewlyActiveUDPServices), len(result.NewlyActiveUDPServices), result.NewlyActiveUDPServices)
-			}
-			for svcName := range tc.expectedNewlyActiveUDPServices {
-				found := false
-				for _, newSvcName := range result.NewlyActiveUDPServices {
-					if newSvcName == svcName {
-						found = true
-					}
-				}
-				if !found {
-					t.Errorf("[%d] expected newlyActiveUDPServices[%v], but didn't find it: %v", tci, svcName, result.NewlyActiveUDPServices)
-				}
-			}
-
 			localReadyEndpoints := fp.endpointsMap.LocalReadyEndpoints()
 			if !reflect.DeepEqual(localReadyEndpoints, tc.expectedLocalEndpoints) {
 				t.Errorf("[%d] expected local ready endpoints %v, got %v", tci, tc.expectedLocalEndpoints, localReadyEndpoints)
diff --git a/pkg/proxy/iptables/proxier_test.go b/pkg/proxy/iptables/proxier_test.go
index a9d3a9e2269d2..b42490764a8ce 100644
--- a/pkg/proxy/iptables/proxier_test.go
+++ b/pkg/proxy/iptables/proxier_test.go
@@ -2966,16 +2966,11 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 	for i := range services {
 		fp.OnServiceAdd(services[i])
 	}
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 10 {
 		t.Errorf("expected service map length 10, got %v", fp.svcPortMap)
 	}
 
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
-
 	// The only-local-loadbalancer ones get added
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
@@ -3000,24 +2995,11 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 	fp.OnServiceDelete(services[2])
 	fp.OnServiceDelete(services[3])
 
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 1 {
 		t.Errorf("expected service map length 1, got %v", fp.svcPortMap)
 	}
 
-	// All services but one were deleted. While you'd expect only the ClusterIPs
-	// from the three deleted services here, we still have the ClusterIP for
-	// the not-deleted service, because one of it's ServicePorts was deleted.
-	expectedStaleUDPServices := []string{"172.30.55.10", "172.30.55.4", "172.30.55.11", "172.30.55.12"}
-	if len(result.DeletedUDPClusterIPs) != len(expectedStaleUDPServices) {
-		t.Errorf("expected stale UDP services length %d, got %v", len(expectedStaleUDPServices), result.DeletedUDPClusterIPs.UnsortedList())
-	}
-	for _, ip := range expectedStaleUDPServices {
-		if !result.DeletedUDPClusterIPs.Has(ip) {
-			t.Errorf("expected stale UDP service service %s", ip)
-		}
-	}
-
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected 0 healthcheck ports, got %v", healthCheckNodePorts)
@@ -3041,15 +3023,11 @@ func TestBuildServiceMapServiceHeadless(t *testing.T) {
 	)
 
 	// Headless service should be ignored
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 0 {
 		t.Errorf("expected service map length 0, got %d", len(fp.svcPortMap))
 	}
 
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
-
 	// No proxied services, so no healthchecks
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
@@ -3070,13 +3048,10 @@ func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
 		}),
 	)
 
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 0 {
 		t.Errorf("expected service map length 0, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs)
-	}
 	// No proxied services, so no healthchecks
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
@@ -3111,14 +3086,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	fp.OnServiceAdd(servicev1)
 
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected healthcheck ports length 0, got %v", healthCheckNodePorts)
@@ -3126,13 +3097,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	// Change service to load-balancer
 	fp.OnServiceUpdate(servicev1, servicev2)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs.UnsortedList())
-	}
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
 		t.Errorf("expected healthcheck ports length 1, got %v", healthCheckNodePorts)
@@ -3141,13 +3109,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 	// No change; make sure the service map stays the same and there are
 	// no health-check changes
 	fp.OnServiceUpdate(servicev2, servicev2)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs.UnsortedList())
-	}
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
 		t.Errorf("expected healthcheck ports length 1, got %v", healthCheckNodePorts)
@@ -3155,14 +3120,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	// And back to ClusterIP
 	fp.OnServiceUpdate(servicev2, servicev1)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected healthcheck ports length 0, got %v", healthCheckNodePorts)
@@ -3564,22 +3525,20 @@ func TestUpdateEndpointsMap(t *testing.T) {
 		// previousEndpoints and currentEndpoints are used to call appropriate
 		// handlers OnEndpoints* (based on whether corresponding values are nil
 		// or non-nil) and must be of equal length.
-		name                           string
-		previousEndpoints              []*discovery.EndpointSlice
-		currentEndpoints               []*discovery.EndpointSlice
-		oldEndpoints                   map[proxy.ServicePortName][]endpointExpectation
-		expectedResult                 map[proxy.ServicePortName][]endpointExpectation
-		expectedDeletedUDPEndpoints    []proxy.ServiceEndpoint
-		expectedNewlyActiveUDPServices map[proxy.ServicePortName]bool
-		expectedLocalEndpoints         map[types.NamespacedName]int
+		name                             string
+		previousEndpoints                []*discovery.EndpointSlice
+		currentEndpoints                 []*discovery.EndpointSlice
+		oldEndpoints                     map[proxy.ServicePortName][]endpointExpectation
+		expectedResult                   map[proxy.ServicePortName][]endpointExpectation
+		expectedConntrackCleanupRequired bool
+		expectedLocalEndpoints           map[types.NamespacedName]int
 	}{{
 		// Case[0]: nothing
-		name:                           "nothing",
-		oldEndpoints:                   map[proxy.ServicePortName][]endpointExpectation{},
-		expectedResult:                 map[proxy.ServicePortName][]endpointExpectation{},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		name:                             "nothing",
+		oldEndpoints:                     map[proxy.ServicePortName][]endpointExpectation{},
+		expectedResult:                   map[proxy.ServicePortName][]endpointExpectation{},
+		expectedConntrackCleanupRequired: false,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[1]: no change, named port, local
 		name:              "no change, named port, local",
@@ -3595,8 +3554,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3621,9 +3579,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.2:12", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: false,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[3]: no change, multiple subsets, multiple ports, local
 		name:              "no change, multiple subsets, multiple ports, local",
@@ -3651,8 +3608,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.3:13", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3713,8 +3669,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.2.2.2:22", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 2,
 			makeNSN("ns2", "ep2"): 1,
@@ -3730,10 +3685,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3747,13 +3699,9 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedResult: map[proxy.ServicePortName][]endpointExpectation{},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedResult:                   map[proxy.ServicePortName][]endpointExpectation{},
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[7]: add an IP and port
 		name:              "add an IP and port",
@@ -3774,10 +3722,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.2:12", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3801,18 +3746,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.2:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.1.1.1:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.1.1.2:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[9]: add a subset
 		name:              "add a subset",
@@ -3831,10 +3766,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.2:12", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3856,12 +3788,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.2:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[11]: rename a port
 		name:              "rename a port",
@@ -3877,14 +3805,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11-2", v1.ProtocolUDP): true,
-		},
-		expectedLocalEndpoints: map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[12]: renumber a port
 		name:              "renumber a port",
@@ -3900,12 +3822,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:22", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[13]: complex add and remove
 		name:              "complex add and remove",
@@ -3948,27 +3866,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.4.4.4:44", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.2.2.2:22",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.2.2.22:22",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.2.2.3:23",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p23", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.4.4.5:44",
-			ServicePortName: makeServicePortName("ns4", "ep4", "p44", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.4.4.6:45",
-			ServicePortName: makeServicePortName("ns4", "ep4", "p45", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP):  true,
-			makeServicePortName("ns1", "ep1", "p122", v1.ProtocolUDP): true,
-			makeServicePortName("ns3", "ep3", "p33", v1.ProtocolUDP):  true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns4", "ep4"): 1,
 		},
@@ -3983,11 +3881,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): true,
-		},
-		expectedLocalEndpoints: map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	},
 	}
 
@@ -4026,34 +3921,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 			result := fp.endpointsMap.Update(fp.endpointsChanges)
 			newMap := fp.endpointsMap
 			checkEndpointExpectations(t, tci, newMap, tc.expectedResult)
-			if len(result.DeletedUDPEndpoints) != len(tc.expectedDeletedUDPEndpoints) {
-				t.Errorf("[%d] expected %d staleEndpoints, got %d: %v", tci, len(tc.expectedDeletedUDPEndpoints), len(result.DeletedUDPEndpoints), result.DeletedUDPEndpoints)
-			}
-			for _, x := range tc.expectedDeletedUDPEndpoints {
-				found := false
-				for _, stale := range result.DeletedUDPEndpoints {
-					if stale == x {
-						found = true
-						break
-					}
-				}
-				if !found {
-					t.Errorf("[%d] expected staleEndpoints[%v], but didn't find it: %v", tci, x, result.DeletedUDPEndpoints)
-				}
-			}
-			if len(result.NewlyActiveUDPServices) != len(tc.expectedNewlyActiveUDPServices) {
-				t.Errorf("[%d] expected %d staleServiceNames, got %d: %v", tci, len(tc.expectedNewlyActiveUDPServices), len(result.NewlyActiveUDPServices), result.NewlyActiveUDPServices)
-			}
-			for svcName := range tc.expectedNewlyActiveUDPServices {
-				found := false
-				for _, stale := range result.NewlyActiveUDPServices {
-					if stale == svcName {
-						found = true
-					}
-				}
-				if !found {
-					t.Errorf("[%d] expected staleServiceNames[%v], but didn't find it: %v", tci, svcName, result.NewlyActiveUDPServices)
-				}
+			if result.ConntrackCleanupRequired != tc.expectedConntrackCleanupRequired {
+				t.Errorf("[%d] expected conntrackCleanupRequired to be %t, got %t", tci, tc.expectedConntrackCleanupRequired, result.ConntrackCleanupRequired)
 			}
 			localReadyEndpoints := fp.endpointsMap.LocalReadyEndpoints()
 			if !reflect.DeepEqual(localReadyEndpoints, tc.expectedLocalEndpoints) {
diff --git a/pkg/proxy/ipvs/proxier_test.go b/pkg/proxy/ipvs/proxier_test.go
index cf6eb6451e72c..1b54f9a285eda 100644
--- a/pkg/proxy/ipvs/proxier_test.go
+++ b/pkg/proxy/ipvs/proxier_test.go
@@ -2517,16 +2517,11 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 	for i := range services {
 		fp.OnServiceAdd(services[i])
 	}
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 12 {
 		t.Errorf("expected service map length 12, got %v", fp.svcPortMap)
 	}
 
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
-
 	// The only-local-loadbalancer ones get added
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
@@ -2551,24 +2546,10 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 	fp.OnServiceDelete(services[2])
 	fp.OnServiceDelete(services[3])
 
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 1 {
 		t.Errorf("expected service map length 1, got %v", fp.svcPortMap)
 	}
-
-	// All services but one were deleted. While you'd expect only the ClusterIPs
-	// from the three deleted services here, we still have the ClusterIP for
-	// the not-deleted service, because one of it's ServicePorts was deleted.
-	expectedStaleUDPServices := []string{"172.16.55.10", "172.16.55.4", "172.16.55.11", "172.16.55.12"}
-	if len(result.DeletedUDPClusterIPs) != len(expectedStaleUDPServices) {
-		t.Errorf("expected stale UDP services length %d, got %v", len(expectedStaleUDPServices), result.DeletedUDPClusterIPs.UnsortedList())
-	}
-	for _, ip := range expectedStaleUDPServices {
-		if !result.DeletedUDPClusterIPs.Has(ip) {
-			t.Errorf("expected stale UDP service service %s", ip)
-		}
-	}
-
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected 0 healthcheck ports, got %v", healthCheckNodePorts)
@@ -2600,15 +2581,11 @@ func TestBuildServiceMapServiceHeadless(t *testing.T) {
 	)
 
 	// Headless service should be ignored
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 0 {
 		t.Errorf("expected service map length 0, got %d", len(fp.svcPortMap))
 	}
 
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
-
 	// No proxied services, so no healthchecks
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
@@ -2632,13 +2609,10 @@ func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
 		}),
 	)
 
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 0 {
 		t.Errorf("expected service map length 0, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs)
-	}
 
 	// No proxied services, so no healthchecks
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
@@ -2677,15 +2651,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	fp.OnServiceAdd(servicev1)
 
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
-
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected healthcheck ports length 0, got %v", healthCheckNodePorts)
@@ -2693,13 +2662,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	// Change service to load-balancer
 	fp.OnServiceUpdate(servicev1, servicev2)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs.UnsortedList())
-	}
 
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
@@ -2709,13 +2675,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 	// No change; make sure the service map stays the same and there are
 	// no health-check changes
 	fp.OnServiceUpdate(servicev2, servicev2)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs.UnsortedList())
-	}
 
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
@@ -2724,14 +2687,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	// And back to ClusterIP
 	fp.OnServiceUpdate(servicev2, servicev1)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
@@ -3127,22 +3086,20 @@ func Test_updateEndpointsMap(t *testing.T) {
 		// previousEndpoints and currentEndpoints are used to call appropriate
 		// handlers OnEndpoints* (based on whether corresponding values are nil
 		// or non-nil) and must be of equal length.
-		name                           string
-		previousEndpoints              []*discovery.EndpointSlice
-		currentEndpoints               []*discovery.EndpointSlice
-		oldEndpoints                   map[proxy.ServicePortName][]endpointExpectation
-		expectedResult                 map[proxy.ServicePortName][]endpointExpectation
-		expectedDeletedUDPEndpoints    []proxy.ServiceEndpoint
-		expectedNewlyActiveUDPServices map[proxy.ServicePortName]bool
-		expectedReadyEndpoints         map[types.NamespacedName]int
+		name                             string
+		previousEndpoints                []*discovery.EndpointSlice
+		currentEndpoints                 []*discovery.EndpointSlice
+		oldEndpoints                     map[proxy.ServicePortName][]endpointExpectation
+		expectedResult                   map[proxy.ServicePortName][]endpointExpectation
+		expectedConntrackCleanupRequired bool
+		expectedReadyEndpoints           map[types.NamespacedName]int
 	}{{
 		// Case[0]: nothing
-		name:                           "nothing",
-		oldEndpoints:                   map[proxy.ServicePortName][]endpointExpectation{},
-		expectedResult:                 map[proxy.ServicePortName][]endpointExpectation{},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedReadyEndpoints:         map[types.NamespacedName]int{},
+		name:                             "nothing",
+		oldEndpoints:                     map[proxy.ServicePortName][]endpointExpectation{},
+		expectedResult:                   map[proxy.ServicePortName][]endpointExpectation{},
+		expectedConntrackCleanupRequired: false,
+		expectedReadyEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[1]: no change, named port, local
 		name:              "no change, named port, local",
@@ -3158,8 +3115,7 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedReadyEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3184,9 +3140,8 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.2:12", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedReadyEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: false,
+		expectedReadyEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[3]: no change, multiple subsets, multiple ports, local
 		name:              "no change, multiple subsets, multiple ports, local",
@@ -3214,8 +3169,7 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.3:13", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedReadyEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3276,8 +3230,7 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "2.2.2.2:22", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedReadyEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 2,
 			makeNSN("ns2", "ep2"): 1,
@@ -3293,10 +3246,7 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedReadyEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3310,13 +3260,9 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedResult: map[proxy.ServicePortName][]endpointExpectation{},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "1.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedReadyEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedResult:                   map[proxy.ServicePortName][]endpointExpectation{},
+		expectedReadyEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[7]: add an IP and port
 		name:              "add an IP and port",
@@ -3337,10 +3283,7 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.2:12", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedReadyEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3364,18 +3307,8 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "1.1.1.2:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "1.1.1.1:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "1.1.1.2:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedReadyEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedReadyEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[9]: add a subset
 		name:              "add a subset",
@@ -3394,10 +3327,7 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.2:12", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedReadyEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -3419,12 +3349,8 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "1.1.1.2:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedReadyEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedReadyEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[11]: rename a port
 		name:              "rename a port",
@@ -3440,14 +3366,8 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "1.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11-2", v1.ProtocolUDP): true,
-		},
-		expectedReadyEndpoints: map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedReadyEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[12]: renumber a port
 		name:              "renumber a port",
@@ -3463,12 +3383,8 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.1:22", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "1.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedReadyEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedReadyEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[13]: complex add and remove
 		name:              "complex add and remove",
@@ -3511,27 +3427,7 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "4.4.4.4:44", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "2.2.2.2:22",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "2.2.2.22:22",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "2.2.2.3:23",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p23", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "4.4.4.5:44",
-			ServicePortName: makeServicePortName("ns4", "ep4", "p44", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "4.4.4.6:45",
-			ServicePortName: makeServicePortName("ns4", "ep4", "p45", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP):  true,
-			makeServicePortName("ns1", "ep1", "p122", v1.ProtocolUDP): true,
-			makeServicePortName("ns3", "ep3", "p33", v1.ProtocolUDP):  true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedReadyEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns4", "ep4"): 1,
 		},
@@ -3546,11 +3442,8 @@ func Test_updateEndpointsMap(t *testing.T) {
 				{endpoint: "1.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): true,
-		},
-		expectedReadyEndpoints: map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedReadyEndpoints:           map[types.NamespacedName]int{},
 	},
 	}
 
@@ -3592,35 +3485,8 @@ func Test_updateEndpointsMap(t *testing.T) {
 			result := fp.endpointsMap.Update(fp.endpointsChanges)
 			newMap := fp.endpointsMap
 			checkEndpointExpectations(t, tci, newMap, tc.expectedResult)
-			if len(result.DeletedUDPEndpoints) != len(tc.expectedDeletedUDPEndpoints) {
-				t.Errorf("[%d] expected %d staleEndpoints, got %d: %v", tci, len(tc.expectedDeletedUDPEndpoints), len(result.DeletedUDPEndpoints), result.DeletedUDPEndpoints)
-			}
-			for _, x := range tc.expectedDeletedUDPEndpoints {
-				found := false
-				for _, stale := range result.DeletedUDPEndpoints {
-					if stale == x {
-						found = true
-						break
-					}
-				}
-				if !found {
-					t.Errorf("[%d] expected staleEndpoints[%v], but didn't find it: %v", tci, x, result.DeletedUDPEndpoints)
-				}
-			}
-			if len(result.NewlyActiveUDPServices) != len(tc.expectedNewlyActiveUDPServices) {
-				t.Errorf("[%d] expected %d staleServiceNames, got %d: %v", tci, len(tc.expectedNewlyActiveUDPServices), len(result.NewlyActiveUDPServices), result.NewlyActiveUDPServices)
-			}
-			for svcName := range tc.expectedNewlyActiveUDPServices {
-				found := false
-				for _, stale := range result.NewlyActiveUDPServices {
-					if stale == svcName {
-						found = true
-						break
-					}
-				}
-				if !found {
-					t.Errorf("[%d] expected staleServiceNames[%v], but didn't find it: %v", tci, svcName, result.NewlyActiveUDPServices)
-				}
+			if result.ConntrackCleanupRequired != tc.expectedConntrackCleanupRequired {
+				t.Errorf("[%d] expected conntrackCleanupRequired to be %t, got %t", tci, tc.expectedConntrackCleanupRequired, result.ConntrackCleanupRequired)
 			}
 			localReadyEndpoints := fp.endpointsMap.LocalReadyEndpoints()
 			if !reflect.DeepEqual(localReadyEndpoints, tc.expectedReadyEndpoints) {
diff --git a/pkg/proxy/nftables/proxier_test.go b/pkg/proxy/nftables/proxier_test.go
index 379070e871e22..d94f8590ed483 100644
--- a/pkg/proxy/nftables/proxier_test.go
+++ b/pkg/proxy/nftables/proxier_test.go
@@ -1374,16 +1374,11 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 	for i := range services {
 		fp.OnServiceAdd(services[i])
 	}
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 10 {
 		t.Errorf("expected service map length 10, got %v", fp.svcPortMap)
 	}
 
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
-
 	// The only-local-loadbalancer ones get added
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
@@ -1408,24 +1403,11 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 	fp.OnServiceDelete(services[2])
 	fp.OnServiceDelete(services[3])
 
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 1 {
 		t.Errorf("expected service map length 1, got %v", fp.svcPortMap)
 	}
 
-	// All services but one were deleted. While you'd expect only the ClusterIPs
-	// from the three deleted services here, we still have the ClusterIP for
-	// the not-deleted service, because one of it's ServicePorts was deleted.
-	expectedStaleUDPServices := []string{"172.30.55.10", "172.30.55.4", "172.30.55.11", "172.30.55.12"}
-	if len(result.DeletedUDPClusterIPs) != len(expectedStaleUDPServices) {
-		t.Errorf("expected stale UDP services length %d, got %v", len(expectedStaleUDPServices), result.DeletedUDPClusterIPs.UnsortedList())
-	}
-	for _, ip := range expectedStaleUDPServices {
-		if !result.DeletedUDPClusterIPs.Has(ip) {
-			t.Errorf("expected stale UDP service service %s", ip)
-		}
-	}
-
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected 0 healthcheck ports, got %v", healthCheckNodePorts)
@@ -1448,15 +1430,11 @@ func TestBuildServiceMapServiceHeadless(t *testing.T) {
 	)
 
 	// Headless service should be ignored
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 0 {
 		t.Errorf("expected service map length 0, got %d", len(fp.svcPortMap))
 	}
 
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
-
 	// No proxied services, so no healthchecks
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
@@ -1476,13 +1454,10 @@ func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
 		}),
 	)
 
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 0 {
 		t.Errorf("expected service map length 0, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs)
-	}
 	// No proxied services, so no healthchecks
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
@@ -1516,14 +1491,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	fp.OnServiceAdd(servicev1)
 
-	result := fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected healthcheck ports length 0, got %v", healthCheckNodePorts)
@@ -1531,13 +1502,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	// Change service to load-balancer
 	fp.OnServiceUpdate(servicev1, servicev2)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs.UnsortedList())
-	}
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
 		t.Errorf("expected healthcheck ports length 1, got %v", healthCheckNodePorts)
@@ -1546,13 +1514,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 	// No change; make sure the service map stays the same and there are
 	// no health-check changes
 	fp.OnServiceUpdate(servicev2, servicev2)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs.UnsortedList())
-	}
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
 		t.Errorf("expected healthcheck ports length 1, got %v", healthCheckNodePorts)
@@ -1560,14 +1525,10 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 
 	// And back to ClusterIP
 	fp.OnServiceUpdate(servicev2, servicev1)
-	result = fp.svcPortMap.Update(fp.serviceChanges)
+	fp.svcPortMap.Update(fp.serviceChanges)
 	if len(fp.svcPortMap) != 2 {
 		t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected healthcheck ports length 0, got %v", healthCheckNodePorts)
@@ -1969,22 +1930,20 @@ func TestUpdateEndpointsMap(t *testing.T) {
 		// previousEndpoints and currentEndpoints are used to call appropriate
 		// handlers OnEndpoints* (based on whether corresponding values are nil
 		// or non-nil) and must be of equal length.
-		name                           string
-		previousEndpoints              []*discovery.EndpointSlice
-		currentEndpoints               []*discovery.EndpointSlice
-		oldEndpoints                   map[proxy.ServicePortName][]endpointExpectation
-		expectedResult                 map[proxy.ServicePortName][]endpointExpectation
-		expectedDeletedUDPEndpoints    []proxy.ServiceEndpoint
-		expectedNewlyActiveUDPServices map[proxy.ServicePortName]bool
-		expectedLocalEndpoints         map[types.NamespacedName]int
+		name                             string
+		previousEndpoints                []*discovery.EndpointSlice
+		currentEndpoints                 []*discovery.EndpointSlice
+		oldEndpoints                     map[proxy.ServicePortName][]endpointExpectation
+		expectedResult                   map[proxy.ServicePortName][]endpointExpectation
+		expectedConntrackCleanupRequired bool
+		expectedLocalEndpoints           map[types.NamespacedName]int
 	}{{
 		// Case[0]: nothing
-		name:                           "nothing",
-		oldEndpoints:                   map[proxy.ServicePortName][]endpointExpectation{},
-		expectedResult:                 map[proxy.ServicePortName][]endpointExpectation{},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		name:                             "nothing",
+		oldEndpoints:                     map[proxy.ServicePortName][]endpointExpectation{},
+		expectedResult:                   map[proxy.ServicePortName][]endpointExpectation{},
+		expectedConntrackCleanupRequired: false,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[1]: no change, named port, local
 		name:              "no change, named port, local",
@@ -2000,8 +1959,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -2026,9 +1984,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.2:12", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: false,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[3]: no change, multiple subsets, multiple ports, local
 		name:              "no change, multiple subsets, multiple ports, local",
@@ -2056,8 +2013,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.3:13", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -2118,8 +2074,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.2.2.2:22", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints:    []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
+		expectedConntrackCleanupRequired: false,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 2,
 			makeNSN("ns2", "ep2"): 1,
@@ -2135,10 +2090,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -2152,13 +2104,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: true},
 			},
 		},
-		expectedResult: map[proxy.ServicePortName][]endpointExpectation{},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[7]: add an IP and port
 		name:              "add an IP and port",
@@ -2179,10 +2126,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.2:12", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -2206,18 +2150,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.2:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.1.1.1:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.1.1.2:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[9]: add a subset
 		name:              "add a subset",
@@ -2236,10 +2170,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.2:12", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP): true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns1", "ep1"): 1,
 		},
@@ -2261,12 +2192,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.2:12",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[11]: rename a port
 		name:              "rename a port",
@@ -2282,14 +2209,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11-2", v1.ProtocolUDP): true,
-		},
-		expectedLocalEndpoints: map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[12]: renumber a port
 		name:              "renumber a port",
@@ -2305,12 +2226,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:22", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.1.1.1:11",
-			ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
-		expectedLocalEndpoints:         map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	}, {
 		// Case[13]: complex add and remove
 		name:              "complex add and remove",
@@ -2353,27 +2270,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.4.4.4:44", isLocal: true},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
-			Endpoint:        "10.2.2.2:22",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.2.2.22:22",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p22", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.2.2.3:23",
-			ServicePortName: makeServicePortName("ns2", "ep2", "p23", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.4.4.5:44",
-			ServicePortName: makeServicePortName("ns4", "ep4", "p44", v1.ProtocolUDP),
-		}, {
-			Endpoint:        "10.4.4.6:45",
-			ServicePortName: makeServicePortName("ns4", "ep4", "p45", v1.ProtocolUDP),
-		}},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p12", v1.ProtocolUDP):  true,
-			makeServicePortName("ns1", "ep1", "p122", v1.ProtocolUDP): true,
-			makeServicePortName("ns3", "ep3", "p33", v1.ProtocolUDP):  true,
-		},
+		expectedConntrackCleanupRequired: true,
 		expectedLocalEndpoints: map[types.NamespacedName]int{
 			makeNSN("ns4", "ep4"): 1,
 		},
@@ -2388,11 +2285,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 				{endpoint: "10.1.1.1:11", isLocal: false},
 			},
 		},
-		expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
-		expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{
-			makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): true,
-		},
-		expectedLocalEndpoints: map[types.NamespacedName]int{},
+		expectedConntrackCleanupRequired: true,
+		expectedLocalEndpoints:           map[types.NamespacedName]int{},
 	},
 	}
 
@@ -2430,34 +2324,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
 			result := fp.endpointsMap.Update(fp.endpointsChanges)
 			newMap := fp.endpointsMap
 			checkEndpointExpectations(t, tci, newMap, tc.expectedResult)
-			if len(result.DeletedUDPEndpoints) != len(tc.expectedDeletedUDPEndpoints) {
-				t.Errorf("[%d] expected %d staleEndpoints, got %d: %v", tci, len(tc.expectedDeletedUDPEndpoints), len(result.DeletedUDPEndpoints), result.DeletedUDPEndpoints)
-			}
-			for _, x := range tc.expectedDeletedUDPEndpoints {
-				found := false
-				for _, stale := range result.DeletedUDPEndpoints {
-					if stale == x {
-						found = true
-						break
-					}
-				}
-				if !found {
-					t.Errorf("[%d] expected staleEndpoints[%v], but didn't find it: %v", tci, x, result.DeletedUDPEndpoints)
-				}
-			}
-			if len(result.NewlyActiveUDPServices) != len(tc.expectedNewlyActiveUDPServices) {
-				t.Errorf("[%d] expected %d staleServiceNames, got %d: %v", tci, len(tc.expectedNewlyActiveUDPServices), len(result.NewlyActiveUDPServices), result.NewlyActiveUDPServices)
-			}
-			for svcName := range tc.expectedNewlyActiveUDPServices {
-				found := false
-				for _, stale := range result.NewlyActiveUDPServices {
-					if stale == svcName {
-						found = true
-					}
-				}
-				if !found {
-					t.Errorf("[%d] expected staleServiceNames[%v], but didn't find it: %v", tci, svcName, result.NewlyActiveUDPServices)
-				}
+			if result.ConntrackCleanupRequired != tc.expectedConntrackCleanupRequired {
+				t.Errorf("[%d] expected conntrackCleanupRequired to be %t, got %t", tci, tc.expectedConntrackCleanupRequired, result.ConntrackCleanupRequired)
 			}
 			localReadyEndpoints := fp.endpointsMap.LocalReadyEndpoints()
 			if !reflect.DeepEqual(localReadyEndpoints, tc.expectedLocalEndpoints) {
diff --git a/pkg/proxy/servicechangetracker_test.go b/pkg/proxy/servicechangetracker_test.go
index a989d097da588..e139d6648247f 100644
--- a/pkg/proxy/servicechangetracker_test.go
+++ b/pkg/proxy/servicechangetracker_test.go
@@ -680,9 +680,6 @@ func TestServiceMapUpdateHeadless(t *testing.T) {
 	if len(result.UpdatedServices) != 0 {
 		t.Errorf("expected 0 updated services, got %d", len(result.UpdatedServices))
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 
 	// No proxied services, so no healthchecks
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
@@ -710,9 +707,6 @@ func TestUpdateServiceTypeExternalName(t *testing.T) {
 	if len(result.UpdatedServices) != 0 {
 		t.Errorf("expected 0 updated services, got %v", result.UpdatedServices)
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs)
-	}
 
 	// No proxied services, so no healthchecks
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
@@ -782,10 +776,6 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 	if len(result.UpdatedServices) != len(services) {
 		t.Errorf("expected %d updated services, got %d", len(services), len(result.UpdatedServices))
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 
 	// The only-local-loadbalancer ones get added
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
@@ -823,19 +813,6 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
 	if len(healthCheckNodePorts) != 0 {
 		t.Errorf("expected 0 healthcheck ports, got %v", healthCheckNodePorts)
 	}
-
-	// All services but one were deleted. While you'd expect only the ClusterIPs
-	// from the three deleted services here, we still have the ClusterIP for
-	// the not-deleted service, because one of it's ServicePorts was deleted.
-	expectedDeletedUDPClusterIPs := []string{"172.16.55.10", "172.16.55.4", "172.16.55.11", "172.16.55.12"}
-	if len(result.DeletedUDPClusterIPs) != len(expectedDeletedUDPClusterIPs) {
-		t.Errorf("expected stale UDP services length %d, got %v", len(expectedDeletedUDPClusterIPs), result.DeletedUDPClusterIPs.UnsortedList())
-	}
-	for _, ip := range expectedDeletedUDPClusterIPs {
-		if !result.DeletedUDPClusterIPs.Has(ip) {
-			t.Errorf("expected stale UDP service service %s", ip)
-		}
-	}
 }
 
 func TestBuildServiceMapServiceUpdate(t *testing.T) {
@@ -871,10 +848,6 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 	if len(result.UpdatedServices) != 1 {
 		t.Errorf("expected 1 updated service, got %d", len(result.UpdatedServices))
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 
 	healthCheckNodePorts := fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {
@@ -890,9 +863,6 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 	if len(result.UpdatedServices) != 1 {
 		t.Errorf("expected 1 updated service, got %d", len(result.UpdatedServices))
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs.UnsortedList())
-	}
 
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
@@ -909,9 +879,6 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 	if len(result.UpdatedServices) != 0 {
 		t.Errorf("expected 0 updated services, got %d", len(result.UpdatedServices))
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		t.Errorf("expected stale UDP services length 0, got %v", result.DeletedUDPClusterIPs.UnsortedList())
-	}
 
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 1 {
@@ -927,10 +894,6 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 	if len(result.UpdatedServices) != 1 {
 		t.Errorf("expected 1 updated service, got %d", len(result.UpdatedServices))
 	}
-	if len(result.DeletedUDPClusterIPs) != 0 {
-		// Services only added, so nothing stale yet
-		t.Errorf("expected stale UDP services length 0, got %d", len(result.DeletedUDPClusterIPs))
-	}
 
 	healthCheckNodePorts = fp.svcPortMap.HealthCheckNodePorts()
 	if len(healthCheckNodePorts) != 0 {