test: make all emqx_bridge_greptimedb_SUITE tests passing

apps/emqx_bridge_greptimedb/src/emqx_bridge_greptimedb_connector.erl

@@ -81,7 +81,7 @@ on_query(InstId, {send_message, Data}, _State = #{write_syntax := SyntaxLines, c
                 #{batch => false, mode => sync, error => ErrorPoints}
             ),
             log_error_points(InstId, ErrorPoints),
-            ErrorPoints
+            {error, ErrorPoints}
     end.
 
 %% Once a Batched Data trans to points failed.
@@ -463,7 +463,7 @@ parse_timestamp([TsBin]) ->
 
 continue_lines_to_points(Data, Item, Rest, ResultPointsAcc, ErrorPointsAcc) ->
     case line_to_point(Data, Item) of
-        #{fields := Fields} when map_size(Fields) =:= 0 ->
+        {_, [#{fields := Fields}]} when map_size(Fields) =:= 0 ->
             %% greptimedb client doesn't like empty field maps...
             ErrorPointsAcc1 = [{error, no_fields} | ErrorPointsAcc],
             lines_to_points(Data, Rest, ResultPointsAcc, ErrorPointsAcc1);

apps/emqx_bridge_greptimedb/test/emqx_bridge_greptimedb_SUITE.erl

@@ -9,6 +9,7 @@
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("emqx/include/logger.hrl").
 
 %%------------------------------------------------------------------------------
 %% CT boilerplate
@@ -284,7 +285,8 @@ send_message(Config, Payload) ->
     Name = ?config(greptimedb_name, Config),
     Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
     BridgeId = emqx_bridge_resource:bridge_id(Type, Name),
-    emqx_bridge:send_message(BridgeId, Payload).
+    Resp = emqx_bridge:send_message(BridgeId, Payload),
+    Resp.
 
 query_by_clientid(Topic, ClientId, Config) ->
     GreptimedbHost = ?config(greptimedb_host, Config),
@@ -308,7 +310,7 @@ query_by_clientid(Topic, ClientId, Config) ->
         {"Authorization", "Basic Z3JlcHRpbWVfdXNlcjpncmVwdGltZV9wd2Q="},
         {"Content-Type", "application/x-www-form-urlencoded"}
     ],
-    Body = <<"sql=select * from ", Topic/binary, " where clientid='", ClientId/binary, "'">>,
+    Body = <<"sql=select * from \"", Topic/binary, "\" where clientid='", ClientId/binary, "'">>,
     {ok, 200, _Headers, RawBody0} =
         ehttpc:request(
             EHttpcPoolName,
@@ -317,6 +319,8 @@ query_by_clientid(Topic, ClientId, Config) ->
             _Timeout = 10_000,
             _Retry = 0
         ),
+
+    case emqx_utils_json:decode(RawBody0, [return_maps]) of
         #{
             <<"code">> := 0,
             <<"output">> := [
@@ -327,19 +331,37 @@ query_by_clientid(Topic, ClientId, Config) ->
                     }
                 }
             ]
-    } = emqx_utils_json:decode(RawBody0, [return_maps]),
+        } ->
+            make_row(Schema, Rows);
+        #{
+            <<"code">> := Code,
+            <<"error">> := Error
+        } ->
+            GreptimedbName = ?config(greptimedb_name, Config),
+            Type = greptimedb_type_bin(?config(greptimedb_type, Config)),
+            BridgeId = emqx_bridge_resource:bridge_id(Type, GreptimedbName),
 
-    case Schema of
+            ?SLOG(error, #{
-        null ->
+                msg => io_lib:format("Failed to query: ~p, ~p", [Code, Error]),
-            #{};
+                connector => BridgeId,
-        #{<<"column_schemas">> := ColumnsSchemas} ->
+                reason => Error
-            Columns = lists:map(fun(#{<<"name">> := Name}) -> Name end, ColumnsSchemas),
+            }),
-            index_by_field(Rows, Columns)
+            %% TODO(dennis): check the error by code
+            case binary:match(Error, <<"Table not found">>) of
+                nomatch ->
+                    {error, Error};
+                _ ->
+                    %% Table not found
+                    #{}
+            end
     end.
 
-index_by_field([], Columns) ->
+make_row(null, _Rows) ->
     #{};
-index_by_field([Row], Columns) ->
+make_row(_Schema, []) ->
+    #{};
+make_row(#{<<"column_schemas">> := ColumnsSchemas}, [Row]) ->
+    Columns = lists:map(fun(#{<<"name">> := Name}) -> Name end, ColumnsSchemas),
     maps:from_list(lists:zip(Columns, Row)).
 
 assert_persisted_data(ClientId, Expected, PersistedData) ->
@@ -784,26 +806,22 @@ t_write_failure(Config) ->
         emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
             case QueryMode of
                 sync ->
-                    {_, {ok, _}} =
                     ?wait_async_action(
                         ?assertMatch(
                             {error, {resource_error, #{reason := timeout}}},
                             send_message(Config, SentData)
                         ),
-                            #{?snk_kind := handle_async_reply, action := nack},
+                        #{?snk_kind := greptimedb_connector_do_query_failure, action := nack},
-                            1_000
+                        16_000
                     )
             end
         end),
-        fun(Trace0) ->
+        fun(Trace) ->
             case QueryMode of
                 sync ->
-                    Trace = ?of_kind(handle_async_reply, Trace0),
+                    ?assertMatch(
-                    ?assertMatch([_ | _], Trace),
+                        [#{error := _} | _],
-                    [#{result := Result} | _] = Trace,
+                        ?of_kind(greptimedb_connector_do_query_failure, Trace)
-                    ?assert(
-                        not emqx_bridge_greptimedb_connector:is_unrecoverable_error(Result),
-                        #{got => Result}
                     )
             end,
             ok
@@ -841,7 +859,7 @@ t_missing_field(Config) ->
                     ?match_n_events(NEvents, #{
                         ?snk_kind := greptimedb_connector_send_query_error
                     }),
-                    _Timeout1 = 10_000
+                    _Timeout1 = 16_000
                 ),
             ok
         end,