yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(kafka_consumer): follow up refactoring requested from previous pull request 

Follow up from https://github.com/emqx/emqx/pull/10273
This commit is contained in:
Thales Macedo Garitezi 2023-04-13 16:26:03 -03:00
parent 26883eec02
commit 4bcfbea056
1 changed files with 17 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
apps/emqx_bridge_kafka/src/emqx_bridge_kafka_impl_consumer.erl
View File

@ -179,7 +179,12 @@ on_get_status(_InstanceID, State) ->
kafka_client_id := ClientID, kafka_client_id := ClientID,
kafka_topics := KafkaTopics kafka_topics := KafkaTopics
} = State, } = State,
do_get_status(State, ClientID, KafkaTopics, SubscriberId). case do_get_status(ClientID, KafkaTopics, SubscriberId) of
{disconnected, Message} ->
{disconnected, State, Message};
Res ->
Res
end.
%%------------------------------------------------------------------------------------- %%-------------------------------------------------------------------------------------
%% `brod_group_subscriber' API %% `brod_group_subscriber' API
@ -376,41 +381,41 @@ stop_client(ClientID) ->
), ),
ok. ok.
do_get_status(State, ClientID, [KafkaTopic | RestTopics], SubscriberId) -> do_get_status(ClientID, [KafkaTopic | RestTopics], SubscriberId) ->
case brod:get_partitions_count(ClientID, KafkaTopic) of case brod:get_partitions_count(ClientID, KafkaTopic) of
{ok, NPartitions} -> {ok, NPartitions} ->
case do_get_status1(ClientID, KafkaTopic, SubscriberId, NPartitions) of case do_get_topic_status(ClientID, KafkaTopic, SubscriberId, NPartitions) of
connected -> do_get_status(State, ClientID, RestTopics, SubscriberId); connected -> do_get_status(ClientID, RestTopics, SubscriberId);
disconnected -> disconnected disconnected -> disconnected
end; end;
{error, {client_down, Context}} -> {error, {client_down, Context}} ->
case infer_client_error(Context) of case infer_client_error(Context) of
auth_error -> auth_error ->
Message = "Authentication error. " ++ ?CLIENT_DOWN_MESSAGE, Message = "Authentication error. " ++ ?CLIENT_DOWN_MESSAGE,
{disconnected, State, Message}; {disconnected, Message};
{auth_error, Message0} -> {auth_error, Message0} ->
Message = binary_to_list(Message0) ++ "; " ++ ?CLIENT_DOWN_MESSAGE, Message = binary_to_list(Message0) ++ "; " ++ ?CLIENT_DOWN_MESSAGE,
{disconnected, State, Message}; {disconnected, Message};
connection_refused -> connection_refused ->
Message = "Connection refused. " ++ ?CLIENT_DOWN_MESSAGE, Message = "Connection refused. " ++ ?CLIENT_DOWN_MESSAGE,
{disconnected, State, Message}; {disconnected, Message};
_ -> _ ->
{disconnected, State, ?CLIENT_DOWN_MESSAGE} {disconnected, ?CLIENT_DOWN_MESSAGE}
end; end;
{error, leader_not_available} -> {error, leader_not_available} ->
Message = Message =
"Leader connection not available. Please check the Kafka topic used," "Leader connection not available. Please check the Kafka topic used,"
" the connection parameters and Kafka cluster health", " the connection parameters and Kafka cluster health",
{disconnected, State, Message}; {disconnected, Message};
_ -> _ ->
disconnected disconnected
end; end;
do_get_status(_State, _ClientID, _KafkaTopics = [], _SubscriberId) -> do_get_status(_ClientID, _KafkaTopics = [], _SubscriberId) ->
connected. connected.
-spec do_get_status1(brod:client_id(), binary(), subscriber_id(), pos_integer()) -> -spec do_get_topic_status(brod:client_id(), binary(), subscriber_id(), pos_integer()) ->
connected | disconnected. connected | disconnected.
do_get_status1(ClientID, KafkaTopic, SubscriberId, NPartitions) -> do_get_topic_status(ClientID, KafkaTopic, SubscriberId, NPartitions) ->
Results = Results =
lists:map( lists:map(
fun(N) -> fun(N) ->