Data bridge target unavailable #10645

paulozulato · 2023-05-08T19:18:38Z

Fixes https://emqx.atlassian.net/browse/EMQX-9026

Summary

`🤖 Generated by Copilot at 6b23c31`

This pull request enhances the error handling and reporting of the MySQL and PostgreSQL connectors when the target table is not created. It also updates the default SQL statement for the MySQL connector to match the new table schema. It affects the files emqx_connector_mysql.erl, emqx_connector_pgsql.erl, emqx_resource_buffer_worker.erl, and emqx_ee_bridge_mysql.erl.

PR Checklist

Please convert it to a draft if any of the following conditions are not met. Reviewers may skip over until all the items are checked:

Added tests for the changes
Changed lines covered in coverage report
Change log has been added to changes/{ce,ee}/(feat|perf|fix)-<PR-id>.en.md files
For internal contributor: there is a jira ticket to track this change
~~If there should be document changes, a PR to emqx-docs.git is sent, or a jira ticket is created to follow up~~
~~Schema changes are backward compatible~~

Checklist for CI (.github/workflows) changes

~~If changed package build workflow, pass this action (manual trigger)~~
~~Change log has been added to changes/ dir for user-facing artifacts update~~

thalesmg · 2023-06-09T20:55:44Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

+    % wolff starts a socket to get metadata and closes it asynchronously (calling a spawn to
+    % actually close it), causing some {'EXIT', tcp_closed} errors.
+    process_flag(trap_exit, true),
+    Ret =
+        case catch wolff_client:check_if_topic_exists(Hosts, KafkaConfig, KafkaTopic) of
+            ok ->
+                ok;
+            {error, unknown_topic_or_partition} ->
+                unhealthy_target;
+            _ ->
+                error
+        end,
+    process_flag(trap_exit, false),
+    Ret.


Suggested change

% wolff starts a socket to get metadata and closes it asynchronously (calling a spawn to

% actually close it), causing some {'EXIT', tcp_closed} errors.

process_flag(trap_exit, true),

Ret =

case catch wolff_client:check_if_topic_exists(Hosts, KafkaConfig, KafkaTopic) of

ok ->

ok;

{error, unknown_topic_or_partition} ->

unhealthy_target;

_ ->

error

end,

process_flag(trap_exit, false),

Ret.

try wolff_client:check_if_topic_exists(Hosts, KafkaConfig, KafkaTopic) of

ok ->

ok;

{error, unknown_topic_or_partition} ->

unhealthy_target;

_ ->

error

catch

_:_ -> error

end.

?

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

thalesmg · 2023-06-09T21:07:21Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

+            lists:foldl(
+                fun
+                    (_, connected) ->
+                        connected;
+                    ({_Partition, Pid}, Acc) ->
+                        case is_pid(Pid) andalso erlang:is_process_alive(Pid) of
+                            false ->
+                                Acc;
+                            true ->
+                                CheckIfTopicExists(Pid, KafkaTopic, Acc)
+                        end
+                end,
+                disconnected,
+                Leaders
+            );


Maybe we could disentangle the two checks to make it more readable, and check the topic existence only once?

on_get_status -> case do_get_status(..., State) of {disconnected, unhealthy} -> {disconnected, State, unhealthy}; Status -> Status end. %% ... do_get_status -> HealthyLeaders = lists:filter(fun(L) -> is_pid ... end, Leaders), case HealthyLeaders of [Pid | _] -> do_get_topic_status(Pid, Topic); [] -> disconnected end. do_get_topic_status -> try wolf_client:check_if_topic_exists(Pid, Topic) of ... catch _:_ -> disconnected end.

thalesmg · 2023-06-09T21:08:58Z

apps/emqx_bridge_oracle/test/emqx_bridge_oracle_SUITE.erl

+    delete_all_bridges(),
+    ?check_trace(


Nit: maybe split this test case into 2.

thalesmg · 2023-06-09T21:09:28Z

apps/emqx_bridge_pgsql/test/emqx_bridge_pgsql_SUITE.erl

+    delete_bridge(Config),
+    ?check_trace(


Split into two separate cases?

zmstone · 2023-06-10T12:39:52Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

    end.

+check_if_topic_exists(Hosts, KafkaConfig, KafkaTopic) ->
+    % wolff starts a socket to get metadata and closes it asynchronously (calling a spawn to


then setting trap_exit immediately after would not help? it could happen that the trap_exist is set back to false before the socket is down.
better solution is to span_monitor a process to do it and drain the DOWN message

zmstone · 2023-06-10T12:41:09Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

+do_get_status(Client, #{kafka_topic := KafkaTopic} = State) ->
+    CheckIfTopicExists =
+        fun(Pid, Topic, Default) ->
+            case wolff_client:check_if_topic_exists(Pid, Topic) of


the return value match of wolff_client:check_if_topic_exists is repeated, maybe create a wrapper function instead.

coveralls · 2023-06-16T00:45:43Z

Pull Request Test Coverage Report for Build 5337384650

83 of 106 (78.3%) changed or added relevant lines in 6 files are covered.
30 unchanged lines in 16 files lost coverage.
Overall coverage increased (+0.05%) to 81.803%

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
apps/emqx_resource/src/emqx_resource_buffer_worker.erl	0	1	0.0%
apps/emqx_connector/src/emqx_connector_mysql.erl	10	14	71.43%
apps/emqx_connector/src/emqx_connector_pgsql.erl	12	17	70.59%
apps/emqx_oracle/src/emqx_oracle.erl	42	47	89.36%
apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl	17	25	68.0%

Files with Coverage Reduction	New Missed Lines	%
apps/emqx_conf/src/emqx_conf_schema.erl	1	95.0%
apps/emqx_dashboard/src/emqx_dashboard_desc_cache.erl	1	94.44%
apps/emqx_dashboard/src/emqx_dashboard_monitor.erl	1	66.21%
apps/emqx_gateway_mqttsn/src/emqx_mqttsn_frame.erl	1	63.48%
apps/emqx_oracle/src/emqx_oracle.erl	1	82.58%
apps/emqx_resource/src/emqx_resource_buffer_worker.erl	1	93.4%
apps/emqx/src/emqx_cm.erl	1	92.05%
apps/emqx/src/emqx_logger_textfmt.erl	1	75.0%
apps/emqx/src/emqx_message.erl	1	94.83%
apps/emqx/src/emqx_quic_data_stream.erl	1	75.21%

Totals
Change from base Build 5333842210:	0.05%
Covered Lines:	29958
Relevant Lines:	36622

💛 - Coveralls

thalesmg · 2023-06-16T20:54:32Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

+    case do_get_topic_status(HealthyLeaders, KafkaTopic) of
+        error ->
+            % it can be that topic is unknown or there was some error getting the leaders (like
+            % some connection issue). Therefore, let's try to get metadata for that topic directly.
+            do_get_topic_status(Hosts, KafkaConfig, KafkaTopic);


In this case, I guess we could just call do_get_topic_status(Hosts, KafkaConfig, KafkaTopic) directly rather than going through do_get_topic_status(HealthyLeaders, KafkaTopic)?

thalesmg · 2023-06-16T20:55:11Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

+            (_Pid, ok) ->
+                ok;
+            (Pid, Acc) ->


Should stop iterating when we already know the topic is unknown.

thalesmg · 2023-06-16T20:59:55Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

+    HealthyLeaders = do_get_healthy_leaders(Client, KafkaTopic),
+    case do_get_topic_status(HealthyLeaders, KafkaTopic) of


Since the unknown topic error seems to take precedence over connection issues, we could check the topic first and short-circuit if it's unknown, instead of checking the leader pids.

thalesmg · 2023-06-16T21:02:47Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

+    RetPid = self(),
+    QueryRef = make_ref(),
+    CheckTopic = fun() ->
+        Ret =
+            case wolff_client:check_if_topic_exists(Hosts, KafkaConfig, KafkaTopic) of
+                ok ->
+                    ok;
+                {error, unknown_topic_or_partition} ->
+                    unhealthy_target;
+                _ ->
+                    error
+            end,
+        RetPid ! {QueryRef, Ret}
+    end,
+    % wolff starts a socket to get metadata and closes it asynchronously (calling a spawn to
+    % actually close it), causing some {'EXIT', tcp_closed} errors. Therefore, it'll be executed
+    % by a spawn process to avoid crashing the main process.
+    {Pid, Ref} = spawn_monitor(CheckTopic),
+    % wait process terminate
+    receive
+        {'DOWN', Ref, process, Pid, _What} -> ok
+    after 5000 ->
+        ?SLOG(error, #{msg => "failed to check kafka topic", pid => Pid}),
+        demonitor(Ref),
+        exit(Pid, kill)
+    end,
+    % check for result
+    receive
+        {QueryRef, Result} -> Result
+    after 500 ->
+        error


It seems like this could use emqx_utils:nolink_apply.

try emqx_utils:nolink_apply(fun() -> check... end, Timeout) catch exit:_ -> ...; error:_ -> ...

thalesmg · 2023-06-16T21:06:07Z

apps/emqx_connector/src/emqx_connector_mysql.erl

+    lists:foldl(
+        fun
+            (WorkerPid, ok) ->
+                {ok, Conn} = ecpool_worker:client(WorkerPid),


Better avoid exploding here?

Fixes https://emqx.atlassian.net/browse/EMQX-9026

thalesmg · 2023-06-20T13:35:45Z

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_producer.erl

+            wolff_client:check_if_topic_exists(Hosts, KafkaConfig, KafkaTopic)
+        end,
+    try
+        case emqx_utils:nolink_apply(CheckTopicFun, 500) of


Maybe 500 ms is a bit too strict and could lead to false positives?

Suggested change

case emqx_utils:nolink_apply(CheckTopicFun, 500) of

case emqx_utils:nolink_apply(CheckTopicFun, 5_000) of

thalesmg · 2023-06-20T13:39:57Z

apps/emqx_connector/src/emqx_connector_mysql.erl

@@ -224,7 +224,10 @@ on_get_status(_InstId, #{pool_name := PoolName} = State) ->
                    {connected, NState};
                {error, _Reason} ->
                    %% do not log error, it is logged in prepare_sql_to_conn
-                    connecting
+                    connecting;
+                {undefined_table, NState} ->


Nit:

Suggested change

{undefined_table, NState} ->

{error, {undefined_table, NState}} ->

(will require reordering with the catch-all clause above)

thalesmg · 2023-06-20T13:41:00Z

apps/emqx_connector/src/emqx_connector_pgsql.erl

+    lists:foldl(
+        fun
+            (WorkerPid, ok) ->
+                {ok, Conn} = ecpool_worker:client(WorkerPid),


This one is still explosive.

thalesmg · 2023-06-20T13:41:27Z

apps/emqx_oracle/src/emqx_oracle.erl

+    lists:foldl(
+        fun
+            (WorkerPid, ok) ->
+                {ok, Conn} = ecpool_worker:client(WorkerPid),


apps/emqx_oracle/src/emqx_oracle.erl

thalesmg

there are still a few occurrences of {undefined_table, _} instead of {error, {undefined_table, _}}, but that's a nit that could be addressed at a later time.

Fixes https://emqx.atlassian.net/browse/EMQX-9026

paulozulato · 2023-06-21T18:06:46Z

there are still a few occurrences of {undefined_table, _} instead of {error, {undefined_table, _}}, but that's a nit that could be addressed at a later time.

fixed =)

paulozulato force-pushed the data-bridge-target-unavailable branch from 6b23c31 to 9dcdc19 Compare May 10, 2023 00:54

paulozulato force-pushed the data-bridge-target-unavailable branch 2 times, most recently from a561f3d to e9bf002 Compare May 18, 2023 00:52

paulozulato force-pushed the data-bridge-target-unavailable branch 3 times, most recently from b5c67e9 to ed91d81 Compare June 9, 2023 20:14

thalesmg reviewed Jun 9, 2023

View reviewed changes

zmstone reviewed Jun 10, 2023

View reviewed changes

paulozulato force-pushed the data-bridge-target-unavailable branch from ed91d81 to a02fa2d Compare June 15, 2023 23:05

paulozulato marked this pull request as ready for review June 16, 2023 15:54

paulozulato requested review from a team and JimMoen as code owners June 16, 2023 15:54

thalesmg reviewed Jun 16, 2023

View reviewed changes

paulozulato added 3 commits June 19, 2023 11:12

feat(postgresql): check whether target table exists

9454af9

Fixes https://emqx.atlassian.net/browse/EMQX-9026

feat(mysql): check whether target table exists

c9a2ddf

Fixes https://emqx.atlassian.net/browse/EMQX-9026

feat(oracle): check whether target table exists

5f10936

Fixes https://emqx.atlassian.net/browse/EMQX-9026

paulozulato force-pushed the data-bridge-target-unavailable branch from a02fa2d to ca40e80 Compare June 19, 2023 20:02

thalesmg reviewed Jun 20, 2023

View reviewed changes

paulozulato force-pushed the data-bridge-target-unavailable branch 3 times, most recently from 1f9ebb3 to a5ceaf3 Compare June 21, 2023 13:57

thalesmg previously approved these changes Jun 21, 2023

View reviewed changes

feat(kafka): check whether target topic exists

8430ec6

Fixes https://emqx.atlassian.net/browse/EMQX-9026

paulozulato dismissed thalesmg’s stale review via 8430ec6 June 21, 2023 18:00

paulozulato force-pushed the data-bridge-target-unavailable branch from a5ceaf3 to 8430ec6 Compare June 21, 2023 18:00

thalesmg approved these changes Jun 21, 2023

View reviewed changes

paulozulato merged commit 62d3766 into emqx:master Jun 21, 2023
130 checks passed

paulozulato deleted the data-bridge-target-unavailable branch June 21, 2023 21:19

thalesmg mentioned this pull request Jul 25, 2023

fix(postgres_bridge): fix table existence check and handle sync_required #11338

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Data bridge target unavailable #10645

Data bridge target unavailable #10645

paulozulato commented May 8, 2023 •

edited

thalesmg Jun 9, 2023

thalesmg Jun 9, 2023

thalesmg Jun 9, 2023

thalesmg Jun 9, 2023

zmstone Jun 10, 2023

zmstone Jun 10, 2023

coveralls commented Jun 16, 2023 •

edited

thalesmg Jun 16, 2023

thalesmg Jun 16, 2023

thalesmg Jun 16, 2023

thalesmg Jun 16, 2023

thalesmg Jun 16, 2023

thalesmg Jun 20, 2023

thalesmg Jun 20, 2023

thalesmg Jun 20, 2023

thalesmg Jun 20, 2023

thalesmg left a comment

paulozulato commented Jun 21, 2023

		HealthyLeaders = do_get_healthy_leaders(Client, KafkaTopic),
		case do_get_topic_status(HealthyLeaders, KafkaTopic) of

	case emqx_utils:nolink_apply(CheckTopicFun, 500) of
	case emqx_utils:nolink_apply(CheckTopicFun, 5_000) of

	{undefined_table, NState} ->
	{error, {undefined_table, NState}} ->

Data bridge target unavailable #10645

Data bridge target unavailable #10645

Conversation

paulozulato commented May 8, 2023 • edited

Summary

🤖 Generated by Copilot at 6b23c31

PR Checklist

Checklist for CI (.github/workflows) changes

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coveralls commented Jun 16, 2023 • edited

Pull Request Test Coverage Report for Build 5337384650

💛 - Coveralls

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

thalesmg left a comment

Choose a reason for hiding this comment

paulozulato commented Jun 21, 2023

paulozulato commented May 8, 2023 •

edited

`🤖 Generated by Copilot at 6b23c31`

coveralls commented Jun 16, 2023 •

edited