Packages
brod
4.4.5
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
6
files changed
+173
additions
-9
deletions
| @@ -23,5 +23,5 @@ | |
| 23 23 | [{<<"kafka_protocol">>, |
| 24 24 | [{<<"app">>,<<"kafka_protocol">>}, |
| 25 25 | {<<"optional">>,false}, |
| 26 | - {<<"requirement">>,<<"4.2.6">>}]}]}. |
| 27 | - {<<"version">>,<<"4.4.4">>}. |
| 26 | + {<<"requirement">>,<<"4.2.7">>}]}]}. |
| 27 | + {<<"version">>,<<"4.4.5">>}. |
| @@ -1,4 +1,4 @@ | |
| 1 | - {deps, [{kafka_protocol, "4.2.6"}]}. |
| 1 | + {deps, [{kafka_protocol, "4.2.7"}]}. |
| 2 2 | {project_plugins, [{rebar3_lint, "~> 3.2.5"}]}. |
| 3 3 | {edoc_opts, [{preprocess, true}]}. |
| 4 4 | {erl_opts, [warnings_as_errors, warn_unused_vars,warn_shadow_vars,warn_obsolete_guard,debug_info]}. |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,brod, |
| 2 2 | [{description,"Apache Kafka Erlang client library"}, |
| 3 | - {vsn,"4.4.4"}, |
| 3 | + {vsn,"4.4.5"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol]}, |
| 6 6 | {env,[]}, |
| @@ -623,6 +623,7 @@ do_get_metadata(FetchMetadataFor, Topics, | |
| 623 623 | #state{ client_id = ClientId |
| 624 624 | , workers_tab = Ets |
| 625 625 | , config = Config |
| 626 | + , producers_sup = ProducersSup |
| 626 627 | } = State0) -> |
| 627 628 | FetchTopics = case FetchMetadataFor of |
| 628 629 | all -> all; %% in case no topic is given, get all |
| @@ -633,11 +634,18 @@ do_get_metadata(FetchMetadataFor, Topics, | |
| 633 634 | Request = brod_kafka_request:metadata(Conn, FetchTopics), |
| 634 635 | case request_sync(State, Request) of |
| 635 636 | {ok, #kpro_rsp{api = metadata, msg = Metadata}} -> |
| 636 | - TopicMetadataArray = kf(topics, Metadata), |
| 637 | + TopicsMetadata = kf(topics, Metadata), |
| 638 | + ok = maybe_start_partition_producer( |
| 639 | + filter_topics(TopicsMetadata), |
| 640 | + brod_producers_sup:count_started_children(ProducersSup), |
| 641 | + Topics, |
| 642 | + ProducersSup |
| 643 | + ), |
| 644 | + |
| 637 645 | UnknownTopicCacheTtl = config(unknown_topic_cache_ttl, Config, |
| 638 646 | ?DEFAULT_UNKNOWN_TOPIC_CACHE_TTL), |
| 639 | - ok = update_partitions_count_cache(Ets, TopicMetadataArray, UnknownTopicCacheTtl), |
| 640 | - ok = maybe_cache_unknown_topic_partition(Ets, Topics, TopicMetadataArray, |
| 647 | + ok = update_partitions_count_cache(Ets, TopicsMetadata, UnknownTopicCacheTtl), |
| 648 | + ok = maybe_cache_unknown_topic_partition(Ets, Topics, TopicsMetadata, |
| 641 649 | UnknownTopicCacheTtl), |
| 642 650 | {{ok, Metadata}, State}; |
| 643 651 | {error, Reason} -> |
| @@ -981,6 +989,67 @@ ensure_partition_workers(TopicName, State, F) -> | |
| 981 989 | end |
| 982 990 | end). |
| 983 991 | |
| 992 | + -spec maybe_start_partition_producer(TopicsMetadata, Producers, Topics, ProducerSup) -> ok |
| 993 | + when |
| 994 | + TopicsMetadata :: #{Key => non_neg_integer()}, |
| 995 | + Producers :: #{Key => non_neg_integer()}, |
| 996 | + Topics :: [topic()], |
| 997 | + ProducerSup :: pid(), |
| 998 | + Key :: topic(). |
| 999 | + maybe_start_partition_producer(_TopicsMetadata, _Producers, [], _ProducerSuf) -> |
| 1000 | + ok; |
| 1001 | + maybe_start_partition_producer(TopicsMetadata, Producers, [Topic | Rest], ProducerSup) when |
| 1002 | + is_map_key(Topic, TopicsMetadata), is_map_key(Topic, Producers) |
| 1003 | + -> |
| 1004 | + OldPartitionCnt = maps:get(Topic, Producers), |
| 1005 | + NewPartitionCnt = maps:get(Topic, TopicsMetadata), |
| 1006 | + |
| 1007 | + case NewPartitionCnt > OldPartitionCnt of |
| 1008 | + true -> |
| 1009 | + do_start_partition_producer(ProducerSup, Topic, OldPartitionCnt, NewPartitionCnt); |
| 1010 | + _ -> |
| 1011 | + ok |
| 1012 | + end, |
| 1013 | + |
| 1014 | + maybe_start_partition_producer(TopicsMetadata, Producers, Rest, ProducerSup); |
| 1015 | + maybe_start_partition_producer(TopicsMetadata, Producers, [_Topic | Rest], ProducerSup) -> |
| 1016 | + maybe_start_partition_producer(TopicsMetadata, Producers, Rest, ProducerSup). |
| 1017 | + |
| 1018 | + -spec do_start_partition_producer(pid(), topic(), non_neg_integer(), non_neg_integer()) -> ok. |
| 1019 | + do_start_partition_producer(ProducerSup, Topic, OldPartitionCnt, NewPartitionCnt) -> |
| 1020 | + %% Get producer config from partition producer which partition index is 0. |
| 1021 | + case brod_producers_sup:get_producer_config(ProducerSup, Topic, 0) of |
| 1022 | + {ok, Config} -> |
| 1023 | + %% start new producers for the partitions that are not started yet |
| 1024 | + lists:foreach( |
| 1025 | + fun |
| 1026 | + (Partition) when is_integer(Partition) -> |
| 1027 | + %% start a new producer for the partition |
| 1028 | + brod_producers_sup:start_producer(ProducerSup, self(), Topic, Partition, Config), |
| 1029 | + ok; |
| 1030 | + (_) -> |
| 1031 | + ok |
| 1032 | + end, |
| 1033 | + lists:seq(OldPartitionCnt, NewPartitionCnt - 1) |
| 1034 | + ); |
| 1035 | + _ -> |
| 1036 | + %% get producer config failed, do not start new producers |
| 1037 | + ok |
| 1038 | + end. |
| 1039 | + |
| 1040 | + -spec filter_topics([kpro:struct()]) -> #{topic() => non_neg_integer()}. |
| 1041 | + filter_topics(TopicsMetadataArray) -> |
| 1042 | + {_, NoErrors1} = |
| 1043 | + lists:partition(fun(#{error_code := E}) -> ?IS_ERROR(E) end, TopicsMetadataArray), |
| 1044 | + |
| 1045 | + NoErrors = [ |
| 1046 | + {TopicName, erlang:length(Partitions)} |
| 1047 | + || #{name := <<TopicName/binary>>, partitions := Partitions} <- NoErrors1, |
| 1048 | + is_list(Partitions) |
| 1049 | + ], |
| 1050 | + |
| 1051 | + maps:from_list(NoErrors). |
| 1052 | + |
| 984 1053 | %% Catches exit exceptions when making gen_server:call. |
| 985 1054 | -spec safe_gen_call(pid() | atom(), Call, Timeout) -> Return |
| 986 1055 | when Call :: term(), |
| @@ -27,7 +27,10 @@ | |
| 27 27 | , start_link/0 |
| 28 28 | , find_producer/3 |
| 29 29 | , start_producer/4 |
| 30 | + , start_producer/5 |
| 30 31 | , stop_producer/2 |
| 32 | + , get_producer_config/3 |
| 33 | + , count_started_children/1 |
| 31 34 | ]). |
| 32 35 | |
| 33 36 | -include("brod_int.hrl"). |
| @@ -60,6 +63,13 @@ start_producer(SupPid, ClientPid, TopicName, Config) -> | |
| 60 63 | Spec = producers_sup_spec(ClientPid, TopicName, Config), |
| 61 64 | brod_supervisor3:start_child(SupPid, Spec). |
| 62 65 | |
| 66 | + %% @doc Dynamically start a partition producer |
| 67 | + -spec start_producer(pid(), pid(), brod:topic(), brod:partition(), brod:producer_config()) -> |
| 68 | + {ok, pid()} | {error, any()}. |
| 69 | + start_producer(SupPid, ClientPid, Topic, Partition, Config) -> |
| 70 | + Spec = producer_spec(ClientPid, Topic, Partition, Config), |
| 71 | + brod_supervisor3:start_child(SupPid, Spec). |
| 72 | + |
| 63 73 | %% @doc Dynamically stop a per-topic supervisor |
| 64 74 | -spec stop_producer(pid(), brod:topic()) -> ok | {}. |
| 65 75 | stop_producer(SupPid, TopicName) -> |
| @@ -92,6 +102,56 @@ find_producer(SupPid, Topic, Partition) -> | |
| 92 102 | end |
| 93 103 | end. |
| 94 104 | |
| 105 | + %% @doc Get topic producer config from producer child specification. |
| 106 | + -spec get_producer_config(pid(), brod:topic(), brod:partition()) -> |
| 107 | + {ok, brod_producer:config()} | {error, Reason} when |
| 108 | + Reason :: {producer_not_found, brod:topic()} |
| 109 | + | {not_found, brod:topic(), brod:partition()} |
| 110 | + | {producer_down, any()}. |
| 111 | + get_producer_config(SupPid, Topic, Partition) -> |
| 112 | + case brod_supervisor3:find_child(SupPid, Topic) of |
| 113 | + [] -> |
| 114 | + {error, {producer_not_found, Topic}}; |
| 115 | + [PartitionsSupPid]-> |
| 116 | + try |
| 117 | + case brod_supervisor3:get_childspec(PartitionsSupPid, Partition) of |
| 118 | + {ok, {_Name, {brod_producer, start_link, [_, _, _, Config]}, _Restart, |
| 119 | + _Shutdown, _Type, _Module}} when is_list(Config) -> |
| 120 | + {ok, Config}; |
| 121 | + _ -> |
| 122 | + {error, {not_found, Topic, Partition}} |
| 123 | + end |
| 124 | + catch exit : {Reason, _} -> |
| 125 | + {error, {producer_down, Reason}} |
| 126 | + end |
| 127 | + end. |
| 128 | + |
| 129 | + %% @doc Count started children under a given supervisor process. |
| 130 | + -spec count_started_children(pid()) -> #{brod:topic() => non_neg_integer()}. |
| 131 | + count_started_children(SupRef) -> |
| 132 | + TopicSups = which_producers(SupRef), |
| 133 | + lists:foldl( |
| 134 | + fun |
| 135 | + ({Topic, Pid, _Type, _Mods}, Acc) when is_binary(Topic), is_pid(Pid) -> |
| 136 | + PartitionWorkers = which_producers(Pid), |
| 137 | + case length(PartitionWorkers) of |
| 138 | + 0 -> |
| 139 | + Acc; |
| 140 | + N -> |
| 141 | + Acc#{Topic => N} |
| 142 | + end; |
| 143 | + (_, Acc) -> |
| 144 | + Acc |
| 145 | + end, #{}, TopicSups). |
| 146 | + |
| 147 | + which_producers(Sup) -> |
| 148 | + try |
| 149 | + brod_supervisor3:which_children(Sup) |
| 150 | + catch |
| 151 | + _:_ -> |
| 152 | + [] |
| 153 | + end. |
| 154 | + |
| 95 155 | %% @doc brod_supervisor3 callback. |
| 96 156 | init(?TOPICS_SUP) -> |
| 97 157 | {ok, {{one_for_one, 0, 1}, []}}; |
Loading more files…