1+ % %% Licensed under the Apache License, Version 2.0 (the "License");
2+ % %% you may not use this file except in compliance with the License.
3+ % %% You may obtain a copy of the License at
4+ % %%
5+ % %% http://www.apache.org/licenses/LICENSE-2.0
6+ % %%
7+ % %% Unless required by applicable law or agreed to in writing, software
8+ % %% distributed under the License is distributed on an "AS IS" BASIS,
9+ % %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+ % %% See the License for the specific language governing permissions and
11+ % %% limitations under the License.
12+ % %%
13+
14+ % % @doc A `brod_partitions_sync' is a `gen_server' that is responsible for fetching
15+ % % the latest partition counts for a client and ensuring external changes to partitions
16+ % % are propogated to the clients and starts a producer for the partition if not present
17+ -module (brod_partitions_sync ).
18+ -behaviour (gen_server ).
19+
20+ -export ([
21+ start_link /3
22+ ]).
23+
24+ -export ([
25+ code_change /3 ,
26+ handle_call /3 ,
27+ handle_cast /2 ,
28+ handle_info /2 ,
29+ init /1 ,
30+ terminate /2
31+ ]).
32+
33+ -include (" brod_int.hrl"
34+
35+ -define (DEFAULT_SYNC_PARTITIONS_INTERVAL_SECONDS , 60 ).
36+
37+ -record (state , {
38+ client_id :: pid (),
39+ producers_sup :: pid (),
40+ interval :: non_neg_integer (),
41+ config :: brod_client :config ()
42+ }).
43+
44+ -type state () :: # state {}.
45+
46+ -spec start_link (pid (), pid (), brod_client :config ()) ->
47+ {ok , pid ()} | {error , any ()}.
48+ start_link (ClientId , ProducersSup , Config ) ->
49+ gen_server :start_link (? MODULE , {ClientId , ProducersSup , Config }, []).
50+
51+ -spec init ({pid (), pid (), brod_client :config ()}) ->
52+ {ok , state ()}.
53+ init ({ClientId , ProducersSup , Config }) ->
54+ Interval = sync_partition_interval (Config ),
55+ schedule_sync (Interval ),
56+ State = # state {
57+ client_id = ClientId ,
58+ producers_sup = ProducersSup ,
59+ interval = Interval ,
60+ config = Config
61+ },
62+
63+ {ok , State }.
64+
65+ % % @private
66+ handle_info (
67+ sync ,
68+ # state {
69+ client_id = Client ,
70+ config = Config ,
71+ producers_sup = Sup ,
72+ interval = Interval
73+ } = State
74+ ) ->
75+ sync_partitions (Client , Sup , Config ),
76+ sync_client (Client ),
77+ schedule_sync (Interval ),
78+ {noreply , State };
79+ handle_info (_Info , # state {} = State ) ->
80+ {noreply , State }.
81+
82+ % % @private
83+ handle_call (Call , _From , # state {} = State ) ->
84+ {reply , {error , {unsupported_call , Call }}, State }.
85+
86+ % % @private
87+ handle_cast (_Cast , # state {} = State ) ->
88+ {noreply , State }.
89+
90+ % % @private
91+ code_change (_OldVsn , # state {} = State , _Extra ) ->
92+ {ok , State }.
93+
94+ % % @private
95+ terminate (_Reason , _State ) ->
96+ ok .
97+
98+ % % @private
99+ sync_client (Client ) ->
100+ case brod_client :get_metadata_safe (Client , []) of
101+ {ok , _ } ->
102+ ok ;
103+ {error , Error } ->
104+ ? BROD_LOG_ERROR (" Partitions Sync Client MetaData Error: ~p " Error ]),
105+ ok
106+ end .
107+
108+ % % @private
109+ sync_partitions (Client , Sup , Config ) ->
110+ Producers = brod_supervisor3 :which_children (Sup ),
111+ TopicsList = lists :map (fun ({Topic , ProducerPid , _ , _ }) -> {Topic , ProducerPid } end , Producers ),
112+ TopicsMap = maps :from_list (TopicsList ),
113+ MetaData = get_metadata_for_topics (Client , maps :keys (TopicsMap )),
114+ lists :foreach (
115+ fun (#{name := Topic , partitions := Partitions }) ->
116+ ProducerPid = maps :get (Topic , TopicsMap ),
117+
118+ lists :foreach (
119+ fun (#{partition_index := Partition }) ->
120+ sync_partition (ProducerPid , Client , Topic , Partition , Config )
121+ end ,
122+ Partitions
123+ )
124+ end ,
125+ MetaData
126+ ),
127+ ok .
128+
129+ % % @private
130+ schedule_sync (Interval ) ->
131+ erlang :send_after (Interval , self (), sync ).
132+
133+ % % @private
134+ sync_partition (Sup , Client , Topic , Partition , Config ) ->
135+ case brod_producers_sup :find_producer (Sup , Topic , Partition ) of
136+ {ok , Pid } ->
137+ {ok , Pid };
138+ {error , _ } ->
139+ brod_producers_sup :start_producer (Sup , Client , Topic , Partition , Config )
140+ end .
141+
142+ % % @private
143+ sync_partition_interval (Config ) ->
144+ T = proplists :get_value (
145+ sync_partitions_interval_seconds , Config , ? DEFAULT_SYNC_PARTITIONS_INTERVAL_SECONDS
146+ ),
147+ timer :seconds (T ).
148+
149+ % % @private
150+ get_metadata_for_topics (_Client , []) ->
151+ [];
152+ % % @private
153+ get_metadata_for_topics (Client , Topics ) ->
154+ case brod_client :get_bootstrap (Client ) of
155+ {ok , {Hosts , Conn }} ->
156+ case brod_utils :get_metadata (Hosts , Topics , Conn ) of
157+ {ok , #{topics := MetaData }} ->
158+ MetaData ;
159+ {error , Error } ->
160+ ? BROD_LOG_ERROR (" Partitions Sync MetaData Error: ~p " Error ]),
161+ []
162+ end ;
163+ {error , Error } ->
164+ ? BROD_LOG_ERROR (" Partitions Sync Bootstrap Error: ~p " Error ]),
165+ []
166+ end .
0 commit comments