Packages
brod
2.2.7
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
Files
Jump to
Current section
Files
src/brod_sup.erl
%%%
%%% Copyright (c) 2015 Klarna AB
%%%
%%% Licensed under the Apache License, Version 2.0 (the "License");
%%% you may not use this file except in compliance with the License.
%%% You may obtain a copy of the License at
%%%
%%% http://www.apache.org/licenses/LICENSE-2.0
%%%
%%% Unless required by applicable law or agreed to in writing, software
%%% distributed under the License is distributed on an "AS IS" BASIS,
%%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%% See the License for the specific language governing permissions and
%%% limitations under the License.
%%%
%%%=============================================================================
%%% @doc brod supervisor
%%%
%%% Hierarchy:
%%% brod_sup (one_for_one)
%%% |
%%% +--client_1
%%% | |
%%% | +-- producers_sup level 1
%%% | | |
%%% | | +-- producers_sup level 2 for topic 1
%%% | | | |
%%% | | | +-- partition_0_worker
%%% | | | |
%%% | | | +-- partition_1_worker
%%% | | | |...
%%% | | |
%%% | | +-- producers_sup level 2 for topic 2
%%% | | | |...
%%% | | |...
%%% | |
%%% | +-- consumers_sup level 1
%%% | |
%%% | +-- consumer_sup level 2 for topic 1
%%% | | |
%%% | | +-- partition_0_worker
%%% | | |
%%% | | +-- partition_1_worker
%%% | | |...
%%% | |
%%% | +-- consumer_sup level 2 for topic 2
%%% | | |...
%%% | |...
%%% |
%%% +-- client_2
%%% | |...
%%% |...
%%%
%%% @copyright 2015 Klarna AB
%%% @end
%%%=============================================================================
-module(brod_sup).
-behaviour(supervisor3).
-export([ init/1
, post_init/1
, start_link/0
, start_client/3
, stop_client/1
, find_client/1
]).
-include("brod_int.hrl").
-define(SUP, ?MODULE).
%% By deafult, restart client process after a 10-seconds delay
-define(DEFAULT_CLIENT_RESTART_DELAY, 10).
%%%_* APIs =====================================================================
%% @doc Start root supervisor.
%%
%% To start permanent clients add 'clients' section in sys.config.
%% So far only 'endpoints' config is mandatory, other options are optional.
%%
%% ```
%% [
%% %% Permanent clients
%% { clients
%% , [ {client_1 %% unique client ID
%% , [ {endpoints, [{"localhost", 9092}]}
%% , {restart_delay_seconds, 10}
%% , {max_metadata_sock_retry, 1}
%% , {get_metadata_timeout_seconds, 5}
%% , {reconnect_cool_down_seconds, 1}
%% , {allow_topic_auto_creation, true}
%% , {auto_start_producers, false}
%% , {default_producer_config, []}
%% ]
%% }
%% ]
%% }
%% ].
%% '''
%% @end
-spec start_link() -> {ok, pid()}.
start_link() ->
supervisor3:start_link({local, ?SUP}, ?MODULE, clients_sup).
-spec start_client([endpoint()], brod_client_id(), client_config()) ->
ok | {error, any()}.
start_client(Endpoints, ClientId, Config) ->
ClientSpec = client_spec(Endpoints, ClientId, Config),
case supervisor3:start_child(?SUP, ClientSpec) of
{ok, _Pid} -> ok;
Error -> Error
end.
-spec stop_client(brod_client_id()) -> ok | {error, any()}.
stop_client(ClientId) ->
_ = supervisor3:terminate_child(?SUP, ClientId),
supervisor3:delete_child(?SUP, ClientId).
-spec find_client(brod_client_id()) -> [pid()].
find_client(Client) ->
supervisor3:find_child(?SUP, Client).
%% @doc supervisor3 callback
init(clients_sup) ->
Clients = application:get_env(brod, clients, []),
ClientSpecs =
lists:map(fun({ClientId, Args}) ->
is_atom(ClientId) orelse exit({bad_client_id, ClientId}),
client_spec(ClientId, Args)
end, Clients),
%% A client may crash and restart due to network failure
%% e.g. when none of the kafka endpoints are reachable.
%% In this case, restart right away will very likely fail again.
%% Hence set MaxR=0 here to cool-down for a configurable N-seconds
%% before supervisor tries to restart it.
{ok, {{one_for_one, 0, 1}, ClientSpecs}}.
%% @doc supervisor3 callback.
post_init(_) ->
ignore.
%%%_* Internal functions =======================================================
client_spec(ClientId, Config) ->
Endpoints = proplists:get_value(endpoints, Config, []),
client_spec(Endpoints, ClientId, Config).
client_spec([], ClientId, _Config) ->
Error = lists:flatten(
io_lib:format("No endpoints found in brod client '~p' config",
[ClientId])),
exit(Error);
client_spec(Endpoints, ClientId, Config0) ->
DelaySecs = proplists:get_value(restart_delay_seconds, Config0,
?DEFAULT_CLIENT_RESTART_DELAY),
Config = proplists:delete(restart_delay_seconds, Config0),
StartArgs = [Endpoints, ClientId, Config],
{ _Id = ClientId
, _Start = {brod_client, start_link, StartArgs}
, _Restart = {permanent, DelaySecs}
, _Shutdown = 5000
, _Type = worker
, _Module = [brod_client]
}.
%%%_* Emacs ====================================================================
%%% Local Variables:
%%% allout-layout: t
%%% erlang-indent-level: 2
%%% End: