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:           M#kafka_group_member_metadata.topics
          end, Members))).
  
  -spec get_partitions(client(), topic()) -> [partition()].
  get_partitions(Client, Topic) ->
    Count = ?ESCALATE(brod_client:get_partitions_count(Client, Topic)),
    lists:seq(0, Count-1).
  
  -spec do_assign_partitions(roundrobin, [member()],
                             [{topic(), partition()}]) ->
                                [{member_id(), [brod_partition_assignment()]}].
  do_assign_partitions(roundrobin, Members, AllPartitions) ->
    F = fun({MemberId, M}) ->
          SubscribedTopics = M#kafka_group_member_metadata.topics,
  
negative.code:   lists:append(CommittedOffsets0).
  
  -spec resolve_begin_offsets(
          TopicPartitions  :: [{brod:topic(), brod:partition()}],
          CommittedOffsets :: [{{brod:topic(), brod:partition()}, brod:offset()}]) ->
              brod:brod_received_assignments().
  resolve_begin_offsets([], _) -> [];
  resolve_begin_offsets([{Topic, Partition} | Rest], CommittedOffsets) ->
    Offset =
  