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:       ["\n", Topic, ":", format_partition_assignments(Partitions) ]
      end, Groupped).
  
  -spec format_partition_assignments([{brod:partition(), brod:offset()}]) -> iodata().
  format_partition_assignments([]) -> [];
  format_partition_assignments([{Partition, BeginOffset} | Rest]) ->
    [ io_lib:format("~n    partition=~p begin_offset=~p",
  