anchor.diff:          , topic             :: topic()
           , partition         :: partition()
           , sock_pid          :: pid()
  +        , sock_mref         :: reference()
           , buffer            :: brod_producer_buffer:buf()
           , retries = 0       :: integer()
           , max_retries       :: integer()
  -        , retry_timeout     :: non_neg_integer()
  +        , retry_backoff_ms  :: non_neg_integer()
           , retry_tref        :: timer:tref()
           , reconnect_timeout :: non_neg_integer()
           }).
  
positive.code:   Partition = State#state.partition, %% assert
    case brod_kafka:is_error(ErrorCode) of
      true ->
        error_logger:error_msg(
          "Error in produce response\n"
          "Topic: ~s\n"
          "Partition: ~B\n"
          "Offset: ~B\n"
          "Error code: ~p",
          [Topic, Partition, Offset, ErrorCode]),
        Error = {produce_response_error, Topic, Partition, Offset, ErrorCode},
        {ok, NewState} = schedule_retry(State, Error),
        {ok, NewBuffer} = brod_producer_buffer:nack(Buffer, CorrId),
        {noreply, NewState#state{buffer = NewBuffer}};
      false ->
        {ok, NewBuffer} = brod_producer_buffer:ack(Buffer, CorrId),
        {noreply, State#state{buffer = NewBuffer}}
    end;
  handle_info({?RETRY_MSG, Msg}, State) ->
    case maybe_retry(State) of
      {ok, NewState} ->
        {noreply, NewState};
      false ->
        Error = io_lib:format("brod_producer ~p failed to recover after ~p retries: ~p",
                              [self(), State#state.retries, Msg]),
        {stop, iolist_to_binary(Error), State}
    end;
  handle_info(_Info, State) ->
    {noreply, State}.
  
negative.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 0.5 second before trying to send
  %% buffered messages again upon receiving a error from kafka
  -define(DEFAULT_RETRY_BACKOFF_MS, 500).
  %% 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).
  
  -define(RETRY_MSG, retry).
  
  -define(config(Key, Default), proplists:get_value(Key, Config, Default)).
  