anchor.diff:                       , high_wm_offset = HighWmOffset
                        , last_offset = LastOffset
                        , messages = Messages} = PM,
  +  CbMod = State0#state.cb_mod,
  +  CbState = State0#state.cb_state,
     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};
  +      Error = {ErrorCode, brod_kafka_errors:desc(ErrorCode)},
  +      case CbMod:handle_error(Topic, Partition, Error, CbState) of
  +        {ok, NewCbState} ->
  +          erlang:send_after(?ERROR_COOLDOWN, self(), ?SEND_FETCH_REQUEST),
  +          {noreply, State0#state{cb_state = NewCbState}};
  +        {ok, NewCbState, NewCbOptions} ->
  +          {ok, State} = update_cb_options(NewCbOptions, State0),
  +          erlang:send_after(?ERROR_COOLDOWN, self(), ?SEND_FETCH_REQUEST),
  +          {noreply, State#state{cb_state = NewCbState}};
  +        stop ->
  +          {stop, Error, State0};
  +        Other ->
  +          {stop, {callback_error, Other, Error}, State0}
  +      end;
       false ->
         SleepTimeout = State0#state.sleep_timeout,
         case Messages of
  
positive.code:   {ok, _} = gen:call(Pid, system, {debug, Debug}, infinity),
    ok.
  
  %%%_* Tests ====================================================================
  
  -include_lib("eunit/include/eunit.hrl").
  
negative.code:                          , CbState      :: any()) ->
    {ok, NewCbState :: any()} |
    {ok, NewCbState :: any(), NewCbOptions :: [{consumer_option(), any()}]}.
  
  -callback handle_error( Topic        :: topic()
                        , Partition    :: partition()
                        , Error        :: {atom(), any()}
                        , CbState      :: any()) ->
    {ok, NewCbState :: any()} |
    {ok, NewCbState :: any(), NewCbOptions :: [{consumer_option(), any()}]} |
    stop.
  
  -record(state, { cb_mod         :: atom()
                 , cb_state       :: any()
  