Packages
brod
4.5.4
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
+104
additions
-24
deletions
| @@ -16,7 +16,7 @@ | |
| 16 16 | <<"src/brod_topic_subscriber.erl">>, |
| 17 17 | <<"src/brod_topic_subscriber_cb_fun.erl">>,<<"src/brod_transaction.erl">>, |
| 18 18 | <<"src/brod_transaction_processor.erl">>,<<"src/brod_utils.erl">>]}. |
| 19 | - {<<"licenses">>,[<<"Apache License 2.0">>]}. |
| 19 | + {<<"licenses">>,[<<"Apache-2.0">>]}. |
| 20 20 | {<<"links">>,[{<<"Github">>,<<"https://github.com/kafka4beam/brod">>}]}. |
| 21 21 | {<<"name">>,<<"brod">>}. |
| 22 22 | {<<"requirements">>, |
| @@ -24,4 +24,4 @@ | |
| 24 24 | [{<<"app">>,<<"kafka_protocol">>}, |
| 25 25 | {<<"optional">>,false}, |
| 26 26 | {<<"requirement">>,<<"4.3.4">>}]}]}. |
| 27 | - {<<"version">>,<<"4.5.3">>}. |
| 27 | + {<<"version">>,<<"4.5.4">>}. |
| @@ -1,12 +1,12 @@ | |
| 1 1 | {application,brod, |
| 2 2 | [{description,"Apache Kafka Erlang client library"}, |
| 3 | - {vsn,"4.5.3"}, |
| 3 | + {vsn,"4.5.4"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol]}, |
| 6 6 | {env,[]}, |
| 7 7 | {mod,{brod,[]}}, |
| 8 8 | {modules,[]}, |
| 9 | - {licenses,["Apache License 2.0"]}, |
| 9 | + {licenses,["Apache-2.0"]}, |
| 10 10 | {links,[{"Github","https://github.com/kafka4beam/brod"}]}, |
| 11 11 | {build_tools,["rebar3","cmake"]}, |
| 12 12 | {files,["src","include","rebar.config","rebar.config.script", |
| @@ -986,7 +986,7 @@ get_topic_assignments(#state{} = State, Assignment) -> | |
| 986 986 | %% or call the consumer callback to read committed offsets. |
| 987 987 | -spec get_committed_offsets(state(), [{brod:topic(), brod:partition()}]) -> |
| 988 988 | [{{brod:topic(), brod:partition()}, |
| 989 | - brod:offset() | {begin_offset, brod:offset_time()}}]. |
| 989 | + brod:offset() | {begin_offset, brod:offset_time()} | ?undef}]. |
| 990 990 | get_committed_offsets(#state{ offset_commit_policy = consumer_managed |
| 991 991 | , member_pid = MemberPid |
| 992 992 | , member_module = MemberModule |
| @@ -1016,9 +1016,13 @@ get_committed_offsets(#state{ offset_commit_policy = commit_to_kafka_v2 | |
| 1016 1016 | EC = kpro:find(error_code, PartitionOffset), |
| 1017 1017 | ?ESCALATE_EC(EC), |
| 1018 1018 | %% Offset -1 in offset_fetch_response is an indicator of 'no-value' |
| 1019 | + %% (e.g. new group, topic recreated, offsets.retention.minutes |
| 1020 | + %% expired). Surface it as ?undef so the subscriber can apply |
| 1021 | + %% offset_reset_policy rather than silently falling back to the |
| 1022 | + %% consumer config begin_offset. |
| 1019 1023 | case Offset0 =:= -1 of |
| 1020 1024 | true -> |
| 1021 | - Acc; |
| 1025 | + [{{Topic, Partition}, ?undef} | Acc]; |
| 1022 1026 | false -> |
| 1023 1027 | Offset = maybe_upgrade_from_roundrobin_v1(Offset0, Metadata), |
| 1024 1028 | [{{Topic, Partition}, Offset} | Acc] |
| @@ -1029,7 +1033,7 @@ get_committed_offsets(#state{ offset_commit_policy = commit_to_kafka_v2 | |
| 1029 1033 | |
| 1030 1034 | -spec resolve_begin_offsets( |
| 1031 1035 | [TP], |
| 1032 | - [{TP, brod:offset() | {begin_offset, brod:offset_time()}}], |
| 1036 | + [{TP, brod:offset() | {begin_offset, brod:offset_time()} | ?undef}], |
| 1033 1037 | boolean()) -> |
| 1034 1038 | brod:received_assignments() |
| 1035 1039 | when TP :: {brod:topic(), brod:partition()}. |
| @@ -1038,6 +1042,12 @@ resolve_begin_offsets([{Topic, Partition} | Rest], CommittedOffsets, | |
| 1038 1042 | IsConsumerManaged) -> |
| 1039 1043 | BeginOffset = |
| 1040 1044 | case lists:keyfind({Topic, Partition}, 1, CommittedOffsets) of |
| 1045 | + {_, ?undef} -> |
| 1046 | + %% No committed offset (OffsetFetch returned -1, or the |
| 1047 | + %% consumer_managed callback returned ?undef explicitly). The |
| 1048 | + %% subscriber will resolve via the consumer config begin_offset / |
| 1049 | + %% offset_reset_policy. |
| 1050 | + ?undef; |
| 1041 1051 | {_, {begin_offset, Offset}} -> |
| 1042 1052 | %% already resolved |
| 1043 1053 | resolve_special_offset(Offset); |
| @@ -1050,9 +1060,10 @@ resolve_begin_offsets([{Topic, Partition} | Rest], CommittedOffsets, | |
| 1050 1060 | %% offsets committed to kafka is already begin_offset |
| 1051 1061 | %% since the introduction of 'roundrobin_v2' protocol |
| 1052 1062 | Offset; |
| 1053 | - false -> |
| 1054 | - %% No commit history found |
| 1055 | - %% Should use default begin_offset in consumer config |
| 1063 | + false -> |
| 1064 | + %% Partition missing from the result. Kafka omits partitions |
| 1065 | + %% with no committed offset in some response versions, so this |
| 1066 | + %% has the same "no committed offset" meaning. |
| 1056 1067 | ?undef |
| 1057 1068 | end, |
| 1058 1069 | Assignment = |
| @@ -578,14 +578,16 @@ consume_ack(Pid, Offset) -> | |
| 578 578 | do_commit_ack(Pid, GenerationId, Topic, Partition, Offset) -> |
| 579 579 | ok = brod_group_coordinator:ack(Pid, GenerationId, Topic, Partition, Offset). |
| 580 580 | |
| 581 | - subscribe_partitions(#state{ client = Client |
| 582 | - , consumers = Consumers0 |
| 581 | + subscribe_partitions(#state{ client = Client |
| 582 | + , consumer_config = ConsumerConfig |
| 583 | + , consumers = Consumers0 |
| 583 584 | } = State) -> |
| 584 585 | Consumers = |
| 585 | - lists:map(fun(C) -> subscribe_partition(Client, C) end, Consumers0), |
| 586 | + lists:map(fun(C) -> subscribe_partition(Client, ConsumerConfig, C) end, |
| 587 | + Consumers0), |
| 586 588 | {ok, State#state{consumers = Consumers}}. |
| 587 589 | |
| 588 | - subscribe_partition(Client, Consumer) -> |
| 590 | + subscribe_partition(Client, ConsumerConfig, Consumer) -> |
| 589 591 | #consumer{ topic_partition = {Topic, Partition} |
| 590 592 | , consumer_pid = Pid |
| 591 593 | , begin_offset = BeginOffset0 |
| @@ -608,11 +610,7 @@ subscribe_partition(Client, Consumer) -> | |
| 608 610 | ?undef -> BeginOffset0; |
| 609 611 | N when N >= 0 -> N + 1 |
| 610 612 | end, |
| 611 | - Options = |
| 612 | - case BeginOffset =:= ?undef of |
| 613 | - true -> []; %% fetch from 'begin_offset' in consumer config |
| 614 | - false -> [{begin_offset, BeginOffset}] |
| 615 | - end, |
| 613 | + Options = subscribe_options(BeginOffset, ConsumerConfig, Topic, Partition), |
| 616 614 | case brod:subscribe(Client, self(), Topic, Partition, Options) of |
| 617 615 | {ok, ConsumerPid} -> |
| 618 616 | Mref = erlang:monitor(process, ConsumerPid), |
| @@ -626,6 +624,39 @@ subscribe_partition(Client, Consumer) -> | |
| 626 624 | end |
| 627 625 | end. |
| 628 626 | |
| 627 | + %% Decide subscribe options based on the assigned begin_offset. |
| 628 | + %% `?undef' means the coordinator has no committed offset for this partition |
| 629 | + %% (OffsetFetch returned -1, the partition was omitted from the response, or |
| 630 | + %% the consumer_managed callback returned ?undef). Honour `begin_offset' in |
| 631 | + %% consumer config first (backwards compatible: pass no options and let the |
| 632 | + %% consumer use its configured start point), otherwise inject |
| 633 | + %% `{begin_offset, X}' derived from `offset_reset_policy'. Either way the |
| 634 | + %% choice is logged so the fallback is no longer silent. |
| 635 | + subscribe_options(?undef, ConsumerConfig, Topic, Partition) -> |
| 636 | + case proplists:lookup(begin_offset, ConsumerConfig) of |
| 637 | + {begin_offset, Offset} -> |
| 638 | + ?BROD_LOG_INFO("No committed offset for ~s-~p, starting from " |
| 639 | + "consumer config begin_offset=~p", |
| 640 | + [Topic, Partition, Offset]), |
| 641 | + []; %% consumer will use its configured begin_offset |
| 642 | + none -> |
| 643 | + Policy = proplists:get_value(offset_reset_policy, ConsumerConfig, |
| 644 | + reset_by_subscriber), |
| 645 | + Offset = reset_policy_to_offset(Policy), |
| 646 | + ?BROD_LOG_INFO("No committed offset for ~s-~p, starting from ~p " |
| 647 | + "(offset_reset_policy=~p)", |
| 648 | + [Topic, Partition, Offset, Policy]), |
| 649 | + [{begin_offset, Offset}] |
| 650 | + end; |
| 651 | + subscribe_options(BeginOffset, _ConsumerConfig, _Topic, _Partition) -> |
| 652 | + [{begin_offset, BeginOffset}]. |
| 653 | + |
| 654 | + %% reset_by_subscriber has no equivalent at assignment time, so default to |
| 655 | + %% ?OFFSET_LATEST — matches brod_consumer's own default begin_offset. |
| 656 | + reset_policy_to_offset(reset_to_earliest) -> ?OFFSET_EARLIEST; |
| 657 | + reset_policy_to_offset(reset_to_latest) -> ?OFFSET_LATEST; |
| 658 | + reset_policy_to_offset(_) -> ?OFFSET_LATEST. |
| 659 | + |
| 629 660 | log(#state{ groupId = GroupId |
| 630 661 | , memberId = MemberId |
| 631 662 | , generationId = GenerationId |
| @@ -490,11 +490,13 @@ terminate_worker(WorkerPid) -> | |
| 490 490 | , brod:consumer_config() |
| 491 491 | , brod:topic() |
| 492 492 | , brod:partition() |
| 493 | - , brod:offset() | undefined |
| 493 | + , undefined |
| 494 | + | brod:offset() |
| 495 | + | {begin_offset, brod:offset_time()} |
| 494 496 | , state() |
| 495 497 | ) -> state(). |
| 496 498 | maybe_start_worker( _MemberId |
| 497 | - , ConsumerConfig |
| 499 | + , ConsumerConfig0 |
| 498 500 | , Topic |
| 499 501 | , Partition |
| 500 502 | , BeginOffset |
| @@ -512,6 +514,8 @@ maybe_start_worker( _MemberId | |
| 512 514 | #{TopicPartition := _Worker} -> |
| 513 515 | State; |
| 514 516 | _ -> |
| 517 | + ConsumerConfig = |
| 518 | + maybe_apply_reset_policy(BeginOffset, ConsumerConfig0, Topic, Partition), |
| 515 519 | Self = self(), |
| 516 520 | CommitFun = fun(Offset) -> commit(Self, Topic, Partition, Offset) end, |
| 517 521 | AckFun = fun(Offset) -> ack(Self, Topic, Partition, Offset) end, |
| @@ -567,6 +571,40 @@ do_ack(Topic, Partition, Offset, #state{ workers = Workers | |
| 567 571 | {error, unknown_topic_or_partition} |
| 568 572 | end. |
| 569 573 | |
| 574 | + %% When the coordinator has no committed offset for this partition |
| 575 | + %% (`BeginOffset = ?undef'), make the fallback explicit: honour |
| 576 | + %% `begin_offset' in consumer config first (backwards compatible |
| 577 | + %% start-point override), otherwise inject `{begin_offset, X}' from |
| 578 | + %% `offset_reset_policy'. Either way the chosen offset is logged so the |
| 579 | + %% start-up choice is no longer silent. `?undef' is then passed through |
| 580 | + %% to the worker — the consumer applies the (possibly updated) |
| 581 | + %% `begin_offset' from its config. |
| 582 | + maybe_apply_reset_policy(?undef, ConsumerConfig, Topic, Partition) -> |
| 583 | + case proplists:is_defined(begin_offset, ConsumerConfig) of |
| 584 | + true -> |
| 585 | + ?BROD_LOG_INFO("No committed offset for ~s-~p, starting from " |
| 586 | + "consumer config begin_offset=~p", |
| 587 | + [Topic, Partition, |
| 588 | + proplists:get_value(begin_offset, ConsumerConfig)]), |
| 589 | + ConsumerConfig; |
| 590 | + false -> |
| 591 | + Policy = proplists:get_value(offset_reset_policy, ConsumerConfig, |
| 592 | + reset_by_subscriber), |
| 593 | + Offset = reset_policy_to_offset(Policy), |
| 594 | + ?BROD_LOG_INFO("No committed offset for ~s-~p, starting from ~p " |
| 595 | + "(offset_reset_policy=~p)", |
| 596 | + [Topic, Partition, Offset, Policy]), |
| 597 | + [{begin_offset, Offset} | ConsumerConfig] |
| 598 | + end; |
| 599 | + maybe_apply_reset_policy(_BeginOffset, ConsumerConfig, _Topic, _Partition) -> |
| 600 | + ConsumerConfig. |
| 601 | + |
| 602 | + %% reset_by_subscriber has no equivalent at assignment time, so default to |
| 603 | + %% ?OFFSET_LATEST — matches brod_consumer's own default begin_offset. |
| 604 | + reset_policy_to_offset(reset_to_earliest) -> ?OFFSET_EARLIEST; |
| 605 | + reset_policy_to_offset(reset_to_latest) -> ?OFFSET_LATEST; |
| 606 | + reset_policy_to_offset(_) -> ?OFFSET_LATEST. |
| 607 | + |
| 570 608 | stop_consumers(Config) -> |
| 571 609 | #{ client := Client |
| 572 610 | , topics := Topics |
Loading more files…