Packages
brod
3.7.8
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
+769
additions
-45
deletions
| @@ -1,7 +1,7 @@ | |
| 1 1 | KAFKA_VERSION ?= 1.1 |
| 2 2 | PROJECT = brod |
| 3 3 | PROJECT_DESCRIPTION = Kafka client library in Erlang |
| 4 | - PROJECT_VERSION = 3.7.7 |
| 4 | + PROJECT_VERSION = 3.7.8 |
| 5 5 | |
| 6 6 | DEPS = supervisor3 kafka_protocol |
| @@ -1,5 +1,5 @@ | |
| 1 1 | {<<"name">>,<<"brod">>}. |
| 2 | - {<<"version">>,<<"3.7.7">>}. |
| 2 | + {<<"version">>,<<"3.7.8">>}. |
| 3 3 | {<<"requirements">>, |
| 4 4 | #{<<"kafka_protocol">> => |
| 5 5 | #{<<"app">> => <<"kafka_protocol">>,<<"optional">> => false, |
| @@ -19,11 +19,11 @@ | |
| 19 19 | <<"src/brod_cli_pipe.erl">>,<<"src/brod_client.erl">>, |
| 20 20 | <<"src/brod_consumer.erl">>,<<"src/brod_consumers_sup.erl">>, |
| 21 21 | <<"src/brod_group_coordinator.erl">>,<<"src/brod_group_member.erl">>, |
| 22 | - <<"src/brod_group_subscriber.erl">>,<<"src/brod_kafka_apis.erl">>, |
| 23 | - <<"src/brod_kafka_request.erl">>,<<"src/brod_producer.erl">>, |
| 24 | - <<"src/brod_producer_buffer.erl">>,<<"src/brod_producers_sup.erl">>, |
| 25 | - <<"src/brod_sup.erl">>,<<"src/brod_topic_subscriber.erl">>, |
| 26 | - <<"src/brod_utils.erl">>]}. |
| 22 | + <<"src/brod_group_subscriber.erl">>,<<"src/brod_group_subscriber2.erl">>, |
| 23 | + <<"src/brod_kafka_apis.erl">>,<<"src/brod_kafka_request.erl">>, |
| 24 | + <<"src/brod_producer.erl">>,<<"src/brod_producer_buffer.erl">>, |
| 25 | + <<"src/brod_producers_sup.erl">>,<<"src/brod_sup.erl">>, |
| 26 | + <<"src/brod_topic_subscriber.erl">>,<<"src/brod_utils.erl">>]}. |
| 27 27 | {<<"licenses">>,[<<"Apache License 2.0">>]}. |
| 28 28 | {<<"links">>,[{<<"Github">>,<<"https://github.com/klarna/brod">>}]}. |
| 29 29 | {<<"build_tools">>,[<<"make">>,<<"rebar">>,<<"rebar3">>]}. |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,brod, |
| 2 2 | [{description,"Apache Kafka Erlang client library"}, |
| 3 | - {vsn,"3.7.7"}, |
| 3 | + {vsn,"3.7.8"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol,supervisor3]}, |
| 6 6 | {env,[]}, |
| @@ -397,31 +397,27 @@ handle_batches(Header, Batches, | |
| 397 397 | , partition = Partition |
| 398 398 | } = State0) -> |
| 399 399 | HighWmOffset = kpro:find(high_watermark, Header), |
| 400 | - State1 = case brod_utils:flatten_batches(BeginOffset, Batches) of |
| 401 | - [] -> |
| 402 | - %% flatten_batches might remove all actual messages |
| 403 | - %% (if they were all before BeginOffset), leaving us |
| 404 | - %% with nothing in this batch. Since there is no way |
| 405 | - %% to know how big the 'hole' is we can only bump |
| 406 | - %% begin_offset with +1 and try again. |
| 407 | - State0#state{ begin_offset = BeginOffset + 1 }; |
| 408 | - Messages -> |
| 409 | - MsgSet = #kafka_message_set{ topic = Topic |
| 410 | - , partition = Partition |
| 411 | - , high_wm_offset = HighWmOffset |
| 412 | - , messages = Messages |
| 413 | - }, |
| 414 | - ok = cast_to_subscriber(Subscriber, MsgSet), |
| 415 | - NewPendingAcks = add_pending_acks(PendingAcks, Messages), |
| 416 | - {value, ?PENDING(LastOffset, _LastMsgSize)} = |
| 417 | - queue:peek_r(NewPendingAcks#pending_acks.queue), |
| 418 | - State2 = State0#state{ pending_acks = NewPendingAcks |
| 419 | - , begin_offset = LastOffset + 1 |
| 420 | - }, |
| 421 | - maybe_shrink_max_bytes(State2, MsgSet#kafka_message_set.messages) |
| 422 | - end, |
| 423 | - State = maybe_send_fetch_request(State1), |
| 424 | - {noreply, State}. |
| 400 | + %% for API version 4 or higher, use last_stable_offset |
| 401 | + LatestOffset = kpro:find(last_stable_offset, Header, HighWmOffset), |
| 402 | + {NewBeginOffset, Messages} = |
| 403 | + brod_utils:flatten_batches(BeginOffset, Header, Batches), |
| 404 | + State1 = State0#state{begin_offset = NewBeginOffset}, |
| 405 | + State = |
| 406 | + case Messages =:= [] of |
| 407 | + true -> |
| 408 | + State1; |
| 409 | + false -> |
| 410 | + MsgSet = #kafka_message_set{ topic = Topic |
| 411 | + , partition = Partition |
| 412 | + , high_wm_offset = LatestOffset |
| 413 | + , messages = Messages |
| 414 | + }, |
| 415 | + ok = cast_to_subscriber(Subscriber, MsgSet), |
| 416 | + NewPendingAcks = add_pending_acks(PendingAcks, Messages), |
| 417 | + State2 = State1#state{pending_acks = NewPendingAcks}, |
| 418 | + maybe_shrink_max_bytes(State2, MsgSet#kafka_message_set.messages) |
| 419 | + end, |
| 420 | + {noreply, maybe_send_fetch_request(State)}. |
| 425 421 | |
| 426 422 | %% Add received offsets to pending queue. |
| 427 423 | add_pending_acks(PendingAcks, Messages) -> |
| @@ -0,0 +1,663 @@ | |
| 1 | + %%% |
| 2 | + %%% Copyright (c) 2019 Klarna Bank AB (publ) |
| 3 | + %%% |
| 4 | + %%% Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | + %%% you may not use this file except in compliance with the License. |
| 6 | + %%% You may obtain a copy of the License at |
| 7 | + %%% |
| 8 | + %%% http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | + %%% |
| 10 | + %%% Unless required by applicable law or agreed to in writing, software |
| 11 | + %%% distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | + %%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | + %%% See the License for the specific language governing permissions and |
| 14 | + %%% limitations under the License. |
| 15 | + %%% |
| 16 | + |
| 17 | + %%%============================================================================= |
| 18 | + %%% @doc |
| 19 | + %%% A group subscriber is a gen_statem which managements partition assignments |
| 20 | + %%% and spawn per-partition workers to process message-sets. |
| 21 | + %%% |
| 22 | + %%% The main difference between `brod_group_subscriber' and |
| 23 | + %%% `brod_group_subscriber2' is that in `brod_group_subscriber2', per-partition |
| 24 | + %%% workers are spawned to subscribe message-sets, while in |
| 25 | + %%% `brod_group_subscriber' it's a single process subscribes all assigned |
| 26 | + %%% partitions. |
| 27 | + %%% |
| 28 | + %%% An overview of what it does behind the scene: |
| 29 | + %%% 1. Start a consumer group coordinator to manage the consumer group states, |
| 30 | + %%% @see brod_group_coordinator:start_link/4. |
| 31 | + %%% 2. Start (if not already started) topic-consumers (pollers), and spawn |
| 32 | + %%% worker processes to subscribe to `brod_consumer' when group assignment |
| 33 | + %%% is received from the group leader, @see brod:start_consumer/3. |
| 34 | + %%% 3. In each worker, call CallbackModule:handle_message_set/4 when messages |
| 35 | + %%% are received from partition consumer. |
| 36 | + %%% 4. Send acknowledged offsets to group coordinator which will be committed |
| 37 | + %%% to kafka periodically. |
| 38 | + %%% @end |
| 39 | + %%%============================================================================= |
| 40 | + |
| 41 | + -module(brod_group_subscriber2). |
| 42 | + |
| 43 | + -behaviour(gen_statem). |
| 44 | + -behaviour(brod_group_member). |
| 45 | + |
| 46 | + -export([ ack/4 |
| 47 | + , ack/5 |
| 48 | + , commit/1 |
| 49 | + , commit/4 |
| 50 | + , start_link/7 |
| 51 | + , start_link/8 |
| 52 | + , stop/1 |
| 53 | + ]). |
| 54 | + |
| 55 | + %% callbacks for brod_group_coordinator |
| 56 | + -export([ get_committed_offsets/2 |
| 57 | + , assignments_received/4 |
| 58 | + , assignments_revoked/1 |
| 59 | + , assign_partitions/3 |
| 60 | + , user_data/1 |
| 61 | + ]). |
| 62 | + |
| 63 | + -export([ code_change/3 |
| 64 | + , handle_call/3 |
| 65 | + , handle_cast/2 |
| 66 | + , handle_info/2 |
| 67 | + , init/1 |
| 68 | + , terminate/2 |
| 69 | + ]). |
| 70 | + |
| 71 | + -include("brod_int.hrl"). |
| 72 | + |
| 73 | + -type cb_state() :: term(). |
| 74 | + -type member_id() :: brod:group_member_id(). |
| 75 | + |
| 76 | + %% Initialize the callback module s state. |
| 77 | + -callback init(brod:group_id(), term()) -> {ok, cb_state()}. |
| 78 | + |
| 79 | + %% Handle a message. Return one of: |
| 80 | + %% |
| 81 | + %% {ok, NewCallbackState}: |
| 82 | + %% The subscriber has received the message for processing async-ly. |
| 83 | + %% It should call brod_group_subscriber:ack/4 to acknowledge later. |
| 84 | + %% |
| 85 | + %% {ok, ack, NewCallbackState} |
| 86 | + %% The subscriber has completed processing the message. |
| 87 | + %% |
| 88 | + %% {ok, ack_no_commit, NewCallbackState} |
| 89 | + %% The subscriber has completed processing the message, but it |
| 90 | + %% is not ready to commit offset yet. It should call |
| 91 | + %% brod_group_subscriber:commit/4 later. |
| 92 | + %% |
| 93 | + %% While this callback function is being evaluated, the fetch-ahead |
| 94 | + %% partition-consumers are fetching more messages behind the scene |
| 95 | + %% unless prefetch_count and prefetch_bytes are set to 0 in consumer config. |
| 96 | + %% |
| 97 | + -callback handle_message(brod:topic(), |
| 98 | + brod:partition(), |
| 99 | + brod:message(), |
| 100 | + cb_state()) -> {ok, cb_state()} | |
| 101 | + {ok, ack, cb_state()} | |
| 102 | + {ok, ack_no_commit, cb_state()}. |
| 103 | + |
| 104 | + %% This callback is called only when subscriber is to commit offsets locally |
| 105 | + %% instead of kafka. |
| 106 | + %% Return {ok, Offsets, cb_state()} where Offsets can be [], |
| 107 | + %% or only the ones that are found in e.g. local storage or database. |
| 108 | + %% For the topic-partitions which have no committed offset found, |
| 109 | + %% the consumer will take 'begin_offset' in consumer config as the start point |
| 110 | + %% of data stream. If 'begin_offset' is not found in consumer config, the |
| 111 | + %% default value -1 (latest) is used. |
| 112 | + % |
| 113 | + % commented out as it's an optional callback |
| 114 | + %-callback get_committed_offsets(brod:group_id(), |
| 115 | + % [{brod:topic(), brod:partition()}], |
| 116 | + % cb_state()) -> {ok, |
| 117 | + % [{{brod:topic(), |
| 118 | + % brod:partition()}, |
| 119 | + % brod:offset()}], |
| 120 | + % cb_state()}. |
| 121 | + % |
| 122 | + %% This function is called only when 'partition_assignment_strategy' is |
| 123 | + %% 'callback_implemented' in group config. |
| 124 | + %% The first element in the group member list is ensured to be the group leader. |
| 125 | + % |
| 126 | + % commented out as it's an optional callback |
| 127 | + %-callback assign_partitions([brod:group_member()], |
| 128 | + % [{brod:topic(), brod:partition()}], |
| 129 | + % cb_state()) -> [{brod:group_member_id(), |
| 130 | + % [brod:partition_assignment()]}]. |
| 131 | + |
| 132 | + -define(DOWN(Reason), {down, brod_utils:os_time_utc_str(), Reason}). |
| 133 | + |
| 134 | + -record(consumer, |
| 135 | + { topic_partition :: {brod:topic(), brod:partition()} |
| 136 | + , consumer_pid :: ?undef %% initial state |
| 137 | + | pid() %% normal state |
| 138 | + | {down, string(), any()} %% consumer restarting |
| 139 | + , consumer_mref :: ?undef | reference() |
| 140 | + , begin_offset :: ?undef | brod:offset() |
| 141 | + , acked_offset :: ?undef | brod:offset() |
| 142 | + , last_offset :: ?undef | brod:offset() |
| 143 | + }). |
| 144 | + |
| 145 | + -type consumer() :: #consumer{}. |
| 146 | + |
| 147 | + -type ack_ref() :: {brod:topic(), brod:partition(), brod:offset()}. |
| 148 | + |
| 149 | + -record(state, |
| 150 | + { client :: brod:client() |
| 151 | + , client_mref :: reference() |
| 152 | + , groupId :: brod:group_id() |
| 153 | + , memberId :: ?undef | member_id() |
| 154 | + , generationId :: ?undef | brod:group_generation_id() |
| 155 | + , coordinator :: pid() |
| 156 | + , consumers = [] :: [consumer()] |
| 157 | + , consumer_config :: brod:consumer_config() |
| 158 | + , is_blocked = false :: boolean() |
| 159 | + , subscribe_tref :: ?undef | reference() |
| 160 | + , cb_module :: module() |
| 161 | + , cb_state :: cb_state() |
| 162 | + , message_type :: message | message_set |
| 163 | + }). |
| 164 | + |
| 165 | + -type state() :: #state{}. |
| 166 | + |
| 167 | + %% delay 2 seconds retry the failed subscription to partition consumer process |
| 168 | + -define(RESUBSCRIBE_DELAY, 2000). |
| 169 | + |
| 170 | + -define(LO_CMD_SUBSCRIBE_PARTITIONS, '$subscribe_partitions'). |
| 171 | + |
| 172 | + %%%_* APIs ===================================================================== |
| 173 | + |
| 174 | + %% @doc Start (link) a group subscriber. |
| 175 | + %% Client: |
| 176 | + %% Client ID (or pid, but not recommended) of the brod client. |
| 177 | + %% GroupId: |
| 178 | + %% Consumer group ID which should be unique per kafka cluster |
| 179 | + %% Topics: |
| 180 | + %% Predefined set of topic names to join the group. |
| 181 | + %% NOTE: The group leader member will collect topics from all members and |
| 182 | + %% assign all collected topic-partitions to members in the group. |
| 183 | + %% i.e. members can join with arbitrary set of topics. |
| 184 | + %% GroupConfig: |
| 185 | + %% For group coordinator, @see brod_group_coordinator:start_link/5 |
| 186 | + %% ConsumerConfig: |
| 187 | + %% For partition consumer, @see brod_consumer:start_link/4 |
| 188 | + %% CbModule: |
| 189 | + %% Callback module which should have the callback functions |
| 190 | + %% implemented for message processing. |
| 191 | + %% CbInitArg: |
| 192 | + %% The term() that is going to be passed to CbModule:init/1 when |
| 193 | + %% initializing the subscriber. |
| 194 | + %% @end |
| 195 | + -spec start_link(brod:client(), brod:group_id(), [brod:topic()], |
| 196 | + brod:group_config(), brod:consumer_config(), |
| 197 | + module(), term()) -> {ok, pid()} | {error, any()}. |
| 198 | + start_link(Client, GroupId, Topics, GroupConfig, |
| 199 | + ConsumerConfig, CbModule, CbInitArg) -> |
| 200 | + start_link(Client, GroupId, Topics, GroupConfig, ConsumerConfig, |
| 201 | + message, CbModule, CbInitArg). |
| 202 | + |
| 203 | + %% @doc Start (link) a group subscriber. |
| 204 | + %% Client: |
| 205 | + %% Client ID (or pid, but not recommended) of the brod client. |
| 206 | + %% GroupId: |
| 207 | + %% Consumer group ID which should be unique per kafka cluster |
| 208 | + %% Topics: |
| 209 | + %% Predefined set of topic names to join the group. |
| 210 | + %% NOTE: The group leader member will collect topics from all members and |
| 211 | + %% assign all collected topic-partitions to members in the group. |
| 212 | + %% i.e. members can join with arbitrary set of topics. |
| 213 | + %% GroupConfig: |
| 214 | + %% For group coordinator, @see brod_group_coordinator:start_link/5 |
| 215 | + %% ConsumerConfig: |
| 216 | + %% For partition consumer, @see brod_consumer:start_link/4 |
| 217 | + %% MessageType: |
| 218 | + %% The type of message that is going to be handled by the callback |
| 219 | + %% module. Can be either message or message set. |
| 220 | + %% CbModule: |
| 221 | + %% Callback module which should have the callback functions |
| 222 | + %% implemented for message processing. |
| 223 | + %% CbInitArg: |
| 224 | + %% The term() that is going to be passed to CbModule:init/1 when |
| 225 | + %% initializing the subscriber. |
| 226 | + %% @end |
| 227 | + -spec start_link(brod:client(), brod:group_id(), [brod:topic()], |
| 228 | + brod:group_config(), brod:consumer_config(), |
| 229 | + message | message_set, |
| 230 | + module(), term()) -> {ok, pid()} | {error, any()}. |
| 231 | + start_link(Client, GroupId, Topics, GroupConfig, |
| 232 | + ConsumerConfig, MessageType, CbModule, CbInitArg) -> |
| 233 | + Args = {Client, GroupId, Topics, GroupConfig, |
| 234 | + ConsumerConfig, MessageType, CbModule, CbInitArg}, |
| 235 | + gen_server:start_link(?MODULE, Args, []). |
| 236 | + |
| 237 | + %% @doc Stop group subscriber, wait for pid DOWN before return. |
| 238 | + -spec stop(pid()) -> ok. |
| 239 | + stop(Pid) -> |
| 240 | + Mref = erlang:monitor(process, Pid), |
| 241 | + ok = gen_server:cast(Pid, stop), |
| 242 | + receive |
| 243 | + {'DOWN', Mref, process, Pid, _Reason} -> |
| 244 | + ok |
| 245 | + end. |
| 246 | + |
| 247 | + %% @doc Acknowledge and commit an offset. |
| 248 | + %% The subscriber may ack a later (greater) offset which will be considered |
| 249 | + %% as multi-acking the earlier (smaller) offsets. This also means that |
| 250 | + %% disordered acks may overwrite offset commits and lead to unnecessary |
| 251 | + %% message re-delivery in case of restart. |
| 252 | + %% @end |
| 253 | + -spec ack(pid(), brod:topic(), brod:partition(), brod:offset()) -> ok. |
| 254 | + ack(Pid, Topic, Partition, Offset) -> |
| 255 | + ack(Pid, Topic, Partition, Offset, true). |
| 256 | + |
| 257 | + %% @doc Acknowledge an offset. |
| 258 | + %% This call may or may not commit group subscriber offset depending on |
| 259 | + %% the value of `Commit' argument |
| 260 | + %% @end |
| 261 | + -spec ack(pid(), brod:topic(), brod:partition(), brod:offset(), boolean()) -> |
| 262 | + ok. |
| 263 | + ack(Pid, Topic, Partition, Offset, Commit) -> |
| 264 | + gen_server:cast(Pid, {ack, Topic, Partition, Offset, Commit}). |
| 265 | + |
| 266 | + %% @doc Commit all acked offsets. NOTE: This is an async call. |
| 267 | + -spec commit(pid()) -> ok. |
| 268 | + commit(Pid) -> |
| 269 | + gen_server:cast(Pid, commit_offsets). |
| 270 | + |
| 271 | + %% @doc Commit offset for a topic. This is an asynchronous call |
| 272 | + -spec commit(pid(), brod:topic(), brod:partition(), brod:offset()) -> ok. |
| 273 | + commit(Pid, Topic, Partition, Offset) -> |
| 274 | + gen_server:cast(Pid, {commit_offset, Topic, Partition, Offset}). |
| 275 | + |
| 276 | + user_data(_Pid) -> <<>>. |
| 277 | + |
| 278 | + %%%_* APIs for group coordinator =============================================== |
| 279 | + |
| 280 | + %% @doc Called by group coordinator when there is new assignment received. |
| 281 | + -spec assignments_received(pid(), member_id(), integer(), |
| 282 | + brod:received_assignments()) -> ok. |
| 283 | + assignments_received(Pid, MemberId, GenerationId, TopicAssignments) -> |
| 284 | + gen_server:cast(Pid, {new_assignments, MemberId, |
| 285 | + GenerationId, TopicAssignments}). |
| 286 | + |
| 287 | + %% @doc Called by group coordinator before re-joining the consumer group. |
| 288 | + -spec assignments_revoked(pid()) -> ok. |
| 289 | + assignments_revoked(Pid) -> |
| 290 | + gen_server:call(Pid, unsubscribe_all_partitions, infinity). |
| 291 | + |
| 292 | + %% @doc This function is called only when `partition_assignment_strategy' |
| 293 | + %% is set for `callback_implemented' in group config. |
| 294 | + %% @end |
| 295 | + -spec assign_partitions(pid(), [brod:group_member()], |
| 296 | + [{brod:topic(), brod:partition()}]) -> |
| 297 | + [{member_id(), [brod:partition_assignment()]}]. |
| 298 | + assign_partitions(Pid, Members, TopicPartitionList) -> |
| 299 | + Call = {assign_partitions, Members, TopicPartitionList}, |
| 300 | + gen_server:call(Pid, Call, infinity). |
| 301 | + |
| 302 | + %% @doc Called by group coordinator when initializing the assignments |
| 303 | + %% for subscriber. |
| 304 | + %% |
| 305 | + %% NOTE: This function is called only when `offset_commit_policy' is set to |
| 306 | + %% `consumer_managed' in group config. |
| 307 | + %% |
| 308 | + %% NOTE: The committed offsets should be the offsets for successfully processed |
| 309 | + %% (acknowledged) messages, not the `begin_offset' to start fetching from. |
| 310 | + %% @end |
| 311 | + -spec get_committed_offsets(pid(), [{brod:topic(), brod:partition()}]) -> |
| 312 | + {ok, [{{brod:topic(), brod:partition()}, brod:offset()}]}. |
| 313 | + get_committed_offsets(Pid, TopicPartitions) -> |
| 314 | + gen_server:call(Pid, {get_committed_offsets, TopicPartitions}, infinity). |
| 315 | + |
| 316 | + %%%_* gen_server callbacks ===================================================== |
| 317 | + |
| 318 | + init({Client, GroupId, Topics, GroupConfig, |
| 319 | + ConsumerConfig, MessageType, CbModule, CbInitArg}) -> |
| 320 | + ok = brod_utils:assert_client(Client), |
| 321 | + ok = brod_utils:assert_group_id(GroupId), |
| 322 | + ok = brod_utils:assert_topics(Topics), |
| 323 | + {ok, CbState} = CbModule:init(GroupId, CbInitArg), |
| 324 | + {ok, Pid} = brod_group_coordinator:start_link(Client, GroupId, Topics, |
| 325 | + GroupConfig, ?MODULE, self()), |
| 326 | + State = #state{ client = Client |
| 327 | + , client_mref = erlang:monitor(process, Client) |
| 328 | + , groupId = GroupId |
| 329 | + , coordinator = Pid |
| 330 | + , consumer_config = ConsumerConfig |
| 331 | + , cb_module = CbModule |
| 332 | + , cb_state = CbState |
| 333 | + , message_type = MessageType |
| 334 | + }, |
| 335 | + {ok, State}. |
| 336 | + |
| 337 | + handle_info({_ConsumerPid, #kafka_message_set{} = MsgSet}, State0) -> |
| 338 | + State = handle_consumer_delivery(MsgSet, State0), |
| 339 | + {noreply, State}; |
| 340 | + handle_info({'DOWN', Mref, process, _Pid, _Reason}, |
| 341 | + #state{client_mref = Mref} = State) -> |
| 342 | + %% restart, my supervisor should restart me |
| 343 | + %% brod_client DOWN reason is discarded as it should have logged |
| 344 | + %% in its crash log |
| 345 | + {stop, client_down, State}; |
| 346 | + handle_info({'DOWN', _Mref, process, Pid, Reason}, |
| 347 | + #state{consumers = Consumers} = State) -> |
| 348 | + case get_consumer(Pid, Consumers) of |
| 349 | + #consumer{} = Consumer -> |
| 350 | + NewConsumer = Consumer#consumer{ consumer_pid = ?DOWN(Reason) |
| 351 | + , consumer_mref = ?undef |
| 352 | + }, |
| 353 | + NewConsumers = put_consumer(NewConsumer, Consumers), |
| 354 | + NewState = State#state{consumers = NewConsumers}, |
| 355 | + {noreply, NewState}; |
| 356 | + false -> |
| 357 | + {noreply, State} |
| 358 | + end; |
| 359 | + handle_info(?LO_CMD_SUBSCRIBE_PARTITIONS, State) -> |
| 360 | + NewState = |
| 361 | + case State#state.is_blocked of |
| 362 | + true -> |
| 363 | + State; |
| 364 | + false -> |
| 365 | + {ok, #state{} = St} = subscribe_partitions(State), |
| 366 | + St |
| 367 | + end, |
| 368 | + Tref = start_subscribe_timer(?undef, ?RESUBSCRIBE_DELAY), |
| 369 | + {noreply, NewState#state{subscribe_tref = Tref}}; |
| 370 | + handle_info(Info, State) -> |
| 371 | + log(State, info, "discarded message:~p", [Info]), |
| 372 | + {noreply, State}. |
| 373 | + |
| 374 | + handle_call({get_committed_offsets, TopicPartitions}, _From, |
| 375 | + #state{ groupId = GroupId |
| 376 | + , cb_module = CbModule |
| 377 | + , cb_state = CbState |
| 378 | + } = State) -> |
| 379 | + case CbModule:get_committed_offsets(GroupId, TopicPartitions, CbState) of |
| 380 | + {ok, Result, NewCbState} -> |
| 381 | + NewState = State#state{cb_state = NewCbState}, |
| 382 | + {reply, {ok, Result}, NewState}; |
| 383 | + Unknown -> |
| 384 | + erlang:error({bad_return_value, |
| 385 | + {CbModule, get_committed_offsets, Unknown}}) |
| 386 | + end; |
| 387 | + handle_call({assign_partitions, Members, TopicPartitions}, _From, |
| 388 | + #state{ cb_module = CbModule |
| 389 | + , cb_state = CbState |
| 390 | + } = State) -> |
| 391 | + case CbModule:assign_partitions(Members, TopicPartitions, CbState) of |
| 392 | + {NewCbState, Result} -> |
| 393 | + {reply, Result, State#state{ cb_state = NewCbState }}; |
| 394 | + %% Returning an updated cb_state is optional and clients that implemented |
| 395 | + %% brod prior to version 3.7.1 need this backwards compatibly case clause |
| 396 | + Result when is_list(Result) -> |
| 397 | + {reply, Result, State} |
| 398 | + end; |
| 399 | + handle_call(unsubscribe_all_partitions, _From, |
| 400 | + #state{ consumers = Consumers |
| 401 | + } = State) -> |
| 402 | + lists:foreach( |
| 403 | + fun(#consumer{ consumer_pid = ConsumerPid |
| 404 | + , consumer_mref = ConsumerMref |
| 405 | + }) -> |
| 406 | + case is_pid(ConsumerPid) of |
| 407 | + true -> |
| 408 | + _ = brod:unsubscribe(ConsumerPid, self()), |
| 409 | + _ = erlang:demonitor(ConsumerMref, [flush]); |
| 410 | + false -> |
| 411 | + ok |
| 412 | + end |
| 413 | + end, Consumers), |
| 414 | + {reply, ok, State#state{ consumers = [] |
| 415 | + , is_blocked = true |
| 416 | + }}; |
| 417 | + handle_call(Call, _From, State) -> |
| 418 | + {reply, {error, {unknown_call, Call}}, State}. |
| 419 | + |
| 420 | + handle_cast({ack, Topic, Partition, Offset, Commit}, State) -> |
| 421 | + AckRef = {Topic, Partition, Offset}, |
| 422 | + NewState = handle_ack(AckRef, State, Commit), |
| 423 | + {noreply, NewState}; |
| 424 | + handle_cast(commit_offsets, State) -> |
| 425 | + ok = brod_group_coordinator:commit_offsets(State#state.coordinator), |
| 426 | + {noreply, State}; |
| 427 | + handle_cast({commit_offset, Topic, Partition, Offset}, State) -> |
| 428 | + #state{ coordinator = Coordinator |
| 429 | + , generationId = GenerationId |
| 430 | + } = State, |
| 431 | + do_commit_ack(Coordinator, GenerationId, Topic, Partition, Offset), |
| 432 | + {noreply, State}; |
| 433 | + handle_cast({new_assignments, MemberId, GenerationId, Assignments}, |
| 434 | + #state{ client = Client |
| 435 | + , consumer_config = ConsumerConfig |
| 436 | + , subscribe_tref = Tref |
| 437 | + } = State) -> |
| 438 | + AllTopics = |
| 439 | + lists:map(fun(#brod_received_assignment{topic = Topic}) -> |
| 440 | + Topic |
| 441 | + end, Assignments), |
| 442 | + lists:foreach( |
| 443 | + fun(Topic) -> |
| 444 | + ok = brod:start_consumer(Client, Topic, ConsumerConfig) |
| 445 | + end, lists:usort(AllTopics)), |
| 446 | + Consumers = |
| 447 | + [ #consumer{ topic_partition = {Topic, Partition} |
| 448 | + , consumer_pid = ?undef |
| 449 | + , begin_offset = BeginOffset |
| 450 | + , acked_offset = ?undef |
| 451 | + } |
| 452 | + || #brod_received_assignment{ topic = Topic |
| 453 | + , partition = Partition |
| 454 | + , begin_offset = BeginOffset |
| 455 | + } <- Assignments |
| 456 | + ], |
| 457 | + NewState = State#state{ consumers = Consumers |
| 458 | + , is_blocked = false |
| 459 | + , memberId = MemberId |
| 460 | + , generationId = GenerationId |
| 461 | + , subscribe_tref = start_subscribe_timer(Tref, 0) |
| 462 | + }, |
| 463 | + {noreply, NewState}; |
| 464 | + handle_cast(stop, State) -> |
| 465 | + {stop, normal, State}; |
| 466 | + handle_cast(_Cast, State) -> |
| 467 | + {noreply, State}. |
| 468 | + |
| 469 | + code_change(_OldVsn, State, _Extra) -> |
| 470 | + {ok, State}. |
| 471 | + |
| 472 | + terminate(_Reason, #state{}) -> |
| 473 | + ok. |
| 474 | + |
| 475 | + %%%_* Internal Functions ======================================================= |
| 476 | + |
| 477 | + handle_consumer_delivery(#kafka_message_set{ topic = Topic |
| 478 | + , partition = Partition |
| 479 | + , messages = Messages |
| 480 | + } = MsgSet, |
| 481 | + #state{ message_type = MsgType |
| 482 | + , consumers = Consumers0 |
| 483 | + } = State0) -> |
| 484 | + case get_consumer({Topic, Partition}, Consumers0) of |
| 485 | + #consumer{} = C -> |
| 486 | + Consumers = update_last_offset(Messages, C, Consumers0), |
| 487 | + State = State0#state{consumers = Consumers}, |
| 488 | + case MsgType of |
| 489 | + message -> handle_messages(Topic, Partition, Messages, State); |
| 490 | + message_set -> handle_message_set(MsgSet, State) |
| 491 | + end; |
| 492 | + false -> |
| 493 | + State0 |
| 494 | + end. |
| 495 | + |
| 496 | + update_last_offset(Messages, Consumer0, Consumers) -> |
| 497 | + %% brod_consumer never delivers empty message set, lists:last is safe |
| 498 | + #kafka_message{offset = LastOffset} = lists:last(Messages), |
| 499 | + Consumer = Consumer0#consumer{last_offset = LastOffset}, |
| 500 | + put_consumer(Consumer, Consumers). |
| 501 | + |
| 502 | + -spec start_subscribe_timer(?undef | reference(), timeout()) -> reference(). |
| 503 | + start_subscribe_timer(?undef, Delay) -> |
| 504 | + erlang:send_after(Delay, self(), ?LO_CMD_SUBSCRIBE_PARTITIONS); |
| 505 | + start_subscribe_timer(Ref, _Delay) when is_reference(Ref) -> |
| 506 | + %% The old timer is not expired, keep waiting |
| 507 | + %% A bit delay on subscribing to brod_consumer is fine |
| 508 | + Ref. |
| 509 | + |
| 510 | + handle_message_set(MessageSet, State) -> |
| 511 | + #kafka_message_set{ topic = Topic |
| 512 | + , partition = Partition |
| 513 | + , messages = Messages |
| 514 | + } = MessageSet, |
| 515 | + #state{cb_module = CbModule, cb_state = CbState} = State, |
| 516 | + {AckNow, CommitNow, NewCbState} = |
| 517 | + case CbModule:handle_message(Topic, Partition, MessageSet, CbState) of |
| 518 | + {ok, NewCbState_} -> |
| 519 | + {false, false, NewCbState_}; |
| 520 | + {ok, ack, NewCbState_} -> |
| 521 | + {true, true, NewCbState_}; |
| 522 | + {ok, ack_no_commit, NewCbState_} -> |
| 523 | + {true, false, NewCbState_}; |
| 524 | + Unknown -> |
| 525 | + erlang:error({bad_return_value, |
| 526 | + {CbModule, handle_message, Unknown}}) |
| 527 | + end, |
| 528 | + State1 = State#state{cb_state = NewCbState}, |
| 529 | + case AckNow of |
| 530 | + true -> |
| 531 | + LastMessage = lists:last(Messages), |
| 532 | + LastOffset = LastMessage#kafka_message.offset, |
| 533 | + AckRef = {Topic, Partition, LastOffset}, |
| 534 | + handle_ack(AckRef, State1, CommitNow); |
| 535 | + false -> State1 |
| 536 | + end. |
| 537 | + |
| 538 | + handle_messages(_Topic, _Partition, [], State) -> |
| 539 | + State; |
| 540 | + handle_messages(Topic, Partition, [Msg | Rest], State) -> |
| 541 | + #kafka_message{offset = Offset} = Msg, |
| 542 | + #state{cb_module = CbModule, cb_state = CbState} = State, |
| 543 | + AckRef = {Topic, Partition, Offset}, |
| 544 | + {AckNow, CommitNow, NewCbState} = |
| 545 | + case CbModule:handle_message(Topic, Partition, Msg, CbState) of |
| 546 | + {ok, NewCbState_} -> |
| 547 | + {false, false, NewCbState_}; |
| 548 | + {ok, ack, NewCbState_} -> |
| 549 | + {true, true, NewCbState_}; |
| 550 | + {ok, ack_no_commit, NewCbState_} -> |
| 551 | + {true, false, NewCbState_}; |
| 552 | + Unknown -> |
| 553 | + erlang:error({bad_return_value, |
| 554 | + {CbModule, handle_message, Unknown}}) |
| 555 | + end, |
| 556 | + State1 = State#state{cb_state = NewCbState}, |
| 557 | + NewState = |
| 558 | + case AckNow of |
| 559 | + true -> handle_ack(AckRef, State1, CommitNow); |
| 560 | + false -> State1 |
| 561 | + end, |
| 562 | + handle_messages(Topic, Partition, Rest, NewState). |
| 563 | + |
| 564 | + -spec handle_ack(ack_ref(), state(), boolean()) -> state(). |
| 565 | + handle_ack(AckRef, #state{ generationId = GenerationId |
| 566 | + , consumers = Consumers |
| 567 | + , coordinator = Coordinator |
| 568 | + } = State, CommitNow) -> |
| 569 | + {Topic, Partition, Offset} = AckRef, |
| 570 | + case get_consumer({Topic, Partition}, Consumers) of |
| 571 | + #consumer{consumer_pid = ConsumerPid} = Consumer when CommitNow -> |
| 572 | + ok = consume_ack(ConsumerPid, Offset), |
| 573 | + ok = do_commit_ack(Coordinator, GenerationId, Topic, Partition, Offset), |
| 574 | + NewConsumer = Consumer#consumer{acked_offset = Offset}, |
| 575 | + NewConsumers = put_consumer(NewConsumer, Consumers), |
| 576 | + State#state{consumers = NewConsumers}; |
| 577 | + #consumer{consumer_pid = ConsumerPid} -> |
| 578 | + ok = consume_ack(ConsumerPid, Offset), |
| 579 | + State; |
| 580 | + false -> |
| 581 | + %% Stale async-ack, discard. |
| 582 | + State |
| 583 | + end. |
| 584 | + |
| 585 | + %% Tell consumer process to fetch more (if pre-fetch count/byte limit allows). |
| 586 | + consume_ack(Pid, Offset) -> |
| 587 | + is_pid(Pid) andalso brod:consume_ack(Pid, Offset), |
| 588 | + ok. |
| 589 | + |
| 590 | + %% Send an async message to group coordinator for offset commit. |
| 591 | + do_commit_ack(Pid, GenerationId, Topic, Partition, Offset) -> |
| 592 | + ok = brod_group_coordinator:ack(Pid, GenerationId, Topic, Partition, Offset). |
| 593 | + |
| 594 | + subscribe_partitions(#state{ client = Client |
| 595 | + , consumers = Consumers0 |
| 596 | + } = State) -> |
| 597 | + Consumers = |
| 598 | + lists:map(fun(C) -> subscribe_partition(Client, C) end, Consumers0), |
| 599 | + {ok, State#state{consumers = Consumers}}. |
| 600 | + |
| 601 | + subscribe_partition(Client, Consumer) -> |
| 602 | + #consumer{ topic_partition = {Topic, Partition} |
| 603 | + , consumer_pid = Pid |
| 604 | + , begin_offset = BeginOffset0 |
| 605 | + , acked_offset = AckedOffset |
| 606 | + , last_offset = LastOffset |
| 607 | + } = Consumer, |
| 608 | + case brod_utils:is_pid_alive(Pid) of |
| 609 | + true -> |
| 610 | + Consumer; |
| 611 | + false when AckedOffset =/= LastOffset andalso LastOffset =/= ?undef -> |
| 612 | + %% The last fetched offset is not yet acked, |
| 613 | + %% do not re-subscribe now to keep it simple and slow. |
| 614 | + %% Otherwise if we subscribe with {begin_offset, LastOffset + 1} |
| 615 | + %% we may exceed pre-fetch window size. |
| 616 | + Consumer; |
| 617 | + false -> |
| 618 | + %% fetch from the last acked offset + 1 |
| 619 | + %% otherwise fetch from the assigned begin_offset |
| 620 | + BeginOffset = case AckedOffset of |
| 621 | + ?undef -> BeginOffset0; |
| 622 | + N when N >= 0 -> N + 1 |
| 623 | + end, |
| 624 | + Options = |
| 625 | + case BeginOffset =:= ?undef of |
| 626 | + true -> []; %% fetch from 'begin_offset' in consumer config |
| 627 | + false -> [{begin_offset, BeginOffset}] |
| 628 | + end, |
| 629 | + case brod:subscribe(Client, self(), Topic, Partition, Options) of |
| 630 | + {ok, ConsumerPid} -> |
| 631 | + Mref = erlang:monitor(process, ConsumerPid), |
| 632 | + Consumer#consumer{ consumer_pid = ConsumerPid |
| 633 | + , consumer_mref = Mref |
| 634 | + }; |
| 635 | + {error, Reason} -> |
| 636 | + Consumer#consumer{ consumer_pid = ?DOWN(Reason) |
| 637 | + , consumer_mref = ?undef |
| 638 | + } |
| 639 | + end |
| 640 | + end. |
| 641 | + |
| 642 | + log(#state{ groupId = GroupId |
| 643 | + , memberId = MemberId |
| 644 | + , generationId = GenerationId |
| 645 | + }, Level, Fmt, Args) -> |
| 646 | + brod_utils:log( |
| 647 | + Level, |
| 648 | + "group subscriber (groupId=~s,memberId=~s,generation=~p,pid=~p):\n" ++ Fmt, |
| 649 | + [GroupId, MemberId, GenerationId, self() | Args]). |
| 650 | + |
| 651 | + get_consumer(Pid, Consumers) when is_pid(Pid) -> |
| 652 | + lists:keyfind(Pid, #consumer.consumer_pid, Consumers); |
| 653 | + get_consumer({_, _} = TP, Consumers) -> |
| 654 | + lists:keyfind(TP, #consumer.topic_partition, Consumers). |
| 655 | + |
| 656 | + put_consumer(#consumer{topic_partition = TP} = C, Consumers) -> |
| 657 | + lists:keyreplace(TP, #consumer.topic_partition, Consumers, C). |
| 658 | + |
| 659 | + %%%_* Emacs ==================================================================== |
| 660 | + %%% Local Variables: |
| 661 | + %%% allout-layout: t |
| 662 | + %%% erlang-indent-level: 2 |
| 663 | + %%% End: |
Loading more files…