anchor.diff:    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, 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),
  
positive.code:     false ->
        {ok, Messages}
    end.
  
  console_consumer(Hosts, Topic, Partition, Offset) ->
    simple_consumer(Hosts, Topic, Partition, Offset, user).
  
  file_consumer(Hosts, Topic, Partition, Offset, Filename) ->
    {ok, File} = file:open(Filename, [write, append]),
    C = simple_consumer(Hosts, Topic, Partition, Offset, File),
    MonitorFun = fun() ->
                     Mref = erlang:monitor(process, C),
                     receive
                       {'DOWN', Mref, process, C, _} ->
                         io:format("~nClosing ~s~n", [Filename]),
                         file:close(File)
                     end
                 end,
    proc_lib:spawn(fun() -> MonitorFun() end),
    C.
  
  simple_consumer(Hosts, Topic, Partition, Offset, Io) ->
    {ok, C} = brod:start_link_consumer(Hosts, Topic, Partition),
    Pid = proc_lib:spawn_link(fun() -> simple_consumer_loop(C, Io) end),
    Callback = fun(MsgOffset, K, V) -> print_message(Io, MsgOffset, K, V) end,
    ok = brod:consume(C, Callback, Offset),
    Pid.
  
  %% escript entry point
  main([]) ->
  
negative.code:              list_to_integer(OffsetStr),
               list_to_integer(MaxWaitTimeStr),
               list_to_integer(MinBytesStr),
               list_to_integer(MaxBytesStr)).
  
  %%%_* Internal functions =======================================================
  
  