anchor.diff:    {ok, _} = brod_sock:send(Socket#socket.pid, Request),
     ok.
   
  -maybe_send_subscriber(_, _, #partition_messages{messages = []}) ->
  -  ok;
  -maybe_send_subscriber(Subscriber, Topic, PM) ->
  +handle_fetch_response(#fetch_response{topics = [TopicFetchData]}, State) ->
  +  #topic_fetch_data{partitions = [PM]} = TopicFetchData,
  +  case has_error(PM) of
  +    {true, Error} ->
  +      {error, Error};
  +    false ->
  +      case PM#partition_messages.messages of
  +        [] ->
  +          {empty, State};
  +        X when is_list(X) ->
  +          LastOffset = PM#partition_messages.last_offset,
  +          Offset = erlang:max(State#state.offset, LastOffset + 1),
  +          {ok, State#state{offset = Offset}}
  +      end
  +  end.
  +
  +send_subscriber(Subscriber, #fetch_response{topics = [TopicFetchData]}) ->
  +  #topic_fetch_data{topic = Topic, partitions = [PM]} = TopicFetchData,
     #partition_messages{ partition = Partition
                        , high_wm_offset = HighWmOffset
                        , messages = Messages} = PM,
  
positive.code:                }).
  
  %%%_* Macros -------------------------------------------------------------------
  
  %%%_* API ----------------------------------------------------------------------
  -spec start_link([{string(), integer()}], binary(), integer()) ->
  
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) ->
  