tests: added parsing of consumer offsets topic to one of the test cases

Signed-off-by: Michal Maslanka <michal@redpanda.com>

tests/rptest/clients/offline_log_viewer.py

@@ -27,15 +27,20 @@ def read_kvstore(self, node):
         kvstore_json = node.account.ssh_output(cmd, combine_stderr=False)
         return json.loads(kvstore_json)
 
-    def read_controller(self, node):
-        cmd = self._cmd("--type controller")
-        controller_json = node.account.ssh_output(cmd, combine_stderr=False)
+    def _json_cmd(self, node, suffix):
+        cmd = self._cmd(suffix=suffix)
+        json_out = node.account.ssh_output(cmd, combine_stderr=False)
         try:
-            return json.loads(controller_json)
+            return json.loads(json_out)
         except json.decoder.JSONDecodeError:
             # Log the bad output before re-raising
-            self._redpanda.logger.error(
-                f"Invalid JSON output: {controller_json}")
+            self._redpanda.logger.error(f"Invalid JSON output: {json_out}")
             import time
             time.sleep(3600)
             raise
+
+    def read_controller(self, node):
+        return self._json_cmd(node, "--type controller")
+
+    def read_consumer_offsets(self, node):
+        return self._json_cmd(node, "--type consumer_offsets")

tests/rptest/tests/consumer_group_test.py

@@ -8,6 +8,7 @@
 # the Business Source License, use of this software will be governed
 # by the Apache License, Version 2.0
 
+from rptest.clients.offline_log_viewer import OfflineLogViewer
 from rptest.services.cluster import cluster
 
 from rptest.clients.rpk import RpkException, RpkTool
@@ -29,7 +30,10 @@ def __init__(self, test_ctx, *args, **kwargs):
             num_brokers=3,
             *args,
             # disable leader balancer to make sure that group will not be realoaded because of leadership changes
-            extra_rp_conf={"enable_leader_balancer": False},
+            extra_rp_conf={
+                "enable_leader_balancer": False,
+                "default_topic_replications": 3
+            },
             **kwargs)
 
     def make_consumer_properties(base_properties, instance_id=None):
@@ -151,6 +155,32 @@ def test_basic_group_join(self, static_members):
             c.wait()
             c.free()
 
+        gd = RpkTool(self.redpanda).group_describe(group=group)
+        viewer = OfflineLogViewer(self.redpanda)
+        for node in self.redpanda.nodes:
+            consumer_offsets_partitions = viewer.read_consumer_offsets(
+                node=node)
+            offsets = {}
+            groups = set()
+            for partition in consumer_offsets_partitions:
+                for r in partition['records']:
+                    self.logger.info(f"{r}")
+                    if r['key']['type'] == 'group_metadata':
+                        groups.add(r['key']['group_id'])
+                    elif r['key']['type'] == 'offset_commit':
+                        tp = f"{r['key']['topic']}/{r['key']['partition']}"
+                        if tp not in offsets:
+                            offsets[tp] = -1
+                        offsets[tp] = max(r['value']['committed_offset'],
+                                          offsets[tp])
+
+            assert len(groups) == 1 and group in groups
+            assert all([
+                f"{p.topic}/{p.partition}" in offsets
+                and offsets[f"{p.topic}/{p.partition}"] == p.current_offset
+                for p in gd.partitions
+            ])
+
     @cluster(num_nodes=6)
     def test_mixed_consumers_join(self):
         """