fix(topicidx): make `get_record/2` simpler to use in concurrent env

The mechanic of `emqx_topic_index` cannot really guarantee atomicity
of reading records associated with index matches, so instead it's
probably better to make the user aware of that lack of this guarantee.

apps/emqx/src/emqx_topic_index.erl

@@ -73,10 +73,16 @@ get_topic(Key) ->
     emqx_trie_search:get_topic(Key).
 
 %% @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.
+%% May return empty list if the index entry was deleted in the meantime.
+%% NOTE: Only really useful for ETS tables where the record data is the last element.
+-spec get_record(match(_ID), ets:table()) -> [_Record].
 get_record(K, Tab) ->
-    ets:lookup_element(Tab, K, 2).
+    case ets:lookup(Tab, K) of
+        [Entry] ->
+            [erlang:element(tuple_size(Entry), Entry)];
+        [] ->
+            []
+    end.
 
 key(TopicOrFilter, ID) ->
     emqx_trie_search:make_key(TopicOrFilter, ID).

apps/emqx_rule_engine/src/emqx_rule_engine.erl

@@ -225,8 +225,9 @@ get_rules_ordered_by_ts() ->
 -spec get_rules_for_topic(Topic :: binary()) -> [rule()].
 get_rules_for_topic(Topic) ->
     [
-        emqx_topic_index:get_record(M, ?RULE_TOPIC_INDEX)
-     || M <- emqx_topic_index:matches(Topic, ?RULE_TOPIC_INDEX, [unique])
+        Rule
+     || M <- emqx_topic_index:matches(Topic, ?RULE_TOPIC_INDEX, [unique]),
+        Rule <- emqx_topic_index:get_record(M, ?RULE_TOPIC_INDEX)
     ].
 
 -spec get_rules_with_same_event(Topic :: binary()) -> [rule()].