Packages
brod
3.7.10
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
+25
additions
-16
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.9 |
| 4 | + PROJECT_VERSION = 3.7.10 |
| 5 5 | |
| 6 6 | DEPS = supervisor3 kafka_protocol |
| @@ -1,5 +1,5 @@ | |
| 1 1 | {<<"name">>,<<"brod">>}. |
| 2 | - {<<"version">>,<<"3.7.9">>}. |
| 2 | + {<<"version">>,<<"3.7.10">>}. |
| 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.9"}, |
| 3 | + {vsn,"3.7.10"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol,supervisor3]}, |
| 6 6 | {env,[]}, |
| @@ -376,18 +376,19 @@ handle_batches(_Header, ?incomplete_batch(Size), | |
| 376 376 | State = maybe_send_fetch_request(State1), |
| 377 377 | {noreply, State}; |
| 378 378 | handle_batches(Header, [], #state{begin_offset = BeginOffset} = State0) -> |
| 379 | - HighWmOffset = kpro:find(high_watermark, Header), |
| 379 | + StableOffset = brod_utils:get_stable_offset(Header), |
| 380 380 | State = |
| 381 | - case BeginOffset < HighWmOffset of |
| 381 | + case BeginOffset < StableOffset of |
| 382 382 | true -> |
| 383 383 | %% There are chances that kafka may return empty message set |
| 384 | - %% when messages are delete from a compacted topic. |
| 384 | + %% when messages are deleted from a compacted topic. |
| 385 385 | %% Since there is no way to know how big the 'hole' is |
| 386 386 | %% we can only bump begin_offset with +1 and try again. |
| 387 387 | State1 = State0#state{begin_offset = BeginOffset + 1}, |
| 388 388 | maybe_send_fetch_request(State1); |
| 389 389 | false -> |
| 390 | - %% we have reached the end of partition |
| 390 | + %% we have either reached the end of a partition |
| 391 | + %% or trying to read uncommitted messages |
| 391 392 | %% try to poll again (maybe after a delay) |
| 392 393 | maybe_delay_fetch_request(State0) |
| 393 394 | end, |
| @@ -399,20 +400,19 @@ handle_batches(Header, Batches, | |
| 399 400 | , topic = Topic |
| 400 401 | , partition = Partition |
| 401 402 | } = State0) -> |
| 402 | - HighWmOffset = kpro:find(high_watermark, Header), |
| 403 | - %% for API version 4 or higher, use last_stable_offset |
| 404 | - LatestOffset = kpro:find(last_stable_offset, Header, HighWmOffset), |
| 403 | + StableOffset = brod_utils:get_stable_offset(Header), |
| 405 404 | {NewBeginOffset, Messages} = |
| 406 405 | brod_utils:flatten_batches(BeginOffset, Header, Batches), |
| 407 406 | State1 = State0#state{begin_offset = NewBeginOffset}, |
| 408 407 | State = |
| 409 408 | case Messages =:= [] of |
| 410 409 | true -> |
| 410 | + %% All messages are before requested offset, hence dropped |
| 411 411 | State1; |
| 412 412 | false -> |
| 413 413 | MsgSet = #kafka_message_set{ topic = Topic |
| 414 414 | , partition = Partition |
| 415 | - , high_wm_offset = LatestOffset |
| 415 | + , high_wm_offset = StableOffset |
| 416 416 | , messages = Messages |
| 417 417 | }, |
| 418 418 | ok = cast_to_subscriber(Subscriber, MsgSet), |
| @@ -31,6 +31,7 @@ | |
| 31 31 | , get_metadata/1 |
| 32 32 | , get_metadata/2 |
| 33 33 | , get_metadata/3 |
| 34 | + , get_stable_offset/1 |
| 34 35 | , group_per_key/1 |
| 35 36 | , group_per_key/2 |
| 36 37 | , init_sasl_opt/1 |
| @@ -322,18 +323,16 @@ fetch(Conn, ReqFun, Offset, MaxBytes) -> | |
| 322 323 | {ok, #{batches := ?incomplete_batch(Size)}} -> |
| 323 324 | fetch(Conn, ReqFun, Offset, Size); |
| 324 325 | {ok, #{header := Header, batches := Batches}} -> |
| 325 | - HighWm = kpro:find(high_watermark, Header), |
| 326 | - %% for API version 4 or higher, use last_stable_offset |
| 327 | - LatestOffset = kpro:find(last_stable_offset, Header, HighWm), |
| 326 | + StableOffset = get_stable_offset(Header), |
| 328 327 | {NewBeginOffset, Msgs} = flatten_batches(Offset, Header, Batches), |
| 329 | - case Offset < LatestOffset andalso Msgs =:= [] of |
| 328 | + case Offset < StableOffset andalso Msgs =:= [] of |
| 330 329 | true -> |
| 331 330 | %% Not reached the latest stable offset yet, |
| 332 331 | %% but received an empty batch-set (all messages are dropped). |
| 333 332 | %% try again with new begin-offset |
| 334 333 | fetch(Conn, ReqFun, NewBeginOffset, MaxBytes); |
| 335 334 | false -> |
| 336 | - {ok, {LatestOffset, Msgs}} |
| 335 | + {ok, {StableOffset, Msgs}} |
| 337 336 | end; |
| 338 337 | {error, Reason} -> |
| 339 338 | {error, Reason} |
| @@ -467,6 +466,16 @@ make_batch_input(Key, Value) -> | |
| 467 466 | false -> [unify_msg(make_msg_input(Key, Value))] |
| 468 467 | end. |
| 469 468 | |
| 469 | + %% @doc last_stable_offset is added in fetch response version 4 |
| 470 | + %% This function takes high watermark offset as last_stable_offset |
| 471 | + %% in case it's missing. |
| 472 | + %% Offsets are considered 'unstable' if they belong to open transactions |
| 473 | + get_stable_offset(Header) -> |
| 474 | + HighWmOffset = kpro:find(high_watermark, Header), |
| 475 | + StableOffset = kpro:find(last_stable_offset, Header, HighWmOffset), |
| 476 | + StableOffset > HighWmOffset andalso error(unexpected_last_stable_offset), |
| 477 | + StableOffset. |
| 478 | + |
| 470 479 | %%%_* Internal functions ======================================================= |
| 471 480 | |
| 472 481 | drop_aborted(#{aborted_transactions := AbortedL}, Batches) -> |