Fix folding over transactions

The brod:fold/8 did not work correctly when the last message in the
message set a transaction commit.

The output looked like this:
```
%%% brod_consumer_SUITE ==> t_fold_transactions: FAILED
%%% brod_consumer_SUITE ==> {function_clause,
    [{lists,last,[[]],[{file,"lists.erl"},{line,228}]},
     {brod_utils,handle_fetch_rsp,7,
         [{file,"/home/indrek/gems/brod/src/brod_utils.erl"},{line,661}]},
     {brod_consumer_SUITE,t_fold_transactions,1,
         [{file,"/home/indrek/gems/brod/test/brod_consumer_SUITE.erl"},
          {line,443}]},
     {test_server,ts_tc,3,[{file,"test_server.erl"},{line,1783}]},
     {test_server,run_test_case_eval1,6,
         [{file,"test_server.erl"},{line,1292}]},
     {test_server,run_test_case_eval,9,
         [{file,"test_server.erl"},{line,1224}]}]}
```

The issue was that `#kafka_message{offset = LastOffset} = lists:last(Msgs),`
was used, when the Msgs list was empty.

Now instead of trying to infer next offset from the kafka_message, we
pass the NextFetchOffset from the brod_utils:fetch/4 function.

Fixes kafka4beam#588

src/brod_utils.erl

@@ -64,7 +64,7 @@
 -include("brod_int.hrl").
 
 -type req_fun() :: fun((offset(), kpro:count()) -> kpro:req()).
--type fetch_fun() :: fun((offset()) -> {ok, {offset(), [brod:message()]}} |
+-type fetch_fun() :: fun((offset()) -> {ok, {offset(), offset(), [brod:message()]}} |
                                        {error, any()}).
 -type connection() :: kpro:connection().
 -type conn_config() :: brod:conn_config().
@@ -301,7 +301,10 @@ fetch(Client, Topic, Partition, Offset, Opts) when is_atom(Client) ->
   end;
 fetch(Conn, Topic, Partition, Offset, Opts) ->
   Fetch = make_fetch_fun(Conn, Topic, Partition, Opts),
-  Fetch(Offset).
+  case Fetch(Offset) of
+    {ok, {HwOffset, _NextOffset, Batch}} -> {ok, {HwOffset, Batch}}; %% backward compatibility
+    {error, Reason} -> {error, Reason}
+  end.
 
 -spec fold(connection() | brod:client_id() | brod:bootstrap(),
            topic(), partition(), offset(), brod:fetch_opts(),
@@ -443,7 +446,7 @@ do_fetch_committed_offsets(Conn, GroupId, Topics) when is_pid(Conn) ->
 %% @doc Fetch a message-set. If the given MaxBytes is not enough to fetch a
 %% single message, expand it to fetch exactly one message
 -spec fetch(connection(), req_fun(), offset(), kpro:count()) ->
-               {ok, {offset(), [brod:message()]}} | {error, any()}.
+               {ok, {offset(), offset(), [brod:message()]}} | {error, any()}.
 fetch(Conn, ReqFun, Offset, MaxBytes) ->
   Request = ReqFun(Offset, MaxBytes),
   case request_sync(Conn, Request, infinity) of
@@ -474,7 +477,7 @@ fetch(Conn, ReqFun, Offset, MaxBytes) ->
             end,
           fetch(Conn, ReqFun, NewBeginOffset, MaxBytes);
         false ->
-          {ok, {StableOffset, Msgs}}
+          {ok, {StableOffset, NewBeginOffset0, Msgs}}
       end;
     {error, Reason} ->
       {error, Reason}
@@ -636,32 +639,33 @@ do_fold(Spawn, {Pid, Mref}, Offset, Acc, Fun, End, Count) ->
 
 handle_fetch_rsp(_Spawn, {error, Reason}, Offset, Acc, _Fun, _, _) ->
   ?BROD_FOLD_RET(Acc, Offset, {fetch_failure, Reason});
-handle_fetch_rsp(_Spawn, {ok, {StableOffset, []}}, Offset, Acc, _Fun,
+handle_fetch_rsp(_Spawn, {ok, {StableOffset, _NextFetchOffset, []}}, Offset, Acc, _Fun,
                 _End, _Count) when Offset >= StableOffset ->
   ?BROD_FOLD_RET(Acc, Offset, reached_end_of_partition);
-handle_fetch_rsp(Spawn, {ok, {_StableOffset, Msgs}}, Offset, Acc, Fun,
+handle_fetch_rsp(Spawn, {ok, {_StableOffset, NextFetchOffset, Msgs}}, Offset, Acc, Fun,
                  End, Count) ->
-  #kafka_message{offset = LastOffset} = lists:last(Msgs),
-  %% start fetching the next batch if not stopping at current
-  Fetcher = case LastOffset < End andalso length(Msgs) < Count of
-              true -> Spawn(LastOffset + 1);
+  Fetcher = case NextFetchOffset < End andalso length(Msgs) < Count of
+              true -> Spawn(NextFetchOffset);
               false -> undefined
             end,
-  do_acc(Spawn, Fetcher, Offset, Acc, Fun, Msgs, End, Count).
+  do_acc(Spawn, Fetcher, NextFetchOffset, Offset, Acc, Fun, Msgs, End, Count).
 
-do_acc(_Spawn, Fetcher, Offset, Acc, _Fun, _, _End, 0) ->
+do_acc(_Spawn, Fetcher, _NextFetchOffset, Offset, Acc, _Fun, _, _End, 0) ->
   undefined = Fetcher, %% assert
   ?BROD_FOLD_RET(Acc, Offset, reached_message_count_limit);
-do_acc(_Spawn, Fetcher, Offset, Acc, _Fun, _, End, _Count) when Offset > End ->
+do_acc(_Spawn, Fetcher, _NextFetchOffset, Offset, Acc, _Fun, _, End, _Count) when Offset > End  ->
   undefined = Fetcher, %% assert
   ?BROD_FOLD_RET(Acc, Offset, reached_target_offset);
-do_acc(Spawn, Fetcher, Offset, Acc, Fun, [], End, Count) ->
-  do_fold(Spawn, Fetcher, Offset, Acc, Fun, End, Count);
-do_acc(Spawn, Fetcher, Offset, Acc, Fun, [Msg | Rest], End, Count) ->
+do_acc(_Spawn, Fetcher, NextFetchOffset, _Offset, Acc, _Fun, [], End, _Count) when NextFetchOffset > End ->
+  undefined = Fetcher, %% assert
+  ?BROD_FOLD_RET(Acc, NextFetchOffset, reached_target_offset);
+do_acc(Spawn, Fetcher, NextFetchOffset, _Offset, Acc, Fun, [], End, Count) ->
+  do_fold(Spawn, Fetcher, NextFetchOffset, Acc, Fun, End, Count);
+do_acc(Spawn, Fetcher, NextFetchOffset, Offset, Acc, Fun, [Msg | Rest], End, Count) ->
   try Fun(Msg, Acc) of
     {ok, NewAcc} ->
       NextOffset = Msg#kafka_message.offset + 1,
-      do_acc(Spawn, Fetcher, NextOffset, NewAcc, Fun, Rest, End, Count - 1);
+      do_acc(Spawn, Fetcher, NextFetchOffset, NextOffset, NewAcc, Fun, Rest, End, Count - 1);
     {error, Reason} ->
       ok = kill_fetcher(Fetcher),
       ?BROD_FOLD_RET(Acc, Offset, Reason)

test/brod_consumer_SUITE.erl

@@ -32,6 +32,7 @@
         , t_fetch_aborted_from_the_middle/1
         , t_direct_fetch/1
         , t_fold/1
+        , t_fold_transactions/1
         , t_direct_fetch_with_small_max_bytes/1
         , t_direct_fetch_expand_max_bytes/1
         , t_resolve_offset/1
@@ -422,6 +423,26 @@ t_fold(Config) when is_list(Config) ->
               0, ErrorFoldF, #{})),
   ok.
 
+t_fold_transactions(kafka_version_match) ->
+  has_txn();
+t_fold_transactions(Config) when is_list(Config) ->
+  Client = ?config(client),
+  Topic = ?TOPIC,
+  Partition = 0,
+  Batch = [#{value => <<"one">>}, #{value => <<"two">>}],
+  {ok, Tx} = brod:transaction(Client, <<"some_transaction">>, []),
+  {ok, Offset} = brod:txn_produce(Tx, ?TOPIC, Partition, Batch),
+  ok = brod:commit(Tx),
+  FoldF =
+    fun F(#kafka_message{value = V}, Acc) -> {ok, F(V, Acc)};
+        F(V, Acc) -> [V | Acc]
+    end,
+  FetchOpts = #{max_bytes => 1},
+  ?assertMatch({Result, O, reached_end_of_partition}
+                when O =:= Offset + length(Batch) + 1 andalso length(Result) =:= 2,
+    brod:fold(Client, Topic, Partition, Offset, FetchOpts, [], FoldF, #{})),
+  ok.
+
 %% This test case does not work with Kafka 0.9, not sure aobut 0.10 and 0.11
 %% since all 0.x versions are old enough, we only try to verify this against
 %% 1.x or newer