fix: revert "feat(iotdb): introduced a thrift driver for IotDB connector"

This reverts commit 75b5d400730f1e7c70b783aa2b0195b38325ff87.

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

@@ -88,33 +88,3 @@ services:
     #     - "18080:18080"
     networks:
       - emqx_bridge
-
-  iotdb-thrift:
-    container_name: iotdb-thrift
-    hostname: iotdb-thrift
-    image: apache/iotdb:1.3.0-standalone
-    restart: always
-    environment:
-      - enable_rest_service=true
-      - cn_internal_address=iotdb-thrift
-      - cn_internal_port=10710
-      - cn_consensus_port=10720
-      - cn_seed_config_node=iotdb-thrift:10710
-      - dn_rpc_address=iotdb-thrift
-      - dn_internal_address=iotdb-thrift
-      - dn_rpc_port=6667
-      - dn_mpp_data_exchange_port=10740
-      - dn_schema_region_consensus_port=10750
-      - dn_data_region_consensus_port=10760
-      - dn_seed_config_node=iotdb-thrift:10710
-    # volumes:
-    #     - ./data:/iotdb/data
-    #     - ./logs:/iotdb/logs
-    expose:
-      - "18080"
-      - "6667"
-    # IoTDB's REST interface, uncomment for local testing
-    # ports:
-    #     - "18080:18080"
-    networks:
-      - emqx_bridge

.ci/docker-compose-file/toxiproxy.json

@@ -156,18 +156,6 @@
     "upstream": "iotdb013:18080",
     "enabled": true
   },
-  {
-    "name": "iotdb_thrift",
-    "listen": "0.0.0.0:46667",
-    "upstream": "iotdb-thrift:6667",
-    "enabled": true
-  },
-  {
-    "name": "iotdb_thrift_rest",
-    "listen": "0.0.0.0:48080",
-    "upstream": "iotdb-thrift:18080",
-    "enabled": true
-  },
   {
     "name": "minio_tcp",
     "listen": "0.0.0.0:19000",

apps/emqx_bridge/test/emqx_bridge_v2_testlib.erl

@@ -988,8 +988,6 @@ t_consume(Config, Opts) ->
     ok.
 
 t_create_via_http(Config) ->
-    t_create_via_http(Config, false).
-t_create_via_http(Config, IsOnlyV2) ->
     ?check_trace(
         begin
             ?assertMatch({ok, _}, create_bridge_api(Config)),
@@ -1002,11 +1000,10 @@ t_create_via_http(Config, IsOnlyV2) ->
             ),
 
             %% check that v1 list API is fine
-            (not IsOnlyV2) andalso
-                ?assertMatch(
-                    {ok, {{_, 200, _}, _, _}},
-                    list_bridges_http_api_v1()
-                ),
+            ?assertMatch(
+                {ok, {{_, 200, _}, _, _}},
+                list_bridges_http_api_v1()
+            ),
 
             ok
         end,

apps/emqx_bridge/test/emqx_bridge_v2_tests.erl

@@ -75,27 +75,10 @@ connector_resource_opts_test() ->
         start_timeout
     ],
     ConnectorSchemasRefs =
-        lists:foldl(
-            fun
-                ({Type, #{type := ?MAP(_, ?R_REF(SchemaMod, FieldName))}}, Acc) ->
-                    [{Type, find_resource_opts_fields(SchemaMod, FieldName)} | Acc];
-                ({Type, #{type := ?MAP(_, ?UNION(UnionType))}}, Acc) ->
-                    Types =
-                        case UnionType of
-                            List when is_list(List) ->
-                                List;
-                            Func when is_function(Func, 1) ->
-                                Func(all_union_members)
-                        end,
-                    lists:foldl(
-                        fun(?R_REF(SchemaMod, FieldName), InAcc) ->
-                            [{Type, find_resource_opts_fields(SchemaMod, FieldName)} | InAcc]
-                        end,
-                        Acc,
-                        Types
-                    )
+        lists:map(
+            fun({Type, #{type := ?MAP(_, ?R_REF(SchemaMod, FieldName))}}) ->
+                {Type, find_resource_opts_fields(SchemaMod, FieldName)}
             end,
-            [],
             emqx_connector_schema:fields(connectors)
         ),
     ConnectorsMissingRO = [Type || {Type, undefined} <- ConnectorSchemasRefs],

apps/emqx_bridge_iotdb/include/emqx_bridge_iotdb.hrl

@@ -11,16 +11,4 @@
 -define(VSN_1_0_X, 'v1.0.x').
 -define(VSN_0_13_X, 'v0.13.x').
 
--define(THRIFT_HOST_OPTIONS, #{
-    default_port => 6667
-}).
-
--define(PROTOCOL_V1, 'protocol_v1').
--define(PROTOCOL_V2, 'protocol_v2').
--define(PROTOCOL_V3, 'protocol_v3').
-
--define(THRIFT_NOT_SUPPORT_ASYNC_MSG, <<"The Thrift backend does not support asynchronous calls">>).
-
--type driver() :: resetapi | thrift.
-
 -endif.

apps/emqx_bridge_iotdb/mix.exs

@@ -5,7 +5,7 @@ defmodule EMQXBridgeIotdb.MixProject do
   def project do
     [
       app: :emqx_bridge_iotdb,
-      version: "0.2.3",
+      version: "0.1.0",
       build_path: "../../_build",
       erlc_options: UMP.erlc_options(),
       erlc_paths: UMP.erlc_paths(),
@@ -23,11 +23,11 @@ defmodule EMQXBridgeIotdb.MixProject do
 
   def deps() do
     [
-      {:iotdb, github: "emqx/iotdb-client-erl", tag: "0.1.4"},
+      {:emqx, in_umbrella: true},
       {:emqx_connector, in_umbrella: true, runtime: false},
       {:emqx_resource, in_umbrella: true},
       {:emqx_bridge, in_umbrella: true, runtime: false},
-      {:emqx_bridge_http, in_umbrella: true, runtime: false}
+      {:emqx_bridge_http, in_umbrella: true}
     ]
   end
 end

apps/emqx_bridge_iotdb/rebar.config

@@ -9,8 +9,7 @@
     {emqx_connector, {path, "../../apps/emqx_connector"}},
     {emqx_resource, {path, "../../apps/emqx_resource"}},
     {emqx_bridge, {path, "../../apps/emqx_bridge"}},
-    {emqx_bridge_http, {path, "../emqx_bridge_http"}},
-    {iotdb, {git, "https://github.com/emqx/iotdb-client-erl.git", {tag, "0.1.4"}}}
+    {emqx_bridge_http, {path, "../emqx_bridge_http"}}
 ]}.
 {plugins, [rebar3_path_deps]}.
 {project_plugins, [erlfmt]}.

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src

@@ -10,8 +10,7 @@
     {applications, [
         kernel,
         stdlib,
-        emqx_resource,
-        iotdb
+        emqx_resource
     ]},
     {env, [
         {emqx_action_info_modules, [emqx_bridge_iotdb_action_info]},

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_action_info.erl

@@ -53,7 +53,7 @@ bridge_v1_config_to_action_config(BridgeV1Config, ConnectorName) ->
     ).
 
 bridge_v1_config_to_connector_config(BridgeV1Config) ->
-    ConnectorKeys = schema_keys(emqx_bridge_iotdb_connector, "config_restapi"),
+    ConnectorKeys = schema_keys(emqx_bridge_iotdb_connector, config),
     emqx_utils_maps:update_if_present(
         <<"resource_opts">>,
         fun emqx_connector_schema:project_to_connector_resource_opts/1,

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_connector.erl

@@ -17,7 +17,6 @@
 -export([
     resource_type/0,
     callback_mode/0,
-    callback_mode/1,
     on_start/2,
     on_stop/2,
     on_get_status/2,
@@ -32,8 +31,6 @@
     on_format_query_result/1
 ]).
 
--export([connect/1, do_get_status/1]).
-
 -export([
     namespace/0,
     roots/0,
@@ -48,8 +45,7 @@
 
 -type config() ::
     #{
-        driver := driver(),
-        request_base => #{
+        request_base := #{
             scheme := http | https,
             host := iolist(),
             port := inet:port_number()
@@ -57,17 +53,18 @@
         connect_timeout := pos_integer(),
         pool_type := random | hash,
         pool_size := pos_integer(),
-        iotdb_version => atom(),
-        protocol_version => atom(),
+        iotdb_version := atom(),
         request => undefined | map(),
         atom() => _
     }.
 
 -type state() ::
     #{
-        driver := driver(),
+        connect_timeout := pos_integer(),
+        pool_type := random | hash,
         channels := map(),
         iotdb_version := atom(),
+        request => undefined | map(),
         atom() => _
     }.
 
@@ -118,13 +115,13 @@ connector_example_values() ->
 namespace() -> "iotdb".
 
 roots() ->
-    [].
+    [{config, #{type => hoconsc:ref(?MODULE, config)}}].
 
-fields("config_restapi") ->
+fields(config) ->
     proplists_without(
         [url, request, retry_interval, headers],
         emqx_bridge_http_schema:fields("config_connector")
-    ) ++ common_fields(restapi) ++
+    ) ++
         fields("connection_fields");
 fields("connection_fields") ->
     [
@@ -146,88 +143,38 @@ fields("connection_fields") ->
             )},
         {authentication,
             mk(
-                hoconsc:union([ref(?MODULE, authentication)]),
+                hoconsc:union([ref(?MODULE, auth_basic)]),
                 #{
                     default => auth_basic, desc => ?DESC("config_authentication")
                 }
             )}
     ];
-fields(authentication) ->
+fields(auth_basic) ->
     [
-        {username, mk(binary(), #{required => true, desc => ?DESC("config_auth_username")})},
+        {username, mk(binary(), #{required => true, desc => ?DESC("config_auth_basic_username")})},
         {password,
             emqx_schema_secret:mk(#{
                 required => true,
-                desc => ?DESC("config_auth_password")
+                desc => ?DESC("config_auth_basic_password")
             })}
     ];
-fields("config_thrift") ->
-    Meta = #{desc => ?DESC("server")},
-    emqx_connector_schema:common_fields() ++
-        common_fields(thrift) ++
-        [
-            {server, emqx_schema:servers_sc(Meta, ?THRIFT_HOST_OPTIONS)},
-            {protocol_version,
-                mk(
-                    hoconsc:enum([?PROTOCOL_V1, ?PROTOCOL_V2, ?PROTOCOL_V3]),
-                    #{
-                        desc => ?DESC("config_protocol_version"),
-                        default => ?PROTOCOL_V3
-                    }
-                )},
-            {'zoneId',
-                mk(
-                    binary(),
-                    #{default => <<"Asia/Shanghai">>, desc => ?DESC("config_zoneId")}
-                )},
-            {pool_size,
-                mk(
-                    pos_integer(),
-                    #{
-                        default => 8,
-                        desc => ?DESC("pool_size")
-                    }
-                )}
-        ] ++ fields(authentication) ++ emqx_connector_schema_lib:ssl_fields() ++
-        emqx_connector_schema:resource_opts_ref(?MODULE, connector_resource_opts);
-fields(connector_resource_opts) ->
-    emqx_connector_schema:resource_opts_fields();
-fields("post_" ++ Driver) ->
-    emqx_connector_schema:type_and_name_fields(enum([iotdb])) ++ fields("config_" ++ Driver);
-fields("put_" ++ Driver) ->
-    fields("config_" ++ Driver);
-fields("get_" ++ Driver) ->
-    emqx_bridge_schema:status_fields() ++ fields("post_" ++ Driver).
-
-common_fields(Driver) ->
-    [
-        {driver,
-            mk(
-                hoconsc:enum([Driver]),
-                #{
-                    desc => ?DESC("config_driver"),
-                    default => <<"restapi">>
-                }
-            )}
-    ].
-
-desc(authentication) ->
-    ?DESC("config_authentication");
-desc(connector_resource_opts) ->
-    "Connector resource options";
-desc(Struct) when is_list(Struct) ->
-    case string:split(Struct, "_") of
-        ["config", _] ->
-            ?DESC("desc_config");
-        [Method, _] when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
-            ["Configuration for IoTDB using `", string:to_upper(Method), "` method."];
-        _ ->
-            undefined
-    end;
+fields("post") ->
+    emqx_connector_schema:type_and_name_fields(enum([iotdb])) ++ fields(config);
+fields("put") ->
+    fields(config);
+fields("get") ->
+    emqx_bridge_schema:status_fields() ++ fields("post").
+
+desc(config) ->
+    ?DESC("desc_config");
+desc(auth_basic) ->
+    "Basic Authentication";
+desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
+    ["Configuration for IoTDB using `", string:to_upper(Method), "` method."];
 desc(_) ->
     undefined.
 
-connector_config(#{driver := restapi} = Conf, #{name := Name, parse_confs := ParseConfs}) ->
+connector_config(Conf, #{name := Name, parse_confs := ParseConfs}) ->
     #{
         base_url := BaseUrl,
         authentication :=
@@ -252,9 +199,7 @@ connector_config(#{driver := restapi} = Conf, #{name := Name, parse_confs := Par
         <<"http">>,
         Name,
         WebhookConfig
-    );
-connector_config(Conf, _) ->
-    Conf.
+    ).
 
 proplists_without(Keys, List) ->
     [El || El = {K, _} <- List, not lists:member(K, Keys)].
@@ -266,13 +211,8 @@ resource_type() -> iotdb.
 
 callback_mode() -> async_if_possible.
 
-callback_mode(#{driver := restapi}) ->
-    async_if_possible;
-callback_mode(#{driver := thrift}) ->
-    always_sync.
-
 -spec on_start(manager_id(), config()) -> {ok, state()} | no_return().
-on_start(InstanceId, #{driver := restapi, iotdb_version := Version} = Config) ->
+on_start(InstanceId, #{iotdb_version := Version} = Config) ->
     %% [FIXME] The configuration passed in here is pre-processed and transformed
     %% in emqx_bridge_resource:parse_confs/2.
     case emqx_bridge_http_connector:on_start(InstanceId, Config) of
@@ -282,78 +222,8 @@ on_start(InstanceId, #{driver := restapi, iotdb_version := Version} = Config) ->
                 instance_id => InstanceId,
                 request => emqx_utils:redact(maps:get(request, State, <<>>))
             }),
-            ?tp(iotdb_bridge_started, #{driver => restapi, instance_id => InstanceId}),
-            {ok, State#{driver => restapi, iotdb_version => Version, channels => #{}}};
-        {error, Reason} ->
-            ?SLOG(error, #{
-                msg => "failed_to_start_iotdb_bridge",
-                instance_id => InstanceId,
-                request => emqx_utils:redact(maps:get(request, Config, <<>>)),
-                reason => Reason
-            }),
-            throw(failed_to_start_iotdb_bridge)
-    end;
-on_start(
-    InstanceId,
-    #{
-        driver := thrift,
-        protocol_version := ProtocolVsn,
-        server := Server,
-        pool_size := PoolSize,
-        ssl := SSL
-    } = Config
-) ->
-    IoTDBOpts0 = maps:with(['zoneId', username, password], Config),
-
-    Version =
-        case ProtocolVsn of
-            ?PROTOCOL_V1 ->
-                0;
-            ?PROTOCOL_V2 ->
-                1;
-            ?PROTOCOL_V3 ->
-                2
-        end,
-
-    #{hostname := Host, port := Port} = emqx_schema:parse_server(Server, ?THRIFT_HOST_OPTIONS),
-
-    TransportOpts =
-        case maps:get(enable, SSL) of
-            true ->
-                #{
-                    ssltransport => true,
-                    ssloptions => emqx_tls_lib:to_client_opts(SSL)
-                };
-            false ->
-                #{}
-        end,
-
-    IoTDBOpts = IoTDBOpts0#{
-        version => Version,
-        host => Host,
-        port => Port,
-        options => TransportOpts
-    },
-
-    Options = [
-        {pool_size, PoolSize},
-        {iotdb_options, IoTDBOpts}
-    ],
-
-    case emqx_resource_pool:start(InstanceId, ?MODULE, Options) of
-        ok ->
-            ?SLOG(info, #{
-                msg => "iotdb_bridge_started",
-                instance_id => InstanceId
-            }),
-
-            ?tp(iotdb_bridge_started, #{driver => thrift, instance_id => InstanceId}),
-
-            {ok, #{
-                driver => thrift,
-                iotdb_version => ProtocolVsn,
-                channels => #{}
-            }};
+            ?tp(iotdb_bridge_started, #{instance_id => InstanceId}),
+            {ok, State#{iotdb_version => Version, channels => #{}}};
         {error, Reason} ->
             ?SLOG(error, #{
                 msg => "failed_to_start_iotdb_bridge",
@@ -365,26 +235,18 @@ on_start(
     end.
 
 -spec on_stop(manager_id(), state()) -> ok | {error, term()}.
-on_stop(InstanceId, #{driver := restapi} = State) ->
+on_stop(InstanceId, State) ->
     ?SLOG(info, #{
         msg => "stopping_iotdb_bridge",
         connector => InstanceId
     }),
     Res = emqx_bridge_http_connector:on_stop(InstanceId, State),
     ?tp(iotdb_bridge_stopped, #{instance_id => InstanceId}),
-    Res;
-on_stop(InstanceId, #{driver := thrift} = _State) ->
-    ?SLOG(info, #{
-        msg => "stopping_iotdb_bridge",
-        connector => InstanceId
-    }),
-
-    ?tp(iotdb_bridge_stopped, #{instance_id => InstanceId}),
-    emqx_resource_pool:stop(InstanceId).
+    Res.
 
 -spec on_get_status(manager_id(), state()) ->
     connected | connecting | {disconnected, state(), term()}.
-on_get_status(InstanceId, #{driver := restapi} = State) ->
+on_get_status(InstanceId, State) ->
     Func = fun(Worker, Timeout) ->
         Request = {?IOTDB_PING_PATH, [], undefined},
         NRequest = emqx_bridge_http_connector:formalize_request(get, Request),
@@ -403,27 +265,7 @@ on_get_status(InstanceId, #{driver := restapi} = State) ->
                 {error, {unexpected_ping_result, Result}}
         end
     end,
-    emqx_bridge_http_connector:on_get_status(InstanceId, State, Func);
-on_get_status(InstanceId, #{driver := thrift} = _State) ->
-    case emqx_resource_pool:health_check_workers(InstanceId, fun ?MODULE:do_get_status/1) of
-        true ->
-            ?status_connected;
-        false ->
-            ?status_disconnected
-    end.
-
-do_get_status(Conn) ->
-    case iotdb:ping(Conn) of
-        {ok, _} ->
-            true;
-        {error, _} ->
-            false
-    end.
-
-connect(Opts) ->
-    {iotdb_options, #{password := Password} = IoTDBOpts0} = lists:keyfind(iotdb_options, 1, Opts),
-    IoTDBOpts = IoTDBOpts0#{password := emqx_secret:unwrap(Password)},
-    iotdb:start_link(IoTDBOpts).
+    emqx_bridge_http_connector:on_get_status(InstanceId, State, Func).
 
 -spec on_query(manager_id(), {send_message, map()}, state()) ->
     {ok, pos_integer(), [term()], term()}
@@ -445,8 +287,8 @@ on_query(
     case try_render_messages([Req], IoTDBVsn, Channels) of
         {ok, [IoTDBPayload]} ->
             handle_response(
-                do_on_query(
-                    InstanceId, ChannelId, IoTDBPayload, State
+                emqx_bridge_http_connector:on_query(
+                    InstanceId, {ChannelId, IoTDBPayload}, State
                 )
             );
         Error ->
@@ -459,7 +301,7 @@ on_query_async(
     InstanceId,
     {ChannelId, _Message} = Req,
     ReplyFunAndArgs0,
-    #{driver := restapi, iotdb_version := IoTDBVsn, channels := Channels} = State
+    #{iotdb_version := IoTDBVsn, channels := Channels} = State
 ) ->
     ?tp(iotdb_bridge_on_query_async, #{instance_id => InstanceId}),
     ?SLOG(debug, #{
@@ -483,27 +325,13 @@ on_query_async(
             );
         Error ->
             Error
-    end;
-on_query_async(
-    InstanceId,
-    Req,
-    _ReplyFunAndArgs0,
-    #{driver := thrift} = State
-) ->
-    ?SLOG(error, #{
-        msg => "iotdb_bridge_async_query_failed",
-        instance_id => InstanceId,
-        send_message => Req,
-        reason => ?THRIFT_NOT_SUPPORT_ASYNC_MSG,
-        state => emqx_utils:redact(State)
-    }),
-    {error, not_support}.
+    end.
 
 on_batch_query_async(
     InstId,
     Requests,
     Callback,
-    #{driver := restapi, iotdb_version := IoTDBVsn, channels := Channels} = State
+    #{iotdb_version := IoTDBVsn, channels := Channels} = State
 ) ->
     ?tp(iotdb_bridge_on_batch_query_async, #{instance_id => InstId}),
     [{ChannelId, _Message} | _] = Requests,
@@ -533,23 +361,8 @@ on_batch_query_async(
             );
         Error ->
             Error
-    end;
-on_batch_query_async(
-    InstanceId,
-    Req,
-    _ReplyFunAndArgs0,
-    #{driver := thrift} = State
-) ->
-    ?SLOG(error, #{
-        msg => "iotdb_bridge_async_query_failed",
-        instance_id => InstanceId,
-        send_message => Req,
-        reason => ?THRIFT_NOT_SUPPORT_ASYNC_MSG,
-        state => emqx_utils:redact(State)
-    }),
-    {error, not_support}.
+    end.
 
-%% todo
 on_batch_query(
     InstId,
     [{ChannelId, _Message}] = Requests,
@@ -568,8 +381,8 @@ on_batch_query(
             lists:map(
                 fun(IoTDBPayload) ->
                     handle_response(
-                        do_on_query(
-                            InstId, ChannelId, IoTDBPayload, State
+                        emqx_bridge_http_connector:on_query(
+                            InstId, {ChannelId, IoTDBPayload}, State
                         )
                     )
                 end,
@@ -584,7 +397,7 @@ on_format_query_result(Result) ->
 
 on_add_channel(
     InstanceId,
-    #{driver := restapi, iotdb_version := Version, channels := Channels} = OldState0,
+    #{iotdb_version := Version, channels := Channels} = OldState0,
     ChannelId,
     #{
         parameters := #{data := Data} = Parameter
@@ -616,27 +429,6 @@ on_add_channel(
                 InstanceId, OldState0, ChannelId, HTTPReq
             ),
 
-            %% update IoTDB channel
-            DeviceId = maps:get(device_id, Parameter, <<>>),
-            Channel = Parameter#{
-                device_id => emqx_placeholder:preproc_tmpl(DeviceId),
-                data := preproc_data_template(Data)
-            },
-            Channels2 = Channels#{ChannelId => Channel},
-            {ok, OldState#{channels := Channels2}}
-    end;
-on_add_channel(
-    _InstanceId,
-    #{driver := thrift, channels := Channels} = OldState,
-    ChannelId,
-    #{
-        parameters := #{data := Data} = Parameter
-    }
-) ->
-    case maps:is_key(ChannelId, Channels) of
-        true ->
-            {error, already_exists};
-        _ ->
             %% update IoTDB channel
             DeviceId = maps:get(device_id, Parameter, <<>>),
             Channel = Parameter#{
@@ -647,11 +439,8 @@ on_add_channel(
             {ok, OldState#{channels := Channels2}}
     end.
 
-on_remove_channel(InstanceId, #{driver := restapi, channels := Channels} = OldState0, ChannelId) ->
+on_remove_channel(InstanceId, #{channels := Channels} = OldState0, ChannelId) ->
     {ok, OldState} = emqx_bridge_http_connector:on_remove_channel(InstanceId, OldState0, ChannelId),
-    Channels2 = maps:remove(ChannelId, Channels),
-    {ok, OldState#{channels => Channels2}};
-on_remove_channel(_InstanceId, #{driver := thrift, channels := Channels} = OldState, ChannelId) ->
     Channels2 = maps:remove(ChannelId, Channels),
     {ok, OldState#{channels => Channels2}}.
 
@@ -747,16 +536,16 @@ proc_data(
     ],
     Msg,
     Nows,
-    IoTDbVsn,
+    IotDbVsn,
     Acc
 ) ->
     DataType = list_to_binary(
         string:uppercase(binary_to_list(emqx_placeholder:proc_tmpl(DataType0, Msg)))
     ),
     try
-        proc_data(T, Msg, Nows, IoTDbVsn, [
+        proc_data(T, Msg, Nows, IotDbVsn, [
             #{
-                timestamp => iot_timestamp(IoTDbVsn, TimestampTkn, Msg, Nows),
+                timestamp => iot_timestamp(IotDbVsn, TimestampTkn, Msg, Nows),
                 measurement => emqx_placeholder:proc_tmpl(Measurement, Msg),
                 data_type => DataType,
                 value => proc_value(DataType, ValueTkn, Msg)
@@ -770,28 +559,28 @@ proc_data(
             ?SLOG(debug, #{exception => Error, reason => Reason, stacktrace => Stacktrace}),
             {error, invalid_data}
     end;
-proc_data([], _Msg, _Nows, _IoTDbVsn, Acc) ->
+proc_data([], _Msg, _Nows, _IotDbVsn, Acc) ->
     {ok, lists:reverse(Acc)}.
 
-iot_timestamp(_IoTDbVsn, Timestamp, _, _) when is_integer(Timestamp) ->
+iot_timestamp(_IotDbVsn, Timestamp, _, _) when is_integer(Timestamp) ->
     Timestamp;
-iot_timestamp(IoTDbVsn, TimestampTkn, Msg, Nows) ->
-    iot_timestamp(IoTDbVsn, emqx_placeholder:proc_tmpl(TimestampTkn, Msg), Nows).
+iot_timestamp(IotDbVsn, TimestampTkn, Msg, Nows) ->
+    iot_timestamp(IotDbVsn, emqx_placeholder:proc_tmpl(TimestampTkn, Msg), Nows).
 
 %% > v1.3.0 don't allow write nanoseconds nor microseconds
 iot_timestamp(?VSN_1_3_X, <<"now_us">>, #{now_ms := NowMs}) ->
     NowMs;
 iot_timestamp(?VSN_1_3_X, <<"now_ns">>, #{now_ms := NowMs}) ->
     NowMs;
-iot_timestamp(_IoTDbVsn, <<"now_us">>, #{now_us := NowUs}) ->
+iot_timestamp(_IotDbVsn, <<"now_us">>, #{now_us := NowUs}) ->
     NowUs;
-iot_timestamp(_IoTDbVsn, <<"now_ns">>, #{now_ns := NowNs}) ->
+iot_timestamp(_IotDbVsn, <<"now_ns">>, #{now_ns := NowNs}) ->
     NowNs;
-iot_timestamp(_IoTDbVsn, Timestamp, #{now_ms := NowMs}) when
+iot_timestamp(_IotDbVsn, Timestamp, #{now_ms := NowMs}) when
     Timestamp =:= <<"now">>; Timestamp =:= <<"now_ms">>; Timestamp =:= <<>>
 ->
     NowMs;
-iot_timestamp(_IoTDbVsn, Timestamp, _) when is_binary(Timestamp) ->
+iot_timestamp(_IotDbVsn, Timestamp, _) when is_binary(Timestamp) ->
     binary_to_integer(Timestamp).
 
 proc_value(<<"TEXT">>, ValueTkn, Msg) ->
@@ -930,10 +719,6 @@ iotdb_field_key(is_aligned, ?VSN_1_0_X) ->
     <<"is_aligned">>;
 iotdb_field_key(is_aligned, ?VSN_0_13_X) ->
     <<"isAligned">>;
-iotdb_field_key(is_aligned, Vsn) when
-    Vsn == ?PROTOCOL_V1; Vsn == ?PROTOCOL_V2; Vsn == ?PROTOCOL_V3
-->
-    'isAligned';
 iotdb_field_key(device_id, ?VSN_1_3_X) ->
     <<"device">>;
 iotdb_field_key(device_id, ?VSN_1_1_X) ->
@@ -942,10 +727,6 @@ iotdb_field_key(device_id, ?VSN_1_0_X) ->
     <<"device">>;
 iotdb_field_key(device_id, ?VSN_0_13_X) ->
     <<"deviceId">>;
-iotdb_field_key(device_id, Vsn) when
-    Vsn == ?PROTOCOL_V1; Vsn == ?PROTOCOL_V2; Vsn == ?PROTOCOL_V3
-->
-    'deviceId';
 iotdb_field_key(data_types, ?VSN_1_3_X) ->
     <<"data_types">>;
 iotdb_field_key(data_types, ?VSN_1_1_X) ->
@@ -953,11 +734,7 @@ iotdb_field_key(data_types, ?VSN_1_1_X) ->
 iotdb_field_key(data_types, ?VSN_1_0_X) ->
     <<"data_types">>;
 iotdb_field_key(data_types, ?VSN_0_13_X) ->
-    <<"dataTypes">>;
-iotdb_field_key(data_types, Vsn) when
-    Vsn == ?PROTOCOL_V1; Vsn == ?PROTOCOL_V2; Vsn == ?PROTOCOL_V3
-->
-    dtypes.
+    <<"dataTypes">>.
 
 to_list(List) when is_list(List) -> List;
 to_list(Data) -> [Data].
@@ -979,8 +756,6 @@ handle_response({ok, Code, _Headers, Body}) ->
     {error, #{code => Code, body => Body}};
 handle_response({ok, Code, Body}) ->
     {error, #{code => Code, body => Body}};
-handle_response({ok, _} = Resp) ->
-    Resp;
 handle_response({error, _} = Error) ->
     Error.
 
@@ -1074,8 +849,3 @@ get_data_template(#{data := Data}, _Payloads) when Data =/= [] ->
 %% This is a self-describing message
 get_data_template(#{data := []}, Payloads) ->
     preproc_data_list(Payloads).
-
-do_on_query(InstanceId, ChannelId, Data, #{driver := restapi} = State) ->
-    emqx_bridge_http_connector:on_query(InstanceId, {ChannelId, Data}, State);
-do_on_query(InstanceId, _ChannelId, Data, #{driver := thrift} = _State) ->
-    ecpool:pick_and_do(InstanceId, {iotdb, insert_tablet, [Data]}, no_handover).

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_connector_info.erl

@@ -16,10 +16,6 @@
     api_schema/1
 ]).
 
--define(CONNECTOR, emqx_bridge_iotdb_connector).
--define(DRIVER_REST, "restapi").
--define(DRIVER_THRIFT, "thrift").
-
 type_name() ->
     iotdb.
 
@@ -35,7 +31,7 @@ config_transform_module() ->
 config_schema() ->
     {iotdb,
         hoconsc:mk(
-            hoconsc:map(name, hoconsc:union(fun driver_union_selector/1)),
+            hoconsc:map(name, hoconsc:ref(emqx_bridge_iotdb_connector, config)),
             #{
                 desc => <<"IoTDB Connector Config">>,
                 required => false
@@ -46,37 +42,6 @@ schema_module() ->
     emqx_bridge_iotdb_connector.
 
 api_schema(Method) ->
-    {<<"iotdb">>, hoconsc:union(mk_api_union_selector(Method))}.
-
-driver_union_selector(all_union_members) ->
-    [
-        ref(?DRIVER_REST, "config"),
-        ref(?DRIVER_THRIFT, "config")
-    ];
-driver_union_selector({value, Value}) ->
-    case Value of
-        #{<<"driver">> := <<"thrift">>} ->
-            [ref(?DRIVER_THRIFT, "config")];
-        _ ->
-            [ref(?DRIVER_REST, "config")]
-    end.
-
-mk_api_union_selector(Method) ->
-    fun
-        (all_union_members) ->
-            [
-                ref(?DRIVER_REST, Method),
-                ref(?DRIVER_THRIFT, Method)
-            ];
-        ({value, Value}) ->
-            case Value of
-                #{<<"driver">> := <<"thrift">>} ->
-                    [ref(?DRIVER_THRIFT, Method)];
-                _ ->
-                    [ref(?DRIVER_REST, Method)]
-            end
-    end.
-
-ref(Driver, Field) ->
-    Name = Field ++ "_" ++ Driver,
-    hoconsc:ref(?CONNECTOR, Name).
+    emqx_connector_schema:api_ref(
+        emqx_bridge_iotdb_connector, <<"iotdb">>, Method
+    ).

apps/emqx_bridge_iotdb/test/emqx_bridge_iotdb_impl_SUITE.erl

@@ -21,25 +21,15 @@ all() ->
     [
         {group, iotdb110},
         {group, iotdb130},
-        {group, legacy},
-        {group, thrift}
+        {group, legacy}
     ].
 
 groups() ->
     AllTCs = emqx_common_test_helpers:all(?MODULE),
-    Async = [
-        t_async_device_id_missing,
-        t_async_invalid_template,
-        t_async_query,
-        t_extract_device_id_from_rule_engine_message,
-        %%todo
-        t_sync_query_aggregated
-    ],
     [
         {iotdb110, AllTCs},
         {iotdb130, AllTCs},
-        {legacy, AllTCs},
-        {thrift, AllTCs -- Async}
+        {legacy, AllTCs}
     ].
 
 init_per_suite(Config) ->
@@ -75,7 +65,6 @@ init_per_group(Type, Config0) when Type =:= iotdb110 orelse Type =:= iotdb130 ->
             [
                 {bridge_host, Host},
                 {bridge_port, Port},
-                {rest_port, Port},
                 {proxy_name, ProxyName},
                 {iotdb_version, IotDbVersion},
                 {iotdb_rest_prefix, <<"/rest/v2/">>}
@@ -99,7 +88,6 @@ init_per_group(legacy = Type, Config0) ->
             [
                 {bridge_host, Host},
                 {bridge_port, Port},
-                {rest_port, Port},
                 {proxy_name, ProxyName},
                 {iotdb_version, ?VSN_0_13_X},
                 {iotdb_rest_prefix, <<"/rest/v1/">>}
@@ -113,30 +101,6 @@ init_per_group(legacy = Type, Config0) ->
                     {skip, no_iotdb}
             end
     end;
-init_per_group(thrift = Type, Config0) ->
-    Host = os:getenv("IOTDB_THRIFT_HOST", "toxiproxy.emqx.net"),
-    Port = list_to_integer(os:getenv("IOTDB_THRIFT_PORT", "46667")),
-    ProxyName = "iotdb_thrift",
-    case emqx_common_test_helpers:is_tcp_server_available(Host, Port) of
-        true ->
-            Config = emqx_bridge_v2_testlib:init_per_group(Type, ?BRIDGE_TYPE_BIN, Config0),
-            [
-                {bridge_host, Host},
-                {bridge_port, Port},
-                {rest_port, 48080},
-                {proxy_name, ProxyName},
-                {iotdb_version, ?PROTOCOL_V3},
-                {iotdb_rest_prefix, <<"/rest/v2/">>}
-                | Config
-            ];
-        false ->
-            case os:getenv("IS_CI") of
-                "yes" ->
-                    throw(no_iotdb);
-                _ ->
-                    {skip, no_iotdb}
-            end
-    end;
 init_per_group(_Group, Config) ->
     Config.
 
@@ -157,7 +121,6 @@ init_per_testcase(TestCase, Config0) ->
         (atom_to_binary(TestCase))/binary, UniqueNum/binary
     >>,
     {_ConfigString, ConnectorConfig} = connector_config(Name, Config0),
-
     {_, ActionConfig} = action_config(Name, Config0),
     Config = [
         {connector_type, Type},
@@ -266,14 +229,16 @@ iotdb_request(Config, Path, Body) ->
     iotdb_request(Config, Path, Body, #{}).
 
 iotdb_request(Config, Path, Body, Opts) ->
-    BridgeConfig = ?config(connector_config, Config),
-    Host = ?config(bridge_host, Config),
-    Port = ?config(rest_port, Config),
-    Username = <<"root">>,
-    Password = <<"root">>,
-    BaseURL = iotdb_server_url(Host, Port),
-
-    ct:pal("bridge config: ~p", [BridgeConfig]),
+    _BridgeConfig =
+        #{
+            <<"base_url">> := BaseURL,
+            <<"authentication">> := #{
+                <<"username">> := Username,
+                <<"password">> := Password
+            }
+        } =
+        ?config(connector_config, Config),
+    ct:pal("bridge config: ~p", [_BridgeConfig]),
     URL = <<BaseURL/binary, Path/binary>>,
     BasicToken = base64:encode(<<Username/binary, ":", Password/binary>>),
     Headers = [
@@ -300,8 +265,6 @@ iotdb_query(Config, Query) ->
 
 is_success_check({ok, 200, _, Body}) ->
     ?assert(is_code(200, emqx_utils_json:decode(Body)));
-is_success_check({ok, _}) ->
-    ok;
 is_success_check(Other) ->
     throw(Other).
 
@@ -340,46 +303,23 @@ connector_config(Name, Config) ->
     Version = ?config(iotdb_version, Config),
     ServerURL = iotdb_server_url(Host, Port),
     ConfigString =
-        case ?config(test_group, Config) of
-            thrift ->
-                io_lib:format(
-                    "connectors.~s.~s {\n"
-                    "  enable = true\n"
-                    "  driver = \"thrift\"\n"
-                    "  server = \"~s:~p\"\n"
-                    "  protocol_version = \"~p\"\n"
-                    "  username = \"root\"\n"
-                    "  password = \"root\"\n"
-                    "  zoneId = \"Asia/Shanghai\"\n"
-                    "  ssl.enable = false\n"
-                    "}\n",
-                    [
-                        Type,
-                        Name,
-                        Host,
-                        Port,
-                        Version
-                    ]
-                );
-            _ ->
-                io_lib:format(
-                    "connectors.~s.~s {\n"
-                    "  enable = true\n"
-                    "  base_url = \"~s\"\n"
-                    "  iotdb_version = \"~s\"\n"
-                    "  authentication = {\n"
-                    "     username = \"root\"\n"
-                    "     password = \"root\"\n"
-                    "  }\n"
-                    "}\n",
-                    [
-                        Type,
-                        Name,
-                        ServerURL,
-                        Version
-                    ]
-                )
-        end,
+        io_lib:format(
+            "connectors.~s.~s {\n"
+            "  enable = true\n"
+            "  base_url = \"~s\"\n"
+            "  iotdb_version = \"~s\"\n"
+            "  authentication = {\n"
+            "     username = \"root\"\n"
+            "     password = \"root\"\n"
+            "  }\n"
+            "}\n",
+            [
+                Type,
+                Name,
+                ServerURL,
+                Version
+            ]
+        ),
     ct:pal("ConnectorConfig:~ts~n", [ConfigString]),
     {ConfigString, parse_connector_and_check(ConfigString, Type, Name)}.
 
@@ -616,10 +556,7 @@ t_async_invalid_template(Config) ->
     ).
 
 t_create_via_http(Config) ->
-    emqx_bridge_v2_testlib:t_create_via_http(
-        Config,
-        thrift =:= ?config(test_group, Config)
-    ).
+    emqx_bridge_v2_testlib:t_create_via_http(Config).
 
 t_start_stop(Config) ->
     emqx_bridge_v2_testlib:t_start_stop(Config, iotdb_bridge_stopped).

apps/emqx_connector/src/schema/emqx_connector_schema.erl

@@ -352,12 +352,7 @@ transform_old_style_bridges_to_connector_and_actions_of_type(
         end,
         RawConfig,
         ActionConnectorTuples
-    );
-transform_old_style_bridges_to_connector_and_actions_of_type(
-    {_ConnectorType, #{type := ?MAP(_Name, _)}},
-    RawConfig
-) ->
-    RawConfig.
+    ).
 
 transform_bridges_v1_to_connectors_and_bridges_v2(RawConfig) ->
     ConnectorFields = ?MODULE:fields(connectors),
@@ -641,11 +636,10 @@ status() ->
 -include_lib("hocon/include/hocon_types.hrl").
 schema_homogeneous_test() ->
     case
-        lists:foldl(
-            fun({_Name, Schema}, Bads) ->
-                is_bad_schema(Schema, Bads)
+        lists:filtermap(
+            fun({_Name, Schema}) ->
+                is_bad_schema(Schema)
             end,
-            [],
             fields(connectors)
         )
     of
@@ -655,32 +649,7 @@ schema_homogeneous_test() ->
             throw(List)
     end.
 
-is_bad_schema(#{type := ?MAP(_, ?R_REF(Module, TypeName))}, Bads) ->
-    is_bad_schema_type(Module, TypeName, Bads);
-is_bad_schema(#{type := ?MAP(_, ?UNION(Types))}, Bads) when is_list(Types) ->
-    is_bad_schema_types(Types, Bads);
-is_bad_schema(#{type := ?MAP(_, ?UNION(Func))}, Bads) when is_function(Func, 1) ->
-    Types = Func(all_union_members),
-    is_bad_schema_types(Types, Bads).
-
-is_bad_schema_types(Types, Bads) ->
-    lists:foldl(
-        fun
-            (?R_REF(Module, TypeName), Acc) ->
-                is_bad_schema_type(Module, TypeName, Acc);
-            (Type, Acc) ->
-                [
-                    #{
-                        type => Type
-                    }
-                    | Acc
-                ]
-        end,
-        Bads,
-        Types
-    ).
-
-is_bad_schema_type(Module, TypeName, Bads) ->
+is_bad_schema(#{type := ?MAP(_, ?R_REF(Module, TypeName))}) ->
     Fields = Module:fields(TypeName),
     ExpectedFieldNames = common_field_names(),
     MissingFields = lists:filter(
@@ -688,16 +657,13 @@ is_bad_schema_type(Module, TypeName, Bads) ->
     ),
     case MissingFields of
         [] ->
-            Bads;
+            false;
         _ ->
-            [
-                #{
-                    schema_module => Module,
-                    type_name => TypeName,
-                    missing_fields => MissingFields
-                }
-                | Bads
-            ]
+            {true, #{
+                schema_module => Module,
+                type_name => TypeName,
+                missing_fields => MissingFields
+            }}
     end.
 
 common_field_names() ->

apps/emqx_resource/src/emqx_resource.erl

@@ -97,7 +97,6 @@
     %% get the callback mode of a specific module
     get_callback_mode/1,
     get_resource_type/1,
-    get_callback_mode/2,
     %% start the instance
     call_start/3,
     %% verify if the resource is working normally
@@ -161,8 +160,7 @@
     on_remove_channel/3,
     on_get_channels/1,
     query_mode/1,
-    on_format_query_result/1,
-    callback_mode/1
+    on_format_query_result/1
 ]).
 
 %% when calling emqx_resource:start/1
@@ -173,10 +171,7 @@
 -callback on_stop(resource_id(), resource_state()) -> term().
 
 %% when calling emqx_resource:get_callback_mode/1
--callback callback_mode() -> callback_mode() | undefined.
-
-%% when calling emqx_resource:get_callback_mode/1
--callback callback_mode(resource_state()) -> callback_mode().
+-callback callback_mode() -> callback_mode().
 
 %% when calling emqx_resource:query/3
 -callback on_query(resource_id(), Request :: term(), resource_state()) -> query_result().
@@ -500,15 +495,6 @@ get_callback_mode(Mod) ->
 get_resource_type(Mod) ->
     Mod:resource_type().
 
--spec get_callback_mode(module(), resource_state()) -> callback_mode() | undefined.
-get_callback_mode(Mod, State) ->
-    case erlang:function_exported(Mod, callback_mode, 1) of
-        true ->
-            Mod:callback_mode(State);
-        _ ->
-            undefined
-    end.
-
 -spec call_start(resource_id(), module(), resource_config()) ->
     {ok, resource_state()} | {error, Reason :: term()}.
 call_start(ResId, Mod, Config) ->

apps/emqx_resource/src/emqx_resource_manager.erl

@@ -795,10 +795,8 @@ start_resource(Data, From) ->
             UpdatedData1 = Data#data{status = ?status_connecting, state = ResourceState},
             %% Perform an initial health_check immediately before transitioning into a connected state
             UpdatedData2 = add_channels(UpdatedData1),
-            UpdatedData3 = maybe_update_callback_mode(UpdatedData2),
-
             Actions = maybe_reply([{state_timeout, 0, health_check}], From, ok),
-            {next_state, ?state_connecting, update_state(UpdatedData3, Data), Actions};
+            {next_state, ?state_connecting, update_state(UpdatedData2, Data), Actions};
         {error, Reason} = Err ->
             IsDryRun = emqx_resource:is_dry_run(ResId),
             ?SLOG(
@@ -838,14 +836,6 @@ add_channels(Data) ->
     ),
     Data#data{added_channels = NewChannels}.
 
-maybe_update_callback_mode(Data = #data{mod = ResourceType, state = ResourceState}) ->
-    case emqx_resource:get_callback_mode(ResourceType, ResourceState) of
-        undefined ->
-            Data;
-        CallMode ->
-            Data#data{callback_mode = CallMode}
-    end.
-
 add_channels_in_list([], Data) ->
     Data;
 add_channels_in_list([{ChannelID, ChannelConfig} | Rest], Data) ->

rel/i18n/emqx_bridge_iotdb_connector.hocon

@@ -12,17 +12,17 @@ auth_basic.desc:
 auth_basic.label:
 """Basic auth params"""
 
-config_auth_username.desc:
-"""The username as configured at the IoTDB REST/Thrift interface"""
+config_auth_basic_username.desc:
+"""The username as configured at the IoTDB REST interface"""
 
-config_auth_username.label:
-  """Auth Username"""
+config_auth_basic_username.label:
+  """HTTP Basic Auth Username"""
 
-config_auth_password.desc:
-"""The password as configured at the IoTDB/Thrift REST interface"""
+config_auth_basic_password.desc:
+"""The password as configured at the IoTDB REST interface"""
 
-config_auth_password.label:
-"""Auth Password"""
+config_auth_basic_password.label:
+"""HTTP Basic Auth Password"""
 
 config_base_url.desc:
 """The base URL of the external IoTDB service's REST interface."""
@@ -41,21 +41,4 @@ desc_config.desc:
 desc_config.label:
 """IoTDB Bridge Configuration"""
 
-config_driver.desc:
-"""The drivers for IoTDB connector, can be:
-- rest
-- thrift"""
-
-server.desc:
-"""The address of the IoTDB Thrift server (host:port)"""
-
-config_protocol_version.desc:
-"""The version of IoTDB Thrift protocol"""
-
-config_zoneId.desc:
-"""Timezone for IoTDB Thrift session"""
-
-pool_size.desc:
-"""Size for the IoTDB Thrift connection pool"""
-
 }