anchor.diff:       %% Check stage:
        fun(Expected, Trace) ->
            check_received_messages(Expected, Trace),
  -         check_state_continuity(Trace)
  +         check_state_continuity(Trace),
  +         check_init_terminate(Trace)
        end).
   
   t_consumer_crash(Config) when is_list(Config) ->
  
positive.code:        %% it may or may not receive the first message (0) depending on when
         %% the consumers starts polling --- before or after the first message
         %% is produced:
         _ = wait_message(<<0>>, 2000),
         %% Send messages:
         Messages = [<<I>> || I <- lists:seq(1, MaxSeqNo)],
         [produce({?topic, 0}, I) || I <- Messages],
         %% Ack messages:
         wait_and_ack(SubscriberPid, Messages),
         ok = brod_topic_subscriber:stop(SubscriberPid),
         Messages
       end,
       %% Check stage:
       fun(Expected, Trace) ->
           Received = ?projection(value, ?of_kind( topic_subscriber_seen_message
                                                 , Trace
                                                 )) -- [<<0>>],
           %% Check that all messages have been received once:
           ?assertEqual(Expected, Received)
       end).
  
  t_begin_offset(Config) when is_list(Config) ->
  
negative.code:      fun(_Ret, Trace) ->
           Expected = [<<I>> || I <- lists:seq(0, 8)],
           check_received_messages(Expected, Trace),
           check_state_continuity(Trace),
           check_init_terminate(Trace)
       end).
  
  %% Test the old way of consuming messages via callback fun:
  