anchor.diff:  %% @doc Find leader broker ID for the given topic-partiton in
   %% the metadata response received from socket.
   %% @end
  --spec find_leader_in_metadata(kpro_MetadataResponse(), topic(), partition()) ->
  -        {ok, endpoint()} | {error, any()}.
  +-spec find_leader_in_metadata(kpro_MetadataResponse(),
  +                              brod:topic(),
  +                              brod:partition()) ->
  +                                     {ok, brod:endpoint()} | {error, any()}.
   find_leader_in_metadata(Metadata, Topic, Partition) ->
     try
       {ok, do_find_leader_in_metadata(Metadata, Topic, Partition)}
  
positive.code: log(error,   Fmt, Args) -> error_logger:error_msg(Fmt, Args).
  
  %% @doc Request (sync) for topic-partition offsets.
  -spec fetch_offsets(pid(), topic(), partition(),
                      offset_time(), pos_integer()) -> {ok, [offset()]}.
  fetch_offsets(SocketPid, Topic, Partition, TimeOrSemanticOffset, NrOfOffsets) ->
    Request = offset_request(Topic, Partition, TimeOrSemanticOffset, NrOfOffsets),
    {ok, Response} = brod_sock:request_sync(SocketPid, Request, 5000),
  
negative.code: %% are semantic 'offset', this is why often a variable named
  %% Offset is used as the Time argument.
  %% @end
  -spec offset_request(brod:topic(), brod:partition(),
                       brod:offset_time(), pos_integer()) -> kpro_OffsetRequest().
  offset_request(Topic, Partition, TimeOrSemanticOffset, MaxOffsets) ->
    Time = ensure_integer_offset_time(TimeOrSemanticOffset),
    kpro:offset_request(Topic, Partition, Time, MaxOffsets).
  