Merge branch 'master' into fix-ssl-existingName-invalid

.ci/docker-compose-file/docker-compose.yaml

@@ -20,6 +20,7 @@ services:
       - ../..:/emqx
     working_dir: /emqx
     tty: true
+    user: "${UID_GID}"
 
 networks:
   emqx_bridge:

.github/workflows/build_and_push_docker_images.yaml

@@ -20,7 +20,7 @@ jobs:
   prepare:
     runs-on: ubuntu-20.04
     # prepare source with any OTP version, no need for a matrix
-    container: "ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04"
+    container: "ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04"
 
     outputs:
       BUILD_PROFILE: ${{ steps.get_profile.outputs.BUILD_PROFILE }}
@@ -112,7 +112,7 @@ jobs:
         # NOTE: 'otp' and 'elixir' are to configure emqx-builder image
         #       only support latest otp and elixir, not a matrix
         otp:
-          - 24.2.1-1 # update to latest
+          - 24.3.4.2-1 # update to latest
         elixir:
           - 1.13.4 # update to latest
 
@@ -164,7 +164,7 @@ jobs:
         tags: ${{ steps.meta.outputs.tags }}
         labels: ${{ steps.meta.outputs.labels }}
         build-args: |
-          BUILD_FROM=ghcr.io/emqx/emqx-builder/5.0-17:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os[0] }}
+          BUILD_FROM=ghcr.io/emqx/emqx-builder/5.0-18:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os[0] }}
           RUN_FROM=${{ matrix.os[1] }}
           EMQX_NAME=${{ steps.meta.outputs.emqx_name }}
         file: source/${{ matrix.os[2] }}
@@ -189,7 +189,7 @@ jobs:
         os:
           - [debian11, "debian:11-slim", "deploy/docker/Dockerfile"]
         otp:
-          - 24.2.1-1 # update to latest
+          - 24.3.4.2-1 # update to latest
         elixir:
           - 1.13.4 # update to latest
 
@@ -232,7 +232,7 @@ jobs:
         tags: ${{ steps.meta.outputs.tags }}
         labels: ${{ steps.meta.outputs.labels }}
         build-args: |
-          BUILD_FROM=ghcr.io/emqx/emqx-builder/5.0-17:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os[0] }}
+          BUILD_FROM=ghcr.io/emqx/emqx-builder/5.0-18:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os[0] }}
           RUN_FROM=${{ matrix.os[1] }}
           EMQX_NAME=${{ steps.meta.outputs.emqx_name }}
         file: source/${{ matrix.os[2] }}
@@ -257,7 +257,7 @@ jobs:
           - [debian11, "debian:11-slim", "deploy/docker/Dockerfile"]
         # NOTE: only support latest otp version, not a matrix
         otp:
-          - 24.2.1-1 # update to latest
+          - 24.3.4.2-1 # update to latest
         registry:
           - 'docker.io'
           - 'public.ecr.aws'
@@ -319,7 +319,7 @@ jobs:
           - ${{ needs.prepare.outputs.BUILD_PROFILE }}
         # NOTE: for docker, only support latest otp version, not a matrix
         otp:
-          - 24.2.1-1 # update to latest
+          - 24.3.4.2-1 # update to latest
         elixir:
           - 1.13.4 # update to latest
         registry:

.github/workflows/build_packages.yaml

@@ -23,7 +23,7 @@ on:
 jobs:
   prepare:
     runs-on: ubuntu-20.04
-    container: ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04
+    container: ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04
     outputs:
       BUILD_PROFILE: ${{ steps.get_profile.outputs.BUILD_PROFILE }}
       IS_EXACT_TAG: ${{ steps.get_profile.outputs.IS_EXACT_TAG }}
@@ -130,7 +130,7 @@ jobs:
     - uses: actions/upload-artifact@v3
       with:
         name: ${{ matrix.profile }}-windows
-        path: source/_packages/${{ matrix.profile }}/.
+        path: source/_packages/${{ matrix.profile }}/
 
   mac:
     needs: prepare
@@ -140,7 +140,7 @@ jobs:
         profile:
           - ${{ needs.prepare.outputs.BUILD_PROFILE }}
         otp:
-          - 24.2.1-1
+          - 24.3.4.2-1
         os:
           - macos-11
     runs-on: ${{ matrix.os }}
@@ -169,13 +169,13 @@ jobs:
     - uses: actions/upload-artifact@v3
       with:
         name: ${{ matrix.profile }}-${{ matrix.otp }}
-        path: _packages/${{ matrix.profile }}/.
+        path: _packages/${{ matrix.profile }}/
 
   linux:
     needs: prepare
     runs-on: ${{ matrix.build_machine }}
     container:
-      image: "ghcr.io/emqx/emqx-builder/5.0-17:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os }}"
+      image: "ghcr.io/emqx/emqx-builder/5.0-18:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os }}"
 
     strategy:
       fail-fast: false
@@ -183,7 +183,7 @@ jobs:
         profile:
           - ${{ needs.prepare.outputs.BUILD_PROFILE }}
         otp:
-          - 24.2.1-1 # we test with OTP 23, but only build package on OTP 24 versions
+          - 24.3.4.2-1 # we test with OTP 23, but only build package on OTP 24 versions
         elixir:
           - 1.13.4
         # used to split elixir packages into a separate job, since the
@@ -232,14 +232,14 @@ jobs:
           profile: emqx-enterprise
         include:
           - profile: emqx
-            otp: 24.2.1-1
+            otp: 24.3.4.2-1
             elixir: 1.13.4
             build_elixir: with_elixir
             arch: amd64
             os: ubuntu20.04
             build_machine: ubuntu-20.04
           - profile: emqx
-            otp: 24.2.1-1
+            otp: 24.3.4.2-1
             elixir: 1.13.4
             build_elixir: with_elixir
             arch: amd64
@@ -290,12 +290,12 @@ jobs:
             --pkgtype "${PKGTYPE}" \
             --arch "${ARCH}" \
             --elixir "${IsElixir}" \
-            --builder "ghcr.io/emqx/emqx-builder/5.0-17:${ELIXIR}-${OTP}-${SYSTEM}"
+            --builder "ghcr.io/emqx/emqx-builder/5.0-18:${ELIXIR}-${OTP}-${SYSTEM}"
         done
     - uses: actions/upload-artifact@v3
       with:
         name: ${{ matrix.profile }}-${{ matrix.otp }}
-        path: source/_packages/${{ matrix.profile }}/.
+        path: source/_packages/${{ matrix.profile }}/
 
   publish_artifacts:
     runs-on: ubuntu-20.04
@@ -307,7 +307,7 @@ jobs:
         profile:
           - ${{ needs.prepare.outputs.BUILD_PROFILE }}
         otp:
-          - 24.2.1-1
+          - 24.3.4.2-1
         include:
           - profile: emqx
             otp: windows # otp version on windows is rather fixed
@@ -320,7 +320,7 @@ jobs:
       run: sudo apt-get update && sudo apt install -y dos2unix
     - name: get packages
       run: |
-        DEFAULT_BEAM_PLATFORM='otp24.2.1-1'
+        DEFAULT_BEAM_PLATFORM='otp24.3.4.2-1'
         set -e -u
         cd packages/${{ matrix.profile }}
         # Make a copy of the default OTP version package to a file without OTP version infix

.github/workflows/build_slim_packages.yaml

@@ -32,14 +32,14 @@ jobs:
         - emqx
         - emqx-enterprise
         otp:
-        - 24.2.1-1
+        - 24.3.4.2-1
         elixir:
         - 1.13.4
         os:
         - ubuntu20.04
         - el8
 
-    container: "ghcr.io/emqx/emqx-builder/5.0-17:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os }}"
+    container: "ghcr.io/emqx/emqx-builder/5.0-18:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os }}"
 
     steps:
     - uses: AutoModality/action-clean@v1
@@ -132,7 +132,7 @@ jobs:
         - emqx
         - emqx-enterprise
         otp:
-        - 24.2.1-1
+        - 24.3.4.2-1
         os:
         - macos-11
 

.github/workflows/check_deps_integrity.yaml

@@ -5,7 +5,7 @@ on: [pull_request, push]
 jobs:
   check_deps_integrity:
     runs-on: ubuntu-20.04
-    container: ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04
+    container: ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04
 
     steps:
       - uses: actions/checkout@v3

.github/workflows/code_style_check.yaml

@@ -5,7 +5,7 @@ on: [pull_request]
 jobs:
   code_style_check:
     runs-on: ubuntu-20.04
-    container: "ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04"
+    container: "ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04"
     steps:
       - uses: actions/checkout@v3
         with:

.github/workflows/elixir_apps_check.yaml

@@ -8,7 +8,7 @@ jobs:
   elixir_apps_check:
     runs-on: ubuntu-latest
     # just use the latest builder
-    container: "ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04"
+    container: "ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04"
 
     strategy:
       fail-fast: false

.github/workflows/elixir_deps_check.yaml

@@ -7,7 +7,7 @@ on: [pull_request, push]
 jobs:
   elixir_deps_check:
     runs-on: ubuntu-20.04
-    container: ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04
+    container: ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04
 
     steps:
       - name: Checkout

.github/workflows/elixir_release.yml

@@ -12,7 +12,7 @@ on:
 jobs:
   elixir_release_build:
     runs-on: ubuntu-latest
-    container: ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04
+    container: ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04
 
     steps:
       - name: Checkout

.github/workflows/release.yaml

@@ -32,7 +32,7 @@ jobs:
           esac
           aws s3 cp --recursive s3://${{ secrets.AWS_S3_BUCKET }}/$s3dir/${{ github.ref_name }} packages
           cd packages
-          DEFAULT_BEAM_PLATFORM='otp24.2.1-1'
+          DEFAULT_BEAM_PLATFORM='otp24.3.4.2-1'
           # all packages including full-name and default-name are uploaded to s3
           # but we only upload default-name packages (and elixir) as github artifacts
           # so we rename (overwrite) non-default packages before uploading
@@ -87,7 +87,7 @@ jobs:
     steps:
       - uses: actions/checkout@v3
         with:
-          ref: ${{ github.ref }
+          ref: ${{ github.ref }}
       - uses: emqx/push-helm-action@v1
         if: startsWith(github.ref_name, 'v')
         with:

.github/workflows/run_emqx_app_tests.yaml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       matrix:
         otp:
-          - 24.2.1-1
+          - 24.3.4.2-1
         # no need to use more than 1 version of Elixir, since tests
         # run using only Erlang code.  This is needed just to specify
         # the base image.
@@ -24,7 +24,7 @@ jobs:
           - amd64
 
     runs-on: aws-amd64
-    container: "ghcr.io/emqx/emqx-builder/5.0-17:${{ matrix.elixir}}-${{ matrix.otp }}-${{ matrix.os }}"
+    container: "ghcr.io/emqx/emqx-builder/5.0-18:${{ matrix.elixir}}-${{ matrix.otp }}-${{ matrix.os }}"
 
     defaults:
       run:

.github/workflows/run_fvt_tests.yaml

@@ -16,7 +16,7 @@ jobs:
   prepare:
     runs-on: ubuntu-20.04
     # prepare source with any OTP version, no need for a matrix
-    container: ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-alpine3.15.1
+    container: ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-alpine3.15.1
 
     steps:
       - uses: actions/checkout@v3
@@ -49,7 +49,7 @@ jobs:
         os:
           - ["alpine3.15.1", "alpine:3.15.1"]
         otp:
-          - 24.2.1-1
+          - 24.3.4.2-1
         elixir:
           - 1.13.4
         arch:
@@ -68,7 +68,7 @@ jobs:
     - name: make docker image
       working-directory: source
       env:
-        EMQX_BUILDER: ghcr.io/emqx/emqx-builder/5.0-17:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os[0] }}
+        EMQX_BUILDER: ghcr.io/emqx/emqx-builder/5.0-18:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os[0] }}
         EMQX_RUNNER: ${{ matrix.os[1] }}
       run: |
         make ${{ matrix.profile }}-docker
@@ -120,7 +120,7 @@ jobs:
         os:
         - ["debian11", "debian:11-slim"]
         otp:
-        - 24.2.1-1
+        - 24.3.4.2-1
         elixir:
         - 1.13.4
         arch:
@@ -141,7 +141,7 @@ jobs:
     - name: make docker image
       working-directory: source
       env:
-        EMQX_BUILDER: ghcr.io/emqx/emqx-builder/5.0-17:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os[0] }}
+        EMQX_BUILDER: ghcr.io/emqx/emqx-builder/5.0-18:${{ matrix.elixir }}-${{ matrix.otp }}-${{ matrix.os[0] }}
         EMQX_RUNNER: ${{ matrix.os[1] }}
       run: |
         make ${{ matrix.profile }}-docker

.github/workflows/run_relup_tests.yaml

@@ -16,7 +16,7 @@ on:
 jobs:
   relup_test_plan:
     runs-on: ubuntu-20.04
-    container: "ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04"
+    container: "ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04"
     outputs:
       CUR_EE_VSN: ${{ steps.find-versions.outputs.CUR_EE_VSN }}
       OLD_VERSIONS: ${{ steps.find-versions.outputs.OLD_VERSIONS }}

.github/workflows/run_test_cases.yaml

@@ -17,7 +17,7 @@ jobs:
     prepare:
         runs-on: ubuntu-20.04
         # prepare source with any OTP version, no need for a matrix
-        container: "ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04"
+        container: "ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04"
         outputs:
           fast_ct_apps: ${{ steps.run_find_apps.outputs.fast_ct_apps }}
           docker_ct_apps: ${{ steps.run_find_apps.outputs.docker_ct_apps }}
@@ -60,7 +60,7 @@ jobs:
         defaults:
           run:
             shell: bash
-        container: "ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04"
+        container: "ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04"
 
         steps:
         - uses: AutoModality/action-clean@v1
@@ -121,6 +121,7 @@ jobs:
             PGSQL_TAG: 13
             REDIS_TAG: 6
           run: |
+            rm _build/default/lib/rocksdb/_build/cmake/CMakeCache.txt
             ./scripts/ct/run.sh --app ${{ matrix.app_name }}
         - uses: actions/upload-artifact@v3
           with:
@@ -143,7 +144,7 @@ jobs:
               - emqx-enterprise
 
         runs-on: aws-amd64
-        container: "ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04"
+        container: "ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04"
         defaults:
           run:
             shell: bash
@@ -200,7 +201,7 @@ jobs:
         - ct
         - ct_docker
       runs-on: ubuntu-20.04
-      container: "ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-ubuntu20.04"
+      container: "ghcr.io/emqx/emqx-builder/5.0-18:1.13.4-24.3.4.2-1-ubuntu20.04"
       steps:
       - uses: AutoModality/action-clean@v1
       - uses: actions/download-artifact@v3

Makefile

@@ -6,7 +6,7 @@ export EMQX_DEFAULT_BUILDER = ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-d
 export EMQX_DEFAULT_RUNNER = debian:11-slim
 export OTP_VSN ?= $(shell $(CURDIR)/scripts/get-otp-vsn.sh)
 export ELIXIR_VSN ?= $(shell $(CURDIR)/scripts/get-elixir-vsn.sh)
-export EMQX_DASHBOARD_VERSION ?= v1.1.0
+export EMQX_DASHBOARD_VERSION ?= v1.1.1
 export EMQX_EE_DASHBOARD_VERSION ?= e1.0.0
 export EMQX_REL_FORM ?= tgz
 export QUICER_DOWNLOAD_FROM_RELEASE = 1

apps/emqx/i18n/emqx_schema_i18n.conf

@@ -494,19 +494,6 @@ emqx_schema {
     }
   }
 
-  flapping_detect_clean_when_banned {
-    desc {
-        en: "Clean retained/delayed messages when client is banned.\n"
-            "Note: This may be expensive and only supports users banned by clientid."
-        zh: "当客户端被禁时删除其保留、延迟消息"
-            "注意: 这个操作开销可能较大,且只支持通过 clientid 封禁的用户数据。"
-    }
-    label: {
-        en: "Clean when banned"
-        zh: "被禁时清理消息"
-    }
-  }
-
   persistent_session_store_enabled {
     desc {
         en: "Use the database to store information about persistent sessions.\n"

apps/emqx/priv/bpapi.versions

@@ -9,7 +9,6 @@
 {emqx_conf,2}.
 {emqx_dashboard,1}.
 {emqx_delayed,1}.
-{emqx_delayed,2}.
 {emqx_exhook,1}.
 {emqx_gateway_api_listeners,1}.
 {emqx_gateway_cm,1}.

apps/emqx/src/emqx_access_control.erl

@@ -24,6 +24,11 @@
     authorize/3
 ]).
 
+-ifdef(TEST).
+-compile(export_all).
+-compile(nowarn_export_all).
+-endif.
+
 %%--------------------------------------------------------------------
 %% APIs
 %%--------------------------------------------------------------------
@@ -45,6 +50,19 @@ authenticate(Credential) ->
 %% @doc Check Authorization
 -spec authorize(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_types:topic()) ->
     allow | deny.
+authorize(ClientInfo, PubSub, <<"$delayed/", Data/binary>> = RawTopic) ->
+    case binary:split(Data, <<"/">>) of
+        [_, Topic] ->
+            authorize(ClientInfo, PubSub, Topic);
+        _ ->
+            ?SLOG(warning, #{
+                msg => "invalid_dealyed_topic_format",
+                expected_example => "$delayed/1/t/foo",
+                got => RawTopic
+            }),
+            inc_authz_metrics(deny),
+            deny
+    end;
 authorize(ClientInfo, PubSub, Topic) ->
     Result =
         case emqx_authz_cache:is_enabled() of

apps/emqx/src/emqx_banned.erl

@@ -32,13 +32,11 @@
 -export([
     check/1,
     create/1,
-    create/2,
     look_up/1,
     delete/1,
     info/1,
     format/1,
-    parse/1,
+    parse/1
-    parse_opts/1
 ]).
 
 %% gen_server callbacks
@@ -65,13 +63,6 @@
 -compile(nowarn_export_all).
 -endif.
 
--type banned_opts() :: #{
-    clean => boolean(),
-    atom() => term()
-}.
-
--export_type([banned_opts/0]).
-
 %%--------------------------------------------------------------------
 %% Mnesia bootstrap
 %%--------------------------------------------------------------------
@@ -150,11 +141,6 @@ parse(Params) ->
                     {error, ErrorReason}
             end
     end.
-
-parse_opts(Params) ->
-    Clean = maps:get(<<"clean">>, Params, false),
-    #{clean => Clean}.
-
 pares_who(#{as := As, who := Who}) ->
     pares_who(#{<<"as">> => As, <<"who">> => Who});
 pares_who(#{<<"as">> := peerhost, <<"who">> := Peerhost0}) ->
@@ -176,15 +162,13 @@ to_rfc3339(Timestamp) ->
 
 -spec create(emqx_types:banned() | map()) ->
     {ok, emqx_types:banned()} | {error, {already_exist, emqx_types:banned()}}.
-create(
+create(#{
-    #{
+    who := Who,
-        who := Who,
+    by := By,
-        by := By,
+    reason := Reason,
-        reason := Reason,
+    at := At,
-        at := At,
+    until := Until
-        until := Until
+}) ->
-    } = Data
-) ->
     Banned = #banned{
         who = Who,
         by = By,
@@ -192,16 +176,11 @@ create(
         at = At,
         until = Until
     },
-    create(Banned, Data);
+    create(Banned);
-create(Banned = #banned{}) ->
+create(Banned = #banned{who = Who}) ->
-    create(Banned, #{clean => false}).
-
--spec create(emqx_types:banned(), banned_opts()) ->
-    {ok, emqx_types:banned()} | {error, {already_exist, emqx_types:banned()}}.
-create(Banned = #banned{who = Who}, Opts) ->
     case look_up(Who) of
         [] ->
-            insert_banned(Banned, Opts),
+            mria:dirty_write(?BANNED_TAB, Banned),
             {ok, Banned};
         [OldBanned = #banned{until = Until}] ->
             %% Don't support shorten or extend the until time by overwrite.
@@ -211,7 +190,7 @@ create(Banned = #banned{who = Who}, Opts) ->
                     {error, {already_exist, OldBanned}};
                 %% overwrite expired one is ok.
                 false ->
-                    insert_banned(Banned, Opts),
+                    mria:dirty_write(?BANNED_TAB, Banned),
                     {ok, Banned}
             end
     end.
@@ -287,12 +266,3 @@ expire_banned_items(Now) ->
         ok,
         ?BANNED_TAB
     ).
-
-insert_banned(Banned, Opts) ->
-    mria:dirty_write(?BANNED_TAB, Banned),
-    run_hooks(Banned, Opts).
-
-run_hooks(Banned, #{clean := true}) ->
-    emqx_hooks:run('client.banned', [Banned]);
-run_hooks(_Banned, _Opts) ->
-    ok.

apps/emqx/src/emqx_channel.erl

@@ -2098,7 +2098,7 @@ parse_topic_filters(TopicFilters) ->
     lists:map(fun emqx_topic:parse/1, TopicFilters).
 
 %%--------------------------------------------------------------------
-%% Ensure disconnected
+%% Maybe & Ensure disconnected
 
 ensure_disconnected(
     Reason,
@@ -2205,6 +2205,7 @@ shutdown(success, Reply, Packet, Channel) ->
 shutdown(Reason, Reply, Packet, Channel) ->
     {shutdown, Reason, Reply, Packet, Channel}.
 
+%% mqtt v5 connected sessions
 disconnect_and_shutdown(
     Reason,
     Reply,
@@ -2214,9 +2215,12 @@ disconnect_and_shutdown(
 ) when
     ConnState =:= connected orelse ConnState =:= reauthenticating
 ->
-    shutdown(Reason, Reply, ?DISCONNECT_PACKET(reason_code(Reason)), Channel);
+    NChannel = ensure_disconnected(Reason, Channel),
+    shutdown(Reason, Reply, ?DISCONNECT_PACKET(reason_code(Reason)), NChannel);
+%% mqtt v3/v4 sessions, mqtt v5 other conn_state sessions
 disconnect_and_shutdown(Reason, Reply, Channel) ->
-    shutdown(Reason, Reply, Channel).
+    NChannel = ensure_disconnected(Reason, Channel),
+    shutdown(Reason, Reply, NChannel).
 
 sp(true) -> 1;
 sp(false) -> 0.

apps/emqx/src/emqx_flapping.erl

@@ -121,7 +121,7 @@ handle_cast(
             started_at = StartedAt,
             detect_cnt = DetectCnt
         },
-        #{window_time := WindTime, ban_time := Interval, clean_when_banned := Clean}},
+        #{window_time := WindTime, ban_time := Interval}},
     State
 ) ->
     case now_diff(StartedAt) < WindTime of
@@ -145,7 +145,7 @@ handle_cast(
                 at = Now,
                 until = Now + (Interval div 1000)
             },
-            {ok, _} = emqx_banned:create(Banned, #{clean => Clean}),
+            {ok, _} = emqx_banned:create(Banned),
             ok;
         false ->
             ?SLOG(

apps/emqx/src/emqx_schema.erl

@@ -640,14 +640,6 @@ fields("flapping_detect") ->
                     default => "5m",
                     desc => ?DESC(flapping_detect_ban_time)
                 }
-            )},
-        {"clean_when_banned",
-            sc(
-                boolean(),
-                #{
-                    default => false,
-                    desc => ?DESC(flapping_detect_clean_when_banned)
-                }
             )}
     ];
 fields("force_shutdown") ->

apps/emqx/test/emqx_access_control_SUITE.erl

@@ -32,6 +32,12 @@ init_per_suite(Config) ->
 end_per_suite(_Config) ->
     emqx_common_test_helpers:stop_apps([]).
 
+end_per_testcase(t_delayed_authorize, Config) ->
+    meck:unload(emqx_access_control),
+    Config;
+end_per_testcase(_, Config) ->
+    Config.
+
 t_authenticate(_) ->
     ?assertMatch({ok, _}, emqx_access_control:authenticate(clientinfo())).
 
@@ -39,6 +45,28 @@ t_authorize(_) ->
     Publish = ?PUBLISH_PACKET(?QOS_0, <<"t">>, 1, <<"payload">>),
     ?assertEqual(allow, emqx_access_control:authorize(clientinfo(), Publish, <<"t">>)).
 
+t_delayed_authorize(_) ->
+    RawTopic = "$dealyed/1/foo/2",
+    InvalidTopic = "$dealyed/1/foo/3",
+    Topic = "foo/2",
+
+    ok = meck:new(emqx_access_control, [passthrough, no_history, no_link]),
+    ok = meck:expect(
+        emqx_access_control,
+        do_authorize,
+        fun
+            (_, _, Topic) -> allow;
+            (_, _, _) -> deny
+        end
+    ),
+
+    Publish1 = ?PUBLISH_PACKET(?QOS_0, RawTopic, 1, <<"payload">>),
+    ?assertEqual(allow, emqx_access_control:authorize(clientinfo(), Publish1, RawTopic)),
+
+    Publish2 = ?PUBLISH_PACKET(?QOS_0, InvalidTopic, 1, <<"payload">>),
+    ?assertEqual(allow, emqx_access_control:authorize(clientinfo(), Publish2, InvalidTopic)),
+    ok.
+
 %%--------------------------------------------------------------------
 %% Helper functions
 %%--------------------------------------------------------------------

apps/emqx/test/emqx_flapping_SUITE.erl

@@ -34,8 +34,7 @@ init_per_suite(Config) ->
             % 0.1s
             window_time => 100,
             %% 2s
-            ban_time => 2000,
+            ban_time => 2000
-            clean_when_banned => false
         }
     ),
     Config.

apps/emqx/test/emqx_takeover_SUITE.erl

@@ -33,6 +33,7 @@
 all() -> emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
+    emqx_common_test_helpers:boot_modules(all),
     emqx_channel_SUITE:set_test_listener_confs(),
     ?check_trace(
         ?wait_async_action(

apps/emqx_authz/src/emqx_authz.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_authz, [
     {description, "An OTP application"},
-    {vsn, "0.1.6"},
+    {vsn, "0.1.7"},
     {registered, []},
     {mod, {emqx_authz_app, []}},
     {applications, [

apps/emqx_authz/src/emqx_authz_api_sources.erl

@@ -40,7 +40,8 @@
 -export([
     api_spec/0,
     paths/0,
-    schema/1
+    schema/1,
+    fields/1
 ]).
 
 -export([
@@ -63,6 +64,9 @@ paths() ->
         "/authorization/sources/:type/move"
     ].
 
+fields(sources) ->
+    [{sources, mk(array(hoconsc:union(authz_sources_type_refs())), #{desc => ?DESC(sources)})}].
+
 %%--------------------------------------------------------------------
 %% Schema for each URI
 %%--------------------------------------------------------------------
@@ -75,10 +79,7 @@ schema("/authorization/sources") ->
                 tags => ?TAGS,
                 responses =>
                     #{
-                        200 => mk(
+                        200 => ref(?MODULE, sources)
-                            array(hoconsc:union(authz_sources_type_refs())),
-                            #{desc => ?DESC(sources)}
-                        )
                     }
             },
         post =>
@@ -241,7 +242,7 @@ source(Method, #{bindings := #{type := Type} = Bindings} = Req) when
 source(get, #{bindings := #{type := Type}}) ->
     case get_raw_source(Type) of
         [] ->
-            {404, #{message => <<"Not found ", Type/binary>>}};
+            {404, #{code => <<"NOT_FOUND">>, message => <<"Not found: ", Type/binary>>}};
         [#{<<"type">> := <<"file">>, <<"enable">> := Enable, <<"path">> := Path}] ->
             case file:read_file(Path) of
                 {ok, Rules} ->

apps/emqx_authz/test/emqx_authz_api_sources_SUITE.erl

@@ -181,6 +181,12 @@ t_api(_) ->
     {ok, 200, Result1} = request(get, uri(["authorization", "sources"]), []),
     ?assertEqual([], get_sources(Result1)),
 
+    {ok, 404, ErrResult} = request(get, uri(["authorization", "sources", "http"]), []),
+    ?assertMatch(
+        #{<<"code">> := <<"NOT_FOUND">>, <<"message">> := <<"Not found: http">>},
+        jsx:decode(ErrResult)
+    ),
+
     [
         begin
             {ok, 204, _} = request(post, uri(["authorization", "sources"]), Source)

apps/emqx_management/i18n/emqx_mgmt_api_banned_i18n.conf

@@ -95,16 +95,4 @@ emqx_mgmt_api_banned {
             zh: """封禁结束时间"""
         }
     }
-    clean {
-        desc {
-            en: """Clean retained/delayed messages when client is banned."""
-                """Note: This may be expensive and only supports users banned by clientid."""
-            zh: """当客户端被禁时删除其保留、延迟消息"""
-                """注意: 这个操作开销可能较大,且只支持通过 clientid 封禁的用户数据。"""
-        }
-        label {
-            en: """Clean when banned"""
-            zh: """被禁时清理消息"""
-        }
-    }
 }

apps/emqx_management/src/emqx_mgmt_api_banned.erl

@@ -150,13 +150,6 @@ fields(ban) ->
                 desc => ?DESC(until),
                 required => false,
                 example => <<"2021-10-25T21:53:47+08:00">>
-            })},
-        {clean,
-            hoconsc:mk(boolean(), #{
-                desc => ?DESC(clean),
-                required => false,
-                default => false,
-                example => false
             })}
     ].
 
@@ -168,8 +161,7 @@ banned(post, #{body := Body}) ->
         {error, Reason} ->
             {400, 'BAD_REQUEST', list_to_binary(Reason)};
         Ban ->
-            Opts = emqx_banned:parse_opts(Body),
+            case emqx_banned:create(Ban) of
-            case emqx_banned:create(Ban, Opts) of
                 {ok, Banned} ->
                     {200, format(Banned)};
                 {error, {already_exist, Old}} ->

apps/emqx_management/src/emqx_mgmt_api_configs.erl

@@ -103,7 +103,9 @@ schema("/configs") ->
                     )}
             ],
             responses => #{
-                200 => lists:map(fun({_, Schema}) -> Schema end, config_list())
+                200 => lists:map(fun({_, Schema}) -> Schema end, config_list()),
+                404 => emqx_dashboard_swagger:error_codes(['NOT_FOUND']),
+                500 => emqx_dashboard_swagger:error_codes(['BAD_NODE'])
             }
         }
     };
@@ -311,14 +313,15 @@ config_reset(post, _Params, Req) ->
     end.
 
 configs(get, Params, _Req) ->
-    Node = maps:get(node, Params, node()),
+    QS = maps:get(query_string, Params, #{}),
+    Node = maps:get(<<"node">>, QS, node()),
     case
         lists:member(Node, mria_mnesia:running_nodes()) andalso
             emqx_management_proto_v2:get_full_config(Node)
     of
         false ->
             Message = list_to_binary(io_lib:format("Bad node ~p, reason not found", [Node])),
-            {500, #{code => 'BAD_NODE', message => Message}};
+            {404, #{code => 'NOT_FOUND', message => Message}};
         {badrpc, R} ->
             Message = list_to_binary(io_lib:format("Bad node ~p, reason ~p", [Node, R])),
             {500, #{code => 'BAD_NODE', message => Message}};

apps/emqx_management/test/emqx_mgmt_api_configs_SUITE.erl

@@ -30,6 +30,16 @@ init_per_suite(Config) ->
 end_per_suite(_) ->
     emqx_mgmt_api_test_util:end_suite([emqx_conf]).
 
+init_per_testcase(TestCase = t_configs_node, Config) ->
+    ?MODULE:TestCase({'init', Config});
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(TestCase = t_configs_node, Config) ->
+    ?MODULE:TestCase({'end', Config});
+end_per_testcase(_TestCase, Config) ->
+    Config.
+
 t_get(_Config) ->
     {ok, Configs} = get_configs(),
     maps:map(
@@ -188,6 +198,37 @@ t_dashboard(_Config) ->
     timer:sleep(1000),
     ok.
 
+t_configs_node({'init', Config}) ->
+    Node = node(),
+    meck:expect(mria_mnesia, running_nodes, fun() -> [Node, bad_node, other_node] end),
+    meck:expect(
+        emqx_management_proto_v2,
+        get_full_config,
+        fun
+            (Node0) when Node0 =:= Node -> <<"\"self\"">>;
+            (other_node) -> <<"\"other\"">>;
+            (bad_node) -> {badrpc, bad}
+        end
+    ),
+    Config;
+t_configs_node({'end', _}) ->
+    meck:unload([mria_mnesia, emqx_management_proto_v2]);
+t_configs_node(_) ->
+    Node = atom_to_list(node()),
+
+    ?assertEqual({ok, <<"self">>}, get_configs(Node, #{return_body => true})),
+    ?assertEqual({ok, <<"other">>}, get_configs("other_node", #{return_body => true})),
+
+    {ExpType, ExpRes} = get_configs("unknown_node", #{return_body => true}),
+    ?assertEqual(error, ExpType),
+    ?assertMatch({{_, 404, _}, _, _}, ExpRes),
+    {_, _, Body} = ExpRes,
+    ?assertMatch(#{<<"code">> := <<"NOT_FOUND">>}, emqx_json:decode(Body, [return_maps])),
+
+    ?assertMatch({error, {_, 500, _}}, get_configs("bad_node")).
+
+%% Helpers
+
 get_config(Name) ->
     Path = emqx_mgmt_api_test_util:api_path(["configs", Name]),
     case emqx_mgmt_api_test_util:request_api(get, Path) of
@@ -198,8 +239,19 @@ get_config(Name) ->
     end.
 
 get_configs() ->
-    Path = emqx_mgmt_api_test_util:api_path(["configs"]),
+    get_configs([], #{}).
-    case emqx_mgmt_api_test_util:request_api(get, Path) of
+
+get_configs(Node) ->
+    get_configs(Node, #{}).
+
+get_configs(Node, Opts) ->
+    Path =
+        case Node of
+            [] -> ["configs"];
+            _ -> ["configs?node=" ++ Node]
+        end,
+    URI = emqx_mgmt_api_test_util:api_path(Path),
+    case emqx_mgmt_api_test_util:request_api(get, URI, [], [], [], Opts) of
         {ok, Res} -> {ok, emqx_json:decode(Res, [return_maps])};
         Error -> Error
     end.

apps/emqx_management/test/emqx_mgmt_api_test_util.erl

@@ -44,15 +44,20 @@ set_special_configs(_App) ->
     ok.
 
 request_api(Method, Url) ->
-    request_api(Method, Url, [], auth_header_(), []).
+    request_api(Method, Url, [], [], [], #{}).
 
 request_api(Method, Url, AuthOrHeaders) ->
-    request_api(Method, Url, [], AuthOrHeaders, []).
+    request_api(Method, Url, [], AuthOrHeaders, [], #{}).
 
 request_api(Method, Url, QueryParams, AuthOrHeaders) ->
-    request_api(Method, Url, QueryParams, AuthOrHeaders, []).
+    request_api(Method, Url, QueryParams, AuthOrHeaders, [], #{}).
 
-request_api(Method, Url, QueryParams, AuthOrHeaders, []) when
+request_api(Method, Url, QueryParams, AuthOrHeaders, Body) ->
+    request_api(Method, Url, QueryParams, AuthOrHeaders, Body, #{}).
+
+request_api(Method, Url, QueryParams, [], Body, Opts) ->
+    request_api(Method, Url, QueryParams, auth_header_(), Body, Opts);
+request_api(Method, Url, QueryParams, AuthOrHeaders, [], Opts) when
     (Method =:= options) orelse
         (Method =:= get) orelse
         (Method =:= put) orelse
@@ -65,10 +70,7 @@ request_api(Method, Url, QueryParams, AuthOrHeaders, []) when
             "" -> Url;
             _ -> Url ++ "?" ++ QueryParams
         end,
-    do_request_api(Method, {NewUrl, build_http_header(AuthOrHeaders)}, #{});
+    do_request_api(Method, {NewUrl, build_http_header(AuthOrHeaders)}, Opts);
-request_api(Method, Url, QueryParams, AuthOrHeaders, Body) ->
-    request_api(Method, Url, QueryParams, AuthOrHeaders, Body, #{}).
-
 request_api(Method, Url, QueryParams, AuthOrHeaders, Body, Opts) when
     (Method =:= post) orelse
         (Method =:= patch) orelse

apps/emqx_modules/src/emqx_delayed.erl

@@ -23,7 +23,6 @@
 -include_lib("emqx/include/logger.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 -include_lib("emqx/include/emqx_hooks.hrl").
--include_lib("stdlib/include/ms_transform.hrl").
 
 %% Mnesia bootstrap
 -export([mnesia/1]).
@@ -32,8 +31,7 @@
 
 -export([
     start_link/0,
-    on_message_publish/1,
+    on_message_publish/1
-    on_client_banned/1
 ]).
 
 %% gen_server callbacks
@@ -46,7 +44,7 @@
     code_change/3
 ]).
 
-%% API
+%% gen_server callbacks
 -export([
     load/0,
     unload/0,
@@ -59,9 +57,7 @@
     delete_delayed_message/1,
     delete_delayed_message/2,
     cluster_list/1,
-    cluster_query/4,
+    cluster_query/4
-    clean_by_clientid/1,
-    do_clean_by_clientid/1
 ]).
 
 -export([
@@ -142,11 +138,6 @@ on_message_publish(
 on_message_publish(Msg) ->
     {ok, Msg}.
 
-on_client_banned(#banned{who = {clientid, ClientId}}) ->
-    clean_by_clientid(ClientId);
-on_client_banned(_) ->
-    ok.
-
 %%--------------------------------------------------------------------
 %% Start delayed publish server
 %%--------------------------------------------------------------------
@@ -237,7 +228,7 @@ get_delayed_message(Id) ->
 get_delayed_message(Node, Id) when Node =:= node() ->
     get_delayed_message(Id);
 get_delayed_message(Node, Id) ->
-    emqx_delayed_proto_v2:get_delayed_message(Node, Id).
+    emqx_delayed_proto_v1:get_delayed_message(Node, Id).
 
 -spec delete_delayed_message(binary()) -> with_id_return().
 delete_delayed_message(Id) ->
@@ -252,7 +243,7 @@ delete_delayed_message(Id) ->
 delete_delayed_message(Node, Id) when Node =:= node() ->
     delete_delayed_message(Id);
 delete_delayed_message(Node, Id) ->
-    emqx_delayed_proto_v2:delete_delayed_message(Node, Id).
+    emqx_delayed_proto_v1:delete_delayed_message(Node, Id).
 
 update_config(Config) ->
     emqx_conf:update([delayed], Config, #{rawconf_with_defaults => true, override_to => cluster}).
@@ -261,15 +252,6 @@ post_config_update(_KeyPath, _ConfigReq, NewConf, _OldConf, _AppEnvs) ->
     Enable = maps:get(enable, NewConf, undefined),
     load_or_unload(Enable).
 
-clean_by_clientid(ClientId) ->
-    Nodes = mria_mnesia:running_nodes(),
-    emqx_delayed_proto_v2:clean_by_clientid(Nodes, ClientId).
-
-do_clean_by_clientid(ClientId) ->
-    ets:select_delete(
-        ?TAB, ets:fun2ms(fun(#delayed_message{msg = Msg}) -> Msg#message.from =:= ClientId end)
-    ).
-
 %%--------------------------------------------------------------------
 %% gen_server callback
 %%--------------------------------------------------------------------
@@ -391,8 +373,23 @@ do_publish({Ts, _Id}, Now, Acc) when Ts > Now ->
     Acc;
 do_publish(Key = {Ts, _Id}, Now, Acc) when Ts =< Now ->
     case mnesia:dirty_read(?TAB, Key) of
-        [] -> ok;
+        [] ->
-        [#delayed_message{msg = Msg}] -> emqx_pool:async_submit(fun emqx:publish/1, [Msg])
+            ok;
+        [#delayed_message{msg = Msg}] ->
+            case emqx_banned:look_up({clientid, Msg#message.from}) of
+                [] ->
+                    emqx_pool:async_submit(fun emqx:publish/1, [Msg]);
+                _ ->
+                    ?tp(
+                        notice,
+                        ignore_delayed_message_publish,
+                        #{
+                            reason => "client is banned",
+                            clienid => Msg#message.from
+                        }
+                    ),
+                    ok
+            end
     end,
     do_publish(mnesia:dirty_next(?TAB, Key), Now, [Key | Acc]).
 
@@ -401,11 +398,9 @@ delayed_count() -> mnesia:table_info(?TAB, size).
 
 do_load_or_unload(true, State) ->
     emqx_hooks:put('message.publish', {?MODULE, on_message_publish, []}, ?HP_DELAY_PUB),
-    ok = emqx_hooks:put('client.banned', {?MODULE, on_client_banned, []}, ?HP_LOWEST),
     State;
 do_load_or_unload(false, #{publish_timer := PubTimer} = State) ->
     emqx_hooks:del('message.publish', {?MODULE, on_message_publish}),
-    ok = emqx_hooks:del('client.banned', {?MODULE, on_client_banned}),
     emqx_misc:cancel_timer(PubTimer),
     ets:delete_all_objects(?TAB),
     State#{publish_timer := undefined, publish_at := 0};

apps/emqx_modules/src/proto/emqx_delayed_proto_v2.erl

@@ -1,47 +0,0 @@
-%%--------------------------------------------------------------------
-%%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.
-%%--------------------------------------------------------------------
-
--module(emqx_delayed_proto_v2).
-
--behaviour(emqx_bpapi).
-
--export([
-    introduced_in/0,
-    get_delayed_message/2,
-    delete_delayed_message/2,
-    clean_by_clientid/2
-]).
-
--include_lib("emqx/include/bpapi.hrl").
-
--define(TIMEOUT, 15000).
-
-introduced_in() ->
-    "5.0.10".
-
--spec get_delayed_message(node(), binary()) ->
-    emqx_delayed:with_id_return(map()) | emqx_rpc:badrpc().
-get_delayed_message(Node, Id) ->
-    rpc:call(Node, emqx_delayed, get_delayed_message, [Id]).
-
--spec delete_delayed_message(node(), binary()) -> emqx_delayed:with_id_return() | emqx_rpc:badrpc().
-delete_delayed_message(Node, Id) ->
-    rpc:call(Node, emqx_delayed, delete_delayed_message, [Id]).
-
--spec clean_by_clientid(list(node()), emqx_types:clientid()) ->
-    emqx_rpc:erpc_multicall().
-clean_by_clientid(Nodes, ClientID) ->
-    erpc:multicall(Nodes, emqx_delayed, do_clean_by_clientid, [ClientID], ?TIMEOUT).

apps/emqx_modules/test/emqx_delayed_SUITE.erl

@@ -26,6 +26,7 @@
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("emqx/include/emqx.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 %%--------------------------------------------------------------------
 %% Setups
@@ -36,7 +37,8 @@
 }).
 
 all() ->
-    emqx_common_test_helpers:all(?MODULE).
+    [t_banned_delayed].
+%%    emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
     ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
@@ -212,66 +214,37 @@ t_delayed_precision(_) ->
     _ = on_message_publish(DelayedMsg0),
     ?assert(FutureDiff() =< MaxSpan).
 
-t_banned_clean(_) ->
+t_banned_delayed(_) ->
     emqx:update_config([delayed, max_delayed_messages], 10000),
     ClientId1 = <<"bc1">>,
     ClientId2 = <<"bc2">>,
-    {ok, C1} = emqtt:start_link([{clientid, ClientId1}, {clean_start, true}, {proto_ver, v5}]),
-    {ok, _} = emqtt:connect(C1),
-
-    {ok, C2} = emqtt:start_link([{clientid, ClientId2}, {clean_start, true}, {proto_ver, v5}]),
-    {ok, _} = emqtt:connect(C2),
-
-    [
-        begin
-            emqtt:publish(
-                Conn,
-                <<"$delayed/60/0/", ClientId/binary>>,
-                <<"">>,
-                [{qos, 0}, {retain, false}]
-            ),
-            emqtt:publish(
-                Conn,
-                <<"$delayed/60/1/", ClientId/binary>>,
-                <<"">>,
-                [{qos, 0}, {retain, false}]
-            )
-        end
-     || {ClientId, Conn} <- lists:zip([ClientId1, ClientId2], [C1, C2])
-    ],
-
-    emqtt:publish(
-        C2,
-        <<"$delayed/60/2/", ClientId2/binary>>,
-        <<"">>,
-        [{qos, 0}, {retain, false}]
-    ),
-
-    timer:sleep(500),
-    ?assertMatch(#{meta := #{count := 5}}, emqx_delayed:list(#{page => 1, limit => 10})),
 
     Now = erlang:system_time(second),
     Who = {clientid, ClientId2},
-    try
+    emqx_banned:create(#{
-        emqx_banned:create(#{
+        who => Who,
-            who => Who,
+        by => <<"test">>,
-            by => <<"test">>,
+        reason => <<"test">>,
-            reason => <<"test">>,
+        at => Now,
-            at => Now,
+        until => Now + 120
-            until => Now + 120,
+    }),
-            clean => true
-        }),
 
-        timer:sleep(500),
+    snabbkaffe:start_trace(),
+    lists:foreach(
+        fun(ClientId) ->
+            Msg = emqx_message:make(ClientId, <<"$delayed/1/bc">>, <<"payload">>),
+            emqx_delayed:on_message_publish(Msg)
+        end,
+        [ClientId1, ClientId1, ClientId1, ClientId2, ClientId2]
+    ),
 
-        ?assertMatch(#{meta := #{count := 2}}, emqx_delayed:list(#{page => 1, limit => 10}))
+    timer:sleep(2000),
-    after
+    Trace = snabbkaffe:collect_trace(),
-        emqx_banned:delete(Who),
+    snabbkaffe:stop(),
-        emqx_delayed:clean_by_clientid(ClientId1)
+    emqx_banned:delete(Who),
-    end,
+    mnesia:clear_table(emqx_delayed),
-    timer:sleep(500),
+
-    ok = emqtt:disconnect(C1),
+    ?assertEqual(2, length(?of_kind(ignore_delayed_message_publish, Trace))).
-    ok = emqtt:disconnect(C2).
 
 subscribe_proc() ->
     Self = self(),

apps/emqx_retainer/src/emqx_retainer.erl

@@ -19,7 +19,6 @@
 -behaviour(gen_server).
 
 -include("emqx_retainer.hrl").
--include_lib("emqx/include/emqx.hrl").
 -include_lib("emqx/include/logger.hrl").
 -include_lib("emqx/include/emqx_hooks.hrl").
 
@@ -27,8 +26,7 @@
 
 -export([
     on_session_subscribed/4,
-    on_message_publish/2,
+    on_message_publish/2
-    on_client_banned/1
 ]).
 
 -export([
@@ -41,7 +39,6 @@
     get_expiry_time/1,
     update_config/1,
     clean/0,
-    clean_by_clientid/1,
     delete/1,
     page_read/3,
     post_config_update/5,
@@ -83,7 +80,6 @@
 -callback match_messages(context(), topic(), cursor()) -> {ok, list(), cursor()}.
 -callback clear_expired(context()) -> ok.
 -callback clean(context()) -> ok.
--callback clean_by_clientid(context(), emqx_types:clientid()) -> ok.
 -callback size(context()) -> non_neg_integer().
 
 %%--------------------------------------------------------------------
@@ -122,11 +118,6 @@ on_message_publish(Msg = #message{flags = #{retain := true}}, Context) ->
 on_message_publish(Msg, _) ->
     {ok, Msg}.
 
-on_client_banned(#banned{who = {clientid, ClientId}}) ->
-    clean_by_clientid(ClientId);
-on_client_banned(_) ->
-    ok.
-
 %%--------------------------------------------------------------------
 %% APIs
 %%--------------------------------------------------------------------
@@ -160,9 +151,6 @@ update_config(Conf) ->
 clean() ->
     call(?FUNCTION_NAME).
 
-clean_by_clientid(ClientId) ->
-    call({?FUNCTION_NAME, ClientId}).
-
 delete(Topic) ->
     call({?FUNCTION_NAME, Topic}).
 
@@ -219,9 +207,6 @@ handle_call({update_config, NewConf, OldConf}, _, State) ->
 handle_call(clean, _, #{context := Context} = State) ->
     clean(Context),
     {reply, ok, State};
-handle_call({clean_by_clientid, ClientId}, _, #{context := Context} = State) ->
-    clean_by_clientid(Context, ClientId),
-    {reply, ok, State};
 handle_call({delete, Topic}, _, #{context := Context} = State) ->
     delete_message(Context, Topic),
     {reply, ok, State};
@@ -313,11 +298,6 @@ clean(Context) ->
     Mod = get_backend_module(),
     Mod:clean(Context).
 
--spec clean_by_clientid(context(), emqx_types:clientid()) -> ok.
-clean_by_clientid(Context, ClientId) ->
-    Mod = get_backend_module(),
-    Mod:clean_by_clientid(Context, ClientId).
-
 -spec update_config(state(), hocons:config(), hocons:config()) -> state().
 update_config(State, Conf, OldConf) ->
     update_config(
@@ -453,13 +433,11 @@ load(Context) ->
         'session.subscribed', {?MODULE, on_session_subscribed, [Context]}, ?HP_RETAINER
     ),
     ok = emqx_hooks:put('message.publish', {?MODULE, on_message_publish, [Context]}, ?HP_RETAINER),
-    ok = emqx_hooks:put('client.banned', {?MODULE, on_client_banned, []}, ?HP_LOWEST),
     emqx_stats:update_interval(emqx_retainer_stats, fun ?MODULE:stats_fun/0),
     ok.
 
 unload() ->
     ok = emqx_hooks:del('message.publish', {?MODULE, on_message_publish}),
     ok = emqx_hooks:del('session.subscribed', {?MODULE, on_session_subscribed}),
-    ok = emqx_hooks:del('client.banned', {?MODULE, on_client_banned}),
     emqx_stats:cancel_update(emqx_retainer_stats),
     ok.

apps/emqx_retainer/src/emqx_retainer_mnesia.erl

@@ -33,14 +33,13 @@
     match_messages/3,
     clear_expired/1,
     clean/1,
-    clean_by_clientid/2,
     size/1
 ]).
 
 %% Internal exports (RPC)
 -export([
     do_store_retained/1,
-    do_clear/1,
+    do_clear_expired/0,
     do_delete_message/1,
     do_populate_index_meta/1,
     do_reindex_batch/2
@@ -62,8 +61,6 @@
 -record(retained_index, {key, expiry_time}).
 -record(retained_index_meta, {key, read_indices, write_indices, reindexing, extra}).
 
--type retained_message() :: #retained_message{}.
-
 -define(META_KEY, index_meta).
 
 -define(CLEAR_BATCH_SIZE, 1000).
@@ -167,22 +164,18 @@ do_store_retained(#message{topic = Topic} = Msg) ->
     end.
 
 clear_expired(_) ->
-    NowMs = erlang:system_time(millisecond),
+    {atomic, _} = mria:transaction(?RETAINER_SHARD, fun ?MODULE:do_clear_expired/0),
-    {atomic, _} = mria:transaction(?RETAINER_SHARD, fun ?MODULE:do_clear/1, [
-        fun(
-            #retained_message{expiry_time = ExpiryTime}
-        ) ->
-            (ExpiryTime =/= 0) and (ExpiryTime < NowMs)
-        end
-    ]),
     ok.
 
--spec do_clear(fun((retained_message()) -> boolean())) -> ok.
+do_clear_expired() ->
-do_clear(Pred) ->
+    NowMs = erlang:system_time(millisecond),
     QH = qlc:q([
         TopicTokens
-     || #retained_message{topic = TopicTokens} = Data <- mnesia:table(?TAB_MESSAGE, [{lock, write}]),
+     || #retained_message{
-        Pred(Data)
+            topic = TopicTokens,
+            expiry_time = ExpiryTime
+        } <- mnesia:table(?TAB_MESSAGE, [{lock, write}]),
+        (ExpiryTime =/= 0) and (ExpiryTime < NowMs)
     ]),
     QC = qlc:cursor(QH),
     clear_batch(db_indices(write), QC).
@@ -270,14 +263,6 @@ clean(_) ->
     _ = mria:clear_table(?TAB_INDEX),
     ok.
 
-clean_by_clientid(_, ClientId) ->
-    {atomic, _} = mria:transaction(?RETAINER_SHARD, fun ?MODULE:do_clear/1, [
-        fun(Msg) ->
-            Msg#retained_message.msg#message.from =:= ClientId
-        end
-    ]),
-    ok.
-
 size(_) ->
     table_size().
 

apps/emqx_retainer/test/emqx_retainer_SUITE.erl

@@ -626,66 +626,6 @@ t_get_basic_usage_info(_Config) ->
     ?assertEqual(#{retained_messages => 5}, emqx_retainer:get_basic_usage_info()),
     ok.
 
-t_banned_clean(_) ->
-    ClientId1 = <<"bc1">>,
-    ClientId2 = <<"bc2">>,
-    {ok, C1} = emqtt:start_link([{clientid, ClientId1}, {clean_start, true}, {proto_ver, v5}]),
-    {ok, _} = emqtt:connect(C1),
-
-    {ok, C2} = emqtt:start_link([{clientid, ClientId2}, {clean_start, true}, {proto_ver, v5}]),
-    {ok, _} = emqtt:connect(C2),
-
-    [
-        begin
-            emqtt:publish(
-                Conn,
-                <<"bc/0/", ClientId/binary>>,
-                <<"this is a retained message 0">>,
-                [{qos, 0}, {retain, true}]
-            ),
-            emqtt:publish(
-                Conn,
-                <<"bc/1/", ClientId/binary>>,
-                <<"this is a retained message 1">>,
-                [{qos, 0}, {retain, true}]
-            )
-        end
-     || {ClientId, Conn} <- lists:zip([ClientId1, ClientId2], [C1, C2])
-    ],
-
-    emqtt:publish(
-        C2,
-        <<"bc/2/", ClientId2/binary>>,
-        <<"this is a retained message 2">>,
-        [{qos, 0}, {retain, true}]
-    ),
-
-    timer:sleep(500),
-    {ok, List} = emqx_retainer:page_read(<<"bc/+/+">>, 1, 10),
-    ?assertEqual(5, length(List)),
-
-    Now = erlang:system_time(second),
-    Who = {clientid, ClientId2},
-    emqx_banned:create(#{
-        who => Who,
-        by => <<"test">>,
-        reason => <<"test">>,
-        at => Now,
-        until => Now + 120,
-        clean => true
-    }),
-
-    timer:sleep(500),
-
-    {ok, List2} = emqx_retainer:page_read(<<"bc/#">>, 1, 10),
-    ?assertEqual(2, length(List2)),
-
-    emqx_banned:delete(Who),
-    emqx_retainer:clean(),
-    timer:sleep(500),
-    ok = emqtt:disconnect(C1),
-    ok = emqtt:disconnect(C2).
-
 %% test whether the app can start normally after disabling emqx_retainer
 %% fix: https://github.com/emqx/emqx/pull/8911
 test_disable_then_start(_Config) ->

apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl

@@ -91,7 +91,13 @@ groups() ->
             t_sqlparse_new_map,
             t_sqlparse_invalid_json
         ]},
-        {events, [], [t_events]},
+        {events, [], [
+            t_events,
+            t_event_client_disconnected_normal,
+            t_event_client_disconnected_kicked,
+            t_event_client_disconnected_discarded,
+            t_event_client_disconnected_takenover
+        ]},
         {telemetry, [], [
             t_get_basic_usage_info_0,
             t_get_basic_usage_info_1
@@ -474,6 +480,165 @@ t_events(_Config) ->
     client_connack_failed(),
     ok.
 
+t_event_client_disconnected_normal(_Config) ->
+    SQL =
+        "select * "
+        "from \"$events/client_disconnected\" ",
+    RepubT = <<"repub/to/disconnected/normal">>,
+
+    {ok, TopicRule} = emqx_rule_engine:create_rule(
+        #{
+            sql => SQL,
+            id => ?TMP_RULEID,
+            actions => [republish_action(RepubT, <<>>)]
+        }
+    ),
+
+    {ok, Client} = emqtt:start_link([{clientid, <<"get_repub_client">>}, {username, <<"emqx0">>}]),
+    {ok, _} = emqtt:connect(Client),
+    {ok, _, _} = emqtt:subscribe(Client, RepubT, 0),
+    ct:sleep(200),
+    {ok, Client1} = emqtt:start_link([{clientid, <<"emqx">>}, {username, <<"emqx">>}]),
+    {ok, _} = emqtt:connect(Client1),
+    emqtt:disconnect(Client1),
+
+    receive
+        {publish, #{topic := T, payload := Payload}} ->
+            ?assertEqual(RepubT, T),
+            ?assertMatch(#{<<"reason">> := <<"normal">>}, emqx_json:decode(Payload, [return_maps]))
+    after 1000 ->
+        ct:fail(wait_for_repub_disconnected_normal)
+    end,
+    emqtt:stop(Client),
+
+    delete_rule(TopicRule).
+
+t_event_client_disconnected_kicked(_Config) ->
+    SQL =
+        "select * "
+        "from \"$events/client_disconnected\" ",
+    RepubT = <<"repub/to/disconnected/kicked">>,
+
+    {ok, TopicRule} = emqx_rule_engine:create_rule(
+        #{
+            sql => SQL,
+            id => ?TMP_RULEID,
+            actions => [republish_action(RepubT, <<>>)]
+        }
+    ),
+
+    {ok, Client} = emqtt:start_link([{clientid, <<"get_repub_client">>}, {username, <<"emqx0">>}]),
+    {ok, _} = emqtt:connect(Client),
+    {ok, _, _} = emqtt:subscribe(Client, RepubT, 0),
+    ct:sleep(200),
+
+    {ok, Client1} = emqtt:start_link([{clientid, <<"emqx">>}, {username, <<"emqx">>}]),
+    {ok, _} = emqtt:connect(Client1),
+    %% the process will receive {'EXIT',{shutdown,tcp_closed}}
+    unlink(Client1),
+
+    emqx_cm:kick_session(<<"emqx">>),
+
+    receive
+        {publish, #{topic := T, payload := Payload}} ->
+            ?assertEqual(RepubT, T),
+            ?assertMatch(#{<<"reason">> := <<"kicked">>}, emqx_json:decode(Payload, [return_maps]))
+    after 1000 ->
+        ct:fail(wait_for_repub_disconnected_kicked)
+    end,
+
+    emqtt:stop(Client),
+    delete_rule(TopicRule).
+
+t_event_client_disconnected_discarded(_Config) ->
+    SQL =
+        "select * "
+        "from \"$events/client_disconnected\" ",
+    RepubT = <<"repub/to/disconnected/discarded">>,
+
+    {ok, TopicRule} = emqx_rule_engine:create_rule(
+        #{
+            sql => SQL,
+            id => ?TMP_RULEID,
+            actions => [republish_action(RepubT, <<>>)]
+        }
+    ),
+
+    {ok, Client} = emqtt:start_link([{clientid, <<"get_repub_client">>}, {username, <<"emqx0">>}]),
+    {ok, _} = emqtt:connect(Client),
+    {ok, _, _} = emqtt:subscribe(Client, RepubT, 0),
+    ct:sleep(200),
+
+    {ok, Client1} = emqtt:start_link([{clientid, <<"emqx">>}, {username, <<"emqx">>}]),
+    {ok, _} = emqtt:connect(Client1),
+    %% the process will receive {'EXIT',{shutdown,tcp_closed}}
+    unlink(Client1),
+
+    {ok, Client2} = emqtt:start_link([
+        {clientid, <<"emqx">>}, {username, <<"emqx">>}, {clean_start, true}
+    ]),
+    {ok, _} = emqtt:connect(Client2),
+
+    receive
+        {publish, #{topic := T, payload := Payload}} ->
+            ?assertEqual(RepubT, T),
+            ?assertMatch(
+                #{<<"reason">> := <<"discarded">>}, emqx_json:decode(Payload, [return_maps])
+            )
+    after 1000 ->
+        ct:fail(wait_for_repub_disconnected_discarded)
+    end,
+    emqtt:stop(Client),
+    emqtt:stop(Client2),
+
+    delete_rule(TopicRule).
+
+t_event_client_disconnected_takenover(_Config) ->
+    SQL =
+        "select * "
+        "from \"$events/client_disconnected\" ",
+    RepubT = <<"repub/to/disconnected/takenover">>,
+
+    {ok, TopicRule} = emqx_rule_engine:create_rule(
+        #{
+            sql => SQL,
+            id => ?TMP_RULEID,
+            actions => [republish_action(RepubT, <<>>)]
+        }
+    ),
+
+    {ok, ClientRecv} = emqtt:start_link([
+        {clientid, <<"get_repub_client">>}, {username, <<"emqx0">>}
+    ]),
+    {ok, _} = emqtt:connect(ClientRecv),
+    {ok, _, _} = emqtt:subscribe(ClientRecv, RepubT, 0),
+    ct:sleep(200),
+
+    {ok, Client1} = emqtt:start_link([{clientid, <<"emqx">>}, {username, <<"emqx">>}]),
+    {ok, _} = emqtt:connect(Client1),
+    %% the process will receive {'EXIT',{shutdown,tcp_closed}}
+    unlink(Client1),
+
+    {ok, Client2} = emqtt:start_link([
+        {clientid, <<"emqx">>}, {username, <<"emqx">>}, {clean_start, false}
+    ]),
+    {ok, _} = emqtt:connect(Client2),
+
+    receive
+        {publish, #{topic := T, payload := Payload}} ->
+            ?assertEqual(RepubT, T),
+            ?assertMatch(
+                #{<<"reason">> := <<"takenover">>}, emqx_json:decode(Payload, [return_maps])
+            )
+    after 1000 ->
+        ct:fail(wait_for_repub_disconnected_discarded)
+    end,
+
+    emqtt:stop(ClientRecv),
+    emqtt:stop(Client2),
+
+    delete_rule(TopicRule).
+
 client_connack_failed() ->
     {ok, Client} = emqtt:start_link(
         [

changes/v5.0.10-en.md

@@ -4,7 +4,8 @@
 
 - Improve `/nodes` API responsiveness [#9221](https://github.com/emqx/emqx/pull/9221).
 
-- Allow clear retained/delayed data when client is banned [#9139](https://github.com/emqx/emqx/pull/9139).
+- Improve the integration of the `banned` and the `delayed` feature [#9326](https://github.com/emqx/emqx/pull/9326).
+  Now when publishing a delayed message will check first if its source client is banned, if true, this publish will be ignored.
 
 - Update `gen_rpc` library to version 3.0 [#9187](https://github.com/emqx/emqx/pull/9187).
 
@@ -15,7 +16,13 @@
 - Now it is possible to opt out VM internal metrics in prometheus stats [#9222](https://github.com/emqx/emqx/pull/9222).
   When system load is high, reporting too much metrics data may cause the prometheus stats API timeout.
 
-- Improve security when converting types such as `binary` `lists` to `atom` types [#9279](https://github.com/emqx/emqx/pull/9279).
+- Improve security when converting types such as `binary` `lists` to `atom` types [#9279](https://github.com/emqx/emqx/pull/9279), [#9286](https://github.com/emqx/emqx/pull/9286).
+
+- Add `/trace/:name/log_detail` HTTP API to return trace file's size and mtime [#9152](https://github.com/emqx/emqx/pull/9152).
+
+- Add `/status` HTTP API endpoint to api documentation [#9230](https://github.com/emqx/emqx/pull/9230).
+
+- Binary packages for all platforms are now built on Erlang/OTP version 24.3.4.2 [#9293](https://github.com/emqx/emqx/pull/9293).
 
 ## Bug fixes
 
@@ -33,4 +40,16 @@
 
 - Fix bad HTTP response status code for `/gateways` API, when Gateway name is unknown, it should return `404` instead of `400` [#9268](https://github.com/emqx/emqx/pull/9268).
 
+
 - Fix `ssl.existingName` option of  helm chart not working [#9307](https://github.com/emqx/emqx/issues/9307).
+
+- Fix incorrect topic authorize checking of delayed messages [#9290](https://github.com/emqx/emqx/pull/9290).
+  Now will determine the actual topic of the delayed messages, e.g. `$delayed/1/t/foo` will be treated as `t/foo` in authorize checks.
+
+- Add property `code` to error response for `/authentication/sources/:type` [9299](https://github.com/emqx/emqx/pull/9299).
+
+- Align documentation for `/authentication/sources` with what we actually send [9299](https://github.com/emqx/emqx/pull/9299).
+
+- Fix query string parameter 'node' to `/configs` resource being ignored, return 404 if node does not exist [#9310](https://github.com/emqx/emqx/pull/9310/).
+
+- Avoid re-dispatching shared-subscription session messages when a session is kicked or taken-over (to a new session) [#9123](https://github.com/emqx/emqx/pull/9123).

changes/v5.0.10-zh.md

@@ -4,7 +4,8 @@
 
 - 提升 `/nodes` API 响应速度 [#9221](https://github.com/emqx/emqx/pull/9221)。
 
-- 支持拉黑客户端并从数据库中删除保留和延迟发布的消息 [#9139](https://github.com/emqx/emqx/pull/9139)。
+- 增强 `封禁` 和 `延迟消息` 这两个功能的集成性 [#9326](https://github.com/emqx/emqx/pull/9326)。
+  现在发送延迟消息前，会先检查消息的来源客户端是否被封禁了，如果是，这条延迟消息将会被忽略。
 
 - 升级 `gen_rpc` 库到 3.0 [#9187](https://github.com/emqx/emqx/pull/9187)。
 
@@ -14,7 +15,13 @@
 
 - 可通过配置关闭 prometheus 中的部分内部指标，如果遇到机器负载过高 prometheus 接口返回超时可考虑关闭部分不关心指标，以提高响应速度 [#9222](https://github.com/emqx/emqx/pull/9222）。
 
-- 提升 `binary` 、`list` 等类型转换为 `atom` 类型时的安全性 [#9279](https://github.com/emqx/emqx/pull/9279)。
+- 提升 `binary` 、`list` 等类型转换为 `atom` 类型时的安全性 [#9279](https://github.com/emqx/emqx/pull/9279)，[#9286](https://github.com/emqx/emqx/pull/9286)。
+
+- 增加了 `/trace/:name/log_detail` HTTP API 用于返回 trace 文件的大小和修改日期等信息 [#9152](https://github.com/emqx/emqx/pull/9152)。
+
+- HTTP API 文档中增加 `/status` 端点的描述 [#9230](https://github.com/emqx/emqx/pull/9230)。
+
+- 为所有平台的二进制包升级了 Erlang/OTP 到 24.3.4.2 [#9293](https://github.com/emqx/emqx/pull/9293)。
 
 ## Bug fixes
 
@@ -31,4 +38,17 @@
 
 - 修复 HTTP API `/gateways` 的返回状态码，未知 Gateway 名字应返回 `404` 而不是 `400` [#9268](https://github.com/emqx/emqx/pull/9268)。
 
+
 - 修复 helm chart 的 `ssl.existingName` 选项不起作用 [#9307](https://github.com/emqx/emqx/issues/9307)。
+
+- 修复延迟消息的主题授权判断不正确的问题 [#9290](https://github.com/emqx/emqx/pull/9290)。
+  现在将会对延迟消息中的真实主题进行授权判断，比如，`$delayed/1/t/foo` 会被当作 `t/foo` 进行判断。
+
+
+- 为 API `/authentication/sources/:type` 的返回值增加 `code` 字段 [9299](https://github.com/emqx/emqx/pull/9299)。
+
+- 对齐文档，`/authentication/sources` 接口的文档仅列出已经支持的资源 [9299](https://github.com/emqx/emqx/pull/9299)。
+
+- 修复 `/configs` API 的 'node' 参数的问题，如果节点不存在，则返回 HTTP 状态码 404 [#9310](https://github.com/emqx/emqx/pull/9310/)。
+
+- 共享订阅的消息在会话被踢出或者迁移时，不向其他订阅组成员进行转发 [#9123](https://github.com/emqx/emqx/pull/9123)。

changes/v5.0.11-en.md

@@ -0,0 +1,6 @@
+# v5.0.11
+
+## Enhancements
+
+## Bug fixes
+

changes/v5.0.11-zh.md

@@ -0,0 +1,5 @@
+# v5.0.11
+
+## 增强
+
+## 修复

scripts/check-nl-at-eof.sh

@@ -13,6 +13,9 @@ nl_at_eof() {
         *.png|*rebar3)
             return
             ;;
+        scripts/erlfmt)
+            return
+            ;;
     esac
     local lastbyte
     lastbyte="$(tail -c 1 "$file" 2>&1)"

scripts/ct/run.sh

@@ -89,8 +89,13 @@ for file in "${FILES[@]}"; do
     F_OPTIONS="$F_OPTIONS -f $file"
 done
 
+# Passing $UID to docker-compose to be used in erlang container
+# as owner of the main process to avoid git repo permissions issue.
+# Permissions issue happens because we are mounting local filesystem
+# where files are owned by $UID to docker container where it's using
+# root (UID=0) by default, and git is not happy about it.
 # shellcheck disable=2086 # no quotes for F_OPTIONS
-docker-compose $F_OPTIONS up -d --build
+UID_GID="$UID:$UID" docker-compose $F_OPTIONS up -d --build
 
 # /emqx is where the source dir is mounted to the Erlang container
 # in .ci/docker-compose-file/docker-compose.yaml
@@ -98,8 +103,11 @@ TTY=''
 if [[ -t 1 ]]; then
     TTY='-t'
 fi
-docker exec -i $TTY "$ERLANG_CONTAINER" bash -c 'git config --global --add safe.directory /emqx'
 
+# rebar and hex cache directory need to be writable by $UID
+docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "mkdir /.cache && chown $UID:$UID /.cache"
+# need to initialize .erlang.cookie manually here because / is not writable by $UID
+docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "openssl rand -base64 16 > /.erlang.cookie && chown $UID:$UID /.erlang.cookie && chmod 0400 /.erlang.cookie"
 if [ "$CONSOLE" = 'yes' ]; then
     docker exec -i $TTY "$ERLANG_CONTAINER" bash -c "make run"
 else
@@ -107,6 +115,6 @@ else
     docker exec -i $TTY "$ERLANG_CONTAINER" bash -c "make ${WHICH_APP}-ct"
     RESULT=$?
     # shellcheck disable=2086 # no quotes for F_OPTIONS
-    docker-compose $F_OPTIONS down
+    UID_GID="$UID:$UID" docker-compose $F_OPTIONS down
     exit $RESULT
 fi