anchor.diff:            {error, {unknown_partition, Topic, Partition}};
           #partition_metadata{leader_id = LeaderId} ->
             Broker = lists:keyfind(LeaderId, #broker_metadata.node_id, Brokers),
  -          #broker_metadata{host = H, port = P} = Broker,
  -          {ok, Pid} = brod_sock:start_link(self(), H, P, State#state.debug),
  -          Socket = #socket{pid = Pid, host = H, port = P, node_id = LeaderId},
  -          Sockets = [Socket | State#state.sockets],
  -          Leaders = [{{Topic, Partition}, Pid} | State#state.leaders],
  -          {ok, Pid, State#state{sockets = Sockets, leaders = Leaders}}
  +          {ok, #broker_metadata{} = Broker}
         end
     end.
   
  
positive.code: -type topic()     :: binary().
  -type partition() :: non_neg_integer().
  -type leader()    :: pid(). % brod_sock pid
  -type corr_id()   :: non_neg_integer().
  -type sender()    :: {pid(), reference()}.
  
  -ifdef(otp_before_17).
  -type data_buffer() :: [{leader(), [{topic(), dict()}]}].
  
negative.code:   {noreply, State}.
  
  handle_info({msg, Leader, CorrId, #produce_response{topics = Topics}}, State0) ->
    Errors = get_error_codes(Topics),
    case handle_produce_response(Leader, CorrId, Errors, State0) of
      {ok, State} ->
        {noreply, State};
      {error, Reason} ->
        %% Errors found in produce response, terminate myself.
        %% In a nicer way, it should recognize at least error code
        %% ?EC_NOT_LEADER_FOR_PARTITION, then try to connect to the
        %% new leader (if not yet connected) and migrate data buffer
        %% and pending list from the old leader to the new leader.
        %% We can not do this now because the payload is not kept in
        %% the pending list for resend, there is no better solution
        %% than crash myself and hope the brod clients would (and they
        %% should) retry after restart.
        {stop, Reason, State0}
    end;
  handle_info({'EXIT', Pid, Reason}, State) ->
    Sockets = lists:keydelete(Pid, #socket.pid, State#state.sockets),
    %% Shutdown because cluster has elected other leaders for
  