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,13 @@ def test_basic_group_join(self, static_members):
             c.wait()
             c.free()
 
+        viewer = OfflineLogViewer(self.redpanda)
+        for node in self.redpanda.nodes:
+            consumer_offsets_records = viewer.read_consumer_offsets(node=node)
+            self.logger.info(
+                f"Read {len(consumer_offsets_records)} consumer_offset records from node {node.name} successfully"
+            )
+
     @cluster(num_nodes=6)
     def test_mixed_consumers_join(self):
         """