anchor.diff:  ensure_integer_offset_time(T) when is_integer(T) -> T.
   
   -spec do_find_leader_in_metadata(kpro_MetadataResponse(),
  -                                 topic(), partition()) -> endpoint().
  +                                 brod:topic(), brod:partition()) -> brod:endpoint().
   do_find_leader_in_metadata(Metadata, Topic, Partition) ->
     #kpro_MetadataResponse{ broker_L        = Brokers
                           , topicMetadata_L = [TopicMetadata]
  
positive.code: %% @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()}.
  find_leader_in_metadata(Metadata, Topic, Partition) ->
    try
      {ok, do_find_leader_in_metadata(Metadata, Topic, Partition)}
  
negative.code: log(error,   Fmt, Args) -> error_logger:error_msg(Fmt, Args).
  
  %% @doc Request (sync) for topic-partition offsets.
  -spec fetch_offsets(pid(), brod:topic(), brod:partition(),
                      brod:offset_time(), pos_integer()) -> {ok, [brod:offset()]}.
  fetch_offsets(SocketPid, Topic, Partition, TimeOrSemanticOffset, NrOfOffsets) ->
    Request = offset_request(Topic, Partition, TimeOrSemanticOffset, NrOfOffsets),
    {ok, Response} = brod_sock:request_sync(SocketPid, Request, 5000),
  