Merge pull request #11472 from thalesmg/sync-r52-master-20230817

sync `release-52` to `master`

.ci/docker-compose-file/.env

@@ -10,6 +10,8 @@ CASSANDRA_TAG=3.11.6
 MINIO_TAG=RELEASE.2023-03-20T20-16-18Z
 OPENTS_TAG=9aa7f88
 KINESIS_TAG=2.1
+HSTREAMDB_TAG=v0.15.0
+HSTREAMDB_ZK_TAG=3.8.1
 
 MS_IMAGE_ADDR=mcr.microsoft.com/mssql/server
 SQLSERVER_TAG=2019-CU19-ubuntu-20.04

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

@@ -2,11 +2,13 @@ version: "3.5"
 
 services:
   hserver:
-    image: hstreamdb/hstream:v0.15.0
+    image: hstreamdb/hstream:${HSTREAMDB_TAG}
     container_name: hstreamdb
     depends_on:
-      - zookeeper
-      - hstore
+      zookeeper:
+        condition: service_started
+      hstore:
+        condition: service_healthy
     # ports:
     #   - "127.0.0.1:6570:6570"
     expose:
@@ -37,7 +39,7 @@ services:
         --io-tasks-network emqx_bridge
 
   hstore:
-    image: hstreamdb/hstream:v0.15.0
+    image: hstreamdb/hstream:${HSTREAMDB_TAG}
     networks:
       - emqx_bridge
     volumes:
@@ -53,10 +55,17 @@ services:
         --use-tcp --tcp-host $$(hostname -I | awk '{print $$1}') \
         --user-admin-port 6440 \
         --param enable-dscp-reflection=false \
-        --no-interactive
+        --no-interactive \
+        > /data/store/hstore.log 2>&1
+    healthcheck:
+      test: ["CMD", "grep", "LogDevice Cluster running", "/data/store/hstore.log"]
+      interval: 10s
+      timeout: 10s
+      retries: 60
+      start_period: 60s
 
   zookeeper:
-    image: zookeeper
+    image: zookeeper:${HSTREAMDB_ZK_TAG}
     expose:
       - 2181
     networks:

Makefile

@@ -16,7 +16,7 @@ endif
 # Dashboard version
 # from https://github.com/emqx/emqx-dashboard5
 export EMQX_DASHBOARD_VERSION ?= v1.3.2
-export EMQX_EE_DASHBOARD_VERSION ?= e1.1.1
+export EMQX_EE_DASHBOARD_VERSION ?= e1.2.0-beta.4
 
 # `:=` should be used here, otherwise the `$(shell ...)` will be executed every time when the variable is used
 # In make 4.4+, for backward-compatibility the value from the original environment is used.

apps/emqx/include/emqx_release.hrl

@@ -35,7 +35,7 @@
 -define(EMQX_RELEASE_CE, "5.1.5-build.3").
 
 %% Enterprise edition
--define(EMQX_RELEASE_EE, "5.2.0-alpha.1").
+-define(EMQX_RELEASE_EE, "5.2.0-alpha.3").
 
 %% The HTTP API version
 -define(EMQX_API_VERSION, "5.0").

apps/emqx/src/emqx_topic_index.erl

@@ -0,0 +1,242 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 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.
+%%--------------------------------------------------------------------
+
+%% @doc Topic index for matching topics to topic filters.
+%%
+%% Works on top of ETS ordered_set table. Keys are tuples constructed from
+%% parsed topic filters and record IDs, wrapped in a tuple to order them
+%% strictly greater than unit tuple (`{}`). Existing table may be used if
+%% existing keys will not collide with index keys.
+%%
+%% Designed to effectively answer questions like:
+%% 1. Does any topic filter match given topic?
+%% 2. Which records are associated with topic filters matching given topic?
+%% 3. Which topic filters match given topic?
+%% 4. Which record IDs are associated with topic filters matching given topic?
+
+-module(emqx_topic_index).
+
+-export([new/0]).
+-export([insert/4]).
+-export([delete/3]).
+-export([match/2]).
+-export([matches/3]).
+
+-export([get_id/1]).
+-export([get_topic/1]).
+-export([get_record/2]).
+
+-type word() :: binary() | '+' | '#'.
+-type key(ID) :: {[word()], {ID}}.
+-type match(ID) :: key(ID).
+
+%% @doc Create a new ETS table suitable for topic index.
+%% Usable mostly for testing purposes.
+-spec new() -> ets:table().
+new() ->
+    ets:new(?MODULE, [public, ordered_set, {read_concurrency, true}]).
+
+%% @doc Insert a new entry into the index that associates given topic filter to given
+%% record ID, and attaches arbitrary record to the entry. This allows users to choose
+%% between regular and "materialized" indexes, for example.
+-spec insert(emqx_types:topic(), _ID, _Record, ets:table()) -> true.
+insert(Filter, ID, Record, Tab) ->
+    ets:insert(Tab, {{words(Filter), {ID}}, Record}).
+
+%% @doc Delete an entry from the index that associates given topic filter to given
+%% record ID. Deleting non-existing entry is not an error.
+-spec delete(emqx_types:topic(), _ID, ets:table()) -> true.
+delete(Filter, ID, Tab) ->
+    ets:delete(Tab, {words(Filter), {ID}}).
+
+%% @doc Match given topic against the index and return the first match, or `false` if
+%% no match is found.
+-spec match(emqx_types:topic(), ets:table()) -> match(_ID) | false.
+match(Topic, Tab) ->
+    {Words, RPrefix} = match_init(Topic),
+    match(Words, RPrefix, Tab).
+
+match(Words, RPrefix, Tab) ->
+    Prefix = lists:reverse(RPrefix),
+    match(ets:next(Tab, {Prefix, {}}), Prefix, Words, RPrefix, Tab).
+
+match(K, Prefix, Words, RPrefix, Tab) ->
+    case match_next(Prefix, K, Words) of
+        true ->
+            K;
+        skip ->
+            match(ets:next(Tab, K), Prefix, Words, RPrefix, Tab);
+        stop ->
+            false;
+        Matched ->
+            match_rest(Matched, Words, RPrefix, Tab)
+    end.
+
+match_rest([W1 | [W2 | _] = SLast], [W1 | [W2 | _] = Rest], RPrefix, Tab) ->
+    % NOTE
+    % Fast-forward through identical words in the topic and the last key suffixes.
+    % This should save us a few redundant `ets:next` calls at the cost of slightly
+    % more complex match patterns.
+    match_rest(SLast, Rest, [W1 | RPrefix], Tab);
+match_rest(SLast, [W | Rest], RPrefix, Tab) when is_list(SLast) ->
+    match(Rest, [W | RPrefix], Tab);
+match_rest(plus, [W | Rest], RPrefix, Tab) ->
+    % NOTE
+    % There's '+' in the key suffix, meaning we should consider 2 alternatives:
+    % 1. Match the rest of the topic as if there was '+' in the current position.
+    % 2. Skip this key and try to match the topic as it is.
+    case match(Rest, ['+' | RPrefix], Tab) of
+        Match = {_, _} ->
+            Match;
+        false ->
+            match(Rest, [W | RPrefix], Tab)
+    end;
+match_rest(_, [], _RPrefix, _Tab) ->
+    false.
+
+%% @doc Match given topic against the index and return _all_ matches.
+%% If `unique` option is given, return only unique matches by record ID.
+-spec matches(emqx_types:topic(), ets:table(), _Opts :: [unique]) -> [match(_ID)].
+matches(Topic, Tab, Opts) ->
+    {Words, RPrefix} = match_init(Topic),
+    AccIn =
+        case Opts of
+            [unique | _] -> #{};
+            [] -> []
+        end,
+    Matches = matches(Words, RPrefix, AccIn, Tab),
+    case Matches of
+        #{} -> maps:values(Matches);
+        _ -> Matches
+    end.
+
+matches(Words, RPrefix, Acc, Tab) ->
+    Prefix = lists:reverse(RPrefix),
+    matches(ets:next(Tab, {Prefix, {}}), Prefix, Words, RPrefix, Acc, Tab).
+
+matches(Words, RPrefix, K = {Filter, _}, Acc, Tab) ->
+    Prefix = lists:reverse(RPrefix),
+    case Prefix > Filter of
+        true ->
+            % NOTE: Prefix already greater than the last key seen, need to `ets:next/2`.
+            matches(ets:next(Tab, {Prefix, {}}), Prefix, Words, RPrefix, Acc, Tab);
+        false ->
+            % NOTE: Prefix is still less than or equal to the last key seen, reuse it.
+            matches(K, Prefix, Words, RPrefix, Acc, Tab)
+    end.
+
+matches(K, Prefix, Words, RPrefix, Acc, Tab) ->
+    case match_next(Prefix, K, Words) of
+        true ->
+            matches(ets:next(Tab, K), Prefix, Words, RPrefix, match_add(K, Acc), Tab);
+        skip ->
+            matches(ets:next(Tab, K), Prefix, Words, RPrefix, Acc, Tab);
+        stop ->
+            Acc;
+        Matched ->
+            % NOTE: Prserve next key on the stack to save on `ets:next/2` calls.
+            matches_rest(Matched, Words, RPrefix, K, Acc, Tab)
+    end.
+
+matches_rest([W1 | [W2 | _] = SLast], [W1 | [W2 | _] = Rest], RPrefix, K, Acc, Tab) ->
+    % NOTE
+    % Fast-forward through identical words in the topic and the last key suffixes.
+    % This should save us a few redundant `ets:next` calls at the cost of slightly
+    % more complex match patterns.
+    matches_rest(SLast, Rest, [W1 | RPrefix], K, Acc, Tab);
+matches_rest(SLast, [W | Rest], RPrefix, K, Acc, Tab) when is_list(SLast) ->
+    matches(Rest, [W | RPrefix], K, Acc, Tab);
+matches_rest(plus, [W | Rest], RPrefix, K, Acc, Tab) ->
+    % NOTE
+    % There's '+' in the key suffix, meaning we should accumulate all matches from
+    % each of 2 branches:
+    % 1. Match the rest of the topic as if there was '+' in the current position.
+    % 2. Skip this key and try to match the topic as it is.
+    NAcc = matches(Rest, ['+' | RPrefix], K, Acc, Tab),
+    matches(Rest, [W | RPrefix], K, NAcc, Tab);
+matches_rest(_, [], _RPrefix, _K, Acc, _Tab) ->
+    Acc.
+
+match_add(K = {_Filter, ID}, Acc = #{}) ->
+    % NOTE: ensuring uniqueness by record ID
+    Acc#{ID => K};
+match_add(K, Acc) ->
+    [K | Acc].
+
+match_next(Prefix, {Filter, _ID}, Suffix) ->
+    match_filter(Prefix, Filter, Suffix);
+match_next(_, '$end_of_table', _) ->
+    stop.
+
+match_filter([], [], []) ->
+    % NOTE: we matched the topic exactly
+    true;
+match_filter([], [], _Suffix) ->
+    % NOTE: we matched the prefix, but there may be more matches next
+    skip;
+match_filter([], ['#'], _Suffix) ->
+    % NOTE: naturally, '#' < '+', so this is already optimal for `match/2`
+    true;
+match_filter([], ['+' | _], _Suffix) ->
+    plus;
+match_filter([], [_H | _] = Rest, _Suffix) ->
+    Rest;
+match_filter([H | T1], [H | T2], Suffix) ->
+    match_filter(T1, T2, Suffix);
+match_filter([H1 | _], [H2 | _], _Suffix) when H2 > H1 ->
+    % NOTE: we're strictly past the prefix, no need to continue
+    stop.
+
+match_init(Topic) ->
+    case words(Topic) of
+        [W = <<"$", _/bytes>> | Rest] ->
+            % NOTE
+            % This will effectively skip attempts to match special topics to `#` or `+/...`.
+            {Rest, [W]};
+        Words ->
+            {Words, []}
+    end.
+
+%% @doc Extract record ID from the match.
+-spec get_id(match(ID)) -> ID.
+get_id({_Filter, {ID}}) ->
+    ID.
+
+%% @doc Extract topic (or topic filter) from the match.
+-spec get_topic(match(_ID)) -> emqx_types:topic().
+get_topic({Filter, _ID}) ->
+    emqx_topic:join(Filter).
+
+%% @doc Fetch the record associated with the match.
+%% NOTE: Only really useful for ETS tables where the record ID is the first element.
+-spec get_record(match(_ID), ets:table()) -> _Record.
+get_record(K, Tab) ->
+    ets:lookup_element(Tab, K, 2).
+
+%%
+
+-spec words(emqx_types:topic()) -> [word()].
+words(Topic) when is_binary(Topic) ->
+    % NOTE
+    % This is almost identical to `emqx_topic:words/1`, but it doesn't convert empty
+    % tokens to ''. This is needed to keep ordering of words consistent with what
+    % `match_filter/3` expects.
+    [word(W) || W <- emqx_topic:tokens(Topic)].
+
+-spec word(binary()) -> word().
+word(<<"+">>) -> '+';
+word(<<"#">>) -> '#';
+word(Bin) -> Bin.

apps/emqx/test/emqx_topic_index_SUITE.erl

@@ -0,0 +1,331 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 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_topic_index_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("proper/include/proper.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+-import(emqx_proper_types, [scaled/2]).
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+t_insert(_) ->
+    Tab = emqx_topic_index:new(),
+    true = emqx_topic_index:insert(<<"sensor/1/metric/2">>, t_insert_1, <<>>, Tab),
+    true = emqx_topic_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab),
+    true = emqx_topic_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab),
+    ?assertEqual(<<"sensor/#">>, topic(match(<<"sensor">>, Tab))),
+    ?assertEqual(t_insert_3, id(match(<<"sensor">>, Tab))).
+
+t_match(_) ->
+    Tab = emqx_topic_index:new(),
+    true = emqx_topic_index:insert(<<"sensor/1/metric/2">>, t_match_1, <<>>, Tab),
+    true = emqx_topic_index:insert(<<"sensor/+/#">>, t_match_2, <<>>, Tab),
+    true = emqx_topic_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab),
+    ?assertMatch(
+        [<<"sensor/#">>, <<"sensor/+/#">>],
+        [topic(M) || M <- matches(<<"sensor/1">>, Tab)]
+    ).
+
+t_match2(_) ->
+    Tab = emqx_topic_index:new(),
+    true = emqx_topic_index:insert(<<"#">>, t_match2_1, <<>>, Tab),
+    true = emqx_topic_index:insert(<<"+/#">>, t_match2_2, <<>>, Tab),
+    true = emqx_topic_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab),
+    ?assertEqual(
+        [<<"#">>, <<"+/#">>, <<"+/+/#">>],
+        [topic(M) || M <- matches(<<"a/b/c">>, Tab)]
+    ),
+    ?assertEqual(
+        false,
+        emqx_topic_index:match(<<"$SYS/broker/zenmq">>, Tab)
+    ).
+
+t_match3(_) ->
+    Tab = emqx_topic_index:new(),
+    Records = [
+        {<<"d/#">>, t_match3_1},
+        {<<"a/b/+">>, t_match3_2},
+        {<<"a/#">>, t_match3_3},
+        {<<"#">>, t_match3_4},
+        {<<"$SYS/#">>, t_match3_sys}
+    ],
+    lists:foreach(
+        fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end,
+        Records
+    ),
+    Matched = matches(<<"a/b/c">>, Tab),
+    case length(Matched) of
+        3 -> ok;
+        _ -> error({unexpected, Matched})
+    end,
+    ?assertEqual(
+        t_match3_sys,
+        id(match(<<"$SYS/a/b/c">>, Tab))
+    ).
+
+t_match4(_) ->
+    Tab = emqx_topic_index:new(),
+    Records = [{<<"/#">>, t_match4_1}, {<<"/+">>, t_match4_2}, {<<"/+/a/b/c">>, t_match4_3}],
+    lists:foreach(
+        fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end,
+        Records
+    ),
+    ?assertEqual(
+        [<<"/#">>, <<"/+">>],
+        [topic(M) || M <- matches(<<"/">>, Tab)]
+    ),
+    ?assertEqual(
+        [<<"/#">>, <<"/+/a/b/c">>],
+        [topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)]
+    ).
+
+t_match5(_) ->
+    Tab = emqx_topic_index:new(),
+    T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>,
+    Records = [
+        {<<"#">>, t_match5_1},
+        {<<T/binary, "/#">>, t_match5_2},
+        {<<T/binary, "/+">>, t_match5_3}
+    ],
+    lists:foreach(
+        fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end,
+        Records
+    ),
+    ?assertEqual(
+        [<<"#">>, <<T/binary, "/#">>],
+        [topic(M) || M <- matches(T, Tab)]
+    ),
+    ?assertEqual(
+        [<<"#">>, <<T/binary, "/#">>, <<T/binary, "/+">>],
+        [topic(M) || M <- matches(<<T/binary, "/1">>, Tab)]
+    ).
+
+t_match6(_) ->
+    Tab = emqx_topic_index:new(),
+    T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>,
+    W = <<"+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/#">>,
+    emqx_topic_index:insert(W, ID = t_match6, <<>>, Tab),
+    ?assertEqual(ID, id(match(T, Tab))).
+
+t_match7(_) ->
+    Tab = emqx_topic_index:new(),
+    T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>,
+    W = <<"a/+/c/+/e/+/g/+/i/+/k/+/m/+/o/+/q/+/s/+/u/+/w/+/y/+/#">>,
+    emqx_topic_index:insert(W, t_match7, <<>>, Tab),
+    ?assertEqual(W, topic(match(T, Tab))).
+
+t_match_fast_forward(_) ->
+    Tab = emqx_topic_index:new(),
+    emqx_topic_index:insert(<<"a/b/1/2/3/4/5/6/7/8/9/#">>, id1, <<>>, Tab),
+    emqx_topic_index:insert(<<"z/y/x/+/+">>, id2, <<>>, Tab),
+    emqx_topic_index:insert(<<"a/b/c/+">>, id3, <<>>, Tab),
+    % dbg:tracer(),
+    % dbg:p(all, c),
+    % dbg:tpl({ets, next, '_'}, x),
+    ?assertEqual(id1, id(match(<<"a/b/1/2/3/4/5/6/7/8/9/0">>, Tab))),
+    ?assertEqual([id1], [id(M) || M <- matches(<<"a/b/1/2/3/4/5/6/7/8/9/0">>, Tab)]).
+
+t_match_unique(_) ->
+    Tab = emqx_topic_index:new(),
+    emqx_topic_index:insert(<<"a/b/c">>, t_match_id1, <<>>, Tab),
+    emqx_topic_index:insert(<<"a/b/+">>, t_match_id1, <<>>, Tab),
+    emqx_topic_index:insert(<<"a/b/c/+">>, t_match_id2, <<>>, Tab),
+    ?assertEqual(
+        [t_match_id1, t_match_id1],
+        [id(M) || M <- emqx_topic_index:matches(<<"a/b/c">>, Tab, [])]
+    ),
+    ?assertEqual(
+        [t_match_id1],
+        [id(M) || M <- emqx_topic_index:matches(<<"a/b/c">>, Tab, [unique])]
+    ).
+
+t_match_wildcard_edge_cases(_) ->
+    CommonTopics = [
+        <<"a/b">>,
+        <<"a/b/#">>,
+        <<"a/b/#">>,
+        <<"a/b/c">>,
+        <<"a/b/+">>,
+        <<"a/b/d">>,
+        <<"a/+/+">>,
+        <<"a/+/#">>
+    ],
+    Datasets =
+        [
+            %% Topics, TopicName, Results
+            {CommonTopics, <<"a/b/c">>, [2, 3, 4, 5, 7, 8]},
+            {CommonTopics, <<"a/b">>, [1, 2, 3, 8]},
+            {[<<"+/b/c">>, <<"/">>], <<"a/b/c">>, [1]},
+            {[<<"#">>, <<"/">>], <<"a">>, [1]},
+            {[<<"/">>, <<"+">>], <<"a">>, [2]}
+        ],
+    F = fun({Topics, TopicName, Expected}) ->
+        Tab = emqx_topic_index:new(),
+        _ = [emqx_topic_index:insert(T, N, <<>>, Tab) || {N, T} <- lists:enumerate(Topics)],
+        ?assertEqual(
+            lists:last(Expected),
+            id(emqx_topic_index:match(TopicName, Tab)),
+            #{"Base topics" => Topics, "Topic name" => TopicName}
+        ),
+        ?assertEqual(
+            Expected,
+            [id(M) || M <- emqx_topic_index:matches(TopicName, Tab, [unique])],
+            #{"Base topics" => Topics, "Topic name" => TopicName}
+        )
+    end,
+    lists:foreach(F, Datasets).
+
+t_prop_matches(_) ->
+    ?assert(
+        proper:quickcheck(
+            topic_matches_prop(),
+            [{max_size, 100}, {numtests, 100}]
+        )
+    ),
+    Statistics = [{C, account(C)} || C <- [filters, topics, matches, maxhits]],
+    ct:pal("Statistics: ~p", [maps:from_list(Statistics)]).
+
+topic_matches_prop() ->
+    ?FORALL(
+        % Generate a longer list of topics and a shorter list of topic filter patterns.
+        #{
+            topics := TTopics,
+            patterns := Pats
+        },
+        emqx_proper_types:fixedmap(#{
+            % NOTE
+            % Beware adding non-empty contraint, proper will have a hard time with `topic_t/1`
+            % for some reason.
+            topics => scaled(4, list(topic_t([1, 2, 3, 4]))),
+            patterns => list(topic_filter_pattern_t())
+        }),
+        begin
+            Tab = emqx_topic_index:new(),
+            Topics = [emqx_topic:join(T) || T <- TTopics],
+            % Produce topic filters from generated topics and patterns.
+            % Number of filters is equal to the number of patterns, most of the time.
+            Filters = lists:enumerate(mk_filters(Pats, TTopics)),
+            _ = [emqx_topic_index:insert(F, N, <<>>, Tab) || {N, F} <- Filters],
+            % Gather some basic statistics
+            _ = account(filters, length(Filters)),
+            _ = account(topics, NTopics = length(Topics)),
+            _ = account(maxhits, NTopics * NTopics),
+            % Verify that matching each topic against index returns the same results as
+            % matching it against the list of filters one by one.
+            lists:all(
+                fun(Topic) ->
+                    Ids1 = [id(M) || M <- emqx_topic_index:matches(Topic, Tab, [unique])],
+                    Ids2 = lists:filtermap(
+                        fun({N, F}) ->
+                            case emqx_topic:match(Topic, F) of
+                                true -> {true, N};
+                                false -> false
+                            end
+                        end,
+                        Filters
+                    ),
+                    % Account a number of matches to compute hitrate later
+                    _ = account(matches, length(Ids1)),
+                    case (Ids2 -- Ids1) ++ (Ids2 -- Ids1) of
+                        [] ->
+                            true;
+                        [_ | _] = _Differences ->
+                            ct:pal(
+                                "Topic name: ~p~n"
+                                "Index results: ~p~n"
+                                "Topic match results:: ~p~n",
+                                [Topic, Ids1, Ids2]
+                            ),
+                            false
+                    end
+                end,
+                Topics
+            )
+        end
+    ).
+
+mk_filters([Pat | PRest], [Topic | TRest]) ->
+    [emqx_topic:join(mk_topic_filter(Pat, Topic)) | mk_filters(PRest, TRest)];
+mk_filters(_, _) ->
+    [].
+
+account(Counter, N) ->
+    put({?MODULE, Counter}, account(Counter) + N).
+
+account(Counter) ->
+    emqx_maybe:define(get({?MODULE, Counter}), 0).
+
+%%
+
+match(T, Tab) ->
+    emqx_topic_index:match(T, Tab).
+
+matches(T, Tab) ->
+    lists:sort(emqx_topic_index:matches(T, Tab, [])).
+
+id(Match) ->
+    emqx_topic_index:get_id(Match).
+
+topic(Match) ->
+    emqx_topic_index:get_topic(Match).
+
+%%
+
+topic_t(EntropyWeights) ->
+    EWLast = lists:last(EntropyWeights),
+    ?LET(L, scaled(1 / 4, list(EWLast)), begin
+        EWs = lists:sublist(EntropyWeights ++ L, length(L)),
+        ?SIZED(S, [oneof([topic_level_t(S * EW), topic_level_fixed_t()]) || EW <- EWs])
+    end).
+
+topic_level_t(Entropy) ->
+    S = floor(1 + math:log2(Entropy) / 4),
+    ?LET(I, range(1, Entropy), iolist_to_binary(io_lib:format("~*.16.0B", [S, I]))).
+
+topic_level_fixed_t() ->
+    oneof([
+        <<"foo">>,
+        <<"bar">>,
+        <<"baz">>,
+        <<"xyzzy">>
+    ]).
+
+topic_filter_pattern_t() ->
+    list(topic_level_pattern_t()).
+
+topic_level_pattern_t() ->
+    frequency([
+        {5, level},
+        {2, '+'},
+        {1, '#'}
+    ]).
+
+mk_topic_filter([], _) ->
+    [];
+mk_topic_filter(_, []) ->
+    [];
+mk_topic_filter(['#' | _], _) ->
+    ['#'];
+mk_topic_filter(['+' | Rest], [_ | Levels]) ->
+    ['+' | mk_topic_filter(Rest, Levels)];
+mk_topic_filter([level | Rest], [L | Levels]) ->
+    [L | mk_topic_filter(Rest, Levels)].

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_client.erl

@@ -205,7 +205,7 @@ get_topic(Topic, ConnectorState) ->
     Path = <<"/v1/projects/", ProjectId/binary, "/topics/", Topic/binary>>,
     Body = <<>>,
     PreparedRequest = {prepared_request, {Method, Path, Body}},
-    query_sync(PreparedRequest, ConnectorState).
+    ?MODULE:query_sync(PreparedRequest, ConnectorState).
 
 %%-------------------------------------------------------------------------------------------------
 %% Helper fns

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_consumer_worker.erl

@@ -217,7 +217,9 @@ handle_continue(?ensure_subscription, State0) ->
             {noreply, State0, {continue, ?ensure_subscription}};
         not_found ->
             %% there's nothing much to do if the topic suddenly doesn't exist anymore.
-            {stop, {error, topic_not_found}, State0}
+            {stop, {error, topic_not_found}, State0};
+        permission_denied ->
+            {stop, {error, permission_denied}, State0}
     end;
 handle_continue(?patch_subscription, State0) ->
     ?tp(gcp_pubsub_consumer_worker_patch_subscription_enter, #{}),
@@ -291,14 +293,17 @@ handle_info(Msg, State0) ->
     }),
     {noreply, State0}.
 
-terminate({error, topic_not_found} = _Reason, State) ->
+terminate({error, Reason}, State) when
+    Reason =:= topic_not_found;
+    Reason =:= permission_denied
+->
     #{
         instance_id := InstanceId,
         topic := _Topic
     } = State,
     optvar:unset(?OPTVAR_SUB_OK(self())),
-    emqx_bridge_gcp_pubsub_impl_consumer:mark_topic_as_nonexistent(InstanceId),
-    ?tp(gcp_pubsub_consumer_worker_terminate, #{reason => _Reason, topic => _Topic}),
+    emqx_bridge_gcp_pubsub_impl_consumer:mark_as_unhealthy(InstanceId, Reason),
+    ?tp(gcp_pubsub_consumer_worker_terminate, #{reason => {error, Reason}, topic => _Topic}),
     ok;
 terminate(_Reason, _State) ->
     optvar:unset(?OPTVAR_SUB_OK(self())),
@@ -329,7 +334,8 @@ ensure_pull_timer(State = #{pull_timer := TRef}) when is_reference(TRef) ->
 ensure_pull_timer(State = #{pull_retry_interval := PullRetryInterval}) ->
     State#{pull_timer := emqx_utils:start_timer(PullRetryInterval, pull)}.
 
--spec ensure_subscription_exists(state()) -> continue | retry | not_found | already_exists.
+-spec ensure_subscription_exists(state()) ->
+    continue | retry | not_found | permission_denied | already_exists.
 ensure_subscription_exists(State) ->
     ?tp(gcp_pubsub_consumer_worker_create_subscription_enter, #{}),
     #{
@@ -367,6 +373,17 @@ ensure_subscription_exists(State) ->
                 }
             ),
             not_found;
+        {error, #{status_code := 403}} ->
+            %% permission denied
+            ?tp(
+                warning,
+                "gcp_pubsub_consumer_worker_permission_denied",
+                #{
+                    instance_id => InstanceId,
+                    topic => Topic
+                }
+            ),
+            permission_denied;
         {ok, #{status_code := 200}} ->
             ?tp(
                 debug,

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_impl_consumer.erl

@@ -17,9 +17,9 @@
 
 %% health check API
 -export([
-    mark_topic_as_nonexistent/1,
-    unset_nonexistent_topic/1,
-    is_nonexistent_topic/1
+    mark_as_unhealthy/2,
+    clear_unhealthy/1,
+    check_if_unhealthy/1
 ]).
 
 -include_lib("emqx/include/logger.hrl").
@@ -47,11 +47,15 @@
 
 -define(AUTO_RECONNECT_S, 2).
 -define(DEFAULT_FORGET_INTERVAL, timer:seconds(60)).
--define(OPTVAR_TOPIC_NOT_FOUND(INSTANCE_ID), {?MODULE, topic_not_found, INSTANCE_ID}).
+-define(OPTVAR_UNHEALTHY(INSTANCE_ID), {?MODULE, topic_not_found, INSTANCE_ID}).
 -define(TOPIC_MESSAGE,
     "GCP PubSub topics are invalid.  Please check the logs, check if the "
     "topics exist in GCP and if the service account has permissions to use them."
 ).
+-define(PERMISSION_MESSAGE,
+    "Permission denied while verifying topic existence.  Please check that the "
+    "provided service account has the correct permissions configured."
+).
 
 %%-------------------------------------------------------------------------------------------------
 %% `emqx_resource' API
@@ -77,7 +81,7 @@ on_start(InstanceId, Config0) ->
 -spec on_stop(resource_id(), state()) -> ok | {error, term()}.
 on_stop(InstanceId, _State) ->
     ?tp(gcp_pubsub_consumer_stop_enter, #{}),
-    unset_nonexistent_topic(InstanceId),
+    clear_unhealthy(InstanceId),
     ok = stop_consumers(InstanceId),
     emqx_bridge_gcp_pubsub_client:stop(InstanceId).
 
@@ -85,10 +89,12 @@ on_stop(InstanceId, _State) ->
 on_get_status(InstanceId, State) ->
     %% We need to check this flag separately because the workers might be gone when we
     %% check them.
-    case is_nonexistent_topic(InstanceId) of
-        true ->
+    case check_if_unhealthy(InstanceId) of
+        {error, topic_not_found} ->
             {disconnected, State, {unhealthy_target, ?TOPIC_MESSAGE}};
-        false ->
+        {error, permission_denied} ->
+            {disconnected, State, {unhealthy_target, ?PERMISSION_MESSAGE}};
+        ok ->
             #{client := Client} = State,
             check_workers(InstanceId, Client)
     end.
@@ -97,24 +103,24 @@ on_get_status(InstanceId, State) ->
 %% Health check API (signalled by consumer worker)
 %%-------------------------------------------------------------------------------------------------
 
--spec mark_topic_as_nonexistent(resource_id()) -> ok.
-mark_topic_as_nonexistent(InstanceId) ->
-    optvar:set(?OPTVAR_TOPIC_NOT_FOUND(InstanceId), true),
+-spec mark_as_unhealthy(resource_id(), topic_not_found | permission_denied) -> ok.
+mark_as_unhealthy(InstanceId, Reason) ->
+    optvar:set(?OPTVAR_UNHEALTHY(InstanceId), Reason),
     ok.
 
--spec unset_nonexistent_topic(resource_id()) -> ok.
-unset_nonexistent_topic(InstanceId) ->
-    optvar:unset(?OPTVAR_TOPIC_NOT_FOUND(InstanceId)),
-    ?tp(gcp_pubsub_consumer_unset_nonexistent_topic, #{}),
+-spec clear_unhealthy(resource_id()) -> ok.
+clear_unhealthy(InstanceId) ->
+    optvar:unset(?OPTVAR_UNHEALTHY(InstanceId)),
+    ?tp(gcp_pubsub_consumer_clear_unhealthy, #{}),
     ok.
 
--spec is_nonexistent_topic(resource_id()) -> boolean().
-is_nonexistent_topic(InstanceId) ->
-    case optvar:peek(?OPTVAR_TOPIC_NOT_FOUND(InstanceId)) of
-        {ok, true} ->
-            true;
-        _ ->
-            false
+-spec check_if_unhealthy(resource_id()) -> ok | {error, topic_not_found | permission_denied}.
+check_if_unhealthy(InstanceId) ->
+    case optvar:peek(?OPTVAR_UNHEALTHY(InstanceId)) of
+        {ok, Reason} ->
+            {error, Reason};
+        undefined ->
+            ok
     end.
 
 %%-------------------------------------------------------------------------------------------------
@@ -153,6 +159,11 @@ start_consumers(InstanceId, Client, Config) ->
             throw(
                 {unhealthy_target, ?TOPIC_MESSAGE}
             );
+        {error, permission_denied} ->
+            _ = emqx_bridge_gcp_pubsub_client:stop(InstanceId),
+            throw(
+                {unhealthy_target, ?PERMISSION_MESSAGE}
+            );
         {error, _} ->
             %% connection might be down; we'll have to check topic existence during health
             %% check, or the workers will kill themselves when they realized there's no
@@ -229,6 +240,8 @@ check_for_topic_existence(Topic, Client) ->
             ok;
         {error, #{status_code := 404}} ->
             {error, not_found};
+        {error, #{status_code := 403}} ->
+            {error, permission_denied};
         {error, Reason} ->
             ?tp(warning, "gcp_pubsub_consumer_check_topic_error", #{reason => Reason}),
             {error, Reason}

apps/emqx_bridge_gcp_pubsub/test/emqx_bridge_gcp_pubsub_consumer_SUITE.erl

@@ -760,6 +760,64 @@ prop_acked_ids_eventually_forgotten(Trace) ->
     ),
     ok.
 
+permission_denied_response() ->
+    Link =
+        <<"https://console.developers.google.com/project/9999/apiui/credential">>,
+    {error, #{
+        status_code => 403,
+        headers =>
+            [
+                {<<"vary">>, <<"X-Origin">>},
+                {<<"vary">>, <<"Referer">>},
+                {<<"content-type">>, <<"application/json; charset=UTF-8">>},
+                {<<"date">>, <<"Tue, 15 Aug 2023 13:59:09 GMT">>},
+                {<<"server">>, <<"ESF">>},
+                {<<"cache-control">>, <<"private">>},
+                {<<"x-xss-protection">>, <<"0">>},
+                {<<"x-frame-options">>, <<"SAMEORIGIN">>},
+                {<<"x-content-type-options">>, <<"nosniff">>},
+                {<<"alt-svc">>, <<"h3=\":443\"; ma=2592000,h3-29=\":443\"; ma=2592000">>},
+                {<<"accept-ranges">>, <<"none">>},
+                {<<"vary">>, <<"Origin,Accept-Encoding">>},
+                {<<"transfer-encoding">>, <<"chunked">>}
+            ],
+        body => emqx_utils_json:encode(
+            #{
+                <<"error">> =>
+                    #{
+                        <<"code">> => 403,
+                        <<"details">> =>
+                            [
+                                #{
+                                    <<"@type">> => <<"type.googleapis.com/google.rpc.Help">>,
+                                    <<"links">> =>
+                                        [
+                                            #{
+                                                <<"description">> =>
+                                                    <<"Google developer console API key">>,
+                                                <<"url">> =>
+                                                    Link
+                                            }
+                                        ]
+                                },
+                                #{
+                                    <<"@type">> => <<"type.googleapis.com/google.rpc.ErrorInfo">>,
+                                    <<"domain">> => <<"googleapis.com">>,
+                                    <<"metadata">> =>
+                                        #{
+                                            <<"consumer">> => <<"projects/9999">>,
+                                            <<"service">> => <<"pubsub.googleapis.com">>
+                                        },
+                                    <<"reason">> => <<"CONSUMER_INVALID">>
+                                }
+                            ],
+                        <<"message">> => <<"Project #9999 has been deleted.">>,
+                        <<"status">> => <<"PERMISSION_DENIED">>
+                    }
+            }
+        )
+    }}.
+
 %%------------------------------------------------------------------------------
 %% Testcases
 %%------------------------------------------------------------------------------
@@ -785,7 +843,7 @@ t_start_stop(Config) ->
             prop_client_stopped(),
             prop_workers_stopped(PubSubTopic),
             fun(Trace) ->
-                ?assertMatch([_], ?of_kind(gcp_pubsub_consumer_unset_nonexistent_topic, Trace)),
+                ?assertMatch([_], ?of_kind(gcp_pubsub_consumer_clear_unhealthy, Trace)),
                 ok
             end
         ]
@@ -1992,6 +2050,81 @@ t_get_subscription(Config) ->
     ),
     ok.
 
+t_permission_denied_topic_check(Config) ->
+    [#{pubsub_topic := PubSubTopic}] = ?config(topic_mapping, Config),
+    ResourceId = resource_id(Config),
+    ?check_trace(
+        begin
+            %% the emulator does not check any credentials
+            emqx_common_test_helpers:with_mock(
+                emqx_bridge_gcp_pubsub_client,
+                query_sync,
+                fun(PreparedRequest = {prepared_request, {Method, Path, _Body}}, Client) ->
+                    RE = iolist_to_binary(["/topics/", PubSubTopic, "$"]),
+                    case {Method =:= get, re:run(Path, RE)} of
+                        {true, {match, _}} ->
+                            permission_denied_response();
+                        _ ->
+                            meck:passthrough([PreparedRequest, Client])
+                    end
+                end,
+                fun() ->
+                    {{ok, _}, {ok, _}} =
+                        ?wait_async_action(
+                            create_bridge(Config),
+                            #{?snk_kind := gcp_pubsub_stop},
+                            5_000
+                        ),
+                    ?assertMatch(
+                        {ok, disconnected},
+                        emqx_resource_manager:health_check(ResourceId)
+                    ),
+                    ?assertMatch(
+                        {ok, _Group, #{error := {unhealthy_target, "Permission denied" ++ _}}},
+                        emqx_resource_manager:lookup_cached(ResourceId)
+                    ),
+                    ok
+                end
+            ),
+            ok
+        end,
+        []
+    ),
+    ok.
+
+t_permission_denied_worker(Config) ->
+    ?check_trace(
+        begin
+            emqx_common_test_helpers:with_mock(
+                emqx_bridge_gcp_pubsub_client,
+                query_sync,
+                fun(PreparedRequest = {prepared_request, {Method, _Path, _Body}}, Client) ->
+                    case Method =:= put of
+                        true ->
+                            permission_denied_response();
+                        false ->
+                            meck:passthrough([PreparedRequest, Client])
+                    end
+                end,
+                fun() ->
+                    {{ok, _}, {ok, _}} =
+                        ?wait_async_action(
+                            create_bridge(
+                                Config
+                            ),
+                            #{?snk_kind := gcp_pubsub_consumer_worker_terminate},
+                            10_000
+                        ),
+
+                    ok
+                end
+            ),
+            ok
+        end,
+        []
+    ),
+    ok.
+
 t_cluster_subscription(Config) ->
     [
         #{

apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_influxdb, [
     {description, "EMQX Enterprise InfluxDB Bridge"},
-    {vsn, "0.1.3"},
+    {vsn, "0.1.4"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb.erl

@@ -168,6 +168,9 @@ write_syntax(format) ->
 write_syntax(_) ->
     undefined.
 
+to_influx_lines(Lines = [#{} | _]) ->
+    %% already parsed/converted (e.g.: bridge_probe, after hocon_tconf:check_plain)
+    Lines;
 to_influx_lines(RawLines) ->
     try
         influx_lines(str(RawLines), [])

apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb_connector.erl

@@ -66,7 +66,9 @@ on_start(InstId, Config) ->
 on_stop(InstId, _State) ->
     case emqx_resource:get_allocated_resources(InstId) of
         #{?influx_client := Client} ->
-            influxdb:stop_client(Client);
+            Res = influxdb:stop_client(Client),
+            ?tp(influxdb_client_stopped, #{instance_id => InstId}),
+            Res;
         _ ->
             ok
     end.

apps/emqx_bridge_influxdb/test/emqx_bridge_influxdb_SUITE.erl

@@ -124,6 +124,9 @@ init_per_group(InfluxDBType, Config0) when
                 {influxdb_config, InfluxDBConfig},
                 {influxdb_config_string, ConfigString},
                 {ehttpc_pool_name, EHttpcPoolName},
+                {bridge_type, influxdb_api_v1},
+                {bridge_name, Name},
+                {bridge_config, InfluxDBConfig},
                 {influxdb_name, Name}
                 | Config
             ];
@@ -193,6 +196,9 @@ init_per_group(InfluxDBType, Config0) when
                 {influxdb_config, InfluxDBConfig},
                 {influxdb_config_string, ConfigString},
                 {ehttpc_pool_name, EHttpcPoolName},
+                {bridge_type, influxdb_api_v2},
+                {bridge_name, Name},
+                {bridge_config, InfluxDBConfig},
                 {influxdb_name, Name}
                 | Config
             ];
@@ -570,6 +576,10 @@ t_start_ok(Config) ->
     ),
     ok.
 
+t_start_stop(Config) ->
+    ok = emqx_bridge_testlib:t_start_stop(Config, influxdb_client_stopped),
+    ok.
+
 t_start_already_started(Config) ->
     Type = influxdb_type_bin(?config(influxdb_type, Config)),
     Name = ?config(influxdb_name, Config),

apps/emqx_bridge_kafka/src/emqx_bridge_kafka.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge_kafka, [
     {description, "EMQX Enterprise Kafka Bridge"},
-    {vsn, "0.1.7"},
+    {vsn, "0.1.8"},
     {registered, [emqx_bridge_kafka_consumer_sup]},
     {applications, [
         kernel,

apps/emqx_bridge_kafka/src/emqx_bridge_kafka.erl

@@ -268,7 +268,8 @@ fields(producer_opts) ->
                 required => true,
                 desc => ?DESC(producer_kafka_opts),
                 validator => fun producer_strategy_key_validator/1
-            })}
+            })},
+        {resource_opts, mk(ref(resource_opts), #{default => #{}})}
     ];
 fields(producer_kafka_opts) ->
     [
@@ -425,7 +426,8 @@ fields(consumer_opts) ->
         {value_encoding_mode,
             mk(enum([none, base64]), #{
                 default => none, desc => ?DESC(consumer_value_encoding_mode)
-            })}
+            })},
+        {resource_opts, mk(ref(resource_opts), #{default => #{}})}
     ];
 fields(consumer_topic_mapping) ->
     [
@@ -460,10 +462,16 @@ fields(consumer_kafka_opts) ->
                 emqx_schema:timeout_duration_s(),
                 #{default => <<"5s">>, desc => ?DESC(consumer_offset_commit_interval_seconds)}
             )}
-    ].
+    ];
+fields(resource_opts) ->
+    SupportedFields = [health_check_interval],
+    CreationOpts = emqx_resource_schema:create_opts(_Overrides = []),
+    lists:filter(fun({Field, _}) -> lists:member(Field, SupportedFields) end, CreationOpts).
 
 desc("config") ->
     ?DESC("desc_config");
+desc(resource_opts) ->
+    ?DESC(emqx_resource_schema, "resource_opts");
 desc("get_" ++ Type) when Type =:= "consumer"; Type =:= "producer" ->
     ["Configuration for Kafka using `GET` method."];
 desc("put_" ++ Type) when Type =:= "consumer"; Type =:= "producer" ->

apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_producer_SUITE.erl

@@ -596,7 +596,6 @@ t_send_message_with_headers(Config) ->
         },
         KafkaMsg
     ),
-    ?assertMatch(#kafka_message{key = BinTime}, KafkaMsg),
     %% TODO: refactor those into init/end per testcase
     ok = ?PRODUCER:on_stop(ResourceId, State),
     ?assertEqual([], supervisor:which_children(wolff_client_sup)),

apps/emqx_bridge_kafka/test/emqx_bridge_kafka_tests.erl

@@ -306,6 +306,9 @@ kafka_producer_new_hocon() ->
     "      sndbuf = \"1024KB\"\n"
     "    }\n"
     "    ssl {enable = false, verify = \"verify_peer\"}\n"
+    "    resource_opts {\n"
+    "      health_check_interval = 10s\n"
+    "    }\n"
     "  }\n"
     "}\n"
     "".
@@ -351,5 +354,8 @@ bridges.kafka_consumer.my_consumer {
     verify = verify_none
     server_name_indication = \"auto\"
   }
+  resource_opts {
+    health_check_interval = 10s
+  }
 }
 """.

apps/emqx_rule_engine/include/rule_engine.hrl

@@ -109,6 +109,7 @@
 
 %% Tables
 -define(RULE_TAB, emqx_rule_engine).
+-define(RULE_TOPIC_INDEX, emqx_rule_engine_topic_index).
 
 %% Allowed sql function provider modules
 -define(DEFAULT_SQL_FUNC_PROVIDER, emqx_rule_funcs).

apps/emqx_rule_engine/src/emqx_rule_engine.erl

@@ -176,7 +176,7 @@ create_rule(Params) ->
 
 create_rule(Params = #{id := RuleId}, CreatedAt) when is_binary(RuleId) ->
     case get_rule(RuleId) of
-        not_found -> parse_and_insert(Params, CreatedAt);
+        not_found -> with_parsed_rule(Params, CreatedAt, fun insert_rule/1);
         {ok, _} -> {error, already_exists}
     end.
 
@@ -185,18 +185,27 @@ update_rule(Params = #{id := RuleId}) when is_binary(RuleId) ->
     case get_rule(RuleId) of
         not_found ->
             {error, not_found};
-        {ok, #{created_at := CreatedAt}} ->
-            parse_and_insert(Params, CreatedAt)
+        {ok, RulePrev = #{created_at := CreatedAt}} ->
+            with_parsed_rule(Params, CreatedAt, fun(Rule) -> update_rule(Rule, RulePrev) end)
     end.
 
 -spec delete_rule(RuleId :: rule_id()) -> ok.
 delete_rule(RuleId) when is_binary(RuleId) ->
-    gen_server:call(?RULE_ENGINE, {delete_rule, RuleId}, ?T_CALL).
+    case get_rule(RuleId) of
+        not_found ->
+            ok;
+        {ok, Rule} ->
+            gen_server:call(?RULE_ENGINE, {delete_rule, Rule}, ?T_CALL)
+    end.
 
 -spec insert_rule(Rule :: rule()) -> ok.
 insert_rule(Rule) ->
     gen_server:call(?RULE_ENGINE, {insert_rule, Rule}, ?T_CALL).
 
+-spec update_rule(Rule :: rule(), RulePrev :: rule()) -> ok.
+update_rule(Rule, RulePrev) ->
+    gen_server:call(?RULE_ENGINE, {update_rule, Rule, RulePrev}, ?T_CALL).
+
 %%----------------------------------------------------------------------------------------
 %% Rule Management
 %%----------------------------------------------------------------------------------------
@@ -216,9 +225,8 @@ get_rules_ordered_by_ts() ->
 -spec get_rules_for_topic(Topic :: binary()) -> [rule()].
 get_rules_for_topic(Topic) ->
     [
-        Rule
-     || Rule = #{from := From} <- get_rules(),
-        emqx_topic:match_any(Topic, From)
+        emqx_topic_index:get_record(M, ?RULE_TOPIC_INDEX)
+     || M <- emqx_topic_index:matches(Topic, ?RULE_TOPIC_INDEX, [unique])
     ].
 
 -spec get_rules_with_same_event(Topic :: binary()) -> [rule()].
@@ -411,10 +419,17 @@ init([]) ->
     {ok, #{}}.
 
 handle_call({insert_rule, Rule}, _From, State) ->
-    do_insert_rule(Rule),
+    ok = do_insert_rule(Rule),
+    ok = do_update_rule_index(Rule),
+    {reply, ok, State};
+handle_call({update_rule, Rule, RulePrev}, _From, State) ->
+    ok = do_delete_rule_index(RulePrev),
+    ok = do_insert_rule(Rule),
+    ok = do_update_rule_index(Rule),
     {reply, ok, State};
 handle_call({delete_rule, Rule}, _From, State) ->
-    do_delete_rule(Rule),
+    ok = do_delete_rule_index(Rule),
+    ok = do_delete_rule(Rule),
     {reply, ok, State};
 handle_call(Req, _From, State) ->
     ?SLOG(error, #{msg => "unexpected_call", request => Req}),
@@ -438,7 +453,7 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal Functions
 %%----------------------------------------------------------------------------------------
 
-parse_and_insert(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt) ->
+with_parsed_rule(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt, Fun) ->
     case emqx_rule_sqlparser:parse(Sql) of
         {ok, Select} ->
             Rule = #{
@@ -459,7 +474,7 @@ parse_and_insert(Params = #{id := RuleId, sql := Sql, actions := Actions}, Creat
                 conditions => emqx_rule_sqlparser:select_where(Select)
                 %% -- calculated fields end
             },
-            ok = insert_rule(Rule),
+            ok = Fun(Rule),
             {ok, Rule};
         {error, Reason} ->
             {error, Reason}
@@ -471,16 +486,27 @@ do_insert_rule(#{id := Id} = Rule) ->
     true = ets:insert(?RULE_TAB, {Id, maps:remove(id, Rule)}),
     ok.
 
-do_delete_rule(RuleId) ->
-    case get_rule(RuleId) of
-        {ok, Rule} ->
-            ok = unload_hooks_for_rule(Rule),
-            ok = clear_metrics_for_rule(RuleId),
-            true = ets:delete(?RULE_TAB, RuleId),
-            ok;
-        not_found ->
-            ok
-    end.
+do_delete_rule(#{id := Id} = Rule) ->
+    ok = unload_hooks_for_rule(Rule),
+    ok = clear_metrics_for_rule(Id),
+    true = ets:delete(?RULE_TAB, Id),
+    ok.
+
+do_update_rule_index(#{id := Id, from := From} = Rule) ->
+    ok = lists:foreach(
+        fun(Topic) ->
+            true = emqx_topic_index:insert(Topic, Id, Rule, ?RULE_TOPIC_INDEX)
+        end,
+        From
+    ).
+
+do_delete_rule_index(#{id := Id, from := From}) ->
+    ok = lists:foreach(
+        fun(Topic) ->
+            true = emqx_topic_index:delete(Topic, Id, ?RULE_TOPIC_INDEX)
+        end,
+        From
+    ).
 
 parse_actions(Actions) ->
     [do_parse_action(Act) || Act <- Actions].

apps/emqx_rule_engine/src/emqx_rule_engine_app.erl

@@ -26,6 +26,7 @@
 
 start(_Type, _Args) ->
     _ = ets:new(?RULE_TAB, [named_table, public, ordered_set, {read_concurrency, true}]),
+    _ = ets:new(?RULE_TOPIC_INDEX, [named_table, public, ordered_set, {read_concurrency, true}]),
     ok = emqx_rule_events:reload(),
     SupRet = emqx_rule_engine_sup:start_link(),
     ok = emqx_rule_engine:load_rules(),

changes/ce/perf-11396.en.md

@@ -0,0 +1 @@
+Introduce topic index for the rule engine runtime that significantly improves the performance of EMQX with a non-trivial number of rules consuming messages matching different topic filters.

changes/ee/feat-11459.en.md

@@ -0,0 +1 @@
+Added the option to configure health check interval for Kafka bridges.

changes/ee/fix-11453.en.md

@@ -0,0 +1 @@
+Fixed an issue which would yield false negatives when testing the connectivity of InfluxDB bridges.

deploy/charts/emqx-enterprise/Chart.yaml

@@ -14,8 +14,8 @@ type: application
 
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
-version: 5.2.0-alpha.1
+version: 5.2.0-alpha.3
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application.
-appVersion: 5.2.0-alpha.1
+appVersion: 5.2.0-alpha.3

scripts/ui-tests/dashboard_test.py

@@ -62,12 +62,13 @@ def test_log(driver, login, dashboard_url):
     ensure_current_url(driver, dest_url)
     title = wait_title(driver)
     assert "Logging" == title.text
-    label = driver.find_element(By.XPATH, "//div[@id='app']//form//label[./label/span[text()='Enable Log Handler']]")
+
+    label = driver.find_element(By.XPATH, "//div[@id='app']//form//label[contains(., 'Enable Log Handler')]")
     assert driver.find_elements(By.ID, label.get_attribute("for"))
-    label = driver.find_element(By.XPATH, "//div[@id='app']//form//label[./label/span[text()='Log Level']]")
+    label = driver.find_element(By.XPATH, "//div[@id='app']//form//label[contains(., 'Log Level')]")
     assert driver.find_elements(By.ID, label.get_attribute("for"))
-    label = driver.find_element(By.XPATH, "//div[@id='app']//form//label[./label/span[text()='Log Formatter']]")
+    label = driver.find_element(By.XPATH, "//div[@id='app']//form//label[contains(., 'Log Formatter')]")
     assert driver.find_elements(By.ID, label.get_attribute("for"))
-    label = driver.find_element(By.XPATH, "//div[@id='app']//form//label[./label/span[text()='Time Offset']]")
+    label = driver.find_element(By.XPATH, "//div[@id='app']//form//label[contains(., 'Time Offset')]")
     assert driver.find_elements(By.ID, label.get_attribute("for"))