Packages
brod
4.5.0
4.5.7
4.5.6
4.5.5
4.5.4
4.5.3
4.5.2
4.5.1
4.5.0
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.3
4.3.2
4.3.1
4.3.0
4.2.0
4.1.1
4.1.0
4.0.0
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.5
3.16.4
3.16.3
3.16.2
3.16.1
3.16.0
3.15.6
3.15.5
3.15.4
3.15.3
3.15.1
3.15.0
3.14.0
3.13.0
3.12.0
3.11.0
3.10.0
3.9.5
3.9.3
3.9.2
3.9.1
3.9.0
3.8.1
3.8.0
3.7.11
3.7.10
3.7.9
3.7.8
3.7.7
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
3.6.2
3.6.1
3.6.0
3.5.2
3.5.1
3.5.0
3.4.0
3.3.5
3.3.4
3.3.3
3.3.2
3.3.1
3.3.0
3.2.0
3.0.0
2.5.0
2.4.1
2.4.0
2.3.7
2.3.6
2.3.5
2.3.4
2.3.3
2.3.1
2.2.16
2.2.15
2.2.14
2.2.12
2.2.11
2.2.10
2.2.9
2.2.8
2.2.7
2.2.6
2.2.5
2.2.4
2.2.3
2.2.2
2.2.1
2.2.0
2.1.12
2.1.11
2.1.10
2.1.8
2.1.7
2.1.4
2.1.2
2.0.0
Apache Kafka Erlang client library
Current section
113 Versions
Jump to
Current section
113 Versions
Compare versions
7
files changed
+142
additions
-2
deletions
| @@ -24,4 +24,4 @@ | |
| 24 24 | [{<<"app">>,<<"kafka_protocol">>}, |
| 25 25 | {<<"optional">>,false}, |
| 26 26 | {<<"requirement">>,<<"4.3.0">>}]}]}. |
| 27 | - {<<"version">>,<<"4.4.7">>}. |
| 27 | + {<<"version">>,<<"4.5.0">>}. |
| @@ -76,6 +76,8 @@ | |
| 76 76 | -define(BROD_LOG_INFO(Fmt, Args), ?LOG_INFO( Fmt, Args, #{domain => [brod]})). |
| 77 77 | -define(BROD_LOG(Level, Fmt, Args), ?LOG(Level, Fmt, Args, #{domain => [brod]})). |
| 78 78 | |
| 79 | + -define(MF_ERR_MAP(Reason), #{cause => Reason, module => ?MODULE, func => ?FUNCTION_NAME}). |
| 80 | + |
| 79 81 | -endif. % include brod_int.hrl |
| 80 82 | |
| 81 83 | %%%_* Emacs ==================================================================== |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,brod, |
| 2 2 | [{description,"Apache Kafka Erlang client library"}, |
| 3 | - {vsn,"4.4.7"}, |
| 3 | + {vsn,"4.5.0"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol]}, |
| 6 6 | {env,[]}, |
| @@ -44,6 +44,7 @@ | |
| 44 44 | , stop_maybe_kill/2 |
| 45 45 | , subscribe/3 |
| 46 46 | , unsubscribe/2 |
| 47 | + , check_connectivity/2 |
| 47 48 | ]). |
| 48 49 | |
| 49 50 | %% Debug API |
| @@ -299,6 +300,21 @@ debug(Pid, File) when is_list(File) -> | |
| 299 300 | get_connection(Pid) -> |
| 300 301 | gen_server:call(Pid, get_connection). |
| 301 302 | |
| 303 | + %% @doc Return 'ok' if the payload connection is alive. |
| 304 | + -spec check_connectivity(pid(), pos_integer()) -> ok | {error, term()}. |
| 305 | + check_connectivity(Pid, Timeout) -> |
| 306 | + try |
| 307 | + case gen_server:call(Pid, is_connected, Timeout) of |
| 308 | + true -> |
| 309 | + ok; |
| 310 | + false -> |
| 311 | + {error, disconnected} |
| 312 | + end |
| 313 | + catch |
| 314 | + exit : Reason -> |
| 315 | + {error, Reason} |
| 316 | + end. |
| 317 | + |
| 302 318 | %%%_* gen_server callbacks ===================================================== |
| 303 319 | |
| 304 320 | init({Bootstrap0, Topic, Partition, Config}) -> |
| @@ -402,6 +418,8 @@ handle_info(Info, State) -> | |
| 402 418 | {noreply, State}. |
| 403 419 | |
| 404 420 | %% @private |
| 421 | + handle_call(is_connected, _From, #state{connection = C} = State) -> |
| 422 | + {reply, is_pid(C) andalso is_process_alive(C), State}; |
| 405 423 | handle_call(get_connection, _From, #state{connection = C} = State) -> |
| 406 424 | {reply, C, State}; |
| 407 425 | handle_call({subscribe, Pid, Options}, _From, |
| @@ -35,6 +35,7 @@ | |
| 35 35 | , start_link/1 |
| 36 36 | , stop/1 |
| 37 37 | , get_workers/1 |
| 38 | + , health_check/2 |
| 38 39 | ]). |
| 39 40 | |
| 40 41 | %% brod_group_coordinator callbacks |
| @@ -208,6 +209,22 @@ get_workers(Pid) -> | |
| 208 209 | get_workers(Pid, Timeout) -> |
| 209 210 | gen_server:call(Pid, get_workers, Timeout). |
| 210 211 | |
| 212 | + %% @doc Returns `ok' if all workers are healthy, |
| 213 | + %% otherwise `{error, Reasons}' where `Reasons' is a list of errors. |
| 214 | + -spec health_check(pid(), timeout()) -> ok | {error, [map()]}. |
| 215 | + health_check(Pid, Timeout) -> |
| 216 | + try |
| 217 | + case gen_server:call(Pid, {health_check, Timeout}, Timeout) of |
| 218 | + [] -> |
| 219 | + ok; |
| 220 | + Errors -> |
| 221 | + {error, Errors} |
| 222 | + end |
| 223 | + catch |
| 224 | + exit : Reason -> |
| 225 | + {error, [?MF_ERR_MAP(#{reason => Reason, group_subscriber => Pid})]} |
| 226 | + end. |
| 227 | + |
| 211 228 | %%%=================================================================== |
| 212 229 | %%% group_coordinator callbacks |
| 213 230 | %%%=================================================================== |
| @@ -315,6 +332,8 @@ handle_call({assign_partitions, Members, TopicPartitionList}, _From, State) -> | |
| 315 332 | {reply, Reply, State}; |
| 316 333 | handle_call(get_workers, _From, State = #state{workers = Workers}) -> |
| 317 334 | {reply, Workers, State}; |
| 335 | + handle_call({health_check, Timeout}, _From, State = #state{workers = Workers}) -> |
| 336 | + {reply, do_health_check(Workers, Timeout), State}; |
| 318 337 | handle_call(Call, _From, State) -> |
| 319 338 | {reply, {error, {unknown_call, Call}}, State}. |
| 320 339 | |
| @@ -545,6 +564,25 @@ stop_consumers(Config) -> | |
| 545 564 | end, |
| 546 565 | Topics). |
| 547 566 | |
| 567 | + do_health_check(Workers0, Timeout) -> |
| 568 | + Workers = lists:keysort(1, maps:to_list(Workers0)), |
| 569 | + Fn = fun({{Topic, _Partition}, Pid}) -> |
| 570 | + case brod_topic_subscriber:health_check(Pid, Timeout) of |
| 571 | + [] -> |
| 572 | + []; |
| 573 | + [Error] -> |
| 574 | + %% start one consumer for each topic |
| 575 | + %% so it's always a one-element list |
| 576 | + [Error#{topic => Topic}] |
| 577 | + end |
| 578 | + end, |
| 579 | + try |
| 580 | + lists:flatten(brod_utils:pmap(Fn, Workers, Timeout)) |
| 581 | + catch |
| 582 | + throw : timeout -> |
| 583 | + [?MF_ERR_MAP(timeout)] |
| 584 | + end. |
| 585 | + |
| 548 586 | %%%_* Emacs ==================================================================== |
| 549 587 | %%% Local Variables: |
| 550 588 | %%% allout-layout: t |
Loading more files…