anchor.diff:            %% i.e. the offsets that are ready for commit.
             %% NOTE: this field is not used if offset_commit_policy is
             %% 'consumer_managed'
  -        , acked_offsets = [] :: [{{topic(), partition()}, offset()}]
  +        , acked_offsets = [] :: [{{brod:topic(), brod:partition()}, brod:offset()}]
             %% The referece of the timer which triggers offset commit
           , offset_commit_timer :: reference()
   
  
positive.code: %% @private Fetch committed offsets from kafka,
  %% or call the consumer callback to read committed offsets.
  %% @end
  -spec get_committed_offsets(#state{}, [{topic(), partition()}]) ->
          [{{topic(), partition()}, offset()}].
  get_committed_offsets(#state{ offset_commit_policy = consumer_managed
                              , member_pid           = MemberPid
                              , member_module        = MemberModule
  
negative.code: %%       meaning if two or more offsets for the same topic-partition exist
  %%       in the list, only the one that is closer the head of the list is kept
  %% @end
  -spec commit_offsets(pid(), [{{brod:topic(), brod:partition()}, brod:offset()}]) ->
          ok | {error, any()}.
  commit_offsets(CoordinatorPid, Offsets0) ->
    %% OBS: do not use 'infinity' timeout here.
  