anchor.diff:              #state{sock_pid = Pid} = State) ->
     {ok, NewState} = schedule_retry(State, Reason),
     {noreply, NewState#state{sock_pid = ?undef}};
  +handle_info({produce, CallRef, Key, Value}, #state{buffer = Buffer} = State) ->
  +  {ok, NewBuffer} = brod_producer_buffer:add(Buffer, CallRef, Key, Value),
  +  State1 = State#state{buffer = NewBuffer},
  +  {ok, NewState} = maybe_produce(State1),
  +  {noreply, NewState};
   handle_info({msg, Pid, CorrId, #kpro_ProduceResponse{} = R},
               #state{ sock_pid = Pid
                     , buffer   = Buffer
  
positive.code: produce(Pid, Key, Value) ->
    CallRef = #brod_call_ref{ caller = self()
                            , callee = Pid
                            , ref    = make_ref()
                            },
    ok = gen_server:cast(Pid, {produce, CallRef, Key, Value}),
    ExpectedReply = #brod_produce_reply{ call_ref = CallRef
                                       , result   = brod_produce_req_buffered
                                       },
    %% Wait until the request is buffered
    case sync_produce_request(ExpectedReply) of
      ok              -> {ok, CallRef};
      {error, Reason} -> {error, Reason}
    end.
  
  %% @doc Block calling process until it receives ExpectedReply.
  
negative.code: handle_call(_Call, _From, State) ->
    {reply, {error, {unsupported_call, _Call}}, State}.
  
  handle_cast(_Cast, State) ->
    {noreply, State}.
  
  