anchor.diff:      proplists:get_value(sleep_timeout, Config, ?DEFAULT_SLEEP_TIMEOUT),
     PrefetchCount =
       proplists:get_value(prefetch_count, Config, ?DEFAULT_PREFETCH_COUNT),
  -  Offset = proplists:get_value(offset, Config, undefined),
  +  Offset = proplists:get_value(begin_offset, Config, undefined),
     {ok, #state{ client_pid     = ClientPid
                , topic          = Topic
                , partition      = Partition
  
positive.code:       State
    end.
  
  send_fetch_request(#state{socket_pid = SocketPid} = State) ->
    true = (is_integer(State#state.begin_offset) andalso
            State#state.begin_offset >= 0), %% assert
    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()}.
  
negative.code: 
  %%%   Copyright (c) 2014-2016, Klarna AB
  %%%
  %%%   Licensed under the Apache License, Version 2.0 (the "License");
  