anchor.diff:        %% 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, ErrorCode}, State0};
  +      {stop, {broker_error, brod_kafka_errors:desc(ErrorCode)}, State0};
       false ->
         SleepTimeout = State0#state.sleep_timeout,
         case Messages of
  
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:     proplists:get_value(sleep_timeout, Config, ?DEFAULT_SLEEP_TIMEOUT),
    SleepTimeout =
      proplists:get_value(sleep_timeout, CbOptions, SleepTimeout0),
    PrefetchCount0 =
      proplists:get_value(prefetch_count, Config, ?DEFAULT_PREFETCH_COUNT),
    PrefetchCount =
      proplists:get_value(prefetch_count, CbOptions, PrefetchCount0),
    Offset0 = proplists:get_value(offset, Config, ?DEFAULT_OFFSET),
    Offset1 = proplists:get_value(offset, CbOptions, Offset0),
  
    %% 3. Retrieve valid starting offset
    case get_valid_offset(SocketPid, Offset1, Topic, Partition) of
      {error, Error} ->
        {stop, {error, Error}, State0};
      {ok, Offset} ->
        State = State0#state{ cb_state       = CbState
                            , socket_pid     = SocketPid
                            , offset         = Offset
                            , max_wait_time  = MaxWaitTime
                            , min_bytes      = MinBytes
                            , max_bytes      = MaxBytes
                            , sleep_timeout  = SleepTimeout
                            , prefetch_count = PrefetchCount
                            },
        %% 4. Start consuming
        {ok, CorrId} = send_fetch_request(State),
  