anchor.diff:      From ! {stopped, Tag}.
   
   exponential_backoff(0) ->
  -    1000;
  +    500;
   exponential_backoff(4000) ->
       timer:sleep(4000),
       4000;
  
positive.code:     erlkaf_nif:consumer_offset_store(ClientRef, Topic, Partition, Offset).
  
  process_events(one_by_one, Msgs, _LastBatchOffset, ClientRef, CbModule, CbState) ->
      process_events_one_by_one(Msgs, 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#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#state{backoff = exponential_backoff(Backoff)});
                  StopMsg ->
                      StopMsg
              end;
  
negative.code:     poll_batch_size,
      dispatch_mode,
      messages = [],
      last_offset = -1
  }).
  
  