tests: added a simple tests that validates consumer offsets cache

Added a test case that validates if `__consumer_offsets` cache is
enabled when cluster property changes.

Signed-off-by: Michał Maślanka <[email protected]>

Author: mmaslankaprv

tests/rptest/tests/consumer_offsets_batch_cache_test.py

@@ -0,0 +1,84 @@
+# Copyright 2025 Redpanda Data, Inc.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.md
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0
+import random
+import time
+
+from ducktape.utils.util import wait_until
+from rptest.services.cluster import cluster
+from rptest.tests.redpanda_test import DefaultClient, RedpandaTest
+
+from rptest.clients.types import TopicSpec
+from rptest.clients.kafka_cli_tools import KafkaCliTools
+from rptest.clients.kcl import KCL
+from confluent_kafka import Consumer, TopicPartition
+import re
+
+
+class ConsumerOffsetsCacheTest(RedpandaTest):
+    def _consumer_offsets_cache_hit_ratio(self):
+        cached_read_metrics = self.redpanda.metrics_sample(
+            "vectorized_storage_log_cached_read_bytes",
+            nodes=self.redpanda.nodes)
+        read_metrics = self.redpanda.metrics_sample(
+            "vectorized_storage_log_read_bytes", nodes=self.redpanda.nodes)
+
+        def _sum_metric_value(metrics):
+            return sum(s.value for s in metrics.samples
+                       if s.labels["topic"] == "__consumer_offsets")
+
+        total_read = _sum_metric_value(read_metrics)
+        cache_read = _sum_metric_value(cached_read_metrics)
+        self.logger.info(
+            f"Read from cache: {cache_read}, total read: {total_read}")
+        self.logger.info(f"Read metrics: {_sum_metric_value(read_metrics)}")
+        return cache_read / total_read if total_read > 0 else 0
+
+    def _commit_random_offsets(self, topic: str, partition: int):
+        # Create a consumer that will commit random offsets without consuming
+        consumer = Consumer({
+            'bootstrap.servers': self.redpanda.brokers(),
+            'group.id': 'test-consumer-group',
+            'enable.auto.commit': False,
+            'auto.offset.reset': 'earliest'
+        })
+
+        # Commit random offsets for the topic partition
+        try:
+            partition = 0
+            for _ in range(100):
+                # Generate a random offset between 0 and 1000
+                random_offset = random.randint(0, 1000)
+
+                # Commit the random offset
+                consumer.commit(offsets=[
+                    TopicPartition(topic, partition, offset=random_offset)
+                ],
+                                asynchronous=False)
+        finally:
+            consumer.close()
+
+    @cluster(num_nodes=3)
+    def test_enabling_consumer_offsets_cache_test(self):
+        topic = TopicSpec(name="test-topic",
+                          partition_count=1,
+                          replication_factor=3)
+
+        DefaultClient(self.redpanda).create_topic(topic)
+        self._commit_random_offsets(topic.name, 0)
+
+        ratio_no_cache = self._consumer_offsets_cache_hit_ratio()
+        assert ratio_no_cache == 0.0, "By default consumer offsets cache should not be enabled"
+        self.redpanda.set_cluster_config(
+            {
+                "consumer_offsets_topic_batch_cache_enabled": True,
+            },
+            expect_restart=True)
+        self._commit_random_offsets(topic.name, 0)
+        hit_ratio = self._consumer_offsets_cache_hit_ratio()
+        assert hit_ratio > 0.2, "Consumer offsets topic should use cache now"