anchor.diff:        heartbeat_loop(SendFun)
     end.
   
  -rand_vsn_request_sync(Connection, API, Body) ->
  -  {Min, Max} = get_api_vsn_range(Connection, API),
  -  Vsn = rand(Min, Max),
  +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.
  
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:   kpro:connect_coordinator(Cluster, ConnCfg, Args).
  
  %% NOTE: MemberId is only relevant for join_group request version 4 or later.
  join_group(Connection, GroupId, StaticMemberId, KafkaVsn) ->
    Meta = #{ version => 0
            , topics => [?TOPIC]
            , user_data => <<"magic-1">>
            },
    Send =
      fun(MemberId) ->
        Fields = #{ group_id => GroupId
                  , session_timeout_ms => timer:seconds(10)
                  , rebalance_timeout_ms => timer:seconds(10)
                  , member_id => MemberId
                  , protocol_type => <<"consumer">>
                  , group_instance_id  => StaticMemberId
                  , protocols =>
                      [ #{ name => <<"test-v1">>
                        , metadata => Meta
                        }
                      ]
                  },
        request_sync(Connection, join_group, Fields, KafkaVsn)
      end,
    #{ error_code := ErrorCode
     , member_id := MyMemberId
     } = Rsp0 = Send(<<>>),
    Rsp =
      case ErrorCode of
        no_error ->
          Rsp0;
        member_id_required ->
          %% https://cwiki.apache.org/confluence/display/KAFKA/KIP-394
          %% Since kafka 2.2, members will maybe receive old member ID
          %% with a member_id_required error code
          Send(MyMemberId)
      end,
    #{ error_code := no_error
     , protocol_name := <<"test-v1">>
     , leader := LeaderId
     , member_id := MyMemberId
     , members := [#{ member_id := MemberId
                    , metadata := Meta
                    }]
     } = Rsp,
    %% only member in grou, leader must be self
    ?assertEqual(LeaderId, MyMemberId),
    ?assertEqual(MyMemberId, MemberId),
    Rsp.
  
  sync_group(Connection, GroupId, MemberId, Generation,
             StaticMemberId, KafkaVsn) ->
    MemberAssignment =
      #{ version => 0
       , topic_partitions => [#{ topic => ?TOPIC
  