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:                          | min_bytes
                           | max_bytes
                           | max_wait_time
                           | sleep_timeout.
  
  %% behaviour definition
  -callback init_consumer( Topic     :: topic()
                         , Partition :: partition()
                         , CbArgs    :: any()) ->
    {ok, CbOptions :: [{consumer_option(), any()}], CbState :: any()}.
  
  -callback handle_messages( Topic        :: topic()
                           , Partition    :: partition()
                           , HighWmOffset :: integer()
                           , Messages     :: [#message{}]
                           , CbState      :: any()) ->
    {ok, NewCbState :: any()}.
  
  -record(state, { cb_mod        :: atom()
                 , cb_state      :: any()
                 , client_id     :: client_id()
                 , config        :: consumer_config()
                 , socket_pid    :: pid()
                 , topic         :: binary()
                 , partition     :: integer()
                 , offset        :: integer()
                 , max_wait_time :: integer()
                 , min_bytes     :: integer()
                 , max_bytes     :: integer()
                 , sleep_timeout :: integer()
                 , corr_id       :: corr_id()
                 }).
  
  -define(DEFAULT_OFFSET, -1).
  
negative.code: 
    %% 2. Get options from callback module and merge with Config
    CbArgs = proplists:get_value(cb_args, Config, []),
    {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),
  