anchor.diff:                           false -> OldBeginOffset
                          end,
     NewBeginOffset = F(begin_offset, DefaultBeginOffset),
  -  NewState =
  +  State1 =
       State#state{ begin_offset   = NewBeginOffset
                  , min_bytes      = F(min_bytes, State#state.min_bytes)
                  , max_bytes      = F(max_bytes, State#state.max_bytes)
  
positive.code: %%%
  %%%   Copyright (c) 2014-2016, Klarna AB
  %%%
  %%%   Licensed under the Apache License, Version 2.0 (the "License");
  
negative.code:       State
    end.
  
  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),
    brod_sock:send(SocketPid, Request).
  
  -spec update_options(options(), #state{}) -> {ok, #state{}} | {error, any()}.
  