anchor.diff:  resolve_begin_offset(State) ->
     {ok, State}.
   
  -fetch_valid_offset(_S, Offset, _T, _P) when Offset >= 0 ->
  -  {ok, Offset};
   fetch_valid_offset(SocketPid, Time, Topic, Partition) ->
     Request = kpro:offset_request(Topic, Partition, Time,
                                         _MaxNoOffsets = 1),
  
positive.code:     suspend ->
        ok = cast_to_subscriber(Subscriber, Error),
        %% Suspend, no more fetch request until the subscriber re-subscribes
        {noreply, State#state{is_suspended = true}};
      restart ->
        ok = cast_to_subscriber(Subscriber, Error),
  
negative.code: %%%_* gen_server callbacks =====================================================
  
  init({ClientPid, Topic, Partition, Config}) ->
    MinBytes = proplists:get_value(min_bytes, Config, ?DEFAULT_MIN_BYTES),
    MaxBytes = proplists:get_value(max_bytes, Config, ?DEFAULT_MAX_BYTES),
    MaxWaitTime =
  