Packages
brod
3.15.6
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
+87
additions
-68
deletions
| @@ -27,4 +27,4 @@ | |
| 27 27 | [{<<"app">>,<<"supervisor3">>}, |
| 28 28 | {<<"optional">>,false}, |
| 29 29 | {<<"requirement">>,<<"1.1.11">>}]}]}. |
| 30 | - {<<"version">>,<<"3.15.5">>}. |
| 30 | + {<<"version">>,<<"3.15.6">>}. |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,brod, |
| 2 2 | [{description,"Apache Kafka Erlang client library"}, |
| 3 | - {vsn,"3.15.5"}, |
| 3 | + {vsn,"3.15.6"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol,supervisor3]}, |
| 6 6 | {env,[]}, |
| @@ -34,6 +34,11 @@ | |
| 34 34 | , format_status/2 |
| 35 35 | ]). |
| 36 36 | |
| 37 | + -export([ do_send_fun/4 |
| 38 | + , do_no_ack/2 |
| 39 | + , do_bufcb/2 |
| 40 | + ]). |
| 41 | + |
| 37 42 | -export_type([ config/0 ]). |
| 38 43 | |
| 39 44 | -include("brod_int.hrl"). |
| @@ -202,7 +207,7 @@ produce(Pid, Key, Value) -> | |
| 202 207 | -spec produce_no_ack(pid(), brod:key(), brod:value()) -> ok. |
| 203 208 | produce_no_ack(Pid, Key, Value) -> |
| 204 209 | CallRef = #brod_call_ref{caller = ?undef}, |
| 205 | - AckCb = fun(_, _) -> ok end, |
| 210 | + AckCb = fun ?MODULE:do_no_ack/2, |
| 206 211 | Batch = brod_utils:make_batch_input(Key, Value), |
| 207 212 | Pid ! {produce, CallRef, Batch, AckCb}, |
| 208 213 | ok. |
| @@ -280,20 +285,7 @@ init({ClientPid, Topic, Partition, Config}) -> | |
| 280 285 | Compression = ?config(compression, ?DEFAULT_COMPRESSION), |
| 281 286 | MaxLingerMs = ?config(max_linger_ms, ?DEFAULT_MAX_LINGER_MS), |
| 282 287 | MaxLingerCount = ?config(max_linger_count, ?DEFAULT_MAX_LINGER_COUNT), |
| 283 | - SendFun = |
| 284 | - fun(Conn, BatchInput, Vsn) -> |
| 285 | - ProduceRequest = |
| 286 | - brod_kafka_request:produce(Vsn, Topic, Partition, BatchInput, |
| 287 | - RequiredAcks, AckTimeout, Compression), |
| 288 | - case send(Conn, ProduceRequest) of |
| 289 | - ok when ProduceRequest#kpro_req.no_ack -> |
| 290 | - ok; |
| 291 | - ok -> |
| 292 | - {ok, ProduceRequest#kpro_req.ref}; |
| 293 | - {error, Reason} -> |
| 294 | - {error, Reason} |
| 295 | - end |
| 296 | - end, |
| 288 | + SendFun = make_send_fun(Topic, Partition, RequiredAcks, AckTimeout, Compression), |
| 297 289 | Buffer = brod_producer_buffer:new(BufferLimit, OnWireLimit, MaxBatchSize, |
| 298 290 | MaxRetries, MaxLingerMs, MaxLingerCount, |
| 299 291 | SendFun), |
| @@ -346,25 +338,7 @@ handle_info({'DOWN', _MonitorRef, process, Pid, Reason}, | |
| 346 338 | {noreply, NewState#state{connection = ?undef, conn_mref = ?undef}} |
| 347 339 | end; |
| 348 340 | handle_info({produce, CallRef, Batch, AckCb}, #state{partition = Partition} = State) -> |
| 349 | - #brod_call_ref{caller = Pid} = CallRef, |
| 350 | - BufCb = |
| 351 | - fun(?buffered) when is_pid(Pid) -> |
| 352 | - Reply = #brod_produce_reply{ call_ref = CallRef |
| 353 | - , result = ?buffered |
| 354 | - }, |
| 355 | - erlang:send(Pid, Reply); |
| 356 | - (?buffered) -> |
| 357 | - %% caller requires no ack |
| 358 | - ok; |
| 359 | - ({?acked, BaseOffset}) when AckCb =:= ?undef -> |
| 360 | - Reply = #brod_produce_reply{ call_ref = CallRef |
| 361 | - , base_offset = BaseOffset |
| 362 | - , result = ?acked |
| 363 | - }, |
| 364 | - erlang:send(Pid, Reply); |
| 365 | - ({?acked, BaseOffset}) when is_function(AckCb, 2) -> |
| 366 | - AckCb(Partition, BaseOffset) |
| 367 | - end, |
| 341 | + BufCb = make_bufcb(CallRef, AckCb, Partition), |
| 368 342 | handle_produce(BufCb, Batch, State); |
| 369 343 | handle_info({msg, Pid, #kpro_rsp{ api = produce |
| 370 344 | , ref = Ref |
| @@ -435,11 +409,55 @@ format_status(terminate, [_PDict, State=#state{buffer = Buffer}]) -> | |
| 435 409 | |
| 436 410 | %%%_* Internal Functions ======================================================= |
| 437 411 | |
| 412 | + make_send_fun(Topic, Partition, RequiredAcks, AckTimeout, Compression) -> |
| 413 | + ExtraArg = {Topic, Partition, RequiredAcks, AckTimeout, Compression}, |
| 414 | + {fun ?MODULE:do_send_fun/4, ExtraArg}. |
| 415 | + |
| 416 | + do_send_fun(ExtraArg, Conn, BatchInput, Vsn) -> |
| 417 | + {Topic, Partition, RequiredAcks, AckTimeout, Compression} = ExtraArg, |
| 418 | + ProduceRequest = |
| 419 | + brod_kafka_request:produce(Vsn, Topic, Partition, BatchInput, |
| 420 | + RequiredAcks, AckTimeout, Compression), |
| 421 | + case send(Conn, ProduceRequest) of |
| 422 | + ok when ProduceRequest#kpro_req.no_ack -> |
| 423 | + ok; |
| 424 | + ok -> |
| 425 | + {ok, ProduceRequest#kpro_req.ref}; |
| 426 | + {error, Reason} -> |
| 427 | + {error, Reason} |
| 428 | + end. |
| 429 | + |
| 430 | + do_no_ack(_Partition, _BaseOffset) -> ok. |
| 431 | + |
| 438 432 | -spec log_error_code(topic(), partition(), offset(), brod:error_code()) -> _. |
| 439 433 | log_error_code(Topic, Partition, Offset, ErrorCode) -> |
| 440 434 | ?BROD_LOG_ERROR("Produce error ~s-~B Offset: ~B Error: ~p", |
| 441 435 | [Topic, Partition, Offset, ErrorCode]). |
| 442 436 | |
| 437 | + make_bufcb(CallRef, AckCb, Partition) -> |
| 438 | + {fun ?MODULE:do_bufcb/2, _ExtraArg = {CallRef, AckCb, Partition}}. |
| 439 | + |
| 440 | + do_bufcb({CallRef, AckCb, Partition}, Arg) -> |
| 441 | + #brod_call_ref{caller = Pid} = CallRef, |
| 442 | + case Arg of |
| 443 | + ?buffered when is_pid(Pid) -> |
| 444 | + Reply = #brod_produce_reply{ call_ref = CallRef |
| 445 | + , result = ?buffered |
| 446 | + }, |
| 447 | + erlang:send(Pid, Reply); |
| 448 | + ?buffered -> |
| 449 | + %% caller requires no ack |
| 450 | + ok; |
| 451 | + {?acked, BaseOffset} when AckCb =:= ?undef -> |
| 452 | + Reply = #brod_produce_reply{ call_ref = CallRef |
| 453 | + , base_offset = BaseOffset |
| 454 | + , result = ?acked |
| 455 | + }, |
| 456 | + erlang:send(Pid, Reply); |
| 457 | + {?acked, BaseOffset} when is_function(AckCb, 2) -> |
| 458 | + AckCb(Partition, BaseOffset) |
| 459 | + end. |
| 460 | + |
| 443 461 | handle_produce(BufCb, Batch, |
| 444 462 | #state{retry_tref = Ref} = State) when is_reference(Ref) -> |
| 445 463 | %% pending on retry, add to buffer regardless of connection state |
| @@ -34,16 +34,17 @@ | |
| 34 34 | |
| 35 35 | -include("brod_int.hrl"). |
| 36 36 | |
| 37 | - %% keep data in fun() to avoid huge log dumps in case of crash etc. |
| 38 | - -type data() :: fun(() -> {brod:key(), brod:value()}). |
| 37 | + -type batch() :: [{brod:key(), brod:value()}]. |
| 39 38 | -type milli_ts() :: pos_integer(). |
| 40 39 | -type milli_sec() :: non_neg_integer(). |
| 41 40 | -type count() :: non_neg_integer(). |
| 42 | - -type cb() :: fun((?buffered | {?acked, offset()}) -> ok). |
| 41 | + -type buf_cb_arg() :: ?buffered | {?acked, offset()}. |
| 42 | + -type buf_cb() :: fun((buf_cb_arg()) -> ok) | |
| 43 | + {fun((any(), buf_cb_arg()) -> ok), any()}. |
| 43 44 | |
| 44 45 | -record(req, |
| 45 | - { cb :: cb() |
| 46 | - , data :: data() |
| 46 | + { buf_cb :: buf_cb() |
| 47 | + , data :: batch() |
| 47 48 | , bytes :: non_neg_integer() |
| 48 49 | , msg_cnt :: non_neg_integer() %% messages in the set |
| 49 50 | , ctime :: milli_ts() %% time when request was created |
| @@ -52,10 +53,11 @@ | |
| 52 53 | -type req() :: #req{}. |
| 53 54 | |
| 54 55 | -type vsn() :: brod_kafka_apis:vsn(). |
| 55 | - -type send_fun() :: fun((pid(), [{brod:key(), brod:value()}], vsn()) -> |
| 56 | - ok | |
| 56 | + -type send_fun_res() :: ok | |
| 57 57 | {ok, reference()} | |
| 58 | - {error, any()}). |
| 58 | + {error, any()}. |
| 59 | + -type send_fun() :: fun((pid(), batch(), vsn()) -> send_fun_res()) | |
| 60 | + {fun((any(), pid(), batch(), vsn()) -> send_fun_res()), any()}. |
| 59 61 | -define(ERR_FUN, fun() -> erlang:error(bad_init) end). |
| 60 62 | |
| 61 63 | -define(NEW_QUEUE, queue:new()). |
| @@ -104,10 +106,10 @@ new(BufferLimit, OnWireLimit, MaxBatchSize, MaxRetry, | |
| 104 106 | |
| 105 107 | %% @doc Buffer a produce request. |
| 106 108 | %% Respond to caller immediately if the buffer limit is not yet reached. |
| 107 | - -spec add(buf(), cb(), brod:batch_input()) -> buf(). |
| 108 | - add(#buf{pending = Pending} = Buf, Cb, Batch) -> |
| 109 | - Req = #req{ cb = Cb |
| 110 | - , data = fun() -> Batch end |
| 109 | + -spec add(buf(), buf_cb(), brod:batch_input()) -> buf(). |
| 110 | + add(#buf{pending = Pending} = Buf, BufCb, Batch) -> |
| 111 | + Req = #req{ buf_cb = BufCb |
| 112 | + , data = Batch |
| 111 113 | , bytes = data_size(Batch) |
| 112 114 | , msg_cnt = length(Batch) |
| 113 115 | , ctime = now_ms() |
| @@ -164,11 +166,11 @@ nack(#buf{onwire = [{Ref, _Reqs} | _]} = Buf, Ref, Reason) -> | |
| 164 166 | %% reached maximum retry limit. |
| 165 167 | -spec nack_all(buf(), any()) -> buf(). |
| 166 168 | nack_all(#buf{onwire = OnWire} = Buf, Reason) -> |
| 167 | - AllOnWireReqs = lists:map(fun({_Ref, Reqs}) -> Reqs end, OnWire), |
| 169 | + AllOnWireReqs = lists:flatmap(fun({_Ref, Reqs}) -> Reqs end, OnWire), |
| 168 170 | NewBuf = Buf#buf{ onwire_count = 0 |
| 169 171 | , onwire = [] |
| 170 172 | }, |
| 171 | - rebuffer_or_crash(lists:append(AllOnWireReqs), NewBuf, Reason). |
| 173 | + rebuffer_or_crash(AllOnWireReqs, NewBuf, Reason). |
| 172 174 | |
| 173 175 | %% @doc Return true if there is no message pending, |
| 174 176 | %% buffered or waiting for ack. |
| @@ -270,8 +272,8 @@ do_send(Reqs, #buf{ onwire_count = OnWireCount | |
| 270 272 | , onwire = OnWire |
| 271 273 | , send_fun = SendFun |
| 272 274 | } = Buf, Conn, Vsn) -> |
| 273 | - Batch = lists:append(lists:map(fun(#req{data = F}) -> F() end, Reqs)), |
| 274 | - case SendFun(Conn, Batch, Vsn) of |
| 275 | + Batch = lists:flatmap(fun(#req{data = Data}) -> Data end, Reqs), |
| 276 | + case apply_sendfun(SendFun, Conn, Batch, Vsn) of |
| 275 277 | ok -> |
| 276 278 | %% fire and forget, do not add onwire counter |
| 277 279 | ok = lists:foreach(fun eval_acked/1, Reqs), |
| @@ -295,6 +297,11 @@ do_send(Reqs, #buf{ onwire_count = OnWireCount | |
| 295 297 | {retry, NewBuf} |
| 296 298 | end. |
| 297 299 | |
| 300 | + apply_sendfun({SendFun, ExtraArg}, Conn, Batch, Vsn) -> |
| 301 | + SendFun(ExtraArg, Conn, Batch, Vsn); |
| 302 | + apply_sendfun(SendFun, Conn, Batch, Vsn) -> |
| 303 | + SendFun(Conn, Batch, Vsn). |
| 304 | + |
| 298 305 | %% Put the produce requests back to buffer. |
| 299 306 | %% raise an 'exit' exception if the first request to send has reached |
| 300 307 | %% retry limit |
| @@ -338,8 +345,8 @@ maybe_buffer(#buf{} = Buf) -> | |
| 338 345 | Buf. |
| 339 346 | |
| 340 347 | -spec eval_buffered(req()) -> ok. |
| 341 | - eval_buffered(#req{cb = CbFun}) -> |
| 342 | - _ = CbFun(?buffered), |
| 348 | + eval_buffered(#req{buf_cb = BufCb}) -> |
| 349 | + _ = apply_bufcb(BufCb, ?buffered), |
| 343 350 | ok. |
| 344 351 | |
| 345 352 | -spec eval_acked(req()) -> ok. |
| @@ -348,10 +355,15 @@ eval_acked(Req) -> | |
| 348 355 | ok. |
| 349 356 | |
| 350 357 | -spec eval_acked(req(), offset()) -> offset(). |
| 351 | - eval_acked(#req{cb = CbFun, msg_cnt = Count}, Offset) -> |
| 352 | - _ = CbFun({?acked, Offset}), |
| 358 | + eval_acked(#req{buf_cb = BufCb, msg_cnt = Count}, Offset) -> |
| 359 | + _ = apply_bufcb(BufCb, {?acked, Offset}), |
| 353 360 | next_base_offset(Offset, Count). |
| 354 361 | |
| 362 | + apply_bufcb({BufCb, ExtraArg}, Arg) -> |
| 363 | + BufCb(ExtraArg, Arg); |
| 364 | + apply_bufcb(BufCb, Arg) -> |
| 365 | + BufCb(Arg). |
| 366 | + |
| 355 367 | next_base_offset(?BROD_PRODUCE_UNKNOWN_OFFSET, _) -> |
| 356 368 | ?BROD_PRODUCE_UNKNOWN_OFFSET; |
| 357 369 | next_base_offset(Offset, Count) -> |
| @@ -412,8 +412,7 @@ get_sasl_opt(Config) -> | |
| 412 412 | %% Module should implement kpro_auth_backend behaviour |
| 413 413 | {callback, Module, Args}; |
| 414 414 | {Mechanism, File} when is_list(File) orelse is_binary(File) -> |
| 415 | - {User, Pass} = read_sasl_file(File), |
| 416 | - {Mechanism, User, Pass}; |
| 415 | + {Mechanism, File}; |
| 417 416 | Other -> |
| 418 417 | Other |
| 419 418 | end. |
| @@ -741,16 +740,6 @@ replace_prop(Key, Value, PropL0) -> | |
| 741 740 | PropL = proplists:delete(Key, PropL0), |
| 742 741 | [{Key, Value} | PropL]. |
| 743 742 | |
| 744 | - %% Read a regular file, assume it has two lines: |
| 745 | - %% First line is the sasl-plain username |
| 746 | - %% Second line is the password |
| 747 | - -spec read_sasl_file(file:name_all()) -> {binary(), binary()}. |
| 748 | - read_sasl_file(File) -> |
| 749 | - {ok, Bin} = file:read_file(File), |
| 750 | - Lines = binary:split(Bin, <<"\n">>, [global]), |
| 751 | - [User, Pass] = lists:filter(fun(Line) -> Line =/= <<>> end, Lines), |
| 752 | - {User, Pass}. |
| 753 | - |
| 754 743 | %% A fetched batch may contain offsets earlier than the |
| 755 744 | %% requested begin-offset because the batch might be compressed on |
| 756 745 | %% kafka side. Here we drop the leading messages. |