fix(kafka): improve shutdown and health check logs during shutdown

apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_consumer.erl

@@ -381,7 +381,13 @@ start_consumer(Config, ResourceId, ClientID) ->
 stop_subscriber(SubscriberId) ->
     _ = log_when_error(
         fun() ->
-            emqx_bridge_kafka_consumer_sup:ensure_child_deleted(SubscriberId)
+            try
+                emqx_bridge_kafka_consumer_sup:ensure_child_deleted(SubscriberId)
+            catch
+                exit:{noproc, _} ->
+                    %% may happen when node is shutting down
+                    ok
+            end
         end,
         #{
             msg => "failed_to_delete_kafka_subscriber",
@@ -465,16 +471,22 @@ do_get_topic_status(ClientID, KafkaTopic, SubscriberId, NPartitions) ->
     end.
 
 are_subscriber_workers_alive(SubscriberId) ->
-    Children = supervisor:which_children(emqx_bridge_kafka_consumer_sup),
+    try
-    case lists:keyfind(SubscriberId, 1, Children) of
+        Children = supervisor:which_children(emqx_bridge_kafka_consumer_sup),
-        false ->
+        case lists:keyfind(SubscriberId, 1, Children) of
-            false;
+            false ->
-        {_, Pid, _, _} ->
+                false;
-            Workers = brod_group_subscriber_v2:get_workers(Pid),
+            {_, Pid, _, _} ->
-            %% we can't enforce the number of partitions on a single
+                Workers = brod_group_subscriber_v2:get_workers(Pid),
-            %% node, as the group might be spread across an emqx
+                %% we can't enforce the number of partitions on a single
-            %% cluster.
+                %% node, as the group might be spread across an emqx
-            lists:all(fun is_process_alive/1, maps:values(Workers))
+                %% cluster.
+                lists:all(fun is_process_alive/1, maps:values(Workers))
+        end
+    catch
+        exit:{shutdown, _} ->
+            %% may happen if node is shutting down
+            false
     end.
 
 log_when_error(Fun, Log) ->

apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl

@@ -335,6 +335,7 @@ init_per_testcase(TestCase, Config) ->
 common_init_per_testcase(TestCase, Config0) ->
     ct:timetrap(timer:seconds(60)),
     delete_all_bridges(),
+    emqx_config:delete_override_conf_files(),
     KafkaTopic =
         <<
             (atom_to_binary(TestCase))/binary,