anchor.diff:    MapFun = fun(#kafka_message{offset = Offset}) -> Offset end,
     Offsets = lists:map(MapFun, Messages),
     LastOffset = lists:last(Offsets),
  -  State = State0#state{ pending_acks = PendingAcks ++ Offsets
  +  NewPendingAcks = handle_add_offset(PendingAcks, Offsets),
  +  State = State0#state{ pending_acks = NewPendingAcks
                         , begin_offset = LastOffset + 1
                         },
     State1 = maybe_shrink_max_bytes(State, MsgSet#kafka_message_set.messages),
     NewState = maybe_send_fetch_request(State1),
     {noreply, NewState}.
  +
  +
  +handle_add_offset(#pending_acks{ offsets_queue = ?EMPTY_QUEUE = Queue
  +                               , count = Count
  +                               } = PendingAcks, [Offset|Offsets]) ->
  +  NewQueue = queue:in({Offset,Offset}, Queue),
  +  handle_add_offset(PendingAcks#pending_acks{ offsets_queue = NewQueue
  +                                            , count = Count + 1
  +                                            }, Offsets);
  +handle_add_offset(#pending_acks{ offsets_queue = Queue
  +                               , count = Count
  +                               } = PendingAcks, [Offset|Offsets]) ->
  +  {{value,{Begin,End}}, Queue1} = queue:out_r(Queue),
  +  case Offset of
  +    _ when Offset == End + 1 ->
  +      NewQueue = queue:in({Begin, Offset}, Queue1),
  +      handle_add_offset(PendingAcks#pending_acks{ offsets_queue = NewQueue
  +                                                , count = Count + 1
  +                                                }, Offsets);
  +    _ when Offset > End + 1 ->
  +      NewQueue = queue:in({Offset, Offset}, Queue),
  +      handle_add_offset(PendingAcks#pending_acks{ offsets_queue = NewQueue
  +                                                , count = Count + 1
  +                                                }, Offsets);
  +    _ ->
  +      handle_add_offset(PendingAcks, Offsets)
  +  end;
  +handle_add_offset(PendingAcks, []) ->
  +  PendingAcks.
   
   maybe_shrink_max_bytes(#state{ prefetch_count = PrefetchCount
                                , max_bytes_orig = MaxBytesOrig
  
positive.code: maybe_send_fetch_request(#state{last_corr_id = I} = State) when is_integer(I) ->
    %% Waiting for the last request
    State;
  maybe_send_fetch_request(#state{ pending_acks   = PendingAcks
                                 , prefetch_count = PrefetchCount
                                 } = State) ->
    case length(PendingAcks) =< PrefetchCount of
      true ->
        case send_fetch_request(State) of
          {ok, CorrId} ->
  
negative.code:     case NewBeginOffset =/= OldBeginOffset of
        true ->
          %% reset buffer in case subscriber wants to fetch from a new offset
          State1#state{pending_acks = #pending_acks{}};
        false ->
          State1
      end,
  