anchor.diff:  %% Discard onwire fetch responses by setting last_corr_id to undefined.
   %% @end
   -spec reset_buffer(#state{}) -> #state{}.
  -reset_buffer(#state{pending_acks = []} = State) ->
  +reset_buffer(#state{pending_acks = #pending_acks{ offsets_queue = ?EMPTY_QUEUE
  +                                                }} = State) ->
     State#state{last_corr_id = ?undef};
  -reset_buffer(#state{pending_acks = [Offset | _]} = State) ->
  -  State#state{ begin_offset = Offset
  -             , pending_acks = []
  +reset_buffer(#state{pending_acks = #pending_acks{ offsets_queue = Queue
  +                                                }} = State) ->
  +  {value, {Begin,_}} = queue:peek(Queue),
  +  State#state{ begin_offset = Begin
  +             , pending_acks = #pending_acks{}
                , last_corr_id = ?undef
                }.
   
  
positive.code:                   reset_to_latest   -> ?OFFSET_LATEST
                  end,
    State1 = State#state{ begin_offset = BeginOffset
                        , pending_acks = []
                        },
    {ok, State2} = resolve_begin_offset(State1),
    NewState = maybe_send_fetch_request(State2),
    {noreply, NewState}.
  
  handle_ack([], _Offset) -> [];
  handle_ack([H | Offsets], Offset) ->
    case H =< Offset of
      true  -> handle_ack(Offsets, Offset);
      false -> [H | Offsets]
    end.
  
  cast_to_subscriber(Pid, Msg) ->
  
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,
  