anchor.diff:    end.
   
   %% @doc Reply 'acked' to callers.
  --spec ack(buf(), corr_id()) -> {ok, buf()} | {error, ignored}.
  +-spec ack(buf(), brod:corr_id()) -> {ok, buf()} | {error, ignored}.
   ack(#buf{ onwire_count = OnWireCount
           , onwire       = [{CorrId, Reqs} | Rest]
           } = Buf, CorrId) ->
  
positive.code: %% @doc Buffer a produce request.
  %% Respond to caller immediately if the buffer limit is not yet reached.
  %% @end
  -spec add(buf(), brod_call_ref(), key(), value()) -> {ok, buf()}.
  add(#buf{pending = Pending} = Buf, CallRef, Key, Value) ->
    Req = #req{ call_ref = CallRef
              , data     = fun() -> {Key, Value} end
  
negative.code: %% guarantees the produce responses are replied in the order the corresponding
  %% produce requests were received from clients.
  %% @end
  -spec assert_corr_id([{brod:corr_id(), [#req{}]}], brod:corr_id()) -> true.
  assert_corr_id(_OnWireRequests = [], _CorrIdReceived) ->
    true;
  assert_corr_id([{CorrId, _Req} | _], CorrIdReceived) ->
  