Packages
brod
3.7.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
8
files changed
+120
additions
-56
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.6.1 |
| 4 | + PROJECT_VERSION = 3.7.0 |
| 5 5 | |
| 6 6 | DEPS = supervisor3 kafka_protocol |
| @@ -1,5 +1,5 @@ | |
| 1 1 | {<<"name">>,<<"brod">>}. |
| 2 | - {<<"version">>,<<"3.6.2">>}. |
| 2 | + {<<"version">>,<<"3.7.0">>}. |
| 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.6.2"}, |
| 3 | + {vsn,"3.7.0"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol,supervisor3]}, |
| 6 6 | {env,[]}, |
| @@ -592,7 +592,7 @@ maybe_connect(#state{client_id = ClientId} = State, | |
| 592 592 | true -> |
| 593 593 | connect(State, Endpoint); |
| 594 594 | false -> |
| 595 | - error_logger:error_msg("~p (re)connect to ~s:~p aborted, " |
| 595 | + error_logger:error_msg("~p (re)connect to ~s:~p aborted.\n" |
| 596 596 | "last failure: ~p", |
| 597 597 | [ClientId, Host, Port, Reason]), |
| 598 598 | {{error, Reason}, State} |
| @@ -87,6 +87,7 @@ | |
| 87 87 | , max_bytes :: bytes() |
| 88 88 | , size_stat_window :: non_neg_integer() |
| 89 89 | , prefetch_bytes :: non_neg_integer() |
| 90 | + , connection_mref :: ?undef | reference() |
| 90 91 | }). |
| 91 92 | |
| 92 93 | -type state() :: #state{}. |
| @@ -235,6 +236,7 @@ init({ClientPid, Topic, Partition, Config}) -> | |
| 235 236 | , avg_bytes = 0 |
| 236 237 | , max_bytes = MaxBytes |
| 237 238 | , size_stat_window = Cfg(size_stat_window, ?DEFAULT_AVG_WINDOW) |
| 239 | + , connection_mref = ?undef |
| 238 240 | }}. |
| 239 241 | |
| 240 242 | handle_info(?INIT_CONNECTION, #state{subscriber = Subscriber} = State0) -> |
| @@ -264,10 +266,10 @@ handle_info({'DOWN', _MonitorRef, process, Pid, _Reason}, | |
| 264 266 | }), |
| 265 267 | {noreply, NewState}; |
| 266 268 | handle_info({'DOWN', _MonitorRef, process, Pid, _Reason}, |
| 267 | - #state{connection = Pid} = State) -> |
| 269 | + #state{connection = Pid} = State0) -> |
| 270 | + State = State0#state{connection = ?undef, connection_mref = ?undef}, |
| 268 271 | ok = maybe_send_init_connection(State), |
| 269 | - State1 = State#state{connection = ?undef}, |
| 270 | - {noreply, State1}; |
| 272 | + {noreply, State}; |
| 271 273 | handle_info(Info, State) -> |
| 272 274 | error_logger:warning_msg("~p ~p got unexpected info: ~p", |
| 273 275 | [?MODULE, self(), Info]), |
| @@ -396,21 +398,30 @@ handle_batches(Header, Batches, | |
| 396 398 | , partition = Partition |
| 397 399 | } = State0) -> |
| 398 400 | HighWmOffset = kpro:find(high_watermark, Header), |
| 399 | - Messages = brod_utils:flatten_batches(BeginOffset, Batches), |
| 400 | - MsgSet = #kafka_message_set{ topic = Topic |
| 401 | - , partition = Partition |
| 402 | - , high_wm_offset = HighWmOffset |
| 403 | - , messages = Messages |
| 404 | - }, |
| 405 | - ok = cast_to_subscriber(Subscriber, MsgSet), |
| 406 | - NewPendingAcks = add_pending_acks(PendingAcks, Messages), |
| 407 | - {value, ?PENDING(LastOffset, _LastMsgSize)} = |
| 408 | - queue:peek_r(NewPendingAcks#pending_acks.queue), |
| 409 | - State1 = State0#state{ pending_acks = NewPendingAcks |
| 410 | - , begin_offset = LastOffset + 1 |
| 411 | - }, |
| 412 | - State2 = maybe_shrink_max_bytes(State1, MsgSet#kafka_message_set.messages), |
| 413 | - State = maybe_send_fetch_request(State2), |
| 401 | + State1 = case brod_utils:flatten_batches(BeginOffset, Batches) of |
| 402 | + [] -> |
| 403 | + %% flatten_batches might remove all actual messages |
| 404 | + %% (if they were all before BeginOffset), leaving us |
| 405 | + %% with nothing in this batch. Since there is no way |
| 406 | + %% to know how big the 'hole' is we can only bump |
| 407 | + %% begin_offset with +1 and try again. |
| 408 | + State0#state{ begin_offset = BeginOffset + 1 }; |
| 409 | + Messages -> |
| 410 | + MsgSet = #kafka_message_set{ topic = Topic |
| 411 | + , partition = Partition |
| 412 | + , high_wm_offset = HighWmOffset |
| 413 | + , messages = Messages |
| 414 | + }, |
| 415 | + ok = cast_to_subscriber(Subscriber, MsgSet), |
| 416 | + NewPendingAcks = add_pending_acks(PendingAcks, Messages), |
| 417 | + {value, ?PENDING(LastOffset, _LastMsgSize)} = |
| 418 | + queue:peek_r(NewPendingAcks#pending_acks.queue), |
| 419 | + State2 = State0#state{ pending_acks = NewPendingAcks |
| 420 | + , begin_offset = LastOffset + 1 |
| 421 | + }, |
| 422 | + maybe_shrink_max_bytes(State2, MsgSet#kafka_message_set.messages) |
| 423 | + end, |
| 424 | + State = maybe_send_fetch_request(State1), |
| 414 425 | {noreply, State}. |
| 415 426 | |
| 416 427 | %% Add received offsets to pending queue. |
| @@ -467,20 +478,37 @@ err_op(?request_timed_out) -> retry; | |
| 467 478 | err_op(?unknown_topic_or_partition) -> stop; |
| 468 479 | err_op(?invalid_topic_exception) -> stop; |
| 469 480 | err_op(?offset_out_of_range) -> reset_offset; |
| 481 | + err_op(?leader_not_available) -> reset_connection; |
| 482 | + err_op(?not_leader_for_partition) -> reset_connection; |
| 470 483 | err_op(_) -> restart. |
| 471 484 | |
| 472 485 | handle_fetch_error(#kafka_fetch_error{error_code = ErrorCode} = Error, |
| 473 | - #state{ topic = Topic |
| 474 | - , partition = Partition |
| 475 | - , subscriber = Subscriber |
| 486 | + #state{ topic = Topic |
| 487 | + , partition = Partition |
| 488 | + , subscriber = Subscriber |
| 489 | + , connection_mref = MRef |
| 476 490 | } = State) -> |
| 477 491 | case err_op(ErrorCode) of |
| 492 | + reset_connection -> |
| 493 | + error_logger:info_msg("Fetch error ~p-~p: ~p", |
| 494 | + [Topic, Partition, ErrorCode]), |
| 495 | + %% The current connection in use is not connected to the partition leader, |
| 496 | + %% so we dereference and demonitor the connection pid, but leave it alive, |
| 497 | + %% Can not kill it because it might be shared with other partition workers |
| 498 | + %% Worst case scenario, kafka will close the connection after it |
| 499 | + %% idles for a few minutes. |
| 500 | + is_reference(MRef) andalso erlang:demonitor(MRef), |
| 501 | + NewState = State#state{ connection = ?undef |
| 502 | + , connection_mref = ?undef |
| 503 | + }, |
| 504 | + ok = maybe_send_init_connection(NewState), |
| 505 | + {noreply, NewState}; |
| 478 506 | retry -> |
| 479 507 | {noreply, maybe_send_fetch_request(State)}; |
| 480 508 | stop -> |
| 481 509 | ok = cast_to_subscriber(Subscriber, Error), |
| 482 | - error_logger:error_msg("consumer of topic ~p partition ~p shutdown, " |
| 483 | - "reason: ~p", [Topic, Partition, ErrorCode]), |
| 510 | + error_logger:error_msg("Consumer ~p-~p shutdown\nReason: ~p", |
| 511 | + [Topic, Partition, ErrorCode]), |
| 484 512 | {stop, normal, State}; |
| 485 513 | reset_offset -> |
| 486 514 | handle_reset_offset(State, Error); |
| @@ -700,11 +728,12 @@ maybe_init_connection(#state{ client_pid = ClientPid | |
| 700 728 | %% Lookup, or maybe (re-)establish a connection to partition leader |
| 701 729 | case brod_client:get_leader_connection(ClientPid, Topic, Partition) of |
| 702 730 | {ok, Connection} -> |
| 703 | - _ = erlang:monitor(process, Connection), |
| 731 | + Mref = erlang:monitor(process, Connection), |
| 704 732 | %% Switching to a new connection |
| 705 733 | %% the response for last_req_ref will be lost forever |
| 706 734 | State = State0#state{ last_req_ref = ?undef |
| 707 735 | , connection = Connection |
| 736 | + , connection_mref = Mref |
| 708 737 | }, |
| 709 738 | {ok, State}; |
| 710 739 | {error, Reason} -> |
Loading more files…