anchor.diff:    ok.
   
   %%%_* Internal Functions =======================================================
  +init_socket(#state{ client_pid = ClientPid
  +                  , sock_mref  = OldSockMref
  +                  , topic      = Topic
  +                  , partition  = Partition
  +                  } = State) ->
  +  %% 1. Lookup, or maybe (re-)establish a connection to partition leader
  +  case brod_client:get_leader_connection(ClientPid, Topic, Partition) of
  +    {ok, SockPid} ->
  +      ok = maybe_demonitor(OldSockMref),
  +      SockMref = erlang:monitor(process, SockPid),
  +      %% 2. Register self() to client.
  +      ok = brod_client:register_producer(ClientPid, Topic, Partition),
  +
  +      %% 3. Update state.
  +      {ok, State#state{sock_pid = SockPid, sock_mref = SockMref}};
  +    {error, Error} ->
  +      {error, Error}
  +  end.
  +
  +maybe_demonitor(undefined) ->
  +  ok;
  +maybe_demonitor(Mref) ->
  +  true = erlang:demonitor(Mref, [flush]),
  +  ok.
   
   schedule_retry(#state{retry_tref = undefined} = State, Msg) ->
  -  {ok, TRef} = timer:send_after(State#state.retry_timeout, {?RETRY_MSG, Msg}),
  +  {ok, TRef} = timer:send_after(State#state.retry_backoff_ms, {?RETRY_MSG, Msg}),
     {ok, State#state{retry_tref = TRef}};
   schedule_retry(State, _Msg) ->
     %% retry timer has been already activated
  
positive.code: -define(DEFAULT_REQUIRED_ACKS, -1).
  %% by default, leader should wait 1 second for replicas to ack
  -define(DEFAULT_ACK_TIMEOUT, 1000).
  %% by default, brod_producer will sleep for 1 second before trying to send
  %% buffered messages again upon receiving a error from kafka
  -define(DEFAULT_RETRY_TIMEOUT, 1000).
  %% by default, brod_producer will try to re-send buffered messages
  %% max 3 times upon receiving error from kafka
  %% -1 means 'retry indefinitely'
  -define(DEFAULT_MAX_RETRIES, 3).
  %% by default, brod_producer will wait 1 second before trying to
  %% accuire a new connection to kafka broker from brod_client
  %% when an old connection crashed
  -define(DEFAULT_RECONNECT_TIMEOUT, 1000).
  
  -define(INIT_SOCKET_MSG, init_socket).
  -define(RETRY_MSG, retry).
  
  -define(config(Key, Default), proplists:get_value(Key, Config, Default)).
  
negative.code:   Partition = State#state.partition, %% assert
    case brod_kafka:is_error(ErrorCode) of
      true ->
        ErrorDesc = brod_kafka_error:desc(ErrorCode),
        error_logger:error_msg(
          "Error in produce response\n"
          "Topic: ~s\n"
          "Partition: ~B\n"
          "Offset: ~B\n"
          "Error: ~s",
          [Topic, Partition, Offset, ErrorDesc]),
        Error = {produce_response_error, Topic, Partition, Offset, ErrorCode, ErrorDesc},
        case is_retriable(ErrorCode) of
          true ->
            {ok, NewState} = schedule_retry(State, Error),
            {ok, NewBuffer} = brod_producer_buffer:nack(Buffer, CorrId),
            {noreply, NewState#state{buffer = NewBuffer}};
          false ->
            {stop, Error, State}
        end;
      false ->
        {ok, NewBuffer} = brod_producer_buffer:ack(Buffer, CorrId),
        {noreply, State#state{buffer = NewBuffer}}
    end;
  handle_info(_Info, State) ->
    {noreply, State}.
  