anchor.diff:    {stop, {unsupported_cast, Msg}, State}.
   
   handle_info({msg, _Pid, _CorrId, #fetch_response{} = R}, State0) ->
  -  #fetch_response{topics = [TopicFetchData]} = R,
  -  #topic_fetch_data{topic = Topic, partitions = [PM]} = TopicFetchData,
  -  case has_error(PM) of
  -    {true, Error} ->
  +  case handle_fetch_response(R, State0) of
  +    {error, Error} ->
         {stop, {broker_error, Error}, State0};
  -    false ->
  -      ok = maybe_send_subscriber(State0#state.subscriber, Topic, PM),
  -      LastOffset = PM#partition_messages.last_offset,
  -      case PM#partition_messages.messages of
  -        [] ->
  -          %% do not bump offset when we get nothing, repeat request later
  -          {noreply, State0, State0#state.max_wait_time};
  -        X when is_list(X) ->
  -          Offset = erlang:max(State0#state.offset, LastOffset + 1),
  -          State = State0#state{offset = Offset},
  -          ok = send_fetch_request(State),
  -          {noreply, State}
  -      end
  +    {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(timeout, State) ->
  +handle_info(?SEND_FETCH_REQUEST, State) ->
     ok = send_fetch_request(State),
     {noreply, State};
   handle_info({'EXIT', _Pid, Reason}, State) ->
  
positive.code:                }).
  
  %%%_* Macros -------------------------------------------------------------------
  
  %%%_* API ----------------------------------------------------------------------
  -spec start_link([{string(), integer()}], binary(), integer()) ->
  
negative.code:   {ok, _} = brod_sock:send(Socket#socket.pid, Request),
    ok.
  
  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,
  