anchor.diff:  
     %% 2. Get options from callback module and merge with Config
     CbArgs = proplists:get_value(cb_args, Config, []),
  -  {ok, CbOptions, CbState} = CbMod:init_consumer(Topic, Partition, CbArgs),
  +  {ok, CbState, CbOptions} = CbMod:init_consumer(Topic, Partition, CbArgs),
     MinBytes0 = proplists:get_value(min_bytes, Config, ?DEFAULT_MIN_BYTES),
     MinBytes = proplists:get_value(min_bytes, CbOptions, MinBytes0),
     MaxBytes0 = proplists:get_value(max_bytes, Config, ?DEFAULT_MAX_BYTES),
  
positive.code: 
  init({CbMod, ClientId, Topic, Partition, Config}) ->
    self() ! init_socket,
    {ok, #state{ cb_mod    = CbMod
               , client_id = ClientId
               , topic     = Topic
               , partition = Partition
               , config    = Config
               }}.
  
  handle_info(init_socket, #state{ cb_mod    = CbMod
  
negative.code:           {noreply, State0};
          [_|_] ->
            CbMod = State0#state.cb_mod,
            CbState = State0#state.cb_state,
            F = fun() ->
                    Res = CbMod:handle_messages(Topic, Partition,HighWmOffset,
                                                Messages, CbState),
                    exit({shutdown, Res})
                end,
            {_, Ref} = erlang:spawn_monitor(F),
            CbPending = [Ref | State0#state.cb_pending],
            CbPendingCnt = State0#state.cb_pending_cnt + 1,
            State1 = State0#state{ offset = LastOffset + 1
                                 , cb_pending = CbPending
                                 , cb_pending_cnt = CbPendingCnt},
            {ok, State} = maybe_send_fetch_request(State1),
            {noreply, State}
          end
    end;
  %% handle obsolete fetch responses in case we got new offset from callback
  handle_info({msg, Pid, CorrId1, _R},
              #state{corr_id = CorrId2, socket_pid = Pid} = State)
    when CorrId1 < CorrId2 ->
    error_logger:info_msg("~p ~p Dropping obsolete fetch response with corr_id = ~p",
                          [?MODULE, self(), CorrId1]),
    {noreply, State};
  handle_info(?SEND_FETCH_REQUEST, State) ->
    {ok, CorrId} = send_fetch_request(State),
    {noreply, State#state{corr_id = CorrId}};
  handle_info({'DOWN', _MonitorRef, process, Pid, Reason},
              #state{socket_pid = Pid} = State) ->
    {stop, {socket_down, Reason}, State};
  %% callback completed
  handle_info({'DOWN', MRef, process, Pid, Reason},
              #state{cb_pending = CbPending0} = State0) ->
    case lists:member(MRef, CbPending0) of
      true ->
        CbPending = lists:delete(MRef, CbPending0),
        CbPendingCnt = State0#state.cb_pending_cnt - 1,
        case Reason of
          {shutdown, {ok, CbState}} ->
            State = State0#state{ cb_pending = CbPending
                                , cb_pending_cnt = CbPendingCnt
                                , cb_state = CbState},
            {noreply, State};
          {shutdown, {ok, CbState, NewCbOptions}} ->
            MinBytes = proplists:get_value(min_bytes,NewCbOptions,
                                           State0#state.min_bytes),
            MaxBytes = proplists:get_value(max_bytes, NewCbOptions,
                                           State0#state.max_bytes),
            MaxWaitTime = proplists:get_value(max_wait_time, NewCbOptions,
                                              State0#state.max_wait_time),
            SleepTimeout = proplists:get_value(sleep_timeout, NewCbOptions,
                                               State0#state.sleep_timeout),
            PrefetchCount = proplists:get_value(prefetch_count, NewCbOptions,
                                                State0#state.prefetch_count),
            Offset = proplists:get_value(offset, NewCbOptions,
                                         State0#state.offset),
            State = State0#state{ cb_pending = CbPending
                                , cb_pending_cnt = CbPendingCnt
                                , cb_state = CbState
                                , offset = Offset
                                , min_bytes = MinBytes
                                , max_bytes = MaxBytes
                                , max_wait_time = MaxWaitTime
                                , sleep_timeout = SleepTimeout
                                , prefetch_count = PrefetchCount},
            %% TODO: call get_valid_offset? what to do if it fails?
            {noreply, State};
          Other ->
            {stop, {callback_error, Other}, State0}
        end;
      false ->
        error_logger:warning_msg("Unexpected 'DOWN' in ~p ~p from ~p, reason: ~p",
                                 [?MODULE, self(), Pid, Reason]),
        {noreply, State0}
    end;
  handle_info(Info, State) ->
    error_logger:warning_msg("~p [~p] ~p got unexpected info: ~p",
                            [?MODULE, self(), State#state.client_id, Info]),
  