anchor.diff:  %
   % commented out as it's an optional callback
   %-callback assign_partitions([kafka_group_member()],
  -%                            [{topic(), partition()}]
  -%                            cb_state()) -> [{kafka_group_member_id(),
  -%                                            [brod_partition_assignment()]}].
  +%                            [{brod:topic(), brod:partition()}]
  +%                            cb_state()) -> [{kafka_group_brod:member_id(),
  +%                                            [brod:brod_partition_assignment()]}].
   
   -define(DOWN(Reason), {down, brod_utils:os_time_utc_str(), Reason}).
   
   -record(consumer,
  -        { topic_partition :: {topic(), partition()}
  +        { topic_partition :: {brod:topic(), brod:partition()}
           , consumer_pid    :: ?undef                  %% initial state
                              | pid()                   %% normal state
                              | {down, string(), any()} %% consumer restarting
           , consumer_mref   :: reference()
  -        , begin_offset    :: offset()
  -        , acked_offset    :: offset()
  +        , begin_offset    :: brod:offset()
  +        , acked_offset    :: brod:offset()
           }).
   
  --type ack_ref() :: {topic(), partition(), offset()}.
  +-type ack_ref() :: {brod:topic(), brod:partition(), brod:offset()}.
   
   -record(state,
  -        { client             :: client()
  +        { client             :: brod:client()
           , client_mref        :: reference()
  -        , groupId            :: group_id()
  -        , memberId           :: member_id()
  +        , groupId            :: brod:group_id()
  +        , memberId           :: brod:member_id()
           , generationId       :: integer()
           , coordinator        :: pid()
           , consumers = []     :: [#consumer{}]
  -        , consumer_config    :: consumer_config()
  +        , consumer_config    :: brod:consumer_config()
           , is_blocked = false :: boolean()
           , cb_module          :: module()
           , cb_state           :: cb_state()
  
positive.code:   gen_server:call(Pid, unsubscribe_all_partitions, infinity).
  
  -spec assign_partitions(pid(), [kpro_GroupMemberMetadata()],
                          [{topic(), partition()}]) ->
                              [{kafka_group_member_id(),
                                [brod_partition_assignment()]}].
  assign_partitions(Pid, MemberMetadataList, TopicPartitionList) ->
    Call = {assign_partitions, MemberMetadataList, TopicPartitionList},
    gen_server:call(Pid, Call, infinity).
  
negative.code: %% disordered acks may overwrite offset commits and lead to unnecessary
  %% message re-delivery in case of restart.
  %% @end
  -spec ack(pid(), brod:topic(), brod:partition(), brod:offset()) -> ok.
  ack(Pid, Topic, Partition, Offset) ->
    gen_server:cast(Pid, {ack, Topic, Partition, Offset}).
  
  