[v23.2.x] metrics: fix kafka_max_offset on read replicas

src/v/cluster/partition_probe.cc

@@ -212,6 +212,14 @@ void replicated_partition_probe::setup_public_metrics(const model::ntp& ntp) {
         sm::make_gauge(
           "max_offset",
           [this] {
+              // TODO: merge code with replicated_partition.h?
+              if (_partition.is_read_replica_mode_enabled()) {
+                  if (_partition.cloud_data_available()) {
+                      return _partition.next_cloud_offset();
+                  }
+                  // This is a read replica and there's no data in the cloud.
+                  return model::offset(0);
+              }
               auto log_offset = _partition.high_watermark();
               auto translator = _partition.get_offset_translator_state();
 

tests/rptest/services/redpanda.py

@@ -954,10 +954,25 @@ def metric_sum(self,
                 for sample in family.samples:
                     if sample.name != metric_name:
                         continue
-                    if ns and sample.labels["namespace"] != ns:
-                        continue
-                    if topic and sample.labels["topic"] != topic:
-                        continue
+                    labels = sample.labels
+                    if ns:
+                        if "redpanda_namespace" in labels:
+                            if labels["redpanda_namespace"] != ns:
+                                continue
+                        elif "namespace" in labels:
+                            if labels["namespace"] != ns:
+                                continue
+                        else:
+                            assert False, f"Missing namespace label: {sample}"
+                    if topic:
+                        if "redpanda_topic" in labels:
+                            if labels["redpanda_topic"] != topic:
+                                continue
+                        elif "topic" in labels:
+                            if labels["topic"] != topic:
+                                continue
+                        else:
+                            assert False, f"Missing topic label: {sample}"
                     count += int(sample.value)
         return count
 

tests/rptest/tests/read_replica_e2e_test.py

@@ -19,7 +19,7 @@
 from ducktape.mark import matrix
 from ducktape.tests.test import TestContext
 
-from rptest.services.redpanda import CloudStorageType, RedpandaService, get_cloud_storage_type, make_redpanda_service
+from rptest.services.redpanda import CloudStorageType, MetricsEndpoint, RedpandaService, get_cloud_storage_type, make_redpanda_service
 from rptest.services.redpanda_installer import InstallOptions, RedpandaInstaller
 from rptest.tests.end_to_end import EndToEndTest
 from rptest.utils.expect_rate import ExpectRate, RateTarget
@@ -238,6 +238,13 @@ def _setup_read_replica(self,
             backoff_sec=5,
             err_msg="Could not create read replica topic. Most likely " +
             "because topic manifest is not in S3.")
+        if num_messages > 0:
+            wait_until(lambda: self.second_cluster.metric_sum(
+                "redpanda_kafka_max_offset",
+                topic=self.topic_name,
+                metrics_endpoint=MetricsEndpoint.PUBLIC_METRICS) > 0,
+                       timeout_sec=10,
+                       backoff_sec=1)
 
     def _bucket_usage(self) -> BucketUsage:
         assert self.redpanda and self.redpanda.cloud_storage_client