anchor.diff:                                , 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
  +          {ok, State1} = update_cb_options(NewCbOptions, State0),
  +          State = State1#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?
  +                              , cb_state = CbState},
             {noreply, State};
           Other ->
             {stop, {callback_error, Other}, State0}
  
positive.code:                      , high_wm_offset = HighWmOffset
                       , last_offset = LastOffset
                       , messages = Messages} = PM,
    case brod_kafka:is_error(ErrorCode) of
      true ->
        %% TODO: do something smarter
        %% 1) handle what we can handle
        %% 2) call CbMod:handle_error for errors we can not handle
        %%    it can for example bump max_bytes option or skip a message
        %% 3) stop if CbMod:handle_error is not defined
        {stop, {broker_error, brod_kafka_errors:desc(ErrorCode)}, State0};
      false ->
        SleepTimeout = State0#state.sleep_timeout,
        case Messages of
  
negative.code:   {ok, _} = gen:call(Pid, system, {debug, Debug}, infinity),
    ok.
  
  update_cb_options(NewCbOptions, State) ->
    MinBytes = proplists:get_value(min_bytes,NewCbOptions,
                                   State#state.min_bytes),
    MaxBytes = proplists:get_value(max_bytes, NewCbOptions,
                                   State#state.max_bytes),
    MaxWaitTime = proplists:get_value(max_wait_time, NewCbOptions,
                                      State#state.max_wait_time),
    SleepTimeout = proplists:get_value(sleep_timeout, NewCbOptions,
                                       State#state.sleep_timeout),
    PrefetchCount = proplists:get_value(prefetch_count, NewCbOptions,
                                        State#state.prefetch_count),
    Offset = proplists:get_value(offset, NewCbOptions,
                                 State#state.offset),
    {ok, State#state{ offset = Offset
                    , min_bytes = MinBytes
                    , max_bytes = MaxBytes
                    , max_wait_time = MaxWaitTime
                    , sleep_timeout = SleepTimeout
                    , prefetch_count = PrefetchCount}}.
  
  %%%_* Tests ====================================================================
  
  -include_lib("eunit/include/eunit.hrl").
  