feat: add resource_type to emqx_resource behaviour

apps/emqx_auth/src/emqx_authn/emqx_authn_utils.erl

@@ -21,8 +21,8 @@
 -include_lib("snabbkaffe/include/trace.hrl").
 
 -export([
-    create_resource/4,
+    create_resource/3,
-    update_resource/4,
+    update_resource/3,
     check_password_from_selected_map/3,
     parse_deep/1,
     parse_str/1,
@@ -66,9 +66,8 @@
 %% APIs
 %%--------------------------------------------------------------------
 
-create_resource(Type, ResourceId, Module, Config) ->
+create_resource(ResourceId, Module, Config) ->
     Result = emqx_resource:create_local(
-        Type,
         ResourceId,
         ?AUTHN_RESOURCE_GROUP,
         Module,
@@ -77,9 +76,9 @@ create_resource(Type, ResourceId, Module, Config) ->
     ),
     start_resource_if_enabled(Result, ResourceId, Config).
 
-update_resource(Type, Module, Config, ResourceId) ->
+update_resource(Module, Config, ResourceId) ->
     Result = emqx_resource:recreate_local(
-        Type, ResourceId, Module, Config, ?DEFAULT_RESOURCE_OPTS
+        ResourceId, Module, Config, ?DEFAULT_RESOURCE_OPTS
     ),
     start_resource_if_enabled(Result, ResourceId, Config).
 

apps/emqx_auth/src/emqx_authz/emqx_authz_utils.erl

@@ -25,9 +25,9 @@
 -export([
     cleanup_resources/0,
     make_resource_id/1,
+    create_resource/2,
     create_resource/3,
-    create_resource/4,
+    update_resource/2,
-    update_resource/3,
     remove_resource/1,
     update_config/2,
     parse_deep/2,
@@ -57,13 +57,12 @@
 %% APIs
 %%--------------------------------------------------------------------
 
-create_resource(Type, Module, Config) ->
+create_resource(Module, Config) ->
     ResourceId = make_resource_id(Module),
-    create_resource(Type, ResourceId, Module, Config).
+    create_resource(ResourceId, Module, Config).
 
-create_resource(Type, ResourceId, Module, Config) ->
+create_resource(ResourceId, Module, Config) ->
     Result = emqx_resource:create_local(
-        Type,
         ResourceId,
         ?AUTHZ_RESOURCE_GROUP,
         Module,
@@ -72,11 +71,10 @@ create_resource(Type, ResourceId, Module, Config) ->
     ),
     start_resource_if_enabled(Result, ResourceId, Config).
 
-update_resource(Type, Module, #{annotations := #{id := ResourceId}} = Config) ->
+update_resource(Module, #{annotations := #{id := ResourceId}} = Config) ->
     Result =
         case
             emqx_resource:recreate_local(
-                Type,
                 ResourceId,
                 Module,
                 Config,

apps/emqx_auth_http/src/emqx_authn_http.erl

@@ -40,7 +40,6 @@ create(Config0) ->
         ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
         % {Config, State} = parse_config(Config0),
         {ok, _Data} = emqx_authn_utils:create_resource(
-            http,
             ResourceId,
             emqx_bridge_http_connector,
             Config
@@ -51,9 +50,7 @@ create(Config0) ->
 update(Config0, #{resource_id := ResourceId} = _State) ->
     with_validated_config(Config0, fun(Config, NState) ->
         % {Config, NState} = parse_config(Config0),
-        case
+        case emqx_authn_utils:update_resource(emqx_bridge_http_connector, Config, ResourceId) of
-            emqx_authn_utils:update_resource(http, emqx_bridge_http_connector, Config, ResourceId)
-        of
             {error, Reason} ->
                 error({load_config_error, Reason});
             {ok, _} ->

apps/emqx_auth_http/src/emqx_authz_http.erl

@@ -67,13 +67,15 @@ create(Config) ->
     NConfig = parse_config(Config),
     ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
     {ok, _Data} = emqx_authz_utils:create_resource(
-        http, ResourceId, emqx_bridge_http_connector, NConfig
+        ResourceId,
+        emqx_bridge_http_connector,
+        NConfig
     ),
     NConfig#{annotations => #{id => ResourceId}}.
 
 update(Config) ->
     NConfig = parse_config(Config),
-    case emqx_authz_utils:update_resource(http, emqx_bridge_http_connector, NConfig) of
+    case emqx_authz_utils:update_resource(emqx_bridge_http_connector, NConfig) of
         {error, Reason} -> error({load_config_error, Reason});
         {ok, Id} -> NConfig#{annotations => #{id => Id}}
     end.

apps/emqx_auth_jwt/src/emqx_authn_jwks_connector.erl

@@ -22,6 +22,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -32,6 +33,8 @@
 
 -define(DEFAULT_POOL_SIZE, 8).
 
+resource_type() -> jwks.
+
 callback_mode() -> always_sync.
 
 on_start(InstId, Opts) ->

apps/emqx_auth_jwt/src/emqx_authn_jwt.erl

@@ -183,7 +183,6 @@ do_create(
 ) ->
     ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
     {ok, _Data} = emqx_resource:create_local(
-        jwt,
         ResourceId,
         ?AUTHN_RESOURCE_GROUP,
         emqx_authn_jwks_connector,

apps/emqx_auth_ldap/src/emqx_authn_ldap.erl

@@ -40,12 +40,12 @@ create(_AuthenticatorID, Config) ->
 do_create(Module, Config) ->
     ResourceId = emqx_authn_utils:make_resource_id(Module),
     State = parse_config(Config),
-    {ok, _Data} = emqx_authn_utils:create_resource(ldap, ResourceId, emqx_ldap, Config),
+    {ok, _Data} = emqx_authn_utils:create_resource(ResourceId, emqx_ldap, Config),
     {ok, State#{resource_id => ResourceId}}.
 
 update(Config, #{resource_id := ResourceId} = _State) ->
     NState = parse_config(Config),
-    case emqx_authn_utils:update_resource(ldap, emqx_ldap, Config, ResourceId) of
+    case emqx_authn_utils:update_resource(emqx_ldap, Config, ResourceId) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, _} ->

apps/emqx_auth_ldap/src/emqx_authz_ldap.erl

@@ -56,12 +56,12 @@ description() ->
 
 create(Source) ->
     ResourceId = emqx_authz_utils:make_resource_id(?MODULE),
-    {ok, _Data} = emqx_authz_utils:create_resource(ldap, ResourceId, emqx_ldap, Source),
+    {ok, _Data} = emqx_authz_utils:create_resource(ResourceId, emqx_ldap, Source),
     Annotations = new_annotations(#{id => ResourceId}, Source),
     Source#{annotations => Annotations}.
 
 update(Source) ->
-    case emqx_authz_utils:update_resource(ldap, emqx_ldap, Source) of
+    case emqx_authz_utils:update_resource(emqx_ldap, Source) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, Id} ->

apps/emqx_auth_ldap/test/emqx_authn_ldap_SUITE.erl

@@ -47,7 +47,6 @@ init_per_suite(Config) ->
                 work_dir => ?config(priv_dir, Config)
             }),
             {ok, _} = emqx_resource:create_local(
-                ldap,
                 ?LDAP_RESOURCE,
                 ?AUTHN_RESOURCE_GROUP,
                 emqx_ldap,

apps/emqx_auth_ldap/test/emqx_authn_ldap_bind_SUITE.erl

@@ -47,7 +47,6 @@ init_per_suite(Config) ->
                 work_dir => ?config(priv_dir, Config)
             }),
             {ok, _} = emqx_resource:create_local(
-                ldap,
                 ?LDAP_RESOURCE,
                 ?AUTHN_RESOURCE_GROUP,
                 emqx_ldap,

apps/emqx_auth_ldap/test/emqx_authz_ldap_SUITE.erl

@@ -178,7 +178,6 @@ stop_apps(Apps) ->
 
 create_ldap_resource() ->
     {ok, _} = emqx_resource:create_local(
-        ldap,
         ?LDAP_RESOURCE,
         ?AUTHZ_RESOURCE_GROUP,
         emqx_ldap,

apps/emqx_auth_mongodb/src/emqx_authn_mongodb.erl

@@ -37,7 +37,6 @@ create(Config0) ->
     ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
     {Config, State} = parse_config(Config0),
     {ok, _Data} = emqx_authn_utils:create_resource(
-        mongodb,
         ResourceId,
         emqx_mongodb,
         Config
@@ -46,7 +45,7 @@ create(Config0) ->
 
 update(Config0, #{resource_id := ResourceId} = _State) ->
     {Config, NState} = parse_config(Config0),
-    case emqx_authn_utils:update_resource(mongodb, emqx_mongodb, Config, ResourceId) of
+    case emqx_authn_utils:update_resource(emqx_mongodb, Config, ResourceId) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, _} ->

apps/emqx_auth_mongodb/src/emqx_authz_mongodb.erl

@@ -49,13 +49,13 @@ description() ->
 
 create(#{filter := Filter} = Source) ->
     ResourceId = emqx_authz_utils:make_resource_id(?MODULE),
-    {ok, _Data} = emqx_authz_utils:create_resource(mongodb, ResourceId, emqx_mongodb, Source),
+    {ok, _Data} = emqx_authz_utils:create_resource(ResourceId, emqx_mongodb, Source),
     FilterTemp = emqx_authz_utils:parse_deep(Filter, ?ALLOWED_VARS),
     Source#{annotations => #{id => ResourceId}, filter_template => FilterTemp}.
 
 update(#{filter := Filter} = Source) ->
     FilterTemp = emqx_authz_utils:parse_deep(Filter, ?ALLOWED_VARS),
-    case emqx_authz_utils:update_resource(mongodb, emqx_mongodb, Source) of
+    case emqx_authz_utils:update_resource(emqx_mongodb, Source) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, Id} ->

apps/emqx_auth_mysql/src/emqx_authn_mysql.erl

@@ -39,12 +39,12 @@ create(_AuthenticatorID, Config) ->
 create(Config0) ->
     ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
     {Config, State} = parse_config(Config0),
-    {ok, _Data} = emqx_authn_utils:create_resource(mysql, ResourceId, emqx_mysql, Config),
+    {ok, _Data} = emqx_authn_utils:create_resource(ResourceId, emqx_mysql, Config),
     {ok, State#{resource_id => ResourceId}}.
 
 update(Config0, #{resource_id := ResourceId} = _State) ->
     {Config, NState} = parse_config(Config0),
-    case emqx_authn_utils:update_resource(mysql, emqx_mysql, Config, ResourceId) of
+    case emqx_authn_utils:update_resource(emqx_mysql, Config, ResourceId) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, _} ->

apps/emqx_auth_mysql/src/emqx_authz_mysql.erl

@@ -53,13 +53,13 @@ create(#{query := SQL} = Source0) ->
     {PrepareSQL, TmplToken} = emqx_authz_utils:parse_sql(SQL, '?', ?ALLOWED_VARS),
     ResourceId = emqx_authz_utils:make_resource_id(?MODULE),
     Source = Source0#{prepare_statement => #{?PREPARE_KEY => PrepareSQL}},
-    {ok, _Data} = emqx_authz_utils:create_resource(mysql, ResourceId, emqx_mysql, Source),
+    {ok, _Data} = emqx_authz_utils:create_resource(ResourceId, emqx_mysql, Source),
     Source#{annotations => #{id => ResourceId, tmpl_token => TmplToken}}.
 
 update(#{query := SQL} = Source0) ->
     {PrepareSQL, TmplToken} = emqx_authz_utils:parse_sql(SQL, '?', ?ALLOWED_VARS),
     Source = Source0#{prepare_statement => #{?PREPARE_KEY => PrepareSQL}},
-    case emqx_authz_utils:update_resource(mysql, emqx_mysql, Source) of
+    case emqx_authz_utils:update_resource(emqx_mysql, Source) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, Id} ->

apps/emqx_auth_mysql/test/emqx_authn_mysql_SUITE.erl

@@ -58,7 +58,6 @@ init_per_suite(Config) ->
                 work_dir => ?config(priv_dir, Config)
             }),
             {ok, _} = emqx_resource:create_local(
-                mysql,
                 ?MYSQL_RESOURCE,
                 ?AUTHN_RESOURCE_GROUP,
                 emqx_mysql,

apps/emqx_auth_mysql/test/emqx_authz_mysql_SUITE.erl

@@ -446,7 +446,6 @@ stop_apps(Apps) ->
 
 create_mysql_resource() ->
     {ok, _} = emqx_resource:create_local(
-        mysql,
         ?MYSQL_RESOURCE,
         ?AUTHZ_RESOURCE_GROUP,
         emqx_mysql,

apps/emqx_auth_postgresql/src/emqx_authn_postgresql.erl

@@ -45,7 +45,6 @@ create(Config0) ->
     ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
     {Config, State} = parse_config(Config0, ResourceId),
     {ok, _Data} = emqx_authn_utils:create_resource(
-        postgresql,
         ResourceId,
         emqx_postgresql,
         Config
@@ -54,7 +53,7 @@ create(Config0) ->
 
 update(Config0, #{resource_id := ResourceId} = _State) ->
     {Config, NState} = parse_config(Config0, ResourceId),
-    case emqx_authn_utils:update_resource(postgresql, emqx_postgresql, Config, ResourceId) of
+    case emqx_authn_utils:update_resource(emqx_postgresql, Config, ResourceId) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, _} ->

apps/emqx_auth_postgresql/src/emqx_authz_postgresql.erl

@@ -53,7 +53,6 @@ create(#{query := SQL0} = Source) ->
     {SQL, PlaceHolders} = emqx_authz_utils:parse_sql(SQL0, '$n', ?ALLOWED_VARS),
     ResourceID = emqx_authz_utils:make_resource_id(emqx_postgresql),
     {ok, _Data} = emqx_authz_utils:create_resource(
-        postgresql,
         ResourceID,
         emqx_postgresql,
         Source#{prepare_statement => #{ResourceID => SQL}}
@@ -64,7 +63,6 @@ update(#{query := SQL0, annotations := #{id := ResourceID}} = Source) ->
     {SQL, PlaceHolders} = emqx_authz_utils:parse_sql(SQL0, '$n', ?ALLOWED_VARS),
     case
         emqx_authz_utils:update_resource(
-            postgresql,
             emqx_postgresql,
             Source#{prepare_statement => #{ResourceID => SQL}}
         )

apps/emqx_auth_postgresql/test/emqx_authn_postgresql_SUITE.erl

@@ -79,7 +79,6 @@ init_per_suite(Config) ->
                 work_dir => ?config(priv_dir, Config)
             }),
             {ok, _} = emqx_resource:create_local(
-                postgresql,
                 ?PGSQL_RESOURCE,
                 ?AUTHN_RESOURCE_GROUP,
                 emqx_postgresql,
@@ -199,13 +198,9 @@ test_user_auth(#{
 
 t_authenticate_disabled_prepared_statements(_Config) ->
     ResConfig = maps:merge(pgsql_config(), #{disable_prepared_statements => true}),
-    {ok, _} = emqx_resource:recreate_local(
+    {ok, _} = emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, ResConfig, #{}),
-        postgresql, ?PGSQL_RESOURCE, emqx_postgresql, ResConfig, #{}
-    ),
     on_exit(fun() ->
-        emqx_resource:recreate_local(
+        emqx_resource:recreate_local(?PGSQL_RESOURCE, emqx_postgresql, pgsql_config(), #{})
-            postgresql, ?PGSQL_RESOURCE, emqx_postgresql, pgsql_config(), #{}
-        )
     end),
     ok = lists:foreach(
         fun(Sample0) ->

apps/emqx_auth_postgresql/test/emqx_authz_postgresql_SUITE.erl

@@ -437,7 +437,6 @@ pgsql_config() ->
 
 create_pgsql_resource() ->
     emqx_resource:create_local(
-        postgresql,
         ?PGSQL_RESOURCE,
         ?AUTHZ_RESOURCE_GROUP,
         emqx_postgresql,

apps/emqx_auth_redis/src/emqx_authn_redis.erl

@@ -42,7 +42,6 @@ create(Config0) ->
             Res;
         {Config, State} ->
             {ok, _Data} = emqx_authn_utils:create_resource(
-                redis,
                 ResourceId,
                 emqx_redis,
                 Config
@@ -52,7 +51,7 @@ create(Config0) ->
 
 update(Config0, #{resource_id := ResourceId} = _State) ->
     {Config, NState} = parse_config(Config0),
-    case emqx_authn_utils:update_resource(redis, emqx_redis, Config, ResourceId) of
+    case emqx_authn_utils:update_resource(emqx_redis, Config, ResourceId) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, _} ->

apps/emqx_auth_redis/src/emqx_authz_redis.erl

@@ -50,12 +50,12 @@ description() ->
 create(#{cmd := CmdStr} = Source) ->
     CmdTemplate = parse_cmd(CmdStr),
     ResourceId = emqx_authz_utils:make_resource_id(?MODULE),
-    {ok, _Data} = emqx_authz_utils:create_resource(redis, ResourceId, emqx_redis, Source),
+    {ok, _Data} = emqx_authz_utils:create_resource(ResourceId, emqx_redis, Source),
     Source#{annotations => #{id => ResourceId}, cmd_template => CmdTemplate}.
 
 update(#{cmd := CmdStr} = Source) ->
     CmdTemplate = parse_cmd(CmdStr),
-    case emqx_authz_utils:update_resource(redis, emqx_redis, Source) of
+    case emqx_authz_utils:update_resource(emqx_redis, Source) of
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, Id} ->

apps/emqx_auth_redis/test/emqx_authn_redis_SUITE.erl

@@ -63,7 +63,6 @@ init_per_suite(Config) ->
                 work_dir => ?config(priv_dir, Config)
             }),
             {ok, _} = emqx_resource:create_local(
-                redis,
                 ?REDIS_RESOURCE,
                 ?AUTHN_RESOURCE_GROUP,
                 emqx_redis,

apps/emqx_auth_redis/test/emqx_authz_redis_SUITE.erl

@@ -384,7 +384,6 @@ stop_apps(Apps) ->
 
 create_redis_resource() ->
     {ok, _} = emqx_resource:create_local(
-        redis,
         ?REDIS_RESOURCE,
         ?AUTHZ_RESOURCE_GROUP,
         emqx_redis,

apps/emqx_bridge/src/emqx_bridge_resource.erl

@@ -195,7 +195,6 @@ create(Type, Name, Conf0, Opts) ->
     TypeBin = bin(Type),
     Conf = Conf0#{bridge_type => TypeBin, bridge_name => Name},
     {ok, _Data} = emqx_resource:create_local(
-        Type,
         resource_id(Type, Name),
         <<"bridge">>,
         bridge_to_resource_type(Type),
@@ -265,7 +264,6 @@ recreate(Type, Name, Conf0, Opts) ->
     TypeBin = bin(Type),
     Conf = Conf0#{bridge_type => TypeBin, bridge_name => Name},
     emqx_resource:recreate_local(
-        Type,
         resource_id(Type, Name),
         bridge_to_resource_type(Type),
         parse_confs(TypeBin, Name, Conf),
@@ -302,7 +300,7 @@ create_dry_run_bridge_v1(Type, Conf0) ->
                 {error, Reason};
             {ok, ConfNew} ->
                 ParseConf = parse_confs(TypeBin, TmpName, ConfNew),
-                emqx_resource:create_dry_run_local(Type, bridge_to_resource_type(Type), ParseConf)
+                emqx_resource:create_dry_run_local(bridge_to_resource_type(Type), ParseConf)
         end
     catch
         %% validation errors

apps/emqx_bridge/test/emqx_bridge_v2_SUITE.erl

@@ -1110,6 +1110,7 @@ t_query_uses_action_query_mode(_Config) ->
 
             %% ... now we use a quite different query mode for the action
             meck:expect(con_mod(), query_mode, 1, simple_async_internal_buffer),
+            meck:expect(con_mod(), resource_type, 0, dummy),
             meck:expect(con_mod(), callback_mode, 0, async_if_possible),
 
             {ok, _} = emqx_bridge_v2:create(bridge_type(), ActionName, ActionConfig),

apps/emqx_bridge/test/emqx_bridge_v2_api_SUITE.erl

@@ -293,6 +293,7 @@ init_mocks() ->
     meck:new(emqx_connector_resource, [passthrough, no_link]),
     meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR_IMPL),
     meck:new(?CONNECTOR_IMPL, [non_strict, no_link]),
+    meck:expect(?CONNECTOR_IMPL, resource_type, 0, dummy),
     meck:expect(?CONNECTOR_IMPL, callback_mode, 0, async_if_possible),
     meck:expect(
         ?CONNECTOR_IMPL,

apps/emqx_bridge/test/emqx_bridge_v2_dummy_connector.erl

@@ -15,15 +15,17 @@
 
 %% this module is only intended to be mocked
 -module(emqx_bridge_v2_dummy_connector).
+-behavior(emqx_resource).
 
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
     on_add_channel/4,
     on_get_channel_status/3
 ]).
-
+resource_type() -> dummy.
 callback_mode() -> error(unexpected).
 on_start(_, _) -> error(unexpected).
 on_stop(_, _) -> error(unexpected).

apps/emqx_bridge/test/emqx_bridge_v2_test_connector.erl

@@ -19,6 +19,7 @@
 
 -export([
     query_mode/1,
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -34,6 +35,8 @@
 query_mode(_Config) ->
     sync.
 
+resource_type() -> test_connector.
+
 callback_mode() ->
     always_sync.
 

apps/emqx_bridge_cassandra/src/emqx_bridge_cassandra.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_cassandra, [
     {description, "EMQX Enterprise Cassandra Bridge"},
-    {vsn, "0.3.1"},
+    {vsn, "0.3.2"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_cassandra/src/emqx_bridge_cassandra_connector.erl

@@ -19,6 +19,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -94,6 +95,7 @@ desc("connector") ->
 
 %%--------------------------------------------------------------------
 %% callbacks for emqx_resource
+resource_type() -> cassandra.
 
 callback_mode() -> async_if_possible.
 

apps/emqx_bridge_cassandra/test/emqx_bridge_cassandra_connector_SUITE.erl

@@ -212,7 +212,6 @@ check_config(Config) ->
 create_local_resource(ResourceId, CheckedConfig) ->
     {ok, Bridge} =
         emqx_resource:create_local(
-            cassandra,
             ResourceId,
             ?CONNECTOR_RESOURCE_GROUP,
             ?CASSANDRA_RESOURCE_MOD,

apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_clickhouse, [
     {description, "EMQX Enterprise ClickHouse Bridge"},
-    {vsn, "0.4.1"},
+    {vsn, "0.4.2"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse_connector.erl

@@ -29,6 +29,7 @@
 
 %% callbacks for behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -128,6 +129,7 @@ values(_) ->
 %% ===================================================================
 %% Callbacks defined in emqx_resource
 %% ===================================================================
+resource_type() -> clickhouse.
 
 callback_mode() -> always_sync.
 

apps/emqx_bridge_clickhouse/test/emqx_bridge_clickhouse_connector_SUITE.erl

@@ -109,7 +109,6 @@ t_start_passfile(Config) ->
     ?assertMatch(
         {ok, #{status := connected}},
         emqx_resource:create_local(
-            clickhouse,
             ResourceID,
             ?CONNECTOR_RESOURCE_GROUP,
             ?CLICKHOUSE_RESOURCE_MOD,
@@ -139,7 +138,6 @@ perform_lifecycle_check(ResourceID, InitialConfig) ->
         status := InitialStatus
     }} =
         emqx_resource:create_local(
-            clickhouse,
             ResourceID,
             ?CONNECTOR_RESOURCE_GROUP,
             ?CLICKHOUSE_RESOURCE_MOD,

apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_dynamo, [
     {description, "EMQX Enterprise Dynamo Bridge"},
-    {vsn, "0.2.2"},
+    {vsn, "0.2.3"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo_connector.erl

@@ -17,6 +17,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -68,6 +69,7 @@ fields(config) ->
 %%========================================================================================
 %% `emqx_resource' API
 %%========================================================================================
+resource_type() -> dynamo.
 
 callback_mode() -> always_sync.
 

apps/emqx_bridge_es/src/emqx_bridge_es.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge_es, [
     {description, "EMQX Enterprise Elastic Search Bridge"},
-    {vsn, "0.1.3"},
+    {vsn, "0.1.4"},
     {modules, [
         emqx_bridge_es,
         emqx_bridge_es_connector

apps/emqx_bridge_es/src/emqx_bridge_es_connector.erl

@@ -14,6 +14,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -207,6 +208,8 @@ base_url(#{server := Server}) -> "http://" ++ Server.
 %%-------------------------------------------------------------------------------------
 %% `emqx_resource' API
 %%-------------------------------------------------------------------------------------
+resource_type() -> elastic_search.
+
 callback_mode() -> async_if_possible.
 
 -spec on_start(manager_id(), config()) -> {ok, state()} | no_return().

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_gcp_pubsub, [
     {description, "EMQX Enterprise GCP Pub/Sub Bridge"},
-    {vsn, "0.3.1"},
+    {vsn, "0.3.2"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_impl_consumer.erl

@@ -8,6 +8,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     query_mode/1,
     on_start/2,
@@ -84,6 +85,8 @@
 %%-------------------------------------------------------------------------------------------------
 %% `emqx_resource' API
 %%-------------------------------------------------------------------------------------------------
+-spec resource_type() -> resource_type().
+resource_type() -> gcp_pubsub_consumer.
 
 -spec callback_mode() -> callback_mode().
 callback_mode() -> async_if_possible.

apps/emqx_bridge_gcp_pubsub/src/emqx_bridge_gcp_pubsub_impl_producer.erl

@@ -41,6 +41,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     query_mode/1,
     on_start/2,
@@ -62,6 +63,7 @@
 %%-------------------------------------------------------------------------------------------------
 %% `emqx_resource' API
 %%-------------------------------------------------------------------------------------------------
+resource_type() -> gcp_pubsub.
 
 callback_mode() -> async_if_possible.
 

apps/emqx_bridge_greptimedb/src/emqx_bridge_greptimedb_connector.erl

@@ -16,6 +16,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -67,6 +68,8 @@
 
 %% -------------------------------------------------------------------------------------------------
 %% resource callback
+resource_type() -> greptimedb.
+
 callback_mode() -> async_if_possible.
 
 on_add_channel(

apps/emqx_bridge_greptimedb/test/emqx_bridge_greptimedb_connector_SUITE.erl

@@ -83,7 +83,6 @@ perform_lifecycle_check(PoolName, InitialConfig) ->
         state := #{client := #{pool := ReturnedPoolName}} = State,
         status := InitialStatus
     }} = emqx_resource:create_local(
-        greptimedb,
         PoolName,
         ?CONNECTOR_RESOURCE_GROUP,
         ?GREPTIMEDB_RESOURCE_MOD,

apps/emqx_bridge_hstreamdb/src/emqx_bridge_hstreamdb.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_hstreamdb, [
     {description, "EMQX Enterprise HStreamDB Bridge"},
-    {vsn, "0.2.1"},
+    {vsn, "0.2.2"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_hstreamdb/src/emqx_bridge_hstreamdb_connector.erl

@@ -16,6 +16,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -44,6 +45,8 @@
 
 %% -------------------------------------------------------------------------------------------------
 %% resource callback
+resource_type() -> hstreamsdb.
+
 callback_mode() -> always_sync.
 
 on_start(InstId, Config) ->

apps/emqx_bridge_http/src/emqx_bridge_http_connector.erl

@@ -26,6 +26,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -183,6 +184,7 @@ sc(Type, Meta) -> hoconsc:mk(Type, Meta).
 ref(Field) -> hoconsc:ref(?MODULE, Field).
 
 %% ===================================================================
+resource_type() -> webhook.
 
 callback_mode() -> async_if_possible.
 

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.2.3"},
+    {vsn, "0.2.4"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_influxdb/src/emqx_bridge_influxdb_connector.erl

@@ -16,6 +16,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -70,6 +71,8 @@
 
 %% -------------------------------------------------------------------------------------------------
 %% resource callback
+resource_type() -> influxdb.
+
 callback_mode() -> async_if_possible.
 
 on_add_channel(

apps/emqx_bridge_influxdb/test/emqx_bridge_influxdb_connector_SUITE.erl

@@ -86,7 +86,6 @@ perform_lifecycle_check(PoolName, InitialConfig) ->
         state := #{client := #{pool := ReturnedPoolName}} = State,
         status := InitialStatus
     }} = emqx_resource:create_local(
-        influxdb,
         PoolName,
         ?CONNECTOR_RESOURCE_GROUP,
         ?INFLUXDB_RESOURCE_MOD,
@@ -198,7 +197,6 @@ perform_tls_opts_check(PoolName, InitialConfig, VerifyReturn) ->
             config := #{ssl := #{enable := SslEnabled}},
             status := Status
         }} = emqx_resource:create_local(
-            influxdb,
             PoolName,
             ?CONNECTOR_RESOURCE_GROUP,
             ?INFLUXDB_RESOURCE_MOD,

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge_iotdb, [
     {description, "EMQX Enterprise Apache IoTDB Bridge"},
-    {vsn, "0.2.2"},
+    {vsn, "0.2.3"},
     {modules, [
         emqx_bridge_iotdb,
         emqx_bridge_iotdb_connector

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_connector.erl

@@ -15,6 +15,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -206,6 +207,8 @@ proplists_without(Keys, List) ->
 %%-------------------------------------------------------------------------------------
 %% `emqx_resource' API
 %%-------------------------------------------------------------------------------------
+resource_type() -> iotdb.
+
 callback_mode() -> async_if_possible.
 
 -spec on_start(manager_id(), config()) -> {ok, state()} | no_return().

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_consumer.erl

@@ -7,6 +7,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     query_mode/1,
     on_start/2,
@@ -125,6 +126,7 @@
 %%-------------------------------------------------------------------------------------
 %% `emqx_resource' API
 %%-------------------------------------------------------------------------------------
+resource_type() -> kafka_consumer.
 
 callback_mode() ->
     async_if_possible.

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

@@ -10,6 +10,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     query_mode/1,
     callback_mode/0,
     on_start/2,
@@ -35,6 +36,8 @@
 -define(kafka_client_id, kafka_client_id).
 -define(kafka_producers, kafka_producers).
 
+resource_type() -> kafka_producer.
+
 query_mode(#{parameters := #{query_mode := sync}}) ->
     simple_sync_internal_buffer;
 query_mode(_) ->

apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_kinesis, [
     {description, "EMQX Enterprise Amazon Kinesis Bridge"},
-    {vsn, "0.2.1"},
+    {vsn, "0.2.2"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_kinesis/src/emqx_bridge_kinesis_impl_producer.erl

@@ -30,6 +30,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -50,6 +51,7 @@
 %%-------------------------------------------------------------------------------------------------
 %% `emqx_resource' API
 %%-------------------------------------------------------------------------------------------------
+resource_type() -> kinesis_producer.
 
 callback_mode() -> always_sync.
 

apps/emqx_bridge_mongodb/src/emqx_bridge_mongodb.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_mongodb, [
     {description, "EMQX Enterprise MongoDB Bridge"},
-    {vsn, "0.3.2"},
+    {vsn, "0.3.3"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_mongodb/src/emqx_bridge_mongodb_connector.erl

@@ -11,6 +11,7 @@
 %% `emqx_resource' API
 -export([
     on_remove_channel/3,
+    resource_type/0,
     callback_mode/0,
     on_add_channel/4,
     on_get_channel_status/3,
@@ -25,6 +26,7 @@
 %%========================================================================================
 %% `emqx_resource' API
 %%========================================================================================
+resource_type() -> emqx_mongodb:resource_type().
 
 callback_mode() -> emqx_mongodb:callback_mode().
 

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_connector.erl

@@ -30,6 +30,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -76,6 +77,8 @@ on_message_received(Msg, HookPoints, ResId) ->
     ok.
 
 %% ===================================================================
+resource_type() -> mqtt.
+
 callback_mode() -> async_if_possible.
 
 on_start(ResourceId, #{server := Server} = Conf) ->

apps/emqx_bridge_mysql/src/emqx_bridge_mysql.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_mysql, [
     {description, "EMQX Enterprise MySQL Bridge"},
-    {vsn, "0.1.7"},
+    {vsn, "0.1.8"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_mysql/src/emqx_bridge_mysql_connector.erl

@@ -10,6 +10,7 @@
 %% `emqx_resource' API
 -export([
     on_remove_channel/3,
+    resource_type/0,
     callback_mode/0,
     on_add_channel/4,
     on_batch_query/3,
@@ -24,6 +25,7 @@
 %%========================================================================================
 %% `emqx_resource' API
 %%========================================================================================
+resource_type() -> emqx_mysql:resource_type().
 
 callback_mode() -> emqx_mysql:callback_mode().
 

apps/emqx_bridge_opents/src/emqx_bridge_opents.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_opents, [
     {description, "EMQX Enterprise OpenTSDB Bridge"},
-    {vsn, "0.2.1"},
+    {vsn, "0.2.2"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_opents/src/emqx_bridge_opents_connector.erl

@@ -18,6 +18,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -114,6 +115,8 @@ connector_example_values() ->
 
 -define(HTTP_CONNECT_TIMEOUT, 1000).
 
+resource_type() -> opents.
+
 callback_mode() -> always_sync.
 
 on_start(

apps/emqx_bridge_pulsar/src/emqx_bridge_pulsar_connector.erl

@@ -10,6 +10,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     query_mode/1,
     on_start/2,
@@ -55,6 +56,7 @@
 %%-------------------------------------------------------------------------------------
 %% `emqx_resource' API
 %%-------------------------------------------------------------------------------------
+resource_type() -> pulsar.
 
 callback_mode() -> async_if_possible.
 

apps/emqx_bridge_rabbitmq/src/emqx_bridge_rabbitmq.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_rabbitmq, [
     {description, "EMQX Enterprise RabbitMQ Bridge"},
-    {vsn, "0.2.1"},
+    {vsn, "0.2.2"},
     {registered, []},
     {mod, {emqx_bridge_rabbitmq_app, []}},
     {applications, [

apps/emqx_bridge_rabbitmq/src/emqx_bridge_rabbitmq_connector.erl

@@ -31,6 +31,7 @@
     on_remove_channel/3,
     on_get_channels/1,
     on_stop/2,
+    resource_type/0,
     callback_mode/0,
     on_get_status/2,
     on_get_channel_status/3,
@@ -60,6 +61,7 @@ fields(config) ->
 %% ===================================================================
 
 %% emqx_resource callback
+resource_type() -> rabbitmq.
 
 callback_mode() -> always_sync.
 

apps/emqx_bridge_rabbitmq/test/emqx_bridge_rabbitmq_connector_SUITE.erl

@@ -135,7 +135,6 @@ check_config(Config) ->
 
 create_local_resource(ResourceID, CheckedConfig) ->
     {ok, Bridge} = emqx_resource:create_local(
-        rabbitmq,
         ResourceID,
         ?CONNECTOR_RESOURCE_GROUP,
         emqx_bridge_rabbitmq_connector,

apps/emqx_bridge_redis/src/emqx_bridge_redis.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_redis, [
     {description, "EMQX Enterprise Redis Bridge"},
-    {vsn, "0.1.8"},
+    {vsn, "0.1.9"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_redis/src/emqx_bridge_redis_connector.erl

@@ -12,6 +12,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_add_channel/4,
     on_remove_channel/3,
@@ -29,7 +30,9 @@
 %% resource callbacks
 %% -------------------------------------------------------------------------------------------------
 
-callback_mode() -> always_sync.
+resource_type() -> emqx_redis:resource_type().
+
+callback_mode() -> emqx_redis:callback_mode().
 
 on_add_channel(
     _InstanceId,

apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_rocketmq, [
     {description, "EMQX Enterprise RocketMQ Bridge"},
-    {vsn, "0.2.2"},
+    {vsn, "0.2.3"},
     {registered, []},
     {applications, [kernel, stdlib, emqx_resource, rocketmq]},
     {env, [

apps/emqx_bridge_rocketmq/src/emqx_bridge_rocketmq_connector.erl

@@ -16,6 +16,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -90,6 +91,8 @@ servers() ->
 %% `emqx_resource' API
 %%========================================================================================
 
+resource_type() -> rocketmq.
+
 callback_mode() -> always_sync.
 
 on_start(

apps/emqx_bridge_s3/src/emqx_bridge_s3_connector.erl

@@ -13,6 +13,7 @@
 
 -behaviour(emqx_resource).
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -92,6 +93,8 @@
 -define(AGGREG_SUP, emqx_bridge_s3_sup).
 
 %%
+-spec resource_type() -> resource_type().
+resource_type() -> s3.
 
 -spec callback_mode() -> callback_mode().
 callback_mode() ->

apps/emqx_bridge_sqlserver/src/emqx_bridge_sqlserver.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_sqlserver, [
     {description, "EMQX Enterprise SQL Server Bridge"},
-    {vsn, "0.2.2"},
+    {vsn, "0.2.3"},
     {registered, []},
     {applications, [kernel, stdlib, emqx_resource, odbc]},
     {env, [

apps/emqx_bridge_sqlserver/src/emqx_bridge_sqlserver_connector.erl

@@ -30,6 +30,7 @@
 
 %% callbacks for behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -173,6 +174,7 @@ server() ->
 %%====================================================================
 %% Callbacks defined in emqx_resource
 %%====================================================================
+resource_type() -> sqlserver.
 
 callback_mode() -> always_sync.
 

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_syskeeper, [
     {description, "EMQX Enterprise Data bridge for Syskeeper"},
-    {vsn, "0.1.3"},
+    {vsn, "0.1.4"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_connector.erl

@@ -18,6 +18,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     query_mode/1,
     on_start/2,
@@ -147,6 +148,7 @@ server() ->
 
 %% -------------------------------------------------------------------------------------------------
 %% `emqx_resource' API
+resource_type() -> syskeeper.
 
 callback_mode() -> always_sync.
 

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_proxy_server.erl

@@ -12,6 +12,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     query_mode/1,
     on_start/2,
     on_stop/2,
@@ -40,6 +41,8 @@
 
 %% -------------------------------------------------------------------------------------------------
 %% emqx_resource
+resource_type() ->
+    syskeeper_proxy_server.
 
 query_mode(_) ->
     no_queries.

apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_tdengine, [
     {description, "EMQX Enterprise TDEngine Bridge"},
-    {vsn, "0.2.1"},
+    {vsn, "0.2.2"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine_connector.erl

@@ -19,6 +19,7 @@
 
 %% `emqx_resource' API
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -140,6 +141,7 @@ connector_example_values() ->
 %%========================================================================================
 %% `emqx_resource' API
 %%========================================================================================
+resource_type() -> tdengine.
 
 callback_mode() -> always_sync.
 

apps/emqx_connector/src/emqx_connector_resource.erl

@@ -126,7 +126,6 @@ create(Type, Name, Conf0, Opts) ->
     ResourceId = resource_id(Type, Name),
     Conf = Conf0#{connector_type => TypeBin, connector_name => Name},
     {ok, _Data} = emqx_resource:create_local(
-        Type,
         ResourceId,
         ?CONNECTOR_RESOURCE_GROUP,
         ?MODULE:connector_to_resource_type(Type),
@@ -200,7 +199,6 @@ recreate(Type, Name, Conf) ->
 recreate(Type, Name, Conf, Opts) ->
     TypeBin = bin(Type),
     emqx_resource:recreate_local(
-        Type,
         resource_id(Type, Name),
         ?MODULE:connector_to_resource_type(Type),
         parse_confs(TypeBin, Name, Conf),
@@ -237,7 +235,7 @@ create_dry_run(Type, Conf0, Callback) ->
             {ok, ConfNew} ->
                 ParseConf = parse_confs(bin(Type), TmpName, ConfNew),
                 emqx_resource:create_dry_run_local(
-                    Type, TmpName, ?MODULE:connector_to_resource_type(Type), ParseConf, Callback
+                    TmpName, ?MODULE:connector_to_resource_type(Type), ParseConf, Callback
                 )
         end
     catch

apps/emqx_connector/test/emqx_connector_SUITE.erl

@@ -50,6 +50,7 @@ t_connector_lifecycle({init, Config}) ->
     meck:new(emqx_connector_resource, [passthrough]),
     meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
     meck:new(?CONNECTOR, [non_strict]),
+    meck:expect(?CONNECTOR, resource_type, 0, dummy),
     meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
     meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
     meck:expect(?CONNECTOR, on_stop, 2, ok),
@@ -171,6 +172,7 @@ t_remove_fail({'init', Config}) ->
     meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
     meck:new(?CONNECTOR, [non_strict]),
     meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
+    meck:expect(?CONNECTOR, resource_type, 0, dummy),
     meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
     meck:expect(?CONNECTOR, on_get_channels, 1, [{<<"my_channel">>, #{enable => true}}]),
     meck:expect(?CONNECTOR, on_add_channel, 4, {ok, connector_state}),
@@ -234,6 +236,7 @@ t_create_with_bad_name_direct_path({init, Config}) ->
     meck:new(emqx_connector_resource, [passthrough]),
     meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
     meck:new(?CONNECTOR, [non_strict]),
+    meck:expect(?CONNECTOR, resource_type, 0, dummy),
     meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
     meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
     meck:expect(?CONNECTOR, on_stop, 2, ok),
@@ -265,6 +268,7 @@ t_create_with_bad_name_root_path({init, Config}) ->
     meck:new(emqx_connector_resource, [passthrough]),
     meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
     meck:new(?CONNECTOR, [non_strict]),
+    meck:expect(?CONNECTOR, resource_type, 0, dummy),
     meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
     meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
     meck:expect(?CONNECTOR, on_stop, 2, ok),
@@ -299,6 +303,7 @@ t_no_buffer_workers({'init', Config}) ->
     meck:new(emqx_connector_resource, [passthrough]),
     meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR),
     meck:new(?CONNECTOR, [non_strict]),
+    meck:expect(?CONNECTOR, resource_type, 0, dummy),
     meck:expect(?CONNECTOR, callback_mode, 0, async_if_possible),
     meck:expect(?CONNECTOR, on_start, 2, {ok, connector_state}),
     meck:expect(?CONNECTOR, on_get_channels, 1, []),

apps/emqx_connector/test/emqx_connector_api_SUITE.erl

@@ -224,6 +224,7 @@ init_mocks(_TestCase) ->
     meck:new(emqx_connector_resource, [passthrough, no_link]),
     meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR_IMPL),
     meck:new(?CONNECTOR_IMPL, [non_strict, no_link]),
+    meck:expect(?CONNECTOR_IMPL, resource_type, 0, dummy),
     meck:expect(?CONNECTOR_IMPL, callback_mode, 0, async_if_possible),
     meck:expect(
         ?CONNECTOR_IMPL,

apps/emqx_connector/test/emqx_connector_dummy_impl.erl

@@ -15,8 +15,10 @@
 
 %% this module is only intended to be mocked
 -module(emqx_connector_dummy_impl).
+-behavior(emqx_resource).
 
 -export([
+    resource_type/0,
     query_mode/1,
     callback_mode/0,
     on_start/2,
@@ -25,6 +27,7 @@
     on_get_channel_status/3
 ]).
 
+resource_type() -> dummy.
 query_mode(_) -> error(unexpected).
 callback_mode() -> error(unexpected).
 on_start(_, _) -> error(unexpected).

apps/emqx_dashboard_sso/src/emqx_dashboard_sso_manager.erl

@@ -132,7 +132,6 @@ make_resource_id(Backend) ->
 
 create_resource(ResourceId, Module, Config) ->
     Result = emqx_resource:create_local(
-        dashboard_sso,
         ResourceId,
         ?RESOURCE_GROUP,
         Module,
@@ -143,7 +142,7 @@ create_resource(ResourceId, Module, Config) ->
 
 update_resource(ResourceId, Module, Config) ->
     Result = emqx_resource:recreate_local(
-        dashboard_sso, ResourceId, Module, Config, ?DEFAULT_RESOURCE_OPTS
+        ResourceId, Module, Config, ?DEFAULT_RESOURCE_OPTS
     ),
     start_resource_if_enabled(ResourceId, Result, Config).
 

apps/emqx_ldap/src/emqx_ldap.app.src

@@ -1,6 +1,6 @@
 {application, emqx_ldap, [
     {description, "EMQX LDAP Connector"},
-    {vsn, "0.1.8"},
+    {vsn, "0.1.9"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_ldap/src/emqx_ldap.erl

@@ -27,6 +27,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -129,6 +130,8 @@ ensure_username(Field) ->
     emqx_connector_schema_lib:username(Field).
 
 %% ===================================================================
+resource_type() -> ldap.
+
 callback_mode() -> always_sync.
 
 -spec on_start(binary(), hocon:config()) -> {ok, state()} | {error, _}.

apps/emqx_ldap/test/emqx_ldap_SUITE.erl

@@ -96,7 +96,6 @@ perform_lifecycle_check(ResourceId, InitialConfig) ->
         state := #{pool_name := PoolName} = State,
         status := InitialStatus
     }} = emqx_resource:create_local(
-        ldap,
         ResourceId,
         ?CONNECTOR_RESOURCE_GROUP,
         ?LDAP_RESOURCE_MOD,
@@ -172,7 +171,6 @@ t_get_status(Config) ->
         ?LDAP_RESOURCE_MOD, ldap_config(Config)
     ),
     {ok, _} = emqx_resource:create_local(
-        ldap,
         ResourceId,
         ?CONNECTOR_RESOURCE_GROUP,
         ?LDAP_RESOURCE_MOD,

apps/emqx_mongodb/src/emqx_mongodb.app.src

@@ -1,6 +1,6 @@
 {application, emqx_mongodb, [
     {description, "EMQX MongoDB Connector"},
-    {vsn, "0.1.6"},
+    {vsn, "0.1.7"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_mongodb/src/emqx_mongodb.erl

@@ -26,6 +26,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -172,6 +173,7 @@ desc(_) ->
     undefined.
 
 %% ===================================================================
+resource_type() -> mongodb.
 
 callback_mode() -> always_sync.
 

apps/emqx_mongodb/test/emqx_mongodb_SUITE.erl

@@ -143,7 +143,6 @@ check_config(Config) ->
 
 create_local_resource(ResourceId, CheckedConfig) ->
     {ok, Bridge} = emqx_resource:create_local(
-        mongodb,
         ResourceId,
         ?CONNECTOR_RESOURCE_GROUP,
         ?MONGO_RESOURCE_MOD,

apps/emqx_mysql/src/emqx_mysql.app.src

@@ -1,6 +1,6 @@
 {application, emqx_mysql, [
     {description, "EMQX MySQL Database Connector"},
-    {vsn, "0.1.9"},
+    {vsn, "0.2.0"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_mysql/src/emqx_mysql.erl

@@ -25,6 +25,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -91,6 +92,8 @@ server() ->
     emqx_schema:servers_sc(Meta, ?MYSQL_HOST_OPTIONS).
 
 %% ===================================================================
+resource_type() -> mysql.
+
 callback_mode() -> always_sync.
 
 -spec on_start(binary(), hocon:config()) -> {ok, state()} | {error, _}.

apps/emqx_mysql/test/emqx_mysql_SUITE.erl

@@ -67,7 +67,6 @@ perform_lifecycle_check(ResourceId, InitialConfig) ->
         state := #{pool_name := PoolName} = State,
         status := InitialStatus
     }} = emqx_resource:create_local(
-        mysql,
         ResourceId,
         ?CONNECTOR_RESOURCE_GROUP,
         ?MYSQL_RESOURCE_MOD,

apps/emqx_oracle/src/emqx_oracle.app.src

@@ -1,6 +1,6 @@
 {application, emqx_oracle, [
     {description, "EMQX Enterprise Oracle Database Connector"},
-    {vsn, "0.2.2"},
+    {vsn, "0.2.3"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_oracle/src/emqx_oracle.erl

@@ -21,6 +21,7 @@
 
 %% callbacks for behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -67,6 +68,8 @@
         batch_params_tokens := params_tokens()
     }.
 
+resource_type() -> oracle.
+
 % As ecpool is not monitoring the worker's PID when doing a handover_async, the
 % request can be lost if worker crashes. Thus, it's better to force requests to
 % be sync for now.

apps/emqx_postgresql/src/emqx_postgresql.app.src

@@ -1,6 +1,6 @@
 {application, emqx_postgresql, [
     {description, "EMQX PostgreSQL Database Connector"},
-    {vsn, "0.2.2"},
+    {vsn, "0.2.3"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_postgresql/src/emqx_postgresql.erl

@@ -29,6 +29,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -120,6 +121,8 @@ adjust_fields(Fields) ->
     ).
 
 %% ===================================================================
+resource_type() -> postgresql.
+
 callback_mode() -> always_sync.
 
 -spec on_start(binary(), hocon:config()) -> {ok, state()} | {error, _}.

apps/emqx_postgresql/test/emqx_postgresql_SUITE.erl

@@ -75,7 +75,6 @@ perform_lifecycle_check(ResourceId, InitialConfig) ->
         status := InitialStatus
     }} =
         emqx_resource:create_local(
-            postgresql,
             ResourceId,
             ?CONNECTOR_RESOURCE_GROUP,
             ?PGSQL_RESOURCE_MOD,

apps/emqx_redis/src/emqx_redis.erl

@@ -28,6 +28,7 @@
 
 %% callbacks of behaviour emqx_resource
 -export([
+    resource_type/0,
     callback_mode/0,
     on_start/2,
     on_stop/2,
@@ -119,6 +120,8 @@ redis_type(Type) ->
         desc => ?DESC(Type)
     }}.
 
+resource_type() -> redis.
+
 callback_mode() -> always_sync.
 
 on_start(InstId, Config0) ->