Repository: antrea-io/antrea
Pull request #3734 (2022-05-06 03:21:38)

Bump docker/setup-qemu-action from 1 to 2
Bumps [docker/setup-qemu-action](https://github.com/docker/setup-qemu-action) from 1 to 2.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/docker/setup-qemu-action/releases">docker/setup-qemu-action's releases</a>.</em></p>
<blockquote>
<h2>v2.0.0</h2>
<ul>
<li>Node 16 as default runtime by <a href="https://github.com/crazy-max"><code>@​crazy-max</code></a> (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/48">#48</a>)
<ul>
<li>This requires a minimum <a href="https://github.com/actions/runner/releases/tag/v2.285.0">Actions Runner</a> version of v2.285.0, which is by default available in GHES 3.4 or later.</li>
</ul>
</li>
<li>chore: update dev dependencies and workflow by <a href="https://github.com/crazy-max"><code>@​crazy-max</code></a> (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/43">#43</a> <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/47">#47</a>)</li>
<li>Bump <code>@​actions/core</code> from 1.3.0 to 1.6.0 (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/37">#37</a> <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/39">#39</a> <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/41">#41</a>)</li>
<li>Bump <code>@​actions/exec</code> from 1.0.4 to 1.1.1 (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/38">#38</a> <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/46">#46</a>)</li>
</ul>
<p><strong>Full Changelog</strong>: <a href="https://github.com/docker/setup-qemu-action/compare/v1.2.0...v2.0.0">https://github.com/docker/setup-qemu-action/compare/v1.2.0...v2.0.0</a></p>
<h2>v1.2.0</h2>
<ul>
<li>Display image information (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/36">#36</a>)</li>
<li>Bump <code>@​actions/core</code> from 1.2.7 to 1.3.0 (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/35">#35</a>)</li>
</ul>
<h2>v1.1.0</h2>
<ul>
<li>Remove os limitation (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/30">#30</a>)</li>
<li>Bump <code>@​actions/core</code> from 1.2.6 to 1.2.7 (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/29">#29</a>)</li>
</ul>
<h2>v1.0.2</h2>
<ul>
<li>Enhance workflow (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/26">#26</a>)</li>
<li>Container based developer flow (<a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/19">#19</a> <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/20">#20</a>)</li>
</ul>
<h2>v1.0.1</h2>
<ul>
<li>Fix CVE-2020-15228</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/docker/setup-qemu-action/commit/8b122486cedac8393e77aa9734c3528886e4a1a8"><code>8b12248</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/48">#48</a> from crazy-max/node-16</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/466d53193ca9a892b43bd0f7fcacd4537854009f"><code>466d531</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/50">#50</a> from crazy-max/update-readme</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/607c1922b58acc78ad41e470aa7f061af7ef1f85"><code>607c192</code></a> simplify usage example</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/d7849ecb9c834e9164bac38b409bad6b31f9fd1f"><code>d7849ec</code></a> Node 16 as default runtime</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/2d4bfe71c9c7c6adb935665efb1a651cc8e799ba"><code>2d4bfe7</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/47">#47</a> from crazy-max/update-dev</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/224b802eb3ef627939fa7cc83532f673a8bec346"><code>224b802</code></a> chore: update dev dependencies and workflow</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/95bd86577812b34f1db6d5e57b728cb63b9aa388"><code>95bd865</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/46">#46</a> from docker/dependabot/npm_and_yarn/actions/exec-1.1.1</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/cfd091faa1bc4166b21275c814f26cfe27f6cb43"><code>cfd091f</code></a> Bump <code>@​actions/exec</code> from 1.1.0 to 1.1.1</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/d2a60302b87194ce3bff607f8a9af18fe5ec1cdc"><code>d2a6030</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/docker/setup-qemu-action/issues/45">#45</a> from docker/dependabot/github_actions/actions/checkout-3</li>
<li><a href="https://github.com/docker/setup-qemu-action/commit/97dc484a91e07bd7fe42cfff1919c2ea5ff6df50"><code>97dc484</code></a> Bump actions/checkout from 2 to 3</li>
<li>Additional commits viewable in <a href="https://github.com/docker/setup-qemu-action/compare/v1...v2">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=docker/setup-qemu-action&package-manager=github_actions&previous-version=1&new-version=2)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

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

[![Impacted file tree graph](https://codecov.io/gh/antrea-io/antrea/pull/3734/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/3734?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)

```diff
@@             Coverage Diff             @@
##             main    #3734       +/-   ##
===========================================
- Coverage   64.60%   43.76%   -20.84%     
===========================================
  Files         278      210       -68     
  Lines       39640    35927     -3713     
===========================================
- Hits        25608    15723     -9885     
- Misses      12043    19063     +7020     
+ Partials     1989     1141      -848     
```

| Flag | Coverage Δ | |
|---|---|---|
| kind-e2e-tests | `?` | |
| unit-tests | `43.76% <ø> (+0.02%)` | :arrow_up: |

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/3734?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pkg/ovs/openflow/default.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vcGVuZmxvdy9kZWZhdWx0Lmdv) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/util/runtime/runtime.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL3V0aWwvcnVudGltZS9ydW50aW1lLmdv) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/agent/openflow/groups.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L29wZW5mbG93L2dyb3Vwcy5nbw==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/ovs/openflow/ofctrl\_pipeline.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vcGVuZmxvdy9vZmN0cmxfcGlwZWxpbmUuZ28=) | `0.00% <0.00%> (-95.84%)` | :arrow_down: |
| [pkg/ovs/openflow/logs.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vcGVuZmxvdy9sb2dzLmdv) | `10.34% <0.00%> (-89.66%)` | :arrow_down: |
| [pkg/agent/controller/networkpolicy/reject.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/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%> (-85.89%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/k8s/annotations.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/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%> (-84.45%)` | :arrow_down: |
| [pkg/apis/controlplane/register.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXMvY29udHJvbHBsYW5lL3JlZ2lzdGVyLmdv) | `0.00% <0.00%> (-83.34%)` | :arrow_down: |
| [pkg/agent/client.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NsaWVudC5nbw==) | `0.00% <0.00%> (-76.75%)` | :arrow_down: |
| [pkg/flowaggregator/certificate.go](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2Zsb3dhZ2dyZWdhdG9yL2NlcnRpZmljYXRlLmdv) | `0.00% <0.00%> (-75.53%)` | :arrow_down: |
| ... and [196 more](https://codecov.io/gh/antrea-io/antrea/pull/3734/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | |


Patch:
diff --git a/.github/workflows/docker_update_ethtool.yml b/.github/workflows/docker_update_ethtool.yml
index 8ec7637d3..548176ac9 100644
--- a/.github/workflows/docker_update_ethtool.yml
+++ b/.github/workflows/docker_update_ethtool.yml
@@ -26,11 +26,11 @@ jobs:
       with:
         ref: ${{ github.event.inputs.antrea-ref }}
     - name: Set up QEMU
-      uses: docker/setup-qemu-action@v1
+      uses: docker/setup-qemu-action@v2
     - name: Set up Docker Buildx
       uses: docker/setup-buildx-action@v1
     - name: Login to DockerHub
-      uses: docker/login-action@v1
+      uses: docker/login-action@v2
       with:
         username: ${{ secrets.DOCKER_USERNAME }}
         password: ${{ secrets.DOCKER_PASSWORD }}
diff --git a/.github/workflows/docker_update_flow_visibility.yml b/.github/workflows/docker_update_flow_visibility.yml
index 0784213c8..95e818325 100644
--- a/.github/workflows/docker_update_flow_visibility.yml
+++ b/.github/workflows/docker_update_flow_visibility.yml
@@ -22,7 +22,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Login to DockerHub
-        uses: docker/login-action@v1
+        uses: docker/login-action@v2
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 6ca34583e..f4fa7ce75 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -34,7 +34,6 @@ contributing to project Antrea:
 2. Check out the [Architecture document](docs/design/architecture.md) for the Antrea
    architecture and design.
 3. Set up necessary [accounts](#accounts-setup).
-4. Set up your [development environment](docs/contributors/manual-installation.md)
 
 Now that you're setup, skip ahead to learn how to [contribute](#contribute).
 
diff --git a/build/charts/antrea/README.md b/build/charts/antrea/README.md
index 9005e7c0b..2b14782d0 100644
--- a/build/charts/antrea/README.md
+++ b/build/charts/antrea/README.md
@@ -61,7 +61,7 @@ Kubernetes: `>= 1.16.0-0`
 | controller.podLabels | object | `{}` | Labels to be added to antrea-controller Pod. |
 | controller.priorityClassName | string | `"system-cluster-critical"` | Prority class to use for the antrea-controller Pod. |
 | controller.selfSignedCert | bool | `true` | Indicates whether to use auto-generated self-signed TLS certificates. If false, a Secret named "antrea-controller-tls" must be provided with the following keys: ca.crt, tls.crt, tls.key. |
-| controller.tolerations | list | `[{"key":"CriticalAddonsOnly","operator":"Exists"},{"effect":"NoSchedule","key":"node-role.kubernetes.io/master"}]` | Tolerations for the antrea-controller Pod. |
+| controller.tolerations | list | `[{"key":"CriticalAddonsOnly","operator":"Exists"},{"effect":"NoSchedule","key":"node-role.kubernetes.io/master"},{"effect":"NoSchedule","key":"node-role.kubernetes.io/control-plane"}]` | Tolerations for the antrea-controller Pod. |
 | defaultMTU | int | `0` | Default MTU to use for the host gateway interface and the network interface of each Pod. By default, antrea-agent will discover the MTU of the Node's primary interface and adjust it to accommodate for tunnel encapsulation overhead if applicable. |
 | egress.exceptCIDRs | list | `[]` | CIDR ranges to which outbound Pod traffic will not be SNAT'd by Egresses. |
 | enableBridgingMode | bool | `false` | Enable bridging mode of Pod network on Nodes, in which the Node's transport interface is connected to the OVS bridge. |
diff --git a/build/charts/antrea/templates/agent/secret.yaml b/build/charts/antrea/templates/agent/secret.yaml
new file mode 100644
index 000000000..90053e027
--- /dev/null
+++ b/build/charts/antrea/templates/agent/secret.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: {{ .Release.Namespace }}
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
diff --git a/build/charts/antrea/templates/antctl/secret.yaml b/build/charts/antrea/templates/antctl/secret.yaml
new file mode 100644
index 000000000..c964bc87f
--- /dev/null
+++ b/build/charts/antrea/templates/antctl/secret.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: {{ .Release.Namespace }}
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
diff --git a/build/charts/antrea/templates/crds/clusternetworkpolicy.yaml b/build/charts/antrea/templates/crds/clusternetworkpolicy.yaml
index 307048661..a2a90da43 100644
--- a/build/charts/antrea/templates/crds/clusternetworkpolicy.yaml
+++ b/build/charts/antrea/templates/crds/clusternetworkpolicy.yaml
@@ -494,6 +494,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
diff --git a/build/charts/antrea/values.yaml b/build/charts/antrea/values.yaml
index 8a587f202..45a1297ba 100644
--- a/build/charts/antrea/values.yaml
+++ b/build/charts/antrea/values.yaml
@@ -203,6 +203,9 @@ controller:
     # Allow it to schedule onto master nodes.
     - key: node-role.kubernetes.io/master
       effect: NoSchedule
+    # Control-plane taint for Kubernetes >= 1.24.
+    - key: node-role.kubernetes.io/control-plane
+      effect: NoSchedule
   # -- Node selector for the antrea-controller Pod.
   nodeSelector:
     kubernetes.io/os: linux
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index c5fb6bfbe..6f04524d2 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -26,6 +26,26 @@ metadata:
   labels:
     app: antrea
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1245,6 +1265,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3723,6 +3745,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index 80487df05..9b6b2b40f 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -26,6 +26,26 @@ metadata:
   labels:
     app: antrea
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1245,6 +1265,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3725,6 +3747,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index a0ed8f188..9d6de7b5a 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -26,6 +26,26 @@ metadata:
   labels:
     app: antrea
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1245,6 +1265,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3723,6 +3745,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index d029f1a8e..681db390b 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -39,6 +39,26 @@ stringData:
   # Preshared Key used by IKE for authentication with peers.
   psk: "changeme"
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1258,6 +1278,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3772,6 +3794,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/antrea-prometheus.yml b/build/yamls/antrea-prometheus.yml
index 3ee508fa0..84a5bbbb7 100644
--- a/build/yamls/antrea-prometheus.yml
+++ b/build/yamls/antrea-prometheus.yml
@@ -12,6 +12,16 @@ metadata:
   name: prometheus
   namespace: monitoring
 ---
+# This is useful for testing and manual access to the /metrics endpoints
+apiVersion: v1
+kind: Secret
+metadata:
+  name: prometheus-service-account-token
+  namespace: monitoring
+  annotations:
+    kubernetes.io/service-account.name: prometheus
+type: kubernetes.io/service-account-token
+---
 # Authorize Prometheus to view Kubernetes cluster components for service discovery purposes
 # Authorize Prometheus to retrieve metrics
 apiVersion: rbac.authorization.k8s.io/v1
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index 1ceb4e174..d9465dfee 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -26,6 +26,26 @@ metadata:
   labels:
     app: antrea
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1245,6 +1265,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3723,6 +3745,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/cmd/antrea-agent/agent.go b/cmd/antrea-agent/agent.go
index 94a91d8fa..8cd98f60d 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -579,11 +579,13 @@ func run(o *Options) error {
 		}
 		mcastController := multicast.NewMulticastController(
 			ofClient,
+			v4GroupIDAllocator,
 			nodeConfig,
 			ifaceStore,
 			multicastSocket,
 			sets.NewString(append(o.config.MulticastInterfaces, nodeConfig.NodeTransportInterfaceName)...),
-			ovsBridgeClient)
+			ovsBridgeClient,
+			podUpdateChannel)
 		if err := mcastController.Initialize(); err != nil {
 			return err
 		}
diff --git a/docs/configuration.md b/docs/configuration.md
index 8a2d70c57..c443eb4d8 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -15,9 +15,9 @@ Use `antrea-agent -h` to see complete options.
 
 The `antrea-agent` configuration file specifies the agent configuration
 parameters. For all the agent configuration parameters of a Linux Node, refer to
-this [base configuration file](/build/yamls/base/conf/antrea-agent.conf).
+this [base configuration file](../build/charts/antrea/conf/antrea-agent.conf).
 For all the configuration parameters of a Windows Node, refer to this [base
-configuration file](/build/yamls/windows/base/conf/antrea-agent.conf)
+configuration file](../build/yamls/windows/base/conf/antrea-agent.conf)
 
 ## antrea-controller
 
@@ -34,7 +34,7 @@ Use `antrea-controller -h` to see complete options.
 
 The `antrea-controller` configuration file specifies the controller
 configuration parameters. For all the controller configuration parameters,
-refer to this [base configuration file](/build/yamls/base/conf/antrea-controller.conf).
+refer to this [base configuration file](../build/charts/antrea/conf/antrea-controller.conf).
 
 ## CNI configuration
 
diff --git a/docs/contributors/manual-installation.md b/docs/contributors/manual-installation.md
deleted file mode 100644
index 13aa595c2..000000000
--- a/docs/contributors/manual-installation.md
+++ /dev/null
@@ -1,171 +0,0 @@
-# Manual Installation
-
-## Overview
-
-There are four components which need to be deployed in order to run Antrea:
-
-* The OpenVSwitch daemons `ovs-vswitchd` and `ovsdb-server`
-
-* The controller `antrea-controller`
-
-* The agent `antrea-agent`
-
-* The CNI plugin `antrea-cni`
-
-## Instructions
-
-Prior to bringing up the individual components, follow the common steps:
-
-* Ensure Go v1.15 is [installed](https://golang.org/doc/install)
-
-* Git clone your forked Antrea repository and `cd` into the `antrea` directory
-
-    ```bash
-    git clone https://github.com/$user/antrea
-    cd antrea
-    ```
-
-* Build the binaries for all components under `bin` directory
-
-    ```bash
-    make bin
-    ```
-
-### OpenVSwitch
-
-Open vSwitch >= 2.8.0 userspace daemon `ovs-vswitchd` and `ovsdb-server` should run on all worker nodes. See
-[Installing Open vSwitch](https://docs.openvswitch.org/en/latest/intro/install/#installation-from-packages) for details.
-
-### antrea-controller
-
-`antrea-controller` is required to implement Kubernetes Network Policies. At any time, there should be only a single
-active replica of `antrea-controller`.
-
-1. Grant the `antrea-controller` ServiceAccount necessary permissions to Kubernetes APIs. You can apply
-[controller-rbac.yaml](/build/yamls/base/controller-rbac.yml) to do it.
-
-    ```bash
-    kubectl apply -f build/yamls/base/controller-rbac.yml
-    ```
-
-2. Create the kubeconfig file that contains the K8s APIServer endpoint and the token of ServiceAccount created in the
-above step. See [Configure Access to Multiple Clusters](
-https://kubernetes.io/docs/tasks/access-application-cluster/configure-access-multiple-clusters/) for more information.
-
-    ```bash
-    APISERVER=$(kubectl config view --minify -o jsonpath='{.clusters[0].cluster.server}')
-    TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antrea-controller')].data.token}"|base64 --decode)
-    kubectl config --kubeconfig=antrea-controller.kubeconfig set-cluster kubernetes --server=$APISERVER --insecure-skip-tls-verify
-    kubectl config --kubeconfig=antrea-controller.kubeconfig set-credentials antrea-controller --token=$TOKEN
-    kubectl config --kubeconfig=antrea-controller.kubeconfig set-context antrea-controller@kubernetes --cluster=kubernetes --user=antrea-controller
-    kubectl config --kubeconfig=antrea-controller.kubeconfig use-context antrea-controller@kubernetes
-    ```
-
-3. Create the `antrea-controller` config file, see [Configuration](../configuration.md) for details.
-
-    ```bash
-    cat >antrea-controller.conf <<EOF
-    clientConnection:
-      kubeconfig: antrea-controller.kubeconfig
-    EOF
-    ```
-
-4. Start `antrea-controller`.
-
-    ```bash
-    bin/antrea-controller --config antrea-controller.conf
-    ```
-
-### antrea-agent
-
-`antrea-agent` must run all worker nodes.
-
-1. Grant the `antrea-agent` ServiceAccount necessary permissions to Kubernetes APIs. You can apply [agent-rbac.yaml](
-/build/yamls/base/agent-rbac.yml) to do it.
-
-    ```bash
-    kubectl apply -f build/yamls/base/agent-rbac.yml
-    ```
-
-2. Create the kubeconfig file that contains the K8s APIServer endpoint and the token of ServiceAccount created in the
-above step. See [Configure Access to Multiple Clusters](
-https://kubernetes.io/docs/tasks/access-application-cluster/configure-access-multiple-clusters/) for more information.
-
-    ```bash
-    APISERVER=$(kubectl config view --minify -o jsonpath='{.clusters[0].cluster.server}')
-    TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antrea-agent')].data.token}"|base64 --decode)
-    kubectl config --kubeconfig=antrea-agent.kubeconfig set-cluster kubernetes --server=$APISERVER --insecure-skip-tls-verify
-    kubectl config --kubeconfig=antrea-agent.kubeconfig set-credentials antrea-agent --token=$TOKEN
-    kubectl config --kubeconfig=antrea-agent.kubeconfig set-context antrea-agent@kubernetes --cluster=kubernetes --user=antrea-agent
-    kubectl config --kubeconfig=antrea-agent.kubeconfig use-context antrea-agent@kubernetes
-    ```
-
-3. Create the kubeconfig file that contains the `antrea-controller` APIServer endpoint and the token of ServiceAccount
-created in the above step.
-
-    ```bash
-    # Change it to the correct endpoint if you are running antrea-controller somewhere else.
-    ANTREA_APISERVER=https://localhost
-    TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antrea-agent')].data.token}"|base64 --decode)
-    kubectl config --kubeconfig=antrea-agent.antrea.kubeconfig set-cluster antrea --server=$ANTREA_APISERVER --insecure-skip-tls-verify
-    kubectl config --kubeconfig=antrea-agent.antrea.kubeconfig set-credentials antrea-agent --token=$TOKEN
-    kubectl config --kubeconfig=antrea-agent.antrea.kubeconfig set-context antrea-agent@antrea --cluster=antrea --user=antrea-agent
-    kubectl config --kubeconfig=antrea-agent.antrea.kubeconfig use-context antrea-agent@antrea
-    ```
-
-4. Create the `antrea-agent` config file, see [Configuration](../configuration.md) for details.
-
-    ```bash
-    cat >antrea-agent.conf <<EOF
-    clientConnection:
-      kubeconfig: antrea-agent.kubeconfig
-    antreaClientConnection:
-      kubeconfig: antrea-agent.antrea.kubeconfig
-    hostProcPathPrefix: "/"
-    EOF
-    ```
-
-5. Start `antrea-agent`.
-
-    ```bash
-    bin/antrea-agent --config antrea-agent.conf
-    ```
-
-### antrea-cni
-
-`antrea-cni` should be installed on all worker nodes.
-
-1. Create the cni config file on all worker nodes.
-
-    ```bash
-    mkdir -p /etc/cni/net.d
-
-    cat >/etc/cni/net.d/10-antrea.conflist <<EOF
-    {
-      "cniVersion":"0.3.0",
-      "name": "antrea",
-      "plugins": [
-        {
-          "type": "antrea",
-          "ipam": {
-            "type": "host-local"
-          }
-        },
-        {
-          "type": "portmap",
-          "capabilities": {"portMappings": true}
-        },
-        {
-          "type": "bandwidth",
-          "capabilities": {"bandwidth": true}
-        }
-      ]
-    }
-    EOF
-    ```
-
-2. Install `antrea-cni` to `/opt/cni/bin/antrea`.
-
-    ```bash
-    cp bin/antrea-cni /opt/cni/bin/antrea
-    ```
diff --git a/docs/getting-started.md b/docs/getting-started.md
index fc892c6e3..758832507 100644
--- a/docs/getting-started.md
+++ b/docs/getting-started.md
@@ -146,10 +146,6 @@ or simply reboot your Nodes.
 To build the image locally, you can follow the instructions in the [Contributor
 Guide](../CONTRIBUTING.md#building-and-testing-your-change).
 
-Antrea components can also be run manually as processes for development
-purposes. See [Manual Installation](contributors/manual-installation.md) for
-information.
-
 ### Deploying Antrea in Kind
 
 To deploy Antrea in a [Kind](https://github.com/kubernetes-sigs/kind) cluster,
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
index ea449e886..65d717601 100644
--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -126,7 +126,7 @@ address and pass an authentication token when accessing it, like this:
 # Get the antrea Service address
 ANTREA_SVC=$(kubectl get service antrea -n kube-system -o jsonpath='{.spec.clusterIP}')
 # Get the token value of antctl account, you can use any ServiceAccount that has permissions to antrea API.
-TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antctl')].data.token}"|base64 --decode)
+TOKEN=$(kubectl get secret/antctl-service-account-token -n kube-system -o jsonpath="{.data.token}"|base64 --decode)
 # Access antrea API with TOKEN
 curl --insecure --header "Authorization: Bearer $TOKEN" https://$ANTREA_SVC/apis
 ```
@@ -176,13 +176,14 @@ using the authentication token of the `antctl` ServiceAccount:
 
 ```bash
 # Get the token value of antctl account.
-TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antctl')].data.token}"|base64 --decode)
+TOKEN=$(kubectl get secret/antctl-service-account-token -n kube-system -o jsonpath="{.data.token}"|base64 --decode)
 # Access antrea API with TOKEN
 curl --insecure --header "Authorization: Bearer $TOKEN" https://<Node IP address>:10350/podinterfaces
 ```
 
 However, in this case you will be limited to the endpoints that `antctl` is
-allowed to access, as defined [here](/build/yamls/base/antctl.yml).
+allowed to access, as defined
+[here](../build/charts/antrea/templates/antctl/clusterrole.yaml).
 
 ## Accessing the flow-aggregator API
 
diff --git a/hack/.notableofcontents b/hack/.notableofcontents
index 6f9811354..720e2f957 100644
--- a/hack/.notableofcontents
+++ b/hack/.notableofcontents
@@ -9,7 +9,6 @@ docs/contributors/code-generation.md
 docs/contributors/docker-desktop-alternatives.md
 docs/contributors/eks-terraform.md
 docs/contributors/github-labels.md
-docs/contributors/manual-installation.md
 docs/configuration.md
 docs/cookbooks/multus/README.md
 docs/cookbooks/multus/build/cni-dhcp-daemon/README.md
diff --git a/hack/windows/Helper.psm1 b/hack/windows/Helper.psm1
index 7f6ac3371..be7768113 100644
--- a/hack/windows/Helper.psm1
+++ b/hack/windows/Helper.psm1
@@ -127,7 +127,7 @@ function Install-AntreaAgent {
     $APIServer=$(kubectl --kubeconfig=$KubeConfig get service kubernetes -o jsonpath='{.spec.clusterIP}')
     $APIServerPort=$(kubectl --kubeconfig=$KubeConfig get service kubernetes -o jsonpath='{.spec.ports[0].port}')
     $APIServer="https://$APIServer" + ":" + $APIServerPort
-    $TOKEN=$(kubectl --kubeconfig=$KubeConfig get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antrea-agent')].data.token}")
+    $TOKEN=$(kubectl --kubeconfig=$KubeConfig get secret/antrea-agent-service-account-token -n kube-system -o jsonpath="{.data.token}")
     $TOKEN=$([System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($TOKEN)))
     kubectl config --kubeconfig=$AntreaEtc\antrea-agent.kubeconfig set-cluster kubernetes --server=$APIServer --insecure-skip-tls-verify
     kubectl config --kubeconfig=$AntreaEtc\antrea-agent.kubeconfig set-credentials antrea-agent --token=$TOKEN
diff --git a/pkg/agent/cniserver/pod_configuration.go b/pkg/agent/cniserver/pod_configuration.go
index 0c2027df6..f5d68c530 100644
--- a/pkg/agent/cniserver/pod_configuration.go
+++ b/pkg/agent/cniserver/pod_configuration.go
@@ -32,6 +32,7 @@ import (
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
 	"antrea.io/antrea/pkg/agent/secondarynetwork/cnipodcache"
+	agenttypes "antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
 	"antrea.io/antrea/pkg/util/channel"
@@ -312,6 +313,7 @@ func (pc *podConfigurator) removeInterfaces(containerID string) error {
 	if err := pc.routeClient.DeleteLocalAntreaFlexibleIPAMPodRule(containerConfig.IPs); err != nil {
 		return err
 	}
+
 	return nil
 }
 
@@ -495,7 +497,13 @@ func (pc *podConfigurator) connectInterfaceToOVSCommon(ovsPortName string, conta
 	// Add containerConfig into local cache
 	pc.ifaceStore.AddInterface(containerConfig)
 	// Notify the Pod update event to required components.
-	pc.podUpdateNotifier.Notify(k8s.NamespacedName(containerConfig.PodNamespace, containerConfig.PodName))
+	event := agenttypes.PodUpdate{
+		PodName:      containerConfig.PodName,
+		PodNamespace: containerConfig.PodNamespace,
+		IsAdd:        true,
+		ContainerID:  containerConfig.ContainerID,
+	}
+	pc.podUpdateNotifier.Notify(event)
 	return nil
 }
 
@@ -518,6 +526,13 @@ func (pc *podConfigurator) disconnectInterfaceFromOVS(containerConfig *interface
 	}
 	// Remove container configuration from cache.
 	pc.ifaceStore.DeleteInterface(containerConfig)
+	event := agenttypes.PodUpdate{
+		PodName:      containerConfig.PodName,
+		PodNamespace: containerConfig.PodNamespace,
+		IsAdd:        false,
+		ContainerID:  containerConfig.ContainerID,
+	}
+	pc.podUpdateNotifier.Notify(event)
 	klog.Infof("Removed interfaces for container %s", containerID)
 	return nil
 }
diff --git a/pkg/agent/cniserver/pod_configuration_windows.go b/pkg/agent/cniserver/pod_configuration_windows.go
index a2dc28340..16f824f84 100644
--- a/pkg/agent/cniserver/pod_configuration_windows.go
+++ b/pkg/agent/cniserver/pod_configuration_windows.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/interfacestore"
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	"antrea.io/antrea/pkg/util/k8s"
 )
@@ -49,7 +50,13 @@ func (pc *podConfigurator) connectInterfaceToOVSAsync(ifConfig *interfacestore.I
 		// Update interface config with the ofPort.
 		ifConfig.OVSPortConfig.OFPort = ofPort
 		// Notify the Pod update event to required components.
-		pc.podUpdateNotifier.Notify(k8s.NamespacedName(ifConfig.PodNamespace, ifConfig.PodName))
+		event := types.PodUpdate{
+			PodName:      ifConfig.PodName,
+			PodNamespace: ifConfig.PodNamespace,
+			IsAdd:        true,
+			ContainerID:  ifConfig.ContainerID,
+		}
+		pc.podUpdateNotifier.Notify(event)
 		return nil
 	})
 }
diff --git a/pkg/agent/controller/egress/egress_controller.go b/pkg/agent/controller/egress/egress_controller.go
index 167d0756a..a683a8d84 100644
--- a/pkg/agent/controller/egress/egress_controller.go
+++ b/pkg/agent/controller/egress/egress_controller.go
@@ -41,6 +41,7 @@ import (
 	"antrea.io/antrea/pkg/agent/memberlist"
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
+	"antrea.io/antrea/pkg/agent/types"
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	clientsetversioned "antrea.io/antrea/pkg/client/clientset/versioned"
@@ -217,9 +218,11 @@ func NewEgressController(
 
 // processPodUpdate will be called when CNIServer publishes a Pod update event.
 // It triggers reconciling the effective Egress of the Pod.
-func (c *EgressController) processPodUpdate(pod string) {
+func (c *EgressController) processPodUpdate(e interface{}) {
 	c.egressBindingsMutex.Lock()
 	defer c.egressBindingsMutex.Unlock()
+	podEvent := e.(types.PodUpdate)
+	pod := k8s.NamespacedName(podEvent.PodNamespace, podEvent.PodName)
 	binding, exists := c.egressBindings[pod]
 	if !exists {
 		return
diff --git a/pkg/agent/controller/networkpolicy/cache.go b/pkg/agent/controller/networkpolicy/cache.go
index 32011eaf8..e0fc264f2 100644
--- a/pkg/agent/controller/networkpolicy/cache.go
+++ b/pkg/agent/controller/networkpolicy/cache.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/metrics"
+	agenttypes "antrea.io/antrea/pkg/agent/types"
 	v1beta "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
 	"antrea.io/antrea/pkg/querier"
@@ -361,12 +362,12 @@ func newRuleCache(dirtyRuleHandler func(string), podUpdateSubscriber channel.Sub
 // done if antrea-controller has computed the Pods' policies and propagated
 // them to this Node by their labels and NodeName, instead of waiting for their
 // IPs are reported to kube-apiserver and processed by antrea-controller.
-func (c *ruleCache) processPodUpdate(pod string) {
-	namespace, name := k8s.SplitNamespacedName(pod)
+func (c *ruleCache) processPodUpdate(e interface{}) {
+	podEvent := e.(agenttypes.PodUpdate)
 	member := &v1beta.GroupMember{
 		Pod: &v1beta.PodReference{
-			Name:      name,
-			Namespace: namespace,
+			Name:      podEvent.PodName,
+			Namespace: podEvent.PodNamespace,
 		},
 	}
 	c.appliedToSetLock.RLock()
diff --git a/pkg/agent/multicast/mcast_controller.go b/pkg/agent/multicast/mcast_controller.go
index c11a92337..481dcebb2 100644
--- a/pkg/agent/multicast/mcast_controller.go
+++ b/pkg/agent/multicast/mcast_controller.go
@@ -28,7 +28,11 @@ import (
 	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/interfacestore"
 	"antrea.io/antrea/pkg/agent/openflow"
+	"antrea.io/antrea/pkg/agent/types"
+	"antrea.io/antrea/pkg/agent/util"
+	binding "antrea.io/antrea/pkg/ovs/openflow"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 type eventType uint8
@@ -59,10 +63,11 @@ type mcastGroupEvent struct {
 
 type GroupMemberStatus struct {
 	group net.IP
-	// localMembers is a set for the local Pod's interface name which has joined in the multicast group.
-	localMembers   sets.String
+	// localMembers is a map for the local Pod member and its last update time, key is the Pod's interface name,
+	// and value is its last update time.
+	localMembers   map[string]time.Time
 	lastIGMPReport time.Time
-	mutex          sync.RWMutex
+	ofGroupID      binding.GroupIDType
 }
 
 // eventHandler process the multicast Group membership report or leave messages.
@@ -82,7 +87,8 @@ func (c *Controller) addGroupMemberStatus(e *mcastGroupEvent) {
 	status := &GroupMemberStatus{
 		group:          e.group,
 		lastIGMPReport: e.time,
-		localMembers:   sets.NewString(e.iface.InterfaceName),
+		localMembers:   map[string]time.Time{e.iface.InterfaceName: e.time},
+		ofGroupID:      c.v4GroupAllocator.Allocate(),
 	}
 	c.groupCache.Add(status)
 	c.queue.Add(e.group.String())
@@ -97,29 +103,37 @@ func (c *Controller) addGroupMemberStatus(e *mcastGroupEvent) {
 // is sent out to check if there are still local members in the group.
 func (c *Controller) updateGroupMemberStatus(obj interface{}, e *mcastGroupEvent) {
 	status := obj.(*GroupMemberStatus)
-	status.mutex.Lock()
-	defer status.mutex.Unlock()
 	newStatus := &GroupMemberStatus{
 		group:          status.group,
-		localMembers:   status.localMembers.Union(nil),
+		localMembers:   make(map[string]time.Time),
 		lastIGMPReport: status.lastIGMPReport,
+		ofGroupID:      status.ofGroupID,
 	}
-	exist := status.localMembers.Has(e.iface.InterfaceName)
+	for m, t := range status.localMembers {
+		newStatus.localMembers[m] = t
+	}
+	_, exist := status.localMembers[e.iface.InterfaceName]
 	switch e.eType {
 	case groupJoin:
 		newStatus.lastIGMPReport = e.time
+		newStatus.localMembers[e.iface.InterfaceName] = e.time
+		c.groupCache.Update(newStatus)
 		if !exist {
-			newStatus.localMembers.Insert(e.iface.InterfaceName)
 			klog.InfoS("Added member to multicast group", "group", e.group.String(), "member", e.iface.InterfaceName)
+			c.queue.Add(newStatus.group.String())
 		}
-		c.groupCache.Update(newStatus)
-		c.queue.Add(newStatus.group.String())
 	case groupLeave:
 		if exist {
-			newStatus.localMembers.Delete(e.iface.InterfaceName)
+			delete(newStatus.localMembers, e.iface.InterfaceName)
 			c.groupCache.Update(newStatus)
 			klog.InfoS("Deleted member from multicast group", "group", e.group.String(), "member", e.iface.InterfaceName)
-			if len(newStatus.localMembers) == 0 {
+			_, found := c.ifaceStore.GetInterfaceByName(e.iface.InterfaceName)
+			// Notify worker immediately about the member leave event if the member doesn't exist on the Node, or there are
+			// other local members in the multicast group.
+			if !found || len(newStatus.localMembers) > 0 {
+				c.queue.Add(newStatus.group.String())
+			} else {
+				// Check if all local members have left the multicast group.
 				klog.InfoS("Check last member in multicast group", "group", e.group.String(), "member", e.iface.InterfaceName)
 				c.checkLastMember(e.group)
 			}
@@ -139,28 +153,70 @@ func (c *Controller) checkLastMember(group net.IP) {
 	c.queue.AddAfter(group.String(), igmpMaxResponseTime)
 }
 
-// clearStaleGroups checks the stale groups which have not been updated for mcastGroupTimeout, and then notifies worker
+// clearStaleGroups checks the stale group members which have not been updated for mcastGroupTimeout, and then notifies worker
 // to remove them from groupCache.
 func (c *Controller) clearStaleGroups() {
 	now := time.Now()
 	for _, obj := range c.groupCache.List() {
 		status := obj.(*GroupMemberStatus)
-		status.mutex.RLock()
 		diff := now.Sub(status.lastIGMPReport)
-		status.mutex.RUnlock()
 		if diff > mcastGroupTimeout {
+			// Notify worker to remove the group from groupCache if all its members are not updated before mcastGroupTimeout.
 			c.queue.Add(status.group.String())
+		} else {
+			// Create a "leave" event for a local member if it is not updated before mcastGroupTimeout.
+			for member, lastUpdate := range status.localMembers {
+				if now.Sub(lastUpdate) > mcastGroupTimeout {
+					ifConfig := &interfacestore.InterfaceConfig{
+						InterfaceName: member,
+					}
+					event := &mcastGroupEvent{
+						group: status.group,
+						eType: groupLeave,
+						time:  now,
+						iface: ifConfig,
+					}
+					c.groupEventCh <- event
+				}
+			}
 		}
 	}
 }
 
+// removeLocalInterface searches the GroupMemberStatus which the deleted interface has joined, and then triggers a member
+// leave event so that Antrea can remove the corresponding interface from local multicast receivers on OVS. This function
+// should be called if the removed Pod receiver fails to send IGMP leave message before deletion.
+func (c *Controller) removeLocalInterface(e interface{}) {
+	podEvent := e.(types.PodUpdate)
+	// Ignore Pod creation event.
+	if podEvent.IsAdd {
+		return
+	}
+	interfaceName := util.GenerateContainerInterfaceName(podEvent.PodName, podEvent.PodNamespace, podEvent.ContainerID)
+	ifConfig := &interfacestore.InterfaceConfig{
+		InterfaceName: interfaceName,
+	}
+	groupStatuses := c.getGroupMemberStatusesByPod(interfaceName)
+	for _, g := range groupStatuses {
+		event := &mcastGroupEvent{
+			group: g.group,
+			eType: groupLeave,
+			time:  time.Now(),
+			iface: ifConfig,
+		}
+		c.groupEventCh <- event
+	}
+}
+
 type Controller struct {
-	ofClient     openflow.Client
-	nodeConfig   *config.NodeConfig
-	igmpSnooper  *IGMPSnooper
-	groupEventCh chan *mcastGroupEvent
-	groupCache   cache.Indexer
-	queue        workqueue.RateLimitingInterface
+	ofClient         openflow.Client
+	v4GroupAllocator openflow.GroupAllocator
+	ifaceStore       interfacestore.InterfaceStore
+	nodeConfig       *config.NodeConfig
+	igmpSnooper      *IGMPSnooper
+	groupEventCh     chan *mcastGroupEvent
+	groupCache       cache.Indexer
+	queue            workqueue.RateLimitingInterface
 	// installedGroups saves the groups which are configured on both OVS and the host.
 	installedGroups      sets.String
 	installedGroupsMutex sync.RWMutex
@@ -168,33 +224,39 @@ type Controller struct {
 	ovsBridgeClient      ovsconfig.OVSBridgeClient
 }
 
-func NewMulticastController(ofClient openflow.Client, nodeConfig *config.NodeConfig, ifaceStore interfacestore.InterfaceStore, multicastSocket RouteInterface, multicastInterfaces sets.String, ovsBridgeClient ovsconfig.OVSBridgeClient) *Controller {
+func NewMulticastController(ofClient openflow.Client,
+	v4GroupAllocator openflow.GroupAllocator,
+	nodeConfig *config.NodeConfig,
+	ifaceStore interfacestore.InterfaceStore,
+	multicastSocket RouteInterface,
+	multicastInterfaces sets.String,
+	ovsBridgeClient ovsconfig.OVSBridgeClient,
+	podUpdateSubscriber channel.Subscriber) *Controller {
 	eventCh := make(chan *mcastGroupEvent, workerCount)
 	groupSnooper := newSnooper(ofClient, ifaceStore, eventCh)
 	groupCache := cache.NewIndexer(getGroupEventKey, cache.Indexers{
 		podInterfaceIndex: podInterfaceIndexFunc,
 	})
 	multicastRouteClient := newRouteClient(nodeConfig, groupCache, multicastSocket, multicastInterfaces)
-	return &Controller{
-		ofClient:        ofClient,
-		nodeConfig:      nodeConfig,
-		igmpSnooper:     groupSnooper,
-		groupEventCh:    eventCh,
-		groupCache:      groupCache,
-		installedGroups: sets.NewString(),
-		queue:           workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "multicastgroup"),
-		mRouteClient:    multicastRouteClient,
-		ovsBridgeClient: ovsBridgeClient,
+	c := &Controller{
+		ofClient:         ofClient,
+		ifaceStore:       ifaceStore,
+		v4GroupAllocator: v4GroupAllocator,
+		nodeConfig:       nodeConfig,
+		igmpSnooper:      groupSnooper,
+		groupEventCh:     eventCh,
+		groupCache:       groupCache,
+		installedGroups:  sets.NewString(),
+		queue:            workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "multicastgroup"),
+		mRouteClient:     multicastRouteClient,
+		ovsBridgeClient:  ovsBridgeClient,
 	}
+	podUpdateSubscriber.Subscribe(c.removeLocalInterface)
+	return c
 }
 
 func (c *Controller) Initialize() error {
-	err := c.configureOVSMulticast()
-	if err != nil {
-		klog.ErrorS(err, "Failed to configure multicast for the OVS bridge")
-		return err
-	}
-	err = c.mRouteClient.Initialize()
+	err := c.mRouteClient.Initialize()
 	if err != nil {
 		return err
 	}
@@ -221,7 +283,6 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
 
 	// Periodically check the group member status, and remove the groups in which no members exist
 	go wait.NonSlidingUntil(c.clearStaleGroups, queryInterval, stopCh)
-
 	go c.eventHandler(stopCh)
 
 	for i := 0; i < int(workerCount); i++ {
@@ -298,15 +359,28 @@ func (c *Controller) syncGroup(groupKey string) error {
 		return nil
 	}
 	status := obj.(*GroupMemberStatus)
+	memberPorts := make([]uint32, 0, len(status.localMembers))
+	for memberInterfaceName := range status.localMembers {
+		obj, found := c.ifaceStore.GetInterfaceByName(memberInterfaceName)
+		if !found {
+			klog.InfoS("Failed to find interface from cache", "interface", memberInterfaceName)
+			continue
+		}
+		memberPorts = append(memberPorts, uint32(obj.OFPort))
+	}
 	if c.groupHasInstalled(groupKey) {
-		status.mutex.Lock()
-		defer status.mutex.Unlock()
 		if c.groupIsStale(status) {
 			// Remove the multicast flow entry if no local Pod is in the group.
-			if err := c.ofClient.UninstallMulticastFlow(status.group); err != nil {
+			if err := c.ofClient.UninstallMulticastFlows(status.group); err != nil {
 				klog.ErrorS(err, "Failed to uninstall multicast flows", "group", groupKey)
 				return err
 			}
+			// Remove the multicast flow entry if no local Pod is in the group.
+			if err := c.ofClient.UninstallGroup(status.ofGroupID); err != nil {
+				klog.ErrorS(err, "Failed to uninstall multicast group", "group", groupKey)
+				return err
+			}
+			c.v4GroupAllocator.Release(status.ofGroupID)
 			err := c.mRouteClient.deleteInboundMrouteEntryByGroup(status.group)
 			if err != nil {
 				klog.ErrorS(err, "Cannot delete multicast group", "group", groupKey)
@@ -320,10 +394,22 @@ func (c *Controller) syncGroup(groupKey string) error {
 			c.delInstalledGroup(groupKey)
 			c.groupCache.Delete(status)
 			klog.InfoS("Removed multicast group from cache after all members left", "group", groupKey)
+			return nil
 		}
+		// Reinstall OpenFlow group because the local Pod receivers have changed.
+		if err := c.ofClient.InstallMulticastGroup(status.ofGroupID, memberPorts); err != nil {
+			return err
+		}
+		klog.V(2).InfoS("Updated OpenFlow group for local receivers", "group", groupKey, "ofGroup", status.ofGroupID, "localReceivers", memberPorts)
 		return nil
 	}
-	if err := c.ofClient.InstallMulticastFlow(status.group); err != nil {
+	// Install OpenFlow group for a new multicast group which has local Pod receivers joined.
+	if err := c.ofClient.InstallMulticastGroup(status.ofGroupID, memberPorts); err != nil {
+		return err
+	}
+	klog.V(2).InfoS("Installed OpenFlow group for local receivers", "group", groupKey, "ofGroup", status.ofGroupID, "localReceivers", memberPorts)
+	// Install OpenFlow flow to forward packets to local Pod receivers which are included in the group.
+	if err := c.ofClient.InstallMulticastFlows(status.group, status.ofGroupID); err != nil {
 		klog.ErrorS(err, "Failed to install multicast flows", "group", status.group)
 		return err
 	}
@@ -388,21 +474,6 @@ func podInterfaceIndexFunc(obj interface{}) ([]string, error) {
 	return podInterfaces, nil
 }
 
-func (c *Controller) configureOVSMulticast() error {
-	// Configure bridge to enable multicast snooping.
-	err := c.ovsBridgeClient.SetBridgeMcastSnooping(true)
-	if err != nil {
-		return err
-	}
-	// Disable flooding of unregistered multicast packets to all ports.
-	otherConfig := map[string]interface{}{"mcast-snooping-disable-flood-unregistered": "true"}
-	err = c.ovsBridgeClient.AddBridgeOtherConfig(otherConfig)
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
 func getGroupEventKey(obj interface{}) (string, error) {
 	groupState := obj.(*GroupMemberStatus)
 	return groupState.group.String(), nil
diff --git a/pkg/agent/multicast/mcast_discovery.go b/pkg/agent/multicast/mcast_discovery.go
index 5f418513c..7d4b7fe91 100644
--- a/pkg/agent/multicast/mcast_discovery.go
+++ b/pkg/agent/multicast/mcast_discovery.go
@@ -131,7 +131,7 @@ func (s *IGMPSnooper) processPacketIn(pktIn *ofctrl.PacketIn) error {
 		fallthrough
 	case protocol.IGMPv2Report:
 		mgroup := igmp.(*protocol.IGMPv1or2).GroupAddress
-		klog.InfoS("Received IGMPv1or2 Report message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName)
+		klog.V(2).InfoS("Received IGMPv1or2 Report message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName)
 		event := &mcastGroupEvent{
 			group: mgroup,
 			eType: groupJoin,
@@ -143,7 +143,7 @@ func (s *IGMPSnooper) processPacketIn(pktIn *ofctrl.PacketIn) error {
 		msg := igmp.(*protocol.IGMPv3MembershipReport)
 		for _, gr := range msg.GroupRecords {
 			mgroup := gr.MulticastAddress
-			klog.InfoS("Received IGMPv3 Report message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName, "recordType", gr.Type, "sourceCount", gr.NumberOfSources)
+			klog.V(2).InfoS("Received IGMPv3 Report message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName, "recordType", gr.Type, "sourceCount", gr.NumberOfSources)
 			evtType := groupJoin
 			if (gr.Type == protocol.IGMPIsIn || gr.Type == protocol.IGMPToIn) && gr.NumberOfSources == 0 {
 				evtType = groupLeave
@@ -159,7 +159,7 @@ func (s *IGMPSnooper) processPacketIn(pktIn *ofctrl.PacketIn) error {
 
 	case protocol.IGMPv2LeaveGroup:
 		mgroup := igmp.(*protocol.IGMPv1or2).GroupAddress
-		klog.InfoS("Received IGMPv2 Leave message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName)
+		klog.V(2).InfoS("Received IGMPv2 Leave message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName)
 		event := &mcastGroupEvent{
 			group: mgroup,
 			eType: groupLeave,
diff --git a/pkg/agent/openflow/client.go b/pkg/agent/openflow/client.go
index 076306829..28980801b 100644
--- a/pkg/agent/openflow/client.go
+++ b/pkg/agent/openflow/client.go
@@ -83,9 +83,9 @@ type Client interface {
 	// InstallServiceGroup installs a group for Service LB. Each endpoint
 	// is a bucket of the group. For now, each bucket has the same weight.
 	InstallServiceGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints []proxy.Endpoint) error
-	// UninstallServiceGroup removes the group and its buckets that are
-	// installed by InstallServiceGroup.
-	UninstallServiceGroup(groupID binding.GroupIDType) error
+	// UninstallGroup removes the group and its buckets that are
+	// installed by InstallServiceGroup or InstallMulticastGroup.
+	UninstallGroup(groupID binding.GroupIDType) error
 
 	// InstallEndpointFlows installs flows for accessing Endpoints.
 	// If an Endpoint is on the current Node, then flows for hairpin and endpoint
@@ -269,11 +269,11 @@ type Client interface {
 	// InstallMulticastInitialFlows installs OpenFlow to packetIn the IGMP messages and output the Multicast traffic to
 	// antrea-gw0 so that local Pods could access external Multicast servers.
 	InstallMulticastInitialFlows(pktInReason uint8) error
-	// InstallMulticastFlow installs the flow to forward Multicast traffic normally, and output it to antrea-gw0
+	// InstallMulticastFlows installs the flow to forward Multicast traffic normally, and output it to antrea-gw0
 	// to ensure it can be forwarded to the external addresses.
-	InstallMulticastFlow(multicastIP net.IP) error
-	// UninstallMulticastFlow removes the flow matching the given multicastIP.
-	UninstallMulticastFlow(multicastIP net.IP) error
+	InstallMulticastFlows(multicastIP net.IP, groupID binding.GroupIDType) error
+	// UninstallMulticastFlows removes the flow matching the given multicastIP.
+	UninstallMulticastFlows(multicastIP net.IP) error
 	// SendIGMPQueryPacketOut sends the IGMPQuery packet as a packet-out to OVS from the gateway port.
 	SendIGMPQueryPacketOut(
 		dstMAC net.HardwareAddr,
@@ -292,6 +292,8 @@ type Client interface {
 
 	// UninstallTrafficControlReturnPortFlow removes the flow to classify the packets from a return port.
 	UninstallTrafficControlReturnPortFlow(returnOFPort uint32) error
+
+	InstallMulticastGroup(ofGroupID binding.GroupIDType, localReceivers []uint32) error
 }
 
 // GetFlowTableStatus returns an array of flow table status.
@@ -545,7 +547,7 @@ func (c *client) InstallServiceGroup(groupID binding.GroupIDType, withSessionAff
 	return nil
 }
 
-func (c *client) UninstallServiceGroup(groupID binding.GroupIDType) error {
+func (c *client) UninstallGroup(groupID binding.GroupIDType) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	if !c.bridge.DeleteGroup(groupID) {
@@ -744,7 +746,7 @@ func (c *client) generatePipelines() {
 
 	if c.enableMulticast {
 		// TODO: add support for IPv6 protocol
-		c.featureMulticast = newFeatureMulticast(c.cookieAllocator, []binding.Protocol{binding.ProtocolIP})
+		c.featureMulticast = newFeatureMulticast(c.cookieAllocator, []binding.Protocol{binding.ProtocolIP}, c.bridge)
 		c.activatedFeatures = append(c.activatedFeatures, c.featureMulticast)
 	}
 	c.featureTraceflow = newFeatureTraceflow()
@@ -834,6 +836,9 @@ func (c *client) ReplayFlows() {
 	}
 
 	c.featureService.replayGroups()
+	if c.enableMulticast {
+		c.featureMulticast.replayGroups()
+	}
 
 	for _, activeFeature := range c.activatedFeatures {
 		if err := c.ofEntryOperations.AddAll(activeFeature.replayFlows()); err != nil {
@@ -1109,15 +1114,15 @@ func (c *client) InstallMulticastInitialFlows(pktInReason uint8) error {
 	return c.addFlows(c.featureMulticast.cachedFlows, cacheKey, flows)
 }
 
-func (c *client) InstallMulticastFlow(multicastIP net.IP) error {
-	flows := c.featureMulticast.localMulticastForwardFlow(multicastIP)
+func (c *client) InstallMulticastFlows(multicastIP net.IP, groupID binding.GroupIDType) error {
+	flows := c.featureMulticast.localMulticastForwardFlows(multicastIP, groupID)
 	cacheKey := fmt.Sprintf("multicast_%s", multicastIP.String())
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	return c.addFlows(c.featureMulticast.cachedFlows, cacheKey, flows)
 }
 
-func (c *client) UninstallMulticastFlow(multicastIP net.IP) error {
+func (c *client) UninstallMulticastFlows(multicastIP net.IP) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	cacheKey := fmt.Sprintf("multicast_%s", multicastIP.String())
@@ -1173,3 +1178,14 @@ func (c *client) UninstallTrafficControlReturnPortFlow(returnOFPort uint32) erro
 	defer c.replayMutex.RUnlock()
 	return c.deleteFlows(c.featurePodConnectivity.tcCachedFlows, cacheKey)
 }
+
+func (c *client) InstallMulticastGroup(groupID binding.GroupIDType, localReceivers []uint32) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+
+	targetPorts := append([]uint32{config.HostGatewayOFPort}, localReceivers...)
+	if err := c.featureMulticast.multicastReceiversGroup(groupID, targetPorts...); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/pkg/agent/openflow/framework.go b/pkg/agent/openflow/framework.go
index 16c9392aa..8007bc976 100644
--- a/pkg/agent/openflow/framework.go
+++ b/pkg/agent/openflow/framework.go
@@ -245,7 +245,8 @@ func (f *featureEgress) getRequiredTables() []*Table {
 
 func (f *featureMulticast) getRequiredTables() []*Table {
 	return []*Table{
-		MulticastTable,
+		MulticastRoutingTable,
+		MulticastOutputTable,
 	}
 }
 
diff --git a/pkg/agent/openflow/multicast.go b/pkg/agent/openflow/multicast.go
index 3d416aa71..8a2adedfc 100644
--- a/pkg/agent/openflow/multicast.go
+++ b/pkg/agent/openflow/multicast.go
@@ -15,7 +15,11 @@
 package openflow
 
 import (
+	"fmt"
 	"net"
+	"sync"
+
+	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/openflow/cookie"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
@@ -26,8 +30,10 @@ var _, mcastCIDR, _ = net.ParseCIDR("224.0.0.0/4")
 type featureMulticast struct {
 	cookieAllocator cookie.Allocator
 	ipProtocols     []binding.Protocol
+	bridge          binding.Bridge
 
 	cachedFlows *flowCategoryCache
+	groupCache  sync.Map
 
 	category cookie.Category
 }
@@ -36,12 +42,14 @@ func (f *featureMulticast) getFeatureName() string {
 	return "Multicast"
 }
 
-func newFeatureMulticast(cookieAllocator cookie.Allocator, ipProtocols []binding.Protocol) *featureMulticast {
+func newFeatureMulticast(cookieAllocator cookie.Allocator, ipProtocols []binding.Protocol, bridge binding.Bridge) *featureMulticast {
 	return &featureMulticast{
 		cookieAllocator: cookieAllocator,
 		ipProtocols:     ipProtocols,
 		cachedFlows:     newFlowCategoryCache(),
+		bridge:          bridge,
 		category:        cookie.Multicast,
+		groupCache:      sync.Map{},
 	}
 }
 
@@ -56,10 +64,48 @@ func multicastPipelineClassifyFlow(cookieID uint64, pipeline binding.Pipeline) b
 }
 
 func (f *featureMulticast) initFlows() []binding.Flow {
-	return []binding.Flow{}
+	cookieID := f.cookieAllocator.Request(f.category).Raw()
+	return []binding.Flow{
+		f.multicastOutputFlow(cookieID),
+	}
 }
 
 func (f *featureMulticast) replayFlows() []binding.Flow {
 	// Get cached flows.
 	return getCachedFlows(f.cachedFlows)
 }
+
+func (f *featureMulticast) multicastReceiversGroup(groupID binding.GroupIDType, ports ...uint32) error {
+	group := f.bridge.CreateGroupTypeAll(groupID).ResetBuckets()
+	for i := range ports {
+		group = group.Bucket().
+			LoadToRegField(OFPortFoundRegMark.GetField(), OFPortFoundRegMark.GetValue()).
+			LoadToRegField(TargetOFPortField, ports[i]).
+			ResubmitToTable(MulticastRoutingTable.GetNext()).
+			Done()
+	}
+	if err := group.Add(); err != nil {
+		return fmt.Errorf("error when installing Multicast receiver Group: %w", err)
+	}
+	f.groupCache.Store(groupID, group)
+	return nil
+}
+
+func (f *featureMulticast) multicastOutputFlow(cookieID uint64) binding.Flow {
+	return MulticastOutputTable.ofTable.BuildFlow(priorityNormal).
+		Cookie(cookieID).
+		MatchRegMark(OFPortFoundRegMark).
+		Action().OutputToRegField(TargetOFPortField).
+		Done()
+}
+
+func (f *featureMulticast) replayGroups() {
+	f.groupCache.Range(func(id, value interface{}) bool {
+		group := value.(binding.Group)
+		group.Reset()
+		if err := group.Add(); err != nil {
+			klog.ErrorS(err, "Error when replaying cached group", "group", id)
+		}
+		return true
+	})
+}
diff --git a/pkg/agent/openflow/pipeline.go b/pkg/agent/openflow/pipeline.go
index 0c3539cf9..d609d2f67 100644
--- a/pkg/agent/openflow/pipeline.go
+++ b/pkg/agent/openflow/pipeline.go
@@ -180,7 +180,10 @@ var (
 	// Tables of pipelineMulticast are declared below. Do don't declare any tables of other pipelines here!
 
 	// Tables in stageRouting:
-	MulticastTable = newTable("Multicast", stageRouting, pipelineMulticast)
+	MulticastRoutingTable = newTable("MulticastRouting", stageRouting, pipelineMulticast)
+
+	// Tables in stageOutput
+	MulticastOutputTable = newTable("MulticastOutput", stageOutput, pipelineMulticast)
 
 	// Flow priority level
 	priorityHigh            = uint16(210)
@@ -2637,8 +2640,8 @@ func pipelineClassifyFlow(cookieID uint64, protocol binding.Protocol, pipeline b
 		Done()
 }
 
-// igmpPktInFlows generates the flow to load CustomReasonIGMPRegMark to mark the IGMP packet in MulticastTable and sends
-// it to antrea-agent on MulticastTable.
+// igmpPktInFlows generates the flow to load CustomReasonIGMPRegMark to mark the IGMP packet in MulticastRoutingTable and sends
+// it to antrea-agent.
 func (f *featureMulticast) igmpPktInFlows(reason uint8) []binding.Flow {
 	flows := []binding.Flow{
 		// Set a custom reason for the IGMP packets, and then send it to antrea-agent and forward it normally in the
@@ -2646,7 +2649,7 @@ func (f *featureMulticast) igmpPktInFlows(reason uint8) []binding.Flow {
 		// group and its members in the meanwhile.
 		// Do not set dst IP address because IGMPv1 report message uses target multicast group as IP destination in
 		// the packet.
-		MulticastTable.ofTable.BuildFlow(priorityHigh).
+		MulticastRoutingTable.ofTable.BuildFlow(priorityHigh).
 			Cookie(f.cookieAllocator.Request(f.category).Raw()).
 			MatchProtocol(binding.ProtocolIGMP).
 			MatchRegMark(FromLocalRegMark).
@@ -2658,33 +2661,34 @@ func (f *featureMulticast) igmpPktInFlows(reason uint8) []binding.Flow {
 	return flows
 }
 
-// localMulticastForwardFlow generates the flow to forward multicast packets with OVS action "normal", and outputs
+// localMulticastForwardFlows generates the flow to forward multicast packets with OVS action "normal", and outputs
 // it to Antrea gateway in the meanwhile, so that the packet can be forwarded to local Pods which have joined the Multicast
 // group and to the external receivers. For external multicast packets accessing to the given multicast IP also hits the
 // flow, and the packet is not sent back to Antrea gateway because OVS datapath will drop it when it finds the output
 // port is the same as the input port.
-func (f *featureMulticast) localMulticastForwardFlow(multicastIP net.IP) []binding.Flow {
+func (f *featureMulticast) localMulticastForwardFlows(multicastIP net.IP, groupID binding.GroupIDType) []binding.Flow {
 	return []binding.Flow{
-		MulticastTable.ofTable.BuildFlow(priorityNormal).
+		MulticastRoutingTable.ofTable.BuildFlow(priorityNormal).
 			Cookie(f.cookieAllocator.Request(f.category).Raw()).
 			MatchProtocol(binding.ProtocolIP).
 			MatchDstIP(multicastIP).
-			Action().Output(config.HostGatewayOFPort).
-			Action().Normal().
+			Action().Group(groupID).
 			Done(),
 	}
 }
 
 // externalMulticastReceiverFlow generates the flow to output multicast packets to Antrea gateway, so that local Pods can
 // send multicast packets to access the external receivers. For the case that one or more local Pods have joined the target
-// multicast group, it is handled by the flows created by function "localMulticastForwardFlow" after local Pods report the
+// multicast group, it is handled by the flows created by function "localMulticastForwardFlows" after local Pods report the
 // IGMP membership.
 func (f *featureMulticast) externalMulticastReceiverFlow() binding.Flow {
-	return MulticastTable.ofTable.BuildFlow(priorityLow).
+	return MulticastRoutingTable.ofTable.BuildFlow(priorityLow).
 		Cookie(f.cookieAllocator.Request(f.category).Raw()).
 		MatchProtocol(binding.ProtocolIP).
 		MatchDstIPNet(*mcastCIDR).
-		Action().Output(config.HostGatewayOFPort).
+		Action().LoadRegMark(OFPortFoundRegMark).
+		Action().LoadToRegField(TargetOFPortField, config.HostGatewayOFPort).
+		Action().NextTable().
 		Done()
 }
 
diff --git a/pkg/agent/proxy/proxier.go b/pkg/agent/proxy/proxier.go
index d3e6a4d31..96f10c3c4 100644
--- a/pkg/agent/proxy/proxier.go
+++ b/pkg/agent/proxy/proxier.go
@@ -173,7 +173,7 @@ func (p *proxier) removeStaleServices() {
 		// Remove Service group whose Endpoints are local.
 		if svcInfo.NodeLocalExternal() {
 			if groupIDLocal, exist := p.groupCounter.Get(svcPortName, true); exist {
-				if err := p.ofClient.UninstallServiceGroup(groupIDLocal); err != nil {
+				if err := p.ofClient.UninstallGroup(groupIDLocal); err != nil {
 					klog.ErrorS(err, "Failed to remove Group of local Endpoints for Service", "Service", svcPortName)
 					continue
 				}
@@ -182,7 +182,7 @@ func (p *proxier) removeStaleServices() {
 		}
 		// Remove Service group which has all Endpoints.
 		if groupID, exist := p.groupCounter.Get(svcPortName, false); exist {
-			if err := p.ofClient.UninstallServiceGroup(groupID); err != nil {
+			if err := p.ofClient.UninstallGroup(groupID); err != nil {
 				klog.ErrorS(err, "Failed to remove Group of all Endpoints for Service", "Service", svcPortName)
 				continue
 			}
@@ -555,7 +555,7 @@ func (p *proxier) installServices() {
 					continue
 				}
 				if groupID, exist := p.groupCounter.Get(svcPortName, !nodeLocalVal); exist {
-					if err := p.ofClient.UninstallServiceGroup(groupID); err != nil {
+					if err := p.ofClient.UninstallGroup(groupID); err != nil {
 						klog.ErrorS(err, "Failed to uninstall Group of all Endpoints for Service", "Service", svcPortName)
 						continue
 					}
diff --git a/pkg/agent/types/event.go b/pkg/agent/types/event.go
new file mode 100644
index 000000000..eea70f66e
--- /dev/null
+++ b/pkg/agent/types/event.go
@@ -0,0 +1,22 @@
+// 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 types
+
+type PodUpdate struct {
+	PodNamespace string
+	PodName      string
+	IsAdd        bool
+	ContainerID  string
+}
diff --git a/pkg/ovs/openflow/interfaces.go b/pkg/ovs/openflow/interfaces.go
index a086f9a40..74e362e69 100644
--- a/pkg/ovs/openflow/interfaces.go
+++ b/pkg/ovs/openflow/interfaces.go
@@ -94,6 +94,7 @@ type Bridge interface {
 	CreateTable(table Table, next uint8, missAction MissActionType) Table
 	// AddTable adds table on the Bridge. Return true if the operation succeeds, otherwise return false.
 	DeleteTable(id uint8) bool
+	CreateGroupTypeAll(id GroupIDType) Group
 	CreateGroup(id GroupIDType) Group
 	DeleteGroup(id GroupIDType) bool
 	CreateMeter(id MeterIDType, flags ofctrl.MeterFlag) Meter
diff --git a/pkg/ovs/openflow/ofctrl_bridge.go b/pkg/ovs/openflow/ofctrl_bridge.go
index c1b5e11df..e5b2b886c 100644
--- a/pkg/ovs/openflow/ofctrl_bridge.go
+++ b/pkg/ovs/openflow/ofctrl_bridge.go
@@ -189,8 +189,16 @@ type OFBridge struct {
 	multipartReplyChs map[uint32]chan *openflow13.MultipartReply
 }
 
+func (b *OFBridge) CreateGroupTypeAll(id GroupIDType) Group {
+	return b.createGroupWithType(id, ofctrl.GroupAll)
+}
+
 func (b *OFBridge) CreateGroup(id GroupIDType) Group {
-	ofctrlGroup, err := b.ofSwitch.NewGroup(uint32(id), ofctrl.GroupSelect)
+	return b.createGroupWithType(id, ofctrl.GroupSelect)
+}
+
+func (b *OFBridge) createGroupWithType(id GroupIDType, groupType ofctrl.GroupType) Group {
+	ofctrlGroup, err := b.ofSwitch.NewGroup(uint32(id), groupType)
 	if err != nil { // group already exists
 		ofctrlGroup = b.ofSwitch.GetGroup(uint32(id))
 	}
diff --git a/pkg/util/channel/channel.go b/pkg/util/channel/channel.go
index b10e9b4d4..019eee536 100644
--- a/pkg/util/channel/channel.go
+++ b/pkg/util/channel/channel.go
@@ -25,7 +25,7 @@ const (
 	notifyTimeout = time.Second
 )
 
-type eventHandler func(string)
+type eventHandler func(interface{})
 
 type Subscriber interface {
 	// Subscribe registers an eventHandler which will be called when an event is sent to the channel.
@@ -37,7 +37,7 @@ type Subscriber interface {
 
 type Notifier interface {
 	// Notify sends an event to the channel.
-	Notify(string) bool
+	Notify(interface{}) bool
 }
 
 // SubscribableChannel is different from the Go channel which dispatches every event to only single consumer regardless
@@ -47,7 +47,7 @@ type SubscribableChannel struct {
 	// The name of the channel, used for logging purpose to differentiate multiple channels.
 	name string
 	// eventCh is the channel used for buffering the pending events.
-	eventCh chan string
+	eventCh chan interface{}
 	// handlers is a slice of callbacks registered by consumers.
 	handlers []eventHandler
 }
@@ -55,7 +55,7 @@ type SubscribableChannel struct {
 func NewSubscribableChannel(name string, bufferSize int) *SubscribableChannel {
 	n := &SubscribableChannel{
 		name:    name,
-		eventCh: make(chan string, bufferSize),
+		eventCh: make(chan interface{}, bufferSize),
 	}
 	return n
 }
@@ -64,7 +64,7 @@ func (n *SubscribableChannel) Subscribe(h eventHandler) {
 	n.handlers = append(n.handlers, h)
 }
 
-func (n *SubscribableChannel) Notify(e string) bool {
+func (n *SubscribableChannel) Notify(e interface{}) bool {
 	timer := time.NewTimer(notifyTimeout)
 	defer timer.Stop()
 	select {


Test patch:
diff --git a/pkg/agent/controller/egress/egress_controller_test.go b/pkg/agent/controller/egress/egress_controller_test.go
index 9c448dd1c..0d241bf7e 100644
--- a/pkg/agent/controller/egress/egress_controller_test.go
+++ b/pkg/agent/controller/egress/egress_controller_test.go
@@ -38,6 +38,7 @@ import (
 	ipassignertest "antrea.io/antrea/pkg/agent/ipassigner/testing"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 	routetest "antrea.io/antrea/pkg/agent/route/testing"
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
@@ -582,7 +583,11 @@ func TestPodUpdateShouldSyncEgress(t *testing.T) {
 	c.mockIPAssigner.EXPECT().UnassignIP(fakeLocalEgressIP1)
 	// Mock CNIServer
 	addPodInterface(c.ifaceStore, "ns1", "pendingPod", 10)
-	c.podUpdateChannel.Notify("ns1/pendingPod")
+	ev := types.PodUpdate{
+		PodName:      "pendingPod",
+		PodNamespace: "ns1",
+	}
+	c.podUpdateChannel.Notify(ev)
 	require.NoError(t, wait.PollImmediate(10*time.Millisecond, time.Second, func() (done bool, err error) {
 		return c.queue.Len() == 1, nil
 	}))
diff --git a/pkg/agent/controller/networkpolicy/cache_test.go b/pkg/agent/controller/networkpolicy/cache_test.go
index 35878d26c..00c01d3ed 100644
--- a/pkg/agent/controller/networkpolicy/cache_test.go
+++ b/pkg/agent/controller/networkpolicy/cache_test.go
@@ -25,8 +25,10 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	"antrea.io/antrea/pkg/util/channel"
+	"antrea.io/antrea/pkg/util/k8s"
 )
 
 var (
@@ -1184,7 +1186,12 @@ func TestRuleCacheProcessPodUpdates(t *testing.T) {
 			stopCh := make(chan struct{})
 			defer close(stopCh)
 			go podUpdateNotifier.Run(stopCh)
-			podUpdateNotifier.Notify(tt.podUpdate)
+			ns, name := k8s.SplitNamespacedName(tt.podUpdate)
+			e := types.PodUpdate{
+				PodNamespace: ns,
+				PodName:      name,
+			}
+			podUpdateNotifier.Notify(e)
 			func() {
 				// Drain the channel with 10 ms timeout so we can know it's done.
 				for {
diff --git a/pkg/agent/multicast/mcast_controller_test.go b/pkg/agent/multicast/mcast_controller_test.go
index 9918b5ffe..0536dfda6 100644
--- a/pkg/agent/multicast/mcast_controller_test.go
+++ b/pkg/agent/multicast/mcast_controller_test.go
@@ -36,8 +36,10 @@ import (
 	"antrea.io/antrea/pkg/agent/interfacestore"
 	ifaceStoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
 	multicasttest "antrea.io/antrea/pkg/agent/multicast/testing"
+	"antrea.io/antrea/pkg/agent/openflow"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 	ovsconfigtest "antrea.io/antrea/pkg/ovs/ovsconfig/testing"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 var (
@@ -45,16 +47,21 @@ var (
 	mockMulticastSocket *multicasttest.MockRouteInterface
 	mockIfaceStore      *ifaceStoretest.MockInterfaceStore
 	ovsClient           *ovsconfigtest.MockOVSBridgeClient
-	mgroup              = net.ParseIP("224.96.1.3")
 	if1                 = &interfacestore.InterfaceConfig{
 		Type:          interfacestore.ContainerInterface,
 		InterfaceName: "if1",
 		IPs:           []net.IP{net.ParseIP("192.168.1.1")},
+		OVSPortConfig: &interfacestore.OVSPortConfig{
+			OFPort: 1,
+		},
 	}
 	if2 = &interfacestore.InterfaceConfig{
 		Type:          interfacestore.ContainerInterface,
 		InterfaceName: "if2",
 		IPs:           []net.IP{net.ParseIP("192.168.1.2")},
+		OVSPortConfig: &interfacestore.OVSPortConfig{
+			OFPort: 2,
+		},
 	}
 	nodeIf1IP           = net.ParseIP("192.168.20.22")
 	externalInterfaceIP = net.ParseIP("192.168.50.23")
@@ -63,6 +70,7 @@ var (
 )
 
 func TestAddGroupMemberStatus(t *testing.T) {
+	mgroup := net.ParseIP("224.96.1.3")
 	event := &mcastGroupEvent{
 		group: mgroup,
 		eType: groupJoin,
@@ -82,7 +90,9 @@ func TestAddGroupMemberStatus(t *testing.T) {
 	key, ok := obj.(string)
 	assert.True(t, ok)
 	assert.Equal(t, mgroup.String(), key)
-	mockOFClient.EXPECT().InstallMulticastFlow(mgroup).Times(1)
+	mockIfaceStore.EXPECT().GetInterfaceByName(if1.InterfaceName).Return(if1, true)
+	mockOFClient.EXPECT().InstallMulticastGroup(gomock.Any(), gomock.Any())
+	mockOFClient.EXPECT().InstallMulticastFlows(mgroup, gomock.Any()).Times(1)
 	mockMulticastSocket.EXPECT().MulticastInterfaceJoinMgroup(mgroup.To4(), nodeIf1IP.To4(), if1.InterfaceName).Times(1)
 	err = mctrl.syncGroup(key)
 	assert.Nil(t, err)
@@ -94,6 +104,7 @@ func TestUpdateGroupMemberStatus(t *testing.T) {
 	err := mctrl.initialize(t)
 	assert.Nil(t, err)
 	igmpMaxResponseTime = time.Second * 1
+	mgroup := net.ParseIP("224.96.1.4")
 	event := &mcastGroupEvent{
 		group: mgroup,
 		eType: groupJoin,
@@ -101,7 +112,6 @@ func TestUpdateGroupMemberStatus(t *testing.T) {
 		iface: if1,
 	}
 	mctrl.addGroupMemberStatus(event)
-	obj, _, _ := mctrl.groupCache.GetByKey(event.group.String())
 	mockOFClient.EXPECT().SendIGMPQueryPacketOut(igmpQueryDstMac, mcastAllHosts, uint32(openflow13.P_NORMAL), gomock.Any()).Times(len(queryVersions))
 	for _, e := range []*mcastGroupEvent{
 		{group: mgroup, eType: groupJoin, time: event.time.Add(time.Second * 20), iface: if1},
@@ -110,6 +120,10 @@ func TestUpdateGroupMemberStatus(t *testing.T) {
 		{group: mgroup, eType: groupLeave, time: event.time.Add(time.Second * 61), iface: if1},
 		{group: mgroup, eType: groupLeave, time: event.time.Add(time.Second * 62), iface: if2},
 	} {
+		obj, _, _ := mctrl.groupCache.GetByKey(event.group.String())
+		if e.eType == groupLeave {
+			mockIfaceStore.EXPECT().GetInterfaceByName(e.iface.InterfaceName).Return(e.iface, true)
+		}
 		mctrl.updateGroupMemberStatus(obj, e)
 		groupCache := mctrl.groupCache
 		compareGroupStatus(t, groupCache, e)
@@ -127,15 +141,22 @@ func TestCheckLastMember(t *testing.T) {
 	workerCount = 1
 	igmpMaxResponseTime = time.Second * 1
 	lastProbe := time.Now()
+	mgroup := net.ParseIP("224.96.1.2")
 	testCheckLastMember := func(ev *mcastGroupEvent, expExist bool) {
 		status := &GroupMemberStatus{
-			localMembers:   sets.NewString(),
+			localMembers:   map[string]time.Time{},
 			lastIGMPReport: lastProbe,
 		}
 		if ev != nil {
 			status.group = ev.group
+			if ev.eType == groupLeave {
+				mockOFClient.EXPECT().UninstallGroup(gomock.Any())
+				mockOFClient.EXPECT().UninstallMulticastFlows(status.group)
+			}
 		} else {
 			status.group = mgroup
+			mockOFClient.EXPECT().UninstallGroup(gomock.Any())
+			mockOFClient.EXPECT().UninstallMulticastFlows(status.group)
 		}
 		_ = mctrl.groupCache.Add(status)
 		mctrl.addInstalledGroup(status.group.String())
@@ -167,7 +188,8 @@ func TestCheckLastMember(t *testing.T) {
 		}
 		mctrl.queue.Forget(obj)
 	}
-	mockOFClient.EXPECT().UninstallMulticastFlow(gomock.Any()).Times(2)
+	mockIfaceStore.EXPECT().GetInterfaceByName(if1.InterfaceName).Return(if1, true).Times(1)
+	mockOFClient.EXPECT().InstallMulticastGroup(gomock.Any(), gomock.Any()).Times(1)
 	for _, tc := range []struct {
 		ev     *mcastGroupEvent
 		exists bool
@@ -194,30 +216,31 @@ func TestClearStaleGroups(t *testing.T) {
 		mctrl.worker()
 		wg.Done()
 	}()
-
 	now := time.Now()
+	validUpdateTime := now.Add(-queryInterval)
 	validGroups := []*GroupMemberStatus{
 		{
 			group:          net.ParseIP("224.96.1.2"),
-			localMembers:   sets.NewString("p1", "p2"),
-			lastIGMPReport: now.Add(-queryInterval),
+			localMembers:   map[string]time.Time{"p1": now, "p2": validUpdateTime},
+			lastIGMPReport: validUpdateTime,
 		},
 		{
 			group:          net.ParseIP("224.96.1.3"),
-			localMembers:   sets.NewString(),
-			lastIGMPReport: now.Add(-queryInterval),
+			localMembers:   map[string]time.Time{"p2": validUpdateTime},
+			lastIGMPReport: validUpdateTime,
 		},
 	}
+	staleUpdateTime := now.Add(-mcastGroupTimeout - time.Second)
 	staleGroups := []*GroupMemberStatus{
 		{
 			group:          net.ParseIP("224.96.1.4"),
-			localMembers:   sets.NewString("p1", "p3"),
-			lastIGMPReport: now.Add(-mcastGroupTimeout - time.Second),
+			localMembers:   map[string]time.Time{"p1": staleUpdateTime, "p3": staleUpdateTime},
+			lastIGMPReport: staleUpdateTime,
 		},
 		{
 			group:          net.ParseIP("224.96.1.5"),
-			localMembers:   sets.NewString(),
-			lastIGMPReport: now.Add(-mcastGroupTimeout - time.Second),
+			localMembers:   map[string]time.Time{},
+			lastIGMPReport: staleUpdateTime,
 		},
 	}
 	for _, g := range validGroups {
@@ -225,12 +248,19 @@ func TestClearStaleGroups(t *testing.T) {
 		assert.Nil(t, err)
 		mctrl.addInstalledGroup(g.group.String())
 	}
+	fakePort := int32(1)
 	for _, g := range staleGroups {
 		err := mctrl.groupCache.Add(g)
 		assert.Nil(t, err)
 		mctrl.addInstalledGroup(g.group.String())
+		for m := range g.localMembers {
+			mockIface := &interfacestore.InterfaceConfig{InterfaceName: m, OVSPortConfig: &interfacestore.OVSPortConfig{OFPort: fakePort}}
+			mockIfaceStore.EXPECT().GetInterfaceByName(m).Return(mockIface, true)
+			fakePort++
+		}
 	}
-	mockOFClient.EXPECT().UninstallMulticastFlow(gomock.Any()).Times(len(staleGroups))
+	mockOFClient.EXPECT().UninstallGroup(gomock.Any()).Times(len(staleGroups))
+	mockOFClient.EXPECT().UninstallMulticastFlows(gomock.Any()).Times(len(staleGroups))
 	mockMulticastSocket.EXPECT().MulticastInterfaceLeaveMgroup(gomock.Any(), gomock.Any(), gomock.Any()).Times(len(staleGroups))
 	mctrl.clearStaleGroups()
 	mctrl.queue.ShutDown()
@@ -251,7 +281,9 @@ func TestProcessPacketIn(t *testing.T) {
 	snooper := mockController.igmpSnooper
 	stopCh := make(chan struct{})
 	defer close(stopCh)
-	go mockController.eventHandler(stopCh)
+	go func() {
+		mockController.eventHandler(stopCh)
+	}()
 
 	getIPs := func(ipStrs []string) []net.IP {
 		ips := make([]net.IP, len(ipStrs))
@@ -285,6 +317,7 @@ func TestProcessPacketIn(t *testing.T) {
 			version:      3,
 		},
 	} {
+		mockIfaceStore.EXPECT().GetInterfaceByName(tc.iface.InterfaceName).Return(tc.iface, true).AnyTimes()
 		packets := createIGMPReportPacketIn(getIPs(tc.joinedGroups.List()), getIPs(tc.leftGroups.List()), tc.version, uint32(tc.iface.OFPort))
 		mockOFClient.EXPECT().SendIGMPQueryPacketOut(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 		for _, pkt := range packets {
@@ -311,11 +344,11 @@ func compareGroupStatus(t *testing.T, cache cache.Indexer, event *mcastGroupEven
 	assert.Equal(t, event.group, status.group)
 	if event.eType == groupJoin {
 		assert.True(t, status.lastIGMPReport.Equal(event.time) || status.lastIGMPReport.After(event.time))
-		exists := status.localMembers.Has(event.iface.InterfaceName)
+		_, exists := status.localMembers[event.iface.InterfaceName]
 		assert.Truef(t, exists, "member is not added into cache")
 	} else {
 		assert.True(t, status.lastIGMPReport.Before(event.time))
-		exists := status.localMembers.Has(event.iface.InterfaceName)
+		_, exists := status.localMembers[event.iface.InterfaceName]
 		assert.Falsef(t, exists, "member is not removed from cache")
 	}
 }
@@ -329,13 +362,13 @@ func newMockMulticastController(t *testing.T) *Controller {
 	addr := &net.IPNet{IP: nodeIf1IP, Mask: net.IPv4Mask(255, 255, 255, 0)}
 	nodeConfig := &config.NodeConfig{GatewayConfig: &config.GatewayConfig{Name: "antrea-gw0"}, NodeIPv4Addr: addr}
 	mockOFClient.EXPECT().RegisterPacketInHandler(gomock.Any(), gomock.Any(), gomock.Any()).Times(1)
-	mctrl := NewMulticastController(mockOFClient, nodeConfig, mockIfaceStore, mockMulticastSocket, sets.NewString(), ovsClient)
+	groupAllocator := openflow.NewGroupAllocator(false)
+	podUpdateSubscriber := channel.NewSubscribableChannel("PodUpdate", 100)
+	mctrl := NewMulticastController(mockOFClient, groupAllocator, nodeConfig, mockIfaceStore, mockMulticastSocket, sets.NewString(), ovsClient, podUpdateSubscriber)
 	return mctrl
 }
 
 func (c *Controller) initialize(t *testing.T) error {
-	ovsClient.EXPECT().SetBridgeMcastSnooping(true).Times(1)
-	ovsClient.EXPECT().AddBridgeOtherConfig(map[string]interface{}{"mcast-snooping-disable-flood-unregistered": "true"}).Times(1)
 	mockOFClient.EXPECT().InstallMulticastInitialFlows(uint8(0)).Times(1)
 	mockMulticastSocket.EXPECT().AllocateVIFs(gomock.Any(), uint16(0)).Times(1).Return([]uint16{0}, nil)
 	mockMulticastSocket.EXPECT().AllocateVIFs(gomock.Any(), uint16(1)).Times(1).Return([]uint16{1, 2}, nil)
diff --git a/pkg/agent/multicast/mcast_route_test.go b/pkg/agent/multicast/mcast_route_test.go
index 081171e7c..b9f4c14d6 100644
--- a/pkg/agent/multicast/mcast_route_test.go
+++ b/pkg/agent/multicast/mcast_route_test.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -76,12 +77,12 @@ func TestProcessIGMPNocacheMsg(t *testing.T) {
 	mRoute.internalInterfaceVIF = uint16(0)
 	status1 := &GroupMemberStatus{
 		group:        net.ParseIP("224.3.3.8"),
-		localMembers: sets.NewString("aa"),
+		localMembers: map[string]time.Time{"aa": time.Now()},
 	}
 	mRoute.groupCache.Add(status1)
 	status2 := &GroupMemberStatus{
 		group:        net.ParseIP("224.3.3.9"),
-		localMembers: sets.NewString(),
+		localMembers: map[string]time.Time{},
 	}
 	mRoute.groupCache.Add(status2)
 	for _, m := range []struct {
diff --git a/pkg/agent/openflow/testing/mock_openflow.go b/pkg/agent/openflow/testing/mock_openflow.go
index 34415abdb..8e47d69fc 100644
--- a/pkg/agent/openflow/testing/mock_openflow.go
+++ b/pkg/agent/openflow/testing/mock_openflow.go
@@ -282,18 +282,32 @@ func (mr *MockClientMockRecorder) InstallEndpointFlows(arg0, arg1 interface{}) *
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallEndpointFlows", reflect.TypeOf((*MockClient)(nil).InstallEndpointFlows), arg0, arg1)
 }
 
-// InstallMulticastFlow mocks base method
-func (m *MockClient) InstallMulticastFlow(arg0 net.IP) error {
+// InstallMulticastFlows mocks base method
+func (m *MockClient) InstallMulticastFlows(arg0 net.IP, arg1 openflow.GroupIDType) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallMulticastFlow", arg0)
+	ret := m.ctrl.Call(m, "InstallMulticastFlows", arg0, arg1)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// InstallMulticastFlow indicates an expected call of InstallMulticastFlow
-func (mr *MockClientMockRecorder) InstallMulticastFlow(arg0 interface{}) *gomock.Call {
+// InstallMulticastFlows indicates an expected call of InstallMulticastFlows
+func (mr *MockClientMockRecorder) InstallMulticastFlows(arg0, arg1 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticastFlow", reflect.TypeOf((*MockClient)(nil).InstallMulticastFlow), arg0)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticastFlows", reflect.TypeOf((*MockClient)(nil).InstallMulticastFlows), arg0, arg1)
+}
+
+// InstallMulticastGroup mocks base method
+func (m *MockClient) InstallMulticastGroup(arg0 openflow.GroupIDType, arg1 []uint32) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "InstallMulticastGroup", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// InstallMulticastGroup indicates an expected call of InstallMulticastGroup
+func (mr *MockClientMockRecorder) InstallMulticastGroup(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticastGroup", reflect.TypeOf((*MockClient)(nil).InstallMulticastGroup), arg0, arg1)
 }
 
 // InstallMulticastInitialFlows mocks base method
@@ -640,18 +654,32 @@ func (mr *MockClientMockRecorder) UninstallEndpointFlows(arg0, arg1 interface{})
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallEndpointFlows", reflect.TypeOf((*MockClient)(nil).UninstallEndpointFlows), arg0, arg1)
 }
 
-// UninstallMulticastFlow mocks base method
-func (m *MockClient) UninstallMulticastFlow(arg0 net.IP) error {
+// UninstallGroup mocks base method
+func (m *MockClient) UninstallGroup(arg0 openflow.GroupIDType) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "UninstallGroup", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// UninstallGroup indicates an expected call of UninstallGroup
+func (mr *MockClientMockRecorder) UninstallGroup(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallGroup", reflect.TypeOf((*MockClient)(nil).UninstallGroup), arg0)
+}
+
+// UninstallMulticastFlows mocks base method
+func (m *MockClient) UninstallMulticastFlows(arg0 net.IP) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UninstallMulticastFlow", arg0)
+	ret := m.ctrl.Call(m, "UninstallMulticastFlows", arg0)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// UninstallMulticastFlow indicates an expected call of UninstallMulticastFlow
-func (mr *MockClientMockRecorder) UninstallMulticastFlow(arg0 interface{}) *gomock.Call {
+// UninstallMulticastFlows indicates an expected call of UninstallMulticastFlows
+func (mr *MockClientMockRecorder) UninstallMulticastFlows(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallMulticastFlow", reflect.TypeOf((*MockClient)(nil).UninstallMulticastFlow), arg0)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallMulticastFlows", reflect.TypeOf((*MockClient)(nil).UninstallMulticastFlows), arg0)
 }
 
 // UninstallNodeFlows mocks base method
@@ -739,20 +767,6 @@ func (mr *MockClientMockRecorder) UninstallServiceFlows(arg0, arg1, arg2 interfa
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallServiceFlows", reflect.TypeOf((*MockClient)(nil).UninstallServiceFlows), arg0, arg1, arg2)
 }
 
-// UninstallServiceGroup mocks base method
-func (m *MockClient) UninstallServiceGroup(arg0 openflow.GroupIDType) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UninstallServiceGroup", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// UninstallServiceGroup indicates an expected call of UninstallServiceGroup
-func (mr *MockClientMockRecorder) UninstallServiceGroup(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallServiceGroup", reflect.TypeOf((*MockClient)(nil).UninstallServiceGroup), arg0)
-}
-
 // UninstallTraceflowFlows mocks base method
 func (m *MockClient) UninstallTraceflowFlows(arg0 byte) error {
 	m.ctrl.T.Helper()
diff --git a/pkg/agent/proxy/proxier_test.go b/pkg/agent/proxy/proxier_test.go
index fc809925f..6c6ec07c2 100644
--- a/pkg/agent/proxy/proxier_test.go
+++ b/pkg/agent/proxy/proxier_test.go
@@ -347,7 +347,7 @@ func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep
 			mockOFClient.EXPECT().InstallServiceFlows(groupID, loadBalancerIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeLoadBalancer).Times(1)
 		}
 		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, !nodeLocalVal)
-		mockOFClient.EXPECT().UninstallServiceGroup(groupID).Times(1)
+		mockOFClient.EXPECT().UninstallGroup(groupID).Times(1)
 	}
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	if proxyLoadBalancerIPs {
@@ -466,7 +466,7 @@ func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP
 		mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeNodePort).Times(1)
 
 		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, !nodeLocalVal)
-		mockOFClient.EXPECT().UninstallServiceGroup(groupID).Times(1)
+		mockOFClient.EXPECT().UninstallGroup(groupID).Times(1)
 	}
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	mockRouteClient.EXPECT().AddNodePort(gomock.Any(), uint16(svcNodePort), bindingProtocol).Times(1)
@@ -748,7 +748,7 @@ func testClusterIPRemoval(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool)
 	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(1)
+	mockOFClient.EXPECT().UninstallGroup(gomock.Any()).Times(1)
 	fp.syncProxyRules()
 
 	fp.serviceChanges.OnServiceUpdate(service, nil)
@@ -1231,8 +1231,8 @@ func TestServicesWithSameEndpoints(t *testing.T) {
 	mockOFClient.EXPECT().InstallServiceFlows(groupID2, svcIP2, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
 	mockOFClient.EXPECT().UninstallServiceFlows(svcIP1, uint16(svcPort), bindingProtocol).Times(1)
 	mockOFClient.EXPECT().UninstallServiceFlows(svcIP2, uint16(svcPort), bindingProtocol).Times(1)
-	mockOFClient.EXPECT().UninstallServiceGroup(groupID1).Times(1)
-	mockOFClient.EXPECT().UninstallServiceGroup(groupID2).Times(1)
+	mockOFClient.EXPECT().UninstallGroup(groupID1).Times(1)
+	mockOFClient.EXPECT().UninstallGroup(groupID2).Times(1)
 	// Since these two Services reference to the same Endpoint, there should only be one operation.
 	mockOFClient.EXPECT().UninstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
 
diff --git a/pkg/ovs/openflow/testing/mock_openflow.go b/pkg/ovs/openflow/testing/mock_openflow.go
index 283a606c5..45a1e3f47 100644
--- a/pkg/ovs/openflow/testing/mock_openflow.go
+++ b/pkg/ovs/openflow/testing/mock_openflow.go
@@ -134,6 +134,20 @@ func (mr *MockBridgeMockRecorder) CreateGroup(arg0 interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateGroup", reflect.TypeOf((*MockBridge)(nil).CreateGroup), arg0)
 }
 
+// CreateGroupTypeAll mocks base method
+func (m *MockBridge) CreateGroupTypeAll(arg0 openflow.GroupIDType) openflow.Group {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CreateGroupTypeAll", arg0)
+	ret0, _ := ret[0].(openflow.Group)
+	return ret0
+}
+
+// CreateGroupTypeAll indicates an expected call of CreateGroupTypeAll
+func (mr *MockBridgeMockRecorder) CreateGroupTypeAll(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateGroupTypeAll", reflect.TypeOf((*MockBridge)(nil).CreateGroupTypeAll), arg0)
+}
+
 // CreateMeter mocks base method
 func (m *MockBridge) CreateMeter(arg0 openflow.MeterIDType, arg1 ofctrl.MeterFlag) openflow.Meter {
 	m.ctrl.T.Helper()
diff --git a/pkg/util/channel/channel_test.go b/pkg/util/channel/channel_test.go
index 2eee503f7..e848bd019 100644
--- a/pkg/util/channel/channel_test.go
+++ b/pkg/util/channel/channel_test.go
@@ -36,10 +36,10 @@ func newEventReceiver() *eventReceiver {
 	}
 }
 
-func (r *eventReceiver) receive(e string) {
+func (r *eventReceiver) receive(e interface{}) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
-	r.receivedEvents.Insert(e)
+	r.receivedEvents.Insert(e.(string))
 }
 
 func (r *eventReceiver) received() sets.String {
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index 7b3a80c52..9a8121103 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -379,12 +379,19 @@ func labelNodeRoleControlPlane() string {
 	return labelNodeRoleControlPlane
 }
 
-func controlPlaneNoScheduleToleration() corev1.Toleration {
+func controlPlaneNoScheduleTolerations() []corev1.Toleration {
 	// the Node taint still uses "master" in K8s v1.20
-	return corev1.Toleration{
-		Key:      "node-role.kubernetes.io/master",
-		Operator: corev1.TolerationOpExists,
-		Effect:   corev1.TaintEffectNoSchedule,
+	return []corev1.Toleration{
+		{
+			Key:      "node-role.kubernetes.io/master",
+			Operator: corev1.TolerationOpExists,
+			Effect:   corev1.TaintEffectNoSchedule,
+		},
+		{
+			Key:      "node-role.kubernetes.io/control-plane",
+			Operator: corev1.TolerationOpExists,
+			Effect:   corev1.TaintEffectNoSchedule,
+		},
 	}
 }
 
@@ -1119,8 +1126,7 @@ func (data *TestData) CreatePodOnNodeInNamespace(name, ns string, nodeName, ctrN
 	}
 	if nodeName == controlPlaneNodeName() {
 		// tolerate NoSchedule taint if we want Pod to run on control-plane Node
-		noScheduleToleration := controlPlaneNoScheduleToleration()
-		podSpec.Tolerations = []corev1.Toleration{noScheduleToleration}
+		podSpec.Tolerations = controlPlaneNoScheduleTolerations()
 	}
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -2390,9 +2396,7 @@ func (data *TestData) createAgnhostPodOnNodeWithAnnotations(name string, ns stri
 
 func (data *TestData) createDaemonSet(name string, ns string, ctrName string, image string, cmd []string, args []string) (*appsv1.DaemonSet, func() error, error) {
 	podSpec := corev1.PodSpec{
-		Tolerations: []corev1.Toleration{
-			controlPlaneNoScheduleToleration(),
-		},
+		Tolerations: controlPlaneNoScheduleTolerations(),
 		Containers: []corev1.Container{
 			{
 				Name:            ctrName,
@@ -2464,9 +2468,7 @@ func (data *TestData) waitForDaemonSetPods(timeout time.Duration, dsName string,
 
 func (data *TestData) createStatefulSet(name string, ns string, size int32, ctrName string, image string, cmd []string, args []string, mutateFunc func(*appsv1.StatefulSet)) (*appsv1.StatefulSet, func() error, error) {
 	podSpec := corev1.PodSpec{
-		Tolerations: []corev1.Toleration{
-			controlPlaneNoScheduleToleration(),
-		},
+		Tolerations: controlPlaneNoScheduleTolerations(),
 		Containers: []corev1.Container{
 			{
 				Name:            ctrName,
diff --git a/test/e2e/multicast_test.go b/test/e2e/multicast_test.go
index 31a49751b..0d7755550 100644
--- a/test/e2e/multicast_test.go
+++ b/test/e2e/multicast_test.go
@@ -83,6 +83,7 @@ func TestMulticast(t *testing.T) {
 		for _, mc := range testcases {
 			mc := mc
 			t.Run(mc.name, func(t *testing.T) {
+				t.Parallel()
 				runTestMulticastBetweenPods(t, data, mc, nodeMulticastInterfaces)
 			})
 		}
@@ -122,6 +123,7 @@ func TestMulticast(t *testing.T) {
 		for _, mc := range testcases {
 			mc := mc
 			t.Run(mc.name, func(t *testing.T) {
+				t.Parallel()
 				runTestMulticastBetweenPods(t, data, mc, nodeMulticastInterfaces)
 			})
 		}
@@ -189,7 +191,7 @@ func runTestMulticastForwardToMultipleInterfaces(t *testing.T, data *TestData, s
 	}
 }
 
-func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestcase, nodeMulticastInterfaces [][]string) {
+func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestcase, nodeMulticastInterfaces map[int][]string) {
 	mcjoinWaitTimeout := defaultTimeout / time.Second
 	gatewayInterface, err := data.GetGatewayInterfaceName(antreaNamespace)
 	failOnError(err, t)
@@ -209,10 +211,10 @@ func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestc
 			defer wg.Done()
 			// The following command joins a multicast group and sets the timeout to 100 seconds(-W 100) before exit.
 			// The command will return after receiving 1 packet(-c 1).
-			receiveMulticastCommand := []string{"/bin/sh", "-c", fmt.Sprintf("mcjoin -c 1 -o -p %d -W %d %s", mc.port, mcjoinWaitTimeout, mc.group.String())}
+			receiveMulticastCommand := []string{"/bin/sh", "-c", fmt.Sprintf("mcjoin -c 10 -o -p %d -W %d %s", mc.port, mcjoinWaitTimeout, mc.group.String())}
 			res, _, err := data.RunCommandFromPod(testNamespace, r, mcjoinContainerName, receiveMulticastCommand)
 			failOnError(err, t)
-			assert.Contains(t, res, "Total: 1 packets")
+			assert.Contains(t, res, "Total: 10 packets")
 		}()
 	}
 	// Wait 2 seconds(-w 2) before sending multicast traffic.
@@ -222,23 +224,32 @@ func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestc
 		data.RunCommandFromPod(testNamespace, senderName, mcjoinContainerName, sendMulticastCommand)
 	}()
 
-	if err := wait.Poll(5*time.Second, defaultTimeout, func() (bool, error) {
-		// Sender pods should add an outbound multicast route except running as HostNetwork.
-		_, mrouteResult, _, err := data.RunCommandOnNode(nodeName(mc.senderConfig.nodeIdx), fmt.Sprintf("ip mroute show to %s iif %s | grep '%s'", mc.group.String(), gatewayInterface, strings.Join(nodeMulticastInterfaces[mc.senderConfig.nodeIdx], " ")))
-		if err != nil {
-			return false, err
-		}
-		if !mc.senderConfig.isHostNetwork {
-			if len(mrouteResult) == 0 {
-				return false, nil
+	readyReceivers := sets.NewInt()
+	senderReady := false
+	if err := wait.Poll(3*time.Second, defaultTimeout, func() (bool, error) {
+		if !senderReady {
+			// Sender pods should add an outbound multicast route except running as HostNetwork.
+			_, mrouteResult, _, err := data.RunCommandOnNode(nodeName(mc.senderConfig.nodeIdx), fmt.Sprintf("ip mroute show to %s iif %s | grep '%s'", mc.group.String(), gatewayInterface, strings.Join(nodeMulticastInterfaces[mc.senderConfig.nodeIdx], " ")))
+			if err != nil {
+				return false, err
 			}
-		} else {
-			if len(mrouteResult) != 0 {
-				return false, nil
+			if !mc.senderConfig.isHostNetwork {
+				if len(mrouteResult) == 0 {
+					return false, nil
+				}
+			} else {
+				if len(mrouteResult) != 0 {
+					return false, nil
+				}
 			}
+			senderReady = true
 		}
+
 		// Check inbound multicast route and whether multicast interfaces has joined the multicast group.
 		for _, receiver := range mc.receiverConfigs {
+			if readyReceivers.Has(receiver.nodeIdx) {
+				continue
+			}
 			for _, receiverMulticastInterface := range nodeMulticastInterfaces[receiver.nodeIdx] {
 				_, mRouteResult, _, err := data.RunCommandOnNode(nodeName(receiver.nodeIdx), fmt.Sprintf("ip mroute show to %s iif %s ", mc.group.String(), receiverMulticastInterface))
 				if err != nil {
@@ -272,6 +283,7 @@ func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestc
 					}
 				}
 			}
+			readyReceivers = readyReceivers.Insert(receiver.nodeIdx)
 		}
 		return true, nil
 	}); err != nil {
@@ -282,7 +294,7 @@ func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestc
 
 // computeMulticastInterfaces computes multicastInterfaces for each node.
 // It returns [][]string with its index as node index and value as multicastInterfaces for this node.
-func computeMulticastInterfaces(t *testing.T, data *TestData) ([][]string, error) {
+func computeMulticastInterfaces(t *testing.T, data *TestData) (map[int][]string, error) {
 	multicastInterfaces, err := data.GetMulticastInterfaces(antreaNamespace)
 	if err != nil {
 		return nil, err
@@ -291,7 +303,7 @@ func computeMulticastInterfaces(t *testing.T, data *TestData) ([][]string, error
 	if err != nil {
 		t.Fatalf("Error getting transport interfaces: %v", err)
 	}
-	nodeMulticastInterfaces := make([][]string, 0, len(clusterInfo.nodes))
+	nodeMulticastInterfaces := make(map[int][]string)
 	for nodeIdx := range clusterInfo.nodes {
 		_, localInterfacesStr, _, err := data.RunCommandOnNode(nodeName(nodeIdx), "ls /sys/class/net")
 		if err != nil {
@@ -303,7 +315,7 @@ func computeMulticastInterfaces(t *testing.T, data *TestData) ([][]string, error
 		externalMulticastInterfaces := localInterfacesSet.Intersection(multicastInterfaceSet)
 		currNodeMulticastInterfaces := externalMulticastInterfaces.Insert(transportInterface).List()
 		t.Logf("Multicast interfaces for node index %d is %+v", nodeIdx, currNodeMulticastInterfaces)
-		nodeMulticastInterfaces = append(nodeMulticastInterfaces, currNodeMulticastInterfaces)
+		nodeMulticastInterfaces[nodeIdx] = currNodeMulticastInterfaces
 	}
 	return nodeMulticastInterfaces, nil
 }
diff --git a/test/e2e/networkpolicy_test.go b/test/e2e/networkpolicy_test.go
index 313ad31b1..c4d3939f4 100644
--- a/test/e2e/networkpolicy_test.go
+++ b/test/e2e/networkpolicy_test.go
@@ -816,8 +816,7 @@ func testIngressPolicyWithEndPort(t *testing.T, data *TestData) {
 		}
 		if nodeName == controlPlaneNodeName() {
 			// tolerate NoSchedule taint if we want Pod to run on control-plane Node
-			noScheduleToleration := controlPlaneNoScheduleToleration()
-			podSpec.Tolerations = []corev1.Toleration{noScheduleToleration}
+			podSpec.Tolerations = controlPlaneNoScheduleTolerations()
 		}
 		pod := &corev1.Pod{
 			ObjectMeta: metav1.ObjectMeta{
diff --git a/test/e2e/performance_test.go b/test/e2e/performance_test.go
index 02b32f666..c8b2c61de 100644
--- a/test/e2e/performance_test.go
+++ b/test/e2e/performance_test.go
@@ -50,8 +50,6 @@ var (
 	customizePolicyRules = flag.Int("perf.http.policy_rules", 0, "Number of CIDRs in the network policy")
 	httpConcurrency      = flag.Int("perf.http.concurrency", 1, "Number of multiple requests to make at a time")
 	realizeTimeout       = flag.Duration("perf.realize.timeout", 5*time.Minute, "Timeout of the realization of network policies")
-	// tolerate NoSchedule taint to let the Pod run on control-plane Node
-	noScheduleToleration = controlPlaneNoScheduleToleration()
 	labelSelector        = &metav1.LabelSelector{
 		MatchLabels: map[string]string{"app": perfTestAppLabel},
 	}
@@ -118,7 +116,7 @@ func createPerfTestPodDefinition(name, containerName, image string) *corev1.Pod
 		"kubernetes.io/hostname": controlPlaneNodeName(),
 	}
 
-	podSpec.Tolerations = []corev1.Toleration{noScheduleToleration}
+	podSpec.Tolerations = controlPlaneNoScheduleTolerations()
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   name,
diff --git a/test/e2e/prometheus_test.go b/test/e2e/prometheus_test.go
index 08571fd34..b3e1732b1 100644
--- a/test/e2e/prometheus_test.go
+++ b/test/e2e/prometheus_test.go
@@ -109,20 +109,14 @@ func skipIfPrometheusDisabled(t *testing.T) {
 // getMonitoringAuthToken retrieves monitoring authorization token, required for access to Antrea apiserver/metrics
 // resource
 func getMonitoringAuthToken(t *testing.T, data *TestData) string {
-	secrets, err := data.clientset.CoreV1().Secrets(monitoringNamespace).List(context.TODO(), metav1.ListOptions{})
+	const secretName = "prometheus-service-account-token"
+	secret, err := data.clientset.CoreV1().Secrets(monitoringNamespace).Get(context.TODO(), secretName, metav1.GetOptions{})
 	if err != nil {
-		t.Fatalf("Error fetching monitoring secrets: %v", err)
+		t.Fatalf("Error fetching monitoring secret '%s': %v", secretName, err)
 	}
-
-	var token string
-	for _, secret := range secrets.Items {
-		if secret.Annotations["kubernetes.io/service-account.name"] == "prometheus" {
-			token = string(secret.Data[v1.ServiceAccountTokenKey])
-		}
-	}
-
+	token := string(secret.Data[v1.ServiceAccountTokenKey])
 	if len(token) == 0 {
-		t.Fatal("Prometheus ServiceAccount secret not found")
+		t.Fatalf("Monitoring secret '%s' does not include token", secretName)
 	}
 
 	return token
diff --git a/test/integration/agent/openflow_test.go b/test/integration/agent/openflow_test.go
index 528aba271..e7c9d557b 100644
--- a/test/integration/agent/openflow_test.go
+++ b/test/integration/agent/openflow_test.go
@@ -704,7 +704,7 @@ func uninstallServiceFlowsFunc(t *testing.T, gid uint32, svc svcConfig, endpoint
 	groupID := ofconfig.GroupIDType(gid)
 	err := c.UninstallServiceFlows(svc.ip, svc.port, svc.protocol)
 	assert.Nil(t, err)
-	err = c.UninstallServiceGroup(groupID)
+	err = c.UninstallGroup(groupID)
 	assert.Nil(t, err)
 	for _, ep := range endpointList {
 		err := c.UninstallEndpointFlows(svc.protocol, ep)
diff --git a/test/integration/ovs/ofctrl_test.go b/test/integration/ovs/ofctrl_test.go
index b646cceeb..73769a10e 100644
--- a/test/integration/ovs/ofctrl_test.go
+++ b/test/integration/ovs/ofctrl_test.go
@@ -269,6 +269,8 @@ func TestOFctrlGroup(t *testing.T) {
 
 	ovsCtlClient := ovsctl.NewClient(brName)
 
+	id := 1
+
 	for name, buckets := range map[string][]struct {
 		weight        uint16      // Must have non-zero value.
 		reg2reg       [][4]uint32 // regNum, data, startIndex, endIndex
@@ -278,12 +280,25 @@ func TestOFctrlGroup(t *testing.T) {
 			{weight: 100, reg2reg: [][4]uint32{{0, 1, 0, 31}, {1, 2, 15, 31}}, resubmitTable: 31},
 			{weight: 110, resubmitTable: 42},
 		},
+		"TypeAll": {
+			{weight: 100, reg2reg: [][4]uint32{{0, 1, 0, 31}, {1, 2, 15, 31}}, resubmitTable: 31},
+			{weight: 110, resubmitTable: 42},
+		},
 	} {
 		t.Run(name, func(t *testing.T) {
-			group := br.CreateGroup(1)
+			var group binding.Group
+			gid := binding.GroupIDType(id)
+			if name == "TypeAll" {
+				group = br.CreateGroupTypeAll(gid)
+			} else {
+				group = br.CreateGroup(gid)
+			}
 			for _, bucket := range buckets {
 				require.NotZero(t, bucket.weight, "Weight value of a bucket must be specified")
-				bucketBuilder := group.Bucket().Weight(bucket.weight)
+				bucketBuilder := group.Bucket()
+				if name == "Normal" {
+					bucketBuilder = bucketBuilder.Weight(bucket.weight)
+				}
 				if bucket.resubmitTable != 0 {
 					bucketBuilder = bucketBuilder.ResubmitToTable(bucket.resubmitTable)
 				}
@@ -303,8 +318,10 @@ func TestOFctrlGroup(t *testing.T) {
 			}), "Failed to install group")
 			dumpedGroup := groups[0]
 			for i, bucket := range buckets {
-				// Must have weight
-				assert.True(t, strings.Contains(dumpedGroup[i+1], fmt.Sprintf("weight:%d", bucket.weight)))
+				if name == "Normal" {
+					// Must have weight
+					assert.True(t, strings.Contains(dumpedGroup[i+1], fmt.Sprintf("weight:%d", bucket.weight)))
+				}
 				for _, loading := range bucket.reg2reg {
 					rngStr := "[]"
 					if !(loading[2] == 0 && loading[3] == 31) {
@@ -326,6 +343,7 @@ func TestOFctrlGroup(t *testing.T) {
 				return len(groups) == 0, nil
 			}), "Failed to delete group")
 		})
+		id++
 	}
 }