anchor.diff:  process_events(batch, Msgs, LastBatchOffset, ClientRef, CbModule, CbState) ->
       process_events_batch(Msgs, LastBatchOffset, ClientRef, CbModule, CbState).
   
  -process_events_batch(Msgs, LastBatchOffset, ClientRef, CbModule, CbState) ->
  +process_events_batch(Msgs, LastBatchOffset, ClientRef, CbModule, #state{backoff = Backoff} = CbState) ->
       case catch CbModule:handle_message(Msgs, CbState) of
           {ok, NewCbState} ->
               {Topic, Partition, Offset} = LastBatchOffset,
               ok = erlkaf_nif:consumer_offset_store(ClientRef, Topic, Partition, Offset),
  -            {ok, NewCbState};
  +            {ok, NewCbState#state{backoff = 0}};
           {error, Reason, NewCbState} ->
               ?LOG_ERROR("~p:handle_message for batch error: ~p", [CbModule, Reason]),
               case recv_stop() of
                   false ->
  -                    process_events_batch(Msgs, LastBatchOffset, ClientRef, CbModule, NewCbState);
  +                    process_events_batch(Msgs, LastBatchOffset, ClientRef, CbModule, NewCbState#state{backoff = exponential_backoff(Backoff)});
                   StopMsg ->
                       StopMsg
               end;
  
positive.code:     ?LOG_INFO("stop consumer for: ~p partition: ~p", [TopicName, Partition]),
      ok = erlkaf_nif:consumer_queue_cleanup(Queue),
      From ! {stopped, Tag}.
  
negative.code:             ?LOG_ERROR("~p:handle_message for batch error: ~p", [CbModule, Error]),
              case recv_stop() of
                  false ->
                      process_events_batch(Msgs, LastBatchOffset, ClientRef, CbModule, CbState#state{backoff = exponential_backoff(Backoff)});
                  StopMsg ->
                      StopMsg
              end
      end.
  
  process_events_one_by_one([H|T] = Msgs, ClientRef, CbModule, #state{backoff = Backoff} = CbState) ->
      case recv_stop() of
          false ->
              case catch CbModule:handle_message(H, CbState) of
                  {ok, NewCbState} ->
                      ok = commit_offset(ClientRef, H),
                      process_events_one_by_one(T, ClientRef, CbModule, NewCbState#state{backoff = 0});
                  {error, Reason, NewCbState} ->
                      ?LOG_ERROR("~p:handle_message for: ~p error: ~p", [CbModule, H, Reason]),
                      process_events_one_by_one(Msgs, ClientRef, CbModule, NewCbState#state{backoff = exponential_backoff(Backoff)});
                  Error ->
                      ?LOG_ERROR("~p:handle_message for: ~p error: ~p", [CbModule, H, Error]),
                      process_events_one_by_one(Msgs, ClientRef, CbModule, CbState#state{backoff = exponential_backoff(Backoff)})
              end;
          StopMsg ->
              StopMsg
  