Packages

brod

3.9.1
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

HexDocs

Current section

113 Versions

Jump to
Readme
113 Versions
2 Dependencies
Dependants
Files
Activity
Hide whitespace

Compare versions

3 files changed
+39 additions
-26 deletions
  @@ -1,5 +1,5 @@
1 1 {<<"name">>,<<"brod">>}.
2 - {<<"version">>,<<"3.9.0">>}.
2 + {<<"version">>,<<"3.9.1">>}.
3 3 {<<"requirements">>,
4 4 #{<<"kafka_protocol">> =>
5 5 #{<<"app">> => <<"kafka_protocol">>,<<"optional">> => false,
  @@ -1,6 +1,6 @@
1 1 {application,brod,
2 2 [{description,"Apache Kafka Erlang client library"},
3 - {vsn,"3.9.0"},
3 + {vsn,"3.9.1"},
4 4 {registered,[]},
5 5 {applications,[kernel,stdlib,kafka_protocol,supervisor3]},
6 6 {env,[]},
  @@ -287,14 +287,7 @@ init({Client, GroupId, Topics, Config, CbModule, MemberPid}) ->
287 287 {ok, State}.
288 288
289 289 handle_info({ack, GenerationId, Topic, Partition, Offset}, State) ->
290 - case GenerationId < State#state.generationId of
291 - true ->
292 - %% Ignore stale acks
293 - {noreply, State};
294 - false ->
295 - {ok, NewState} = handle_ack(State, Topic, Partition, Offset),
296 - {noreply, NewState}
297 - end;
290 + {noreply, handle_ack(State, GenerationId, Topic, Partition, Offset)};
298 291 handle_info(?LO_CMD_COMMIT_OFFSETS, #state{is_in_group = true} = State) ->
299 292 {ok, NewState} =
300 293 try
  @@ -434,29 +427,36 @@ stabilize(#state{ rejoin_delay_seconds = RejoinDelaySeconds
434 427 %% 1. unsubscribe all currently assigned partitions
435 428 ok = MemberModule:assignments_revoked(MemberPid),
436 429
437 - %% 2. try to commit current offsets before re-joinning the group.
430 + %% 2. some brod_group_member implementations may wait for messages
431 + %% to finish processing when assignments_revoked is called.
432 + %% The acknowledments of those messages would then be sitting
433 + %% in our inbox. So we do an explicit pass to collect all pending
434 + %% acks so they are included in the best-effort commit below.
435 + State1 = receive_pending_acks(State0),
436 +
437 + %% 3. try to commit current offsets before re-joinning the group.
438 438 %% try only on the first re-join attempt
439 439 %% do not try if it was illegal generation exception received
440 440 %% because it will fail on the same exception again
441 - State1 =
441 + State2 =
442 442 case AttemptNo =:= 0 andalso
443 443 Reason =/= ?illegal_generation of
444 444 true ->
445 - {ok, #state{} = State1_} = try_commit_offsets(State0),
446 - State1_;
445 + {ok, #state{} = State2_} = try_commit_offsets(State1),
446 + State2_;
447 447 false ->
448 - State0
448 + State1
449 449 end,
450 - State2 = State1#state{is_in_group = false},
450 + State3 = State2#state{is_in_group = false},
451 451
452 - %$ 3. Clean up state based on the last failure reason
453 - State = maybe_reset_member_id(State2, Reason),
452 + %$ 4. Clean up state based on the last failure reason
453 + State = maybe_reset_member_id(State3, Reason),
454 454
455 - %% 4. ensure we have a connection to the (maybe new) group coordinator
455 + %% 5. ensure we have a connection to the (maybe new) group coordinator
456 456 F1 = fun discover_coordinator/1,
457 - %% 5. join group
457 + %% 6. join group
458 458 F2 = fun join_group/1,
459 - %% 6. sync assignemnts
459 + %% 7. sync assignemnts
460 460 F3 = fun sync_group/1,
461 461
462 462 RetryFun =
  @@ -474,6 +474,16 @@ stabilize(#state{ rejoin_delay_seconds = RejoinDelaySeconds
474 474 end,
475 475 do_stabilize([F1, F2, F3], RetryFun, State).
476 476
477 + -spec receive_pending_acks(state()) -> state().
478 + receive_pending_acks(State) ->
479 + receive
480 + {ack, GenerationId, Topic, Partition, Offset} ->
481 + NewState = handle_ack(State, GenerationId, Topic, Partition, Offset),
482 + receive_pending_acks(NewState)
483 + after
484 + 0 -> State
485 + end.
486 +
477 487 do_stabilize([], _RetryFun, State) ->
478 488 {ok, State};
479 489 do_stabilize([F | Rest], RetryFun, State) ->
  @@ -579,13 +589,16 @@ sync_group(#state{ groupId = GroupId
579 589 [format_assignments(TopicAssignments)]),
580 590 start_offset_commit_timer(NewState).
581 591
582 - -spec handle_ack(state(), brod:topic(), brod:partition(), brod:offset()) ->
583 - {ok, state()}.
584 - handle_ack(#state{ acked_offsets = AckedOffsets
585 - } = State, Topic, Partition, Offset) ->
592 + -spec handle_ack(state(), brod:group_generation_id(), brod:topic(),
593 + brod:partition(), brod:offset()) -> state().
594 + handle_ack(State, GenerationId, _Topic, _Partition, _Offset)
595 + when GenerationId < State#state.generationId ->
596 + State;
597 + handle_ack(#state{acked_offsets = AckedOffsets} = State,
598 + _GenerationId, Topic, Partition, Offset) ->
586 599 NewAckedOffsets =
587 600 merge_acked_offsets(AckedOffsets, [{{Topic, Partition}, Offset}]),
588 - {ok, State#state{acked_offsets = NewAckedOffsets}}.
601 + State#state{acked_offsets = NewAckedOffsets}.
589 602
590 603 %% Add new offsets to be acked into the acked offsets collection.
591 604 -spec merge_acked_offsets(Offsets, Offsets) -> Offsets when