Packages
brod
2.1.10
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_producers_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 producers supervisor
%%%
%%% @copyright 2015 Klarna AB
%%% @end
%%%=============================================================================
-module(brod_producers_sup).
-behaviour(supervisor3).
-export([ init/1
, post_init/1
, start_link/0
, find_producer/3
, start_producer/4
, stop_producer/2
]).
-include("brod_int.hrl").
-define(SUP, brod_producers_sup).
-define(SUP2, brod_producers_sup2).
%% Minimum delay seconds to work with supervisor3
-define(MIN_SUPERVISOR3_DELAY_SECS, 1).
%% By default, restart sup2 after a 10-seconds delay
-define(DEFAULT_SUP2_RESTART_DELAY, 10).
%% By default, restart partition producer worker process after a 5-seconds delay
-define(DEFAULT_PRODUCER_RESTART_DELAY, 5).
%%%_* APIs =====================================================================
%% @doc Start a root producers supervisor.
%% For more details: @see brod_producer:start_link/4
%% @end
-spec start_link() -> {ok, pid()}.
start_link() ->
supervisor3:start_link(?MODULE, ?SUP).
%% @doc Dynamically start a per-topic supervisor
-spec start_producer(pid(), pid(), topic(), producer_config()) ->
{ok, pid()} | {error, any()}.
start_producer(SupPid, ClientPid, TopicName, Config) ->
Spec = producers_sup_spec(ClientPid, TopicName, Config),
supervisor3:start_child(SupPid, Spec).
%% @doc Dynamically stop a per-topic supervisor
-spec stop_producer(pid(), topic()) -> ok | {}.
stop_producer(SupPid, TopicName) ->
supervisor3:terminate_child(SupPid, TopicName).
%% @doc Find a brod_producer process pid running under sup2.
-spec find_producer(pid(), topic(), partition()) ->
{ok, pid()} | {error, Reason} when
Reason :: {producer_not_found, topic()}
| {producer_not_found, topic(), partition()}
| {producer_down, noproc}.
find_producer(SupPid, Topic, Partition) ->
case supervisor3:find_child(SupPid, Topic) of
[] ->
%% no such topic worker started,
%% check sys.config or brod:start_link_client args
{error, {producer_not_found, Topic}};
[Sup2Pid] ->
try
case supervisor3:find_child(Sup2Pid, Partition) of
[] ->
%% no such partition?
{error, {producer_not_found, Topic, Partition}};
[Pid] ->
{ok, Pid}
end
catch exit : {noproc, _} ->
{error, {producer_down, noproc}}
end
end.
%% @doc supervisor3 callback.
init(?SUP) ->
{ok, {{one_for_one, 0, 1}, []}};
init({?SUP2, _ClientPid, _Topic, _Config}) ->
post_init.
post_init({?SUP2, ClientPid, Topic, Config}) ->
case brod_client:get_partitions_count(ClientPid, Topic) of
{ok, PartitionsCnt} ->
Children = [ producer_spec(ClientPid, Topic, Partition, Config)
|| Partition <- lists:seq(0, PartitionsCnt - 1) ],
%% Producer may crash in case of exception in case of network failure,
%% or error code received in produce response (e.g. leader transition)
%% In any case, restart right away will erry 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}, Children}};
Error ->
Error
end.
producers_sup_spec(ClientPid, TopicName, Config0) ->
{Config, DelaySecs} =
take_delay_secs(Config0, topic_restart_delay_seconds,
?DEFAULT_SUP2_RESTART_DELAY),
Args = [?MODULE, {?SUP2, ClientPid, TopicName, Config}],
{ _Id = TopicName
, _Start = {supervisor3, start_link, Args}
, _Restart = {permanent, DelaySecs}
, _Shutdown = infinity
, _Type = supervisor
, _Module = [?MODULE]
}.
producer_spec(ClientPid, Topic, Partition, Config0) ->
{Config, DelaySecs} =
take_delay_secs(Config0, partition_restart_delay_seconds,
?DEFAULT_PRODUCER_RESTART_DELAY),
Args = [ClientPid, Topic, Partition, Config],
{ _Id = Partition
, _Start = {brod_producer, start_link, Args}
, _Restart = {permanent, DelaySecs}
, _Shutdown = 5000
, _Type = worker
, _Module = [brod_producer]
}.
%%%_* Internal Functions =======================================================
-spec take_delay_secs(producer_config(), atom(), integer()) ->
{producer_config(), integer()}.
take_delay_secs(Config, Name, DefaultValue) ->
Secs =
case proplists:get_value(Name, Config) of
N when is_integer(N) andalso N >= ?MIN_SUPERVISOR3_DELAY_SECS ->
N;
_ ->
DefaultValue
end,
{proplists:delete(Name, Config), Secs}.
%%%_* Emacs ====================================================================
%%% Local Variables:
%%% allout-layout: t
%%% erlang-indent-level: 2
%%% End: