anchor.diff:  -type cb_state() :: term().
   
   %% Initialize the callback module s state.
  --callback init(group_id(), term()) -> {ok, cb_state()}.
  +-callback init(brod:group_id(), term()) -> {ok, cb_state()}.
   
   %% Handle a message. Return one of:
   %%
  
positive.code: %% partition-consumers are fetching more messages behind the scene
  %% unless prefetch_count is set to 0 in consumer config.
  %%
  -callback handle_message(topic(), partition(), #kafka_message{}, cb_state()) ->
              {ok, cb_state()} | {ok, ack, cb_state()}.
  
  %% This callback is called only when subscriber is to commit offsets locally
  
negative.code: %
  % commented out as it's an optional callback
  %-callback assign_partitions([kafka_group_member()],
  %                            [{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 :: {brod:topic(), brod:partition()}
          , consumer_pid    :: ?undef                  %% initial state
                             | pid()                   %% normal state
                             | {down, string(), any()} %% consumer restarting
          , consumer_mref   :: reference()
          , begin_offset    :: brod:offset()
          , acked_offset    :: brod:offset()
          }).
  
  -type ack_ref() :: {brod:topic(), brod:partition(), brod:offset()}.
  
  -record(state,
          { client             :: brod:client()
          , client_mref        :: reference()
          , groupId            :: brod:group_id()
          , memberId           :: brod:member_id()
          , generationId       :: integer()
          , coordinator        :: pid()
          , consumers = []     :: [#consumer{}]
          , consumer_config    :: brod:consumer_config()
          , is_blocked = false :: boolean()
          , cb_module          :: module()
          , cb_state           :: cb_state()
  