anchor.diff:    {_, _, I} = os:timestamp(),
     I.
   
  -rand(Min, Max) ->
  -  rand() rem (Max - Min + 1) + Min.
  -
  -get_api_vsn_range(Connection, API) ->
  -  {ok, KafkaSupportedRange} = kpro:get_api_vsn_range(Connection, API),
  -  kpro_api_vsn:intersect(KafkaSupportedRange, range(API)).
  -
  -%% Max version for consumer group APIs are all before the introduction of
  -%% group_instance_id (see KIP-345). we are not testing it so far.
  -%% TODO: Add tests to support group_instance_id
  -range(join_group) -> {0, 4};
  -range(heartbeat) -> {0, 2};
  -range(leave_group) -> {0, 2};
  -range(sync_group) -> {0, 2};
  -range(describe_groups) -> {0, 4};
  -range(list_groups) -> {0, 100}.
  +kafka_vsns() ->
  +  [v0_9, v0_10, v0_11, v1_0, v2_0, v2_1, v2_2
  +   %%,v2_3, v2_4 % commented out, TODO: test group_instance_id
  +  ].
  +
  +max_vsn(v0_9) -> max_vsn(v0_10);
  +max_vsn(v0_10) ->
  +  #{ join_group => 0
  +   , heartbeat => 0
  +   , leave_group => 0
  +   , sync_group => 0
  +   , describe_groups => 0
  +   , list_groups => 0
  +   };
  +max_vsn(v0_11) -> max_vsn(v1_0);
  +max_vsn(v1_0) -> max_vsn(v1_1);
  +max_vsn(v1_1) ->
  +  #{ join_group => 2
  +   , heartbeat => 1
  +   , leave_group => 1
  +   , sync_group => 1
  +   , describe_groups => 1
  +   , list_groups => 1
  +   };
  +max_vsn(v2_0) -> max_vsn(v2_1);
  +max_vsn(v2_1) ->
  +  #{ join_group => 3
  +   , heartbeat => 2
  +   , leave_group => 2
  +   , sync_group => 2
  +   , describe_groups => 2
  +   , list_groups => 2
  +   };
  +max_vsn(v2_2) ->
  +  #{ join_group => 4
  +   , heartbeat => 2
  +   , leave_group => 2
  +   , sync_group => 2
  +   , describe_groups => 2
  +   , list_groups => 2
  +   };
  +max_vsn(v2_3) ->
  +  #{ join_group => 5
  +   , heartbeat => 3
  +   , leave_group => 2
  +   , sync_group => 3
  +   , describe_groups => 3
  +   , list_groups => 2
  +   };
  +max_vsn(v2_4) ->
  +  #{ join_group => 6
  +   , heartbeat => 4
  +   , leave_group => 4
  +   , sync_group => 4
  +   , describe_groups => 5
  +   , list_groups => 3
  +   }.
   
   %%%_* Emacs ====================================================================
   %%% Local Variables:
  
positive.code:             , assignments => Assignment
              , group_instance_id => StaticMemberId
              },
    Rsp = rand_vsn_request_sync(Connection, sync_group, Fields),
    #{ error_code := no_error
     , assignment := MemberAssignment
     } = Rsp,
    ok.
  
  describe_groups(Connection, GroupId) ->
    Groups = [GroupId, <<"unknown-group">>],
    Body = #{groups => Groups, include_authorized_operations => true},
    Rsp = rand_vsn_request_sync(Connection, describe_groups, Body),
    #{groups := RspGroups} = Rsp,
    lists:foreach(
      fun(#{error_code := ErrorCode}) ->
          ?assertEqual(no_error, ErrorCode)
      end, RspGroups).
  
  list_groups(Connection) ->
    Rsp = rand_vsn_request_sync(Connection, list_groups, #{}),
    ?assertMatch(#{error_code := no_error}, Rsp),
    ok.
  
  leave_group(Connection, GroupId, MemberId) ->
    Fields = #{ group_id => GroupId
              , member_id => MemberId
              },
    Rsp = rand_vsn_request_sync(Connection, leave_group, Fields),
    ?assertMatch(#{error_code := no_error}, Rsp),
    ok.
  
  heartbeat(Connection, GroupId, MemberId, Generation) ->
    {Min, Max} = get_api_vsn_range(Connection, heartbeat),
    SendFun =
      fun() ->
          Vsn = rand(Min, Max),
          Req = kpro:make_request(heartbeat, Vsn,
                                  [ {group_id, GroupId}
                                  , {generation_id, Generation}
                                  , {member_id, MemberId}
                                  ]),
          {ok, #kpro_rsp{msg = Rsp}} =
            kpro:request_sync(Connection, Req, ?TIMEOUT),
          ErrorCode = kpro:find(error_code, Rsp),
  
negative.code:       heartbeat_loop(SendFun)
    end.
  
  request_sync(Connection, API, Body, KafkaVsn) ->
    Vsn = maps:get(API, max_vsn(KafkaVsn)),
    Req = kpro:make_request(API, Vsn, Body),
    {ok, #kpro_rsp{msg = Rsp}} = kpro:request_sync(Connection, Req, ?TIMEOUT),
    Rsp.
  