%%% %%% Copyright (c) 2016 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 %%% Kafka consumer group membership controller %%% %%% @copyright 2016 Klarna AB %%% @end %%%============================================================================= -module(brod_group_controller). -behaviour(gen_server). -export([ ack/5 , commit_offsets/1 , start_link/4 ]). -export([ code_change/3 , handle_call/3 , handle_cast/2 , handle_info/2 , init/1 , terminate/2 ]). -include("brod_int.hrl"). -define(PARTITION_ASSIGMENT_STRATEGY_ROUNDROBIN, roundrobin). %% default -type partition_assignment_strategy() :: atom(). %% default configs -define(SESSION_TIMEOUT_SECONDS, 10). -define(HEARTBEAT_RATE_SECONDS, 2). -define(PROTOCOL_TYPE, <<"consumer">>). -define(MAX_REJOIN_ATTEMPTS, 5). -define(REJOIN_DELAY_SECONDS, 1). -define(OFFSET_COMMIT_POLICY, commit_to_kafka_v2). -define(OFFSET_COMMIT_INTERVAL_SECONDS, 5). %% use kfaka's offset meta-topic retention policy -define(OFFSET_RETENTION_DEFAULT, -1). -define(ESCALATE_EC(EC), kpro_ErrorCode:is_error(EC) andalso erlang:throw(EC)). -define(ESCALATE(Expr), fun() -> case Expr of {ok, Result} -> Result; {error, Reason} -> throw(Reason) end end()). %% loopback commands -define(LO_CMD_SEND_HB, lo_cmd_send_heartbeat). -define(LO_CMD_COMMIT_OFFSETS, lo_cmd_commit_offsets). -define(LO_CMD_STABILIZE(AttemptCount, Reason), {lo_cmd_stabilize, AttemptCount, Reason}). -type config() :: group_config(). -type ts() :: erlang:timestamp(). -type member() :: kpro_GroupMemberMetadata(). -type offset_commit_policy() :: brod_offset_commit_policy(). -record(state, { client :: client() , groupId :: group_id() %% Group member ID, which should be set to empty in the first %% join group request, then a new member id is assigned by the %% group coordinator and in join group response. %% This field may change if the member has lost connection %% to the coordinator and received 'UnknownMemberId' exception %% in response messages. , memberId = <<"">> :: member_id() %% State#state.memberId =:= State#state.leaderId if %% elected as group leader by the coordinator. , leaderId :: member_id() %% Generation ID is used by the group coordinator to sync state %% of the group members, e.g. kick out stale members who have not %% kept up with the latest generation ID bumps. , generationId = 0 :: integer() %% A set of topic names where the group members consumes from , topics = [] :: [topic()] %% This is the result of group coordinator discovery. %% It may change when the coordinator is down then a new one %% is elected among the kafka cluster members. , coordinator :: endpoint() %% The socket pid to the group coordinator broker. %% This socket is dedicated for group management and %% offset commit requests. %% We can not just get a payload socket from client %% because the socket might be shared with other group %% members in the same client, however group members are %% distinguished by connections to coordinator , sock_pid :: pid() %% heartbeat reference, to discard stale responses , hb_ref :: {corr_id(), ts()} %% all group members received in the join group response %% this field is currently not used, but the binary encoded %% kpro_GroupMemberMetadata.protocolMetadata.userData field %% can be useful for 'sticky' assignments etc. in the future , members = [] :: [member()] %% Set to false before joining the group %% then set to true when sucessfully joined the group. %% This is by far only used to prevent the timer-triggered %% loopback command message sending a HeartbeatRequest to %% the group coordinator broker. , is_in_group = false :: boolean() %% The message-set subscriber which subscribes to all %% assigned topic-partitions. , subscriber :: pid() %% The offsets that has been acknowledged by the subscriber %% i.e. the offsets that are ready for commit. %% NOTE: this field is not used if offset_commit_policy is %% 'consumer_managed' , acked_offsets = [] :: [{{topic(), partition()}, offset()}] %% configs, see start_link/4 doc for details , partition_assignment_strategy :: partition_assignment_strategy() , session_timeout_seconds :: pos_integer() , heartbeat_rate_seconds :: pos_integer() , max_rejoin_attempts :: non_neg_integer() , rejoin_delay_seconds :: non_neg_integer() , offset_retention_seconds :: ?undef | integer() , offset_commit_policy :: offset_commit_policy() , offset_commit_interval_seconds :: pos_integer() }). -define(IS_LEADER(S), (S#state.leaderId =:= S#state.memberId)). %%%_* APIs ===================================================================== %% @doc To be called by group subscriber. %% Client: ClientId (or pid, but not recommended) %% GroupId: Predefined globally unique (in a kafka cluster) binary string. %% Topics: Predefined set of topic names in the group. %% Config: The group controller configs in a proplist, possible entries: %% - partition_assignment_strategy (optional, default = roundrobin) %% roundrobin: Take all topic-offset (sorted [{TopicName, Partition}] list) %% assign one to each member in a roundrobin fashion. %% TODO: support sticky assignments %% - session_timeout_seconds (optional, default = 10) %% Time in seconds for the group coordinator broker to consider a member %% 'down' if no heartbeat or any kind of requests received from a broker %% in the past N seconds. %% A group member may also consider the coordinator broker 'down' if no %% heartbeat response response received in the past N seconds. %% - heartbeat_rate_seconds (optional, default = 2) %% Time in seconds for the member to 'ping' the group coordinator. %% OBS: Care should be taken when picking the number, on one hand, we do %% not want to flush the broker with requests if we set it too low, %% on the other hand, if set it too high, it may take too long for %% the members to realise status changes of the group such as %% assignment rebalacing or group coordinator switchover etc. %% - max_rejoin_attempts (optional, default = 5) %% Maximum number of times allowd for a member to re-join the group. %% The gen_server will stop if it reached the maximum number of retries. %% OBS: 'let it crash' may not be the optimal strategy here because %% the group member id is kept in the gen_server looping state and %% it is reused when re-joining the group. %% - rejoin_delay_seconds (optional, default = 1) %% Delay in seconds before re-joining the group. %% - offset_commit_policy (optional, default = commit_to_kafka_v2) %% How/where to commit offsets, possible values: %% - commit_to_kafka_v2: %% Group controller will commit the offsets to kafka using %% version 2 OffsetCommitRequest. %% - consumer_managed: %% The subscirber (brod_group_subscriber.erl) is responsible %% for persisting offsets to a local or centralized storage. %% And the callback get_committed_offsets should be implemented %% to allow group controller to retrieve the commited offsets. %% - offset_commit_interval_seconds (optional, default = 5) %% The time interval between two OffsetCommitRequest messages. %% This config is irrelevant if offset_commit_policy is consumer_managed. %% - offset_retention_seconds (optional, default = -1) %% How long the time is to be kept in kafka before it is deleted. %% The default special value -1 indicates that the __consumer_offsets %% topic retention policy is used. %% This config is irrelevant if offset_commit_policy is consumer_managed. %% @end -spec start_link(client(), group_id(), [topic()], config()) -> {ok, pid()} | {error, any()}. start_link(Client, GroupId, Topics, Config) -> Subscriber = self(), Args = {Client, GroupId, Topics, Config, Subscriber}, gen_server:start_link(?MODULE, Args, []). %% @doc For group subscriber to call to acknowledge. -spec ack(pid(), integer(), topic(), partition(), offset()) -> ok. ack(Pid, GenerationId, Topic, Partition, Offset) -> Pid ! {ack, GenerationId, Topic, Partition, Offset}, ok. %% @doc Force commit offsets immediately. -spec commit_offsets(pid()) -> ok | {error, any()}. commit_offsets(ControllerPid) -> gen_server:call(ControllerPid, commit_offsets, infinity). %%%_* gen_server callbacks ===================================================== init({Client, GroupId, Topics, Config, Subscriber}) -> process_flag(trap_exit, true), GetCfg = fun(Name, Default) -> proplists:get_value(Name, Config, Default) end, PaStrategy = GetCfg(partition_assignment_strategy, ?PARTITION_ASSIGMENT_STRATEGY_ROUNDROBIN), SessionTimeoutSec = GetCfg(session_timeout_seconds, ?SESSION_TIMEOUT_SECONDS), HbRateSec = GetCfg(heartbeat_rate_seconds, ?HEARTBEAT_RATE_SECONDS), MaxRejoinAttempts = GetCfg(max_rejoin_attempts, ?MAX_REJOIN_ATTEMPTS), RejoinDelaySeconds = GetCfg(rejoin_delay_seconds, ?REJOIN_DELAY_SECONDS), OffsetRetentionSeconds = GetCfg(offset_retention_seconds, ?undef), OffsetCommitPolicy = GetCfg(offset_commit_policy, ?OFFSET_COMMIT_POLICY), OffsetCommitIntervalSeconds = GetCfg(offset_commit_interval_seconds, ?OFFSET_COMMIT_INTERVAL_SECONDS), self() ! ?LO_CMD_STABILIZE(0, ?undef), ok = start_heartbeat_timer(HbRateSec), State = #state{ client = Client , groupId = GroupId , topics = Topics , subscriber = Subscriber , partition_assignment_strategy = PaStrategy , session_timeout_seconds = SessionTimeoutSec , heartbeat_rate_seconds = HbRateSec , max_rejoin_attempts = MaxRejoinAttempts , rejoin_delay_seconds = RejoinDelaySeconds , offset_retention_seconds = OffsetRetentionSeconds , offset_commit_policy = OffsetCommitPolicy , offset_commit_interval_seconds = OffsetCommitIntervalSeconds }, ok = maybe_start_offset_commit_timer(State), {ok, State}. handle_info({ack, GenerationId, Topic, Partition, Offset}, State) -> case GenerationId < State#state.generationId of true -> %% Ignore stale acks {noreply, State}; false -> {ok, NewState} = handle_ack(State, Topic, Partition, Offset), {noreply, NewState} end; handle_info(?LO_CMD_COMMIT_OFFSETS, State) -> try {ok, NewState} = do_commit_offsets(State), ok = maybe_start_offset_commit_timer(NewState), {noreply, NewState} catch throw : Reason -> {stop, {failed_to_commit_offsets, Reason}, State} end; handle_info(?LO_CMD_STABILIZE(N, _Reason), #state{max_rejoin_attempts = Max} = State) when N >= Max -> {stop, max_rejoin_attempts, State}; handle_info(?LO_CMD_STABILIZE(N, Reason), State) -> {ok, NewState} = stabilize(State, N, Reason), {noreply, NewState}; handle_info({'EXIT', Pid, Reason}, #state{sock_pid = Pid} = State) -> {ok, NewState} = stabilize(State, 0, {sockent_down, Reason}), {noreply, NewState}; handle_info({'EXIT', Pid, Reason}, #state{subscriber = Pid} = State) -> case Reason of shutdown -> {stop, shutdown, State}; {shutdown, _} -> {stop, shutdown, State}; normal -> {stop, normal, State}; _ -> {stop, subscriber_down, State} end; handle_info(?LO_CMD_SEND_HB, #state{ hb_ref = HbRef , session_timeout_seconds = SessionTimeoutSec } = State) -> _ = start_heartbeat_timer(State#state.heartbeat_rate_seconds), case HbRef of ?undef -> {ok, NewState} = maybe_send_heartbeat(State), {noreply, NewState}; {_HbCorrId, SentTime} -> Elapsed = timer:now_diff(os:timestamp(), SentTime), case Elapsed < SessionTimeoutSec * 1000000 of true -> %% keep waiting for heartbeat response {noreply, State}; false -> %% time to re-discover a new coordinator ? {ok, NewState} = stabilize(State, 0, hb_timeout), {noreply, NewState} end end; handle_info({msg, _Pid, HbCorrId, #kpro_HeartbeatResponse{errorCode = EC}}, #state{hb_ref = {HbCorrId, _SentTime}} = State0) -> State = State0#state{hb_ref = ?undef}, case kpro_ErrorCode:is_error(EC) of true -> {ok, NewState} = stabilize(State, 0, EC), {noreply, NewState}; false -> {noreply, State} end; handle_info(_Info, State) -> {noreply, State}. handle_call(commit_offsets, _From, #state{offset_commit_policy = consumer_managed} = State) -> %% the subscriber is responsible for commiting offsets in handle_message {reply, {error, consumer_managed}, State}; handle_call(commit_offsets, From, State) -> try {ok, NewState} = do_commit_offsets(State), {reply, ok, NewState} catch throw : Reason -> gen_server:reply(From, {error, Reason}), {stop, {failed_to_commit_offsets, Reason}, State} end; handle_call(Call, _From, State) -> {reply, {error, {unknown_call, Call}}, State}. handle_cast(_Cast, State) -> {noreply, State}. code_change(_OldVsn, State, _Extra) -> {ok, State}. terminate(Reason, #state{ sock_pid = SockPid , groupId = GroupId , memberId = MemberId } = State) -> log(State, info, "leaving group, reason ~p\n", [Reason]), Request = #kpro_LeaveGroupRequest { groupId = GroupId , memberId = MemberId }, try send_sync(SockPid, Request, 1000) catch _ : _ -> ok end, ok = stop_socket(SockPid). %%%_* Internal Functions ======================================================= -spec discover_coordinator(#state{}) -> {ok, #state{}}. discover_coordinator(#state{ client = Client , coordinator = Coordinator , sock_pid = SockPid , groupId = GroupId } = State) -> {Host, Port} = ?ESCALATE(brod_client:get_group_coordinator(Client, GroupId)), HasConnectionToCoordinator = case Coordinator =:= {Host, Port} of true -> is_pid(SockPid) andalso is_process_alive(SockPid); false -> false end, case HasConnectionToCoordinator of true -> {ok, State}; false -> %% close old socket _ = brod_sock:stop(SockPid), ClientId = make_group_connection_client_id(), NewSockPid = ?ESCALATE(brod_sock:start_link(self(), Host, Port, ClientId, [])), log(State, info, "connected to group coordinator ~s:~p", [Host, Port]), NewState = State#state{ coordinator = {Host, Port} , sock_pid = NewSockPid }, {ok, NewState} end. -spec stabilize(#state{}, integer(), any()) -> {ok, #state{}}. stabilize(#state{ rejoin_delay_seconds = RejoinDelaySeconds , subscriber = Subscriber , offset_commit_policy = CommitPolicy } = State0, AttemptNo, Reason) -> Reason =/= ?undef andalso log(State0, info, "re-joining group, reason:~p", [Reason]), %% 1. unsubscribe all currently assigned partitions ok = brod_group_subscriber:unsubscribe_all_partitions(Subscriber), %% 2. if it is illegal generation error code received, try to commit current %% current offsets before re-joinning the group. State1 = case AttemptNo =:= 0 andalso Reason =:= ?EC_ILLEGAL_GENERATION andalso CommitPolicy =/= consumer_managed of true -> {ok, #state{} = State1_} = try_commit_offsets(State0), State1_; false -> State0 end, State2 = State1#state{is_in_group = false}, %$ 3. Clean up state based on the last failure reason State3 = maybe_reset_member_id(State2, Reason), State = maybe_reset_socket(State3, Reason), %% 4. ensure we have a connection to the (maybe new) group coordinator F1 = fun discover_coordinator/1, %% 5. join group F2 = fun join_group/1, %% 6. sync assignemnts F3 = fun sync_group/1, RetryFun = fun(StateIn, NewReason) -> log(StateIn, info, "failed to join group\nreason:~p", [NewReason]), _ = case AttemptNo =:= 0 of true -> %% do not delay before the first retry self() ! ?LO_CMD_STABILIZE(AttemptNo + 1, NewReason); false -> erlang:send_after(timer:seconds(RejoinDelaySeconds), self(), ?LO_CMD_STABILIZE(AttemptNo + 1, NewReason)) end, {ok, StateIn} end, do_stabilize([F1, F2, F3], RetryFun, State). do_stabilize([], _RetryFun, State) -> {ok, State}; do_stabilize([F | Rest], RetryFun, State) -> try {ok, #state{} = NewState} = F(State), do_stabilize(Rest, RetryFun, NewState) catch throw : Reason -> RetryFun(State, Reason) end. maybe_reset_member_id(State, Reason) -> case should_reset_member_id(Reason) of true -> State#state{memberId = <<>>}; false -> State end. should_reset_member_id(?EC_UNKNOWN_MEMBER_ID) -> %% we are likely kicked out from the group %% rejoin with empty member id true; should_reset_member_id(?EC_NOT_COORDINATOR_FOR_GROUP) -> %% the coordinator have moved to another broker %% set it to ?undef to trigger a socket restart true; should_reset_member_id({socket_down, _Reason}) -> %% old socket was down, new connection will lead %% to a new member id true; should_reset_member_id(_) -> false. maybe_reset_socket(State, ?EC_NOT_COORDINATOR_FOR_GROUP) -> ok = stop_socket(State#state.sock_pid), State#state{sock_pid = ?undef}; maybe_reset_socket(State, _OtherReason) -> State. stop_socket(SockPid) -> catch unlink(SockPid), ok = brod_sock:stop(SockPid). -spec join_group(#state{}) -> {ok, #state{}}. join_group(#state{ groupId = GroupId , memberId = MemberId0 , topics = Topics , sock_pid = SockPid , partition_assignment_strategy = PaStrategy , session_timeout_seconds = SessionTimeoutSec } = State0) -> ConsumerGroupProtocolMeta = #kpro_ConsumerGroupProtocolMetadata { version = ?BROD_CONSUMER_GROUP_PROTOCOL_VERSION , topicName_L = Topics , userData = make_user_data() }, ConsumerGroupProtocol = #kpro_GroupProtocol { protocolName = atom_to_list(PaStrategy) , protocolMetadata = ConsumerGroupProtocolMeta }, SessionTimeout = timer:seconds(SessionTimeoutSec), JoinReq = #kpro_JoinGroupRequest { groupId = GroupId , sessionTimeout = SessionTimeout , memberId = MemberId0 , protocolType = ?PROTOCOL_TYPE , groupProtocol_L = [ConsumerGroupProtocol] }, %% send join group request and wait for response %% as long as the session timeout config JoinRsp = send_sync(SockPid, JoinReq, SessionTimeout), ?ESCALATE_EC(JoinRsp#kpro_JoinGroupResponse.errorCode), #kpro_JoinGroupResponse { generationId = GenerationId , protocolName = _PaStrategyBinStr , leaderId = LeaderId , memberId = MemberId , groupMemberMetadata_L = Members } = JoinRsp, IsGroupLeader = (LeaderId =:= MemberId), State = State0#state{ memberId = MemberId , leaderId = LeaderId , generationId = GenerationId , members = Members }, log(State, info, "elected=~p", [IsGroupLeader]), {ok, State}. -spec sync_group(#state{}) -> {ok, #state{}}. sync_group(#state{ groupId = GroupId , generationId = GenerationId , memberId = MemberId , sock_pid = SockPid , subscriber = Subscriber } = State) -> SyncReq = #kpro_SyncGroupRequest { groupId = GroupId , generationId = GenerationId , memberId = MemberId , groupAssignment_L = assign_partitions(State) }, %% send sync group request and wait for response #kpro_SyncGroupResponse { errorCode = SyncErrorCode , memberAssignment = Assignment } = send_sync(SockPid, SyncReq), ?ESCALATE_EC(SyncErrorCode), %% get my partition assignments TopicAssignments = get_topic_assignments(State, Assignment), ok = brod_group_subscriber:new_assignments(Subscriber, MemberId, GenerationId, TopicAssignments), NewState = State#state{is_in_group = true}, log(NewState, info, "assignments received:~s", [format_assignments(TopicAssignments)]), {ok, NewState}. -spec handle_ack(#state{}, topic(), partition(), offset()) -> {ok, #state{}}. handle_ack(#state{ acked_offsets = AckedOffsets } = State, Topic, Partition, Offset) -> NewAckedOffsets = lists:keystore({Topic, Partition}, 1, AckedOffsets, {{Topic, Partition}, Offset}), {ok, State#state{acked_offsets = NewAckedOffsets}}. -spec format_assignments([topic_assignment()]) -> iodata(). format_assignments([]) -> ""; format_assignments([{Topic, Partitions} | Rest]) -> ["\n", Topic, ":", format_partition_assignments(Partitions), format_assignments(Rest)]. format_partition_assignments([]) -> ""; format_partition_assignments([PA | Rest]) -> #partition_assignment{ partition = Partition , begin_offset = BeginOffset , metadata = Metadata } = PA, [ io_lib:format("~n partition=~p begin_offset=~p metadata='~s'", [Partition, BeginOffset, Metadata]) , format_partition_assignments(Rest)]. %% @private Commit the current offsets before re-join the group. %% NOTE: this is a 'best-effort' attempt, failing to commit offset %% at this stage should be fine, after all, the consumers will %% refresh their start point offsets when new assignment is %% received. %% @end -spec try_commit_offsets(#state{}) -> {ok, #state{}}. try_commit_offsets(#state{} = State) -> try {ok, #state{}} = do_commit_offsets(State) catch _ : _ -> {ok, State} end. -spec do_commit_offsets(#state{}) -> {ok, #state{}}. do_commit_offsets(#state{acked_offsets = []} = State) -> {ok, State}; do_commit_offsets(#state{ groupId = GroupId , memberId = MemberId , generationId = GenerationId , sock_pid = SockPid , offset_retention_seconds = OffsetRetentionSecs , acked_offsets = AckedOffsets } = State) -> Metadata = make_offset_commit_metadata(), TopicOffsets = lists:foldl( fun({{Topic, Partition}, Offset}, Acc) -> PartitionOffset = #kpro_OCReqV2Partition{ partition = Partition , offset = Offset , metadata = Metadata }, orddict:append_list(Topic, [PartitionOffset], Acc) end, [], AckedOffsets), Offsets = lists:map( fun({Topic, PartitionOffsets}) -> #kpro_OCReqV2Topic{ topicName = Topic , oCReqV2Partition_L = PartitionOffsets } end, TopicOffsets), Req = #kpro_OffsetCommitRequestV2 { consumerGroupId = GroupId , consumerGroupGenerationId = GenerationId , consumerId = MemberId , retentionTime = case OffsetRetentionSecs =/= ?undef of true -> timer:seconds(OffsetRetentionSecs); false -> ?OFFSET_RETENTION_DEFAULT end , oCReqV2Topic_L = Offsets }, Rsp = send_sync(SockPid, Req), #kpro_OffsetCommitResponse{oCRspTopic_L = Topics} = Rsp, lists:foreach( fun(#kpro_OCRspTopic{topicName = Topic, oCRspPartition_L = Partitions}) -> lists:foreach( fun(#kpro_OCRspPartition{partition = Partition, errorCode = EC}) -> kpro_ErrorCode:is_error(EC) andalso begin log(State, error, "failed to commit offset for topic=~s, partition=~p\n" "~p:~s", [Topic, Partition, EC, kpro_ErrorCode:desc(EC)]), erlang:error(EC) end end, Partitions) end, Topics), {ok, State#state{acked_offsets = []}}. -spec assign_partitions(#state{}) -> [kpro_GroupAssignment()]. assign_partitions(State) when ?IS_LEADER(State) -> #state{ client = Client , topics = Topics , members = Members , partition_assignment_strategy = Strategy } = State, AllPartitions = [ {Topic, Partition} || Topic <- lists:usort(Topics), Partition <- get_partitions(Client, Topic) ], Assignments = do_assign_partitions(Strategy, Members, AllPartitions), lists:map( fun({MemberId, Topics_}) -> PartitionAssignments = lists:map(fun({Topic, Partitions}) -> #kpro_ConsumerGroupPartitionAssignment { topicName = Topic , partition_L = Partitions } end, Topics_), #kpro_GroupAssignment { memberId = MemberId , memberAssignment = #kpro_ConsumerGroupMemberAssignment { version = ?BROD_CONSUMER_GROUP_PROTOCOL_VERSION , consumerGroupPartitionAssignment_L = PartitionAssignments , userData = <<0>> %% null is not allowed before 0.9.0.1 } } end, Assignments); assign_partitions(#state{}) -> %% only leader can assign partitions to members []. -spec get_partitions(client(), topic()) -> [partition()]. get_partitions(Client, Topic) -> Count = ?ESCALATE(brod_client:get_partitions_count(Client, Topic)), lists:seq(0, Count-1). -spec do_assign_partitions(partition_assignment_strategy(), [kpro_GroupMemberMetadata()], [{topic(), partition()}]) -> [member_assignment()]. do_assign_partitions(roundrobin, Members, AllPartitions) -> %% round robin, we only care about the member id F = fun(#kpro_GroupMemberMetadata{memberId = MemberId}) -> {MemberId, []} end, MemberAssignment = lists:map(F, Members), roundrobin_assign_loop(AllPartitions, MemberAssignment, []). -spec roundrobin_assign_loop([{topic(), partition()}], [member_assignment()], [member_assignment()]) -> [member_assignment()]. roundrobin_assign_loop([], PendingMembers, AssignedMembers) -> lists:reverse(AssignedMembers) ++ PendingMembers; roundrobin_assign_loop(Partitions, [], AssignedMembers) -> %% all members have received assignments, continue the next round roundrobin_assign_loop(Partitions, lists:reverse(AssignedMembers), []); roundrobin_assign_loop([{Topic, Partition} | Rest], [Member0 | PendingMembers], AssignedMembers) -> Member = assign_partition(Member0, Topic, Partition), roundrobin_assign_loop(Rest, PendingMembers, [Member | AssignedMembers]). -spec assign_partition(member_assignment(), topic(), partition()) -> member_assignment(). assign_partition({MemberId, Topics0}, Topic, Partition) -> Topics = orddict:append_list(Topic, [Partition], Topics0), {MemberId, Topics}. %% @private Extract the partition assignemts from SyncGroupResponse %% then fetch the committed offsets of each partition. %% @end -spec get_topic_assignments(#state{}, kpro_ConsumerGroupMemberAssignment()) -> [topic_assignment()]. get_topic_assignments(#state{}, <<>>) -> []; get_topic_assignments(#state{} = State, Assignment) -> #kpro_ConsumerGroupMemberAssignment { version = _VersionIgnored , consumerGroupPartitionAssignment_L = PartitionAssignments } = Assignment, TopicPartitions0 = lists:map( fun(#kpro_ConsumerGroupPartitionAssignment{ topicName = Topic , partition_L = Partitions }) -> [{Topic, Partition} || Partition <- Partitions] end, PartitionAssignments), TopicPartitions = lists:append(TopicPartitions0), CommittedOffsets = get_committed_offsets(State, TopicPartitions), resolve_begin_offsets(TopicPartitions, CommittedOffsets, orddict:from_list([])). %% @private Fetch committed offsets from kafka, %% or call the consumer callback to read committed offsets. %% @end -spec get_committed_offsets(#state{}, [{topic(), partition()}]) -> [{{topic(), partition()}, OffsetOrWithMetadata}] when OffsetOrWithMetadata :: offset() | {offset(), binary()}. get_committed_offsets(#state{ offset_commit_policy = consumer_managed , subscriber = Subscriber }, TopicPartitions) -> brod_group_subscriber:get_committed_offsets(Subscriber, TopicPartitions); get_committed_offsets(#state{ offset_commit_policy = commit_to_kafka_v2 , groupId = GroupId , sock_pid = SockPid }, TopicPartitions) -> GrouppedPartitions = lists:foldl(fun({T, P}, Dict) -> orddict:append_list(T, [P], Dict) end, [], TopicPartitions), OffsetFetchRequestTopics = lists:map( fun({Topic, Partitions}) -> #kpro_OFReqTopic{ topicName = Topic , partition_L = Partitions } end, GrouppedPartitions), OffsetFetchRequest = #kpro_OffsetFetchRequest { consumerGroup = GroupId , oFReqTopic_L = OffsetFetchRequestTopics }, Rsp = send_sync(SockPid, OffsetFetchRequest), #kpro_OffsetFetchResponse{topicOffset_L = TopicOffsets} = Rsp, CommittedOffsets0 = lists:map( fun(#kpro_TopicOffset{ topicName = Topic , partitionOffset_L = Partitions }) -> lists:foldl( fun(#kpro_PartitionOffset{ partition = Partition , offset = Offset , metadata = Metadata , errorCode = EC }, Acc) -> case EC =:= ?EC_UNKNOWN_TOPIC_OR_PARTITION of true -> %% OffsetFetchResponse v0 if no commit history found Acc; false -> case EC =:= ?EC_NONE andalso Offset =:= -1 andalso Metadata =:= <<>> of true -> %% OffsetFetchResponse v1 if no commit history found Acc; false -> ?ESCALATE_EC(EC), [{{Topic, Partition}, {Offset, Metadata}} | Acc] end end end, [], Partitions) end, TopicOffsets), lists:append(CommittedOffsets0). -spec resolve_begin_offsets( TopicPartitions :: [{topic(), partition()}], CommittedOffsets :: [{{topic(), partition()}, OffsetOrWithMetadata}], [topic_assignment()]) -> [topic_assignment()] when OffsetOrWithMetadata :: offset() | {offset(), binary()}. resolve_begin_offsets([], _, Acc) -> Acc; resolve_begin_offsets([{Topic, Partition} | Rest], CommittedOffsets, Acc) -> {Offset, Metadata} = case lists:keyfind({Topic, Partition}, 1, CommittedOffsets) of {_, Tuple} when is_tuple(Tuple) -> %% Commit history found in kafka Tuple; {_, Offset_} when is_integer(Offset_) -> %% Commit history found from subscriber callback {Offset_, <<>>}; false -> %% No commit history found {?undef, <<>>} end, BeginOffset = case is_integer(Offset) andalso Offset > 0 of true -> Offset + 1; false -> Offset end, PartitionAssignment = #partition_assignment{ partition = Partition , begin_offset = BeginOffset , metadata = Metadata }, NewAcc = orddict:append_list(Topic, [PartitionAssignment], Acc), resolve_begin_offsets(Rest, CommittedOffsets, NewAcc). %% @private Start a timer to send a loopback command to self() to trigger %% a heartbeat request to the group coordinator. %% NOTE: the heartbeat requests are sent only when it is in group, %% but the timer is always restarted after expiration. %% @end -spec start_heartbeat_timer(pos_integer()) -> ok. start_heartbeat_timer(HbRateSec) -> erlang:send_after(timer:seconds(HbRateSec), self(), ?LO_CMD_SEND_HB), ok. %% @private Start a timer to send a loopback command to self() to trigger %% a offset commit request to group coordinator. %% @end -spec maybe_start_offset_commit_timer(#state{} | offset_commit_policy()) -> ok. maybe_start_offset_commit_timer(#state{} = State) -> #state{ offset_commit_policy = Policy , offset_commit_interval_seconds = Seconds } = State, case Policy of consumer_managed -> ok; commit_to_kafka_v2 -> Timeout = timer:seconds(Seconds), _ = erlang:send_after(Timeout, self(), ?LO_CMD_COMMIT_OFFSETS), ok end. %% @private Send heartbeat request if it has joined the group. -spec maybe_send_heartbeat(#state{}) -> {ok, #state{}}. maybe_send_heartbeat(#state{ is_in_group = true , groupId = GroupId , memberId = MemberId , generationId = GenerationId , sock_pid = SockPid } = State) -> Request = #kpro_HeartbeatRequest{ groupId = GroupId , memberId = MemberId , generationId = GenerationId }, {ok, CorrId} = brod_sock:request_async(SockPid, Request), NewState = State#state{hb_ref = {CorrId, os:timestamp()}}, {ok, NewState}; maybe_send_heartbeat(#state{} = State) -> %% do not send heartbeat when not in group {ok, State#state{hb_ref = ?undef}}. send_sync(SockPid, Request) -> send_sync(SockPid, Request, 5000). send_sync(SockPid, Request, Timeout) -> ?ESCALATE(brod_sock:request_sync(SockPid, Request, Timeout)). log(#state{ groupId = GroupId , memberId = MemberId , generationId = GenerationId }, Level, Fmt, Args) -> brod_utils:log( Level, "group controller (groupId=~s,memberId=~s,generation=~p,pid=~p):\n" ++ Fmt, [GroupId, MemberId, GenerationId, self() | Args]). %% @private Make metata to be committed together with offsets. -spec make_offset_commit_metadata() -> iodata(). make_offset_commit_metadata() -> io_lib:format("~s ~p ~p", [brod_utils:os_time_utc_str(), node(), self()]). %% @private Make group member's user data in JoinGroupRequest -spec make_user_data() -> iodata(). make_user_data() -> controller_id(). %% @private Make a client_id() to be used in the requests sent over the group %% controller's socket (group coordinator on the other end), this id will be %% displayed when describing the group status with admin client/script. %% e.g. brod@localhost/<0.45.0>_/172.18.0.1 %% @end -spec make_group_connection_client_id() -> binary(). make_group_connection_client_id() -> controller_id(). %% @private Use 'node()/pid()' as unique identifier of each group controller. -spec controller_id() -> binary(). controller_id() -> iolist_to_binary(io_lib:format("~p/~p", [node(), self()])). %%%_* Emacs ==================================================================== %%% Local Variables: %%% allout-layout: t %%% erlang-indent-level: 2 %%% End: