Packages
brod
3.7.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
+30
additions
-11
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.3 |
| 4 | + PROJECT_VERSION = 3.7.4 |
| 5 5 | |
| 6 6 | DEPS = supervisor3 kafka_protocol |
| @@ -1,5 +1,5 @@ | |
| 1 1 | {<<"name">>,<<"brod">>}. |
| 2 | - {<<"version">>,<<"3.7.3">>}. |
| 2 | + {<<"version">>,<<"3.7.4">>}. |
| 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.7.3"}, |
| 3 | + {vsn,"3.7.4"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol,supervisor3]}, |
| 6 6 | {env,[]}, |
| @@ -512,11 +512,13 @@ join_group(#state{ groupId = GroupId | |
| 512 512 | , connection = Connection |
| 513 513 | , session_timeout_seconds = SessionTimeoutSec |
| 514 514 | , protocol_name = ProtocolName |
| 515 | + , member_module = MemberModule |
| 516 | + , member_pid = MemberPid |
| 515 517 | } = State0) -> |
| 516 518 | Meta = |
| 517 519 | [ {version, ?BROD_CONSUMER_GROUP_PROTOCOL_VERSION} |
| 518 520 | , {topics, Topics} |
| 519 | - , {user_data, user_data(join)} |
| 521 | + , {user_data, user_data(MemberModule, MemberPid)} |
| 520 522 | ], |
| 521 523 | Protocol = |
| 522 524 | [ {protocol_name, ProtocolName} |
| @@ -743,7 +745,7 @@ assign_partitions(State) when ?IS_LEADER(State) -> | |
| 743 745 | , {member_assignment, |
| 744 746 | [ {version, ?BROD_CONSUMER_GROUP_PROTOCOL_VERSION} |
| 745 747 | , {topic_partitions, PartitionAssignments} |
| 746 | - , {user_data, user_data(assign)} |
| 748 | + , {user_data, <<>>} |
| 747 749 | ]} |
| 748 750 | ] |
| 749 751 | end, Assignments); |
| @@ -1007,12 +1009,17 @@ make_offset_commit_metadata() -> | |
| 1007 1009 | %% then received by group leader, group leader (may mutate it and) |
| 1008 1010 | %% assigns it back to members in topic-partition assignments. |
| 1009 1011 | %% |
| 1010 | - %% Currently user_data is created in this module and terminated in this |
| 1011 | - %% module, when needed for advanced features, we can originate it from |
| 1012 | - %% member's init callback, and pass it to members via |
| 1013 | - %% `brod_received_assignments()' |
| 1014 | - -spec user_data(join | assign) -> binary(). |
| 1015 | - user_data(_Action) -> <<"">>. |
| 1012 | + %% Currently user_data in join requests can be created from a |
| 1013 | + %% callback to member module. |
| 1014 | + %% user_data assigned back to members is always an empty binary |
| 1015 | + %% if assign_partitions callback is not implemented, otherwise |
| 1016 | + %% should be set in the assign_partitions callback implemenation. |
| 1017 | + user_data(Module, Pid) -> |
| 1018 | + %% Module is ensured to be loaded already |
| 1019 | + case erlang:function_exported(Module, user_data, 1) of |
| 1020 | + true -> Module:user_data(Pid); |
| 1021 | + false -> <<>> |
| 1022 | + end. |
| 1016 1023 | |
| 1017 1024 | %% Make a client ID to be used in the requests sent over the group |
| 1018 1025 | %% coordinator's connection (group coordinator broker on the other end), |
| @@ -46,6 +46,10 @@ | |
| 46 46 | |
| 47 47 | -include("brod_int.hrl"). |
| 48 48 | |
| 49 | + -optional_callbacks([assign_partitions/3, |
| 50 | + user_data/1 |
| 51 | + ]). |
| 52 | + |
| 49 53 | %% Call the callback module to initialize assignments. |
| 50 54 | %% NOTE: This function is called only when `offset_commit_policy' is |
| 51 55 | %% `consumer_managed' in group config. |
| @@ -76,6 +80,11 @@ | |
| 76 80 | %% brod:unsubscribe/3 to unsubscribe from all currently subscribed partitions. |
| 77 81 | -callback assignments_revoked(pid()) -> ok. |
| 78 82 | |
| 83 | + %% Called when making join request. This metadata is to let group leader know |
| 84 | + %% more details about the member. e.g. its location and or capacity etc. |
| 85 | + %% so that leader can make smarter decisions when assigning partitions to it. |
| 86 | + -callback user_data(pid()) -> binary(). |
| 87 | + |
| 79 88 | %%%_* Emacs ==================================================================== |
| 80 89 | %%% Local Variables: |
| 81 90 | %%% allout-layout: t |
Loading more files…