Merge branch 'main-v4.3' into test/mqtt_sn

2022-10-08 14:28:01 +08:00 · 2022-10-08 14:28:01 +08:00 · 92d5caf908
parent 04c52aa5f4 120905403c
commit 92d5caf908
47 changed files with 1605 additions and 389 deletions
--- a/.ci/docker-compose-file/conf.env
+++ b/.ci/docker-compose-file/conf.env
@ -1,5 +1,5 @@
 EMQX_AUTH__LDAP__SERVERS=ldap_server
-EMQX_AUTH__MONGO__SERVER=mongo_server:27017
+EMQX_AUTH__MONGO__SERVER=toxiproxy:27017
 EMQX_AUTH__MYSQL__SERVER=mysql_server:3306
 EMQX_AUTH__MYSQL__USERNAME=root
 EMQX_AUTH__MYSQL__PASSWORD=public
--- a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml
+++ b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml
@ -0,0 +1,16 @@
+version: '3.9'
+
+services:
+  toxiproxy:
+    container_name: toxiproxy
+    image: ghcr.io/shopify/toxiproxy:2.5.0
+    restart: always
+    networks:
+      - emqx_bridge
+    volumes:
+      - "./toxiproxy.json:/config/toxiproxy.json"
+    ports:
+      - 8474:8474
+    command:
+      - "-host=0.0.0.0"
+      - "-config=/config/toxiproxy.json"
--- a/.ci/docker-compose-file/toxiproxy.json
+++ b/.ci/docker-compose-file/toxiproxy.json
@ -0,0 +1,8 @@
+[
+  {
+    "name": "mongo_single",
+    "listen": "0.0.0.0:27017",
+    "upstream": "mongo:27017",
+    "enabled": true
+  }
+]
--- a/.github/actions/detect-profiles/action.yaml
+++ b/.github/actions/detect-profiles/action.yaml
@ -0,0 +1,25 @@
+name: 'Detect profiles'
+inputs:
+  ci_git_token:
+    required: true
+    type: string
+outputs:
+  profiles:
+    description: 'Detected profiles'
+    value: ${{ steps.detect-profiles.outputs.profiles}}
+
+runs:
+  using: composite
+  steps:
+    - id: detect-profiles
+      shell: bash
+      run: |
+        if make emqx-ee --dry-run > /dev/null 2>&1; then
+          echo "::set-output name=profiles::[\"emqx-ee\"]"
+          echo "https://ci%40emqx.io:${{ inputs.ci_git_token }}@github.com" > $HOME/.git-credentials
+          git config --global credential.helper store
+          echo "EMQX_NAME=emqx-ee" >> $GITHUB_ENV
+        else
+          echo "::set-output name=profiles::[\"emqx\", \"emqx-edge\"]"
+          echo "EMQX_NAME=emqx" >> $GITHUB_ENV
+        fi
--- a/.github/actions/package-macos/action.yaml
+++ b/.github/actions/package-macos/action.yaml
@ -0,0 +1,96 @@
+name: 'Create MacOS package'
+inputs:
+  otp: # 24.2.1-1, 23.3.4.9-3
+    required: true
+    type: string
+  os:
+    required: false
+    type: string
+    default: macos-11
+  apple_id_password:
+    required: true
+    type: string
+  apple_developer_identity:
+    required: true
+    type: string
+  apple_developer_id_bundle:
+    required: true
+    type: string
+  apple_developer_id_bundle_password:
+    required: true
+    type: string
+
+runs:
+  using: composite
+  steps:
+    - name: prepare
+      shell: bash
+      run: |
+        brew update
+        brew install curl zip unzip gnu-sed kerl coreutils unixodbc freetds openssl@1.1
+        echo "/usr/local/opt/bison/bin" >> $GITHUB_PATH
+        echo "/usr/local/bin" >> $GITHUB_PATH
+    - uses: actions/cache@v2
+      id: cache
+      with:
+        path: ~/.kerl/${{ inputs.otp }}
+        key: otp-install-${{ inputs.otp }}-${{ inputs.os }}-static-ssl-disable-hipe-disable-jit
+    - name: build erlang
+      if: steps.cache.outputs.cache-hit != 'true'
+      shell: bash
+      env:
+        KERL_BUILD_BACKEND: git
+        OTP_GITHUB_URL: https://github.com/emqx/otp
+        KERL_CONFIGURE_OPTIONS: --disable-dynamic-ssl-lib --with-ssl=/usr/local/opt/openssl@1.1 --disable-hipe --disable-jit
+      run: |
+        kerl update releases
+        kerl build ${{ inputs.otp }}
+        kerl install ${{ inputs.otp }} $HOME/.kerl/${{ inputs.otp }}
+    - name: build
+      env:
+        AUTO_INSTALL_BUILD_DEPS: 1
+        APPLE_SIGN_BINARIES: 1
+        APPLE_ID: developers@emqx.io
+        APPLE_TEAM_ID: 26N6HYJLZA
+        APPLE_ID_PASSWORD: ${{ inputs.apple_id_password }}
+        APPLE_DEVELOPER_IDENTITY: ${{ inputs.apple_developer_identity }}
+        APPLE_DEVELOPER_ID_BUNDLE: ${{ inputs.apple_developer_id_bundle }}
+        APPLE_DEVELOPER_ID_BUNDLE_PASSWORD: ${{ inputs.apple_developer_id_bundle_password }}
+      shell: bash
+      run: |
+        . $HOME/.kerl/${{ inputs.otp }}/activate
+        make ensure-rebar3
+        sudo cp rebar3 /usr/local/bin/rebar3
+        make ${EMQX_NAME}-zip
+    - name: test
+      shell: bash
+      run: |
+        pkg_name=$(basename _packages/${EMQX_NAME}/${EMQX_NAME}-*.zip)
+        unzip -q _packages/${EMQX_NAME}/$pkg_name
+        # test with a spaces in path
+        mv ./emqx "./emqx  home/"
+        cd "./emqx  home/"
+        gsed -i '/emqx_telemetry/d' data/loaded_plugins
+        ./bin/emqx start || cat log/erlang.log.1
+        ready='no'
+        for i in {1..10}; do
+          if curl -fs 127.0.0.1:18083 > /dev/null; then
+            ready='yes'
+            break
+          fi
+          sleep 1
+        done
+        if [ "$ready" != "yes" ]; then
+          echo "Timed out waiting for emqx to be ready"
+          cat log/erlang.log.1
+          exit 1
+        fi
+        ./bin/emqx_ctl status
+        if ! ./bin/emqx stop; then
+          cat log/erlang.log.1 || true
+          cat log/emqx.log.1 || true
+          echo "failed to stop emqx"
+          exit 1
+        fi
+        cd ..
+        rm -rf "emqx  home"
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@ -23,23 +23,18 @@ jobs:
    container: emqx/build-env:erl23.3.4.9-3-ubuntu20.04

    outputs:
-      profiles: ${{ steps.set_profile.outputs.profiles}}
+      profiles: ${{ steps.detect-profiles.outputs.profiles}}

    steps:
      - uses: actions/checkout@v2
        with:
          path: source
          fetch-depth: 0
-      - name: set profile
-        id: set_profile
-        shell: bash
-        run: |
-          cd source
-          if make emqx-ee --dry-run > /dev/null 2>&1; then
-            echo "::set-output name=profiles::[\"emqx-ee\"]"
-          else
-            echo "::set-output name=profiles::[\"emqx\", \"emqx-edge\"]"
-          fi
+      - id: detect-profiles
+        working-directory: source
+        uses: ./.github/actions/detect-profiles
+        with:
+          ci_git_token: ${{ secrets.CI_GIT_TOKEN }}
      - name: get_all_deps
        if: endsWith(github.repository, 'emqx')
        run: |
@ -125,14 +120,11 @@ jobs:

    strategy:
      matrix:
-        profile: ${{fromJSON(needs.prepare.outputs.profiles)}}
-        erl_otp:
+        otp:
          - 23.3.4.9-3
-        exclude:
-          - profile: emqx-edge
-        macos:
-          - macos-10.15
-    runs-on: ${{ matrix.macos  }}
+        os:
+          - macos-11
+    runs-on: ${{ matrix.os }}

    steps:
    - uses: actions/download-artifact@v2
@ -140,70 +132,26 @@ jobs:
        name: source
        path: .
    - name: unzip source code
-      run: unzip -q source.zip
-    - name: prepare
      run: |
-        brew update
-        brew install curl zip unzip gnu-sed kerl unixodbc freetds
-        echo "/usr/local/bin" >> $GITHUB_PATH
-        git config --global credential.helper store
-    - uses: actions/cache@v2
-      id: cache
+        ln -s . source
+        unzip -q source.zip
+        rm source source.zip
+    - id: detect-profiles
+      uses: ./.github/actions/detect-profiles
      with:
-        path: ~/.kerl/${{ matrix.erl_otp }}
-        key: otp-install-${{ matrix.erl_otp }}-${{ matrix.macos }}
-    - name: build erlang
-      if: steps.cache.outputs.cache-hit != 'true'
-      timeout-minutes: 60
-      env:
-        KERL_BUILD_BACKEND: git
-        OTP_GITHUB_URL: https://github.com/emqx/otp
-      run: |
-        kerl update releases
-        kerl build ${{ matrix.erl_otp }}
-        kerl install ${{ matrix.erl_otp }} $HOME/.kerl/${{ matrix.erl_otp }}
-    - name: build
-      run: |
-        . $HOME/.kerl/${{ matrix.erl_otp }}/activate
-        cd source
-        make ensure-rebar3
-        sudo cp rebar3 /usr/local/bin/rebar3
-        rm -rf _build/${{ matrix.profile }}/lib
-        make ${{ matrix.profile }}-zip
-    - name: test
-      run: |
-        cd source
-        pkg_name=$(basename _packages/${{ matrix.profile }}/${{ matrix.profile }}-*.zip)
-        unzip -q _packages/${{ matrix.profile }}/$pkg_name
-        gsed -i '/emqx_telemetry/d' ./emqx/data/loaded_plugins
-        ./emqx/bin/emqx start || cat emqx/log/erlang.log.1
-        ready='no'
-        for i in {1..10}; do
-          if curl -fs 127.0.0.1:18083 > /dev/null; then
-            ready='yes'
-            break
-          fi
-          sleep 1
-        done
-        if [ "$ready" != "yes" ]; then
-          echo "Timed out waiting for emqx to be ready"
-          cat emqx/log/erlang.log.1
-          exit 1
-        fi
-        ./emqx/bin/emqx_ctl status
-        if ! ./emqx/bin/emqx stop; then
-          cat emqx/log/erlang.log.1 || true
-          cat emqx/log/emqx.log.1 || true
-          echo "failed to stop emqx"
-          exit 1
-        fi
-        rm -rf emqx
-        #sha256sum ./_packages/${{ matrix.profile }}/$pkg_name | head -c64  > ./_packages/${{ matrix.profile }}/$pkg_name.sha256
-        openssl dgst -sha256 ./_packages/${{ matrix.profile }}/$pkg_name | awk '{print $2}'  > ./_packages/${{ matrix.profile }}/$pkg_name.sha256
+        ci_git_token: ${{ secrets.CI_GIT_TOKEN }}
+    - uses: ./.github/actions/package-macos
+      with:
+        otp: ${{ matrix.otp }}
+        os: ${{ matrix.os }}
+        apple_id_password: ${{ secrets.APPLE_ID_PASSWORD }}
+        apple_developer_identity: ${{ secrets.APPLE_DEVELOPER_IDENTITY }}
+        apple_developer_id_bundle: ${{ secrets.APPLE_DEVELOPER_ID_BUNDLE }}
+        apple_developer_id_bundle_password: ${{ secrets.APPLE_DEVELOPER_ID_BUNDLE_PASSWORD }}
    - uses: actions/upload-artifact@v1
      with:
-        name: ${{ matrix.profile }}
-        path: source/_packages/${{ matrix.profile }}/.
+        name: ${EMQX_NAME}-${{ matrix.otp }}
+        path: _packages/${EMQX_NAME}/.

  linux:
    runs-on: ubuntu-20.04
--- a/.github/workflows/build_slim_packages.yaml
+++ b/.github/workflows/build_slim_packages.yaml
@ -10,11 +10,11 @@ jobs:

    strategy:
      matrix:
-        erl_otp:
-        - erl23.3.4.9-3
+        otp:
+          - erl23.3.4.9-3
        os:
-        - ubuntu20.04
-        - centos7
+          - ubuntu20.04
+          - centos7
        runs-on:
          - aws-amd64
          - ubuntu-20.04
@ -26,26 +26,20 @@ jobs:
          - runs-on: aws-amd64
            use-self-hosted: false

-    container: emqx/build-env:${{ matrix.erl_otp }}-${{ matrix.os }}
+    container: emqx/build-env:${{ matrix.otp }}-${{ matrix.os }}

    steps:
    - uses: actions/checkout@v1
-    - name: prepare
-      run: |
-        if make emqx-ee --dry-run > /dev/null 2>&1; then
-          echo "https://ci%40emqx.io:${{ secrets.CI_GIT_TOKEN }}@github.com" > $HOME/.git-credentials
-          git config --global credential.helper store
-          echo "EMQX_NAME=emqx-ee" >> $GITHUB_ENV
-        else
-          echo "EMQX_NAME=emqx" >> $GITHUB_ENV
-        fi
+    - uses: ./.github/actions/detect-profiles
+      with:
+        ci_git_token: ${{ secrets.CI_GIT_TOKEN }}
    - name: fix-git-unsafe-repository
      run: git config --global --add safe.directory /__w/emqx/emqx
    - uses: actions/cache@v2
      with:
        # dialyzer PLTs
        path: ~/.cache/rebar3/
-        key: dialyer-${{ matrix.erl_otp }}
+        key: dialyer-${{ matrix.otp }}
    - name: make xref
      run: make xref
    - name: make dialyzer
@ -71,85 +65,29 @@ jobs:
  mac:
    strategy:
      matrix:
-        erl_otp:
-        - 23.3.4.9-3
-        macos:
-        - macos-11
-    runs-on: ${{ matrix.macos }}
+        otp:
+          - 23.3.4.9-3
+        os:
+          - macos-11
+    runs-on: ${{ matrix.os }}
    steps:
    - uses: actions/checkout@v1
-    - name: prepare
-      run: |
-        if make emqx-ee --dry-run > /dev/null 2>&1; then
-          echo "https://ci%40emqx.io:${{ secrets.CI_GIT_TOKEN }}@github.com" > $HOME/.git-credentials
-          git config --global credential.helper store
-          echo "EMQX_NAME=emqx-ee" >> $GITHUB_ENV
-        else
-          echo "EMQX_NAME=emqx" >> $GITHUB_ENV
-        fi
-    - name: prepare
-      run: |
-        brew update
-        brew install curl zip unzip gnu-sed kerl unixodbc freetds
-        echo "/usr/local/bin" >> $GITHUB_PATH
-        git config --global credential.helper store
-    - uses: actions/cache@v2
-      id: cache
+    - uses: ./.github/actions/detect-profiles
      with:
-        path: ~/.kerl/${{ matrix.erl_otp }}
-        key: otp-install-${{ matrix.erl_otp }}-${{ matrix.macos }}-static-ssl-disable-hipe-disable-jit
-    - name: build erlang
-      if: steps.cache.outputs.cache-hit != 'true'
-      timeout-minutes: 60
-      env:
-        KERL_BUILD_BACKEND: git
-        OTP_GITHUB_URL: https://github.com/emqx/otp
-        KERL_CONFIGURE_OPTIONS: --disable-dynamic-ssl-lib --with-ssl=/usr/local/opt/openssl@1.1 --disable-hipe --disable-jit
-      run: |
-        kerl update releases
-        kerl build ${{ matrix.erl_otp }}
-        kerl install ${{ matrix.erl_otp }} $HOME/.kerl/${{ matrix.erl_otp }}
-    - name: build
-      env:
-        APPLE_SIGN_BINARIES: 1
-        APPLE_ID: developers@emqx.io
-        APPLE_TEAM_ID: 26N6HYJLZA
-        APPLE_ID_PASSWORD: ${{ secrets.APPLE_ID_PASSWORD }}
-        APPLE_DEVELOPER_IDENTITY: ${{ secrets.APPLE_DEVELOPER_IDENTITY }}
-        APPLE_DEVELOPER_ID_BUNDLE: ${{ secrets.APPLE_DEVELOPER_ID_BUNDLE }}
-        APPLE_DEVELOPER_ID_BUNDLE_PASSWORD: ${{ secrets.APPLE_DEVELOPER_ID_BUNDLE_PASSWORD }}
-      run: |
-        . $HOME/.kerl/${{ matrix.erl_otp }}/activate
-        make ensure-rebar3
-        sudo cp rebar3 /usr/local/bin/rebar3
-        make ${EMQX_NAME}-zip
+        ci_git_token: ${{ secrets.CI_GIT_TOKEN }}
+    - uses: ./.github/actions/package-macos
+      with:
+        otp: ${{ matrix.otp }}
+        os: ${{ matrix.os }}
+        apple_id_password: ${{ secrets.APPLE_ID_PASSWORD }}
+        apple_developer_identity: ${{ secrets.APPLE_DEVELOPER_IDENTITY }}
+        apple_developer_id_bundle: ${{ secrets.APPLE_DEVELOPER_ID_BUNDLE }}
+        apple_developer_id_bundle_password: ${{ secrets.APPLE_DEVELOPER_ID_BUNDLE_PASSWORD }}
    - uses: actions/upload-artifact@v1
      if: failure()
      with:
        name: rebar3.crashdump
        path: ./rebar3.crashdump
-    - name: test
-      run: |
-        pkg_name=$(basename _packages/${EMQX_NAME}/emqx-*.zip)
-        unzip -q _packages/${EMQX_NAME}/$pkg_name
-        gsed -i '/emqx_telemetry/d' ./emqx/data/loaded_plugins
-        ./emqx/bin/emqx start || cat emqx/log/erlang.log.1
-        ready='no'
-        for i in {1..10}; do
-          if curl -fs 127.0.0.1:18083 > /dev/null; then
-            ready='yes'
-            break
-          fi
-          sleep 1
-        done
-        if [ "$ready" != "yes" ]; then
-          echo "Timed out waiting for emqx to be ready"
-          cat emqx/log/erlang.log.1
-          exit 1
-        fi
-        ./emqx/bin/emqx_ctl status
-        ./emqx/bin/emqx stop
-        rm -rf emqx
    - uses: actions/upload-artifact@v2
      with:
        name: macos
--- a/.github/workflows/release.yaml
+++ b/.github/workflows/release.yaml
@ -10,24 +10,18 @@ jobs:
    container: emqx/build-env:erl23.3.4.9-3-ubuntu20.04

    outputs:
-      profiles: ${{ steps.set_profile.outputs.profiles}}
-      s3dir: ${{ steps.set_profile.outputs.s3dir}}
+      profiles: ${{ steps.detect-profiles.outputs.profiles}}

    steps:
      - uses: actions/checkout@v2
        with:
          path: source
          fetch-depth: 0
-      - name: set profile
-        id: set_profile
-        shell: bash
-        run: |
-          cd source
-          if make emqx-ee --dry-run > /dev/null 2>&1; then
-            echo "::set-output name=profiles::[\"emqx-ee\"]"
-          else
-            echo "::set-output name=profiles::[\"emqx\", \"emqx-edge\"]"
-          fi
+      - id: detect-profiles
+        working-directory: source
+        uses: ./.github/actions/detect-profiles
+        with:
+          ci_git_token: ${{ secrets.CI_GIT_TOKEN }}

  upload:
    runs-on: ubuntu-20.04
--- a/.github/workflows/run_cts_tests.yaml
+++ b/.github/workflows/run_cts_tests.yaml
@ -82,6 +82,7 @@ jobs:
      - name: docker-compose up
        run: |
          docker-compose \
+            -f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
            -f .ci/docker-compose-file/docker-compose-mongo-${{ matrix.connect_type }}.yaml \
            -f .ci/docker-compose-file/docker-compose.yaml \
            up -d --build
--- a/.github/workflows/run_test_cases.yaml
+++ b/.github/workflows/run_test_cases.yaml
@ -54,6 +54,7 @@ jobs:
          run: |
            docker-compose \
                -f .ci/docker-compose-file/docker-compose.yaml \
+                -f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
                -f .ci/docker-compose-file/docker-compose-ldap-tcp.yaml \
                -f .ci/docker-compose-file/docker-compose-mongo-tcp.yaml \
                -f .ci/docker-compose-file/docker-compose-mysql-tcp.yaml \
@ -81,6 +82,7 @@ jobs:
          run: |
            docker-compose \
                -f .ci/docker-compose-file/docker-compose.yaml \
+                -f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
                -f .ci/docker-compose-file/docker-compose-ldap-tcp.yaml \
                -f .ci/docker-compose-file/docker-compose-mongo-tcp.yaml \
                -f .ci/docker-compose-file/docker-compose-mysql-tcp.yaml \
--- a/CHANGES-4.3.md
+++ b/CHANGES-4.3.md
@ -29,13 +29,27 @@ File format:
 - TLS listener default buffer size to 4KB [#9007](https://github.com/emqx/emqx/pull/9007)
  Eliminate uncertainty that the buffer size is set by OS default.

- Fix delayed publish inaccurate caused by os time change. [#8908](https://github.com/emqx/emqx/pull/8908)
-
 - Disable authorization for `api/v4/emqx_prometheus` endpoint. [8955](https://github.com/emqx/emqx/pull/8955)

 - Added a test to prevent a last will testament message to be
  published when a client is denied connection. [#8894](https://github.com/emqx/emqx/pull/8894)

+### Bug fixes
+
+- Fix delayed publish inaccurate caused by os time change. [#8908](https://github.com/emqx/emqx/pull/8908)
+
+- Hide redis password in error logs [#9071](https://github.com/emqx/emqx/pull/9071)
+  In this change, it also included more changes in redis client:
+  - Improve redis connection error logging [eredis:19](https://github.com/emqx/eredis/pull/19).
+    Also added support for eredis to accept an anonymous function as password instead of
+    passing around plaintext args which may get dumpped to crash logs (hard to predict where).
+    This change also added `format_status` callback for `gen_server` states which hold plaintext
+    password so the process termination log and `sys:get_status` will print '******' instead of
+    the password to console.
+  - Avoid pool name clashing [eredis_cluster#22](https://github.com/emqx/eredis_cluster/pull/22)
+    Same `format_status` callback is added here too for `gen_server`s which hold password in
+    their state.
+
 ## v4.3.20

 ### Bug fixes
--- a/apps/emqx_auth_mnesia/test/emqx_acl_mnesia_SUITE.erl
+++ b/apps/emqx_auth_mnesia/test/emqx_acl_mnesia_SUITE.erl
@ -20,6 +20,7 @@
 -compile(export_all).

 -include("emqx_auth_mnesia.hrl").
+-include_lib("emqx/include/emqx.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
@ -77,15 +78,37 @@ init_per_testcase_migration(_, Config) ->
    emqx_acl_mnesia_migrator:migrate_records(),
    Config.

+init_per_testcase_other(t_last_will_testament_message_check_acl, Config) ->
+    OriginalACLNoMatch = application:get_env(emqx, acl_nomatch),
+    application:set_env(emqx, acl_nomatch, deny),
+    emqx_mod_acl_internal:unload([]),
+    %% deny all for this client
+    ClientID = <<"lwt_client">>,
+    ok = emqx_acl_mnesia_db:add_acl({clientid, ClientID}, <<"#">>, pubsub, deny),
+    [ {original_acl_nomatch, OriginalACLNoMatch}
+    , {clientid, ClientID}
+    | Config];
+init_per_testcase_other(_TestCase, Config) ->
+    Config.
+
 init_per_testcase(Case, Config) ->
    PerTestInitializers = [
        fun init_per_testcase_clean/2,
        fun init_per_testcase_migration/2,
-        fun init_per_testcase_emqx_hook/2
+        fun init_per_testcase_emqx_hook/2,
+        fun init_per_testcase_other/2
    ],
    lists:foldl(fun(Init, Conf) -> Init(Case, Conf) end, Config, PerTestInitializers).

-end_per_testcase(_, Config) ->
+end_per_testcase(t_last_will_testament_message_check_acl, Config) ->
+    emqx:unhook('client.check_acl', fun emqx_acl_mnesia:check_acl/5),
+    case ?config(original_acl_nomatch, Config) of
+        {ok, Original} -> application:set_env(emqx, acl_nomatch, Original);
+        _ -> ok
+    end,
+    emqx_mod_acl_internal:load([]),
+    ok;
+end_per_testcase(_TestCase, Config) ->
    emqx:unhook('client.check_acl', fun emqx_acl_mnesia:check_acl/5),
    Config.

@ -464,6 +487,35 @@ t_rest_api(_Config) ->
    {ok, Res3} = request_http_rest_list(["$all"]),
    ?assertMatch([], get_http_data(Res3)).

+%% asserts that we check ACL for the LWT topic before publishing the
+%% LWT.
+t_last_will_testament_message_check_acl(Config) ->
+    ClientID = ?config(clientid, Config),
+    {ok, C} = emqtt:start_link([
+        {clientid, ClientID},
+        {will_topic, <<"$SYS/lwt">>},
+        {will_payload, <<"should not be published">>}
+    ]),
+    {ok, _} = emqtt:connect(C),
+    ok = emqx:subscribe(<<"$SYS/lwt">>),
+    unlink(C),
+    ok = snabbkaffe:start_trace(),
+    {true, {ok, _}} =
+        ?wait_async_action(
+        exit(C, kill),
+        #{?snk_kind := last_will_testament_publish_denied},
+        1_000
+    ),
+    ok = snabbkaffe:stop(),
+
+    receive
+        {deliver, <<"$SYS/lwt">>, #message{payload = <<"should not be published">>}} ->
+            error(lwt_should_not_be_published_to_forbidden_topic)
+    after 1_000 ->
+        ok
+    end,
+
+    ok.

 create_conflicting_records() ->
    Records = [
--- a/apps/emqx_auth_mnesia/test/emqx_auth_mnesia_SUITE.erl
+++ b/apps/emqx_auth_mnesia/test/emqx_auth_mnesia_SUITE.erl
@ -408,7 +408,7 @@ t_password_hash(_) ->
    ok = application:start(emqx_auth_mnesia).

 t_will_message_connection_denied(Config) when is_list(Config) ->
-    ClientId = Username = <<"subscriber">>,
+    ClientId = <<"subscriber">>,
    Password = <<"p">>,
    application:stop(emqx_auth_mnesia),
    ok = emqx_ct_helpers:start_apps([emqx_auth_mnesia]),
--- a/apps/emqx_auth_mongo/src/emqx_acl_mongo.erl
+++ b/apps/emqx_auth_mongo/src/emqx_acl_mongo.erl
@ -79,4 +79,3 @@ feedvar(Str, Var, Val) ->
    re:replace(Str, Var, Val, [global, {return, binary}]).

 description() -> "ACL with MongoDB".
-
--- a/apps/emqx_auth_mongo/src/emqx_auth_mongo.app.src
+++ b/apps/emqx_auth_mongo/src/emqx_auth_mongo.app.src
@ -1,6 +1,6 @@
 {application, emqx_auth_mongo,
 [{description, "EMQ X Authentication/ACL with MongoDB"},
-  {vsn, "4.3.4"}, % strict semver, bump manually!
+  {vsn, "4.3.5"}, % strict semver, bump manually!
  {modules, []},
  {registered, [emqx_auth_mongo_sup]},
  {applications, [kernel,stdlib,mongodb,ecpool]},
--- a/apps/emqx_auth_mongo/src/emqx_auth_mongo.appup.src
+++ b/apps/emqx_auth_mongo/src/emqx_auth_mongo.appup.src
@ -1,7 +1,8 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{<<"4\\.3\\.[1-3]">>,
+  [{"4.3.4",[{load_module,emqx_auth_mongo,brutal_purge,soft_purge,[]}]},
+   {<<"4\\.3\\.[1-3]">>,
    [{load_module,emqx_auth_mongo_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_auth_mongo,brutal_purge,soft_purge,[]}]},
   {"4.3.0",
@ -9,7 +10,8 @@
     {load_module,emqx_auth_mongo,brutal_purge,soft_purge,[]},
     {load_module,emqx_acl_mongo,brutal_purge,soft_purge,[]}]},
   {<<".*">>,[]}],
-  [{<<"4\\.3\\.[1-3]">>,
+  [{"4.3.4",[{load_module,emqx_auth_mongo,brutal_purge,soft_purge,[]}]},
+   {<<"4\\.3\\.[1-3]">>,
    [{load_module,emqx_auth_mongo_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_auth_mongo,brutal_purge,soft_purge,[]}]},
   {"4.3.0",
--- a/apps/emqx_auth_mongo/src/emqx_auth_mongo.erl
+++ b/apps/emqx_auth_mongo/src/emqx_auth_mongo.erl
@ -22,6 +22,7 @@
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("emqx/include/logger.hrl").
 -include_lib("emqx/include/types.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").

 -export([ check/3
        , description/0
@ -38,14 +39,22 @@
        , available/3
        ]).

+-ifdef(TEST).
+-export([ is_superuser/3
+        , available/4
+        ]).
+-endif.
+
 check(ClientInfo = #{password := Password}, AuthResult,
      Env = #{authquery := AuthQuery, superquery := SuperQuery}) ->
+    ?tp(emqx_auth_mongo_superuser_check_authn_enter, #{}),
    #authquery{collection = Collection, field = Fields,
               hash = HashType, selector = Selector} = AuthQuery,
    Pool = maps:get(pool, Env, ?APP),
    case query(Pool, Collection, maps:from_list(replvars(Selector, ClientInfo))) of
        undefined -> ok;
        {error, Reason} ->
+            ?tp(emqx_auth_mongo_check_authn_error, #{error => Reason}),
            ?LOG(error, "[MongoDB] Can't connect to MongoDB server: ~0p", [Reason]),
            {stop, AuthResult#{auth_result => not_authorized, anonymous => false}};
        UserMap ->
@ -58,6 +67,7 @@ check(ClientInfo = #{password := Password}, AuthResult,
                     end,
            case Result of
                ok ->
+                    ?tp(emqx_auth_mongo_superuser_check_authn_ok, #{}),
                    {stop, AuthResult#{is_superuser => is_superuser(Pool, SuperQuery, ClientInfo),
                                       anonymous => false,
                                       auth_result => success}};
@ -81,17 +91,24 @@ description() -> "Authentication with MongoDB".
 is_superuser(_Pool, undefined, _ClientInfo) ->
    false;
 is_superuser(Pool, #superquery{collection = Coll, field = Field, selector = Selector}, ClientInfo) ->
-    case query(Pool, Coll, maps:from_list(replvars(Selector, ClientInfo))) of
-        undefined -> false;
-        {error, Reason} ->
-            ?LOG(error, "[MongoDB] Can't connect to MongoDB server: ~0p", [Reason]),
-            false;
-        Row ->
-            case maps:get(Field, Row, false) of
-                true   -> true;
-                _False -> false
-            end
-    end.
+    ?tp(emqx_auth_mongo_superuser_query_enter, #{}),
+    Res =
+      case query(Pool, Coll, maps:from_list(replvars(Selector, ClientInfo))) of
+          undefined ->
+              %% returned when there are no returned documents
+              false;
+          {error, Reason} ->
+              ?tp(emqx_auth_mongo_superuser_query_error, #{error => Reason}),
+              ?LOG(error, "[MongoDB] Can't connect to MongoDB server: ~0p", [Reason]),
+              false;
+          Row ->
+              case maps:get(Field, Row, false) of
+                  true   -> true;
+                  _False -> false
+              end
+      end,
+    ?tp(emqx_auth_mongo_superuser_query_result, #{is_superuser => Res}),
+    Res.

 %%--------------------------------------------------------------------
 %% Availability Test
@ -114,6 +131,7 @@ available(Pool, Collection, Query) ->
 available(Pool, Collection, Query, Fun) ->
    try Fun(Pool, Collection, Query) of
        {error, Reason} ->
+            ?tp(emqx_auth_mongo_available_error, #{error => Reason}),
            ?LOG(error, "[MongoDB] ~p availability test error: ~0p", [Collection, Reason]),
            {error, Reason};
        Error = #{<<"code">> := Code} ->
@ -144,7 +162,16 @@ test_client_info() ->
 %%--------------------------------------------------------------------

 replvars(VarList, ClientInfo) ->
-    lists:map(fun(Var) -> replvar(Var, ClientInfo) end, VarList).
+    lists:foldl(
+     fun(Var, Selector) ->
+       case replvar(Var, ClientInfo) of
+           %% assumes that all fields are binaries...
+           {unmatchable, Field} -> [{Field, []} | Selector];
+           Interpolated -> [Interpolated | Selector]
+       end
+     end,
+     [],
+     VarList).

 replvar({Field, <<"%u">>}, #{username := Username}) ->
    {Field, Username};
@ -154,8 +181,8 @@ replvar({Field, <<"%C">>}, #{cn := CN}) ->
    {Field, CN};
 replvar({Field, <<"%d">>}, #{dn := DN}) ->
    {Field, DN};
-replvar(Selector, _ClientInfo) ->
-    Selector.
+replvar({Field, _PlaceHolder}, _ClientInfo) ->
+    {unmatchable, Field}.

 %%--------------------------------------------------------------------
 %% MongoDB Connect/Query
@ -169,19 +196,57 @@ connect(Opts) ->
    mongo_api:connect(Type, Hosts, Options, WorkerOptions).

 query(Pool, Collection, Selector) ->
-    ecpool:with_client(Pool, fun(Conn) -> mongo_api:find_one(Conn, Collection, Selector, #{}) end).
+    Timeout = timer:seconds(15),
+    with_timeout(Timeout, fun() ->
+      ecpool:with_client(Pool, fun(Conn) -> mongo_api:find_one(Conn, Collection, Selector, #{}) end)
+    end).

 query_multi(Pool, Collection, SelectorList) ->
+    ?tp(emqx_auth_mongo_query_multi_enter, #{}),
+    Timeout = timer:seconds(45),
    lists:reverse(lists:flatten(lists:foldl(fun(Selector, Acc1) ->
-        Batch = ecpool:with_client(Pool, fun(Conn) ->
-                  case mongo_api:find(Conn, Collection, Selector, #{}) of
-                      {error, Reason} ->
-                          ?LOG(error, "[MongoDB] query_multi failed, got error: ~p", [Reason]),
-                          [];
-                      [] -> [];
-                      {ok, Cursor} ->
-                          mc_cursor:foldl(fun(O, Acc2) -> [O|Acc2] end, [], Cursor, 1000)
-                  end
-                end),
-        [Batch|Acc1]
+        Res =
+          with_timeout(Timeout, fun() ->
+            ecpool:with_client(Pool, fun(Conn) ->
+              ?tp(emqx_auth_mongo_query_multi_find_selector, #{}),
+              case find(Conn, Collection, Selector) of
+                  {error, Reason} ->
+                      ?tp(emqx_auth_mongo_query_multi_error,
+                          #{error => Reason}),
+                      ?LOG(error, "[MongoDB] query_multi failed, got error: ~p", [Reason]),
+                      [];
+                  [] ->
+                      ?tp(emqx_auth_mongo_query_multi_no_results, #{}),
+                      [];
+                  {ok, Cursor} ->
+                      mc_cursor:foldl(fun(O, Acc2) -> [O | Acc2] end, [], Cursor, 1000)
+              end
+            end)
+          end),
+        case Res of
+            {error, timeout} ->
+                ?tp(emqx_auth_mongo_query_multi_error, #{error => timeout}),
+                ?LOG(error, "[MongoDB] query_multi timeout", []),
+                Acc1;
+            Batch ->
+                [Batch | Acc1]
+        end
    end, [], SelectorList))).
+
+find(Conn, Collection, Selector) ->
+    try
+        mongo_api:find(Conn, Collection, Selector, #{})
+    catch
+        K:E:S ->
+            {error, {K, E, S}}
+    end.
+
+with_timeout(Timeout, Fun) ->
+    try
+        emqx_misc:nolink_apply(Fun, Timeout)
+    catch
+        exit:timeout ->
+            {error, timeout};
+        K:E:S ->
+            erlang:raise(K, E, S)
+    end.
--- a/apps/emqx_auth_mongo/src/emqx_auth_mongo_app.erl
+++ b/apps/emqx_auth_mongo/src/emqx_auth_mongo_app.erl
@ -30,6 +30,10 @@
        , stop/1
        ]).

+-ifdef(TEST).
+-export([with_env/2]).
+-endif.
+
 %%--------------------------------------------------------------------
 %% Application callbacks
 %%--------------------------------------------------------------------
--- a/apps/emqx_auth_mongo/test/emqx_auth_mongo_SUITE.erl
+++ b/apps/emqx_auth_mongo/test/emqx_auth_mongo_SUITE.erl
@ -19,47 +19,103 @@
 -compile(export_all).
 -compile(nowarn_export_all).

+-include("emqx_auth_mongo.hrl").
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
-
-define(APP, emqx_auth_mongo).
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").

 -define(POOL(App),  ecpool_worker:client(gproc_pool:pick_worker({ecpool, App}))).

 -define(MONGO_CL_ACL, <<"mqtt_acl">>).
 -define(MONGO_CL_USER, <<"mqtt_user">>).

-define(INIT_ACL, [{<<"username">>, <<"testuser">>, <<"clientid">>, <<"null">>, <<"subscribe">>, [<<"#">>]},
-                   {<<"username">>, <<"dashboard">>, <<"clientid">>, <<"null">>, <<"pubsub">>, [<<"$SYS/#">>]},
-                   {<<"username">>, <<"user3">>, <<"clientid">>, <<"null">>, <<"publish">>, [<<"a/b/c">>]}]).
+-define(INIT_ACL, [ { <<"username">>, <<"testuser">>
+                    , <<"clientid">>, <<"null">>
+                    , <<"subscribe">>, [<<"#">>]
+                    }
+                  , { <<"username">>, <<"dashboard">>
+                    , <<"clientid">>, <<"null">>
+                    , <<"pubsub">>, [<<"$SYS/#">>]
+                    }
+                  , { <<"username">>, <<"user3">>
+                    , <<"clientid">>, <<"null">>
+                    , <<"publish">>, [<<"a/b/c">>]
+                    }
+                  ]).

-define(INIT_AUTH, [{<<"username">>, <<"plain">>, <<"password">>, <<"plain">>, <<"salt">>, <<"salt">>, <<"is_superuser">>, true},
-                    {<<"username">>, <<"md5">>, <<"password">>, <<"1bc29b36f623ba82aaf6724fd3b16718">>, <<"salt">>, <<"salt">>, <<"is_superuser">>, false},
-                    {<<"username">>, <<"sha">>, <<"password">>, <<"d8f4590320e1343a915b6394170650a8f35d6926">>, <<"salt">>, <<"salt">>, <<"is_superuser">>, false},
-                    {<<"username">>, <<"sha256">>, <<"password">>, <<"5d5b09f6dcb2d53a5fffc60c4ac0d55fabdf556069d6631545f42aa6e3500f2e">>, <<"salt">>, <<"salt">>, <<"is_superuser">>, false},
-                    {<<"username">>, <<"pbkdf2_password">>, <<"password">>, <<"cdedb5281bb2f801565a1122b2563515">>, <<"salt">>, <<"ATHENA.MIT.EDUraeburn">>, <<"is_superuser">>, false},
-                    {<<"username">>, <<"bcrypt_foo">>, <<"password">>, <<"$2a$12$sSS8Eg.ovVzaHzi1nUHYK.HbUIOdlQI0iS22Q5rd5z.JVVYH6sfm6">>, <<"salt">>, <<"$2a$12$sSS8Eg.ovVzaHzi1nUHYK.">>, <<"is_superuser">>, false}
-                    ]).
+-define(INIT_AUTH, [ { <<"username">>, <<"plain">>
+                     , <<"password">>, <<"plain">>
+                     , <<"salt">>, <<"salt">>
+                     , <<"is_superuser">>, true
+                     }
+                   , { <<"username">>, <<"md5">>
+                     , <<"password">>, <<"1bc29b36f623ba82aaf6724fd3b16718">>
+                     , <<"salt">>, <<"salt">>
+                     , <<"is_superuser">>, false
+                     }
+                   , { <<"username">>, <<"sha">>
+                     , <<"password">>, <<"d8f4590320e1343a915b6394170650a8f35d6926">>
+                     , <<"salt">>, <<"salt">>
+                     , <<"is_superuser">>, false
+                     }
+                   , { <<"username">>, <<"sha256">>
+                     , <<"password">>, <<"5d5b09f6dcb2d53a5fffc60c4ac0d55fabdf556069d6631545f42aa6e3500f2e">>
+                     , <<"salt">>, <<"salt">>
+                     , <<"is_superuser">>, false
+                     }
+                   , { <<"username">>, <<"pbkdf2_password">>
+                     , <<"password">>, <<"cdedb5281bb2f801565a1122b2563515">>
+                     , <<"salt">>, <<"ATHENA.MIT.EDUraeburn">>
+                     , <<"is_superuser">>, false
+                     }
+                   , { <<"username">>, <<"bcrypt_foo">>
+                     , <<"password">>, <<"$2a$12$sSS8Eg.ovVzaHzi1nUHYK.HbUIOdlQI0iS22Q5rd5z.JVVYH6sfm6">>
+                     , <<"salt">>, <<"$2a$12$sSS8Eg.ovVzaHzi1nUHYK.">>
+                     , <<"is_superuser">>, false
+                     }
+                   , { <<"username">>, <<"user_full">>
+                     , <<"clientid">>, <<"client_full">>
+                     , <<"common_name">>, <<"cn_full">>
+                     , <<"distinguished_name">>, <<"dn_full">>
+                     , <<"password">>, <<"plain">>
+                     , <<"salt">>, <<"salt">>
+                     , <<"is_superuser">>, false
+                     }
+                   ]).

 %%--------------------------------------------------------------------
 %% Setups
 %%--------------------------------------------------------------------

 all() ->
-    emqx_ct:all(?MODULE).
+    OtherTCs = emqx_ct:all(?MODULE) -- resilience_tests(),
+    [ {group, resilience}
+    | OtherTCs].

-init_per_suite(Cfg) ->
+resilience_tests() ->
+    [ t_acl_superuser_timeout
+    , t_available_acl_query_no_connection
+    , t_available_acl_query_timeout
+    , t_available_authn_query_timeout
+    , t_authn_timeout
+    , t_available
+    ].
+
+groups() ->
+    [ {resilience, resilience_tests()}
+    ].
+
+init_per_suite(Config) ->
    emqx_ct_helpers:start_apps([emqx_auth_mongo], fun set_special_confs/1),
-    init_mongo_data(),
    %% avoid inter-suite flakiness
    ok = emqx_mod_acl_internal:unload([]),
-    Cfg.
+    Config.

 end_per_suite(_Cfg) ->
    deinit_mongo_data(),
    %% avoid inter-suite flakiness
-    ok = emqx_mod_acl_internal:load([]),
+    emqx_mod_acl_internal:load([]),
    emqx_ct_helpers:stop_apps([emqx_auth_mongo]).

 set_special_confs(emqx) ->
@ -69,6 +125,83 @@ set_special_confs(emqx) ->
 set_special_confs(_App) ->
    ok.

+init_per_group(resilience, Config) ->
+    ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"),
+    ProxyPortStr = os:getenv("PROXY_PORT", "8474"),
+    ProxyPort = list_to_integer(ProxyPortStr),
+    reset_proxy(ProxyHost, ProxyPort),
+    ProxyServer = ProxyHost ++ ":27017",
+    {ok, OriginalServer} = application:get_env(emqx_auth_mongo, server),
+    OriginalServerMap = maps:from_list(OriginalServer),
+    NewServerMap = OriginalServerMap#{hosts => [ProxyServer]},
+    NewServer = maps:to_list(NewServerMap),
+    emqx_ct_helpers:stop_apps([emqx_auth_mongo]),
+    Handler =
+        fun(App = emqx_auth_mongo) ->
+             application:set_env(emqx_auth_mongo, server, NewServer),
+             set_special_confs(App);
+           (App)->
+             set_special_confs(App)
+        end,
+    emqx_ct_helpers:start_apps([emqx_auth_mongo], Handler),
+    [ {original_server, OriginalServer}
+    , {proxy_host, ProxyHost}
+    , {proxy_port, ProxyPort}
+    | Config];
+init_per_group(_Group, Config) ->
+    Config.
+
+end_per_group(resilience, Config) ->
+    OriginalServer = ?config(original_server, Config),
+    application:set_env(emqx_auth_mongo, server, OriginalServer),
+    emqx_ct_helpers:stop_apps([emqx_auth_mongo]),
+    emqx_ct_helpers:start_apps([emqx_auth_mongo], fun set_special_confs/1),
+    ok;
+end_per_group(_Group, _Config) ->
+    ok.
+
+init_per_testcase(t_authn_full_selector_variables, Config) ->
+    {ok, AuthQuery} = application:get_env(emqx_auth_mongo, auth_query),
+    OriginalSelector = proplists:get_value(selector, AuthQuery),
+    Selector = [ {<<"username">>, <<"%u">>}
+               , {<<"clientid">>, <<"%c">>}
+               , {<<"common_name">>, <<"%C">>}
+               , {<<"distinguished_name">>, <<"%d">>}
+               ],
+    reload({auth_query, [{selector, Selector}]}),
+    init_mongo_data(),
+    [ {original_selector, OriginalSelector}
+    , {selector, Selector}
+    | Config];
+init_per_testcase(_TestCase, Config) ->
+    init_mongo_data(),
+    Config.
+
+end_per_testcase(t_authn_full_selector_variables, Config) ->
+    OriginalSelector = ?config(original_selector, Config),
+    reload({auth_query, [{selector, OriginalSelector}]}),
+    deinit_mongo_data(),
+    ok;
+end_per_testcase(TestCase, Config)
+ when TestCase =:= t_available_acl_query_timeout;
+      TestCase =:= t_acl_superuser_timeout;
+      TestCase =:= t_authn_no_connection;
+      TestCase =:= t_available_authn_query_timeout;
+      TestCase =:= t_authn_timeout;
+      TestCase =:= t_available_acl_query_no_connection ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    reset_proxy(ProxyHost, ProxyPort),
+    %% force restart of clients because CI tends to get stuck...
+    application:stop(emqx_auth_mongo),
+    application:start(emqx_auth_mongo),
+    wait_for_stabilization(#{attempts => 10, interval_ms => 500}),
+    deinit_mongo_data(),
+    ok;
+end_per_testcase(_TestCase, _Config) ->
+    deinit_mongo_data(),
+    ok.
+
 init_mongo_data() ->
    %% Users
    {ok, Connection} = ?POOL(?APP),
@ -87,6 +220,14 @@ deinit_mongo_data() ->
 %% Test cases
 %%--------------------------------------------------------------------

+%% for full coverage ;-)
+t_authn_description(_Config) ->
+    ?assert(is_list(emqx_auth_mongo:description())).
+
+%% for full coverage ;-)
+t_acl_description(_Config) ->
+    ?assert(is_list(emqx_acl_mongo:description())).
+
 t_check_auth(_) ->
    Plain = #{zone => external, clientid => <<"client1">>, username => <<"plain">>},
    Plain1 = #{zone => external, clientid => <<"client1">>, username => <<"plain2">>},
@ -116,7 +257,124 @@ t_check_auth(_) ->
    {ok, #{is_superuser := false}} = emqx_access_control:authenticate(Pbkdf2#{password => <<"password">>}),
    reload({auth_query, [{password_hash, {salt, bcrypt}}]}),
    {ok, #{is_superuser := false}} = emqx_access_control:authenticate(Bcrypt#{password => <<"foo">>}),
-    {error, _} = emqx_access_control:authenticate(User1#{password => <<"foo">>}).
+    {error, _} = emqx_access_control:authenticate(User1#{password => <<"foo">>}),
+    %% bad field config
+    reload({auth_query, [{password_field, [<<"bad_field">>]}]}),
+    ?assertEqual({error, password_error},
+                 emqx_access_control:authenticate(Plain#{password => <<"plain">>})),
+    %% unknown username
+    Unknown = #{zone => unknown, clientid => <<"?">>, username => <<"?">>, password => <<"">>},
+    ?assertEqual({error, not_authorized}, emqx_access_control:authenticate(Unknown)),
+    ok.
+
+t_authn_full_selector_variables(Config) ->
+    Selector = ?config(selector, Config),
+    ClientInfo = #{ zone => external
+                  , clientid => <<"client_full">>
+                  , username => <<"user_full">>
+                  , cn => <<"cn_full">>
+                  , dn => <<"dn_full">>
+                  , password => <<"plain">>
+                  },
+    ?assertMatch({ok, _}, emqx_access_control:authenticate(ClientInfo)),
+    EnvFields = [ clientid
+                , username
+                , cn
+                , dn
+                ],
+    lists:foreach(
+     fun(Field) ->
+       UnauthorizedClientInfo = ClientInfo#{Field => <<"wrong">>},
+       ?assertEqual({error, not_authorized},
+                    emqx_access_control:authenticate(UnauthorizedClientInfo),
+                    #{ field => Field
+                     , client_info => UnauthorizedClientInfo
+                     , selector => Selector
+                     })
+     end,
+     EnvFields),
+    ok.
+
+t_authn_interpolation_no_info(_Config) ->
+    Valid = #{zone => external, clientid => <<"client1">>,
+              username => <<"plain">>, password => <<"plain">>},
+    ?assertMatch({ok, _}, emqx_access_control:authenticate(Valid)),
+    try
+        %% has values that are equal to placeholders
+        InterpolationUser = #{ <<"username">> => <<"%u">>
+                             , <<"password">> => <<"plain">>
+                             , <<"salt">> => <<"salt">>
+                             , <<"is_superuser">> => true
+                             },
+        {ok, Conn} = ?POOL(?APP),
+        {{true, _}, _} = mongo_api:insert(Conn, ?MONGO_CL_USER, InterpolationUser),
+        Invalid = maps:without([username], Valid),
+        ?assertMatch({error, not_authorized}, emqx_access_control:authenticate(Invalid))
+    after
+        deinit_mongo_data(),
+        init_mongo_data()
+    end.
+
+%% authenticates, but superquery returns no documents
+t_authn_empty_is_superuser_collection(_Config) ->
+    {ok, SuperQuery} = application:get_env(emqx_auth_mongo, super_query),
+    Collection = list_to_binary(proplists:get_value(collection, SuperQuery)),
+    reload({auth_query, [{password_hash, plain}]}),
+    Plain = #{zone => external, clientid => <<"client1">>,
+              username => <<"plain">>, password => <<"plain">>},
+    ok = snabbkaffe:start_trace(),
+    ?force_ordering(
+      #{?snk_kind := emqx_auth_mongo_superuser_check_authn_ok},
+      #{?snk_kind := truncate_coll_enter}),
+    ?force_ordering(
+      #{?snk_kind := truncate_coll_done},
+      #{?snk_kind := emqx_auth_mongo_superuser_query_enter}),
+    try
+        spawn_link(fun() ->
+          ?tp(truncate_coll_enter, #{}),
+          {ok, Conn} = ?POOL(?APP),
+          {true, _} = mongo_api:delete(Conn, Collection, _Selector = #{}),
+          ?tp(truncate_coll_done, #{})
+        end),
+        ?assertMatch({ok, #{is_superuser := false}}, emqx_access_control:authenticate(Plain)),
+        ok = snabbkaffe:stop(),
+        ok
+    after
+        init_mongo_data()
+    end.
+
+t_available(Config) ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    Pool = ?APP,
+    SuperQuery = #superquery{collection = SuperCollection} = superquery(),
+    %% success;
+    ?assertEqual(ok, emqx_auth_mongo:available(Pool, SuperQuery)),
+    %% error with code;
+    EmptySelector = #{},
+    ?assertEqual(
+       {error, {mongo_error, 2}},
+       emqx_auth_mongo:available(Pool, SuperCollection, EmptySelector, fun error_code_query/3)),
+    %% exception (in query)
+    ?assertMatch(
+       {error, _},
+       with_failure(down, ProxyHost, ProxyPort,
+         fun() ->
+           Collection = <<"mqtt_user">>,
+           Selector = #{},
+           emqx_auth_mongo:available(Pool, Collection, Selector)
+         end)),
+    %% exception (arbitrary function)
+    ?assertMatch(
+       {error, _},
+       with_failure(down, ProxyHost, ProxyPort,
+         fun() ->
+           Collection = <<"mqtt_user">>,
+           Selector = #{},
+           RaisingFun = fun(_, _, _) -> error(some_error) end,
+           emqx_auth_mongo:available(Pool, Collection, Selector, RaisingFun)
+         end)),
+    ok.

 t_check_acl(_) ->
    {ok, Connection} = ?POOL(?APP),
@ -132,7 +390,30 @@ t_check_acl(_) ->
    allow = emqx_access_control:check_acl(User2, subscribe, <<"$SYS/testuser/1">>),
    allow = emqx_access_control:check_acl(User3, publish, <<"a/b/c">>),
    deny = emqx_access_control:check_acl(User3, publish, <<"c">>),
-    deny = emqx_access_control:check_acl(User4, publish, <<"a/b/c">>).
+    deny = emqx_access_control:check_acl(User4, publish, <<"a/b/c">>),
+    %% undefined value to interpolate
+    User1Undef = User1#{clientid => undefined},
+    allow = emqx_access_control:check_acl(User1Undef, subscribe, <<"users/testuser/1">>),
+    ok.
+
+t_acl_empty_results(_Config) ->
+    #aclquery{selector = Selector} = aclquery(),
+    User1 = #{zone => external, clientid => <<"client1">>, username => <<"testuser">>},
+    try
+        reload({acl_query, [{selector, []}]}),
+        ?assertEqual(deny, emqx_access_control:check_acl(User1, subscribe, <<"users/testuser/1">>)),
+        ok
+    after
+        reload({acl_query, [{selector, Selector}]})
+    end,
+    ok.
+
+t_acl_exception(_Config) ->
+    %% FIXME: is there a more authentic way to produce an exception in
+    %% `match'???
+    User1 = #{zone => external, clientid => not_a_binary, username => <<"testuser">>},
+    ?assertEqual(deny, emqx_access_control:check_acl(User1, subscribe, <<"users/testuser/1">>)),
+    ok.

 t_acl_super(_) ->
    reload({auth_query, [{password_hash, plain}, {password_field, [<<"password">>]}]}),
@ -155,10 +436,175 @@ t_acl_super(_) ->
    end,
    emqtt:disconnect(C).

+%% apparently, if the config is undefined in `emqx_auth_mongo_app:r',
+%% this is allowed...
+t_is_superuser_undefined(_Config) ->
+    Pool = ClientInfo = unused_in_this_case,
+    SuperQuery = undefined,
+    ?assertNot(emqx_auth_mongo:is_superuser(Pool, SuperQuery, ClientInfo)),
+    ok.
+
+t_authn_timeout(Config) ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    FailureType = timeout,
+    {ok, C} = emqtt:start_link([{clientid, <<"simpleClient">>},
+                                {username, <<"plain">>},
+                                {password, <<"plain">>}]),
+    unlink(C),
+
+    ?check_trace(
+       try
+           enable_failure(FailureType, ProxyHost, ProxyPort),
+           {error, {unauthorized_client, _}} = emqtt:connect(C),
+           ok
+       after
+           heal_failure(FailureType, ProxyHost, ProxyPort)
+       end,
+       fun(Trace) ->
+           %% fails with `{exit,{{{badmatch,{{error,closed},...'
+           ?assertMatch([_], ?of_kind(emqx_auth_mongo_check_authn_error, Trace)),
+           ok
+       end),
+
+    ok.
+
+%% tests query timeout failure
+t_available_authn_query_timeout(Config) ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    FailureType = timeout,
+    SuperQuery = superquery(),
+
+    ?check_trace(
+       #{timetrap => timer:seconds(60)},
+       try
+           enable_failure(FailureType, ProxyHost, ProxyPort),
+           Pool = ?APP,
+           %% query_multi returns an empty list even on failures.
+           ?assertEqual({error, timeout}, emqx_auth_mongo:available(Pool, SuperQuery)),
+           ok
+       after
+           heal_failure(FailureType, ProxyHost, ProxyPort)
+       end,
+       fun(Trace) ->
+         ?assertMatch(
+            [#{?snk_kind := emqx_auth_mongo_available_error , error := _}],
+            ?of_kind(emqx_auth_mongo_available_error, Trace))
+       end),
+
+    ok.
+
+%% tests query_multi failure
+t_available_acl_query_no_connection(Config) ->
+    test_acl_query_failure(down, Config).
+
+%% ensure query_multi has a timeout
+t_available_acl_query_timeout(Config) ->
+    ct:timetrap(90000),
+    test_acl_query_failure(timeout, Config).
+
+%% checks that `with_timeout' lets unknown errors pass through
+t_query_multi_unknown_exception(_Config) ->
+    ok = meck:new(ecpool, [no_link, no_history, non_strict, passthrough]),
+    ok = meck:expect(ecpool, with_client, fun(_, _) -> throw(some_error) end),
+    Pool = ?APP,
+    Collection = ?MONGO_CL_ACL,
+    SelectorList = [#{<<"username">> => <<"user">>}],
+    try
+        ?assertThrow(some_error, emqx_auth_mongo:query_multi(Pool, Collection, SelectorList))
+    after
+        meck:unload(ecpool)
+    end.
+
+t_acl_superuser_timeout(Config) ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    FailureType = timeout,
+    reload({auth_query, [{password_hash, plain}, {password_field, [<<"password">>]}]}),
+    {ok, C} = emqtt:start_link([{clientid, <<"simpleClient">>},
+                                {username, <<"plain">>},
+                                {password, <<"plain">>}]),
+    unlink(C),
+
+    ?check_trace(
+       try
+           ?force_ordering(
+              #{?snk_kind := emqx_auth_mongo_superuser_check_authn_ok},
+              #{?snk_kind := connection_will_cut}
+             ),
+           ?force_ordering(
+              #{?snk_kind := connection_cut},
+              #{?snk_kind := emqx_auth_mongo_superuser_query_enter}
+             ),
+           spawn(fun() ->
+                   ?tp(connection_will_cut, #{}),
+                   enable_failure(FailureType, ProxyHost, ProxyPort),
+                   ?tp(connection_cut, #{})
+                 end),
+
+           {ok, _} = emqtt:connect(C),
+           ok = emqtt:disconnect(C),
+           ok
+       after
+           heal_failure(FailureType, ProxyHost, ProxyPort)
+       end,
+       fun(Trace) ->
+         ?assertMatch(
+            [ #{ ?snk_kind := emqx_auth_mongo_superuser_query_error
+               , error := _
+               }
+            , #{ ?snk_kind := emqx_auth_mongo_superuser_query_result
+               , is_superuser := false
+               }
+            ],
+            ?of_kind([ emqx_auth_mongo_superuser_query_error
+                     , emqx_auth_mongo_superuser_query_result
+                     ], Trace))
+       end),
+
+    ok.
+
 %%--------------------------------------------------------------------
 %% Utils
 %%--------------------------------------------------------------------

+test_acl_query_failure(FailureType, Config) ->
+    ProxyHost = ?config(proxy_host, Config),
+    ProxyPort = ?config(proxy_port, Config),
+    ACLQuery = aclquery(),
+
+    ?check_trace(
+       #{timetrap => timer:seconds(60)},
+       try
+           ?force_ordering(
+              #{?snk_kind := emqx_auth_mongo_query_multi_enter},
+              #{?snk_kind := connection_will_cut}
+             ),
+           ?force_ordering(
+              #{?snk_kind := connection_cut},
+              #{?snk_kind := emqx_auth_mongo_query_multi_find_selector}
+             ),
+           spawn(fun() ->
+                   ?tp(connection_will_cut, #{}),
+                   enable_failure(FailureType, ProxyHost, ProxyPort),
+                   ?tp(connection_cut, #{})
+                 end),
+           Pool = ?APP,
+           %% query_multi returns an empty list even on failures.
+           ?assertMatch(ok, emqx_auth_mongo:available(Pool, ACLQuery)),
+           ok
+       after
+           heal_failure(FailureType, ProxyHost, ProxyPort)
+       end,
+       fun(Trace) ->
+         ?assertMatch(
+            [#{?snk_kind := emqx_auth_mongo_query_multi_error , error := _}],
+            ?of_kind(emqx_auth_mongo_query_multi_error, Trace))
+       end),
+
+    ok.
+
 reload({Par, Vals}) when is_list(Vals) ->
    application:stop(?APP),
    {ok, TupleVals} = application:get_env(?APP, Par),
@ -171,3 +617,105 @@ reload({Par, Vals}) when is_list(Vals) ->
    end, TupleVals),
    application:set_env(?APP, Par, lists:append(NewVals, Vals)),
    application:start(?APP).
+
+superquery() ->
+    emqx_auth_mongo_app:with_env(super_query, fun(SQ) -> SQ end).
+
+aclquery() ->
+    emqx_auth_mongo_app:with_env(acl_query, fun(SQ) -> SQ end).
+
+%% TODO: any easier way to make mongo return a map with an error code???
+error_code_query(Pool, Collection, Selector) ->
+    %% should be a query; this is to provoke an error return from
+    %% mongo.
+    WrongLimit = {},
+    ecpool:with_client(
+      Pool,
+      fun(Conn) ->
+        mongoc:transaction_query(
+          Conn,
+          fun(Conf = #{pool := Worker}) ->
+            Query = mongoc:count_query(Conf, Collection, Selector, WrongLimit),
+            {_, Res} = mc_worker_api:command(Worker, Query),
+            Res
+          end)
+      end).
+
+wait_for_stabilization(#{attempts := Attempts, interval_ms := IntervalMS})
+  when Attempts > 0 ->
+    try
+        {ok, Conn} = ?POOL(?APP),
+        #{} = mongo_api:find_one(Conn, ?MONGO_CL_USER, #{}, #{}),
+        ok
+    catch
+        _:_ ->
+            ct:pal("mongodb connection still stabilizing... sleeping for ~b ms", [IntervalMS]),
+            ct:sleep(IntervalMS),
+            wait_for_stabilization(#{attempts => Attempts - 1, interval_ms => IntervalMS})
+    end;
+wait_for_stabilization(_) ->
+    error(mongo_connection_did_not_stabilize).
+
+%% TODO: move to ct helpers???
+reset_proxy(ProxyHost, ProxyPort) ->
+    Url = "http://" ++ ProxyHost ++ ":" ++ integer_to_list(ProxyPort) ++ "/reset",
+    Body = <<>>,
+    {ok, {{_, 204, _}, _, _}} = httpc:request(post, {Url, [], "application/json", Body}, [],
+                                              [{body_format, binary}]).
+
+with_failure(FailureType, ProxyHost, ProxyPort, Fun) ->
+    enable_failure(FailureType, ProxyHost, ProxyPort),
+    try
+        Fun()
+    after
+        heal_failure(FailureType, ProxyHost, ProxyPort)
+    end.
+
+enable_failure(FailureType, ProxyHost, ProxyPort) ->
+    case FailureType of
+        down -> switch_proxy(off, ProxyHost, ProxyPort);
+        timeout -> timeout_proxy(on, ProxyHost, ProxyPort);
+        latency_up -> latency_up_proxy(on, ProxyHost, ProxyPort)
+    end.
+
+heal_failure(FailureType, ProxyHost, ProxyPort) ->
+    case FailureType of
+        down -> switch_proxy(on, ProxyHost, ProxyPort);
+        timeout -> timeout_proxy(off, ProxyHost, ProxyPort);
+        latency_up -> latency_up_proxy(off, ProxyHost, ProxyPort)
+    end.
+
+switch_proxy(Switch, ProxyHost, ProxyPort) ->
+    Url = "http://" ++ ProxyHost ++ ":" ++ integer_to_list(ProxyPort) ++ "/proxies/mongo_single",
+    Body = case Switch of
+               off -> <<"{\"enabled\":false}">>;
+               on  -> <<"{\"enabled\":true}">>
+           end,
+    {ok, {{_, 200, _}, _, _}} = httpc:request(post, {Url, [], "application/json", Body}, [],
+                                              [{body_format, binary}]).
+
+timeout_proxy(on, ProxyHost, ProxyPort) ->
+    Url = "http://" ++ ProxyHost ++ ":" ++ integer_to_list(ProxyPort) ++ "/proxies/mongo_single/toxics",
+    Body = <<"{\"name\":\"timeout\",\"type\":\"timeout\","
+             "\"stream\":\"upstream\",\"toxicity\":1.0,"
+             "\"attributes\":{\"timeout\":0}}">>,
+    {ok, {{_, 200, _}, _, _}} = httpc:request(post, {Url, [], "application/json", Body}, [],
+                                              [{body_format, binary}]);
+timeout_proxy(off, ProxyHost, ProxyPort) ->
+    Url = "http://" ++ ProxyHost ++ ":" ++ integer_to_list(ProxyPort) ++ "/proxies/mongo_single/toxics/timeout",
+    Body = <<>>,
+    {ok, {{_, 204, _}, _, _}} = httpc:request(delete, {Url, [], "application/json", Body}, [],
+                                              [{body_format, binary}]).
+
+latency_up_proxy(on, ProxyHost, ProxyPort) ->
+    Url = "http://" ++ ProxyHost ++ ":" ++ integer_to_list(ProxyPort) ++ "/proxies/mongo_single/toxics",
+    Body = <<"{\"name\":\"latency_up\",\"type\":\"latency\","
+             "\"stream\":\"upstream\",\"toxicity\":1.0,"
+             "\"attributes\":{\"latency\":20000,\"jitter\":3000}}">>,
+    {ok, {{_, 200, _}, _, _}} = httpc:request(post, {Url, [], "application/json", Body}, [],
+                                              [{body_format, binary}]);
+latency_up_proxy(off, ProxyHost, ProxyPort) ->
+    Url = "http://" ++ ProxyHost ++ ":" ++ integer_to_list(ProxyPort) ++ "/proxies/mongo_single/toxics/latency_up",
+    Body = <<>>,
+    {ok, {{_, 204, _}, _, _}} = httpc:request(delete, {Url, [], "application/json", Body}, [],
+                                              [{body_format, binary}]).
--- a/apps/emqx_auth_pgsql/test/emqx_auth_pgsql_SUITE.erl
+++ b/apps/emqx_auth_pgsql/test/emqx_auth_pgsql_SUITE.erl
@ -70,6 +70,9 @@
 all() ->
    emqx_ct:all(?MODULE).

+suite() ->
+    [{timetrap, {seconds, 120}}].
+
 init_per_suite(Config) ->
    emqx_ct_helpers:start_apps([emqx_auth_pgsql]),
    drop_acl(),
--- a/apps/emqx_exhook/docs/design-cn.md
+++ b/apps/emqx_exhook/docs/design-cn.md
@ -47,7 +47,7 @@

 用户需要实现的方法，和数据类型的定义在 `priv/protos/exhook.proto` 文件中：

-```protobuff
+```protobuf
 syntax = "proto3";

 package emqx.exhook.v1;
--- a/apps/emqx_exproto/docs/design-cn.md
+++ b/apps/emqx_exproto/docs/design-cn.md
@ -44,7 +44,7 @@

 详情参见：`priv/protos/exproto.proto`，例如接口的定义有：

-```protobuff
+```protobuf
 syntax = "proto3";

 package emqx.exproto.v1;
--- a/apps/emqx_psk_file/include/emqx_psk_file.hrl
+++ b/apps/emqx_psk_file/include/emqx_psk_file.hrl
@ -0,0 +1,25 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% 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.
+%%--------------------------------------------------------------------
+
+-ifndef(EMQX_PSK_FILE).
+-define(EMQX_PSK_FILE, true).
+
+-define(PSK_FILE_TAB, emqx_psk_file).
+
+-record(psk_entry, {psk_id :: binary(),
+                    psk_str :: binary()}).
+
+-endif.
--- a/apps/emqx_psk_file/src/emqx_psk_file.app.src
+++ b/apps/emqx_psk_file/src/emqx_psk_file.app.src
@ -1,6 +1,6 @@
 {application, emqx_psk_file,
 [{description,"EMQX PSK Plugin from File"},
-  {vsn, "4.3.0"}, % strict semver, bump manually!
+  {vsn, "4.3.1"}, % strict semver, bump manually!
  {modules,[]},
  {registered,[emqx_psk_file_sup]},
  {applications,[kernel,stdlib]},
--- a/apps/emqx_psk_file/src/emqx_psk_file.appup.src
+++ b/apps/emqx_psk_file/src/emqx_psk_file.appup.src
@ -0,0 +1,10 @@
+%% -*- mode: erlang -*-
+{VSN,
+  [{"4.3.0",
+    [{load_module,emqx_psk_file,brutal_purge,soft_purge,[]},
+     {load_module,emqx_psk_file_sup,brutal_purge,soft_purge,[]}]}
+  ],
+  [{"4.3.0",
+    [{load_module,emqx_psk_file,brutal_purge,soft_purge,[]},
+     {load_module,emqx_psk_file_sup,brutal_purge,soft_purge,[]}]}
+  ]}.
--- a/apps/emqx_psk_file/src/emqx_psk_file.erl
+++ b/apps/emqx_psk_file/src/emqx_psk_file.erl
@ -16,6 +16,7 @@

 -module(emqx_psk_file).

+-include("emqx_psk_file.hrl").
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("emqx/include/logger.hrl").

@ -26,15 +27,10 @@
 %% Hooks functions
 -export([on_psk_lookup/2]).

-define(TAB, ?MODULE).
 -define(LF, 10).

-record(psk_entry, {psk_id :: binary(),
-                    psk_str :: binary()}).
-
 %% Called when the plugin application start
 load(Env) ->
-    _ = ets:new(?TAB, [set, named_table, {keypos, #psk_entry.psk_id}]),
    {ok, PskFile} = file:open(get_value(path, Env), [read, raw, binary, read_ahead]),
    preload_psks(PskFile, bin(get_value(delimiter, Env))),
    _ = file:close(PskFile),
@ -45,7 +41,7 @@ unload() ->
    emqx:unhook('tls_handshake.psk_lookup', fun ?MODULE:on_psk_lookup/2).

 on_psk_lookup(ClientPSKID, UserState) ->
-    case ets:lookup(?TAB, ClientPSKID) of
+    case ets:lookup(?PSK_FILE_TAB, ClientPSKID) of
        [#psk_entry{psk_str = PskStr}] ->
            {stop, PskStr};
        [] ->
@ -57,7 +53,9 @@ preload_psks(FileHandler, Delimiter) ->
        {ok, Line} ->
            case binary:split(Line, Delimiter) of
                [Key, Rem] ->
-                    ets:insert(?TAB, #psk_entry{psk_id = Key, psk_str = trim_lf(Rem)}),
+                    ets:insert(
+                      ?PSK_FILE_TAB,
+                      #psk_entry{psk_id = Key, psk_str = trim_lf(Rem)}),
                    preload_psks(FileHandler, Delimiter);
                [Line] ->
                    ?LOG(warning, "[~p] - Invalid line: ~p, delimiter: ~p", [?MODULE, Line, Delimiter])
--- a/apps/emqx_psk_file/src/emqx_psk_file_sup.erl
+++ b/apps/emqx_psk_file/src/emqx_psk_file_sup.erl
@ -16,6 +16,8 @@

 -module(emqx_psk_file_sup).

+-include("emqx_psk_file.hrl").
+
 -behaviour(supervisor).

 %% API
@ -25,8 +27,11 @@
 -export([init/1]).

 start_link() ->
+    _ = ets:new(
+          ?PSK_FILE_TAB,
+          [set, named_table, public, {keypos, #psk_entry.psk_id}]
+         ),
    supervisor:start_link({local, ?MODULE}, ?MODULE, []).

 init([]) ->
    {ok, { {one_for_one, 0, 1}, []} }.
-
--- a/apps/emqx_retainer/src/emqx_retainer.app.src
+++ b/apps/emqx_retainer/src/emqx_retainer.app.src
@ -1,6 +1,6 @@
 {application, emqx_retainer,
 [{description, "EMQ X Retainer"},
-  {vsn, "4.3.4"}, % strict semver, bump manually!
+  {vsn, "4.3.5"}, % strict semver, bump manually!
  {modules, []},
  {registered, [emqx_retainer_sup]},
  {applications, [kernel,stdlib]},
--- a/apps/emqx_retainer/src/emqx_retainer_sup.erl
+++ b/apps/emqx_retainer/src/emqx_retainer_sup.erl
@ -34,6 +34,23 @@ init([Env]) ->
             type     => worker,
             modules  => [emqx_retainer]} || not is_managed_by_modules()]}}.

+-ifdef(EMQX_ENTERPRISE).
+
+is_managed_by_modules() ->
+    try
+        case supervisor:get_childspec(emqx_modules_sup, emqx_retainer) of
+            {ok, _} -> true;
+            _ -> false
+        end
+    catch
+        exit : {noproc, _} ->
+            false
+    end.
+
+-else.
+
 is_managed_by_modules() ->
    %% always false for opensource edition
    false.
+
+-endif.
--- a/apps/emqx_retainer/test/emqx_retainer_ct_helper.erl
+++ b/apps/emqx_retainer/test/emqx_retainer_ct_helper.erl
@ -20,7 +20,9 @@
 -export([ensure_start/0, ensure_stop/0]).
 -ifdef(EMQX_ENTERPRISE).
 ensure_start() ->
+    %% for enterprise edition, retainer is started by modules
    application:stop(emqx_modules),
+    ensure_stop(),
    init_conf(),
    emqx_ct_helpers:start_apps([emqx_retainer]),
    ok.
@ -29,6 +31,7 @@ ensure_start() ->

 ensure_start() ->
    init_conf(),
+    ensure_stop(),
    emqx_ct_helpers:start_apps([emqx_retainer]),
    ok.

--- a/apps/emqx_rule_engine/include/rule_engine.hrl
+++ b/apps/emqx_rule_engine/include/rule_engine.hrl
@ -131,7 +131,7 @@

 -record(resource_params,
        { id :: resource_id()
-        , params :: #{} %% the params got after initializing the resource
+        , params :: map() %% the params got after initializing the resource
        , status = #{is_alive => false} :: #{is_alive := boolean(), atom() => term()}
        }).

--- a/bin/emqx
+++ b/bin/emqx
@ -481,6 +481,16 @@ case "$1" in
        ;;
 esac

+if [ "$IS_BOOT_COMMAND" = 'no' ]; then
+    # for non-boot commands, inspect vm.<time>.args for node name
+    # shellcheck disable=SC2012,SC2086
+    LATEST_VM_ARGS_FILE="$(ls -t "$RUNNER_DATA_DIR"/configs/vm.*.args 2>/dev/null | head -1)"
+    if [ -z "$LATEST_VM_ARGS_FILE" ]; then
+        echoerr "There is no vm.*.args config file found in '$RUNNER_DATA_DIR/configs/'"
+        echoerr "Please make sure the node is initialized (started for at least once)"
+        exit 1
+    fi
+fi

 if [ -z "$NAME_ARG" ]; then
    NODENAME="${EMQX_NODE_NAME:-}"
@ -488,15 +498,7 @@ if [ -z "$NAME_ARG" ]; then
    [ -z "$NODENAME" ] && [ -n "$EMQX_NAME" ] && [ -n "$EMQX_HOST" ] && NODENAME="${EMQX_NAME}@${EMQX_HOST}"
    if [ -z "$NODENAME" ]; then
        if [ "$IS_BOOT_COMMAND" = 'no' ]; then
-            # for non-boot commands, inspect vm.<time>.args for node name
-            # shellcheck disable=SC2012,SC2086
-            LATEST_VM_ARGS="$(ls -t $RUNNER_DATA_DIR/configs/vm.*.args 2>/dev/null | head -1)"
-            if [ -z "$LATEST_VM_ARGS" ]; then
-                echoerr "For command $1, there is no vm.*.args config file found in $RUNNER_DATA_DIR/configs/"
-                echoerr "Please make sure the node is initialized (started for at least once)"
-                exit 1
-            fi
-            NODENAME="$(grep -E '^-name' "$LATEST_VM_ARGS" | awk '{print $2}')"
+            NODENAME="$(grep -E '^-name' "$LATEST_VM_ARGS_FILE" | awk '{print $2}')"
        else
            # for boot commands, inspect emqx.conf for node name
            NODENAME=$("$ERTS_PATH"/escript "$RUNNER_ROOT_DIR"/bin/cuttlefish -i "$REL_DIR"/emqx.schema -c "$RUNNER_ETC_DIR"/emqx.conf get node.name)
@ -531,13 +533,7 @@ if [ -z "$COOKIE" ]; then
    if [ "$IS_BOOT_COMMAND" = 'yes' ]; then
        COOKIE=$("$ERTS_PATH"/escript "$RUNNER_ROOT_DIR"/bin/cuttlefish -i "$REL_DIR"/emqx.schema -c "$RUNNER_ETC_DIR"/emqx.conf get node.cookie)
    else
-        # shellcheck disable=SC2012,SC2086
-        LATEST_VM_ARGS="$(ls -t $RUNNER_DATA_DIR/configs/vm.*.args | head -1)"
-        if [ -z "$LATEST_VM_ARGS" ]; then
-            echo "For command $1, there is no vm.*.args config file found in $RUNNER_DATA_DIR/configs/"
-            exit 1
-        fi
-        COOKIE="$(grep -E '^-setcookie' "$LATEST_VM_ARGS" | awk '{print $2}')"
+        COOKIE="$(grep -E '^-setcookie' "$LATEST_VM_ARGS_FILE" | awk '{print $2}')"
    fi
 fi

--- a/rebar.config
+++ b/rebar.config
@ -42,7 +42,7 @@
    , {redbug, "2.0.7"}
    , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.2.0"}}}
    , {gun, {git, "https://github.com/emqx/gun", {tag, "1.3.7"}}}
-    , {eredis_cluster, {git, "https://github.com/emqx/eredis_cluster", {tag, "0.7.3"}}}
+    , {eredis_cluster, {git, "https://github.com/emqx/eredis_cluster", {tag, "0.7.4"}}}
    , {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}}
    , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}
    , {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.8.2"}}}
@ -59,7 +59,7 @@
    , {recon, {git, "https://github.com/ferd/recon", {tag, "2.5.1"}}}
    , {observer_cli, "1.6.1"} % NOTE: depends on recon 2.5.1
    , {getopt, "1.0.1"}
-    , {snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe.git", {tag, "0.15.0"}}}
+    , {snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe.git", {tag, "1.0.1"}}}
    , {lc, {git, "https://github.com/emqx/lc.git", {tag, "0.3.1"}}}
    , {mongodb, {git,"https://github.com/emqx/mongodb-erlang", {tag, "v3.0.13"}}}
    , {epgsql, {git, "https://github.com/emqx/epgsql.git", {tag, "4.6.0"}}}
--- a/scripts/rel/cut4x.sh
+++ b/scripts/rel/cut4x.sh
@ -40,10 +40,10 @@ EOF
 }

 logerr() {
-    echo -e "\e[31mERROR: $1\e[39m"
+    echo "$(tput setaf 1)ERROR: $1$(tput sgr0)"
 }
 logmsg() {
-    echo -e "\e[33mINFO: $1\e[39m"
+    echo "INFO: $1"
 }

 REL_BRANCH_CE="${REL_BRANCH_CE:-release-v43}"
--- a/scripts/rel/sync-remotes.sh
+++ b/scripts/rel/sync-remotes.sh
@ -45,11 +45,10 @@ EOF
 }

 logerr() {
-    echo -e "\e[31mERROR: $1\e[39m"
+    echo "$(tput setaf 1)ERROR: $1$(tput sgr0)"
 }
-
 logwarn() {
-    echo -e "\e[33mINFO: $1\e[39m"
+    echo "$(tput setaf 3)WARNING: $1$(tput sgr0)"
 }

 logmsg() {
@ -170,27 +169,29 @@ remote_refs() {
 upstream_branches() {
    local base="$1"
    case "$base" in
-        release-v43|main-v4.3)
-            ## no upstream for 4.3 opensource
+        release-v43)
            remote_ref "$base"
            ;;
        release-v44)
            remote_refs "$base" 'release-v43'
            ;;
-        main-v4.4)
-            remote_refs "$base" 'main-v4.3'
-            ;;
        release-e43)
            remote_refs "$base" 'release-v43'
            ;;
-        main-v4.3-enterprise)
-            remote_refs "$base" 'main-v4.3'
-            ;;
        release-e44)
            remote_refs "$base" 'release-v44' 'release-e43' 'release-v43'
            ;;
+        main-v4.3)
+            remote_refs "$base" 'release-v43'
+            ;;
+        main-v4.4)
+            remote_refs "$base" 'release-v44' 'main-v4.3'
+            ;;
+        main-v4.3-enterprise)
+            remote_refs "$base" 'release-e43' 'main-v4.3'
+            ;;
        main-v4.4-enterprise)
-            remote_refs "$base" 'main-v4.4' 'main-v4.3-enterprise' 'main-v4.3'
+            remote_refs "$base" 'release-e44' 'main-v4.4' 'main-v4.3-enterprise' 'main-v4.3'
            ;;
    esac
 }
--- a/scripts/update-appup.sh
+++ b/scripts/update-appup.sh
@ -7,12 +7,20 @@

 set -euo pipefail

-usage() {
-    echo "$0 PROFILE"
-}
 # ensure dir
 cd -P -- "$(dirname -- "${BASH_SOURCE[0]}")/.."

+usage() {
+    echo "$0 PROFILE [options]"
+    echo "options:"
+    echo "--skip-build:      Skip building the profile only to re-generate the appup files."
+    echo "--skip-build-base: This script by default forces a git clean before rebuilding on the base version "
+    echo "                   this option is useful when you are sure the past builds can be trusted,"
+    echo "                   that is, there were no re-tags or anything."
+    echo "--check:           Exit with non-zero code if there is git diff after the execution."
+    echo "                   Mostly used in CI."
+}
+
 PROFILE="${1:-}"
 case "$PROFILE" in
    emqx-ee)
@ -48,7 +56,7 @@ ESCRIPT_ARGS=( '' )
 while [ "$#" -gt 0 ]; do
    case $1 in
    -h|--help)
-        help
+        usage
        exit 0
        ;;
    --skip-build)
@ -100,7 +108,7 @@ else
    pushd "${PREV_DIR_BASE}/${PREV_TAG}"
    if [ "$NEW_COPY" = 'no' ]; then
        REMOTE="$(git remote -v | grep "${GIT_REPO}" | head -1 | awk '{print $1}')"
-        git fetch "$REMOTE"
+        git fetch "$REMOTE" --tags -f
    fi
    git reset --hard
    git clean -ffdx
--- a/scripts/update_appup.escript
+++ b/scripts/update_appup.escript
@ -62,7 +62,8 @@ app_specific_actions(_) ->
 ignored_apps() ->
    [gpb, %% only a build tool
     emqx_dashboard, %% generic appup file for all versions
-     emqx_management %% generic appup file for all versions
+     emqx_management, %% generic appup file for all versions
+     emqx_modules_spec %% generic appup file for all versions
    ] ++ otp_standard_apps().

 main(Args) ->
@ -284,9 +285,9 @@ merge_update_actions(App, Changes, Vsns, PrevVersion) ->
 %% but there is a 1.1.2 in appup we may skip merging instructions for
 %% 1.1.2 because it's not used and no way to know what has been changed
 is_skipped_version(App, Vsn, PrevVersion) when is_list(Vsn) andalso is_list(PrevVersion) ->
-    case is_app_external(App) andalso parse_version_number(Vsn) of
+    case is_app_external(App) andalso parse_version(Vsn, non_strict_semver) of
        {ok, VsnTuple} ->
-            case parse_version_number(PrevVersion) of
+            case parse_version(PrevVersion, non_strict_semver) of
                {ok, PrevVsnTuple} ->
                    VsnTuple > PrevVsnTuple;
                _ ->
@ -312,7 +313,7 @@ do_merge_update_actions(App, {New0, Changed0, Deleted0}, OldActions) ->
            true ->
                [];
            false ->
-                [{load_module, M, brutal_purge, soft_purge, []} || M <- Changed] ++
+                [{load_module, M, brutal_purge, soft_purge, []} || M <- Changed, not is_secret_module(M)] ++
                [{add_module, M} || M <- New]
        end,
    {OldActionsWithStop, OldActionsAfterStop} =
@ -324,10 +325,18 @@ do_merge_update_actions(App, {New0, Changed0, Deleted0}, OldActions) ->
            true ->
                [];
            false ->
-                [{delete_module, M} || M <- Deleted]
+                [{delete_module, M} || M <- Deleted, not is_secret_module(M)]
        end ++
        AppSpecific.

+%% Do not reload or delet _secret modules
+is_secret_module(Module) ->
+    Suffix = "_secret",
+    case string:right(atom_to_list(Module), length(Suffix)) of
+        Suffix -> true;
+        _ -> false
+    end.
+
 %% If an entry restarts an application, there's no need to use
 %% `load_module' instructions.
 contains_restart_application(Application, Actions) ->
@ -397,7 +406,7 @@ contains_version(Needle, Haystack) when is_list(Needle) ->
 %% past versions that should be covered by regexes in .appup file
 %% instructions.
 enumerate_past_versions(Vsn) when is_list(Vsn) ->
-    case parse_version_number(Vsn) of
+    case parse_version(Vsn) of
        {ok, ParsedVsn} ->
            {ok, enumerate_past_versions(ParsedVsn)};
        Error ->
@ -406,14 +415,39 @@ enumerate_past_versions(Vsn) when is_list(Vsn) ->
 enumerate_past_versions({Major, Minor, Patch}) ->
    [{Major, Minor, P} || P <- lists:seq(Patch - 1, 0, -1)].

-parse_version_number(Vsn) when is_list(Vsn) ->
-    Nums = string:split(Vsn, ".", all),
-    Results = lists:map(fun string:to_integer/1, Nums),
-    case Results of
-        [{Major, []}, {Minor, []}, {Patch, []}] ->
-            {ok, {Major, Minor, Patch}};
-        _ ->
-            {error, bad_version}
+parse_version(Vsn) ->
+    parse_version(Vsn, strict_semver).
+
+parse_version(Vsn, MaybeSemver) when is_list(Vsn) ->
+    case parse_dot_separated_numbers(Vsn) of
+        {ok, {_Major, _Minor, _Patch}} = Res ->
+            Res;
+        {ok, Nums} ->
+            case MaybeSemver of
+                strict_semver ->
+                    {error, {bad_semver, Vsn}};
+                non_strict_semver ->
+                    {ok, Nums}
+            end;
+        {error, Reason} ->
+            {error, {Reason, Vsn}}
+    end.
+
+parse_dot_separated_numbers(Str) when is_list(Str) ->
+    try
+        Split = string:split(Str, ".", all),
+        IntL = lists:map(fun(SubStr) ->
+                                 case string:to_integer(SubStr) of
+                                     {Int, []} when is_integer(Int) ->
+                                         Int;
+                                     _ ->
+                                         throw(no_integer)
+                                 end
+                         end, Split),
+        {ok, list_to_tuple(IntL)}
+    catch
+        _ : _ ->
+            {error, bad_version_string}
    end.

 vsn_number_to_string({Major, Minor, Patch}) ->
--- a/src/emqx.appup.src
+++ b/src/emqx.appup.src
@ -2,7 +2,8 @@
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
  [{"4.3.21",
-    [{load_module,emqx_alarm,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_alarm,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_connection,brutal_purge,soft_purge,[]},
     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
@ -11,9 +12,13 @@
     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
     {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
-     {load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]}]},
   {"4.3.20",
-    [{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
@ -24,9 +29,13 @@
     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
-     {load_module,emqx_cm,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]}]},
   {"4.3.19",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
@ -37,11 +46,14 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]},
     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
   {"4.3.18",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
@ -54,9 +66,12 @@
     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_plugins,brutal_purge,soft_purge,[]}]},
   {"4.3.17",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
@ -74,9 +89,11 @@
     {update,emqx_broker_sup,supervisor},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_access_control,brutal_purge,soft_purge,[]}]},
   {"4.3.16",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
@ -101,9 +118,11 @@
     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
     {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
     {load_module,emqx_mqtt_caps,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_topic,brutal_purge,soft_purge,[]}]},
   {"4.3.15",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
@ -135,9 +154,11 @@
     {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
     {load_module,emqx_alarm,brutal_purge,soft_purge,[]},
     {update,emqx_os_mon,{advanced,[]}},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]}]},
   {"4.3.14",
-    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
@ -171,9 +192,11 @@
     {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
     {load_module,emqx_alarm,brutal_purge,soft_purge,[]},
     {update,emqx_os_mon,{advanced,[]}},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_hooks,brutal_purge,soft_purge,[]}]},
   {"4.3.13",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -210,9 +233,11 @@
     {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
     {load_module,emqx_alarm,brutal_purge,soft_purge,[]},
     {update,emqx_os_mon,{advanced,[]}},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
   {"4.3.12",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -252,9 +277,11 @@
     {load_module,emqx,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.11",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -298,7 +325,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.10",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -342,7 +370,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.9",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -389,7 +418,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.8",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -436,7 +466,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.7",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -483,7 +514,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.6",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -530,7 +562,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.5",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -577,7 +610,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.4",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -624,7 +658,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.3",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -671,7 +706,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.2",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -718,7 +754,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.1",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -767,7 +804,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.0",
-    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
+    [{add_module,emqx_secret},
+     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
     {load_module,emqx_router,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
     {add_module,emqx_calendar},
@ -819,7 +857,8 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {<<".*">>,[]}],
-  [{"4.3.21",
+  [
+   {"4.3.21",
    [{load_module,emqx_alarm,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_connection,brutal_purge,soft_purge,[]},
@ -829,7 +868,10 @@
     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
     {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
     {load_module,emqx_tracer,brutal_purge,soft_purge,[]},
-     {load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]}]},
   {"4.3.20",
    [{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
@ -842,7 +884,10 @@
     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
-     {load_module,emqx_cm,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]}]},
   {"4.3.19",
    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
     {load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
@ -855,8 +900,10 @@
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]},
     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
   {"4.3.18",
    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
@ -872,6 +919,8 @@
     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_plugins,brutal_purge,soft_purge,[]}]},
   {"4.3.17",
    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
@ -892,6 +941,7 @@
     {update,emqx_broker_sup,supervisor},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_access_control,brutal_purge,soft_purge,[]}]},
   {"4.3.16",
    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
@ -919,6 +969,7 @@
     {load_module,emqx_mqtt_caps,brutal_purge,soft_purge,[]},
     {load_module,emqx_topic,brutal_purge,soft_purge,[]},
     {apply,{emqx_exclusive_subscription,on_delete_module,[]}},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {delete_module,emqx_exclusive_subscription}]},
   {"4.3.15",
    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
@ -952,6 +1003,7 @@
     {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
     {load_module,emqx_access_control,brutal_purge,soft_purge,[]},
     {load_module,emqx_os_mon,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]}]},
   {"4.3.14",
    [{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
@ -987,6 +1039,7 @@
     {load_module,emqx_frame,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_os_mon,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_hooks,brutal_purge,soft_purge,[]}]},
   {"4.3.13",
    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
@ -1025,6 +1078,7 @@
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_connection,brutal_purge,soft_purge,[]}]},
   {"4.3.12",
    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
@ -1065,6 +1119,7 @@
     {load_module,emqx,brutal_purge,soft_purge,[]},
     {load_module,emqx_app,brutal_purge,soft_purge,[]},
     {load_module,emqx_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_http_lib,brutal_purge,soft_purge,[]},
     {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
   {"4.3.11",
    [{load_module,emqx_router_helper,brutal_purge,soft_purge,[]},
--- a/src/emqx.erl
+++ b/src/emqx.erl
@ -228,6 +228,7 @@ shutdown() ->

 shutdown(Reason) ->
    ?LOG(critical, "emqx shutdown for ~s", [Reason]),
+    on_shutdown(Reason),
    _ = emqx_plugins:unload(),
    lists:foreach(fun application:stop/1
                 , lists:reverse(default_started_applications())
@ -238,10 +239,12 @@ reboot() ->
        true ->
            _ = application:stop(emqx_dashboard), %% dashboard must be started after mnesia
            lists:foreach(fun application:start/1 , default_started_applications()),
-            application:start(emqx_dashboard);
+            _ = application:start(emqx_dashboard),
+            on_reboot();

        false ->
-            lists:foreach(fun application:start/1 , default_started_applications())
+            lists:foreach(fun application:start/1 , default_started_applications()),
+            on_reboot()
    end.

 is_application_running(App) ->
@ -256,6 +259,32 @@ default_started_applications() ->
    [gproc, esockd, ranch, cowboy, ekka, emqx, emqx_modules].
 -endif.

+-ifdef(EMQX_ENTERPRISE).
+on_reboot() ->
+    try
+        _ = emqx_license_api:bootstrap_license(),
+        ok
+    catch
+        Kind:Reason:Stack ->
+            ?LOG(critical, "~p while rebooting: ~p, ~p", [Kind, Reason, Stack]),
+            ok
+    end,
+    ok.
+
+on_shutdown(join) ->
+    emqx_modules:sync_load_modules_file(),
+    ok;
+on_shutdown(_) ->
+    ok.
+
+-else.
+on_reboot() ->
+    ok.
+
+on_shutdown(_) ->
+    ok.
+-endif.
+
 %%--------------------------------------------------------------------
 %% Internal functions
 %%--------------------------------------------------------------------
--- a/src/emqx_channel.erl
+++ b/src/emqx_channel.erl
@ -951,9 +951,10 @@ return_sub_unsub_ack(Packet, Channel) ->
 handle_call(kick, Channel = #channel{
                               conn_state = ConnState,
                               will_msg = WillMsg,
+                               clientinfo = ClientInfo,
                               conninfo = #{proto_ver := ProtoVer}
                              }) ->
-    (WillMsg =/= undefined) andalso publish_will_msg(WillMsg),
+    (WillMsg =/= undefined) andalso publish_will_msg(ClientInfo, WillMsg),
    Channel1 = case ConnState of
                   connected -> ensure_disconnected(kicked, Channel);
                   _ -> Channel
@ -1102,8 +1103,9 @@ handle_timeout(_TRef, expire_awaiting_rel,
 handle_timeout(_TRef, expire_session, Channel) ->
    shutdown(expired, Channel);

-handle_timeout(_TRef, will_message, Channel = #channel{will_msg = WillMsg}) ->
-    (WillMsg =/= undefined) andalso publish_will_msg(WillMsg),
+handle_timeout(_TRef, will_message, Channel = #channel{will_msg = WillMsg,
+                                                       clientinfo = ClientInfo}) ->
+    (WillMsg =/= undefined) andalso publish_will_msg(ClientInfo, WillMsg),
    {ok, clean_timer(will_timer, Channel#channel{will_msg = undefined})};

 handle_timeout(_TRef, expire_quota_limit, Channel) ->
@ -1159,9 +1161,10 @@ terminate(_Reason, #channel{conn_state = idle} = _Channel) ->
    ok;
 terminate(normal, Channel) ->
    run_terminate_hook(normal, Channel);
-terminate(Reason, Channel = #channel{will_msg = WillMsg}) ->
+terminate(Reason, Channel = #channel{will_msg = WillMsg,
+                                     clientinfo = ClientInfo}) ->
    should_publish_will_message(Reason, Channel)
-        andalso publish_will_msg(WillMsg),
+        andalso publish_will_msg(ClientInfo, WillMsg),
    run_terminate_hook(Reason, Channel).

 run_terminate_hook(_Reason, #channel{session = undefined} = _Channel) ->
@ -1701,10 +1704,11 @@ ensure_disconnected(Reason, Channel = #channel{conninfo = ConnInfo,

 maybe_publish_will_msg(Channel = #channel{will_msg = undefined}) ->
    Channel;
-maybe_publish_will_msg(Channel = #channel{will_msg = WillMsg}) ->
+maybe_publish_will_msg(Channel = #channel{will_msg = WillMsg,
+                                          clientinfo = ClientInfo}) ->
    case will_delay_interval(WillMsg) of
        0 ->
-            ok = publish_will_msg(WillMsg),
+            ok = publish_will_msg(ClientInfo, WillMsg),
            Channel#channel{will_msg = undefined};
        I ->
            ensure_timer(will_timer, timer:seconds(I), Channel)
@ -1714,9 +1718,19 @@ will_delay_interval(WillMsg) ->
    maps:get('Will-Delay-Interval',
             emqx_message:get_header(properties, WillMsg, #{}), 0).

-publish_will_msg(Msg) ->
-    _ = emqx_broker:publish(Msg),
-    ok.
+publish_will_msg(ClientInfo, Msg = #message{topic = Topic}) ->
+    case emqx_access_control:check_acl(ClientInfo, publish, Topic) of
+        allow ->
+            _ = emqx_broker:publish(Msg),
+            ok;
+        deny ->
+            ?tp(
+                warning,
+                last_will_testament_publish_denied,
+                #{topic => Topic}
+            ),
+            ok
+    end.

 %%--------------------------------------------------------------------
 %% Disconnect Reason
--- a/src/emqx_http_lib.erl
+++ b/src/emqx_http_lib.erl
@ -96,8 +96,8 @@ do_parse(URI) ->
 %% underscores replaced with hyphens
 %% NOTE: assuming the input Headers list is a proplists,
 %% that is, when a key is duplicated, list header overrides tail
-%% e.g. [{"Content_Type", "applicaiton/binary"}, {<<"content-type">>, "applicaiton/json"}]
-%% results in: [{"content-type", "applicaiton/binary"}]
+%% e.g. [{"Content_Type", "applicaiton/binary"}, {"content-type", "applicaiton/json"}]
+%% results in: [{<<"content-type">>, "applicaiton/binary"}]
 normalise_headers(Headers0) ->
    F = fun({K0, V}) ->
                K = re:replace(K0, "_", "-", [{return,binary}]),
--- a/src/emqx_misc.erl
+++ b/src/emqx_misc.erl
@ -45,6 +45,9 @@
        , index_of/2
        , maybe_parse_ip/1
        , ipv6_probe/1
+        , ipv6_probe/2
+        , pmap/2
+        , pmap/3
        ]).

 -export([ bin2hexstr_A_F/1
@ -55,7 +58,13 @@
 -export([ is_sane_id/1
        ]).

+-export([
+    nolink_apply/1,
+    nolink_apply/2
+]).
+
 -define(VALID_STR_RE, "^[A-Za-z0-9]+[A-Za-z0-9-_]*$").
+-define(DEFAULT_PMAP_TIMEOUT, 5000).

 -spec is_sane_id(list() | binary()) -> ok | {error, Reason::binary()}.
 is_sane_id(Str) ->
@ -84,12 +93,15 @@ maybe_parse_ip(Host) ->

 %% @doc Add `ipv6_probe' socket option if it's supported.
 ipv6_probe(Opts) ->
+    ipv6_probe(Opts, true).
+
+ipv6_probe(Opts, Ipv6Probe) when is_boolean(Ipv6Probe) orelse is_integer(Ipv6Probe) ->
    Bool = try gen_tcp:ipv6_probe()
           catch _ : _ -> false end,
-    ipv6_probe(Bool, Opts).
+    ipv6_probe(Bool, Opts, Ipv6Probe).

-ipv6_probe(false, Opts) -> Opts;
-ipv6_probe(true, Opts) -> [{ipv6_probe, true} | Opts].
+ipv6_probe(false, Opts, _) -> Opts;
+ipv6_probe(true, Opts, Ipv6Probe) -> [{ipv6_probe, Ipv6Probe} | Opts].

 %% @doc Merge options
 -spec(merge_opts(Opts, Opts) -> Opts when Opts :: proplists:proplist()).
@ -328,6 +340,110 @@ hexchar2int(I) when I >= $0 andalso I =< $9 -> I - $0;
 hexchar2int(I) when I >= $A andalso I =< $F -> I - $A + 10;
 hexchar2int(I) when I >= $a andalso I =< $f -> I - $a + 10.

+%% @doc Like lists:map/2, only the callback function is evaluated
+%% concurrently.
+-spec pmap(fun((A) -> B), list(A)) -> list(B).
+pmap(Fun, List) when is_function(Fun, 1), is_list(List) ->
+    pmap(Fun, List, ?DEFAULT_PMAP_TIMEOUT).
+
+-spec pmap(fun((A) -> B), list(A), timeout()) -> list(B).
+pmap(Fun, List, Timeout) when
+    is_function(Fun, 1), is_list(List), is_integer(Timeout), Timeout >= 0
+->
+    nolink_apply(fun() -> do_parallel_map(Fun, List) end, Timeout).
+
+%% @doc Delegate a function to a worker process.
+%% The function may spawn_link other processes but we do not
+%% want the caller process to be linked.
+%% This is done by isolating the possible link with a not-linked
+%% middleman process.
+nolink_apply(Fun) -> nolink_apply(Fun, infinity).
+
+%% @doc Same as `nolink_apply/1', with a timeout.
+-spec nolink_apply(function(), timer:timeout()) -> term().
+nolink_apply(Fun, Timeout) when is_function(Fun, 0) ->
+    Caller = self(),
+    ResRef = make_ref(),
+    Middleman = erlang:spawn(make_middleman_fn(Caller, Fun, ResRef)),
+    receive
+        {ResRef, {normal, Result}} ->
+            Result;
+        {ResRef, {exception, {C, E, S}}} ->
+            erlang:raise(C, E, S);
+        {ResRef, {'EXIT', Reason}} ->
+            exit(Reason)
+    after Timeout ->
+        exit(Middleman, kill),
+        exit(timeout)
+    end.
+
+-spec make_middleman_fn(pid(), fun(() -> any()), reference()) -> fun(() -> no_return()).
+make_middleman_fn(Caller, Fun, ResRef) ->
+    fun() ->
+        process_flag(trap_exit, true),
+        CallerMRef = erlang:monitor(process, Caller),
+        Worker = erlang:spawn_link(make_worker_fn(Caller, Fun, ResRef)),
+        receive
+            {'DOWN', CallerMRef, process, _, _} ->
+                %% For whatever reason, if the caller is dead,
+                %% there is no reason to continue
+                exit(Worker, kill),
+                exit(normal);
+            {'EXIT', Worker, normal} ->
+                exit(normal);
+            {'EXIT', Worker, Reason} ->
+                %% worker exited with some reason other than 'normal'
+                _ = erlang:send(Caller, {ResRef, {'EXIT', Reason}}),
+                exit(normal)
+        end
+    end.
+
+-spec make_worker_fn(pid(), fun(() -> any()), reference()) -> fun(() -> no_return()).
+make_worker_fn(Caller, Fun, ResRef) ->
+    fun() ->
+        Res =
+            try
+                {normal, Fun()}
+            catch
+                C:E:S ->
+                    {exception, {C, E, S}}
+            end,
+        _ = erlang:send(Caller, {ResRef, Res}),
+        exit(normal)
+    end.
+
+do_parallel_map(Fun, List) ->
+    Parent = self(),
+    PidList = lists:map(
+        fun(Item) ->
+            erlang:spawn_link(
+                fun() ->
+                    Res =
+                        try
+                            {normal, Fun(Item)}
+                        catch
+                            C:E:St ->
+                                {exception, {C, E, St}}
+                        end,
+                    Parent ! {self(), Res}
+                end
+            )
+        end,
+        List
+    ),
+    lists:foldr(
+        fun(Pid, Acc) ->
+            receive
+                {Pid, {normal, Result}} ->
+                    [Result | Acc];
+                {Pid, {exception, {C, E, St}}} ->
+                    erlang:raise(C, E, St)
+            end
+        end,
+        [],
+        PidList
+    ).
+
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").

--- a/src/emqx_plugins.erl
+++ b/src/emqx_plugins.erl
@ -24,6 +24,7 @@
 -export([init/0]).

 -export([ load/0
+        , force_load/0
        , load/1
        , unload/0
        , unload/1
@ -59,12 +60,17 @@ init() ->
 %% @doc Load all plugins when the broker started.
 -spec(load() -> ok | ignore | {error, term()}).
 load() ->
+    do_load(#{force_load => false}).
+force_load() ->
+    do_load(#{force_load => true}).
+
+do_load(Options) ->
    ok = load_ext_plugins(emqx:get_env(expand_plugins_dir)),
    case emqx:get_env(plugins_loaded_file) of
        undefined -> ignore; %% No plugins available
        File ->
            _ = ensure_file(File),
-            with_loaded_file(File, fun(Names) -> load_plugins(Names, false) end)
+            with_loaded_file(File, fun(Names) -> load_plugins(Names, Options, false) end)
    end.

 %% @doc Load a Plugin
@ -282,18 +288,23 @@ filter_plugins([{Name, Load} | Names], Plugins) ->
 filter_plugins([Name | Names], Plugins) when is_atom(Name) ->
    filter_plugins([{Name, true} | Names], Plugins).

-load_plugins(Names, Persistent) ->
+load_plugins(Names, Options, Persistent) ->
    Plugins = list(),
    NotFound = Names -- names(Plugins),
    case NotFound of
        []       -> ok;
        NotFound -> ?LOG(alert, "cannot_find_plugins: ~p", [NotFound])
    end,
-    NeedToLoad = (Names -- NotFound) -- names(started_app),
+    NeedToLoad0 = Names -- NotFound,
+    NeedToLoad1 =
+        case Options of
+            #{force_load := true} -> NeedToLoad0;
+            _ -> NeedToLoad0 -- names(started_app)
+        end,
    lists:foreach(fun(Name) ->
                      Plugin = find_plugin(Name, Plugins),
                      load_plugin(Plugin#plugin.name, Persistent)
-                  end, NeedToLoad).
+                  end, NeedToLoad1).

 generate_configs(App) ->
    ConfigFile = filename:join([emqx:get_env(plugins_etc_dir), App]) ++ ".config",
--- a/src/emqx_secret.erl
+++ b/src/emqx_secret.erl
@ -0,0 +1,41 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% 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.
+%%--------------------------------------------------------------------
+
+%% Note: this module CAN'T be hot-patched to avoid invalidating the
+%% closures, so it must not be changed.
+-module(emqx_secret).
+
+%% API:
+-export([wrap/1, unwrap/1]).
+
+%%================================================================================
+%% API funcions
+%%================================================================================
+
+wrap(undefined) ->
+    undefined;
+wrap(Func) when is_function(Func) ->
+    Func;
+wrap(Term) ->
+    fun() ->
+        Term
+    end.
+
+unwrap(Term) when is_function(Term, 0) ->
+    %% Handle potentially nested funs
+    unwrap(Term());
+unwrap(Term) ->
+    Term.
--- a/test/emqx_misc_SUITE.erl
+++ b/test/emqx_misc_SUITE.erl
@ -146,3 +146,36 @@ t_now_to_secs(_) ->
 t_now_to_ms(_) ->
    ?assert(is_integer(emqx_misc:now_to_ms(os:timestamp()))).

+t_pmap_normal(_) ->
+    ?assertEqual(
+        [5, 7, 9],
+        emqx_misc:pmap(
+            fun({A, B}) -> A + B end,
+            [{2, 3}, {3, 4}, {4, 5}]
+        )
+    ).
+
+t_pmap_timeout(_) ->
+    ?assertExit(
+        timeout,
+        emqx_misc:pmap(
+            fun
+                (timeout) -> ct:sleep(1000);
+                ({A, B}) -> A + B
+            end,
+            [{2, 3}, {3, 4}, timeout],
+            100
+        )
+    ).
+
+t_pmap_exception(_) ->
+    ?assertError(
+        foobar,
+        emqx_misc:pmap(
+            fun
+                (error) -> error(foobar);
+                ({A, B}) -> A + B
+            end,
+            [{2, 3}, {3, 4}, error]
+        )
+    ).
--- a/test/emqx_node_helpers.erl
+++ b/test/emqx_node_helpers.erl
@ -18,33 +18,59 @@

 -include_lib("eunit/include/eunit.hrl").

-define(SLAVE_START_APPS, [emqx]).
+%% modules is included because code is called before cluster join
+-define(SLAVE_START_APPS, [emqx, emqx_modules]).

 -export([start_slave/1,
         start_slave/2,
-         stop_slave/1]).
+         stop_slave/1,
+         wait_for_synced_routes/3
+        ]).

 start_slave(Name) ->
    start_slave(Name, #{}).

 start_slave(Name, Opts) ->
-    {ok, Node} = ct_slave:start(list_to_atom(atom_to_list(Name) ++ "@" ++ host()),
-                                [{kill_if_fail, true},
-                                 {monitor_master, true},
-                                 {init_timeout, 10000},
-                                 {startup_timeout, 10000},
-                                 {erl_flags, ebin_path()}]),
-
+    Node = make_node_name(Name),
+    case ct_slave:start(Node, [{kill_if_fail, true},
+                               {monitor_master, true},
+                               {init_timeout, 10000},
+                               {startup_timeout, 10000},
+                               {erl_flags, ebin_path()}]) of
+        {ok, _} ->
+            ok;
+        {error, started_not_connected, _} ->
+            ok;
+        Other ->
+            throw(Other)
+    end,
    pong = net_adm:ping(Node),
    setup_node(Node, Opts),
    Node.

-stop_slave(Node) ->
-    rpc:call(Node, ekka, leave, []),
-    ct_slave:stop(Node).
+make_node_name(Name) ->
+    case string:tokens(atom_to_list(Name), "@") of
+        [_Name, _Host] ->
+            %% the name already has a @
+            Name;
+        _ ->
+            list_to_atom(atom_to_list(Name) ++ "@" ++ host())
+    end.
+
+stop_slave(Node0) ->
+    Node = make_node_name(Node0),
+    case rpc:call(Node, ekka, leave, []) of
+        ok -> ok;
+        {badrpc, nodedown} -> ok
+    end,
+    case ct_slave:stop(Node) of
+        {ok, _} -> ok;
+        {error, not_started, _} -> ok
+    end.

 host() ->
-    [_, Host] = string:tokens(atom_to_list(node()), "@"), Host.
+    [_, Host] = string:tokens(atom_to_list(node()), "@"),
+    Host.

 ebin_path() ->
    string:join(["-pa" | lists:filter(fun is_lib/1, code:get_path())], " ").
@ -68,10 +94,19 @@ setup_node(Node, #{} = Opts) ->
        end,
    EnvHandler = maps:get(env_handler, Opts, DefaultEnvHandler),

-    [ok = rpc:call(Node, application, load, [App]) || App <- [gen_rpc, emqx]],
+    %% apps need to be loaded before starting for ekka to find and create mnesia tables
+    LoadApps = lists:usort([gen_rcp, emqx] ++ ?SLAVE_START_APPS),
+    lists:foreach(fun(App) ->
+                          rpc:call(Node, application, load, [App])
+                  end, LoadApps),
    ok = rpc:call(Node, emqx_ct_helpers, start_apps, [StartApps, EnvHandler]),

-    rpc:call(Node, ekka, join, [node()]),
+    case maps:get(no_join, Opts, false) of
+        true ->
+            ok;
+        false ->
+            ok = rpc:call(Node, ekka, join, [node()])
+    end,

    %% Sanity check. Assert that `gen_rpc' is set up correctly:
    ?assertEqual( Node
@ -81,3 +116,40 @@ setup_node(Node, #{} = Opts) ->
                , gen_rpc:call(Node, gen_rpc, call, [node(), erlang, node, []])
                ),
    ok.
+
+%% Routes are replicated async.
+%% Call this function to wait for nodes in the cluster to have the same view
+%% for a given topic.
+wait_for_synced_routes(Nodes, Topic, Timeout) ->
+    F = fun() -> do_wait_for_synced_routes(Nodes, Topic) end,
+    emqx_misc:nolink_apply(F, Timeout).
+
+do_wait_for_synced_routes(Nodes, Topic) ->
+    PerNodeView0 =
+        lists:map(
+          fun(Node) ->
+                  {rpc:call(Node, emqx_router, match_routes, [Topic]), Node}
+          end, Nodes),
+    PerNodeView = lists:keysort(1, PerNodeView0),
+    case check_consistent_view(PerNodeView) of
+        {ok, OneView} ->
+            ct:pal("consistent_routes_view~n~p", [OneView]),
+            ok;
+        {error, Reason}->
+            ct:pal("inconsistent_routes_view~n~p", [Reason]),
+            timer:sleep(10),
+            do_wait_for_synced_routes(Nodes, Topic)
+    end.
+
+check_consistent_view(PerNodeView) ->
+    check_consistent_view(PerNodeView, []).
+
+check_consistent_view([], [OneView]) -> {ok, OneView};
+check_consistent_view([], MoreThanOneView) -> {error, MoreThanOneView};
+check_consistent_view([{View, Node} | Rest], [{View, Nodes} | Acc]) ->
+    check_consistent_view(Rest, [{View, add_to_list(Node, Nodes)} | Acc]);
+check_consistent_view([{View, Node} | Rest], Acc) ->
+    check_consistent_view(Rest, [{View, Node} | Acc]).
+
+add_to_list(Node, Nodes) when is_list(Nodes) -> [Node | Nodes];
+add_to_list(Node, Node1) -> [Node, Node1].
--- a/test/emqx_shared_sub_SUITE.erl
+++ b/test/emqx_shared_sub_SUITE.erl
@ -40,6 +40,9 @@ init_per_suite(Config) ->
    PortDiscovery = application:get_env(gen_rpc, port_discovery),
    application:set_env(gen_rpc, port_discovery, stateless),
    application:ensure_all_started(gen_rpc),
+    %% ensure emqx_moduels' app modules are loaded
+    %% so the mnesia tables are created
+    ok = load_app(emqx_modules),
    emqx_ct_helpers:start_apps([]),
    [{port_discovery, PortDiscovery} | Config].

@ -50,32 +53,45 @@ end_per_suite(Config) ->
      _ -> ok
    end.

-t_is_ack_required(_) ->
+init_per_testcase(Case, Config) ->
+    try
+        ?MODULE:Case({'init', Config})
+    catch
+        error : function_clause ->
+            Config
+    end.
+
+end_per_testcase(Case, Config) ->
+    try
+        ?MODULE:Case({'end', Config})
+    catch
+        error : function_clause ->
+            ok
+    end.
+
+t_is_ack_required(Config) when is_list(Config) ->
    ?assertEqual(false, emqx_shared_sub:is_ack_required(#message{headers = #{}})).

-t_maybe_nack_dropped(_) ->
+t_maybe_nack_dropped(Config) when is_list(Config) ->
    ?assertEqual(store, emqx_shared_sub:maybe_nack_dropped(#message{headers = #{}})),
    Msg = #message{headers = #{shared_dispatch_ack => {self(), {fresh, <<"group">>, for_test}}}},
    ?assertEqual(drop, emqx_shared_sub:maybe_nack_dropped(Msg)),
    ?assertEqual(ok,receive {for_test, {shared_sub_nack, dropped}} -> ok after 100 -> timeout end).

-t_nack_no_connection(_) ->
+t_nack_no_connection(Config) when is_list(Config) ->
    Msg = #message{headers = #{shared_dispatch_ack => {self(), {fresh, <<"group">>, for_test}}}},
    ?assertEqual(ok, emqx_shared_sub:nack_no_connection(Msg)),
    ?assertEqual(ok,receive {for_test, {shared_sub_nack, no_connection}} -> ok
                    after 100 -> timeout end).

-t_maybe_ack(_) ->
+t_maybe_ack(Config) when is_list(Config) ->
    ?assertEqual(#message{headers = #{}}, emqx_shared_sub:maybe_ack(#message{headers = #{}})),
    Msg = #message{headers = #{shared_dispatch_ack => {self(), {fresh, <<"group">>, for_test}}}},
    ?assertEqual(#message{headers = #{shared_dispatch_ack => ?no_ack}},
                 emqx_shared_sub:maybe_ack(Msg)),
    ?assertEqual(ok,receive {for_test, ?ack} -> ok after 100 -> timeout end).

-% t_subscribers(_) ->
-%     error('TODO').
-
-t_random_basic(_) ->
+t_random_basic(Config) when is_list(Config) ->
    ok = ensure_config(random),
    ClientId = <<"ClientId">>,
    Topic = <<"foo">>,
@ -105,7 +121,7 @@ t_random_basic(_) ->
 %% After the connection for the 2nd session is also closed,
 %% i.e. when all clients are offline, the following message(s)
 %% should be delivered randomly.
-t_no_connection_nack(_) ->
+t_no_connection_nack(Config) when is_list(Config) ->
    ok = ensure_config(sticky),
    Publisher = <<"publisher">>,
    Subscriber1 = <<"Subscriber1">>,
@ -171,27 +187,27 @@ t_no_connection_nack(_) ->
    %   emqx_sm:close_session(SPid2),
    ok.

-t_random(_) ->
+t_random(Config) when is_list(Config) ->
    ok = ensure_config(random, true),
    test_two_messages(random).

-t_round_robin(_) ->
+t_round_robin(Config) when is_list(Config) ->
    ok = ensure_config(round_robin, true),
    test_two_messages(round_robin).

-t_sticky(_) ->
+t_sticky(Config) when is_list(Config) ->
    ok = ensure_config(sticky, true),
    test_two_messages(sticky).

-t_hash(_) ->
+t_hash(Config) when is_list(Config) ->
    ok = ensure_config(hash, false),
    test_two_messages(hash).

-t_hash_clinetid(_) ->
+t_hash_clinetid(Config) when is_list(Config) ->
    ok = ensure_config(hash_clientid, false),
    test_two_messages(hash_clientid).

-t_hash_topic(_) ->
+t_hash_topic(Config) when is_list(Config) ->
    ok = ensure_config(hash_topic, false),
    ClientId1 = <<"ClientId1">>,
    ClientId2 = <<"ClientId2">>,
@ -230,7 +246,7 @@ t_hash_topic(_) ->
    ok.

 %% if the original subscriber dies, change to another one alive
-t_not_so_sticky(_) ->
+t_not_so_sticky(Config) when is_list(Config) ->
    ok = ensure_config(sticky),
    ClientId1 = <<"ClientId1">>,
    ClientId2 = <<"ClientId2">>,
@ -303,7 +319,7 @@ last_message(ExpectedPayload, Pids, Timeout) ->
        <<"not yet?">>
    end.

-t_dispatch(_) ->
+t_dispatch(Config) when is_list(Config) ->
    ok = ensure_config(random),
    Topic = <<"foo">>,
    ?assertEqual({error, no_subscribers},
@ -312,18 +328,13 @@ t_dispatch(_) ->
    ?assertEqual({ok, 1},
                 emqx_shared_sub:dispatch(<<"group1">>, Topic, #delivery{message = #message{}})).

-% t_unsubscribe(_) ->
-%     error('TODO').
-
-% t_subscribe(_) ->
-%     error('TODO').
-t_uncovered_func(_) ->
+t_uncovered_func(Config) when is_list(Config) ->
    ignored = gen_server:call(emqx_shared_sub, ignored),
    ok = gen_server:cast(emqx_shared_sub, ignored),
    ignored = emqx_shared_sub ! ignored,
    {mnesia_table_event, []} = emqx_shared_sub ! {mnesia_table_event, []}.

-t_per_group_config(_) ->
+t_per_group_config(Config) when is_list(Config) ->
    ok = ensure_group_config(#{
                               <<"local_group_fallback">> => local,
                               <<"local_group">> => local,
@ -342,8 +353,8 @@ t_per_group_config(_) ->
    test_two_messages(round_robin, <<"round_robin_group">>),
    test_two_messages(round_robin, <<"round_robin_group">>).

-t_local(_) ->
-    Node = start_slave('local_shared_sub_test19', 21884),
+t_local({'init', Config}) ->
+    Node = start_slave(local_shared_sub_test19, 21884),
    GroupConfig = #{
                    <<"local_group_fallback">> => local,
                    <<"local_group">> => local,
@ -352,7 +363,11 @@ t_local(_) ->
                   },
    ok = ensure_group_config(Node, GroupConfig),
    ok = ensure_group_config(GroupConfig),
-
+    [{slave_node, Node} | Config];
+t_local({'end', _Config}) ->
+    ok = stop_slave(local_shared_sub_test19);
+t_local(Config) when is_list(Config) ->
+    Node = proplists:get_value(slave_node, Config),
    Topic = <<"local_foo1/bar">>,
    ClientId1 = <<"ClientId1">>,
    ClientId2 = <<"ClientId2">>,
@ -388,7 +403,7 @@ t_local(_) ->
    ?assertNotEqual(UsedSubPid1, UsedSubPid2),
    ok.

-t_local_fallback(_) ->
+t_local_fallback({'init', Config}) ->
    ok = ensure_group_config(#{
                               <<"local_group_fallback">> => local,
                               <<"local_group">> => local,
@ -396,10 +411,15 @@ t_local_fallback(_) ->
                               <<"sticky_group">> => sticky
                              }),

+    Node = start_slave(local_fallback_shared_sub_test19, 11885),
+    [{slave_node, Node} | Config];
+t_local_fallback({'end', _}) ->
+    ok = stop_slave(local_fallback_shared_sub_test19);
+t_local_fallback(Config) when is_list(Config) ->
    Topic = <<"local_foo2/bar">>,
    ClientId1 = <<"ClientId1">>,
    ClientId2 = <<"ClientId2">>,
-    Node = start_slave('local_fallback_shared_sub_test19', 11885),
+    Node = proplists:get_value(slave_node, Config),

    {ok, ConnPid1} = emqtt:start_link([{clientid, ClientId1}]),
    {ok, _} = emqtt:connect(ConnPid1),
@ -407,6 +427,7 @@ t_local_fallback(_) ->
    Message2 = emqx_message:make(ClientId2, 0, Topic, <<"hello2">>),

    emqtt:subscribe(ConnPid1, {<<"$share/local_group_fallback/", Topic/binary>>, 0}),
+    ok = emqx_node_helpers:wait_for_synced_routes([node(), Node], Topic, timer:seconds(10)),

    [{share, Topic, {ok, 1}}] = emqx:publish(Message1),
    {true, UsedSubPid1} = last_message(<<"hello1">>, [ConnPid1]),
@ -422,7 +443,7 @@ t_local_fallback(_) ->

 %% This one tests that broker tries to select another shared subscriber
 %% If the first one doesn't return an ACK
-t_redispatch(_) ->
+t_redispatch(Config) when is_list(Config) ->
    ok = ensure_config(sticky, true),
    application:set_env(emqx, shared_dispatch_ack_enabled, true),

@ -453,7 +474,7 @@ t_redispatch(_) ->
    emqtt:stop(UsedSubPid2),
    ok.

-t_dispatch_when_inflights_are_full(_) ->
+t_dispatch_when_inflights_are_full(Config) when is_list(Config) ->
    ok = ensure_config(round_robin, true),
    Topic = <<"foo/bar">>,
    ClientId1 = <<"ClientId1">>,
@ -536,8 +557,20 @@ recv_msgs(Count, Msgs) ->
    end.

 start_slave(Name, Port) ->
+    ok = emqx_ct_helpers:start_apps([emqx_modules]),
    Listeners = [#{listen_on => {{127,0,0,1}, Port},
+                   start_apps => [emqx, emqx_modules],
                   name => "internal",
                   opts => [{zone,internal}],
                   proto => tcp}],
    emqx_node_helpers:start_slave(Name, #{listeners => Listeners}).
+
+stop_slave(Name) ->
+    emqx_node_helpers:stop_slave(Name).
+
+load_app(App) ->
+    case application:load(App) of
+        ok -> ok;
+        {error, {already_loaded, _}} -> ok;
+        {error, Reason} -> error({failed_to_load_app, App, Reason})
+    end.