From 75770f2842d7417a59fb836dd824cff72c3866dc Mon Sep 17 00:00:00 2001
From: Ivan Dyachkov <dev@dyachkov.org>
Date: Fri, 30 Dec 2022 10:18:52 +0100
Subject: [PATCH] fix(kafka): detect connectivity in on_get_status

---
 .../src/kafka/emqx_bridge_impl_kafka_producer.erl          | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl b/lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl
index b46bdb486..e0a37bed9 100644
--- a/lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl
+++ b/lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl
@@ -173,8 +173,11 @@ on_kafka_ack(_Partition, _Offset, _Extra) ->
     %% Maybe need to bump some counters?
     ok.
 
-on_get_status(_InstId, _State) ->
-    connected.
+on_get_status(_InstId, #{client_id := ClientID}) ->
+    case wolff:check_connectivity(ClientID) of
+        ok -> connected;
+        _ -> disconnected
+    end.
 
 %% Parse comma separated host:port list into a [{Host,Port}] list
 hosts(Hosts) when is_binary(Hosts) ->