Packages
brod
4.2.0
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
5
files changed
+88
additions
-30
deletions
| @@ -25,4 +25,4 @@ | |
| 25 25 | [{<<"app">>,<<"kafka_protocol">>}, |
| 26 26 | {<<"optional">>,false}, |
| 27 27 | {<<"requirement">>,<<"4.1.9">>}]}]}. |
| 28 | - {<<"version">>,<<"4.1.1">>}. |
| 28 | + {<<"version">>,<<"4.2.0">>}. |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,brod, |
| 2 2 | [{description,"Apache Kafka Erlang client library"}, |
| 3 | - {vsn,"4.1.1"}, |
| 3 | + {vsn,"4.2.0"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol]}, |
| 6 6 | {env,[]}, |
| @@ -262,6 +262,7 @@ | |
| 262 262 | | {offset_reset_policy, brod_consumer:offset_reset_policy()} |
| 263 263 | | {size_stat_window, non_neg_integer()} |
| 264 264 | | {isolation_level, brod_consumer:isolation_level()} |
| 265 | + | {share_leader_conn, boolean()} |
| 265 266 | ]. |
| 266 267 | %% Consumer configuration. |
| 267 268 | %% |
| @@ -1,5 +1,5 @@ | |
| 1 | - %%% |
| 2 1 | %%% Copyright (c) 2015-2021 Klarna Bank AB (publ) |
| 2 | + %%% Copyright (c) 2022-2024 kafka4beam contributors |
| 3 3 | %%% |
| 4 4 | %%% Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 5 | %%% you may not use this file except in compliance with the License. |
| @@ -30,6 +30,7 @@ | |
| 30 30 | , get_group_coordinator/2 |
| 31 31 | , get_transactional_coordinator/2 |
| 32 32 | , get_leader_connection/3 |
| 33 | + , get_bootstrap/1 |
| 33 34 | , get_metadata/2 |
| 34 35 | , get_metadata_safe/2 |
| 35 36 | , get_partitions_count/2 |
| @@ -227,6 +228,10 @@ stop_consumer(Client, TopicName) -> | |
| 227 228 | get_leader_connection(Client, Topic, Partition) -> |
| 228 229 | safe_gen_call(Client, {get_leader_connection, Topic, Partition}, infinity). |
| 229 230 | |
| 231 | + -spec get_bootstrap(client()) -> {ok, brod:bootstrap()} | {error, any()}. |
| 232 | + get_bootstrap(Client) -> |
| 233 | + safe_gen_call(Client, get_bootstrap, infinity). |
| 234 | + |
| 230 235 | %% @doc Get connection to a kafka broker. |
| 231 236 | %% |
| 232 237 | %% Return already established connection towards the broker, |
| @@ -388,6 +393,10 @@ handle_call({stop_consumer, Topic}, _From, State) -> | |
| 388 393 | handle_call({get_leader_connection, Topic, Partition}, _From, State) -> |
| 389 394 | {Result, NewState} = do_get_leader_connection(State, Topic, Partition), |
| 390 395 | {reply, Result, NewState}; |
| 396 | + handle_call(get_bootstrap, _From, State) -> |
| 397 | + #state{bootstrap_endpoints = Endpoints} = State, |
| 398 | + ConnConfig = conn_config(State), |
| 399 | + {reply, {ok, {Endpoints, ConnConfig}}, State}; |
| 391 400 | handle_call({get_connection, Host, Port}, _From, State) -> |
| 392 401 | {Result, NewState} = maybe_connect(State, {Host, Port}), |
| 393 402 | {reply, Result, NewState}; |
| @@ -1,4 +1,5 @@ | |
| 1 1 | %%% Copyright (c) 2014-2021 Klarna Bank AB (publ) |
| 2 | + %%% Copyright (c) 2022-2024 kafka4beam contributors |
| 2 3 | %%% |
| 3 4 | %%% Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 5 | %%% you may not use this file except in compliance with the License. |
| @@ -93,7 +94,10 @@ | |
| 93 94 | -type pending_acks() :: #pending_acks{}. |
| 94 95 | -type isolation_level() :: kpro:isolation_level(). |
| 95 96 | |
| 96 | - -record(state, { bootstrap :: pid() | brod:bootstrap() |
| 97 | + -define(GET_FROM_CLIENT, get). |
| 98 | + -define(IGNORE, ignore). |
| 99 | + -record(state, { client_pid :: ?IGNORE | pid() |
| 100 | + , bootstrap :: ?IGNORE | ?GET_FROM_CLIENT | brod:bootstrap() |
| 97 101 | , connection :: ?undef | pid() |
| 98 102 | , topic :: binary() |
| 99 103 | , partition :: integer() |
| @@ -136,6 +140,7 @@ | |
| 136 140 | -define(INIT_CONNECTION, init_connection). |
| 137 141 | -define(DEFAULT_AVG_WINDOW, 5). |
| 138 142 | -define(DEFAULT_ISOLATION_LEVEL, ?kpro_read_committed). |
| 143 | + -define(DEFAULT_SHARE_LEADER_CONN, false). |
| 139 144 | |
| 140 145 | %%%_* APIs ===================================================================== |
| 141 146 | %% @equiv start_link(ClientPid, Topic, Partition, Config, []) |
| @@ -220,6 +225,16 @@ start_link(Bootstrap, Topic, Partition, Config) -> | |
| 220 225 | %% and `read_committed' to get only the records from committed |
| 221 226 | %% transactions</li> |
| 222 227 | %% |
| 228 | + %% <li>`share_leader_conn': (optional, default = `false') |
| 229 | + %% |
| 230 | + %% Whether or not share the partition leader connection with |
| 231 | + %% other producers or consumers. |
| 232 | + %% Set to `true' to consume less TCP connections towards Kafka, |
| 233 | + %% but may lead to higher fetch latency. This is because Kafka can |
| 234 | + %% ony accumulate messages for the oldest fetch request, later |
| 235 | + %% requests behind it may get blocked until `max_wait_time' expires |
| 236 | + %% for the oldest one</li> |
| 237 | + %% |
| 223 238 | %% </ul> |
| 224 239 | %% @end |
| 225 240 | -spec start_link(pid() | brod:bootstrap(), |
| @@ -286,7 +301,7 @@ get_connection(Pid) -> | |
| 286 301 | |
| 287 302 | %%%_* gen_server callbacks ===================================================== |
| 288 303 | |
| 289 | - init({Bootstrap, Topic, Partition, Config}) -> |
| 304 | + init({Bootstrap0, Topic, Partition, Config}) -> |
| 290 305 | erlang:process_flag(trap_exit, true), |
| 291 306 | Cfg = fun(Name, Default) -> |
| 292 307 | proplists:get_value(Name, Config, Default) |
| @@ -300,15 +315,33 @@ init({Bootstrap, Topic, Partition, Config}) -> | |
| 300 315 | BeginOffset = Cfg(begin_offset, ?DEFAULT_BEGIN_OFFSET), |
| 301 316 | OffsetResetPolicy = Cfg(offset_reset_policy, ?DEFAULT_OFFSET_RESET_POLICY), |
| 302 317 | IsolationLevel = Cfg(isolation_level, ?DEFAULT_ISOLATION_LEVEL), |
| 318 | + IsShareConn = Cfg(share_leader_conn, ?DEFAULT_SHARE_LEADER_CONN), |
| 303 319 | |
| 304 | - %% If bootstrap is a client pid, register self to the client |
| 305 | - case is_shared_conn(Bootstrap) of |
| 320 | + %% resolve connection bootstrap args |
| 321 | + {ClientPid, Bootstrap} = |
| 322 | + case is_pid(Bootstrap0) of |
| 323 | + true when IsShareConn -> |
| 324 | + %% share leader connection with other producers/consumers |
| 325 | + %% the connection is to be managed by brod_client |
| 326 | + {Bootstrap0, ?IGNORE}; |
| 327 | + true -> |
| 328 | + %% not sharing leader connection with other producers/consumers |
| 329 | + %% the bootstrap args will be resolved later when it's |
| 330 | + %% time to establish a connection to partition leader |
| 331 | + {Bootstrap0, ?GET_FROM_CLIENT}; |
| 332 | + false -> |
| 333 | + %% this consumer process is not started from `brod' APIs |
| 334 | + %% maybe managed by other supervisors. |
| 335 | + {?IGNORE, Bootstrap0} |
| 336 | + end, |
| 337 | + case is_pid(ClientPid) of |
| 306 338 | true -> |
| 307 339 | ok = brod_client:register_consumer(Bootstrap, Topic, Partition); |
| 308 340 | false -> |
| 309 341 | ok |
| 310 342 | end, |
| 311 | - {ok, #state{ bootstrap = Bootstrap |
| 343 | + {ok, #state{ client_pid = ClientPid |
| 344 | + , bootstrap = Bootstrap |
| 312 345 | , topic = Topic |
| 313 346 | , partition = Partition |
| 314 347 | , begin_offset = BeginOffset |
| @@ -418,20 +451,26 @@ handle_cast(Cast, State) -> | |
| 418 451 | {noreply, State}. |
| 419 452 | |
| 420 453 | %% @private |
| 421 | - terminate(Reason, #state{ bootstrap = Bootstrap |
| 454 | + terminate(Reason, #state{ client_pid = ClientPid |
| 422 455 | , topic = Topic |
| 423 456 | , partition = Partition |
| 424 457 | , connection = Connection |
| 458 | + , connection_mref = Mref |
| 425 459 | }) -> |
| 426 | - IsShared = is_shared_conn(Bootstrap), |
| 427 460 | IsNormal = brod_utils:is_normal_reason(Reason), |
| 428 461 | %% deregister consumer if it's shared connection and normal shutdown |
| 429 | - IsShared andalso IsNormal andalso |
| 430 | - brod_client:deregister_consumer(Bootstrap, Topic, Partition), |
| 431 | - %% close connection if it's working standalone |
| 432 | - case not IsShared andalso is_pid(Connection) of |
| 433 | - true -> kpro:close_connection(Connection); |
| 434 | - false -> ok |
| 462 | + case is_pid(ClientPid) andalso IsNormal of |
| 463 | + true -> |
| 464 | + brod_client:deregister_consumer(ClientPid, Topic, Partition); |
| 465 | + false -> |
| 466 | + ok |
| 467 | + end, |
| 468 | + %% close connection if it's owned by this consumer |
| 469 | + case Mref =:= ?undef andalso is_pid(Connection) andalso is_process_alive(Connection) of |
| 470 | + true -> |
| 471 | + kpro:close_connection(Connection); |
| 472 | + false -> |
| 473 | + ok |
| 435 474 | end, |
| 436 475 | %% write a log if it's not a normal reason |
| 437 476 | IsNormal orelse ?BROD_LOG_ERROR("Consumer ~s-~w terminate reason: ~p", |
| @@ -858,17 +897,19 @@ safe_gen_call(Server, Call, Timeout) -> | |
| 858 897 | -spec maybe_init_connection(state()) -> |
| 859 898 | {ok, state()} | {{error, any()}, state()}. |
| 860 899 | maybe_init_connection( |
| 861 | - #state{ bootstrap = Bootstrap |
| 900 | + #state{ client_pid = ClientPid |
| 901 | + , bootstrap = Bootstrap |
| 862 902 | , topic = Topic |
| 863 903 | , partition = Partition |
| 864 904 | , connection = ?undef |
| 865 905 | } = State0) -> |
| 866 906 | %% Lookup, or maybe (re-)establish a connection to partition leader |
| 867 | - case connect_leader(Bootstrap, Topic, Partition) of |
| 907 | + {MonitorOrLink, Result} = connect_leader(ClientPid, Bootstrap, Topic, Partition), |
| 908 | + case Result of |
| 868 909 | {ok, Connection} -> |
| 869 | - Mref = case is_shared_conn(Bootstrap) of |
| 870 | - true -> erlang:monitor(process, Connection); |
| 871 | - false -> ?undef %% linked |
| 910 | + Mref = case MonitorOrLink of |
| 911 | + monitor -> erlang:monitor(process, Connection); |
| 912 | + linked -> ?undef |
| 872 913 | end, |
| 873 914 | %% Switching to a new connection |
| 874 915 | %% the response for last_req_ref will be lost forever |
| @@ -883,13 +924,23 @@ maybe_init_connection( | |
| 883 924 | maybe_init_connection(State) -> |
| 884 925 | {ok, State}. |
| 885 926 | |
| 886 | - connect_leader(ClientPid, Topic, Partition) when is_pid(ClientPid) -> |
| 887 | - brod_client:get_leader_connection(ClientPid, Topic, Partition); |
| 888 | - connect_leader(Endpoints, Topic, Partition) when is_list(Endpoints) -> |
| 889 | - connect_leader({Endpoints, []}, Topic, Partition); |
| 890 | - connect_leader({Endpoints, ConnCfg}, Topic, Partition) -> |
| 927 | + connect_leader(ClientPid, ?IGNORE, Topic, Partition) when is_pid(ClientPid) -> |
| 928 | + {monitor, brod_client:get_leader_connection(ClientPid, Topic, Partition)}; |
| 929 | + connect_leader(ClientPid, ?GET_FROM_CLIENT, Topic, Partition) when is_pid(ClientPid) -> |
| 930 | + case brod_client:get_bootstrap(ClientPid) of |
| 931 | + {ok, Bootstrap} -> |
| 932 | + link_connect_leader(Bootstrap, Topic, Partition); |
| 933 | + {error, Reason} -> |
| 934 | + {linked, {error, Reason}} |
| 935 | + end; |
| 936 | + connect_leader(?IGNORE, Bootstrap, Topic, Partition) -> |
| 937 | + link_connect_leader(Bootstrap, Topic, Partition). |
| 938 | + |
| 939 | + link_connect_leader(Endpoints, Topic, Partition) when is_list(Endpoints) -> |
| 940 | + link_connect_leader({Endpoints, []}, Topic, Partition); |
| 941 | + link_connect_leader({Endpoints, ConnCfg}, Topic, Partition) -> |
| 891 942 | %% connection pid is linked to self() |
| 892 | - kpro:connect_partition_leader(Endpoints, ConnCfg, Topic, Partition). |
| 943 | + {linked, kpro:connect_partition_leader(Endpoints, ConnCfg, Topic, Partition)}. |
| 893 944 | |
| 894 945 | %% Send a ?INIT_CONNECTION delayed loopback message to re-init. |
| 895 946 | -spec maybe_send_init_connection(state()) -> ok. |
| @@ -900,9 +951,6 @@ maybe_send_init_connection(#state{subscriber = Subscriber}) -> | |
| 900 951 | erlang:send_after(Timeout, self(), ?INIT_CONNECTION), |
| 901 952 | ok. |
| 902 953 | |
| 903 | - %% In case 'bootstrap' is a client pid, connection is shared with other workers. |
| 904 | - is_shared_conn(Bootstrap) -> is_pid(Bootstrap). |
| 905 | - |
| 906 954 | %%%_* Tests ==================================================================== |
| 907 955 | |
| 908 956 | -ifdef(TEST). |