anchor.diff:                 }).
   
   %%%_* Macros -------------------------------------------------------------------
  +%% how much time to wait before sending next fetch request when we got
  +%% no messages in the last one
  +-define(SLEEP_TIMEOUT, 1000).
  +
  +-define(SEND_FETCH_REQUEST, send_fetch_request).
   
   %%%_* API ----------------------------------------------------------------------
   -spec start_link([{string(), integer()}], binary(), integer()) ->
  
positive.code:   {ok, _} = brod_sock:send(Socket#socket.pid, Request),
    ok.
  
  maybe_send_subscriber(_, _, #partition_messages{messages = []}) ->
    ok;
  maybe_send_subscriber(Subscriber, Topic, PM) ->
    #partition_messages{ partition = Partition
                       , high_wm_offset = HighWmOffset
                       , messages = Messages} = PM,
  
negative.code:   {stop, {unsupported_cast, Msg}, State}.
  
  handle_info({msg, _Pid, _CorrId, #fetch_response{} = R}, State0) ->
    case handle_fetch_response(R, State0) of
      {error, Error} ->
        {stop, {broker_error, Error}, State0};
      {ok, State} ->
        ok = send_subscriber(State#state.subscriber, R),
        ok = send_fetch_request(State),
        {noreply, State};
      {empty, State} ->
        erlang:send_after(?SLEEP_TIMEOUT, self(), ?SEND_FETCH_REQUEST),
        {noreply, State}
    end;
  handle_info(?SEND_FETCH_REQUEST, State) ->
    ok = send_fetch_request(State),
    {noreply, State};
  handle_info({'EXIT', _Pid, Reason}, State) ->
  