diff --git a/pkg/proxy/conntrack/cleanup.go b/pkg/proxy/conntrack/cleanup.go
index 8035fcbdb90..f04bba05938 100644
--- a/pkg/proxy/conntrack/cleanup.go
+++ b/pkg/proxy/conntrack/cleanup.go
@@ -86,10 +86,15 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 			}
 		}
 
-		// a Service without endpoints does not require to clean the conntrack entries associated.
-		if endpoints.Len() == 0 {
-			continue
-		}
+		// Note: a Service without any serving endpoints is processed too, with an
+		// empty endpoints set, so that all the existing entries directed to its
+		// frontends are removed. The REJECT (iptables) / reject (nftables) rule
+		// installed for a Service with no endpoints does not cover the previously
+		// established flows: those are DNATed to the (deleted) endpoint IP before
+		// the reject rule, which matches on the Service IP, can be evaluated.
+		// One-way UDP flows (e.g. statsd) refresh the conntrack entry timeout with
+		// every packet, so without this cleanup they keep sending traffic to the
+		// deleted endpoint IP indefinitely.
 
 		// we need to filter entries that are directed to a Service IP:Port frontend
 		// that does not have an Endpoint IP:Port backend as part of the serving endpoints
@@ -105,8 +110,13 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 			serviceIPEndpoints[net.JoinHostPort(externalIP.String(), portStr)] = endpoints
 		}
 		// we need to filter entries that are directed to a *:NodePort that does not have
-		// an Endpoint IP:Port backend as part of the serving endpoints
-		if svc.NodePort() != 0 {
+		// an Endpoint IP:Port backend as part of the serving endpoints.
+		// NodePort entries are matched on the destination port only, so with an
+		// empty endpoints set every UDP flow towards that port number would be
+		// removed, including flows not owned by kube-proxy (e.g. traffic to an
+		// unrelated host on the same port). Skip NodePort cleanup for services
+		// without serving endpoints until the match can be restricted to node IPs.
+		if svc.NodePort() != 0 && endpoints.Len() > 0 {
 			// *:NodePort
 			serviceNodePortEndpoints[svc.NodePort()] = endpoints
 		}
diff --git a/pkg/proxy/conntrack/cleanup_test.go b/pkg/proxy/conntrack/cleanup_test.go
index 98afa791064..1213dc25a2d 100644
--- a/pkg/proxy/conntrack/cleanup_test.go
+++ b/pkg/proxy/conntrack/cleanup_test.go
@@ -458,6 +458,7 @@ func TestServiceWithoutEndpoints(t *testing.T) {
 				{
 					Name:     "test-udp",
 					Port:     testServicePort,
+					NodePort: testServiceNodePort,
 					Protocol: v1.ProtocolUDP,
 				},
 			},
@@ -508,10 +509,38 @@ func TestServiceWithoutEndpoints(t *testing.T) {
 	_ = endpointsMap.Update(ect)
 
 	flows := []*netlink.ConntrackFlow{}
-	// 1 valid entry
-	flows = append(flows, generateConntrackEntry(testExternalIP, testServicePort, testServingEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
-	// 1 stale entry
-	flows = append(flows, generateConntrackEntry(testExternalIP, testServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
+	var entriesAfterCleanup []*netlink.ConntrackFlow
+
+	// All UDP entries directed to the frontends of a service without serving
+	// endpoints are stale and must be cleared, regardless of which (deleted)
+	// endpoint they are DNATed to: established flows bypass the REJECT rule
+	// installed for services with no endpoints and would otherwise keep
+	// blackholing traffic to the deleted endpoint IP indefinitely.
+	for _, origDest := range []string{testClusterIP, testLoadBalancerIP, testExternalIP} {
+		for _, dnatDest := range []string{testServingEndpointIP, testDeletedEndpointIP} {
+			flows = append(flows, generateConntrackEntry(origDest, testServicePort, dnatDest, testEndpointPort, unix.IPPROTO_UDP))
+		}
+	}
+
+	// UDP entries not directed to a service frontend are not owned by
+	// kube-proxy and must be preserved.
+	entry := generateConntrackEntry(testExternalIP, testNonServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP)
+	flows = append(flows, entry)
+	entriesAfterCleanup = append(entriesAfterCleanup, entry)
+
+	// non-UDP entries must be preserved.
+	entry = generateConntrackEntry(testExternalIP, testServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_TCP)
+	flows = append(flows, entry)
+	entriesAfterCleanup = append(entriesAfterCleanup, entry)
+
+	// *:NodePort entries are matched on the destination port only, which with an
+	// empty endpoints set would also remove flows not owned by kube-proxy, so
+	// NodePort cleanup is skipped for services without serving endpoints and
+	// these entries must be preserved.
+	entry = generateConntrackEntry("", testServiceNodePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP)
+	flows = append(flows, entry)
+	entriesAfterCleanup = append(entriesAfterCleanup, entry)
+
 	ct := newConntracker(
 		&fakeHandler{
 			entries: flows,
@@ -520,7 +549,19 @@ func TestServiceWithoutEndpoints(t *testing.T) {
 
 	CleanStaleEntries(ct, testIPFamily, svcPortMap, endpointsMap)
 	actualEntries, _ := ct.ListEntries(ipFamilyMap[testIPFamily])
-	if len(actualEntries) != 2 {
-		t.Errorf("unexpected number of entries, got %d expected %d", len(actualEntries), 2)
+
+	require.Len(t, actualEntries, len(entriesAfterCleanup))
+
+	// 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(entriesAfterCleanup, func(i, j int) bool {
+		return entriesAfterCleanup[i].String() < entriesAfterCleanup[j].String()
+	})
+
+	if diff := cmp.Diff(entriesAfterCleanup, actualEntries); len(diff) > 0 {
+		t.Errorf("unexpected entries after cleanup: %s", diff)
 	}
 }