anchor.diff:      case NewBeginOffset =/= OldBeginOffset of
         true ->
           %% reset buffer in case subscriber wants to fetch from a new offset
  -        State1#state{pending_acks = []};
  +        State1#state{pending_acks = #pending_acks{}};
         false ->
           State1
       end,
  
positive.code: -define(SEND_FETCH_REQUEST, send_fetch_request).
  -define(INIT_SOCKET, init_socket).
  -define(DEFAULT_AVG_WINDOW, 5).
  
  %%%_* APIs =====================================================================
  %% @equiv start_link(ClientPid, Topic, Partition, Config, [])
  
negative.code:                , last_corr_id        :: ?undef | corr_id()
                 , subscriber          :: ?undef | pid()
                 , subscriber_mref     :: ?undef | reference()
                 , pending_acks        :: #pending_acks{}
                 , is_suspended        :: boolean()
                 , offset_reset_policy :: offset_reset_policy()
                 , avg_bytes           :: number()
  