Repository: antrea-io/antrea
Pull request #3313 (2022-02-12 00:10:45)

Fix FlowAggregator e2e test random failures
In this commit, we fixed some random failure cases of the FlowAggregator e2e test, which are previously raised by Jianjun in the email thread.

One fix is for the ToExternalFlows test, the reason this test fails randomly is that `wget` is used instead of `iperf -t 12` to generate traffic, so 6 seconds of waiting time is sometimes not enough because the first record theoretically arrives after 5.5 seconds `(exporterActiveFlowExportTimeout+aggregatorActiveFlowRecordTimeout)`. So we changed to it 9 seconds (5.5s plus one more `aggregatorActiveFlowRecordTimeout`) for waiting for flow records received on the IPFIX collector.

Another fix is adding `flowEndReason==3` in the condition of receiving the last flow record. We found in a low possibility, the `flowEndSeconds - flowStartSeconds` of the last flow record is equal to `11` instead of `>= 12`. So we add `flowEndReason==3` in the condition of finding the last flow record. 

Added 2/18:
The third fix is setting a bandwidth limit of iperf command to make the throughput test more stable. We found when running iperf command with unlimited bandwidth by default, the throughput fluctuates greatly (±20%) at the beginning. So we set a limit of 10Mbits/s for it and the throughput test becomes more stable.

I have triggered 5 times of Kind test of Flow Aggregator on GitHub and no failures were found: https://github.com/antrea-io/antrea/runs/5221164243?check_suite_focus=true

Signed-off-by: Yongming Ding <dyongming@vmware.com>

codecov-commenter: # [Codecov](https://codecov.io/gh/antrea-io/antrea/pull/3313?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#3313](https://codecov.io/gh/antrea-io/antrea/pull/3313?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (6649e26) into [main](https://codecov.io/gh/antrea-io/antrea/commit/f384995edf7dd660d898c46175eee2a665cbaeda?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (f384995) will **decrease** coverage by `11.37%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/antrea-io/antrea/pull/3313/graphs/tree.svg?width=650&height=150&src=pr&token=TjWaLL2GoC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)](https://codecov.io/gh/antrea-io/antrea/pull/3313?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)

```diff
@@             Coverage Diff             @@
##             main    #3313       +/-   ##
===========================================
- Coverage   60.13%   48.76%   -11.38%     
===========================================
  Files         331      331               
  Lines       28420    28420               
===========================================
- Hits        17090    13858     -3232     
- Misses       9460    13036     +3576     
+ Partials     1870     1526      -344     
```

| Flag | Coverage Δ | |
|---|---|---|
| kind-e2e-tests | `26.94% <ø> (-20.94%)` | :arrow_down: |
| unit-tests | `41.80% <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/antrea-io/antrea/pull/3313?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pkg/agent/nodeportlocal/rules/rules.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvcnVsZXMvcnVsZXMuZ28=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/k8s/annotations.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvazhzL2Fubm90YXRpb25zLmdv) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/agent/controller/networkpolicy/reject.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbnRyb2xsZXIvbmV0d29ya3BvbGljeS9yZWplY3QuZ28=) | `0.00% <0.00%> (-87.91%)` | :arrow_down: |
| [...g/agent/apiserver/handlers/featuregates/handler.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2FwaXNlcnZlci9oYW5kbGVycy9mZWF0dXJlZ2F0ZXMvaGFuZGxlci5nbw==) | `0.00% <0.00%> (-82.36%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/util/parse\_port.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvdXRpbC9wYXJzZV9wb3J0Lmdv) | `0.00% <0.00%> (-71.43%)` | :arrow_down: |
| [pkg/apis/controlplane/v1beta2/helper.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXMvY29udHJvbHBsYW5lL3YxYmV0YTIvaGVscGVyLmdv) | `33.33% <0.00%> (-66.67%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/npl\_agent\_init.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvbnBsX2FnZW50X2luaXQuZ28=) | `0.00% <0.00%> (-66.67%)` | :arrow_down: |
| [pkg/agent/ipassigner/local\_ip\_detector\_linux.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2lwYXNzaWduZXIvbG9jYWxfaXBfZGV0ZWN0b3JfbGludXguZ28=) | `0.00% <0.00%> (-65.22%)` | :arrow_down: |
| [...formers/externalversions/security/v1alpha1/tier.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2xlZ2FjeWNsaWVudC9pbmZvcm1lcnMvZXh0ZXJuYWx2ZXJzaW9ucy9zZWN1cml0eS92MWFscGhhMS90aWVyLmdv) | `0.00% <0.00%> (-64.29%)` | :arrow_down: |
| [...ers/externalversions/core/v1alpha2/clustergroup.go](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2xlZ2FjeWNsaWVudC9pbmZvcm1lcnMvZXh0ZXJuYWx2ZXJzaW9ucy9jb3JlL3YxYWxwaGEyL2NsdXN0ZXJncm91cC5nbw==) | `0.00% <0.00%> (-64.29%)` | :arrow_down: |
| ... and [122 more](https://codecov.io/gh/antrea-io/antrea/pull/3313/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | |

dreamtalen: > I do not see the fix of "flowEndReason==3" in code?
> 
> In commit message:
> 
> > One fix is for ToExternalFlows test, we gave it 2 more seconds for
> 
> test, we gave -> test. We give

Thanks Jianjun, I squashed two commits and addressed the message, I changed the code from 
`if exportTime >= flowStartTime+iperfTimeSec {`
to
`if exportTime >= flowStartTime+iperfTimeSec || flowEndReason == 3 {`

dreamtalen: /test-all
dreamtalen: @antoninbas @jianjuns Hi Antonin and Jianjun, may I have a second round of reviews from you? We added another fix for the throughput test. Since this PR could be beneficial for merging other PRs by reducing the test failure, we would like to merge it sooner, thanks!
dreamtalen: /test-all
dreamtalen: /test-integration

Patch:
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index 5236bb1a4..856808189 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -567,6 +567,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -829,6 +833,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1020,6 +1028,10 @@ spec:
                     port:
                       x-kubernetes-int-or-string: true
                     protocol:
+                      enum:
+                      - TCP
+                      - UDP
+                      - SCTP
                       type: string
                   type: object
                 type: array
@@ -1369,6 +1381,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1606,6 +1622,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index ca40c4ed6..8cc152dd2 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -567,6 +567,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -829,6 +833,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1020,6 +1028,10 @@ spec:
                     port:
                       x-kubernetes-int-or-string: true
                     protocol:
+                      enum:
+                      - TCP
+                      - UDP
+                      - SCTP
                       type: string
                   type: object
                 type: array
@@ -1369,6 +1381,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1606,6 +1622,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index 3284925ca..021e6e739 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -567,6 +567,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -829,6 +833,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1020,6 +1028,10 @@ spec:
                     port:
                       x-kubernetes-int-or-string: true
                     protocol:
+                      enum:
+                      - TCP
+                      - UDP
+                      - SCTP
                       type: string
                   type: object
                 type: array
@@ -1369,6 +1381,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1606,6 +1622,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index dc952b7b1..47302e9df 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -567,6 +567,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -829,6 +833,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1020,6 +1028,10 @@ spec:
                     port:
                       x-kubernetes-int-or-string: true
                     protocol:
+                      enum:
+                      - TCP
+                      - UDP
+                      - SCTP
                       type: string
                   type: object
                 type: array
@@ -1369,6 +1381,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1606,6 +1622,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
diff --git a/build/yamls/antrea-kind.yml b/build/yamls/antrea-kind.yml
index e3ae7cdfa..4d49f2be9 100644
--- a/build/yamls/antrea-kind.yml
+++ b/build/yamls/antrea-kind.yml
@@ -567,6 +567,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -829,6 +833,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1020,6 +1028,10 @@ spec:
                     port:
                       x-kubernetes-int-or-string: true
                     protocol:
+                      enum:
+                      - TCP
+                      - UDP
+                      - SCTP
                       type: string
                   type: object
                 type: array
@@ -1369,6 +1381,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1606,6 +1622,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index 3b181e7cd..bc4c8ef89 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -567,6 +567,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -829,6 +833,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1020,6 +1028,10 @@ spec:
                     port:
                       x-kubernetes-int-or-string: true
                     protocol:
+                      enum:
+                      - TCP
+                      - UDP
+                      - SCTP
                       type: string
                   type: object
                 type: array
@@ -1369,6 +1381,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
@@ -1606,6 +1622,10 @@ spec:
                           port:
                             x-kubernetes-int-or-string: true
                           protocol:
+                            enum:
+                            - TCP
+                            - UDP
+                            - SCTP
                             type: string
                         type: object
                       type: array
diff --git a/build/yamls/base/crds.yml b/build/yamls/base/crds.yml
index 8fbca85dc..eb653d73f 100644
--- a/build/yamls/base/crds.yml
+++ b/build/yamls/base/crds.yml
@@ -910,6 +910,7 @@ spec:
                           properties:
                             protocol:
                               type: string
+                              enum: ['TCP', 'UDP', 'SCTP']
                             port:
                               x-kubernetes-int-or-string: true
                             endPort:
@@ -1080,6 +1081,7 @@ spec:
                           properties:
                             protocol:
                               type: string
+                              enum: ['TCP', 'UDP', 'SCTP']
                             port:
                               x-kubernetes-int-or-string: true
                             endPort:
@@ -1328,6 +1330,7 @@ spec:
                           properties:
                             protocol:
                               type: string
+                              enum: ['TCP', 'UDP', 'SCTP']
                             port:
                               x-kubernetes-int-or-string: true
                             endPort:
@@ -1465,8 +1468,10 @@ spec:
                         items:
                           type: object
                           properties:
+                           # Ensure that Protocol field allows only TCP, UDP and SCTP values
                             protocol:
                               type: string
+                              enum: ['TCP', 'UDP', 'SCTP']
                             port:
                               x-kubernetes-int-or-string: true
                             endPort:
@@ -1628,6 +1633,7 @@ spec:
                     properties:
                       protocol:
                         type: string
+                        enum: ['TCP', 'UDP', 'SCTP']
                       port:
                         x-kubernetes-int-or-string: true
                       name:
diff --git a/cmd/antrea-agent/agent.go b/cmd/antrea-agent/agent.go
index e7fc702e7..f1776f749 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -41,7 +41,6 @@ import (
 	"antrea.io/antrea/pkg/agent/flowexporter"
 	"antrea.io/antrea/pkg/agent/flowexporter/exporter"
 	"antrea.io/antrea/pkg/agent/interfacestore"
-	"antrea.io/antrea/pkg/agent/ipassigner"
 	"antrea.io/antrea/pkg/agent/memberlist"
 	"antrea.io/antrea/pkg/agent/metrics"
 	"antrea.io/antrea/pkg/agent/multicast"
@@ -231,8 +230,10 @@ func run(o *Options) error {
 
 	var groupCounters []proxytypes.GroupCounter
 	groupIDUpdates := make(chan string, 100)
-	v4GroupCounter := proxytypes.NewGroupCounter(false, groupIDUpdates)
-	v6GroupCounter := proxytypes.NewGroupCounter(true, groupIDUpdates)
+	v4GroupIDAllocator := openflow.NewGroupAllocator(false)
+	v4GroupCounter := proxytypes.NewGroupCounter(v4GroupIDAllocator, groupIDUpdates)
+	v6GroupIDAllocator := openflow.NewGroupAllocator(true)
+	v6GroupCounter := proxytypes.NewGroupCounter(v6GroupIDAllocator, groupIDUpdates)
 
 	v4Enabled := networkConfig.IPv4Enabled
 	v6Enabled := networkConfig.IPv6Enabled
@@ -305,13 +306,11 @@ func run(o *Options) error {
 	var externalIPPoolController *externalippool.ExternalIPPoolController
 	var externalIPController *serviceexternalip.ServiceExternalIPController
 	var memberlistCluster *memberlist.Cluster
-	var localIPDetector ipassigner.LocalIPDetector
 
 	if features.DefaultFeatureGate.Enabled(features.Egress) || features.DefaultFeatureGate.Enabled(features.ServiceExternalIP) {
 		externalIPPoolController = externalippool.NewExternalIPPoolController(
 			crdClient, externalIPPoolInformer,
 		)
-		localIPDetector = ipassigner.NewLocalIPDetector()
 		var nodeTransportIP net.IP
 		if nodeConfig.NodeTransportIPv4Addr != nil {
 			nodeTransportIP = nodeConfig.NodeTransportIPv4Addr.IP
@@ -330,7 +329,7 @@ func run(o *Options) error {
 	if features.DefaultFeatureGate.Enabled(features.Egress) {
 		egressController, err = egress.NewEgressController(
 			ofClient, antreaClientProvider, crdClient, ifaceStore, routeClient, nodeConfig.Name, nodeConfig.NodeTransportInterfaceName,
-			memberlistCluster, egressInformer, nodeInformer, localIPDetector,
+			memberlistCluster, egressInformer, nodeInformer,
 		)
 		if err != nil {
 			return fmt.Errorf("error creating new Egress controller: %v", err)
@@ -344,7 +343,6 @@ func run(o *Options) error {
 			memberlistCluster,
 			serviceInformer,
 			endpointsInformer,
-			localIPDetector,
 		)
 		if err != nil {
 			return fmt.Errorf("error creating new ServiceExternalIP controller: %v", err)
@@ -529,7 +527,6 @@ func run(o *Options) error {
 
 	if features.DefaultFeatureGate.Enabled(features.Egress) || features.DefaultFeatureGate.Enabled(features.ServiceExternalIP) {
 		go externalIPPoolController.Run(stopCh)
-		go localIPDetector.Run(stopCh)
 		go memberlistCluster.Run(stopCh)
 	}
 
diff --git a/docs/antrea-network-policy.md b/docs/antrea-network-policy.md
index aded74ef4..af3a6be54 100644
--- a/docs/antrea-network-policy.md
+++ b/docs/antrea-network-policy.md
@@ -509,13 +509,13 @@ deduplication is applied to simplify multiple logs. Duplication buffer length is
 The rules are logged in the following format:
 
 ```text
-    <yyyy/mm/dd> <time> <ovs-table-name> <antrea-native-policy-reference> <action> <openflow-priority> <source-ip> <destination-ip> <packet-length> <protocol>
+    <yyyy/mm/dd> <time> <ovs-table-name> <antrea-native-policy-reference> <action> <openflow-priority> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length>
     Deduplication:
-    <yyyy/mm/dd> <time> <ovs-table-name> <antrea-native-policy-reference> <action> <openflow-priority> <source-ip> <destination-ip> <packet-length> <protocol> [<num of packets> packets in <duplicate duration>]
+    <yyyy/mm/dd> <time> <ovs-table-name> <antrea-native-policy-reference> <action> <openflow-priority> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length> [<num of packets> packets in <duplicate duration>]
 
     Example:
-    2020/11/02 22:21:21.148395 AntreaPolicyAppTierIngressRule AntreaNetworkPolicy:default/test-anp Allow 61800 10.0.0.4 10.0.0.5 60 TCP
-    2021/06/24 23:56:41.346165 AntreaPolicyEgressRule AntreaNetworkPolicy:default/test-anp Drop 44900 10.0.0.5 10.0.0.4 60 TCP [3 packets in 1.011379442s]
+    2020/11/02 22:21:21.148395 AntreaPolicyAppTierIngressRule AntreaNetworkPolicy:default/test-anp Allow 61800 10.10.1.65 35402 10.0.0.5 80 TCP 60
+    2021/06/24 23:56:41.346165 AntreaPolicyEgressRule AntreaNetworkPolicy:default/test-anp Drop 44900 10.10.1.65 35402 10.0.0.5 80 TCP 60 [3 packets in 1.011379442s]
 ```
 
 **`appliedTo` per rule**: A ClusterNetworkPolicy ingress or egress rule may
diff --git a/go.mod b/go.mod
index d0ac58a1b..dbbe609a6 100644
--- a/go.mod
+++ b/go.mod
@@ -27,6 +27,10 @@ require (
 	github.com/k8snetworkplumbingwg/network-attachment-definition-client v1.1.0
 	github.com/k8snetworkplumbingwg/sriov-cni v2.1.0+incompatible
 	github.com/kevinburke/ssh_config v0.0.0-20190725054713-01f96b0aa0cd
+	github.com/mdlayher/arp v0.0.0-20191213142603-f72070a231fc
+	github.com/mdlayher/ethernet v0.0.0-20190606142754-0394541c37b7
+	github.com/mdlayher/ndp v0.0.0-20210831201139-f982b8766fb5
+	github.com/mdlayher/raw v0.0.0-20211126142749-4eae47f3d54b
 	github.com/miekg/dns v1.1.43
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
 	github.com/onsi/ginkgo v1.16.4
@@ -146,6 +150,7 @@ require (
 	github.com/streamrail/concurrent-map v0.0.0-20160823150647-8bf1e9bacbf6 // indirect
 	github.com/ti-mo/netfilter v0.3.1 // indirect
 	github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae // indirect
+	gitlab.com/golang-commonmark/puny v0.0.0-20191124015043-9f83538fa04f // indirect
 	go.etcd.io/etcd v0.5.0-alpha.5.0.20200910180754-dd1b699fc489 // indirect
 	go.opencensus.io v0.22.3 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
diff --git a/go.sum b/go.sum
index dd5b13ae5..930c3d8c0 100644
--- a/go.sum
+++ b/go.sum
@@ -356,6 +356,7 @@ github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5a
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
@@ -518,10 +519,17 @@ github.com/mattn/go-shellwords v1.0.3/go.mod h1:3xCvwCdWdlDJUrvuMn7Wuy9eWs4pE8vq
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
+github.com/mdlayher/arp v0.0.0-20191213142603-f72070a231fc h1:m7rJJJeXrYCFpsxXYapkDW53wJCDmf9bsIXUg0HoeQY=
+github.com/mdlayher/arp v0.0.0-20191213142603-f72070a231fc/go.mod h1:eOj1DDj3NAZ6yv+WafaKzY37MFZ58TdfIhQ+8nQbiis=
+github.com/mdlayher/ethernet v0.0.0-20190313224307-5b5fc417d966/go.mod h1:5s5p/sMJ6sNsFl6uCh85lkFGV8kLuIYJCRJLavVJwvg=
+github.com/mdlayher/ethernet v0.0.0-20190606142754-0394541c37b7 h1:lez6TS6aAau+8wXUP3G9I3TGlmPFEq2CTxBaRqY6AGE=
+github.com/mdlayher/ethernet v0.0.0-20190606142754-0394541c37b7/go.mod h1:U6ZQobyTjI/tJyq2HG+i/dfSoFUt8/aZCM+GKtmFk/Y=
 github.com/mdlayher/ethtool v0.0.0-20210210192532-2b88debcdd43 h1:WgyLFv10Ov49JAQI/ZLUkCZ7VJS3r74hwFIGXJsgZlY=
 github.com/mdlayher/ethtool v0.0.0-20210210192532-2b88debcdd43/go.mod h1:+t7E0lkKfbBsebllff1xdTmyJt8lH37niI6kwFk9OTo=
 github.com/mdlayher/genetlink v1.0.0 h1:OoHN1OdyEIkScEmRgxLEe2M9U8ClMytqA5niynLtfj0=
 github.com/mdlayher/genetlink v1.0.0/go.mod h1:0rJ0h4itni50A86M2kHcgS85ttZazNt7a8H2a2cw0Gc=
+github.com/mdlayher/ndp v0.0.0-20210831201139-f982b8766fb5 h1:w1Lne6x7QKrrkAviNvSd0M3NYblM1n76Pv2FsGkNzzI=
+github.com/mdlayher/ndp v0.0.0-20210831201139-f982b8766fb5/go.mod h1:32w/5dDZWVSEOxyniAgKK4d7dHTuO6TCxWmUznQe3f8=
 github.com/mdlayher/netlink v0.0.0-20190409211403-11939a169225/go.mod h1:eQB3mZE4aiYnlUsyGGCOpPETfdQq4Jhsgf1fk3cwQaA=
 github.com/mdlayher/netlink v1.0.0/go.mod h1:KxeJAFOFLG6AjpyDkQ/iIhxygIUKD+vcwqcnu43w/+M=
 github.com/mdlayher/netlink v1.1.0/go.mod h1:H4WCitaheIsdF9yOYu8CFmCgQthAPIWZmcKp9uZHgmY=
@@ -533,6 +541,10 @@ github.com/mdlayher/netlink v1.2.2-0.20210123213345-5cc92139ae3e/go.mod h1:bacnN
 github.com/mdlayher/netlink v1.3.0/go.mod h1:xK/BssKuwcRXHrtN04UBkwQ6dY9VviGGuriDdoPSWys=
 github.com/mdlayher/netlink v1.4.0 h1:n3ARR+Fm0dDv37dj5wSWZXDKcy+U0zwcXS3zKMnSiT0=
 github.com/mdlayher/netlink v1.4.0/go.mod h1:dRJi5IABcZpBD2A3D0Mv/AiX8I9uDEu5oGkAVrekmf8=
+github.com/mdlayher/raw v0.0.0-20190313224157-43dbcdd7739d/go.mod h1:r1fbeITl2xL/zLbVnNHFyOzQJTgr/3fpf1lJX/cjzR8=
+github.com/mdlayher/raw v0.0.0-20190606142536-fef19f00fc18/go.mod h1:7EpbotpCmVZcu+KCX4g9WaRNuu11uyhiW7+Le1dKawg=
+github.com/mdlayher/raw v0.0.0-20211126142749-4eae47f3d54b h1:MHcTarUMC4sFA7eiyR8IEJ6j2PgmgXR+B9X2IIMjh7A=
+github.com/mdlayher/raw v0.0.0-20211126142749-4eae47f3d54b/go.mod h1:7EpbotpCmVZcu+KCX4g9WaRNuu11uyhiW7+Le1dKawg=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso=
 github.com/miekg/dns v1.1.43 h1:JKfpVSCB84vrAmHzyrsxB5NAr5kLoMXZArPSw7Qlgyg=
@@ -746,6 +758,8 @@ github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+gitlab.com/golang-commonmark/puny v0.0.0-20191124015043-9f83538fa04f h1:Wku8eEdeJqIOFHtrfkYUByc4bCaTeA6fL0UJgfEiFMI=
+gitlab.com/golang-commonmark/puny v0.0.0-20191124015043-9f83538fa04f/go.mod h1:Tiuhl+njh/JIg0uS/sOJVYi0x2HEa5rc1OAaVsb5tAs=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0=
@@ -843,8 +857,10 @@ golang.org/x/net v0.0.0-20181220203305-927f97764cc3/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190313220215-9f648a60d977/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190320064053-1272bf9dcd53/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190419010253-1f3472d942ba/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190501004415-9ce7a6920f09/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
@@ -865,6 +881,7 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200301022130-244492dfa37a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200602114024-627f9648deb9/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200904194848-62affa334b73/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
@@ -893,6 +910,7 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
@@ -913,10 +931,12 @@ golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190321052220-f7bb7a8bee54/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190411185658-b44545bcd369/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190418153312-f0ce4c0180be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190514135907-3a4b5fb9f71f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190606122018-79a91cf218c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190616124812-15dcb6c0061f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -949,6 +969,7 @@ golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200331124033-c3d80250170d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200602100848-8d3cce7afc34/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200728102440-3e129f6d46b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
diff --git a/hack/windows/Helper.psm1 b/hack/windows/Helper.psm1
index 92ba4ec65..09d0b7f3d 100644
--- a/hack/windows/Helper.psm1
+++ b/hack/windows/Helper.psm1
@@ -163,7 +163,7 @@ function New-KubeProxyServiceInterface {
         Write-Host "Network adapter $INTERFACE_TO_ADD_SERVICE_IP exists, exit."
         return
     }
-    if (!(Get-VMSwitch -Name $hnsSwitchName -ErrorAction SilentlyContinue)) {
+    if (!(Get-VMSwitch -ComputerName $(hostname) -Name $hnsSwitchName -ErrorAction SilentlyContinue)) {
         Write-Host "Creating internal switch: $hnsSwitchName for kube-proxy"
         New-VMSwitch -name $hnsSwitchName -SwitchType Internal
     }
@@ -216,7 +216,7 @@ function Start-OVSServices {
     }
     # Try to cleanup ovsdb-server configurations if the antrea-hnsnetwork is not existing. Or ovs-vswitchd service
     # will can not get started.
-    if (!(Get-VMswitch -Name "antrea-hnsnetwork" -SwitchType External -ErrorAction SilentlyContinue)) {
+    if (!(Get-VMswitch -ComputerName $(hostname) -Name "antrea-hnsnetwork" -SwitchType External -ErrorAction SilentlyContinue)) {
         & ovs-vsctl.exe --no-wait --if-exists del-br br-int
         if ($LASTEXITCODE) {
             return $false
diff --git a/hack/windows/Prepare-ServiceInterface.ps1 b/hack/windows/Prepare-ServiceInterface.ps1
index 33159867e..c962c2cb8 100644
--- a/hack/windows/Prepare-ServiceInterface.ps1
+++ b/hack/windows/Prepare-ServiceInterface.ps1
@@ -25,7 +25,7 @@ if (Get-NetAdapter -InterfaceAlias $INTERFACE_TO_ADD_SERVICE_IP -ErrorAction Sil
     return
 }
 [Environment]::SetEnvironmentVariable("INTERFACE_TO_ADD_SERVICE_IP", $INTERFACE_TO_ADD_SERVICE_IP, [System.EnvironmentVariableTarget]::Machine)
-$hnsSwitchName = $(Get-VMSwitch -SwitchType Internal).Name
+$hnsSwitchName = $(Get-VMSwitch -ComputerName $(hostname) -SwitchType Internal).Name
 Add-VMNetworkAdapter -ManagementOS -Name $InterfaceAlias -SwitchName $hnsSwitchName
 Set-NetIPInterface -ifAlias $INTERFACE_TO_ADD_SERVICE_IP -Forwarding Enabled
 
diff --git a/pkg/agent/cniserver/ipam/antrea_ipam_controller.go b/pkg/agent/cniserver/ipam/antrea_ipam_controller.go
index 62bdf887b..d03a625e6 100644
--- a/pkg/agent/cniserver/ipam/antrea_ipam_controller.go
+++ b/pkg/agent/cniserver/ipam/antrea_ipam_controller.go
@@ -148,6 +148,20 @@ func (c *AntreaIPAMController) getIPPoolsByPod(namespace, name string) ([]string
 		}
 		return nil, nil, nil, err
 	}
+
+	annotations, exists := pod.Annotations[AntreaIPAMAnnotationKey]
+	if !exists {
+		// Find IPPool by Namespace
+		ns, err := c.namespaceLister.Get(namespace)
+		if err != nil {
+			return nil, nil, nil, nil
+		}
+		annotations, exists = ns.Annotations[AntreaIPAMAnnotationKey]
+		if !exists {
+			return nil, nil, nil, nil
+		}
+	}
+
 	// Collect specified IPs if exist
 	ipStrings, _ := pod.Annotations[AntreaIPAMPodIPAnnotationKey]
 	ipStrings = strings.ReplaceAll(ipStrings, " ", "")
@@ -187,20 +201,6 @@ ownerReferenceLoop:
 		}
 	}
 
-	annotations, exists := pod.Annotations[AntreaIPAMAnnotationKey]
-	if exists {
-		return strings.Split(annotations, AntreaIPAMAnnotationDelimiter), ips, reservedOwner, ipErr
-	}
-
-	// Find IPPool by Namespace
-	ns, err := c.namespaceLister.Get(namespace)
-	if err != nil {
-		return nil, nil, nil, nil
-	}
-	annotations, exists = ns.Annotations[AntreaIPAMAnnotationKey]
-	if !exists {
-		return nil, nil, nil, nil
-	}
 	return strings.Split(annotations, AntreaIPAMAnnotationDelimiter), ips, reservedOwner, ipErr
 }
 
diff --git a/pkg/agent/controller/egress/egress_controller.go b/pkg/agent/controller/egress/egress_controller.go
index 4addbc95c..510c952ad 100644
--- a/pkg/agent/controller/egress/egress_controller.go
+++ b/pkg/agent/controller/egress/egress_controller.go
@@ -155,7 +155,6 @@ func NewEgressController(
 	cluster *memberlist.Cluster,
 	egressInformer crdinformers.EgressInformer,
 	nodeInformer coreinformers.NodeInformer,
-	localIPDetector ipassigner.LocalIPDetector,
 ) (*EgressController, error) {
 	c := &EgressController{
 		ofClient:             ofClient,
@@ -172,7 +171,7 @@ func NewEgressController(
 		egressStates:         map[string]*egressState{},
 		egressIPStates:       map[string]*egressIPState{},
 		egressBindings:       map[string]*egressBinding{},
-		localIPDetector:      localIPDetector,
+		localIPDetector:      ipassigner.NewLocalIPDetector(),
 		idAllocator:          newIDAllocator(minEgressMark, maxEgressMark),
 		cluster:              cluster,
 	}
@@ -292,6 +291,7 @@ func (c *EgressController) Run(stopCh <-chan struct{}) {
 
 	go c.localIPDetector.Run(stopCh)
 
+	go c.ipAssigner.Run(stopCh)
 	if !cache.WaitForNamedCacheSync(controllerName, stopCh, c.egressListerSynced, c.localIPDetector.HasSynced) {
 		return
 	}
diff --git a/pkg/agent/controller/networkpolicy/audit_logging.go b/pkg/agent/controller/networkpolicy/audit_logging.go
index 1b15656ba..97265d12b 100644
--- a/pkg/agent/controller/networkpolicy/audit_logging.go
+++ b/pkg/agent/controller/networkpolicy/audit_logging.go
@@ -19,13 +19,17 @@ import (
 	"log"
 	"os"
 	"path/filepath"
+	"strconv"
 	"sync"
 	"time"
 
+	"antrea.io/ofnet/ofctrl"
 	"gopkg.in/natefinch/lumberjack.v2"
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/openflow"
+	binding "antrea.io/antrea/pkg/ovs/openflow"
+	"antrea.io/antrea/pkg/util/ip"
 	"antrea.io/antrea/pkg/util/logdir"
 )
 
@@ -65,7 +69,9 @@ type logInfo struct {
 	disposition string // Allow/Drop of the rule sending packetin
 	ofPriority  string // openflow priority of the flow sending packetin
 	srcIP       string // source IP of the traffic logged
+	srcPort     string // source port of the traffic logged
 	destIP      string // destination IP of the traffic logged
+	destPort    string // destination port of the traffic logged
 	pktLength   uint16 // packet length of packetin
 	protocolStr string // protocol of the traffic logged
 }
@@ -127,7 +133,7 @@ func (l *AntreaPolicyLogger) updateLogKey(logMsg string, bufferLength time.Durat
 // LogDedupPacket logs information in ob based on disposition and duplication conditions.
 func (l *AntreaPolicyLogger) LogDedupPacket(ob *logInfo) {
 	// Deduplicate non-Allow packet log.
-	logMsg := fmt.Sprintf("%s %s %s %s %s %s %d %s", ob.tableName, ob.npRef, ob.disposition, ob.ofPriority, ob.srcIP, ob.destIP, ob.pktLength, ob.protocolStr)
+	logMsg := fmt.Sprintf("%s %s %s %s %s %s %s %s %s %d", ob.tableName, ob.npRef, ob.disposition, ob.ofPriority, ob.srcIP, ob.srcPort, ob.destIP, ob.destPort, ob.protocolStr, ob.pktLength)
 	if ob.disposition == openflow.DispositionToString[openflow.DispositionAllow] {
 		l.anpLogger.Printf(logMsg)
 	} else {
@@ -170,3 +176,64 @@ func newAntreaPolicyLogger() (*AntreaPolicyLogger, error) {
 	klog.InfoS("Initialized Antrea-native Policy Logger for audit logging", "logFile", logFile)
 	return antreaPolicyLogger, nil
 }
+
+// getNetworkPolicyInfo fills in tableName, npName, ofPriority, disposition of logInfo ob.
+func getNetworkPolicyInfo(pktIn *ofctrl.PacketIn, c *Controller, ob *logInfo) error {
+	matchers := pktIn.GetMatches()
+	var match *ofctrl.MatchField
+	// Get table name.
+	tableID := pktIn.TableId
+	ob.tableName = openflow.GetFlowTableName(tableID)
+
+	// Get disposition Allow or Drop.
+	match = getMatchRegField(matchers, openflow.APDispositionField)
+	info, err := getInfoInReg(match, openflow.APDispositionField.GetRange().ToNXRange())
+	if err != nil {
+		return fmt.Errorf("received error while unloading disposition from reg: %v", err)
+	}
+	ob.disposition = openflow.DispositionToString[info]
+
+	// Set match to corresponding ingress/egress reg according to disposition.
+	match = getMatch(matchers, tableID, info)
+
+	// Get Network Policy full name and OF priority of the conjunction.
+	info, err = getInfoInReg(match, nil)
+	if err != nil {
+		return fmt.Errorf("received error while unloading conjunction id from reg: %v", err)
+	}
+	ob.npRef, ob.ofPriority = c.ofClient.GetPolicyInfoFromConjunction(info)
+
+	return nil
+}
+
+// logPacket retrieves information from openflow reg, controller cache, packet-in
+// packet to log. Log is deduplicated for non-Allow packets from record in logDeduplication.
+// Deduplication is safe guarded by logRecordDedupMap mutex.
+func (c *Controller) logPacket(pktIn *ofctrl.PacketIn) error {
+	ob := new(logInfo)
+	packet, err := binding.ParsePacketIn(pktIn)
+	if err != nil {
+		return fmt.Errorf("received error while parsing packetin: %v", err)
+	}
+
+	// Set Network Policy and packet info to log.
+	err = getNetworkPolicyInfo(pktIn, c, ob)
+	if err != nil {
+		return fmt.Errorf("received error while retrieving NetworkPolicy info: %v", err)
+	}
+	ob.srcIP = packet.SourceIP.String()
+	ob.destIP = packet.DestinationIP.String()
+	ob.pktLength = packet.IPLength
+	ob.protocolStr = ip.IPProtocolNumberToString(packet.IPProto, "UnknownProtocol")
+	if ob.protocolStr == "TCP" || ob.protocolStr == "UDP" {
+		ob.srcPort = strconv.Itoa(int(packet.SourcePort))
+		ob.destPort = strconv.Itoa(int(packet.DestinationPort))
+	} else {
+		// Placeholders for ICMP packets without port numbers.
+		ob.srcPort, ob.destPort = "<nil>", "<nil>"
+	}
+
+	// Log the ob info to corresponding file w/ deduplication.
+	c.antreaPolicyLogger.LogDedupPacket(ob)
+	return nil
+}
diff --git a/pkg/agent/controller/networkpolicy/packetin.go b/pkg/agent/controller/networkpolicy/packetin.go
index 140529f82..a60d23522 100644
--- a/pkg/agent/controller/networkpolicy/packetin.go
+++ b/pkg/agent/controller/networkpolicy/packetin.go
@@ -20,7 +20,6 @@ import (
 	"time"
 
 	"antrea.io/libOpenflow/openflow13"
-	"antrea.io/libOpenflow/protocol"
 	"antrea.io/ofnet/ofctrl"
 	"github.com/vmware/go-ipfix/pkg/registry"
 	"k8s.io/klog/v2"
@@ -28,7 +27,6 @@ import (
 	"antrea.io/antrea/pkg/agent/flowexporter"
 	"antrea.io/antrea/pkg/agent/openflow"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
-	"antrea.io/antrea/pkg/util/ip"
 )
 
 // HandlePacketIn is the packetin handler registered to openflow by Antrea network
@@ -107,80 +105,6 @@ func getInfoInReg(regMatch *ofctrl.MatchField, rng *openflow13.NXRange) (uint32,
 	return regValue.Data, nil
 }
 
-// getNetworkPolicyInfo fills in tableName, npName, ofPriority, disposition of logInfo ob.
-func getNetworkPolicyInfo(pktIn *ofctrl.PacketIn, c *Controller, ob *logInfo) error {
-	matchers := pktIn.GetMatches()
-	var match *ofctrl.MatchField
-	// Get table name
-	tableID := pktIn.TableId
-	ob.tableName = openflow.GetFlowTableName(tableID)
-
-	// Get disposition Allow or Drop
-	match = getMatchRegField(matchers, openflow.APDispositionField)
-	info, err := getInfoInReg(match, openflow.APDispositionField.GetRange().ToNXRange())
-	if err != nil {
-		return fmt.Errorf("received error while unloading disposition from reg: %v", err)
-	}
-	ob.disposition = openflow.DispositionToString[info]
-
-	// Set match to corresponding ingress/egress reg according to disposition
-	match = getMatch(matchers, tableID, info)
-
-	// Get Network Policy full name and OF priority of the conjunction
-	info, err = getInfoInReg(match, nil)
-	if err != nil {
-		return fmt.Errorf("received error while unloading conjunction id from reg: %v", err)
-	}
-	ob.npRef, ob.ofPriority = c.ofClient.GetPolicyInfoFromConjunction(info)
-
-	return nil
-}
-
-// getPacketInfo fills in srcIP, destIP, pktLength, protocol of logInfo ob.
-func getPacketInfo(pktIn *ofctrl.PacketIn, ob *logInfo) error {
-	var prot uint8
-	switch ipPkt := pktIn.Data.Data.(type) {
-	case *protocol.IPv4:
-		ob.srcIP = ipPkt.NWSrc.String()
-		ob.destIP = ipPkt.NWDst.String()
-		ob.pktLength = ipPkt.Length
-		prot = ipPkt.Protocol
-	case *protocol.IPv6:
-		ob.srcIP = ipPkt.NWSrc.String()
-		ob.destIP = ipPkt.NWDst.String()
-		ob.pktLength = ipPkt.Length
-		prot = ipPkt.NextHeader
-	default:
-		return errors.New("unsupported packet-in: should be a valid IPv4 or IPv6 packet")
-	}
-
-	ob.protocolStr = ip.IPProtocolNumberToString(prot, "UnknownProtocol")
-
-	return nil
-}
-
-// logPacket retrieves information from openflow reg, controller cache, packet-in
-// packet to log. Log is deduplicated for non-Allow packets from record in logDeduplication.
-// Deduplication is safe guarded by logRecordDedupMap mutex.
-func (c *Controller) logPacket(pktIn *ofctrl.PacketIn) error {
-	ob := new(logInfo)
-
-	// Get Network Policy log info
-	err := getNetworkPolicyInfo(pktIn, c, ob)
-	if err != nil {
-		return fmt.Errorf("received error while retrieving NetworkPolicy info: %v", err)
-	}
-	// Get packet log info
-	err = getPacketInfo(pktIn, ob)
-	if err != nil {
-		return fmt.Errorf("received error while retrieving NetworkPolicy info: %v", err)
-	}
-
-	// Log the ob info to corresponding file w/ deduplication
-	c.antreaPolicyLogger.LogDedupPacket(ob)
-	return nil
-}
-
 func (c *Controller) storeDenyConnection(pktIn *ofctrl.PacketIn) error {
 	packet, err := binding.ParsePacketIn(pktIn)
 	if err != nil {
diff --git a/pkg/agent/controller/serviceexternalip/controller.go b/pkg/agent/controller/serviceexternalip/controller.go
index f8ac8c38d..e458b5e98 100644
--- a/pkg/agent/controller/serviceexternalip/controller.go
+++ b/pkg/agent/controller/serviceexternalip/controller.go
@@ -23,7 +23,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	apimachinerytypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -51,9 +50,6 @@ const (
 
 	externalIPIndex     = "externalIP"
 	externalIPPoolIndex = "externalIPPool"
-
-	// ingressDummyDevice is the dummy device that holds the Service external IPs configured to the system by antrea-agent.
-	ingressDummyDevice = "antrea-ingress0"
 )
 
 type externalIPState struct {
@@ -78,9 +74,11 @@ type ServiceExternalIPController struct {
 	externalIPStates      map[apimachinerytypes.NamespacedName]externalIPState
 	externalIPStatesMutex sync.RWMutex
 
-	cluster         memberlist.Interface
-	ipAssigner      ipassigner.IPAssigner
-	localIPDetector ipassigner.LocalIPDetector
+	cluster    memberlist.Interface
+	ipAssigner ipassigner.IPAssigner
+
+	assignedIPs      map[string]sets.String
+	assignedIPsMutex sync.Mutex
 }
 
 func NewServiceExternalIPController(
@@ -90,7 +88,6 @@ func NewServiceExternalIPController(
 	cluster memberlist.Interface,
 	serviceInformer coreinformers.ServiceInformer,
 	endpointsInformer coreinformers.EndpointsInformer,
-	localIPDetector ipassigner.LocalIPDetector,
 ) (*ServiceExternalIPController, error) {
 	c := &ServiceExternalIPController{
 		nodeName:              nodeName,
@@ -104,9 +101,9 @@ func NewServiceExternalIPController(
 		endpointsLister:       endpointsInformer.Lister(),
 		endpointsListerSynced: endpointsInformer.Informer().HasSynced,
 		externalIPStates:      make(map[apimachinerytypes.NamespacedName]externalIPState),
-		localIPDetector:       localIPDetector,
+		assignedIPs:           make(map[string]sets.String),
 	}
-	ipAssigner, err := ipassigner.NewIPAssigner(nodeTransportInterface, ingressDummyDevice)
+	ipAssigner, err := ipassigner.NewIPAssigner(nodeTransportInterface, "")
 	if err != nil {
 		return nil, fmt.Errorf("initializing service external IP assigner failed: %v", err)
 	}
@@ -158,7 +155,6 @@ func NewServiceExternalIPController(
 		resyncPeriod,
 	)
 
-	c.localIPDetector.AddEventHandler(c.onLocalIPUpdate)
 	c.cluster.AddClusterEventHandler(c.enqueueServicesByExternalIPPool)
 	return c, nil
 }
@@ -236,6 +232,7 @@ func (c *ServiceExternalIPController) enqueueServicesByExternalIPPool(eipName st
 // workqueue.
 func (c *ServiceExternalIPController) Run(stopCh <-chan struct{}) {
 	defer c.queue.ShutDown()
+	go c.ipAssigner.Run(stopCh)
 
 	klog.Infof("Starting %s", controllerName)
 	defer klog.Infof("Shutting down %s", controllerName)
@@ -244,50 +241,12 @@ func (c *ServiceExternalIPController) Run(stopCh <-chan struct{}) {
 		return
 	}
 
-	c.removeStaleExternalIPs()
-
 	for i := 0; i < defaultWorkers; i++ {
 		go wait.Until(c.worker, time.Second, stopCh)
 	}
 	<-stopCh
 }
 
-// removeStaleExternalIPs unassigns stale external IPs that shouldn't be present on this Node.
-// This function will only delete IPs which are caused by Service changes when the agent on this Node was
-// not running. Those IPs should be deleted caused by migration will be deleted by processNextWorkItem.
-func (c *ServiceExternalIPController) removeStaleExternalIPs() {
-	desiredExternalIPs := sets.NewString()
-	services, _ := c.serviceLister.List(labels.Everything())
-	for _, service := range services {
-		if service.Spec.Type == corev1.ServiceTypeLoadBalancer &&
-			service.ObjectMeta.Annotations[types.ServiceExternalIPPoolAnnotationKey] != "" &&
-			len(service.Status.LoadBalancer.Ingress) != 0 {
-			desiredExternalIPs.Insert(service.Status.LoadBalancer.Ingress[0].IP)
-		}
-	}
-	actualExternalIPs := c.ipAssigner.AssignedIPs()
-	for ip := range actualExternalIPs.Difference(desiredExternalIPs) {
-		if err := c.ipAssigner.UnassignIP(ip); err != nil {
-			klog.ErrorS(err, "Failed to clean up stale service external IP", "ip", ip)
-		}
-	}
-}
-
-func (c *ServiceExternalIPController) onLocalIPUpdate(ip string, added bool) {
-	services, _ := c.serviceInformer.GetIndexer().ByIndex(externalIPIndex, ip)
-	if len(services) == 0 {
-		return
-	}
-	if added {
-		klog.Infof("Detected service external IP address %s added to this Node", ip)
-	} else {
-		klog.Infof("Detected service external IP address %s deleted from this Node", ip)
-	}
-	for _, s := range services {
-		c.enqueueService(s)
-	}
-}
-
 // worker is a long-running function that will continually call the processNextWorkItem function in
 // order to read and process a message on the workqueue.
 func (c *ServiceExternalIPController) worker() {
@@ -325,7 +284,7 @@ func (c *ServiceExternalIPController) deleteService(service apimachinerytypes.Na
 	if state, exist = c.externalIPStates[service]; !exist {
 		return nil
 	}
-	if err := c.ipAssigner.UnassignIP(state.ip); err != nil {
+	if err := c.unassignIP(state.ip, service); err != nil {
 		return err
 	}
 	delete(c.externalIPStates, service)
@@ -440,9 +399,41 @@ func (c *ServiceExternalIPController) syncService(key apimachinerytypes.Namespac
 				return err
 			}
 		}
-		return c.ipAssigner.AssignIP(currentExternalIP)
+		return c.assignIP(currentExternalIP, key)
 	}
-	return c.ipAssigner.UnassignIP(currentExternalIP)
+	return c.unassignIP(currentExternalIP, key)
+}
+
+func (c *ServiceExternalIPController) assignIP(ip string, service apimachinerytypes.NamespacedName) error {
+	c.assignedIPsMutex.Lock()
+	defer c.assignedIPsMutex.Unlock()
+	if _, ok := c.assignedIPs[ip]; !ok {
+		if err := c.ipAssigner.AssignIP(ip); err != nil {
+			return err
+		}
+		c.assignedIPs[ip] = sets.NewString(service.String())
+	} else {
+		c.assignedIPs[ip].Insert(service.String())
+	}
+	return nil
+}
+
+func (c *ServiceExternalIPController) unassignIP(ip string, service apimachinerytypes.NamespacedName) error {
+	c.assignedIPsMutex.Lock()
+	defer c.assignedIPsMutex.Unlock()
+	assigned, ok := c.assignedIPs[ip]
+	if !ok {
+		return nil
+	}
+	if assigned.Len() == 1 && assigned.Has(service.String()) {
+		if err := c.ipAssigner.UnassignIP(ip); err != nil {
+			return err
+		}
+		delete(c.assignedIPs, ip)
+		return nil
+	}
+	assigned.Delete(service.String())
+	return nil
 }
 
 // nodesHasHealthyServiceEndpoint returns the set of Nodes which has at least one healthy endpoint.
diff --git a/pkg/agent/ipassigner/ip_assigner.go b/pkg/agent/ipassigner/ip_assigner.go
index 9428a0f18..ac1e6862f 100644
--- a/pkg/agent/ipassigner/ip_assigner.go
+++ b/pkg/agent/ipassigner/ip_assigner.go
@@ -24,4 +24,6 @@ type IPAssigner interface {
 	UnassignIP(ip string) error
 	// AssignedIPs return the IPs that are assigned to the system by this IPAssigner.
 	AssignedIPs() sets.String
+	// Run starts the IP assigner.
+	Run(<-chan struct{})
 }
diff --git a/pkg/agent/ipassigner/ip_assigner_linux.go b/pkg/agent/ipassigner/ip_assigner_linux.go
index 04ed9d1af..80e976c8e 100644
--- a/pkg/agent/ipassigner/ip_assigner_linux.go
+++ b/pkg/agent/ipassigner/ip_assigner_linux.go
@@ -15,17 +15,19 @@
 package ipassigner
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"sync"
 
 	"github.com/vishvananda/netlink"
+	"golang.org/x/sys/unix"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
+	utilnet "k8s.io/utils/net"
 
+	"antrea.io/antrea/pkg/agent/ipassigner/responder"
 	"antrea.io/antrea/pkg/agent/util"
-	"antrea.io/antrea/pkg/agent/util/arping"
-	"antrea.io/antrea/pkg/agent/util/ndp"
 )
 
 // ipAssigner creates a dummy device and assigns IPs to it.
@@ -41,29 +43,45 @@ type ipAssigner struct {
 	// assignIPs caches the IPs that are assigned to the dummy device.
 	// TODO: Add a goroutine to ensure that the cache is in sync with the IPs assigned to the dummy device in case the
 	// IPs are removed by users accidentally.
-	assignedIPs sets.String
-	mutex       sync.RWMutex
+	assignedIPs  sets.String
+	mutex        sync.RWMutex
+	arpResponder responder.Responder
+	ndpResponder responder.Responder
 }
 
 // NewIPAssigner returns an *ipAssigner.
 func NewIPAssigner(nodeTransportInterface string, dummyDeviceName string) (*ipAssigner, error) {
-	_, _, externalInterface, err := util.GetIPNetDeviceByName(nodeTransportInterface)
+	ipv4, ipv6, externalInterface, err := util.GetIPNetDeviceByName(nodeTransportInterface)
 	if err != nil {
 		return nil, fmt.Errorf("get IPNetDevice from name %s error: %+v", nodeTransportInterface, err)
 	}
-
-	dummyDevice, err := ensureDummyDevice(dummyDeviceName)
-	if err != nil {
-		return nil, fmt.Errorf("error when ensuring dummy device exist: %v", err)
-	}
-
 	a := &ipAssigner{
 		externalInterface: externalInterface,
-		dummyDevice:       dummyDevice,
 		assignedIPs:       sets.NewString(),
 	}
-	if err := a.loadIPAddresses(); err != nil {
-		return nil, fmt.Errorf("error when loading IP addresses from the system: %v", err)
+	if ipv4 != nil {
+		arpResonder, err := responder.NewARPResponder(externalInterface)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create ARP responder for link %s: %v", externalInterface.Name, err)
+		}
+		a.arpResponder = arpResonder
+	}
+	if ipv6 != nil {
+		ndpResponder, err := responder.NewNDPResponder(externalInterface)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create NDP responder for link %s: %v", externalInterface.Name, err)
+		}
+		a.ndpResponder = ndpResponder
+	}
+	if dummyDeviceName != "" {
+		dummyDevice, err := ensureDummyDevice(dummyDeviceName)
+		if err != nil {
+			return nil, fmt.Errorf("error when ensuring dummy device exists: %v", err)
+		}
+		a.dummyDevice = dummyDevice
+		if err := a.loadIPAddresses(); err != nil {
+			return nil, fmt.Errorf("error when loading IP addresses from the system: %v", err)
+		}
 	}
 	return a, nil
 }
@@ -99,45 +117,45 @@ func (a *ipAssigner) loadIPAddresses() error {
 	return nil
 }
 
-// AssignIP ensures the provided IP is assigned to the dummy device.
+// AssignIP ensures the provided IP is assigned to the dummy device and the ARP/NDP responders.
 func (a *ipAssigner) AssignIP(ip string) error {
 	parsedIP := net.ParseIP(ip)
 	if parsedIP == nil {
 		return fmt.Errorf("invalid IP %s", ip)
 	}
-	addr := util.NewIPNet(parsedIP)
-
-	if err := func() error {
-		a.mutex.Lock()
-		defer a.mutex.Unlock()
+	a.mutex.Lock()
+	defer a.mutex.Unlock()
 
-		if a.assignedIPs.Has(ip) {
-			klog.V(2).InfoS("The IP is already assigned", "ip", ip)
-			return nil
-		}
+	if a.assignedIPs.Has(ip) {
+		klog.V(2).InfoS("The IP is already assigned", "ip", ip)
+		return nil
+	}
 
+	if a.dummyDevice != nil {
+		addr := util.NewIPNet(parsedIP)
 		if err := netlink.AddrAdd(a.dummyDevice, &netlink.Addr{IPNet: addr}); err != nil {
-			return fmt.Errorf("failed to add IP %v to interface %s: %v", ip, a.dummyDevice.Attrs().Name, err)
+			if !errors.Is(err, unix.EEXIST) {
+				return fmt.Errorf("failed to add IP %v to interface %s: %v", ip, a.dummyDevice.Attrs().Name, err)
+			} else {
+				klog.InfoS("IP was already assigned to interface", "ip", parsedIP, "interface", a.dummyDevice.Attrs().Name)
+			}
+		} else {
+			klog.InfoS("Assigned IP to interface", "ip", parsedIP, "interface", a.dummyDevice.Attrs().Name)
 		}
-		klog.InfoS("Assigned IP to interface", "ip", parsedIP, "interface", a.dummyDevice.Attrs().Name)
-
-		a.assignedIPs.Insert(ip)
-		return nil
-	}(); err != nil {
-		return err
 	}
 
-	if addr.IP.To4() != nil {
-		if err := arping.GratuitousARPOverIface(addr.IP.To4(), a.externalInterface); err != nil {
-			return fmt.Errorf("failed to send gratuitous ARP: %v", err)
+	if utilnet.IsIPv4(parsedIP) && a.arpResponder != nil {
+		if err := a.arpResponder.AddIP(parsedIP); err != nil {
+			return fmt.Errorf("failed to assign IP %v to ARP responder: %v", ip, err)
 		}
-		klog.V(2).InfoS("Sent gratuitous ARP", "ip", parsedIP)
-	} else {
-		if err := ndp.NeighborAdvertisement(addr.IP.To16(), a.externalInterface); err != nil {
-			return fmt.Errorf("failed to send neighbor advertisement: %v", err)
+	}
+	if utilnet.IsIPv6(parsedIP) && a.ndpResponder != nil {
+		if err := a.ndpResponder.AddIP(parsedIP); err != nil {
+			return fmt.Errorf("failed to assign IP %v to NDP responder: %v", ip, err)
 		}
-		klog.V(2).InfoS("Sent NDP neighbor advertisement", "ip", parsedIP)
 	}
+
+	a.assignedIPs.Insert(ip)
 	return nil
 }
 
@@ -147,7 +165,6 @@ func (a *ipAssigner) UnassignIP(ip string) error {
 	if parsedIP == nil {
 		return fmt.Errorf("invalid IP %s", ip)
 	}
-
 	a.mutex.Lock()
 	defer a.mutex.Unlock()
 
@@ -156,11 +173,28 @@ func (a *ipAssigner) UnassignIP(ip string) error {
 		return nil
 	}
 
-	addr := util.NewIPNet(parsedIP)
-	if err := netlink.AddrDel(a.dummyDevice, &netlink.Addr{IPNet: addr}); err != nil {
-		return fmt.Errorf("failed to delete IP %v from interface %s: %v", ip, a.dummyDevice.Attrs().Name, err)
+	if a.dummyDevice != nil {
+		addr := util.NewIPNet(parsedIP)
+		if err := netlink.AddrDel(a.dummyDevice, &netlink.Addr{IPNet: addr}); err != nil {
+			if !errors.Is(err, unix.EADDRNOTAVAIL) {
+				return fmt.Errorf("failed to delete IP %v from interface %s: %v", ip, a.dummyDevice.Attrs().Name, err)
+			} else {
+				klog.InfoS("IP does not exist on interface", "ip", parsedIP, "interface", a.dummyDevice.Attrs().Name)
+			}
+		}
+		klog.InfoS("Deleted IP from interface", "ip", ip, "interface", a.dummyDevice.Attrs().Name)
+	}
+
+	if utilnet.IsIPv4(parsedIP) && a.arpResponder != nil {
+		if err := a.arpResponder.RemoveIP(parsedIP); err != nil {
+			return fmt.Errorf("failed to remove IP %v from ARP responder: %v", ip, err)
+		}
+	}
+	if utilnet.IsIPv6(parsedIP) && a.ndpResponder != nil {
+		if err := a.ndpResponder.RemoveIP(parsedIP); err != nil {
+			return fmt.Errorf("failed to remove IP %v from NDP responder: %v", ip, err)
+		}
 	}
-	klog.InfoS("Deleted IP from interface", "ip", ip, "interface", a.dummyDevice.Attrs().Name)
 
 	a.assignedIPs.Delete(ip)
 	return nil
@@ -173,3 +207,18 @@ func (a *ipAssigner) AssignedIPs() sets.String {
 	// Return a copy.
 	return a.assignedIPs.Union(nil)
 }
+
+// Run starts the ARP responder and NDP responder.
+func (a *ipAssigner) Run(ch <-chan struct{}) {
+	// Start the ARP responder only when the dummy device is not created. The kernel will handle ARP requests
+	// for IPs assigned to the dummy devices by default.
+	// TODO: Check the arp_ignore sysctl parameter of the transport interface to determine whether to start
+	// the ARP responder or not.
+	if a.dummyDevice == nil && a.arpResponder != nil {
+		go a.arpResponder.Run(ch)
+	}
+	if a.ndpResponder != nil {
+		go a.ndpResponder.Run(ch)
+	}
+	<-ch
+}
diff --git a/pkg/agent/ipassigner/ip_assigner_windows.go b/pkg/agent/ipassigner/ip_assigner_windows.go
index 00caa6917..bba7e0cc5 100644
--- a/pkg/agent/ipassigner/ip_assigner_windows.go
+++ b/pkg/agent/ipassigner/ip_assigner_windows.go
@@ -36,3 +36,6 @@ func (a *ipAssigner) UnassignIP(ip string) error {
 func (a *ipAssigner) AssignedIPs() sets.String {
 	return nil
 }
+
+func (a *ipAssigner) Run(ch <-chan struct{}) {
+}
diff --git a/pkg/agent/ipassigner/responder/arp_responder.go b/pkg/agent/ipassigner/responder/arp_responder.go
new file mode 100644
index 000000000..4aacd1a43
--- /dev/null
+++ b/pkg/agent/ipassigner/responder/arp_responder.go
@@ -0,0 +1,145 @@
+// Copyright 2022 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package responder
+
+import (
+	"fmt"
+	"net"
+	"sync"
+
+	"github.com/mdlayher/arp"
+	"github.com/mdlayher/ethernet"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog/v2"
+	utilnet "k8s.io/utils/net"
+)
+
+type arpResponder struct {
+	iface       *net.Interface
+	conn        *arp.Client
+	assignedIPs sets.String
+	mutex       sync.Mutex
+}
+
+var _ Responder = (*arpResponder)(nil)
+
+func NewARPResponder(iface *net.Interface) (*arpResponder, error) {
+	conn, err := arp.Dial(iface)
+	if err != nil {
+		return nil, fmt.Errorf("creating ARP responder for %q: %s", iface.Name, err)
+	}
+	return &arpResponder{
+		iface:       iface,
+		conn:        conn,
+		assignedIPs: sets.NewString(),
+	}, nil
+}
+
+// advertise sends an gratuitous ARP packet for the IP.
+func (r *arpResponder) advertise(ip net.IP) error {
+	pkt, err := arp.NewPacket(arp.OperationRequest, r.iface.HardwareAddr, ip, ethernet.Broadcast, ip)
+	if err != nil {
+		return err
+	}
+	return r.conn.WriteTo(pkt, ethernet.Broadcast)
+}
+
+func (r *arpResponder) InterfaceName() string {
+	return r.iface.Name
+}
+
+func (r *arpResponder) AddIP(ip net.IP) error {
+	if !utilnet.IsIPv4(ip) {
+		return fmt.Errorf("only IPv4 is supported")
+	}
+	if r.addIP(ip) {
+		klog.InfoS("Assigned IP to ARP responder", "ip", ip, "interface", r.iface.Name)
+		err := r.advertise(ip)
+		if err != nil {
+			klog.ErrorS(err, "Failed to advertise", "ip", ip, "interface", r.iface.Name)
+		}
+	}
+	return nil
+}
+
+func (r *arpResponder) RemoveIP(ip net.IP) error {
+	if !utilnet.IsIPv4(ip) {
+		return fmt.Errorf("only IPv4 is supported")
+	}
+	if r.deleteIP(ip) {
+		klog.InfoS("Removed IP from ARP responder", "ip", ip, "interface", r.iface.Name)
+	}
+	return nil
+}
+
+func (r *arpResponder) handleARPRequest() error {
+	pkt, _, err := r.conn.Read()
+	if err != nil {
+		return err
+	}
+	if pkt.Operation != arp.OperationRequest {
+		return nil
+	}
+	if !r.isIPAssigned(pkt.TargetIP) {
+		klog.V(4).InfoS("Ignored ARP request", "ip", pkt.TargetIP, "interface", r.iface.Name)
+		return nil
+	}
+	if err := r.conn.Reply(pkt, r.iface.HardwareAddr, pkt.TargetIP); err != nil {
+		return fmt.Errorf("failed to reply ARP packet for IP %s: %v", pkt.TargetIP, err)
+	}
+	klog.V(4).InfoS("Sent ARP response", "ip", pkt.TargetIP, "interface", r.iface.Name)
+	return nil
+}
+
+func (r *arpResponder) Run(stopCh <-chan struct{}) {
+	for {
+		select {
+		case <-stopCh:
+			r.conn.Close()
+			return
+		default:
+			err := r.handleARPRequest()
+			if err != nil {
+				klog.ErrorS(err, "Failed to handle ARP request", "deviceName", r.iface.Name)
+			}
+		}
+	}
+}
+
+func (r *arpResponder) isIPAssigned(ip net.IP) bool {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	return r.assignedIPs.Has(ip.String())
+}
+
+func (r *arpResponder) deleteIP(ip net.IP) bool {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	exist := r.assignedIPs.Has(ip.String())
+	if exist {
+		r.assignedIPs.Delete(ip.String())
+	}
+	return exist
+}
+
+func (r *arpResponder) addIP(ip net.IP) bool {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	exist := r.assignedIPs.Has(ip.String())
+	if !exist {
+		r.assignedIPs.Insert(ip.String())
+	}
+	return !exist
+}
diff --git a/pkg/agent/ipassigner/responder/interface.go b/pkg/agent/ipassigner/responder/interface.go
new file mode 100644
index 000000000..4e8d7917d
--- /dev/null
+++ b/pkg/agent/ipassigner/responder/interface.go
@@ -0,0 +1,29 @@
+// Copyright 2022 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package responder
+
+import "net"
+
+// Responder is an interface to handle ARP (IPv4)/NS (IPv6) queries using raw sockets.
+type Responder interface {
+	// InterfaceName returns the name of the network interface which the raw sockets binds on.
+	InterfaceName() string
+	// AddIP assigns the IP to the responder.
+	AddIP(net.IP) error
+	// RemoveIP removes the IP from the responder.
+	RemoveIP(net.IP) error
+	// Run starts the responder.
+	Run(<-chan struct{})
+}
diff --git a/pkg/agent/ipassigner/responder/ndp_responder.go b/pkg/agent/ipassigner/responder/ndp_responder.go
new file mode 100644
index 000000000..35acc6dca
--- /dev/null
+++ b/pkg/agent/ipassigner/responder/ndp_responder.go
@@ -0,0 +1,210 @@
+// Copyright 2022 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package responder
+
+import (
+	"fmt"
+	"net"
+	"sync"
+
+	"github.com/mdlayher/ndp"
+	"golang.org/x/net/ipv6"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog/v2"
+	utilnet "k8s.io/utils/net"
+)
+
+const (
+	solicitedNodeMulticastAddressPrefix = "ff02::1:ff00:0"
+)
+
+type ndpConn interface {
+	WriteTo(message ndp.Message, cm *ipv6.ControlMessage, dstIP net.IP) error
+	ReadFrom() (ndp.Message, *ipv6.ControlMessage, net.IP, error)
+	JoinGroup(net.IP) error
+	LeaveGroup(net.IP) error
+	Close() error
+}
+
+type ndpResponder struct {
+	iface           *net.Interface
+	conn            ndpConn
+	assignedIPs     sets.String
+	multicastGroups map[int]int
+	mutex           sync.Mutex
+}
+
+var _ Responder = (*ndpResponder)(nil)
+
+func parseIPv6SolicitedNodeMulticastAddress(ip net.IP) (net.IP, int) {
+	group := net.ParseIP(solicitedNodeMulticastAddressPrefix)
+	// copy lower 24 bits
+	copy(group[13:], ip[13:])
+	key := int(group[13])<<16 | int(group[14])<<8 | int(group[15])
+	return group, key
+}
+
+func NewNDPResponder(iface *net.Interface) (*ndpResponder, error) {
+	conn, _, err := ndp.Listen(iface, ndp.LinkLocal)
+	if err != nil {
+		return nil, err
+	}
+	return &ndpResponder{
+		iface:           iface,
+		conn:            conn,
+		multicastGroups: make(map[int]int),
+		assignedIPs:     sets.NewString(),
+	}, nil
+}
+
+func (r *ndpResponder) InterfaceName() string {
+	return r.iface.Name
+}
+
+// advertise sends Neighbor Advertisement for the IP.
+func (r *ndpResponder) advertise(ip net.IP) error {
+	na := &ndp.NeighborAdvertisement{
+		Override:      true,
+		TargetAddress: ip,
+		Options: []ndp.Option{
+			&ndp.LinkLayerAddress{
+				Direction: ndp.Target,
+				Addr:      r.iface.HardwareAddr,
+			},
+		},
+	}
+	return r.conn.WriteTo(na, nil, net.IPv6linklocalallnodes)
+}
+
+func (r *ndpResponder) handleNeighborSolicitation() error {
+	pkt, _, srcIP, err := r.conn.ReadFrom()
+	if err != nil {
+		return err
+	}
+	ns, ok := pkt.(*ndp.NeighborSolicitation)
+	if !ok {
+		return nil
+	}
+	var nsSourceHWAddr net.HardwareAddr
+	for _, o := range ns.Options {
+		addr, ok := o.(*ndp.LinkLayerAddress)
+		if !ok {
+			continue
+		}
+		if addr.Direction != ndp.Source {
+			continue
+		}
+		nsSourceHWAddr = addr.Addr
+		break
+	}
+	if nsSourceHWAddr == nil {
+		return nil
+	}
+	if !r.isIPAssigned(ns.TargetAddress) {
+		klog.V(4).InfoS("Ignored Neighbor Solicitation", "ip", ns.TargetAddress.String(), "interface", r.iface.Name)
+		return nil
+	}
+	na := &ndp.NeighborAdvertisement{
+		Solicited:     true,
+		Override:      true,
+		TargetAddress: ns.TargetAddress,
+		Options: []ndp.Option{
+			&ndp.LinkLayerAddress{
+				Direction: ndp.Target,
+				Addr:      r.iface.HardwareAddr,
+			},
+		},
+	}
+	if err := r.conn.WriteTo(na, nil, srcIP); err != nil {
+		return err
+	}
+	klog.V(4).InfoS("Sent Neighbor Advertisement", "ip", ns.TargetAddress.String(), "interface", r.iface.Name)
+	return nil
+}
+
+func (r *ndpResponder) Run(stopCh <-chan struct{}) {
+	for {
+		select {
+		case <-stopCh:
+			r.conn.Close()
+			return
+		default:
+			err := r.handleNeighborSolicitation()
+			if err != nil {
+				klog.ErrorS(err, "Failed to handle Neighbor Solicitation", "deviceName", r.iface.Name)
+			}
+		}
+	}
+}
+
+func (r *ndpResponder) AddIP(ip net.IP) error {
+	if !utilnet.IsIPv6(ip) {
+		return fmt.Errorf("only IPv6 is supported")
+	}
+	if r.isIPAssigned(ip) {
+		return nil
+	}
+	group, key := parseIPv6SolicitedNodeMulticastAddress(ip)
+	if err := func() error {
+		r.mutex.Lock()
+		defer r.mutex.Unlock()
+		if r.multicastGroups[key] == 0 {
+			if err := r.conn.JoinGroup(group); err != nil {
+				return fmt.Errorf("joining solicited-node multicast group %s for %q failed: %v", group, ip, err)
+			}
+			klog.InfoS("Joined solicited-node multicast group", "group", group, "interface", r.iface.Name)
+		}
+		klog.InfoS("Assigned IP to NDP responder", "ip", ip, "interface", r.iface.Name)
+		r.multicastGroups[key]++
+		r.assignedIPs.Insert(ip.String())
+		return nil
+	}(); err != nil {
+		return err
+	}
+	if err := r.advertise(ip); err != nil {
+		klog.ErrorS(err, "Failed to advertise", "ip", ip, "interface", r.iface.Name)
+	}
+	return nil
+}
+
+func (r *ndpResponder) RemoveIP(ip net.IP) error {
+	if !utilnet.IsIPv6(ip) {
+		return fmt.Errorf("only IPv6 is supported")
+	}
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	if !r.assignedIPs.Has(ip.String()) {
+		return nil
+	}
+	group, key := parseIPv6SolicitedNodeMulticastAddress(ip)
+	if r.multicastGroups[key] == 1 {
+		if err := r.conn.LeaveGroup(group); err != nil {
+			return fmt.Errorf("leaving solicited-node multicast group %s for %q failed: %v", group, ip, err)
+		}
+		klog.InfoS("Left solicited-node multicast group", "group", group, "interface", r.iface.Name)
+		delete(r.multicastGroups, key)
+	} else {
+		r.multicastGroups[key]--
+	}
+	r.assignedIPs.Delete(ip.String())
+	klog.InfoS("Removed IP from NDP responder", "ip", ip, "interface", r.iface.Name)
+	return nil
+}
+
+func (r *ndpResponder) isIPAssigned(ip net.IP) bool {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	return r.assignedIPs.Has(ip.String())
+}
diff --git a/pkg/agent/openflow/client.go b/pkg/agent/openflow/client.go
index 09627844d..33b11e48b 100644
--- a/pkg/agent/openflow/client.go
+++ b/pkg/agent/openflow/client.go
@@ -30,6 +30,7 @@ import (
 	"antrea.io/antrea/pkg/agent/util"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 	utilip "antrea.io/antrea/pkg/util/ip"
+	"antrea.io/antrea/pkg/util/runtime"
 	"antrea.io/antrea/third_party/proxy"
 )
 
@@ -113,13 +114,6 @@ type Client interface {
 	InstallServiceFlows(groupID binding.GroupIDType, svcIP net.IP, svcPort uint16, protocol binding.Protocol, affinityTimeout uint16, nodeLocalExternal bool, svcType v1.ServiceType) error
 	// UninstallServiceFlows removes flows installed by InstallServiceFlows.
 	UninstallServiceFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error
-	// InstallLoadBalancerServiceFromOutsideFlows installs flows for LoadBalancer Service traffic from outside node.
-	// The traffic is received from uplink port and will be forwarded to gateway by the installed flows. And then
-	// kube-proxy will handle the traffic.
-	// This function is only used for Windows platform.
-	InstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error
-	// UninstallLoadBalancerServiceFromOutsideFlows removes flows installed by InstallLoadBalancerServiceFromOutsideFlows.
-	UninstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error
 
 	// GetFlowTableStatus should return an array of flow table status, all existing flow tables should be included in the list.
 	GetFlowTableStatus() []binding.TableStatus
@@ -146,9 +140,7 @@ type Client interface {
 	// are removed from PolicyRule.From, else from PolicyRule.To.
 	DeletePolicyRuleAddress(ruleID uint32, addrType types.AddressType, addresses []types.Address, priority *uint16) error
 
-	// InstallBridgeUplinkFlows installs Openflow flows between bridge local port and uplink port to support
-	// host networking.
-	// This function is only used for Windows platform.
+	// InstallBridgeUplinkFlows installs Openflow flows between bridge local port and uplink port to support host networking.
 	InstallBridgeUplinkFlows() error
 
 	// InstallExternalFlows sets up flows to enable Pods to communicate to
@@ -160,8 +152,7 @@ type Client interface {
 
 	// InstallSNATMarkFlows installs flows for a local SNAT IP. On Linux, a
 	// single flow is added to mark the packets tunnelled from remote Nodes
-	// that should be SNAT'd with the SNAT IP. On Windows, an extra flow is
-	// added to perform SNAT for the marked packets with the SNAT IP.
+	// that should be SNAT'd with the SNAT IP.
 	InstallSNATMarkFlows(snatIP net.IP, mark uint32) error
 
 	// UninstallSNATMarkFlows removes the flows installed to set the packet
@@ -850,11 +841,11 @@ func (c *client) InstallExternalFlows(exceptCIDRs []net.IPNet) error {
 }
 
 func (c *client) InstallSNATMarkFlows(snatIP net.IP, mark uint32) error {
-	flows := c.snatMarkFlows(snatIP, mark)
+	flow := c.snatIPFromTunnelFlow(snatIP, mark)
 	cacheKey := fmt.Sprintf("s%x", mark)
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
-	return c.addFlows(c.snatFlowCache, cacheKey, flows)
+	return c.addFlows(c.snatFlowCache, cacheKey, []binding.Flow{flow})
 }
 
 func (c *client) UninstallSNATMarkFlows(mark uint32) error {
@@ -1247,3 +1238,19 @@ func (c *client) SendIGMPQueryPacketOut(
 	packetOutObj := packetOutBuilder.Done()
 	return c.bridge.SendPacketOut(packetOutObj)
 }
+
+func (c *client) InstallBridgeUplinkFlows() error {
+	if runtime.IsWindowsPlatform() || c.connectUplinkToBridge {
+		podCIDRMap := make(map[binding.Protocol]net.IPNet)
+		if c.nodeConfig.PodIPv4CIDR != nil {
+			podCIDRMap[binding.ProtocolIP] = *c.nodeConfig.PodIPv4CIDR
+		}
+		//TODO: support IPv6
+		flows := c.hostBridgeUplinkFlows(podCIDRMap, cookie.Default)
+		if err := c.ofEntryOperations.AddAll(flows); err != nil {
+			return err
+		}
+		c.hostNetworkingFlows = flows
+	}
+	return nil
+}
diff --git a/pkg/agent/openflow/client_other.go b/pkg/agent/openflow/client_other.go
deleted file mode 100644
index 3e5562d50..000000000
--- a/pkg/agent/openflow/client_other.go
+++ /dev/null
@@ -1,50 +0,0 @@
-//go:build !windows
-// +build !windows
-
-// package openflow is needed by antctl which is compiled for macOS too.
-
-// Copyright 2021 Antrea Authors
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package openflow
-
-import (
-	"net"
-
-	"antrea.io/antrea/pkg/agent/openflow/cookie"
-	binding "antrea.io/antrea/pkg/ovs/openflow"
-)
-
-func (c *client) InstallBridgeUplinkFlows() error {
-	if !c.connectUplinkToBridge {
-		return nil
-	}
-	// TODO(gran): support IPv6
-	if c.nodeConfig.PodIPv4CIDR != nil {
-		flows := c.hostBridgeUplinkFlows(*c.nodeConfig.PodIPv4CIDR, cookie.Default)
-		if err := c.ofEntryOperations.AddAll(flows); err != nil {
-			return err
-		}
-		c.hostNetworkingFlows = flows
-	}
-	return nil
-}
-
-func (c *client) InstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error {
-	return nil
-}
-
-func (c *client) UninstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error {
-	return nil
-}
diff --git a/pkg/agent/openflow/client_windows.go b/pkg/agent/openflow/client_windows.go
deleted file mode 100644
index bfe651df2..000000000
--- a/pkg/agent/openflow/client_windows.go
+++ /dev/null
@@ -1,53 +0,0 @@
-//go:build windows
-// +build windows
-
-// package openflow is needed by antctl which is compiled for macOS too.
-
-// Copyright 2021 Antrea Authors
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package openflow
-
-import (
-	"fmt"
-	"net"
-
-	"antrea.io/antrea/pkg/agent/openflow/cookie"
-	binding "antrea.io/antrea/pkg/ovs/openflow"
-)
-
-func (c *client) InstallBridgeUplinkFlows() error {
-	flows := c.hostBridgeUplinkFlows(*c.nodeConfig.PodIPv4CIDR, cookie.Default)
-	if err := c.ofEntryOperations.AddAll(flows); err != nil {
-		return err
-	}
-	c.hostNetworkingFlows = flows
-	return nil
-}
-
-func (c *client) InstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error {
-	c.replayMutex.RLock()
-	defer c.replayMutex.RUnlock()
-	var flows []binding.Flow
-	flows = append(flows, c.loadBalancerServiceFromOutsideFlow(svcIP, svcPort, protocol))
-	cacheKey := fmt.Sprintf("L%s%s%x", svcIP, protocol, svcPort)
-	return c.addFlows(c.serviceFlowCache, cacheKey, flows)
-}
-
-func (c *client) UninstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error {
-	c.replayMutex.RLock()
-	defer c.replayMutex.RUnlock()
-	cacheKey := fmt.Sprintf("L%s%s%x", svcIP, protocol, svcPort)
-	return c.deleteFlows(c.serviceFlowCache, cacheKey)
-}
diff --git a/pkg/agent/openflow/groups.go b/pkg/agent/openflow/groups.go
new file mode 100644
index 000000000..277cafad3
--- /dev/null
+++ b/pkg/agent/openflow/groups.go
@@ -0,0 +1,65 @@
+// Copyright 2022 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package openflow
+
+import (
+	"sync"
+
+	binding "antrea.io/antrea/pkg/ovs/openflow"
+)
+
+type GroupAllocator interface {
+	Allocate() binding.GroupIDType
+	Release(id binding.GroupIDType)
+}
+
+type groupAllocator struct {
+	// mu is a lock for the groupAllocator.
+	mu sync.Mutex
+
+	groupIDCounter binding.GroupIDType
+	recycled       []binding.GroupIDType
+}
+
+// Allocate allocates a new group ID. It allocates id from the "recycled" slices first, then increases the groupIDCounter if no
+// recycled ids exist.
+func (a *groupAllocator) Allocate() binding.GroupIDType {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	var id binding.GroupIDType
+	if len(a.recycled) != 0 {
+		id = a.recycled[len(a.recycled)-1]
+		a.recycled = a.recycled[:len(a.recycled)-1]
+	} else {
+		a.groupIDCounter += 1
+		id = a.groupIDCounter
+	}
+	return id
+}
+
+func (a *groupAllocator) Release(id binding.GroupIDType) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	a.recycled = append(a.recycled, id)
+}
+
+func NewGroupAllocator(isIPv6 bool) GroupAllocator {
+	var groupIDCounter binding.GroupIDType
+	if isIPv6 {
+		groupIDCounter = 0x10000000
+	}
+	return &groupAllocator{groupIDCounter: groupIDCounter}
+}
diff --git a/pkg/agent/openflow/pipeline.go b/pkg/agent/openflow/pipeline.go
index 784799b54..04c7feb29 100644
--- a/pkg/agent/openflow/pipeline.go
+++ b/pkg/agent/openflow/pipeline.go
@@ -42,7 +42,6 @@ import (
 
 var (
 	ClassifierTable              = binding.NewOFTable(0, "Classification")
-	UplinkTable                  = binding.NewOFTable(5, "Uplink")
 	SpoofGuardTable              = binding.NewOFTable(10, "SpoofGuard")
 	arpResponderTable            = binding.NewOFTable(20, "ARPResponder")
 	IPv6Table                    = binding.NewOFTable(21, "IPv6")
@@ -181,6 +180,23 @@ func GetTableList() []binding.Table {
 	return tables
 }
 
+// CleanOFTableCache is used to reset ofTableCache and only used in integration tests. When all integration tests about
+// openflow run in batch, unexpected flows could be installed on OVS due to stale ofTableCache, which may cause some tests
+// to fail. For example, for TestFuncA, SNATTable is needed; for TestFuncB, SNATTable is not needed. If TestFuncB is run
+// after TestFuncA, since ofTableCache (SNATTable is added by TestFuncA) is not reset, default flow of SNATTable will also
+// be realized on OVS when running TestFuncB (see "func (c *client) defaultFlows() (flows []binding.Flow)"). Note that,
+// the unexpected flows are not included in the map tableCache of OFBridge defined in pkg/ovs/openflow/ofctrl_bridge.go,
+// because the bridge will be destroyed after every test. For some tests, function checkOVSFlowMetrics (defined in
+// test/integration/agent/openflow_test.go) is used to check the flow number of every installed table. The expected table
+// list is read from the map tableCache of OFBridge, but the actual table list is dumped from OVS bridge (including the
+// unexpected flow). They are different, and as a result, TestFuncB will fail.
+func CleanOFTableCache() {
+	objs := ofTableCache.List()
+	for i := 0; i < len(objs); i++ {
+		ofTableCache.Delete(objs[i])
+	}
+}
+
 func GetAntreaPolicyEgressTables() []binding.Table {
 	return []binding.Table{
 		AntreaPolicyEgressRuleTable,
@@ -2225,20 +2241,6 @@ func (c *client) snatRuleFlow(ofPort uint32, snatIP net.IP, snatMark uint32, loc
 		Done()
 }
 
-// loadBalancerServiceFromOutsideFlow generates the flow to forward LoadBalancer service traffic from outside node
-// to gateway. kube-proxy will then handle the traffic.
-// This flow is for Windows Node only.
-func (c *client) loadBalancerServiceFromOutsideFlow(svcIP net.IP, svcPort uint16, protocol binding.Protocol) binding.Flow {
-	return UplinkTable.BuildFlow(priorityHigh).
-		MatchProtocol(protocol).
-		MatchDstPort(svcPort, nil).
-		MatchRegMark(FromUplinkRegMark).
-		MatchDstIP(svcIP).
-		Action().Output(config.HostGatewayOFPort).
-		Cookie(c.cookieAllocator.Request(cookie.Service).Raw()).
-		Done()
-}
-
 // serviceClassifierFlows generate the flows to match the first packet of Service NodePort and set a bit of a register
 // to mark the Service type as NodePort.
 func (c *client) serviceClassifierFlows(nodePortAddresses []net.IP, ipProtocol binding.Protocol) []binding.Flow {
@@ -2584,13 +2586,9 @@ func (c *client) generatePipeline() {
 	} else {
 		c.createOFTable(DNATTable, c.egressEntryTable, binding.TableMissActionNext)
 	}
-	// The default SNAT is implemented with OVS on Windows.
-	if c.enableEgress || runtime.IsWindowsPlatform() {
+	if c.enableEgress {
 		c.createOFTable(SNATTable, L2ForwardingCalcTable.GetID(), binding.TableMissActionNext)
 	}
-	if runtime.IsWindowsPlatform() || c.connectUplinkToBridge {
-		c.createOFTable(UplinkTable, SpoofGuardTable.GetID(), binding.TableMissActionNone)
-	}
 	if c.enableAntreaPolicy {
 		c.createOFTable(AntreaPolicyEgressRuleTable, EgressRuleTable.GetID(), binding.TableMissActionNext)
 		c.createOFTable(AntreaPolicyIngressRuleTable, IngressRuleTable.GetID(), binding.TableMissActionNext)
diff --git a/pkg/agent/openflow/pipeline_other.go b/pkg/agent/openflow/pipeline_other.go
index 379428004..4c1f72bde 100644
--- a/pkg/agent/openflow/pipeline_other.go
+++ b/pkg/agent/openflow/pipeline_other.go
@@ -27,14 +27,10 @@ import (
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 )
 
-func (c *client) snatMarkFlows(snatIP net.IP, mark uint32) []binding.Flow {
-	return []binding.Flow{c.snatIPFromTunnelFlow(snatIP, mark)}
-}
-
 // hostBridgeUplinkFlows generates the flows that forward traffic between the
 // bridge local port and the uplink port to support the host traffic.
 // TODO(gran): sync latest changes from pipeline_windows.go
-func (c *client) hostBridgeUplinkFlows(localSubnet net.IPNet, category cookie.Category) (flows []binding.Flow) {
+func (c *client) hostBridgeUplinkFlows(localSubnetMap map[binding.Protocol]net.IPNet, category cookie.Category) (flows []binding.Flow) {
 	flows = []binding.Flow{
 		ClassifierTable.BuildFlow(priorityNormal).
 			MatchInPort(config.UplinkOFPort).
diff --git a/pkg/agent/openflow/pipeline_windows.go b/pkg/agent/openflow/pipeline_windows.go
index ccec12503..a3df5f328 100644
--- a/pkg/agent/openflow/pipeline_windows.go
+++ b/pkg/agent/openflow/pipeline_windows.go
@@ -22,65 +22,13 @@ import (
 
 	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/openflow/cookie"
-	"antrea.io/antrea/pkg/agent/types"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 )
 
-const (
-	// ctZoneSNAT is only used on Windows and only when AntreaProxy is enabled.
-	// When a Pod access a ClusterIP Service, and the IP of the selected endpoint
-	// is not in "cluster-cidr". The request packets need to be SNAT'd(set src IP to local Node IP)
-	// after have been DNAT'd(set dst IP to endpoint IP).
-	// For example, the endpoint Pod may run in hostNetwork mode and the IP of the endpoint
-	// will be the current Node IP.
-	// We need to use a different ct_zone to track the SNAT'd connection because OVS
-	// does not support doing both DNAT and SNAT in the same ct_zone.
-	//
-	// An example of the connection is a Pod accesses kubernetes API service:
-	// Pod --> DNAT(CtZone) --> SNAT(ctZoneSNAT) --> Endpoint(API server NodeIP)
-	// Pod <-- unDNAT(CtZone) <-- unSNAT(ctZoneSNAT) <-- Endpoint(API server NodeIP)
-	ctZoneSNAT = 0xffdc
-)
-
-var (
-	// snatCTMark indicates SNAT is performed for packets of the connection.
-	snatCTMark = binding.NewCTMark(0x40, 0, 31)
-)
-
-func (c *client) snatMarkFlows(snatIP net.IP, mark uint32) []binding.Flow {
-	snatIPRange := &binding.IPRange{StartIP: snatIP, EndIP: snatIP}
-	nextTable := ConntrackCommitTable.GetNext()
-	flows := []binding.Flow{
-		c.snatIPFromTunnelFlow(snatIP, mark),
-		ConntrackCommitTable.BuildFlow(priorityNormal).
-			MatchProtocol(binding.ProtocolIP).
-			MatchCTStateNew(true).MatchCTStateTrk(true).MatchCTStateDNAT(false).
-			MatchPktMark(mark, &types.SNATIPMarkMask).
-			Action().CT(true, nextTable, CtZone).
-			SNAT(snatIPRange, nil).
-			LoadToCtMark(snatCTMark).CTDone().
-			Cookie(c.cookieAllocator.Request(cookie.SNAT).Raw()).
-			Done(),
-	}
-
-	if c.enableProxy {
-		flows = append(flows, ConntrackCommitTable.BuildFlow(priorityNormal).
-			MatchProtocol(binding.ProtocolIP).
-			MatchCTStateNew(true).MatchCTStateTrk(true).MatchCTStateDNAT(true).
-			MatchPktMark(mark, &types.SNATIPMarkMask).
-			Action().CT(true, nextTable, ctZoneSNAT).
-			SNAT(snatIPRange, nil).
-			LoadToCtMark(snatCTMark).CTDone().
-			Cookie(c.cookieAllocator.Request(cookie.SNAT).Raw()).
-			Done())
-	}
-	return flows
-}
-
 // hostBridgeUplinkFlows generates the flows that forward traffic between the
 // bridge local port and the uplink port to support the host traffic with
 // outside.
-func (c *client) hostBridgeUplinkFlows(localSubnet net.IPNet, category cookie.Category) (flows []binding.Flow) {
+func (c *client) hostBridgeUplinkFlows(localSubnetMap map[binding.Protocol]net.IPNet, category cookie.Category) (flows []binding.Flow) {
 	flows = []binding.Flow{
 		ClassifierTable.BuildFlow(priorityNormal).
 			MatchInPort(config.UplinkOFPort).
@@ -96,14 +44,17 @@ func (c *client) hostBridgeUplinkFlows(localSubnet net.IPNet, category cookie.Ca
 	if c.networkConfig.TrafficEncapMode.SupportsNoEncap() {
 		// If NoEncap is enabled, the reply packets from remote Pod can be forwarded to local Pod directly.
 		// by explicitly resubmitting them to ServiceHairpinTable and marking "macRewriteMark" at same time.
-		flows = append(flows, ClassifierTable.BuildFlow(priorityHigh).MatchProtocol(binding.ProtocolIP).
-			MatchInPort(config.UplinkOFPort).
-			MatchDstIPNet(localSubnet).
-			Action().LoadRegMark(FromUplinkRegMark).
-			Action().LoadRegMark(RewriteMACRegMark).
-			Action().GotoTable(ServiceHairpinTable.GetID()).
-			Cookie(c.cookieAllocator.Request(category).Raw()).
-			Done())
+		for ipProtocol, localSubnet := range localSubnetMap {
+			flows = append(flows, ClassifierTable.BuildFlow(priorityHigh).
+				MatchProtocol(ipProtocol).
+				MatchInPort(config.UplinkOFPort).
+				MatchDstIPNet(localSubnet).
+				Action().LoadRegMark(FromUplinkRegMark).
+				Action().LoadRegMark(RewriteMACRegMark).
+				Action().GotoTable(ServiceHairpinTable.GetID()).
+				Cookie(c.cookieAllocator.Request(category).Raw()).
+				Done())
+		}
 	}
 	return flows
 }
diff --git a/pkg/agent/proxy/proxier.go b/pkg/agent/proxy/proxier.go
index 571abc29c..2ab8f7eeb 100644
--- a/pkg/agent/proxy/proxier.go
+++ b/pkg/agent/proxy/proxier.go
@@ -168,23 +168,25 @@ func (p *proxier) removeStaleServices() {
 		}
 		// Remove Service group whose Endpoints are local.
 		if svcInfo.NodeLocalExternal() {
-			groupIDLocal, _ := p.groupCounter.Get(svcPortName, true)
-			if err := p.ofClient.UninstallServiceGroup(groupIDLocal); err != nil {
-				klog.ErrorS(err, "Failed to remove Group of local Endpoints for Service", "Service", svcPortName)
-				continue
+			if groupIDLocal, exist := p.groupCounter.Get(svcPortName, true); exist {
+				if err := p.ofClient.UninstallServiceGroup(groupIDLocal); err != nil {
+					klog.ErrorS(err, "Failed to remove Group of local Endpoints for Service", "Service", svcPortName)
+					continue
+				}
+				p.groupCounter.Recycle(svcPortName, true)
 			}
-			p.groupCounter.Recycle(svcPortName, true)
 		}
 		// Remove Service group which has all Endpoints.
-		groupID, _ := p.groupCounter.Get(svcPortName, false)
-		if err := p.ofClient.UninstallServiceGroup(groupID); err != nil {
-			klog.ErrorS(err, "Failed to remove Group of all Endpoints for Service", "Service", svcPortName)
-			continue
+		if groupID, exist := p.groupCounter.Get(svcPortName, false); exist {
+			if err := p.ofClient.UninstallServiceGroup(groupID); err != nil {
+				klog.ErrorS(err, "Failed to remove Group of all Endpoints for Service", "Service", svcPortName)
+				continue
+			}
+			p.groupCounter.Recycle(svcPortName, false)
 		}
 
 		delete(p.serviceInstalledMap, svcPortName)
 		p.deleteServiceByIP(svcInfo.String())
-		p.groupCounter.Recycle(svcPortName, false)
 	}
 }
 
@@ -314,9 +316,6 @@ func (p *proxier) installLoadBalancerService(groupID binding.GroupIDType, loadBa
 			if err := p.ofClient.InstallServiceFlows(groupID, net.ParseIP(ingress), svcPort, protocol, affinityTimeout, nodeLocalExternal, corev1.ServiceTypeLoadBalancer); err != nil {
 				return fmt.Errorf("failed to install Service LoadBalancer load balancing flows: %w", err)
 			}
-			if err := p.ofClient.InstallLoadBalancerServiceFromOutsideFlows(net.ParseIP(ingress), svcPort, protocol); err != nil {
-				return fmt.Errorf("failed to install Service LoadBalancer flows: %w", err)
-			}
 		}
 	}
 	if p.proxyAll {
@@ -334,9 +333,6 @@ func (p *proxier) uninstallLoadBalancerService(loadBalancerIPStrings []string, s
 			if err := p.ofClient.UninstallServiceFlows(net.ParseIP(ingress), svcPort, protocol); err != nil {
 				return fmt.Errorf("failed to remove Service LoadBalancer load balancing flows: %w", err)
 			}
-			if err := p.ofClient.UninstallLoadBalancerServiceFromOutsideFlows(net.ParseIP(ingress), svcPort, protocol); err != nil {
-				return fmt.Errorf("failed to remove Service LoadBalancer flows: %w", err)
-			}
 		}
 	}
 	if p.proxyAll {
@@ -351,7 +347,7 @@ func (p *proxier) uninstallLoadBalancerService(loadBalancerIPStrings []string, s
 func (p *proxier) installServices() {
 	for svcPortName, svcPort := range p.serviceMap {
 		svcInfo := svcPort.(*types.ServiceInfo)
-		groupID, _ := p.groupCounter.Get(svcPortName, false)
+		groupID := p.groupCounter.AllocateIfNotExist(svcPortName, false)
 		endpointsInstalled, ok := p.endpointsInstalledMap[svcPortName]
 		if !ok {
 			endpointsInstalled = map[string]k8sproxy.Endpoint{}
@@ -455,7 +451,7 @@ func (p *proxier) installServices() {
 			if p.proxyAll {
 				if svcInfo.NodeLocalExternal() {
 					// Install another group when Service externalTrafficPolicy is Local.
-					groupIDLocal, _ := p.groupCounter.Get(svcPortName, true)
+					groupIDLocal := p.groupCounter.AllocateIfNotExist(svcPortName, true)
 					var localEndpointList []k8sproxy.Endpoint
 					for _, ed := range endpointUpdateList {
 						if !ed.GetIsLocal() {
@@ -472,12 +468,13 @@ func (p *proxier) installServices() {
 					// unconditionally. If the group doesn't exist on OVS, then the return value will be nil; if the
 					// group exists on OVS, and after it is uninstalled successfully, then the return value will be also
 					// nil.
-					groupIDLocal, _ := p.groupCounter.Get(svcPortName, true)
-					if err := p.ofClient.UninstallServiceGroup(groupIDLocal); err != nil {
-						klog.ErrorS(err, "Failed to remove Group of local Endpoints for Service", "Service", svcPortName)
-						continue
+					if groupIDLocal, exist := p.groupCounter.Get(svcPortName, true); exist {
+						if err := p.ofClient.UninstallServiceGroup(groupIDLocal); err != nil {
+							klog.ErrorS(err, "Failed to remove Group of local Endpoints for Service", "Service", svcPortName)
+							continue
+						}
+						p.groupCounter.Recycle(svcPortName, true)
 					}
-					p.groupCounter.Recycle(svcPortName, true)
 				}
 			}
 
@@ -528,7 +525,7 @@ func (p *proxier) installServices() {
 			// group whose Endpoints are local.
 			nGroupID := groupID
 			if svcInfo.NodeLocalExternal() {
-				nGroupID, _ = p.groupCounter.Get(svcPortName, true)
+				nGroupID = p.groupCounter.AllocateIfNotExist(svcPortName, true)
 			}
 
 			if p.proxyAll {
@@ -792,9 +789,12 @@ func (p *proxier) GetServiceFlowKeys(serviceName, namespace string) ([]string, [
 		svcFlows := p.ofClient.GetServiceFlowKeys(svcInfo.ClusterIP(), uint16(svcInfo.Port()), svcInfo.OFProtocol, epList)
 		flows = append(flows, svcFlows...)
 
-		// TODO: This causes antctl command output incomplete groups, fix it latter.
-		groupID, _ := p.groupCounter.Get(svcPortName, false)
-		groups = append(groups, groupID)
+		if groupID, ok := p.groupCounter.Get(svcPortName, false); ok {
+			groups = append(groups, groupID)
+		}
+		if groupID, ok := p.groupCounter.Get(svcPortName, true); ok {
+			groups = append(groups, groupID)
+		}
 	}
 
 	return flows, groups, found
diff --git a/pkg/agent/proxy/types/groupcounter.go b/pkg/agent/proxy/types/groupcounter.go
index 657f95224..9005f9cb1 100644
--- a/pkg/agent/proxy/types/groupcounter.go
+++ b/pkg/agent/proxy/types/groupcounter.go
@@ -20,40 +20,35 @@ import (
 
 	"k8s.io/apimachinery/pkg/util/sets"
 
+	"antrea.io/antrea/pkg/agent/openflow"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 	k8sproxy "antrea.io/antrea/third_party/proxy"
 )
 
 // GroupCounter generates and manages global unique group ID.
 type GroupCounter interface {
-	// Get generates a global unique group ID for a specific service.
-	// If the group ID of the service has been generated, then return the
-	// prior one. The bool return value indicates whether the groupID is newly
-	// generated.
+	// AllocateIfNotExist generates a global unique group ID for a Service if the group ID has not been generated, then
+	// return the group ID (newly allocated or already allocated).
+	AllocateIfNotExist(svcPortName k8sproxy.ServicePortName, isEndpointsLocal bool) binding.GroupIDType
+	// Get gets the group ID for the Service.
 	Get(svcPortName k8sproxy.ServicePortName, isEndpointsLocal bool) (binding.GroupIDType, bool)
-	// Recycle removes a Service Group ID mapping. The recycled groupID can be
-	// reused.
+	// Recycle removes the Service group ID mapping. The recycled group ID can be reused.
 	Recycle(svcPortName k8sproxy.ServicePortName, isEndpointsLocal bool) bool
-	// GetAllGroupIDs gets all groupID related to a Service.
+	// GetAllGroupIDs gets all group IDs related to the Service.
 	GetAllGroupIDs(svcNamespacedName string) []binding.GroupIDType
 }
 
 type groupCounter struct {
 	mu             sync.Mutex
-	groupIDCounter binding.GroupIDType
-	recycled       []binding.GroupIDType
+	groupAllocator openflow.GroupAllocator
 	groupIDUpdates chan<- string
 
 	servicePortNamesMap map[string]sets.String
 	groupMap            map[string]binding.GroupIDType
 }
 
-func NewGroupCounter(isIPv6 bool, groupIDUpdates chan<- string) *groupCounter {
-	var groupIDCounter binding.GroupIDType
-	if isIPv6 {
-		groupIDCounter = 0x10000000
-	}
-	return &groupCounter{groupMap: map[string]binding.GroupIDType{}, groupIDCounter: groupIDCounter, groupIDUpdates: groupIDUpdates, servicePortNamesMap: map[string]sets.String{}}
+func NewGroupCounter(groupAllocator openflow.GroupAllocator, groupIDUpdates chan<- string) *groupCounter {
+	return &groupCounter{groupMap: map[string]binding.GroupIDType{}, groupAllocator: groupAllocator, groupIDUpdates: groupIDUpdates, servicePortNamesMap: map[string]sets.String{}}
 }
 
 func keyString(svcPortName k8sproxy.ServicePortName, isEndpointsLocal bool) string {
@@ -84,26 +79,26 @@ func (c *groupCounter) deleteServicePortNameMap(svcNamespacedName string, svcKey
 	}
 }
 
-func (c *groupCounter) Get(svcPortName k8sproxy.ServicePortName, isEndpointsLocal bool) (binding.GroupIDType, bool) {
+func (c *groupCounter) AllocateIfNotExist(svcPortName k8sproxy.ServicePortName, isEndpointsLocal bool) binding.GroupIDType {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	key := keyString(svcPortName, isEndpointsLocal)
 	if id, ok := c.groupMap[key]; ok {
-		return id, false
-	} else if len(c.recycled) != 0 {
-		id = c.recycled[len(c.recycled)-1]
-		c.recycled = c.recycled[:len(c.recycled)-1]
-		c.groupMap[key] = id
-		c.updateServicePortNameMap(svcPortName.NamespacedName.String(), key)
-		c.groupIDUpdates <- svcPortName.NamespacedName.String()
-		return id, true
-	} else {
-		c.groupIDCounter += 1
-		c.groupMap[key] = c.groupIDCounter
-		c.updateServicePortNameMap(svcPortName.NamespacedName.String(), key)
-		c.groupIDUpdates <- svcPortName.NamespacedName.String()
-		return c.groupIDCounter, true
+		return id
 	}
+	id := c.groupAllocator.Allocate()
+	c.groupMap[key] = id
+	c.updateServicePortNameMap(svcPortName.NamespacedName.String(), key)
+	c.groupIDUpdates <- svcPortName.NamespacedName.String()
+	return id
+}
+
+func (c *groupCounter) Get(svcPortName k8sproxy.ServicePortName, isEndpointsLocal bool) (binding.GroupIDType, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	key := keyString(svcPortName, isEndpointsLocal)
+	id, exist := c.groupMap[key]
+	return id, exist
 }
 
 func (c *groupCounter) Recycle(svcPortName k8sproxy.ServicePortName, isEndpointsLocal bool) bool {
@@ -113,7 +108,7 @@ func (c *groupCounter) Recycle(svcPortName k8sproxy.ServicePortName, isEndpoints
 	key := keyString(svcPortName, isEndpointsLocal)
 	if id, ok := c.groupMap[key]; ok {
 		delete(c.groupMap, key)
-		c.recycled = append(c.recycled, id)
+		c.groupAllocator.Release(id)
 		c.deleteServicePortNameMap(svcPortName.NamespacedName.String(), key)
 		c.groupIDUpdates <- svcPortName.NamespacedName.String()
 		return true
diff --git a/pkg/apis/crd/v1alpha1/doc.go b/pkg/apis/crd/v1alpha1/doc.go
index 2914f1ed7..bf32d20df 100644
--- a/pkg/apis/crd/v1alpha1/doc.go
+++ b/pkg/apis/crd/v1alpha1/doc.go
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// +k8s:openapi-gen=true
 // +k8s:deepcopy-gen=package
 // +k8s:defaulter-gen=TypeMeta
 // +groupName=crd.antrea.io
diff --git a/pkg/apis/crd/v1alpha2/doc.go b/pkg/apis/crd/v1alpha2/doc.go
index 8c64c9798..6a8a7cccb 100644
--- a/pkg/apis/crd/v1alpha2/doc.go
+++ b/pkg/apis/crd/v1alpha2/doc.go
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// +k8s:openapi-gen=true
 // +k8s:deepcopy-gen=package
 // +groupName=crd.antrea.io
 
diff --git a/pkg/apis/crd/v1alpha3/doc.go b/pkg/apis/crd/v1alpha3/doc.go
index 5ad2c22cb..42fa8ee6b 100644
--- a/pkg/apis/crd/v1alpha3/doc.go
+++ b/pkg/apis/crd/v1alpha3/doc.go
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// +k8s:openapi-gen=true
 // +k8s:deepcopy-gen=package
 // +groupName=crd.antrea.io
 
diff --git a/pkg/controller/networkpolicy/validate.go b/pkg/controller/networkpolicy/validate.go
index bdab883d2..2188bf7d8 100644
--- a/pkg/controller/networkpolicy/validate.go
+++ b/pkg/controller/networkpolicy/validate.go
@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apiserver/pkg/authentication/serviceaccount"
 	"k8s.io/klog/v2"
 
@@ -472,6 +473,9 @@ func (v *antreaPolicyValidator) validateAppliedTo(ingress, egress []crdv1alpha1.
 			if eachAppliedTo.ServiceAccount != nil && appliedToFieldsNum > 1 {
 				return "serviceAccount cannot be set with other peers in appliedTo", false
 			}
+			if reason, allowed := checkSelectorsLabels(eachAppliedTo.PodSelector, eachAppliedTo.NamespaceSelector, eachAppliedTo.ExternalEntitySelector); !allowed {
+				return reason, allowed
+			}
 		}
 		return "", true
 	}
@@ -511,6 +515,9 @@ func (v *antreaPolicyValidator) validatePeers(ingress, egress []crdv1alpha1.Rule
 			if peer.ServiceAccount != nil && peerFieldsNum > 1 {
 				return "serviceAccount cannot be set with other peers in rules", false
 			}
+			if reason, allowed := checkSelectorsLabels(peer.PodSelector, peer.NamespaceSelector, peer.ExternalEntitySelector); !allowed {
+				return reason, allowed
+			}
 		}
 		return "", true
 	}
@@ -549,6 +556,31 @@ func numFieldsSetInPeer(peer crdv1alpha1.NetworkPolicyPeer) int {
 	return num
 }
 
+// checkSelectorsLabels validates labels used in all selectors passed in.
+func checkSelectorsLabels(selectors ...*metav1.LabelSelector) (string, bool) {
+	validateLabels := func(labels map[string]string) (string, bool) {
+		for k, v := range labels {
+			err := validation.IsQualifiedName(k)
+			if err != nil {
+				return fmt.Sprintf("Invalid label key: %s: %s", k, strings.Join(err, "; ")), false
+			}
+			err = validation.IsValidLabelValue(v)
+			if err != nil {
+				return fmt.Sprintf("Invalid label value: %s: %s", v, strings.Join(err, "; ")), false
+			}
+		}
+		return "", true
+	}
+	for _, selector := range selectors {
+		if selector != nil {
+			if reason, allowed := validateLabels(selector.MatchLabels); !allowed {
+				return reason, allowed
+			}
+		}
+	}
+	return "", true
+}
+
 // validateTierForPolicy validates whether a referenced Tier exists.
 func (v *antreaPolicyValidator) validateTierForPolicy(tier string) (string, bool) {
 	// "tier" must exist before referencing
@@ -706,6 +738,9 @@ func validateAntreaGroupSpec(s crdv1alpha2.GroupSpec) (string, bool) {
 	}
 	selector, serviceRef, ipBlock, ipBlocks, childGroups := 0, 0, 0, 0, 0
 	if s.NamespaceSelector != nil || s.ExternalEntitySelector != nil || s.PodSelector != nil {
+		if reason, allowed := checkSelectorsLabels(s.PodSelector, s.NamespaceSelector, s.ExternalEntitySelector); !allowed {
+			return reason, allowed
+		}
 		selector = 1
 	}
 	if s.IPBlock != nil {
diff --git a/plugins/octant/go.sum b/plugins/octant/go.sum
index c85b029bb..e674d8959 100644
--- a/plugins/octant/go.sum
+++ b/plugins/octant/go.sum
@@ -730,8 +730,12 @@ github.com/mattn/go-shellwords v1.0.12/go.mod h1:EZzvwXDESEeg03EKmM+RmDnNOPKG4lL
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
+github.com/mdlayher/arp v0.0.0-20191213142603-f72070a231fc/go.mod h1:eOj1DDj3NAZ6yv+WafaKzY37MFZ58TdfIhQ+8nQbiis=
+github.com/mdlayher/ethernet v0.0.0-20190313224307-5b5fc417d966/go.mod h1:5s5p/sMJ6sNsFl6uCh85lkFGV8kLuIYJCRJLavVJwvg=
+github.com/mdlayher/ethernet v0.0.0-20190606142754-0394541c37b7/go.mod h1:U6ZQobyTjI/tJyq2HG+i/dfSoFUt8/aZCM+GKtmFk/Y=
 github.com/mdlayher/ethtool v0.0.0-20210210192532-2b88debcdd43/go.mod h1:+t7E0lkKfbBsebllff1xdTmyJt8lH37niI6kwFk9OTo=
 github.com/mdlayher/genetlink v1.0.0/go.mod h1:0rJ0h4itni50A86M2kHcgS85ttZazNt7a8H2a2cw0Gc=
+github.com/mdlayher/ndp v0.0.0-20210831201139-f982b8766fb5/go.mod h1:32w/5dDZWVSEOxyniAgKK4d7dHTuO6TCxWmUznQe3f8=
 github.com/mdlayher/netlink v0.0.0-20190409211403-11939a169225/go.mod h1:eQB3mZE4aiYnlUsyGGCOpPETfdQq4Jhsgf1fk3cwQaA=
 github.com/mdlayher/netlink v1.0.0/go.mod h1:KxeJAFOFLG6AjpyDkQ/iIhxygIUKD+vcwqcnu43w/+M=
 github.com/mdlayher/netlink v1.1.0/go.mod h1:H4WCitaheIsdF9yOYu8CFmCgQthAPIWZmcKp9uZHgmY=
@@ -742,6 +746,9 @@ github.com/mdlayher/netlink v1.2.1/go.mod h1:bacnNlfhqHqqLo4WsYeXSqfyXkInQ9JneWI
 github.com/mdlayher/netlink v1.2.2-0.20210123213345-5cc92139ae3e/go.mod h1:bacnNlfhqHqqLo4WsYeXSqfyXkInQ9JneWI68v1KwSU=
 github.com/mdlayher/netlink v1.3.0/go.mod h1:xK/BssKuwcRXHrtN04UBkwQ6dY9VviGGuriDdoPSWys=
 github.com/mdlayher/netlink v1.4.0/go.mod h1:dRJi5IABcZpBD2A3D0Mv/AiX8I9uDEu5oGkAVrekmf8=
+github.com/mdlayher/raw v0.0.0-20190313224157-43dbcdd7739d/go.mod h1:r1fbeITl2xL/zLbVnNHFyOzQJTgr/3fpf1lJX/cjzR8=
+github.com/mdlayher/raw v0.0.0-20190606142536-fef19f00fc18/go.mod h1:7EpbotpCmVZcu+KCX4g9WaRNuu11uyhiW7+Le1dKawg=
+github.com/mdlayher/raw v0.0.0-20211126142749-4eae47f3d54b/go.mod h1:7EpbotpCmVZcu+KCX4g9WaRNuu11uyhiW7+Le1dKawg=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso=
 github.com/miekg/dns v1.1.43/go.mod h1:+evo5L0630/F6ca/Z9+GAqzhjGyn8/c+TBaOyfEl0V4=
@@ -1046,6 +1053,7 @@ github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1
 github.com/yvasiyarov/go-metrics v0.0.0-20140926110328-57bccd1ccd43/go.mod h1:aX5oPXxHm3bOH+xeAttToC8pqch2ScQN/JoXYupl6xs=
 github.com/yvasiyarov/gorelic v0.0.0-20141212073537-a9bba5b9ab50/go.mod h1:NUSPSUX/bi6SeDMUh6brw0nXpxHnc96TguQh0+r/ssA=
 github.com/yvasiyarov/newrelic_platform_go v0.0.0-20140908184405-b21fdbd4370f/go.mod h1:GlGEuHIJweS1mbCqG+7vt2nvWLzLLnRHbXz5JKd/Qbg=
+gitlab.com/golang-commonmark/puny v0.0.0-20191124015043-9f83538fa04f/go.mod h1:Tiuhl+njh/JIg0uS/sOJVYi0x2HEa5rc1OAaVsb5tAs=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
@@ -1157,8 +1165,10 @@ golang.org/x/net v0.0.0-20181220203305-927f97764cc3/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190313220215-9f648a60d977/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190320064053-1272bf9dcd53/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190419010253-1f3472d942ba/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190501004415-9ce7a6920f09/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
@@ -1185,6 +1195,7 @@ golang.org/x/net v0.0.0-20200506145744-7e3656a0809f/go.mod h1:qpuaurCH72eLCgpAm/
 golang.org/x/net v0.0.0-20200513185701-a91f0712d120/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200520182314-0ba52f642ac2/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200602114024-627f9648deb9/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
@@ -1249,12 +1260,14 @@ golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190321052220-f7bb7a8bee54/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190411185658-b44545bcd369/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190418153312-f0ce4c0180be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190514135907-3a4b5fb9f71f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190522044717-8097e1b27ff5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190602015325-4c4f7f33c9ed/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190606122018-79a91cf218c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190616124812-15dcb6c0061f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1296,6 +1309,7 @@ golang.org/x/sys v0.0.0-20200511232937-7e40ca221e25/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200515095857-1151b9dac4a9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200523222454-059865788121/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200602100848-8d3cce7afc34/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200622214017-ed371f2e16b4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=


Test patch:
diff --git a/pkg/agent/controller/networkpolicy/audit_logging_test.go b/pkg/agent/controller/networkpolicy/audit_logging_test.go
index 66547d10f..57438c374 100644
--- a/pkg/agent/controller/networkpolicy/audit_logging_test.go
+++ b/pkg/agent/controller/networkpolicy/audit_logging_test.go
@@ -112,17 +112,22 @@ func newTestAntreaPolicyLogger(bufferLength time.Duration, clock Clock) (*Antrea
 }
 
 func newLogInfo(disposition string) (*logInfo, string) {
-	expected := fmt.Sprintf("AntreaPolicyIngressRule AntreaNetworkPolicy:default/test %s 0 0.0.0.0 1.1.1.1 60 TCP", disposition)
-	return &logInfo{
+	testLogInfo := &logInfo{
 		tableName:   "AntreaPolicyIngressRule",
 		npRef:       "AntreaNetworkPolicy:default/test",
 		ofPriority:  "0",
 		disposition: disposition,
 		srcIP:       "0.0.0.0",
+		srcPort:     "35402",
 		destIP:      "1.1.1.1",
-		pktLength:   60,
+		destPort:    "80",
 		protocolStr: "TCP",
-	}, expected
+		pktLength:   60,
+	}
+	expected := fmt.Sprintf("%s %s %s %s %s %s %s %s %s %d", testLogInfo.tableName, testLogInfo.npRef, testLogInfo.disposition,
+		testLogInfo.ofPriority, testLogInfo.srcIP, testLogInfo.srcPort, testLogInfo.destIP, testLogInfo.destPort,
+		testLogInfo.protocolStr, testLogInfo.pktLength)
+	return testLogInfo, expected
 }
 
 func sendMultiplePackets(antreaLogger *AntreaPolicyLogger, ob *logInfo, numPackets int, sendInterval time.Duration) {
diff --git a/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go b/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
index 09fff89c3..ac17fd167 100644
--- a/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
+++ b/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
@@ -32,6 +32,7 @@ import (
 	"k8s.io/component-base/metrics/legacyregistry"
 
 	"antrea.io/antrea/pkg/agent/metrics"
+	"antrea.io/antrea/pkg/agent/openflow"
 	proxytypes "antrea.io/antrea/pkg/agent/proxy/types"
 	agenttypes "antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
@@ -54,7 +55,8 @@ func newTestController() (*Controller, *fake.Clientset, *mockReconciler) {
 	clientset := &fake.Clientset{}
 	ch := make(chan agenttypes.EntityReference, 100)
 	ch2 := make(chan string, 100)
-	groupCounters := []proxytypes.GroupCounter{proxytypes.NewGroupCounter(false, ch2)}
+	groupIDAllocator := openflow.NewGroupAllocator(false)
+	groupCounters := []proxytypes.GroupCounter{proxytypes.NewGroupCounter(groupIDAllocator, ch2)}
 	controller, _ := NewNetworkPolicyController(&antreaClientGetter{clientset}, nil, nil, "node1", ch, groupCounters, ch2,
 		true, true, true, true, testAsyncDeleteInterval, "8.8.8.8:53", true, false)
 	reconciler := newMockReconciler()
diff --git a/pkg/agent/controller/networkpolicy/packetin_test.go b/pkg/agent/controller/networkpolicy/packetin_test.go
deleted file mode 100644
index c7cd3f684..000000000
--- a/pkg/agent/controller/networkpolicy/packetin_test.go
+++ /dev/null
@@ -1,61 +0,0 @@
-// Copyright 2020 Antrea Authors
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package networkpolicy
-
-import (
-	"net"
-	"testing"
-
-	"antrea.io/libOpenflow/protocol"
-	"antrea.io/libOpenflow/util"
-	"antrea.io/ofnet/ofctrl"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestGetPacketInfo(t *testing.T) {
-	tests := []struct {
-		name       string
-		pktIn      *ofctrl.PacketIn
-		expectedOb logInfo
-		wantErr    bool
-	}{
-		{
-			"ipv4",
-			&ofctrl.PacketIn{
-				Reason: 1,
-				Data: protocol.Ethernet{
-					Ethertype: 0x0800,
-					Data: util.Message(&protocol.IPv4{
-						NWSrc:    net.IPv4(1, 1, 1, 1),
-						NWDst:    net.IPv4(2, 2, 2, 2),
-						Length:   1,
-						Protocol: 6,
-					}),
-				},
-			},
-			logInfo{srcIP: "1.1.1.1", destIP: "2.2.2.2", pktLength: 1, protocolStr: "TCP"},
-			false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			actualOb := logInfo{}
-			if err := getPacketInfo(tt.pktIn, &actualOb); (err != nil) != tt.wantErr {
-				t.Errorf("getPacketInfo() error = %v, wantErr %v", err, tt.wantErr)
-			}
-			assert.Equal(t, tt.expectedOb, actualOb, "Expect to retrieve exact packet info while differed")
-		})
-	}
-}
diff --git a/pkg/agent/controller/networkpolicy/reconciler_test.go b/pkg/agent/controller/networkpolicy/reconciler_test.go
index 4ffd07571..bd8f1cf49 100644
--- a/pkg/agent/controller/networkpolicy/reconciler_test.go
+++ b/pkg/agent/controller/networkpolicy/reconciler_test.go
@@ -101,7 +101,8 @@ func newCIDR(cidrStr string) *net.IPNet {
 func newTestReconciler(t *testing.T, controller *gomock.Controller, ifaceStore interfacestore.InterfaceStore, ofClient *openflowtest.MockClient, v4Enabled, v6Enabled bool) *reconciler {
 	f, _ := newMockFQDNController(t, controller, nil)
 	ch := make(chan string, 100)
-	groupCounters := []proxytypes.GroupCounter{proxytypes.NewGroupCounter(false, ch)}
+	groupIDAllocator := openflow.NewGroupAllocator(v6Enabled)
+	groupCounters := []proxytypes.GroupCounter{proxytypes.NewGroupCounter(groupIDAllocator, ch)}
 	r := newReconciler(ofClient, ifaceStore, newIDAllocator(testAsyncDeleteInterval), f, groupCounters, v4Enabled, v6Enabled)
 	return r
 }
diff --git a/pkg/agent/controller/serviceexternalip/controller_test.go b/pkg/agent/controller/serviceexternalip/controller_test.go
index 38ecf7c93..5c9ff627d 100644
--- a/pkg/agent/controller/serviceexternalip/controller_test.go
+++ b/pkg/agent/controller/serviceexternalip/controller_test.go
@@ -73,7 +73,6 @@ func fakeHashFn(invert bool) func([]string) []string {
 var _ memberlist.Interface = (*fakeMemberlistCluster)(nil)
 
 func (f *fakeMemberlistCluster) AddClusterEventHandler(h memberlist.ClusterNodeEventHandler) {
-
 }
 
 func (f *fakeMemberlistCluster) AliveNodes() sets.String {
@@ -136,6 +135,7 @@ func newFakeController(t *testing.T, objs ...runtime.Object) *fakeController {
 		externalIPStates:      make(map[apimachinerytypes.NamespacedName]externalIPState),
 		cluster:               memberlistCluster,
 		ipAssigner:            mockIPAssigner,
+		assignedIPs:           make(map[string]sets.String),
 	}
 	return &fakeController{
 		ServiceExternalIPController: eipController,
@@ -207,6 +207,7 @@ func makeEndpoints(name, namespace string, addresses, notReadyAddresses map[stri
 func TestCreateService(t *testing.T) {
 	tests := []struct {
 		name                     string
+		previousExternalIPStates map[apimachinerytypes.NamespacedName]externalIPState
 		existingEndpoints        []*corev1.Endpoints
 		serviceToCreate          *corev1.Service
 		healthyNodes             []string
@@ -216,10 +217,11 @@ func TestCreateService(t *testing.T) {
 		expectError              bool
 	}{
 		{
-			name:              "new Service created and local Node selected",
-			existingEndpoints: nil,
-			serviceToCreate:   servicePolicyCluster,
-			healthyNodes:      []string{fakeNode1, fakeNode2},
+			name:                     "new Service created and local Node selected",
+			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
+			existingEndpoints:        nil,
+			serviceToCreate:          servicePolicyCluster,
+			healthyNodes:             []string{fakeNode1, fakeNode2},
 			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockIPAssigner.EXPECT().AssignIP(fakeServiceExternalIP1)
 			},
@@ -232,14 +234,13 @@ func TestCreateService(t *testing.T) {
 			expectError: false,
 		},
 		{
-			name:              "new Service created and local Node not selected",
-			existingEndpoints: nil,
-			serviceToCreate:   servicePolicyCluster,
-			healthyNodes:      []string{fakeNode1, fakeNode2},
-			overrideHashFn:    fakeHashFn(true),
-			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
-				mockIPAssigner.EXPECT().UnassignIP(fakeServiceExternalIP1)
-			},
+			name:                     "new Service created and local Node not selected",
+			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
+			existingEndpoints:        nil,
+			serviceToCreate:          servicePolicyCluster,
+			healthyNodes:             []string{fakeNode1, fakeNode2},
+			overrideHashFn:           fakeHashFn(true),
+			expectedCalls:            func(mockIPAssigner *ipassignertest.MockIPAssigner) {},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
 				keyFor(servicePolicyCluster): {
 					ip:           fakeServiceExternalIP1,
@@ -249,7 +250,8 @@ func TestCreateService(t *testing.T) {
 			expectError: false,
 		},
 		{
-			name: "new Service created with ExternalTrafficPolicy=Local and local Node selected",
+			name:                     "new Service created with ExternalTrafficPolicy=Local and local Node selected",
+			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
 			existingEndpoints: []*corev1.Endpoints{
 				makeEndpoints(servicePolicyLocal.Name, servicePolicyLocal.Namespace,
 					map[string]string{
@@ -271,7 +273,8 @@ func TestCreateService(t *testing.T) {
 			expectError: false,
 		},
 		{
-			name: "new Service created with ExternalTrafficPolicy=Local and local Node not Selected",
+			name:                     "new Service created with ExternalTrafficPolicy=Local and local Node not Selected",
+			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
 			existingEndpoints: []*corev1.Endpoints{
 				makeEndpoints(servicePolicyLocal.Name, servicePolicyLocal.Namespace,
 					map[string]string{
@@ -287,9 +290,7 @@ func TestCreateService(t *testing.T) {
 			healthyNodes:    []string{fakeNode1, fakeNode2},
 			// invert the sorted string to select other Nodes.
 			overrideHashFn: fakeHashFn(true),
-			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
-				mockIPAssigner.EXPECT().UnassignIP(fakeServiceExternalIP1)
-			},
+			expectedCalls:  func(mockIPAssigner *ipassignertest.MockIPAssigner) {},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
 				keyFor(servicePolicyLocal): {
 					ip:           fakeServiceExternalIP1,
@@ -299,7 +300,8 @@ func TestCreateService(t *testing.T) {
 			expectError: false,
 		},
 		{
-			name: "new Service created with ExternalTrafficPolicy=Local and local Node has no healthy endpoints",
+			name:                     "new Service created with ExternalTrafficPolicy=Local and local Node has no healthy endpoints",
+			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
 			existingEndpoints: []*corev1.Endpoints{
 				makeEndpoints(servicePolicyLocal.Name, servicePolicyLocal.Namespace,
 					map[string]string{
@@ -311,9 +313,7 @@ func TestCreateService(t *testing.T) {
 			},
 			serviceToCreate: servicePolicyLocal,
 			healthyNodes:    []string{fakeNode1, fakeNode2},
-			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
-				mockIPAssigner.EXPECT().UnassignIP(fakeServiceExternalIP1)
-			},
+			expectedCalls:   func(mockIPAssigner *ipassignertest.MockIPAssigner) {},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
 				keyFor(servicePolicyLocal): {
 					ip:           fakeServiceExternalIP1,
@@ -323,7 +323,8 @@ func TestCreateService(t *testing.T) {
 			expectError: false,
 		},
 		{
-			name: "new Service created with ExternalTrafficPolicy=Local and no Nodes has healthy endpoints",
+			name:                     "new Service created with ExternalTrafficPolicy=Local and no Nodes has healthy endpoints",
+			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
 			existingEndpoints: []*corev1.Endpoints{
 				makeEndpoints(servicePolicyLocal.Name, servicePolicyLocal.Namespace,
 					nil,
@@ -332,13 +333,33 @@ func TestCreateService(t *testing.T) {
 						"2.3.4.6": fakeNode2,
 					}),
 			},
-			serviceToCreate: servicePolicyLocal,
-			healthyNodes:    []string{fakeNode1, fakeNode2},
-			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
-			},
+			serviceToCreate:          servicePolicyLocal,
+			healthyNodes:             []string{fakeNode1, fakeNode2},
+			expectedCalls:            func(mockIPAssigner *ipassignertest.MockIPAssigner) {},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
 			expectError:              true,
 		},
+		{
+			name:              "new Service created and local Node selected and IP already assigned by other Service",
+			existingEndpoints: nil,
+			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
+				keyFor(servicePolicyLocal): {fakeServiceExternalIP1, fakeNode1},
+			},
+			serviceToCreate: servicePolicyCluster,
+			healthyNodes:    []string{fakeNode1, fakeNode2},
+			expectedCalls:   func(mockIPAssigner *ipassignertest.MockIPAssigner) {},
+			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
+				keyFor(servicePolicyLocal): {
+					ip:           fakeServiceExternalIP1,
+					assignedNode: fakeNode1,
+				},
+				keyFor(servicePolicyCluster): {
+					ip:           fakeServiceExternalIP1,
+					assignedNode: fakeNode1,
+				},
+			},
+			expectError: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -357,6 +378,13 @@ func TestCreateService(t *testing.T) {
 			if tt.overrideHashFn != nil {
 				c.fakeMemberlistCluster.hashFn = tt.overrideHashFn
 			}
+			c.externalIPStates = tt.previousExternalIPStates
+			for service, state := range tt.previousExternalIPStates {
+				if c.assignedIPs[state.ip] == nil {
+					c.assignedIPs[state.ip] = sets.NewString()
+				}
+				c.assignedIPs[state.ip].Insert(service.String())
+			}
 			tt.expectedCalls(c.mockIPAssigner)
 			err := c.syncService(keyFor(tt.serviceToCreate))
 			if tt.expectError {
@@ -370,11 +398,15 @@ func TestCreateService(t *testing.T) {
 }
 
 func TestUpdateService(t *testing.T) {
-	serviceExternlTrafficPolicyClusterUpdatedExternalIP := servicePolicyCluster.DeepCopy()
-	serviceExternlTrafficPolicyClusterUpdatedExternalIP.Status.LoadBalancer.Ingress[0].IP = fakeServiceExternalIP2
+	serviceExternalTrafficPolicyClusterUpdatedExternalIP := servicePolicyCluster.DeepCopy()
+	serviceExternalTrafficPolicyClusterUpdatedExternalIP.Status.LoadBalancer.Ingress[0].IP = fakeServiceExternalIP2
+
+	serviceExternalTrafficPolicyClusterWithSameExternalIP := servicePolicyCluster.DeepCopy()
+	serviceExternalTrafficPolicyClusterWithSameExternalIP.Name = "svc-same-eip"
+	serviceExternalTrafficPolicyClusterWithSameExternalIP.Status.LoadBalancer.Ingress[0].IP = fakeServiceExternalIP2
 
-	serviceExternalTrafficLocalWithNodeSelectd := servicePolicyLocal.DeepCopy()
-	serviceExternalTrafficLocalWithNodeSelectd.Status.LoadBalancer.Ingress[0].Hostname = fakeNode1
+	serviceExternalTrafficLocalWithNodeSelected := servicePolicyLocal.DeepCopy()
+	serviceExternalTrafficLocalWithNodeSelected.Status.LoadBalancer.Ingress[0].Hostname = fakeNode1
 
 	serviceExternalTrafficLocalUpdatedHostname := servicePolicyLocal.DeepCopy()
 	serviceExternalTrafficLocalUpdatedHostname.Status.LoadBalancer.Ingress[0].Hostname = fakeNode2
@@ -402,12 +434,12 @@ func TestUpdateService(t *testing.T) {
 		{
 			name:            "Service updated external IP and local Node selected",
 			endpoints:       nil,
-			serviceToUpdate: serviceExternlTrafficPolicyClusterUpdatedExternalIP,
+			serviceToUpdate: serviceExternalTrafficPolicyClusterUpdatedExternalIP,
 			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
-				keyFor(serviceExternlTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP1, fakeNode1},
+				keyFor(serviceExternalTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP1, fakeNode1},
 			},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
-				keyFor(serviceExternlTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP2, fakeNode1},
+				keyFor(serviceExternalTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP2, fakeNode1},
 			},
 			healthyNodes: []string{fakeNode1, fakeNode2},
 			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
@@ -416,21 +448,38 @@ func TestUpdateService(t *testing.T) {
 			},
 			expectError: false,
 		},
+		{
+			name:            "Service updated external IP and local Node selected but other Service still owns the assigned IP",
+			endpoints:       nil,
+			serviceToUpdate: serviceExternalTrafficPolicyClusterUpdatedExternalIP,
+			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
+				keyFor(serviceExternalTrafficPolicyClusterWithSameExternalIP): {fakeServiceExternalIP1, fakeNode1},
+				keyFor(serviceExternalTrafficPolicyClusterUpdatedExternalIP):  {fakeServiceExternalIP1, fakeNode1},
+			},
+			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
+				keyFor(serviceExternalTrafficPolicyClusterWithSameExternalIP): {fakeServiceExternalIP1, fakeNode1},
+				keyFor(serviceExternalTrafficPolicyClusterUpdatedExternalIP):  {fakeServiceExternalIP2, fakeNode1},
+			},
+			healthyNodes: []string{fakeNode1, fakeNode2},
+			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
+				mockIPAssigner.EXPECT().AssignIP(fakeServiceExternalIP2)
+			},
+			expectError: false,
+		},
 		{
 			name:            "Service updated external IP and local Node not selected",
 			endpoints:       nil,
-			serviceToUpdate: serviceExternlTrafficPolicyClusterUpdatedExternalIP,
+			serviceToUpdate: serviceExternalTrafficPolicyClusterUpdatedExternalIP,
 			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
-				keyFor(serviceExternlTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP1, fakeNode1},
+				keyFor(serviceExternalTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP1, fakeNode1},
 			},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
-				keyFor(serviceExternlTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP2, fakeNode2},
+				keyFor(serviceExternalTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP2, fakeNode2},
 			},
 			healthyNodes:   []string{fakeNode1, fakeNode2},
 			overrideHashFn: fakeHashFn(true),
 			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockIPAssigner.EXPECT().UnassignIP(fakeServiceExternalIP1)
-				mockIPAssigner.EXPECT().UnassignIP(fakeServiceExternalIP2)
 			},
 			expectError: false,
 		},
@@ -439,7 +488,7 @@ func TestUpdateService(t *testing.T) {
 			endpoints:       nil,
 			serviceToUpdate: serviceChangedType,
 			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
-				keyFor(serviceExternlTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP1, fakeNode1},
+				keyFor(serviceExternalTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP1, fakeNode1},
 			},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
 			healthyNodes:             []string{fakeNode1, fakeNode2},
@@ -453,7 +502,7 @@ func TestUpdateService(t *testing.T) {
 			endpoints:       nil,
 			serviceToUpdate: serviceChangedType,
 			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
-				keyFor(serviceExternlTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP1, fakeNode1},
+				keyFor(serviceExternalTrafficPolicyClusterUpdatedExternalIP): {fakeServiceExternalIP1, fakeNode1},
 			},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
 			healthyNodes:             []string{fakeNode1, fakeNode2},
@@ -503,7 +552,7 @@ func TestUpdateService(t *testing.T) {
 			expectError: false,
 		},
 		{
-			name: "local Node no longer have healy endpoints",
+			name: "local Node no longer have healthy endpoints",
 			endpoints: []*corev1.Endpoints{
 				makeEndpoints(servicePolicyLocal.Name, servicePolicyLocal.Namespace,
 					map[string]string{
@@ -512,7 +561,7 @@ func TestUpdateService(t *testing.T) {
 						"2.3.4.5": fakeNode1,
 					}),
 			},
-			serviceToUpdate: serviceExternalTrafficLocalWithNodeSelectd,
+			serviceToUpdate: serviceExternalTrafficLocalWithNodeSelected,
 			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
 				keyFor(servicePolicyLocal): {fakeServiceExternalIP1, fakeNode1},
 			},
@@ -537,18 +586,16 @@ func TestUpdateService(t *testing.T) {
 				),
 			},
 			overrideHashFn:  fakeHashFn(true),
-			serviceToUpdate: serviceExternalTrafficLocalWithNodeSelectd,
+			serviceToUpdate: serviceExternalTrafficLocalWithNodeSelected,
 			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
 				keyFor(servicePolicyLocal): {fakeServiceExternalIP1, fakeNode1},
 			},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
 				keyFor(servicePolicyLocal): {fakeServiceExternalIP1, fakeNode1},
 			},
-			healthyNodes: []string{fakeNode1, fakeNode2},
-			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
-				mockIPAssigner.EXPECT().AssignIP(fakeServiceExternalIP1)
-			},
-			expectError: false,
+			healthyNodes:  []string{fakeNode1, fakeNode2},
+			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {},
+			expectError:   false,
 		},
 		{
 			name: "other Node could promote itself as the new owner",
@@ -586,7 +633,7 @@ func TestUpdateService(t *testing.T) {
 				),
 			},
 			overrideHashFn:           fakeHashFn(true),
-			serviceToUpdate:          serviceExternalTrafficLocalWithNodeSelectd,
+			serviceToUpdate:          serviceExternalTrafficLocalWithNodeSelected,
 			previousExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{},
 			expectedExternalIPStates: map[apimachinerytypes.NamespacedName]externalIPState{
 				keyFor(servicePolicyLocal): {fakeServiceExternalIP1, fakeNode1},
@@ -616,6 +663,12 @@ func TestUpdateService(t *testing.T) {
 			if tt.overrideHashFn != nil {
 				c.fakeMemberlistCluster.hashFn = tt.overrideHashFn
 			}
+			for service, state := range tt.previousExternalIPStates {
+				if c.assignedIPs[state.ip] == nil {
+					c.assignedIPs[state.ip] = sets.NewString()
+				}
+				c.assignedIPs[state.ip].Insert(service.String())
+			}
 			tt.expectedCalls(c.mockIPAssigner)
 			err := c.syncService(keyFor(tt.serviceToUpdate))
 			if tt.expectError {
@@ -628,62 +681,6 @@ func TestUpdateService(t *testing.T) {
 	}
 }
 
-func TestStaleServiceExternalIPRemoval(t *testing.T) {
-	service3 := servicePolicyCluster.DeepCopy()
-	service3.Name = "svc3"
-	service3.Status.LoadBalancer.Ingress = []corev1.LoadBalancerIngress{
-		{IP: fakeServiceExternalIP2},
-	}
-	tests := []struct {
-		name             string
-		existingServices []*corev1.Service
-		assignedIPs      []string
-		expectedCalls    func(mockIPAssigner *ipassignertest.MockIPAssigner)
-	}{
-		{
-			name:             "should keep assigned IPs if corresponding Services are present",
-			existingServices: []*corev1.Service{servicePolicyCluster, service3},
-			assignedIPs:      []string{fakeNode1, fakeNode2},
-			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
-				mockIPAssigner.EXPECT().AssignedIPs().DoAndReturn(
-					func() interface{} {
-						return sets.NewString(fakeServiceExternalIP1, fakeServiceExternalIP2)
-					},
-				)
-			},
-		},
-		{
-			name:             "should cleanup stale assigned IPs",
-			existingServices: []*corev1.Service{servicePolicyCluster},
-			assignedIPs:      []string{fakeNode1, fakeNode2},
-			expectedCalls: func(mockIPAssigner *ipassignertest.MockIPAssigner) {
-				mockIPAssigner.EXPECT().AssignedIPs().DoAndReturn(
-					func() interface{} {
-						return sets.NewString(fakeServiceExternalIP1, fakeServiceExternalIP2)
-					},
-				)
-				mockIPAssigner.EXPECT().UnassignIP(fakeServiceExternalIP2)
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			objs := []runtime.Object{}
-			for _, s := range tt.existingServices {
-				objs = append(objs, s)
-			}
-			c := newFakeController(t, objs...)
-			defer c.mockController.Finish()
-			stopCh := make(chan struct{})
-			defer close(stopCh)
-			c.informerFactory.Start(stopCh)
-			c.informerFactory.WaitForCacheSync(stopCh)
-			tt.expectedCalls(c.mockIPAssigner)
-			c.removeStaleExternalIPs()
-		})
-	}
-}
-
 func keyFor(svc *corev1.Service) apimachinerytypes.NamespacedName {
 	return apimachinerytypes.NamespacedName{
 		Namespace: svc.Namespace,
diff --git a/pkg/agent/ipassigner/responder/arp_responder_test.go b/pkg/agent/ipassigner/responder/arp_responder_test.go
new file mode 100644
index 000000000..02988a585
--- /dev/null
+++ b/pkg/agent/ipassigner/responder/arp_responder_test.go
@@ -0,0 +1,320 @@
+// Copyright 2022 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package responder
+
+import (
+	"bytes"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/mdlayher/arp"
+	"github.com/mdlayher/ethernet"
+	"github.com/mdlayher/raw"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+type fakePacketConn struct {
+	addr   raw.Addr
+	buffer *bytes.Buffer
+}
+
+var _ net.PacketConn = (*fakePacketConn)(nil)
+
+func (pc *fakePacketConn) ReadFrom(p []byte) (int, net.Addr, error) {
+	n, err := pc.buffer.Read(p)
+	return n, &pc.addr, err
+}
+
+func (pc *fakePacketConn) WriteTo(p []byte, addr net.Addr) (int, error) {
+	return pc.buffer.Write(p)
+}
+
+func (pc *fakePacketConn) Close() error {
+	return nil
+}
+
+func (pc *fakePacketConn) LocalAddr() net.Addr {
+	return &pc.addr
+}
+
+func (pc *fakePacketConn) SetDeadline(t time.Time) error {
+	return nil
+}
+
+func (pc *fakePacketConn) SetReadDeadline(t time.Time) error {
+	return nil
+}
+
+func (pc *fakePacketConn) SetWriteDeadline(t time.Time) error {
+	return nil
+}
+
+func newFakeARPClient(iface *net.Interface, conn *fakePacketConn) (*arp.Client, error) {
+	return arp.New(iface, conn)
+}
+
+func newFakeNetworkInterface() *net.Interface {
+	return &net.Interface{
+		Index:        0,
+		MTU:          1500,
+		Name:         "eth0",
+		HardwareAddr: []byte{0x00, 0x11, 0x22, 0x33, 0x44, 0x55},
+	}
+}
+
+func TestARPResponder_Advertise(t *testing.T) {
+	tests := []struct {
+		name          string
+		iface         *net.Interface
+		ip            net.IP
+		expectError   bool
+		expectedBytes []byte
+	}{
+		{
+			name:        "GratuitousARP for IPv4",
+			iface:       newFakeNetworkInterface(),
+			ip:          net.ParseIP("192.168.10.1").To4(),
+			expectError: false,
+			expectedBytes: []byte{
+				// ethernet header (16 bytes)
+				0xff, 0xff, 0xff, 0xff, 0xff, 0xff, // 6 bytes: destination hardware address
+				0x00, 0x11, 0x22, 0x33, 0x44, 0x55, // 6 bytes: source hardware address
+				0x08, 0x06, // 2 bytes: ethernet type
+				// arp payload (46 bytes)
+				0x00, 0x01, // 2 bytes: hardware type
+				0x08, 0x00, // 2 bytes: protocol type
+				0x06,       // 1 byte : hardware address length
+				0x04,       // 1 byte : protocol length
+				0x00, 0x01, // 2 bytes: operation
+				0x00, 0x11, 0x22, 0x33, 0x44, 0x55, // 6 bytes: source hardware address
+				0xc0, 0xa8, 0x0a, 0x01, // 4 bytes: source protocol address
+				0xff, 0xff, 0xff, 0xff, 0xff, 0xff, // 6 bytes: target hardware address
+				0xc0, 0xa8, 0x0a, 0x01, // 4 bytes: target protocol address
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // 18 bytes: padding
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			conn := &fakePacketConn{
+				buffer: bytes.NewBuffer(nil),
+				addr: raw.Addr{
+					HardwareAddr: tt.iface.HardwareAddr,
+				},
+			}
+			fakeARPClient, err := newFakeARPClient(tt.iface, conn)
+			require.NoError(t, err)
+
+			r := arpResponder{
+				iface: tt.iface,
+				conn:  fakeARPClient,
+			}
+			err = r.advertise(tt.ip)
+			if tt.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedBytes, conn.buffer.Bytes())
+			}
+		})
+	}
+}
+
+func TestARPResponder_HandleARPRequest(t *testing.T) {
+	tests := []struct {
+		name                 string
+		iface                *net.Interface
+		arpOperation         arp.Operation
+		srcHWAddr, dstHWAddr net.HardwareAddr
+		srcIP, dstIP         net.IP
+		assignedIPs          []net.IP
+		expectError          bool
+		expectedBytes        []byte
+	}{
+		{
+			name:         "Response for assigned IP",
+			iface:        newFakeNetworkInterface(),
+			arpOperation: arp.OperationRequest,
+			srcHWAddr:    net.HardwareAddr{0x00, 0x01, 0x02, 0x03, 0x04, 0x05},
+			dstHWAddr:    ethernet.Broadcast,
+			srcIP:        net.ParseIP("192.168.10.2"),
+			dstIP:        net.ParseIP("192.168.10.1"),
+			assignedIPs:  []net.IP{net.ParseIP("192.168.10.1")},
+			expectError:  false,
+			expectedBytes: []byte{
+				// ethernet header (16 bytes)
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, // 6 bytes: destination hardware address
+				0x00, 0x11, 0x22, 0x33, 0x44, 0x55, // 6 bytes: source hardware address
+				0x08, 0x06, // 2 bytes: ethernet type
+				// arp payload (46 bytes)
+				0x00, 0x01, // 2 bytes: hardware type
+				0x08, 0x00, // 2 bytes: protocol type
+				0x06,       // 1 byte : hardware address length
+				0x04,       // 1 byte : protocol length
+				0x00, 0x02, // 2 bytes: operation
+				0x00, 0x11, 0x22, 0x33, 0x44, 0x55, // 6 bytes: source hardware address
+				0xc0, 0xa8, 0x0a, 0x01, // 4 bytes: source protocol address
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, // 6 bytes: target hardware address
+				0xc0, 0xa8, 0x0a, 0x02, // 4 bytes: target protocol address
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // 18 bytes: padding
+			},
+		},
+		{
+			name:          "Response for not assigned IP",
+			iface:         newFakeNetworkInterface(),
+			arpOperation:  arp.OperationRequest,
+			srcHWAddr:     net.HardwareAddr{0x00, 0x01, 0x02, 0x03, 0x04, 0x05},
+			dstHWAddr:     ethernet.Broadcast,
+			srcIP:         net.ParseIP("192.168.10.2"),
+			dstIP:         net.ParseIP("192.168.10.3"),
+			assignedIPs:   []net.IP{net.ParseIP("192.168.10.1")},
+			expectError:   false,
+			expectedBytes: []byte{},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			conn := &fakePacketConn{
+				buffer: bytes.NewBuffer(nil),
+				addr: raw.Addr{
+					HardwareAddr: tt.iface.HardwareAddr,
+				},
+			}
+			fakeARPClient, err := newFakeARPClient(tt.iface, conn)
+			require.NoError(t, err)
+			packet, err := arp.NewPacket(tt.arpOperation, tt.srcHWAddr, tt.srcIP, tt.dstHWAddr, tt.dstIP)
+			require.NoError(t, err)
+			err = fakeARPClient.WriteTo(packet, tt.dstHWAddr)
+			require.NoError(t, err)
+			assignedIPs := sets.NewString()
+			for _, ip := range tt.assignedIPs {
+				assignedIPs.Insert(ip.String())
+			}
+			r := arpResponder{
+				iface:       tt.iface,
+				conn:        fakeARPClient,
+				assignedIPs: sets.NewString(),
+			}
+			for _, ip := range tt.assignedIPs {
+				r.AddIP(ip)
+			}
+			err = r.handleARPRequest()
+			assert.NoError(t, err)
+			assert.Equal(t, tt.expectedBytes, conn.buffer.Bytes())
+		})
+	}
+}
+
+func Test_arpResponder_addIP(t *testing.T) {
+	tests := []struct {
+		name                string
+		ip                  net.IP
+		assignedIPs         sets.String
+		expectedError       bool
+		expectedAssignedIPs sets.String
+	}{
+		{
+			name:                "Add new IP",
+			ip:                  net.ParseIP("2.0.2.2"),
+			assignedIPs:         sets.NewString(),
+			expectedAssignedIPs: sets.NewString("2.0.2.2"),
+		},
+		{
+			name:                "Add new IP with some IPs added",
+			ip:                  net.ParseIP("2.0.2.2"),
+			assignedIPs:         sets.NewString("2.0.2.1"),
+			expectedAssignedIPs: sets.NewString("2.0.2.1", "2.0.2.2"),
+		},
+		{
+			name:                "Add invalid IP",
+			ip:                  net.ParseIP("2022::abcd"),
+			assignedIPs:         sets.NewString("2.0.2.1"),
+			expectedAssignedIPs: sets.NewString("2.0.2.1"),
+			expectedError:       true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := &arpResponder{
+				iface:       newFakeNetworkInterface(),
+				assignedIPs: tt.assignedIPs,
+			}
+			conn := &fakePacketConn{
+				buffer: bytes.NewBuffer(nil),
+				addr: raw.Addr{
+					HardwareAddr: r.iface.HardwareAddr,
+				},
+			}
+			fakeARPClient, err := newFakeARPClient(r.iface, conn)
+			require.NoError(t, err)
+			r.conn = fakeARPClient
+			err = r.AddIP(tt.ip)
+			if tt.expectedError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+			assert.Equal(t, tt.expectedAssignedIPs, r.assignedIPs)
+		})
+	}
+}
+
+func Test_arpResponder_removeIP(t *testing.T) {
+	tests := []struct {
+		name                string
+		ip                  net.IP
+		assignedIPs         sets.String
+		expectedError       bool
+		expectedAssignedIPs sets.String
+	}{
+		{
+			name:                "Remove existing IP",
+			ip:                  net.ParseIP("2.0.2.2"),
+			assignedIPs:         sets.NewString("2.0.2.2"),
+			expectedAssignedIPs: sets.NewString(),
+		},
+		{
+			name:                "Remove non-existent IP",
+			ip:                  net.ParseIP("2.0.2.2"),
+			assignedIPs:         sets.NewString("2.0.2.1"),
+			expectedAssignedIPs: sets.NewString("2.0.2.1"),
+		},
+		{
+			name:                "Remove invalid IP",
+			ip:                  net.ParseIP("2022::abcd"),
+			assignedIPs:         sets.NewString("2.0.2.1"),
+			expectedAssignedIPs: sets.NewString("2.0.2.1"),
+			expectedError:       true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := &arpResponder{
+				iface:       newFakeNetworkInterface(),
+				assignedIPs: tt.assignedIPs,
+			}
+			err := r.RemoveIP(tt.ip)
+			if tt.expectedError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+			assert.Equal(t, tt.expectedAssignedIPs, r.assignedIPs)
+		})
+	}
+}
diff --git a/pkg/agent/ipassigner/responder/ndp_responder_test.go b/pkg/agent/ipassigner/responder/ndp_responder_test.go
new file mode 100644
index 000000000..40859c2b5
--- /dev/null
+++ b/pkg/agent/ipassigner/responder/ndp_responder_test.go
@@ -0,0 +1,435 @@
+// Copyright 2022 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package responder
+
+import (
+	"bytes"
+	"net"
+	"testing"
+
+	"github.com/mdlayher/ndp"
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/net/ipv6"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+type fakeNDPConn struct {
+	readFrom   func() (ndp.Message, *ipv6.ControlMessage, net.IP, error)
+	writeTo    func(ndp.Message, *ipv6.ControlMessage, net.IP) error
+	joinGroup  func(ip net.IP) error
+	leaveGroup func(ip net.IP) error
+}
+
+func (c *fakeNDPConn) ReadFrom() (ndp.Message, *ipv6.ControlMessage, net.IP, error) {
+	return c.readFrom()
+}
+
+func (c *fakeNDPConn) WriteTo(message ndp.Message, cm *ipv6.ControlMessage, dstIP net.IP) error {
+	return c.writeTo(message, cm, dstIP)
+}
+
+func (c *fakeNDPConn) Close() error {
+	return nil
+}
+
+func (c *fakeNDPConn) JoinGroup(ip net.IP) error {
+	return c.joinGroup(ip)
+}
+
+func (c *fakeNDPConn) LeaveGroup(ip net.IP) error {
+	return c.leaveGroup(ip)
+}
+
+func TestNDPResponder_Advertise(t *testing.T) {
+	buffer := bytes.NewBuffer(nil)
+	fakeConn := &fakeNDPConn{
+		writeTo: func(msg ndp.Message, _ *ipv6.ControlMessage, _ net.IP) error {
+			bs, err := ndp.MarshalMessage(msg)
+			assert.NoError(t, err)
+			buffer.Write(bs)
+			return nil
+		},
+	}
+	responder := &ndpResponder{
+		iface: newFakeNetworkInterface(),
+		conn:  fakeConn,
+	}
+	err := responder.advertise(net.ParseIP("fe80::250:56ff:fea7:e29d"))
+	assert.NoError(t, err)
+	//  Neighbor Advertisement Message Format - RFC 4861 Section 4.4.
+	//   0                   1                   2                   3
+	//   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+	//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+	//  |     Type      |     Code      |          Checksum             |
+	//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+	//  |R|S|O|                     Reserved                            |
+	//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+	//  |                                                               |
+	//  +                                                               +
+	//  |                                                               |
+	//  +                       Target Address                          +
+	//  |                                                               |
+	//  +                                                               +
+	//  |                                                               |
+	//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+	//  |   Options ...
+	//  +-+-+-+-+-+-+-+-+-+-+-+-
+	//
+	//  Options formats - Source/Target Link-layer Address. RFC 4861 Section 4.6.1.
+	//   0                   1                   2                   3
+	//   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+	//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+	//  |     Type      |    Length     |    Link-Layer Address ...
+	//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+	expectedBytes := []byte{
+		0x88,       // type - 136 for Neighbor Advertisement
+		0x00,       // code
+		0x00, 0x00, // checksum
+		0x20, 0x00, 0x00, 0x00, // flags and reserved bits. Override bit is set.
+		0xfe, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x50, 0x56, 0xff, 0xfe, 0xa7, 0xe2, 0x9d, // IPv6 address
+		0x02,                               // option - 2 for Target Link-layer Address
+		0x01,                               // length (units of 8 octets including type and length fields)
+		0x00, 0x11, 0x22, 0x33, 0x44, 0x55, // hardware address
+	}
+	assert.Equal(t, expectedBytes, buffer.Bytes())
+}
+
+func TestNDPResponder_handleNeighborSolicitation(t *testing.T) {
+	tests := []struct {
+		name           string
+		requestMessage []byte
+		requestIP      net.IP
+		assignedIPs    []net.IP
+		expectError    bool
+		expectedReply  []byte
+	}{
+		{
+			name: "request to assigned IP",
+			requestMessage: []byte{
+				0x87,       // type - 135 for Neighbor Solicitation
+				0x00,       // code
+				0x00, 0x00, // checksum
+				0x00, 0x00, 0x00, 0x00, // reserved bits.
+				0xfe, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xa1, // IPv6 address
+				0x01,                               // option - 1 for Source Link-layer Address
+				0x01,                               // length (units of 8 octets including type and length fields)
+				0x00, 0x11, 0x22, 0x33, 0x44, 0x66, // hardware address
+			},
+			requestIP: net.ParseIP("fe80::c1"),
+			assignedIPs: []net.IP{
+				net.ParseIP("fe80::a1"),
+				net.ParseIP("fe80::a2"),
+			},
+			expectError: false,
+			expectedReply: []byte{
+				0x88,       // type - 136 for Neighbor Advertisement
+				0x00,       // code
+				0x00, 0x00, // checksum
+				0x60, 0x00, 0x00, 0x00, // flags and reserved bits. Solicited and Override bits are set.
+				0xfe, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xa1, // IPv6 address
+				0x02,                               // option - 2 for Target Link-layer Address
+				0x01,                               // length (units of 8 octets including type and length fields)
+				0x00, 0x11, 0x22, 0x33, 0x44, 0x55, // hardware address
+			},
+		},
+		{
+			name: "request to not assigned IP",
+			requestMessage: []byte{
+				0x87,       // type - 135 for Neighbor Solicitation
+				0x00,       // code
+				0x00, 0x00, // checksum
+				0x00, 0x00, 0x00, 0x00, // reserved bits.
+				0xfe, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xa3, // IPv6 address
+				0x01,                               // option - 1 for Source Link-layer Address
+				0x01,                               // length (units of 8 octets including type and length fields)
+				0x00, 0x11, 0x22, 0x33, 0x44, 0x66, // hardware address
+			},
+			requestIP: net.ParseIP("fe80::c1"),
+			assignedIPs: []net.IP{
+				net.ParseIP("fe80::a1"),
+				net.ParseIP("fe80::a2"),
+			},
+			expectError:   false,
+			expectedReply: nil,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			buffer := bytes.NewBuffer(nil)
+			fakeConn := &fakeNDPConn{
+				writeTo: func(msg ndp.Message, _ *ipv6.ControlMessage, _ net.IP) error {
+					bs, err := ndp.MarshalMessage(msg)
+					assert.NoError(t, err)
+					buffer.Write(bs)
+					return nil
+				},
+				readFrom: func() (ndp.Message, *ipv6.ControlMessage, net.IP, error) {
+					msg, err := ndp.ParseMessage(tt.requestMessage)
+					return msg, nil, tt.requestIP, err
+				},
+			}
+			assignedIPs := sets.NewString()
+			for _, ip := range tt.assignedIPs {
+				assignedIPs.Insert(ip.String())
+			}
+			responder := &ndpResponder{
+				iface:       newFakeNetworkInterface(),
+				conn:        fakeConn,
+				assignedIPs: sets.NewString(),
+			}
+			for _, ip := range tt.assignedIPs {
+				responder.assignedIPs[ip.String()] = struct{}{}
+			}
+			err := responder.handleNeighborSolicitation()
+			if tt.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expectedReply, buffer.Bytes())
+			}
+		})
+	}
+}
+
+func Test_parseIPv6SolicitedNodeMulticastAddress(t *testing.T) {
+	tests := []struct {
+		name          string
+		ip            net.IP
+		expectedGroup net.IP
+		expectedKey   int
+	}{
+		{
+			name:          "global unicast IPv6 address 1",
+			ip:            net.ParseIP("2022:abcd::11:1111"),
+			expectedGroup: net.ParseIP("ff02::1:ff11:1111"),
+			expectedKey:   0x111111,
+		},
+		{
+			name:          "global unicast IPv6 address 2",
+			ip:            net.ParseIP("2022:ffff::1234:5678"),
+			expectedGroup: net.ParseIP("ff02::1:ff34:5678"),
+			expectedKey:   0x345678,
+		},
+		{
+			name:          "link-local unicast IPv6 address",
+			ip:            net.ParseIP("fe80::1122:3344"),
+			expectedGroup: net.ParseIP("ff02::1:ff22:3344"),
+			expectedKey:   0x223344,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			group, key := parseIPv6SolicitedNodeMulticastAddress(tt.ip)
+			assert.Equal(t, tt.expectedGroup, group)
+			assert.Equal(t, tt.expectedKey, key)
+		})
+	}
+}
+
+func Test_ndpResponder_addIP(t *testing.T) {
+	tests := []struct {
+		name                    string
+		ip                      net.IP
+		conn                    ndpConn
+		assignedIPs             sets.String
+		multicastGroups         map[int]int
+		expectedJoinedGroups    []net.IP
+		expectedLeftGroups      []net.IP
+		expectedMulticastGroups map[int]int
+		expectedAssigndIPs      sets.String
+		expectedError           bool
+	}{
+		{
+			name:                 "Add new IP from a new multicast group 1",
+			ip:                   net.ParseIP("2022::beaf"),
+			assignedIPs:          sets.NewString(),
+			multicastGroups:      map[int]int{},
+			expectedJoinedGroups: []net.IP{net.ParseIP("ff02::1:ff00:beaf")},
+			expectedLeftGroups:   nil,
+			expectedMulticastGroups: map[int]int{
+				0xbeaf: 1,
+			},
+			expectedAssigndIPs: sets.NewString("2022::beaf"),
+		},
+		{
+			name:                 "Add new IP from a new multicast group 2",
+			ip:                   net.ParseIP("2022::beaf:beaf"),
+			assignedIPs:          sets.NewString(),
+			multicastGroups:      map[int]int{},
+			expectedJoinedGroups: []net.IP{net.ParseIP("ff02::1:ffaf:beaf")},
+			expectedLeftGroups:   nil,
+			expectedMulticastGroups: map[int]int{
+				0xafbeaf: 1,
+			},
+			expectedAssigndIPs: sets.NewString("2022::beaf:beaf"),
+		},
+		{
+			name:        "Add new IP from an existing multicast group",
+			ip:          net.ParseIP("2021::beaf"),
+			assignedIPs: sets.NewString("2022::beaf"),
+			multicastGroups: map[int]int{
+				0xbeaf: 1,
+			},
+			expectedJoinedGroups: nil,
+			expectedLeftGroups:   nil,
+			expectedMulticastGroups: map[int]int{
+				0xbeaf: 2,
+			},
+			expectedAssigndIPs: sets.NewString("2021::beaf", "2022::beaf"),
+		},
+		{
+			name:                    "Add invalid IP",
+			ip:                      net.ParseIP("1.2.3.4"),
+			assignedIPs:             sets.NewString(),
+			multicastGroups:         map[int]int{},
+			expectedError:           true,
+			expectedMulticastGroups: map[int]int{},
+			expectedAssigndIPs:      sets.NewString(),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var joinedGroup, leftGroup []net.IP
+			r := &ndpResponder{
+				iface: newFakeNetworkInterface(),
+				conn: &fakeNDPConn{
+					joinGroup: func(ip net.IP) error {
+						joinedGroup = append(joinedGroup, ip)
+						return nil
+					},
+					leaveGroup: func(ip net.IP) error {
+						leftGroup = append(leftGroup, ip)
+						return nil
+					},
+					writeTo: func(_ ndp.Message, _ *ipv6.ControlMessage, _ net.IP) error {
+						return nil
+					},
+				},
+				assignedIPs:     tt.assignedIPs,
+				multicastGroups: tt.multicastGroups,
+			}
+			err := r.AddIP(tt.ip)
+			if tt.expectedError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+			assert.Equal(t, tt.expectedJoinedGroups, joinedGroup)
+			assert.Equal(t, tt.expectedLeftGroups, leftGroup)
+			assert.Equal(t, tt.expectedAssigndIPs, r.assignedIPs)
+			assert.Equal(t, tt.expectedMulticastGroups, r.multicastGroups)
+		})
+	}
+}
+
+func Test_ndpResponder_removeIP(t *testing.T) {
+	tests := []struct {
+		name                    string
+		ip                      net.IP
+		conn                    ndpConn
+		assignedIPs             sets.String
+		multicastGroups         map[int]int
+		expectedJoinedGroups    []net.IP
+		expectedLeftGroups      []net.IP
+		expectedMulticastGroups map[int]int
+		expectedAssigndIPs      sets.String
+		expectedError           bool
+	}{
+		{
+			name:        "Remove IP and leave multicast group",
+			ip:          net.ParseIP("2022::beaf"),
+			assignedIPs: sets.NewString("2022::beaf"),
+			multicastGroups: map[int]int{
+				0xbeaf: 1,
+			},
+			expectedJoinedGroups:    nil,
+			expectedLeftGroups:      []net.IP{net.ParseIP("ff02::1:ff00:beaf")},
+			expectedMulticastGroups: map[int]int{},
+			expectedAssigndIPs:      sets.NewString(),
+		},
+		{
+			name:        "Remove IP and should not leave multicast group",
+			ip:          net.ParseIP("2022::beaf"),
+			assignedIPs: sets.NewString("2022::beaf", "2021::beaf"),
+			multicastGroups: map[int]int{
+				0xbeaf: 2,
+			},
+			expectedJoinedGroups: nil,
+			expectedLeftGroups:   nil,
+			expectedMulticastGroups: map[int]int{
+				0xbeaf: 1,
+			},
+			expectedAssigndIPs: sets.NewString("2021::beaf"),
+		},
+		{
+			name:        "Remove non-existent IP",
+			ip:          net.ParseIP("2022::beaf"),
+			assignedIPs: sets.NewString("2021::beaf"),
+			multicastGroups: map[int]int{
+				0xbeaf: 1,
+			},
+			expectedJoinedGroups: nil,
+			expectedLeftGroups:   nil,
+			expectedMulticastGroups: map[int]int{
+				0xbeaf: 1,
+			},
+			expectedAssigndIPs: sets.NewString("2021::beaf"),
+		}, {
+			name:        "Remove invalid IP",
+			ip:          net.ParseIP("1.2.3.4"),
+			assignedIPs: sets.NewString("2021::beaf"),
+			multicastGroups: map[int]int{
+				0xbeaf: 1,
+			},
+			expectedJoinedGroups: nil,
+			expectedLeftGroups:   nil,
+			expectedMulticastGroups: map[int]int{
+				0xbeaf: 1,
+			},
+			expectedAssigndIPs: sets.NewString("2021::beaf"),
+			expectedError:      true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var joinedGroup, leftGroup []net.IP
+			r := &ndpResponder{
+				iface: newFakeNetworkInterface(),
+				conn: &fakeNDPConn{
+					joinGroup: func(ip net.IP) error {
+						joinedGroup = append(joinedGroup, ip)
+						return nil
+					},
+					leaveGroup: func(ip net.IP) error {
+						leftGroup = append(leftGroup, ip)
+						return nil
+					},
+				},
+				assignedIPs:     tt.assignedIPs,
+				multicastGroups: tt.multicastGroups,
+			}
+			err := r.RemoveIP(tt.ip)
+			if tt.expectedError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+			assert.Equal(t, tt.expectedJoinedGroups, joinedGroup)
+			assert.Equal(t, tt.expectedLeftGroups, leftGroup)
+			assert.Equal(t, tt.expectedAssigndIPs, r.assignedIPs)
+			assert.Equal(t, tt.expectedMulticastGroups, r.multicastGroups)
+		})
+	}
+}
diff --git a/pkg/agent/ipassigner/testing/mock_ipassigner.go b/pkg/agent/ipassigner/testing/mock_ipassigner.go
index 4baecb05c..d52683951 100644
--- a/pkg/agent/ipassigner/testing/mock_ipassigner.go
+++ b/pkg/agent/ipassigner/testing/mock_ipassigner.go
@@ -76,6 +76,18 @@ func (mr *MockIPAssignerMockRecorder) AssignedIPs() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AssignedIPs", reflect.TypeOf((*MockIPAssigner)(nil).AssignedIPs))
 }
 
+// Run mocks base method
+func (m *MockIPAssigner) Run(arg0 <-chan struct{}) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "Run", arg0)
+}
+
+// Run indicates an expected call of Run
+func (mr *MockIPAssignerMockRecorder) Run(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockIPAssigner)(nil).Run), arg0)
+}
+
 // UnassignIP mocks base method
 func (m *MockIPAssigner) UnassignIP(arg0 string) error {
 	m.ctrl.T.Helper()
diff --git a/pkg/agent/openflow/testing/mock_openflow.go b/pkg/agent/openflow/testing/mock_openflow.go
index ab49be308..5ca27c7bc 100644
--- a/pkg/agent/openflow/testing/mock_openflow.go
+++ b/pkg/agent/openflow/testing/mock_openflow.go
@@ -1,4 +1,4 @@
-// Copyright 2021 Antrea Authors
+// Copyright 2022 Antrea Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -365,20 +365,6 @@ func (mr *MockClientMockRecorder) InstallGatewayFlows() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallGatewayFlows", reflect.TypeOf((*MockClient)(nil).InstallGatewayFlows))
 }
 
-// InstallLoadBalancerServiceFromOutsideFlows mocks base method
-func (m *MockClient) InstallLoadBalancerServiceFromOutsideFlows(arg0 net.IP, arg1 uint16, arg2 openflow.Protocol) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallLoadBalancerServiceFromOutsideFlows", arg0, arg1, arg2)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// InstallLoadBalancerServiceFromOutsideFlows indicates an expected call of InstallLoadBalancerServiceFromOutsideFlows
-func (mr *MockClientMockRecorder) InstallLoadBalancerServiceFromOutsideFlows(arg0, arg1, arg2 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallLoadBalancerServiceFromOutsideFlows", reflect.TypeOf((*MockClient)(nil).InstallLoadBalancerServiceFromOutsideFlows), arg0, arg1, arg2)
-}
-
 // InstallMulticastFlow mocks base method
 func (m *MockClient) InstallMulticastFlow(arg0 net.IP) error {
 	m.ctrl.T.Helper()
@@ -709,20 +695,6 @@ func (mr *MockClientMockRecorder) UninstallEndpointFlows(arg0, arg1 interface{})
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallEndpointFlows", reflect.TypeOf((*MockClient)(nil).UninstallEndpointFlows), arg0, arg1)
 }
 
-// UninstallLoadBalancerServiceFromOutsideFlows mocks base method
-func (m *MockClient) UninstallLoadBalancerServiceFromOutsideFlows(arg0 net.IP, arg1 uint16, arg2 openflow.Protocol) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UninstallLoadBalancerServiceFromOutsideFlows", arg0, arg1, arg2)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// UninstallLoadBalancerServiceFromOutsideFlows indicates an expected call of UninstallLoadBalancerServiceFromOutsideFlows
-func (mr *MockClientMockRecorder) UninstallLoadBalancerServiceFromOutsideFlows(arg0, arg1, arg2 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallLoadBalancerServiceFromOutsideFlows", reflect.TypeOf((*MockClient)(nil).UninstallLoadBalancerServiceFromOutsideFlows), arg0, arg1, arg2)
-}
-
 // UninstallMulticastFlow mocks base method
 func (m *MockClient) UninstallMulticastFlow(arg0 net.IP) error {
 	m.ctrl.T.Helper()
diff --git a/pkg/agent/proxy/proxier_test.go b/pkg/agent/proxy/proxier_test.go
index 78ea0c725..f5685e3cb 100644
--- a/pkg/agent/proxy/proxier_test.go
+++ b/pkg/agent/proxy/proxier_test.go
@@ -114,7 +114,7 @@ func withoutProxyLoadBalancerIPs(o *proxyOptions) {
 	o.proxyLoadBalancerIPs = false
 }
 
-func NewFakeProxier(routeClient route.Interface, ofClient openflow.Client, nodePortAddresses []net.IP, isIPv6 bool, options ...proxyOptionsFn) *proxier {
+func NewFakeProxier(routeClient route.Interface, ofClient openflow.Client, nodePortAddresses []net.IP, groupIDAllocator openflow.GroupAllocator, isIPv6 bool, options ...proxyOptionsFn) *proxier {
 	hostname := "localhost"
 	eventBroadcaster := record.NewBroadcaster()
 	recorder := eventBroadcaster.NewRecorder(
@@ -144,7 +144,7 @@ func NewFakeProxier(routeClient route.Interface, ofClient openflow.Client, nodeP
 		endpointsInstalledMap:    types.EndpointsMap{},
 		endpointReferenceCounter: map[string]int{},
 		endpointsMap:             types.EndpointsMap{},
-		groupCounter:             types.NewGroupCounter(isIPv6, make(chan string, 100)),
+		groupCounter:             types.NewGroupCounter(groupIDAllocator, make(chan string, 100)),
 		ofClient:                 ofClient,
 		routeClient:              routeClient,
 		serviceStringMap:         map[string]k8sproxy.ServicePortName{},
@@ -162,7 +162,8 @@ func testClusterIP(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool, extraSv
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, isIPv6, withProxyAll)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, isIPv6, withProxyAll)
 
 	svcPort := 80
 	svcPortName := k8sproxy.ServicePortName{
@@ -195,7 +196,7 @@ func testClusterIP(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool, extraSv
 	}))
 	makeEndpointsMap(fp, allEps...)
 
-	groupID, _ := fp.groupCounter.Get(svcPortName, false)
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
 	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
 	bindingProtocol := binding.ProtocolTCP
 	if isIPv6 {
@@ -204,7 +205,6 @@ func testClusterIP(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool, extraSv
 	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
-	mockOFClient.EXPECT().UninstallServiceGroup(gomock.Any()).Times(1)
 
 	fp.syncProxyRules()
 }
@@ -214,11 +214,12 @@ func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
 	options := []proxyOptionsFn{withProxyAll}
 	if !proxyLoadBalancerIPs {
 		options = append(options, withoutProxyLoadBalancerIPs)
 	}
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nodePortAddresses, isIPv6, options...)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nodePortAddresses, groupAllocator, isIPv6, options...)
 
 	svcPort := 80
 	svcNodePort := 30008
@@ -281,8 +282,7 @@ func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep
 		eps = append(eps, makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc))
 	}
 	makeEndpointsMap(fp, eps...)
-
-	groupID, _ := fp.groupCounter.Get(svcPortName, false)
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
 	bindingProtocol := binding.ProtocolTCP
 	if isIPv6 {
 		bindingProtocol = binding.ProtocolTCPv6
@@ -291,18 +291,13 @@ func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep
 	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
 	if nodeLocalExternal {
-		groupID, _ = fp.groupCounter.Get(svcPortName, true)
+		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, true)
 		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
-	} else {
-		mockOFClient.EXPECT().UninstallServiceGroup(gomock.Any()).Times(1)
 	}
 	if proxyLoadBalancerIPs {
 		mockOFClient.EXPECT().InstallServiceFlows(groupID, loadBalancerIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeLoadBalancer).Times(1)
 	}
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeNodePort).Times(1)
-	if proxyLoadBalancerIPs {
-		mockOFClient.EXPECT().InstallLoadBalancerServiceFromOutsideFlows(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
-	}
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	if proxyLoadBalancerIPs {
 		mockRouteClient.EXPECT().AddLoadBalancer([]string{loadBalancerIP.String()}).Times(1)
@@ -317,7 +312,8 @@ func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nodePortAddresses, isIPv6, withProxyAll)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nodePortAddresses, groupAllocator, isIPv6, withProxyAll)
 
 	svcPort := 80
 	svcNodePort := 31000
@@ -378,7 +374,7 @@ func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP
 	}
 	makeEndpointsMap(fp, eps...)
 
-	groupID, _ := fp.groupCounter.Get(svcPortName, false)
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
 	bindingProtocol := binding.ProtocolTCP
 	if isIPv6 {
 		bindingProtocol = binding.ProtocolTCPv6
@@ -387,16 +383,13 @@ func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP
 	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
 	if nodeLocalExternal {
-		groupID, _ = fp.groupCounter.Get(svcPortName, true)
+		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, true)
 		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
-	} else {
-		mockOFClient.EXPECT().UninstallServiceGroup(gomock.Any()).Times(1)
 	}
 
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeNodePort).Times(1)
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	mockRouteClient.EXPECT().AddNodePort(gomock.Any(), uint16(svcNodePort), bindingProtocol).Times(1)
-	mockOFClient.EXPECT().InstallLoadBalancerServiceFromOutsideFlows(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 
 	fp.syncProxyRules()
 }
@@ -509,8 +502,10 @@ func TestDualStackService(t *testing.T) {
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fpv4 := NewFakeProxier(mockRouteClient, mockOFClient, nil, false)
-	fpv6 := NewFakeProxier(mockRouteClient, mockOFClient, nil, true)
+	ipv4GGroupAllocator := openflow.NewGroupAllocator(false)
+	ipv6GGroupAllocator := openflow.NewGroupAllocator(true)
+	fpv4 := NewFakeProxier(mockRouteClient, mockOFClient, nil, ipv4GGroupAllocator, false)
+	fpv6 := NewFakeProxier(mockRouteClient, mockOFClient, nil, ipv6GGroupAllocator, true)
 	metaProxier := k8sproxy.NewMetaProxier(fpv4, fpv6)
 
 	svcPort := 80
@@ -565,8 +560,8 @@ func TestDualStackService(t *testing.T) {
 	metaProxier.OnEndpointsUpdate(nil, epv6)
 	metaProxier.OnEndpointsSynced()
 
-	groupIDv4, _ := fpv4.groupCounter.Get(svcPortName, false)
-	groupIDv6, _ := fpv6.groupCounter.Get(svcPortName, false)
+	groupIDv4 := fpv4.groupCounter.AllocateIfNotExist(svcPortName, false)
+	groupIDv6 := fpv6.groupCounter.AllocateIfNotExist(svcPortName, false)
 
 	mockOFClient.EXPECT().InstallServiceGroup(groupIDv4, false, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallEndpointFlows(binding.ProtocolTCP, gomock.Any()).Times(1)
@@ -585,7 +580,8 @@ func testClusterIPRemoval(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool)
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, isIPv6, withProxyAll)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, isIPv6, withProxyAll)
 
 	svcPort := 80
 	svcPortName := k8sproxy.ServicePortName{
@@ -622,14 +618,14 @@ func testClusterIPRemoval(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool)
 	}
 	ep := makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc)
 	makeEndpointsMap(fp, ep)
-	groupID, _ := fp.groupCounter.Get(svcPortName, false)
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
 	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	mockOFClient.EXPECT().UninstallServiceFlows(svcIP, uint16(svcPort), bindingProtocol).Times(1)
 	mockOFClient.EXPECT().UninstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
-	mockOFClient.EXPECT().UninstallServiceGroup(gomock.Any()).Times(2)
+	mockOFClient.EXPECT().UninstallServiceGroup(gomock.Any()).Times(1)
 	fp.syncProxyRules()
 
 	fp.serviceChanges.OnServiceUpdate(service, nil)
@@ -650,7 +646,8 @@ func testClusterIPNoEndpoint(t *testing.T, svcIP net.IP, isIPv6 bool) {
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, isIPv6)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, isIPv6)
 
 	svcPort := 80
 	svcNodePort := 3001
@@ -688,7 +685,8 @@ func testClusterIPRemoveSamePortEndpoint(t *testing.T, svcIP net.IP, epIP net.IP
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, isIPv6)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, isIPv6)
 
 	svcPort := 80
 	svcPortName := k8sproxy.ServicePortName{
@@ -752,8 +750,8 @@ func testClusterIPRemoveSamePortEndpoint(t *testing.T, svcIP net.IP, epIP net.IP
 	}
 	makeEndpointsMap(fp, ep, epUDP)
 
-	groupID, _ := fp.groupCounter.Get(svcPortName, false)
-	groupIDUDP, _ := fp.groupCounter.Get(svcPortNameUDP, false)
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
+	groupIDUDP := fp.groupCounter.AllocateIfNotExist(svcPortNameUDP, false)
 	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallServiceGroup(groupIDUDP, false, gomock.Any()).Times(2)
 	mockOFClient.EXPECT().InstallEndpointFlows(protocolTCP, gomock.Any()).Times(1)
@@ -780,7 +778,8 @@ func testClusterIPRemoveEndpoints(t *testing.T, svcIP net.IP, epIP net.IP, isIPv
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, isIPv6)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, isIPv6)
 
 	svcPort := 80
 	svcPortName := k8sproxy.ServicePortName{
@@ -816,7 +815,7 @@ func testClusterIPRemoveEndpoints(t *testing.T, svcIP net.IP, epIP net.IP, isIPv
 		bindingProtocol = binding.ProtocolTCPv6
 	}
 	makeEndpointsMap(fp, ep)
-	groupID, _ := fp.groupCounter.Get(svcPortName, false)
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
 	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(2)
 	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(2)
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
@@ -840,7 +839,8 @@ func testSessionAffinityNoEndpoint(t *testing.T, svcExternalIPs net.IP, svcIP ne
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, isIPv6)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, isIPv6)
 
 	svcPort := 80
 	svcNodePort := 3001
@@ -888,7 +888,7 @@ func testSessionAffinityNoEndpoint(t *testing.T, svcExternalIPs net.IP, svcIP ne
 	if isIPv6 {
 		bindingProtocol = binding.ProtocolTCPv6
 	}
-	groupID, _ := fp.groupCounter.Get(svcPortName, false)
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
 	mockOFClient.EXPECT().InstallServiceGroup(groupID, true, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(corev1.DefaultClientIPServiceAffinitySeconds), false, corev1.ServiceTypeClusterIP).Times(1)
@@ -909,7 +909,8 @@ func testSessionAffinity(t *testing.T, svcExternalIPs net.IP, svcIP net.IP, isIP
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, isIPv6)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, isIPv6)
 
 	svcPort := 80
 	svcNodePort := 3001
@@ -940,7 +941,6 @@ func testSessionAffinity(t *testing.T, svcExternalIPs net.IP, svcIP net.IP, isIP
 		}),
 	)
 	makeEndpointsMap(fp)
-
 	fp.syncProxyRules()
 }
 
@@ -957,7 +957,8 @@ func testPortChange(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool) {
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, isIPv6)
+	groupAllocator := openflow.NewGroupAllocator(isIPv6)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, isIPv6)
 
 	svcPort1 := 80
 	svcPort2 := 8080
@@ -996,7 +997,7 @@ func testPortChange(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool) {
 	}
 	ep := makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc)
 	makeEndpointsMap(fp, ep)
-	groupID, _ := fp.groupCounter.Get(svcPortName, false)
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
 	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort1), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP)
@@ -1031,7 +1032,8 @@ func TestServicesWithSameEndpoints(t *testing.T) {
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
 	mockRouteClient := routemock.NewMockInterface(ctrl)
-	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, false)
+	groupAllocator := openflow.NewGroupAllocator(false)
+	fp := NewFakeProxier(mockRouteClient, mockOFClient, nil, groupAllocator, false)
 	epIP := net.ParseIP("10.50.60.71")
 	svcIP1 := net.ParseIP("10.180.30.41")
 	svcIP2 := net.ParseIP("10.180.30.42")
@@ -1082,8 +1084,8 @@ func TestServicesWithSameEndpoints(t *testing.T) {
 	ep1 := epMapFactory(svcPortName1, epIP.String())
 	ep2 := epMapFactory(svcPortName2, epIP.String())
 
-	groupID1, _ := fp.groupCounter.Get(svcPortName1, false)
-	groupID2, _ := fp.groupCounter.Get(svcPortName2, false)
+	groupID1 := fp.groupCounter.AllocateIfNotExist(svcPortName1, false)
+	groupID2 := fp.groupCounter.AllocateIfNotExist(svcPortName2, false)
 	mockOFClient.EXPECT().InstallServiceGroup(groupID1, false, gomock.Any()).Times(1)
 	mockOFClient.EXPECT().InstallServiceGroup(groupID2, false, gomock.Any()).Times(1)
 	bindingProtocol := binding.ProtocolTCP
diff --git a/pkg/controller/networkpolicy/validate_test.go b/pkg/controller/networkpolicy/validate_test.go
index d5f65466f..8297815b0 100644
--- a/pkg/controller/networkpolicy/validate_test.go
+++ b/pkg/controller/networkpolicy/validate_test.go
@@ -23,6 +23,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	crdv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
+	crdv1alpha2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
 )
 
 func TestValidateAntreaPolicy(t *testing.T) {
@@ -940,6 +941,84 @@ func TestValidateAntreaPolicy(t *testing.T) {
 			},
 			expectedReason: "",
 		},
+		{
+			name: "acnp-invalid-label-key-applied-to",
+			policy: &crdv1alpha1.ClusterNetworkPolicy{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "acnp-invalid-label-key-applied-to",
+				},
+				Spec: crdv1alpha1.ClusterNetworkPolicySpec{
+					AppliedTo: []crdv1alpha1.NetworkPolicyPeer{
+						{
+							NamespaceSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{"foo=": "bar"},
+							},
+						},
+					},
+				},
+			},
+			expectedReason: "Invalid label key: foo=: name part must consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyName',  or 'my.name',  or '123-abc', regex used for validation is '([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9]')",
+		},
+		{
+			name: "acnp-invalid-label-value-applied-to",
+			policy: &crdv1alpha1.ClusterNetworkPolicy{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "acnp-invalid-label-value-applied-to",
+				},
+				Spec: crdv1alpha1.ClusterNetworkPolicySpec{
+					Ingress: []crdv1alpha1.Rule{
+						{
+							Action: &allowAction,
+							From: []crdv1alpha1.NetworkPolicyPeer{
+								{
+									PodSelector: &metav1.LabelSelector{
+										MatchLabels: map[string]string{"foo": "bar"},
+									},
+								},
+							},
+							AppliedTo: []crdv1alpha1.NetworkPolicyPeer{
+								{
+									NamespaceSelector: &metav1.LabelSelector{
+										MatchLabels: map[string]string{"foo1": "bar="},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedReason: "Invalid label value: bar=: a valid label must be an empty string or consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyValue',  or 'my_value',  or '12345', regex used for validation is '(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?')",
+		},
+		{
+			name: "acnp-invalid-label-key-rule",
+			policy: &crdv1alpha1.ClusterNetworkPolicy{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "acnp-invalid-label-key-rule",
+				},
+				Spec: crdv1alpha1.ClusterNetworkPolicySpec{
+					AppliedTo: []crdv1alpha1.NetworkPolicyPeer{
+						{
+							NamespaceSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{"foo": "bar"},
+							},
+						},
+					},
+					Ingress: []crdv1alpha1.Rule{
+						{
+							Action: &allowAction,
+							From: []crdv1alpha1.NetworkPolicyPeer{
+								{
+									PodSelector: &metav1.LabelSelector{
+										MatchLabels: map[string]string{"foo=": "bar1"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedReason: "Invalid label key: foo=: name part must consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyName',  or 'my.name',  or '123-abc', regex used for validation is '([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9]')",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -955,3 +1034,53 @@ func TestValidateAntreaPolicy(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateClusterAntreaGroup(t *testing.T) {
+	tests := []struct {
+		name           string
+		group          *crdv1alpha2.ClusterGroup
+		expectedReason string
+	}{
+		{
+			name: "cg-invalid-label-key",
+			group: &crdv1alpha2.ClusterGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cg-invalid-label-key",
+				},
+				Spec: crdv1alpha2.GroupSpec{
+					PodSelector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{"foo=": "bar"},
+					},
+				},
+			},
+			expectedReason: "Invalid label key: foo=: name part must consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyName',  or 'my.name',  or '123-abc', regex used for validation is '([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9]')",
+		},
+		{
+			name: "cg-invalid-label-value",
+			group: &crdv1alpha2.ClusterGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cg-invalid-label-value",
+				},
+				Spec: crdv1alpha2.GroupSpec{
+					PodSelector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{"foo": "bar="},
+					},
+				},
+			},
+			expectedReason: "Invalid label value: bar=: a valid label must be an empty string or consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyValue',  or 'my_value',  or '12345', regex used for validation is '(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?')",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, c := newController()
+			v := NewNetworkPolicyValidator(c.NetworkPolicyController)
+			actualReason, allowed := v.validateAntreaGroup(tt.group, nil, admv1.Create, authenticationv1.UserInfo{})
+			assert.Equal(t, tt.expectedReason, actualReason)
+			if tt.expectedReason == "" {
+				assert.True(t, allowed)
+			} else {
+				assert.False(t, allowed)
+			}
+		})
+	}
+}
diff --git a/pkg/ovs/openflow/ofctrl_packetin_test.go b/pkg/ovs/openflow/ofctrl_packetin_test.go
index 106856b43..b4d82de21 100644
--- a/pkg/ovs/openflow/ofctrl_packetin_test.go
+++ b/pkg/ovs/openflow/ofctrl_packetin_test.go
@@ -15,12 +15,15 @@
 package openflow
 
 import (
+	"net"
 	"testing"
 
 	"antrea.io/libOpenflow/protocol"
 	"antrea.io/libOpenflow/util"
+	"antrea.io/ofnet/ofctrl"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/net/icmp"
 )
 
 func TestGetTCPHeaderData(t *testing.T) {
@@ -37,7 +40,7 @@ func TestGetTCPHeaderData(t *testing.T) {
 		args args
 	}{
 		{
-			name: "ipv4",
+			name: "GetTCPHeader-ipv4",
 			args: args{
 				tcp: protocol.TCP{
 					PortSrc: 1080,
@@ -73,3 +76,145 @@ func TestGetTCPHeaderData(t *testing.T) {
 		})
 	}
 }
+
+func TestGetUDPHeaderData(t *testing.T) {
+	type args struct {
+		udp              protocol.UDP
+		expectUDPSrcPort uint16
+		expectUDPDstPort uint16
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "GetUDPHeader-ipv4",
+			args: args{
+				udp: protocol.UDP{
+					PortSrc: 1080,
+					PortDst: 80,
+				},
+				expectUDPSrcPort: 1080,
+				expectUDPDstPort: 80,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pktIn := new(protocol.IPv4)
+			pktIn.Data = &tt.args.udp
+
+			udpSrcPort, udpDstPort, err := GetUDPHeaderData(pktIn)
+			require.NoError(t, err, "GetUDPHeaderData() returned an error")
+			assert.Equal(t, tt.args.expectUDPSrcPort, udpSrcPort)
+			assert.Equal(t, tt.args.expectUDPDstPort, udpDstPort)
+		})
+	}
+}
+
+func TestGetICMPHeaderData(t *testing.T) {
+	testEcho, _ := (&icmp.Echo{ID: 1, Seq: 2}).Marshal(0)
+	type args struct {
+		icmp              protocol.ICMP
+		expectICMPType    uint8
+		expectICMPCode    uint8
+		expectICMPEchoID  uint16
+		expectICMPEchoSeq uint16
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "GetICMPHeader-ipv4",
+			args: args{
+				icmp: protocol.ICMP{
+					Type: icmpEchoRequestType,
+					Code: 0,
+					Data: testEcho,
+				},
+				expectICMPType:    icmpEchoRequestType,
+				expectICMPCode:    0,
+				expectICMPEchoID:  1,
+				expectICMPEchoSeq: 2,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pktIn := new(protocol.IPv4)
+			pktIn.Data = &tt.args.icmp
+
+			icmpType, icmpCode, icmpEchoID, icmpEchoSeq, err := getICMPHeaderData(pktIn)
+			require.NoError(t, err, "GetICMPHeaderData() returned an error")
+			assert.Equal(t, tt.args.expectICMPType, icmpType)
+			assert.Equal(t, tt.args.expectICMPCode, icmpCode)
+			assert.Equal(t, tt.args.expectICMPEchoID, icmpEchoID)
+			assert.Equal(t, tt.args.expectICMPEchoSeq, icmpEchoSeq)
+		})
+	}
+}
+
+func TestParsePacketIn(t *testing.T) {
+	testMac1, _ := net.ParseMAC("00:00:5e:00:53:01")
+	testMac2, _ := net.ParseMAC("00:00:5e:00:53:00")
+	testIP1 := net.ParseIP("2001:db8::68")
+	testIP2 := net.ParseIP("2001:db8::69")
+	testTCP := protocol.TCP{
+		PortSrc: 1080,
+		PortDst: 80,
+		Code:    uint8(net.FlagBroadcast),
+	}
+	testBytes, _ := testTCP.MarshalBinary()
+	testBuffer := new(util.Buffer)
+	testBuffer.UnmarshalBinary(testBytes)
+	tests := []struct {
+		name       string
+		pktIn      *ofctrl.PacketIn
+		expectedOb *Packet
+		wantErr    bool
+	}{
+		{
+			"ParsePacketIn-ipv6",
+			&ofctrl.PacketIn{
+				Reason: 1,
+				Data: protocol.Ethernet{
+					Ethertype: protocol.IPv6_MSG,
+					HWDst:     testMac1,
+					HWSrc:     testMac2,
+					Data: util.Message(&protocol.IPv6{
+						Length:     1,
+						NextHeader: protocol.Type_TCP,
+						HopLimit:   0,
+						NWSrc:      testIP1,
+						NWDst:      testIP2,
+						Data:       testBuffer,
+					}),
+				},
+			},
+			&Packet{
+				IsIPv6:          true,
+				DestinationMAC:  testMac1,
+				SourceMAC:       testMac2,
+				DestinationIP:   testIP2,
+				SourceIP:        testIP1,
+				IPLength:        41,
+				IPProto:         protocol.Type_TCP,
+				TTL:             0,
+				DestinationPort: 80,
+				SourcePort:      1080,
+				TCPFlags:        uint8(net.FlagBroadcast),
+			},
+			false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actualOb, err := ParsePacketIn(tt.pktIn)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("ParsePacketIn() returned an error = %v, expected error = %v", err, tt.wantErr)
+			}
+			assert.Equal(t, tt.expectedOb, actualOb)
+		})
+	}
+}
diff --git a/test/e2e/antreapolicy_test.go b/test/e2e/antreapolicy_test.go
index 262bf8746..3b2b80317 100644
--- a/test/e2e/antreapolicy_test.go
+++ b/test/e2e/antreapolicy_test.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -1938,7 +1939,7 @@ func testAuditLoggingBasic(t *testing.T, data *TestData) {
 					}
 					expectedNumEntries += 1
 					// The audit log should contain log entry `... Drop <ofPriority> <x/a IP> <z/* IP> ...`
-					re := regexp.MustCompile(`Drop [0-9]+ ` + srcIPs[i] + ` ` + dstIPs[j])
+					re := regexp.MustCompile(`Drop [0-9]+ ` + srcIPs[i] + ` [0-9]+ ` + dstIPs[j] + ` ` + strconv.Itoa(int(p80)))
 					if re.MatchString(stdout) {
 						actualNumEntries += 1
 					} else {
diff --git a/test/e2e/flowaggregator_test.go b/test/e2e/flowaggregator_test.go
index f585ccd04..ad2e6c413 100644
--- a/test/e2e/flowaggregator_test.go
+++ b/test/e2e/flowaggregator_test.go
@@ -129,6 +129,10 @@ const (
 	testEgressRuleName             = "test-egress-rule-name"
 	iperfTimeSec                   = 12
 	protocolIdentifierTCP          = 6
+	// Set target bandwidth(bits/sec) of iPerf traffic to a relatively small value
+	// (default unlimited for TCP), to reduce the variances caused by network performance
+	// during 12s, and make the throughput test more stable.
+	iperfBandwidth = "10m"
 )
 
 var (
@@ -611,9 +615,9 @@ func checkAntctlRecord(t *testing.T, record map[string]interface{}, srcIP, dstIP
 func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP string, isIPv6 bool, isIntraNode bool, checkService bool, checkK8sNetworkPolicy bool, checkAntreaNetworkPolicy bool, checkBandwidth bool) {
 	var cmdStr string
 	if !isIPv6 {
-		cmdStr = fmt.Sprintf("iperf3 -c %s -t %d", dstIP, iperfTimeSec)
+		cmdStr = fmt.Sprintf("iperf3 -c %s -t %d -b %s", dstIP, iperfTimeSec, iperfBandwidth)
 	} else {
-		cmdStr = fmt.Sprintf("iperf3 -6 -c %s -t %d", dstIP, iperfTimeSec)
+		cmdStr = fmt.Sprintf("iperf3 -6 -c %s -t %d -b %s", dstIP, iperfTimeSec, iperfBandwidth)
 	}
 	stdout, _, err := data.runCommandFromPod(testNamespace, "perftest-a", "perftool", []string{"bash", "-c", cmdStr})
 	require.NoErrorf(t, err, "Error when running iperf3 client: %v", err)
@@ -625,10 +629,8 @@ func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP stri
 	var bandwidthInMbps float64
 	if strings.Contains(bwSlice[1], "Mbits") {
 		bandwidthInMbps = bandwidthInFloat
-	} else if strings.Contains(bwSlice[1], "Gbits") {
-		bandwidthInMbps = bandwidthInFloat * float64(1024)
 	} else {
-		t.Fatalf("Unit of the traffic bandwidth reported by iperf should either be Mbits or Gbits, failing the test.")
+		t.Fatalf("Unit of the traffic bandwidth reported by iperf should be Mbits.")
 	}
 
 	collectorOutput, recordSlices := getCollectorOutput(t, srcIP, dstIP, srcPort, checkService, true, isIPv6)
@@ -684,19 +686,21 @@ func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP stri
 			if checkBandwidth && !strings.Contains(record, "throughput: 0") {
 				flowStartTime := int64(getUint64FieldFromRecord(t, record, "flowStartSeconds"))
 				exportTime := int64(getUint64FieldFromRecord(t, record, "flowEndSeconds"))
+				flowEndReason := int64(getUint64FieldFromRecord(t, record, "flowEndReason"))
 				var recBandwidth float64
-				// Check bandwidth with the field "throughput" except for the last record,
-				// as its throughput is significantly lower than the average Iperf throughput.
-				if exportTime < flowStartTime+iperfTimeSec {
+				// flowEndReason == 3 means the end of flow detected
+				if exportTime >= flowStartTime+iperfTimeSec || flowEndReason == 3 {
+					// Check average bandwidth on the last record.
+					octetTotalCount := getUint64FieldFromRecord(t, record, "octetTotalCount")
+					recBandwidth = float64(octetTotalCount) * 8 / float64(iperfTimeSec) / 1000000
+				} else {
+					// Check bandwidth with the field "throughput" except for the last record,
+					// as their throughput may be significantly lower than the average Iperf throughput.
 					throughput := getUint64FieldFromRecord(t, record, "throughput")
 					recBandwidth = float64(throughput) / 1000000
-				} else {
-					// Check average bandwidth after receiving all the records.
-					octetTotalCount := getUint64FieldFromRecord(t, record, "octetTotalCount")
-					recBandwidth = float64(octetTotalCount) * 8 / float64(exportTime-flowStartTime) / 1000000
 				}
-				t.Logf("Iperf throughput: %.2f Mbits/s, IPFIX record throughput: %.2f Mbits/s", bandwidthInMbps, recBandwidth)
-				assert.InDeltaf(recBandwidth, bandwidthInMbps, bandwidthInMbps*0.15, "Difference between Iperf bandwidth and IPFIX record bandwidth should be lower than 15%%")
+				t.Logf("Throughput check on record with flowEndSeconds-flowStartSeconds: %v, Iperf throughput: %.2f Mbits/s, IPFIX record throughput: %.2f Mbits/s", exportTime-flowStartTime, bandwidthInMbps, recBandwidth)
+				assert.InDeltaf(recBandwidth, bandwidthInMbps, bandwidthInMbps*0.15, "Difference between Iperf bandwidth and IPFIX record bandwidth should be lower than 15%%, record: %s", record)
 			}
 		}
 	}
@@ -852,7 +856,9 @@ func getUint64FieldFromRecord(t *testing.T, record string, field string) uint64
 func getCollectorOutput(t *testing.T, srcIP, dstIP, srcPort string, isDstService bool, checkAllRecords bool, isIPv6 bool) (string, []string) {
 	var collectorOutput string
 	var recordSlices []string
-	err := wait.PollImmediate(500*time.Millisecond, aggregatorInactiveFlowRecordTimeout, func() (bool, error) {
+	// In the ToExternalFlows test, flow record will arrive 5.5s (exporterActiveFlowExportTimeout+aggregatorActiveFlowRecordTimeout) after executing wget command
+	// We set the timeout to 9s (5.5s plus one more aggregatorActiveFlowRecordTimeout) to make the ToExternalFlows test more stable
+	err := wait.PollImmediate(500*time.Millisecond, exporterActiveFlowExportTimeout+aggregatorActiveFlowRecordTimeout*2, func() (bool, error) {
 		var rc int
 		var err error
 		// `pod-running-timeout` option is added to cover scenarios where ipfix flow-collector has crashed after being deployed
@@ -867,8 +873,10 @@ func getCollectorOutput(t *testing.T, srcIP, dstIP, srcPort string, isDstService
 			for _, record := range recordSlices {
 				flowStartTime := int64(getUint64FieldFromRecord(t, record, "flowStartSeconds"))
 				exportTime := int64(getUint64FieldFromRecord(t, record, "flowEndSeconds"))
+				flowEndReason := int64(getUint64FieldFromRecord(t, record, "flowEndReason"))
 				if strings.Contains(record, src) && strings.Contains(record, dst) && strings.Contains(record, srcPort) {
-					if exportTime >= flowStartTime+iperfTimeSec {
+					// flowEndReason == 3 means the end of flow detected
+					if exportTime >= flowStartTime+iperfTimeSec || flowEndReason == 3 {
 						return true, nil
 					}
 				}
diff --git a/test/e2e/proxy_test.go b/test/e2e/proxy_test.go
index 75e3b97e8..aef107a25 100644
--- a/test/e2e/proxy_test.go
+++ b/test/e2e/proxy_test.go
@@ -87,16 +87,6 @@ func skipIfProxyAllDisabled(t *testing.T, data *TestData) {
 	}
 }
 
-func skipIfProxyAllEnabled(t *testing.T, data *TestData) {
-	isProxyAll, err := data.isProxyAll()
-	if err != nil {
-		t.Fatalf("Error getting option antreaProxy.proxyAll value")
-	}
-	if isProxyAll {
-		t.Skipf("Skipping test because option antreaProxy.proxyAll is enabled")
-	}
-}
-
 func skipIfKubeProxyEnabled(t *testing.T, data *TestData) {
 	_, err := data.clientset.AppsV1().DaemonSets(kubeNamespace).Get(context.TODO(), "kube-proxy", metav1.GetOptions{})
 	if err == nil {
diff --git a/test/e2e/service_externalip_test.go b/test/e2e/service_externalip_test.go
index 800789f92..7bebb3d96 100644
--- a/test/e2e/service_externalip_test.go
+++ b/test/e2e/service_externalip_test.go
@@ -107,6 +107,33 @@ func testServiceExternalTrafficPolicyLocal(t *testing.T, data *TestData) {
 			},
 			expectedNodeUpdated: nodeName(0),
 		},
+		{
+			name:    "endpoint created IPv6",
+			ipRange: v1alpha2.IPRange{CIDR: "2021:1::aaa0/124"},
+			nodeSelector: metav1.LabelSelector{
+				MatchExpressions: []metav1.LabelSelectorRequirement{
+					{
+						Key:      v1.LabelHostname,
+						Operator: metav1.LabelSelectorOpIn,
+						Values:   []string{nodeName(0), nodeName(1)},
+					},
+				},
+			},
+			expectedExternalIP:      "2021:1::aaa1",
+			originalEndpointSubsets: nil,
+			expectedNodeOrigin:      "",
+			updatedEndpointSubsets: []v1.EndpointSubset{
+				{
+					Addresses: []v1.EndpointAddress{
+						{
+							IP:       "2021:8::aaa1",
+							NodeName: stringPtr(nodeName(0)),
+						},
+					},
+				},
+			},
+			expectedNodeUpdated: nodeName(0),
+		},
 		{
 			name:    "endpoint changed",
 			ipRange: v1alpha2.IPRange{CIDR: "169.254.100.0/30"},
@@ -143,6 +170,42 @@ func testServiceExternalTrafficPolicyLocal(t *testing.T, data *TestData) {
 			},
 			expectedNodeUpdated: nodeName(1),
 		},
+		{
+			name:    "endpoint changed IPv6",
+			ipRange: v1alpha2.IPRange{CIDR: "2021:1::aaa0/124"},
+			nodeSelector: metav1.LabelSelector{
+				MatchExpressions: []metav1.LabelSelectorRequirement{
+					{
+						Key:      v1.LabelHostname,
+						Operator: metav1.LabelSelectorOpIn,
+						Values:   []string{nodeName(0), nodeName(1)},
+					},
+				},
+			},
+			expectedExternalIP: "2021:1::aaa1",
+			originalEndpointSubsets: []v1.EndpointSubset{
+				{
+					Addresses: []v1.EndpointAddress{
+						{
+							IP:       "2021:8::aaa1",
+							NodeName: stringPtr(nodeName(0)),
+						},
+					},
+				},
+			},
+			expectedNodeOrigin: nodeName(0),
+			updatedEndpointSubsets: []v1.EndpointSubset{
+				{
+					Addresses: []v1.EndpointAddress{
+						{
+							IP:       "2021:8::aaa1",
+							NodeName: stringPtr(nodeName(1)),
+						},
+					},
+				},
+			},
+			expectedNodeUpdated: nodeName(1),
+		},
 	}
 	for idx, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -223,6 +286,18 @@ func testServiceWithExternalIPCRUD(t *testing.T, data *TestData) {
 			expectedNodes:      sets.NewString(nodeName(0)),
 			expectedTotal:      2,
 		},
+		{
+			name:    "single matching Node with IPv6 range",
+			ipRange: v1alpha2.IPRange{CIDR: "2021:1::aaa0/124"},
+			nodeSelector: metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					v1.LabelHostname: nodeName(0),
+				},
+			},
+			expectedExternalIP: "2021:1::aaa1",
+			expectedNodes:      sets.NewString(nodeName(0)),
+			expectedTotal:      15,
+		},
 		{
 			name:    "two matching Nodes",
 			ipRange: v1alpha2.IPRange{Start: "169.254.101.10", End: "169.254.101.11"},
@@ -335,6 +410,15 @@ func testServiceUpdateExternalIP(t *testing.T, data *TestData) {
 			newIPRange:         v1alpha2.IPRange{CIDR: "169.254.101.0/30"},
 			newExternalIP:      "169.254.101.1",
 		},
+		{
+			name:               "different Nodes in IPv6 cluster",
+			originalNode:       nodeName(0),
+			newNode:            nodeName(1),
+			originalIPRange:    v1alpha2.IPRange{CIDR: "2021:2::aaa0/124"},
+			originalExternalIP: "2021:2::aaa1",
+			newIPRange:         v1alpha2.IPRange{CIDR: "2021:2::bbb0/124"},
+			newExternalIP:      "2021:2::bbb1",
+		},
 	}
 	for idx, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -391,6 +475,11 @@ func testServiceNodeFailure(t *testing.T, data *TestData) {
 			ipRange:    v1alpha2.IPRange{CIDR: "169.254.100.0/30"},
 			expectedIP: "169.254.100.1",
 		},
+		{
+			name:       "IPv6 cluster",
+			ipRange:    v1alpha2.IPRange{CIDR: "2021:4::aaa0/124"},
+			expectedIP: "2021:4::aaa1",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -470,12 +559,19 @@ func testExternalIPAccess(t *testing.T, data *TestData) {
 			clientName:      "eth-ipv4",
 			clientIP:        "169.254.170.1",
 			localIP:         "169.254.170.2",
-			clientIPMaskLen: 24,
+			clientIPMaskLen: 25,
+		},
+		{
+			name:            "IPv6 cluster",
+			externalIPCIDR:  "2021:4::aab0/124",
+			clientName:      "eth-ipv6",
+			clientIP:        "2021:4::aaa1",
+			localIP:         "2021:4::aaa2",
+			clientIPMaskLen: 124,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			skipIfProxyAllEnabled(t, data)
 			ipFamily := corev1.IPv4Protocol
 			if utilnet.IsIPv6CIDRString(tt.externalIPCIDR) {
 				skipIfNotIPv6Cluster(t)
@@ -491,6 +587,7 @@ func testExternalIPAccess(t *testing.T, data *TestData) {
 			// Create agnhost Pods on each Node.
 			for idx, node := range nodes {
 				createAgnhostPod(t, data, agnhosts[idx], node, false)
+				defer data.deletePodAndWait(defaultTimeout, agnhosts[idx], testNamespace)
 			}
 			var port int32 = 8080
 			externalIPTestCases := []struct {
@@ -533,6 +630,8 @@ func testExternalIPAccess(t *testing.T, data *TestData) {
 					}
 					service, err := data.createServiceWithAnnotations(et.serviceName, testNamespace, port, port, corev1.ProtocolTCP, map[string]string{"app": "agnhost"}, false, et.externalTrafficPolicyLocal, corev1.ServiceTypeLoadBalancer, &ipFamily, annotations)
 					require.NoError(t, err)
+					defer data.deleteService(service.Name)
+
 					externalIP, host, err := waitExternalIPConfigured(service)
 					require.NoError(t, err)
 
@@ -544,6 +643,7 @@ ip addr add %[3]s/%[4]d dev %[1]s-b && \
 ip link set dev %[1]s-b up && \
 ip netns exec %[1]s ip addr add %[2]s/%[4]d dev %[1]s-a && \
 ip netns exec %[1]s ip link set dev %[1]s-a up && \
+ip netns exec %[1]s ip route replace default via %[3]s && \
 ip netns exec %[1]s \
 sleep 3600`, tt.clientName, tt.clientIP, tt.localIP, tt.clientIPMaskLen)
 
@@ -611,7 +711,7 @@ func getServiceExternalIPAndHost(service *v1.Service) (string, string) {
 	return service.Status.LoadBalancer.Ingress[0].IP, service.Status.LoadBalancer.Ingress[0].Hostname
 }
 
-func (data *TestData) waitForServiceConfigured(service *v1.Service, expectedExternalIP string, waitForNodeConfigured bool, expectedNodeName string, otherNodes ...string) (*corev1.Service, error) {
+func (data *TestData) waitForServiceConfigured(service *v1.Service, expectedExternalIP string, waitForNodeConfigured bool, expectedNodeName string) (*corev1.Service, error) {
 	err := wait.PollImmediate(200*time.Millisecond, 15*time.Second, func() (done bool, err error) {
 		service, err = data.clientset.CoreV1().Services(service.Namespace).Get(context.TODO(), service.Name, metav1.GetOptions{})
 		if err != nil {
@@ -634,27 +734,5 @@ func (data *TestData) waitForServiceConfigured(service *v1.Service, expectedExte
 		return service, fmt.Errorf("wait for Service %q configured failed: %v. Expected external IP %s on Node %s, actual status %#v",
 			service.Name, err, expectedExternalIP, expectedNodeName, service.Status)
 	}
-	err = wait.PollImmediate(200*time.Millisecond, 10*time.Second, func() (done bool, err error) {
-		// Make sure the IP is configured on the expected Node.
-		if expectedNodeName != "" {
-			exists, err := hasIP(data, expectedNodeName, expectedExternalIP)
-			if err != nil || !exists {
-				return false, fmt.Errorf("expected external IP %s to be assigned to Node %s: %v", expectedExternalIP, expectedNodeName, err)
-			}
-		}
-		// Make sure the IP is not configured on the other Nodes.
-		if len(otherNodes) != 0 {
-			for _, node := range otherNodes {
-				exists, err := hasIP(data, node, expectedExternalIP)
-				if err != nil || exists {
-					return false, fmt.Errorf("expected external IP %s not to be assigned to Node %s: %v", expectedExternalIP, expectedNodeName, err)
-				}
-			}
-		}
-		return true, nil
-	})
-	if err != nil {
-		return service, fmt.Errorf("wait for IP assigned to Node failed: %v", err)
-	}
 	return service, nil
 }
diff --git a/test/integration/agent/openflow_test.go b/test/integration/agent/openflow_test.go
index e4ddef735..237aca507 100644
--- a/test/integration/agent/openflow_test.go
+++ b/test/integration/agent/openflow_test.go
@@ -106,7 +106,8 @@ var (
 )
 
 func TestConnectivityFlows(t *testing.T) {
-	// Initialize ovs metrics (Prometheus) to test them
+	// Reset OVS metrics (Prometheus) and reinitialize them to test.
+	legacyregistry.Reset()
 	metrics.InitializeOVSMetrics()
 
 	// Hack the OS type if we run the test not on Windows Node.
@@ -123,6 +124,7 @@ func TestConnectivityFlows(t *testing.T) {
 		assert.Nil(t, err, fmt.Sprintf("Error while disconnecting from OVS bridge: %v", err))
 		err = ofTestUtils.DeleteOVSBridge(br)
 		assert.Nil(t, err, fmt.Sprintf("Error while deleting OVS bridge: %v", err))
+		ofClient.CleanOFTableCache()
 	}()
 
 	config := prepareConfiguration()
@@ -156,7 +158,8 @@ func TestConnectivityFlows(t *testing.T) {
 }
 
 func TestAntreaFlexibleIPAMConnectivityFlows(t *testing.T) {
-	// Initialize ovs metrics (Prometheus) to test them
+	// Reset OVS metrics (Prometheus) and reinitialize them to test.
+	legacyregistry.Reset()
 	metrics.InitializeOVSMetrics()
 
 	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, true, false, false, true, false)
@@ -167,6 +170,7 @@ func TestAntreaFlexibleIPAMConnectivityFlows(t *testing.T) {
 		assert.Nil(t, err, fmt.Sprintf("Error while disconnecting from OVS bridge: %v", err))
 		err = ofTestUtils.DeleteOVSBridge(br)
 		assert.Nil(t, err, fmt.Sprintf("Error while deleting OVS bridge: %v", err))
+		ofClient.CleanOFTableCache()
 	}()
 
 	config := prepareConfiguration()
@@ -201,6 +205,10 @@ func TestAntreaFlexibleIPAMConnectivityFlows(t *testing.T) {
 }
 
 func TestReplayFlowsConnectivityFlows(t *testing.T) {
+	// Reset OVS metrics (Prometheus) and reinitialize them to test.
+	legacyregistry.Reset()
+	metrics.InitializeOVSMetrics()
+
 	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false, false, false, false, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge: %v", err))
@@ -210,6 +218,7 @@ func TestReplayFlowsConnectivityFlows(t *testing.T) {
 		assert.Nil(t, err, fmt.Sprintf("Error while disconnecting from OVS bridge: %v", err))
 		err = ofTestUtils.DeleteOVSBridge(br)
 		assert.Nil(t, err, fmt.Sprintf("Error while deleting OVS bridge: %v", err))
+		ofClient.CleanOFTableCache()
 	}()
 
 	config := prepareConfiguration()
@@ -237,6 +246,10 @@ func TestReplayFlowsConnectivityFlows(t *testing.T) {
 }
 
 func TestReplayFlowsNetworkPolicyFlows(t *testing.T) {
+	// Reset OVS metrics (Prometheus) and reinitialize them to test.
+	legacyregistry.Reset()
+	metrics.InitializeOVSMetrics()
+
 	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false, false, false, false, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge: %v", err))
@@ -249,6 +262,7 @@ func TestReplayFlowsNetworkPolicyFlows(t *testing.T) {
 		assert.Nil(t, err, fmt.Sprintf("Error while disconnecting from OVS bridge: %v", err))
 		err = ofTestUtils.DeleteOVSBridge(br)
 		assert.Nil(t, err, fmt.Sprintf("Error while deleting OVS bridge: %v", err))
+		ofClient.CleanOFTableCache()
 	}()
 
 	ruleID := uint32(100)
@@ -420,7 +434,8 @@ func testUninstallPodFlows(t *testing.T, config *testConfig) {
 }
 
 func TestNetworkPolicyFlows(t *testing.T) {
-	// Initialize ovs metrics (Prometheus) to test them
+	// Reset OVS metrics (Prometheus) and reinitialize them to test.
+	legacyregistry.Reset()
 	metrics.InitializeOVSMetrics()
 
 	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false, false, false, false, false)
@@ -435,6 +450,7 @@ func TestNetworkPolicyFlows(t *testing.T) {
 		assert.Nil(t, err, fmt.Sprintf("Error while disconnecting from OVS bridge: %v", err))
 		err = ofTestUtils.DeleteOVSBridge(br)
 		assert.Nil(t, err, fmt.Sprintf("Error while deleting OVS bridge: %v", err))
+		ofClient.CleanOFTableCache()
 	}()
 
 	ruleID := uint32(100)
@@ -530,7 +546,8 @@ func TestNetworkPolicyFlows(t *testing.T) {
 }
 
 func TestIPv6ConnectivityFlows(t *testing.T) {
-	// Initialize ovs metrics (Prometheus) to test them
+	// Reset OVS metrics (Prometheus) and reinitialize them to test.
+	legacyregistry.Reset()
 	metrics.InitializeOVSMetrics()
 
 	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, true, false, false, false, false)
@@ -542,6 +559,7 @@ func TestIPv6ConnectivityFlows(t *testing.T) {
 		assert.Nil(t, err, fmt.Sprintf("Error while disconnecting from OVS bridge: %v", err))
 		err = ofTestUtils.DeleteOVSBridge(br)
 		assert.Nil(t, err, fmt.Sprintf("Error while deleting OVS bridge: %v", err))
+		ofClient.CleanOFTableCache()
 	}()
 	config := prepareIPv6Configuration()
 	t.Run("testInitialize", func(t *testing.T) {
@@ -575,6 +593,10 @@ type svcConfig struct {
 }
 
 func TestProxyServiceFlows(t *testing.T) {
+	// Reset OVS metrics (Prometheus) and reinitialize them to test.
+	legacyregistry.Reset()
+	metrics.InitializeOVSMetrics()
+
 	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false, false, false, false, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
@@ -587,6 +609,7 @@ func TestProxyServiceFlows(t *testing.T) {
 		assert.Nil(t, err, fmt.Sprintf("Error while disconnecting from OVS bridge: %v", err))
 		err = ofTestUtils.DeleteOVSBridge(br)
 		assert.Nil(t, err, fmt.Sprintf("Error while deleting OVS bridge: %v", err))
+		ofClient.CleanOFTableCache()
 	}()
 
 	endpoints := []k8sproxy.Endpoint{
@@ -1514,6 +1537,10 @@ func prepareSNATFlows(snatIP net.IP, mark, podOFPort, podOFPortRemote uint32, vM
 }
 
 func TestSNATFlows(t *testing.T) {
+	// Reset OVS metrics (Prometheus) and reinitialize them to test.
+	legacyregistry.Reset()
+	metrics.InitializeOVSMetrics()
+
 	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, false, false, true, false, false, false, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
@@ -1527,6 +1554,7 @@ func TestSNATFlows(t *testing.T) {
 		assert.Nil(t, err, fmt.Sprintf("Error while disconnecting from OVS bridge: %v", err))
 		err = ofTestUtils.DeleteOVSBridge(br)
 		assert.Nil(t, err, fmt.Sprintf("Error while deleting OVS bridge: %v", err))
+		ofClient.CleanOFTableCache()
 	}()
 
 	snatIP := net.ParseIP("10.10.10.14")