anchor.diff:    true.
   
   produce_loop(FakeKafka, Buf, [], Buffered, Acked) ->
  -  %% no more to send, block wait for all acks if still pending
  -  try
  -    FakeKafka ! flush
  -  catch _ : _ ->
  -    ok
  -  end,
  -  {NewBuffered, NewAcked, NewBuf} =
  -    collect_replies(Buffered, Acked, Buf, ?MAX_DELAY),
  -  brod_producer_buffer:is_empty(NewBuf) orelse
  -    erlang:error({buffer_not_empty, NewBuf}),
  -  {NewBuffered, NewAcked};
  +  case brod_producer_buffer:is_empty(Buf) of
  +    true ->
  +      {Buffered, Acked};
  +    false ->
  +      {NewBuffered, NewAcked, NewBuf} =
  +        collect_replies(Buffered, Acked, Buf, ?MAX_DELAY),
  +      produce_loop(FakeKafka, NewBuf, [], NewBuffered, NewAcked)
  +  end;
   produce_loop(FakeKafka, Buf0, [{Key, Value} | Rest], Buffered, Acked) ->
     CallRef = #brod_call_ref{ caller = self()
                             , callee = ignore
  
positive.code: 
  random_latency_ack_test_() ->
    {timeout, 60,
     fun() -> ?assert(proper:quickcheck(prop_random_latency_ack_run(), 1000)) end
    }.
  
  %%%_* Help functions ===========================================================
  
  -define(MAX_DELAY, 5).
  
  prop_buffer_limit() -> proper_types:pos_integer().
  prop_onwire_limit() -> proper_types:pos_integer().
  
negative.code:     erlang:error(timeout)
    end.
  
  fake_kafka_loop() ->
    receive
      {produce, FromPid, CorrId, KvList} ->
        ok = fake_kafka_process_msgs(KvList),
        FromPid ! {ack_from_kafka, CorrId},
        fake_kafka_loop();
      stop ->
        exit(normal);
      Msg ->
        exit({fake_kafka, unexpected, Msg})
    end.
  
  fake_kafka_process_msgs([]) -> ok;
  fake_kafka_process_msgs([{_Key, {DelayMs, _Value}} | Rest]) ->
    timer:sleep(DelayMs),
    fake_kafka_process_msgs(Rest).
  
  %%%_* Emacs ====================================================================
  %%% Local Variables:
  