Packages
brod
2.2.12
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
4
files changed
+20
additions
-9
deletions
| @@ -1,6 +1,6 @@ | |
| 1 1 | PROJECT = brod |
| 2 2 | PROJECT_DESCRIPTION = Kafka client library in Erlang |
| 3 | - PROJECT_VERSION = 2.2.11 |
| 3 | + PROJECT_VERSION = 2.2.12 |
| 4 4 | |
| 5 5 | DEPS = supervisor3 kafka_protocol |
| @@ -1,5 +1,5 @@ | |
| 1 1 | {<<"name">>,<<"brod">>}. |
| 2 | - {<<"version">>,<<"2.2.11">>}. |
| 2 | + {<<"version">>,<<"2.2.12">>}. |
| 3 3 | {<<"requirements">>, |
| 4 4 | #{<<"kafka_protocol">> => #{<<"app">> => <<"kafka_protocol">>, |
| 5 5 | <<"optional">> => false, |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,brod, |
| 2 2 | [{description,"Apache Kafka Erlang client library"}, |
| 3 | - {vsn,"2.2.11"}, |
| 3 | + {vsn,"2.2.12"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,ssl,kafka_protocol,supervisor3]}, |
| 6 6 | {env,[]}, |
| @@ -65,7 +65,7 @@ | |
| 65 65 | , max_bytes_orig :: bytes() |
| 66 66 | , sleep_timeout :: integer() |
| 67 67 | , prefetch_count :: integer() |
| 68 | - , last_corr_id :: corr_id() |
| 68 | + , last_corr_id :: ?undef | corr_id() |
| 69 69 | , subscriber :: ?undef | pid() |
| 70 70 | , subscriber_mref :: ?undef | reference() |
| 71 71 | , pending_acks = [] :: [offset()] |
| @@ -293,16 +293,18 @@ do_debug(Pid, Debug) -> | |
| 293 293 | ok. |
| 294 294 | |
| 295 295 | handle_fetch_response(_Response, _CorrId, |
| 296 | - #state{subscriber = ?undef} = State) -> |
| 296 | + #state{subscriber = ?undef} = State0) -> |
| 297 297 | %% discard fetch response when there is no (dead?) subscriber |
| 298 | + State = State0#state{last_corr_id = ?undef}, |
| 298 299 | {noreply, State}; |
| 299 300 | handle_fetch_response(_Response, CorrId1, |
| 300 301 | #state{ last_corr_id = CorrId2 |
| 301 302 | } = State) when CorrId1 =/= CorrId2 -> |
| 302 303 | {noreply, State}; |
| 303 304 | handle_fetch_response(#kpro_FetchResponse{ fetchResponseTopic_L = [TopicData] |
| 304 | - }, CorrId, State) -> |
| 305 | - CorrId = State#state.last_corr_id, %% assert |
| 305 | + }, CorrId, State0) -> |
| 306 | + CorrId = State0#state.last_corr_id, %% assert |
| 307 | + State = State0#state{last_corr_id = ?undef}, |
| 306 308 | #kpro_FetchResponseTopic{ topicName = Topic |
| 307 309 | , fetchResponsePartition_L = [PartitionResponse] |
| 308 310 | } = TopicData, |
| @@ -478,6 +480,9 @@ maybe_send_fetch_request(#state{socket_pid = ?undef} = State) -> | |
| 478 480 | maybe_send_fetch_request(#state{is_suspended = true} = State) -> |
| 479 481 | %% waiting for subscriber to re-subscribe |
| 480 482 | State; |
| 483 | + maybe_send_fetch_request(#state{last_corr_id = I} = State) when is_integer(I) -> |
| 484 | + %% Waiting for the last request |
| 485 | + State; |
| 481 486 | maybe_send_fetch_request(#state{ pending_acks = PendingAcks |
| 482 487 | , prefetch_count = PrefetchCount |
| 483 488 | } = State) -> |
| @@ -575,13 +580,15 @@ fetch_valid_offset(SocketPid, BeginOffset, Topic, Partition) -> | |
| 575 580 | |
| 576 581 | %% @private Reset fetch buffer, use the last unacked offset as the next begin |
| 577 582 | %% offset to fetch data from. |
| 583 | + %% Discard onwire fetch responses by setting last_corr_id to undefined. |
| 578 584 | %% @end |
| 579 585 | -spec reset_buffer(#state{}) -> #state{}. |
| 580 586 | reset_buffer(#state{pending_acks = []} = State) -> |
| 581 | - State; |
| 587 | + State#state{last_corr_id = ?undef}; |
| 582 588 | reset_buffer(#state{pending_acks = [Offset | _]} = State) -> |
| 583 589 | State#state{ begin_offset = Offset |
| 584 590 | , pending_acks = [] |
| 591 | + , last_corr_id = ?undef |
| 585 592 | }. |
| 586 593 | |
| 587 594 | %% @private Catch noproc exit exception when making gen_server:call. |
| @@ -608,7 +615,11 @@ maybe_init_socket(#state{ client_pid = ClientPid | |
| 608 615 | case brod_client:get_leader_connection(ClientPid, Topic, Partition) of |
| 609 616 | {ok, SocketPid} -> |
| 610 617 | _ = erlang:monitor(process, SocketPid), |
| 611 | - State = State0#state{socket_pid = SocketPid}, |
| 618 | + %% Switching to a new socket |
| 619 | + %% the response for last_coor_id will be lost forever |
| 620 | + State = State0#state{ last_corr_id = ?undef |
| 621 | + , socket_pid = SocketPid |
| 622 | + }, |
| 612 623 | {ok, State}; |
| 613 624 | {error, Reason} -> |
| 614 625 | {{error, Reason}, State0} |