anchor.diff:    io:format(user, "  ./brod fetch Hosts Topic Partition Offset MaxWaitTime MinBytes MaxBytes\n", []),
     io:format(user, "  ./brod fetch kafka-1 topic1 0 -1\n", []),
     io:format(user, "  ./brod fetch kafka-1 topic1 0 -1 1000 0 100000\n", []),
  -  io:format(user, "Start a console consumer:\n", []),
  -  io:format(user, "  ./brod console_consumer Hosts Topic Partition Offset\n", []),
  -  io:format(user, "  ./brod console_consumer kafka-1 topic1 0 -1\n", []),
  -  io:format(user, "Start a file consumer:\n", []),
  -  io:format(user, "  ./brod file_consumer Hosts Topic Partition Offset Filepath\n", []),
  -  io:format(user, "  ./brod file_consumer kafka-1 topic1 0 -1 /tmp/kafka-data\n", []),
     ok.
   
   call_api(get_metadata, [HostsStr]) ->
  
positive.code:                               },
    brod_producer:sync_produce_request(Expect).
  
  %% @deprecated
  %% @equiv start_link_consumer(Hosts, Topic, Partition)
  -spec start_consumer([endpoint()], binary(), integer()) ->
                     {ok, pid()} | {error, any()}.
  start_consumer(Hosts, Topic, Partition) ->
    start_link_consumer(Hosts, Topic, Partition).
  
  %% @equiv start_link_consumer(Hosts, Topic, Partition, 1000)
  -spec start_link_consumer([endpoint()], binary(), integer()) ->
                          {ok, pid()} | {error, any()}.
  start_link_consumer(Hosts, Topic, Partition) ->
    start_link_consumer(Hosts, Topic, Partition, 1000).
  
  %% @doc Start consumer process
  %%      SleepTimeout: how much time to wait before sending next fetch
  %%      request when we got no messages in the last one
  -spec start_link_consumer([endpoint()], binary(), integer(), integer()) ->
                          {ok, pid()} | {error, any()}.
  start_link_consumer(Hosts, Topic, Partition, SleepTimeout) ->
    brod_consumer:start_link(Hosts, Topic, Partition, SleepTimeout).
  
  %% @doc Stop consumer process
  -spec stop_consumer(pid()) -> ok.
  stop_consumer(Pid) ->
    brod_consumer:stop(Pid).
  
  %% @doc Fetch broker metadata
  -spec get_metadata([endpoint()]) -> {ok, #metadata_response{}} | {error, any()}.
  get_metadata(Hosts) ->
  
negative.code:   Pos = string:chr(KVStr, $:),
    Key = iolist_to_binary(string:left(KVStr, Pos - 1)),
    Value = iolist_to_binary(string:right(KVStr, length(KVStr) - Pos)),
    {ok, Client} = brod:start_link_client(Hosts, [{Topic, []}], []),
    {ok, ProducerPid} = brod:get_producer(Client, Topic, Partition),
    Res = brod:produce_sync(ProducerPid, Key, Value),
    brod:stop_client(Client),
  