anchor.diff:    assert_reply_sequence(Rest, N-1).
   
   spawn_fake_kafka() ->
  -  erlang:spawn_link(fun() -> fake_kafka_loop(_IsFlushing = false) end).
  +  erlang:spawn_link(fun() -> fake_kafka_loop() end).
   
   stop_fake_kafka(FakeKafka) when is_pid(FakeKafka) ->
     MRef = monitor(process, FakeKafka),
  
positive.code:     erlang:error(timeout)
    end.
  
  fake_kafka_loop(IsFlushing) ->
    receive
      {produce, FromPid, CorrId, KvList} ->
        NewIsFlushing = fake_kafka_process_msgs(IsFlushing, KvList),
        FromPid ! {ack_from_kafka, CorrId},
        fake_kafka_loop(NewIsFlushing);
      stop ->
        exit(normal);
      flush ->
        fake_kafka_loop(true);
      Msg ->
        exit({fake_kafka, unexpected, Msg})
    end.
  
  fake_kafka_process_msgs(IsFlushing, []) -> IsFlushing;
  fake_kafka_process_msgs(IsFlushing, [{_Key, {DelayMs0, _Value}} | Rest]) ->
    DelayMs = case IsFlushing of
                true  -> 0;
                false -> DelayMs0
              end,
    receive
      flush ->
        fake_kafka_process_msgs(true, Rest)
    after DelayMs ->
      fake_kafka_process_msgs(IsFlushing, Rest)
    end.
  
  %%%_* Emacs ====================================================================
  %%% Local Variables:
  
negative.code:   true.
  
  produce_loop(FakeKafka, Buf, [], Buffered, Acked) ->
    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
  