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:           {noreply, State0};
          [_|_] ->
            CbMod = State0#state.cb_mod,
            CbState0 = State0#state.cb_state,
            {ok, CbState} = CbMod:handle_messages(Topic, Partition, HighWmOffset,
                                                  Messages, CbState0),
            State = State0#state{cb_state = CbState, offset = LastOffset + 1},
            {ok, NewCorrId} = send_fetch_request(State),
            {noreply, State#state{corr_id = NewCorrId}}
          end
    end;
  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};
  handle_info(Info, State) ->
    error_logger:warning_msg("~p [~p] ~p got unexpected info: ~p",
                            [?MODULE, self(), State#state.client_id, Info]),
  
negative.code: 
  init({CbMod, ClientId, Topic, Partition, Config}) ->
    self() ! init_socket,
    {ok, #state{ cb_mod         = CbMod
               , cb_pending     = []
               , cb_pending_cnt = 0
               , client_id      = ClientId
               , topic          = Topic
               , partition      = Partition
               , config         = Config
               }}.
  
  handle_info(init_socket, #state{ cb_mod    = CbMod
  