Merge pull request #642 from thalesmg/20250725-fix-list-groups

fix: `ListGroups` request V3

Author: thalesmg

.github/workflows/build.yml

@@ -7,7 +7,7 @@ on:
     branches:
       - master
 env:
-  OTP_VERSION: "27"
+  OTP_VERSION: "27.2"
   REBAR_VERSION: "3.24.0"
 
 jobs:
@@ -18,7 +18,7 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v2
       - name: OTP
-        uses: erlef/setup-beam@v1
+        uses: erlef/setup-beam@v1.20.4
         with:
           version-type: strict
           otp-version: ${{ env.OTP_VERSION }}

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+- 4.4.4
+  - Fixed `ListGroups` API request for Kafka Protocol API version 3.
+
 - 4.4.3
   - Modify`brod_client:get_metadata` and `brod_client:get_metadata_safe` function for support multiple topics.
 

src/brod_kafka_request.erl

@@ -134,7 +134,12 @@ offset_fetch(Connection, GroupId, Topics0) ->
 -spec list_groups(conn()) -> kpro:req().
 list_groups(Connection) ->
   Vsn = pick_version(list_groups, Connection),
-  kpro:make_request(list_groups, Vsn, []).
+  case Vsn >= 3 of
+    true ->
+      kpro:make_request(list_groups, Vsn, #{tagged_fields => []});
+    false ->
+      kpro:make_request(list_groups, Vsn, [])
+  end.
 
 %% @doc Make a `join_group' request.
 -spec join_group(conn(), kpro:struct()) -> kpro:req().

test/brod_SUITE.erl

@@ -43,7 +43,7 @@ suite() -> [{timetrap, {minutes, 5}}].
 init_per_suite(Config) ->
   case kafka_test_helper:kafka_version() of
     {0, 9} ->
-      {skip, "no_topic_manaegment_apis"};
+      {skip, "no_topic_management_apis"};
     _ ->
       {ok, _} = application:ensure_all_started(brod),
       Config

test/brod_consumer_SUITE.erl

@@ -46,6 +46,7 @@
         , t_subscribe_with_unknown_offset/1
         , t_offset_reset_policy/1
         , t_stop_kill/1
+        , t_smoke_list_groups/1
         ]).
 
 
@@ -884,6 +885,13 @@ t_stop_kill(Config) when is_list(Config) ->
   ?WAIT_ONLY({'DOWN', Mref, process, Pid, killed}, ok),
   ok = brod_consumer:stop_maybe_kill(Pid, 100).
 
+%% Smoke test for `ListGroups` API.
+t_smoke_list_groups(Config) when is_list(Config) ->
+  [Endpoint] = ?HOSTS,
+  ConnOpts = kafka_test_helper:client_config(),
+  ?assertMatch({ok, Groups} when is_list(Groups), brod:list_groups(Endpoint, ConnOpts)),
+  ok.
+
 %%%_* Help functions ===========================================================
 
 %% os:timestamp should be unique enough for testing