anchor.diff:        State
     end.
   
  -send_fetch_request(#state{socket_pid = SocketPid} = State) ->
  -  true = (is_integer(State#state.begin_offset) andalso
  -          State#state.begin_offset >= 0), %% assert
  +send_fetch_request(#state{ begin_offset = BeginOffset
  +                         , socket_pid   = SocketPid
  +                         } = State) ->
  +  (is_integer(BeginOffset) andalso BeginOffset >= 0) orelse
  +    erlang:error({bad_begin_offset, BeginOffset}),
     Request =
       kpro:fetch_request(State#state.topic,
                          State#state.partition,
                          State#state.begin_offset,
                          State#state.max_wait_time,
                          State#state.min_bytes,
  -                       State#state.max_bytes
  -                       ),
  +                       State#state.max_bytes),
     brod_sock:send(SocketPid, Request).
   
   -spec update_options(options(), #state{}) -> {ok, #state{}} | {error, any()}.
  
positive.code: %%%
  %%%   Copyright (c) 2014-2016, Klarna AB
  %%%
  %%%   Licensed under the Apache License, Version 2.0 (the "License");
  
negative.code:                , sleep_timeout  = F(sleep_timeout, State#state.sleep_timeout)
                 , prefetch_count = F(prefetch_count, State#state.prefetch_count)
                 },
    NewState =
      case NewBeginOffset =/= OldBeginOffset of
        true ->
          %% reset buffer in case subscriber wants to fetch from a new offset
          State1#state{pending_acks = []};
        false ->
          State1
      end,
    resolve_begin_offset(NewState).
  
  -spec resolve_begin_offset(#state{}) -> {ok, #state{}} | {error, any()}.
  resolve_begin_offset(#state{ begin_offset = BeginOffset
                             , socket_pid   = SocketPid
                             , topic        = Topic
                             , partition    = Partition
                             } = State) when BeginOffset < 0 ->
    case fetch_valid_offset(SocketPid, BeginOffset, Topic, Partition) of
      {ok, NewBeginOffset} ->
        {ok, State#state{begin_offset = NewBeginOffset}};
      {error, Reason} ->
        {error, Reason}
    end;
  resolve_begin_offset(State) ->
    {ok, State}.
  
  fetch_valid_offset(_S, Offset, _T, _P) when Offset >= 0 ->
    {ok, Offset};
  