diff --git a/Cargo.toml b/Cargo.toml
index 2a9028fa..f2b7e096 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -18,6 +18,7 @@ ssl = ["rdkafka/ssl"]
 [dependencies]
 chrono = "0.4.26"
 coarsetime = "0.1.33"
+metrics = "0.24.6"
 once_cell = "1.18.0"
 rand = "0.8.5"
 rdkafka = { version = ">=0.37.0,<0.40", features = ["cmake-build", "tracing"] }
diff --git a/rust-arroyo/src/lib.rs b/rust-arroyo/src/lib.rs
index a8f6e89c..8ddff0d8 100644
--- a/rust-arroyo/src/lib.rs
+++ b/rust-arroyo/src/lib.rs
@@ -4,3 +4,6 @@ pub mod processing;
 pub mod testutils;
 pub mod types;
 pub mod utils;
+
+#[doc(hidden)]
+pub use ::metrics as __metrics;
diff --git a/rust-arroyo/src/metrics/globals.rs b/rust-arroyo/src/metrics/globals.rs
index 19764855..b8861e87 100644
--- a/rust-arroyo/src/metrics/globals.rs
+++ b/rust-arroyo/src/metrics/globals.rs
@@ -1,10 +1,25 @@
-use std::sync::OnceLock;
+#![allow(clippy::mutable_key_type)]
 
-use super::Metric;
+use std::collections::HashMap;
+use std::fmt::Display;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
 
-/// The global [`Recorder`] which will receive [`Metric`] to be recorded.
+use ::metrics::{
+    Counter as MetricsCounter, CounterFn, Gauge as MetricsGauge, GaugeFn,
+    Histogram as MetricsHistogram, HistogramFn, Key, KeyName, Label, Level, Metadata,
+    Recorder as MetricsRecorder, SharedString, Unit,
+};
+use parking_lot::Mutex;
+
+use super::{Metric, MetricType, MetricValue};
+
+/// A compatibility recorder for Arroyo's original metrics API.
+///
+/// New applications should install a [`metrics-rs`](super::facade) recorder
+/// directly. Recorders installed through [`init`] are adapted to that same facade.
 pub trait Recorder {
-    /// Instructs the recorder to record the given [`Metric`].
+    /// Records the given metric.
     fn record_metric(&self, metric: Metric<'_>);
 }
 
@@ -14,31 +29,247 @@ impl<T: Recorder + ?Sized> Recorder for Box<T> {
     }
 }
 
-static GLOBAL_RECORDER: OnceLock<Box<dyn Recorder + Send + Sync + 'static>> = OnceLock::new();
-
-/// Initialize the global [`Recorder`].
+/// Installs a legacy Arroyo recorder as the process-wide `metrics-rs` recorder.
 ///
-/// This will register the given `recorder` as the single global [`Recorder`] instance.
-///
-/// This function can only be called once, and subsequent calls will return an
-/// [`Err`] in case a global [`Recorder`] has already been initialized.
+/// This function can only succeed when no other global `metrics-rs` recorder has
+/// been installed. New applications should install their recorder through their
+/// chosen `metrics-rs` exporter instead.
 pub fn init<R: Recorder + Send + Sync + 'static>(recorder: R) -> Result<(), R> {
-    let mut result = Err(recorder);
-    {
-        let result = &mut result;
-        let _ = GLOBAL_RECORDER.get_or_init(|| {
-            let recorder = std::mem::replace(result, Ok(())).unwrap_err();
-            Box::new(recorder)
-        });
+    match ::metrics::set_global_recorder(LegacyRecorderAdapter::new(recorder)) {
+        Ok(()) => Ok(()),
+        Err(error) => Err(error.into_inner().into_recorder()),
     }
-    result
 }
 
-/// Records a [`Metric`] with the globally configured [`Recorder`].
-///
-/// This function will be a noop in case no global [`Recorder`] is configured.
+/// Records a compatibility [`Metric`] through the active `metrics-rs` recorder.
 pub fn record_metric(metric: Metric<'_>) {
-    if let Some(recorder) = GLOBAL_RECORDER.get() {
-        recorder.record_metric(metric)
+    static METADATA: Metadata<'static> =
+        Metadata::new("sentry_arroyo", Level::INFO, Some(module_path!()));
+
+    let labels = metric
+        .tags
+        .iter()
+        .map(|(key, value)| Label::new((*key).to_owned(), value.to_string()))
+        .collect::<Vec<_>>();
+    let key = Key::from_parts(metric.key.to_string(), labels);
+
+    ::metrics::with_recorder(|recorder| match metric.ty {
+        MetricType::Counter => recorder
+            .register_counter(&key, &METADATA)
+            .increment(metric_counter_value(metric.value)),
+        MetricType::Gauge => recorder
+            .register_gauge(&key, &METADATA)
+            .set(metric_number(metric.value)),
+        MetricType::Timer => recorder
+            .register_histogram(&key, &METADATA)
+            .record(metric_milliseconds(metric.value)),
+    });
+}
+
+pub(crate) fn metric_counter_value(value: MetricValue) -> u64 {
+    match value {
+        MetricValue::I64(value) => value
+            .try_into()
+            .unwrap_or_else(|_| panic!("counter values must be non-negative integers")),
+        MetricValue::U64(value) => value,
+        MetricValue::F64(value) if value.is_finite() && value >= 0.0 && value.fract() == 0.0 => {
+            assert!(value <= u64::MAX as f64, "counter value exceeds u64::MAX");
+            value as u64
+        }
+        MetricValue::F64(_) => panic!("counter values must be non-negative integers"),
+        MetricValue::Duration(value) => value
+            .as_millis()
+            .try_into()
+            .unwrap_or_else(|_| panic!("counter value exceeds u64::MAX")),
+    }
+}
+
+pub(crate) fn metric_number(value: MetricValue) -> f64 {
+    match value {
+        MetricValue::I64(value) => value as f64,
+        MetricValue::U64(value) => value as f64,
+        MetricValue::F64(value) => value,
+        MetricValue::Duration(value) => value.as_millis() as f64,
+    }
+}
+
+pub(crate) fn metric_milliseconds(value: MetricValue) -> f64 {
+    metric_number(value)
+}
+
+struct LegacyRecorderAdapter<R> {
+    recorder: Arc<R>,
+    counters: Mutex<HashMap<Key, Arc<LegacyCounter<R>>>>,
+    gauges: Mutex<HashMap<Key, Arc<LegacyGauge<R>>>>,
+    histograms: Mutex<HashMap<Key, Arc<LegacyHistogram<R>>>>,
+}
+
+impl<R> LegacyRecorderAdapter<R> {
+    fn new(recorder: R) -> Self {
+        Self {
+            recorder: Arc::new(recorder),
+            counters: Mutex::new(HashMap::new()),
+            gauges: Mutex::new(HashMap::new()),
+            histograms: Mutex::new(HashMap::new()),
+        }
+    }
+
+    fn into_recorder(self) -> R {
+        match Arc::try_unwrap(self.recorder) {
+            Ok(recorder) => recorder,
+            Err(_) => unreachable!("a rejected recorder cannot have registered metric handles"),
+        }
+    }
+}
+
+impl<R: Recorder + Send + Sync + 'static> MetricsRecorder for LegacyRecorderAdapter<R> {
+    fn describe_counter(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
+
+    fn describe_gauge(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
+
+    fn describe_histogram(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
+
+    fn register_counter(&self, key: &Key, _: &Metadata<'_>) -> MetricsCounter {
+        let counter = self
+            .counters
+            .lock()
+            .entry(key.to_retained())
+            .or_insert_with(|| {
+                Arc::new(LegacyCounter {
+                    emitter: LegacyEmitter::new(key, Arc::clone(&self.recorder)),
+                    value: AtomicU64::new(0),
+                })
+            })
+            .clone();
+        MetricsCounter::from_arc(counter)
+    }
+
+    fn register_gauge(&self, key: &Key, _: &Metadata<'_>) -> MetricsGauge {
+        let gauge = self
+            .gauges
+            .lock()
+            .entry(key.to_retained())
+            .or_insert_with(|| {
+                Arc::new(LegacyGauge {
+                    emitter: LegacyEmitter::new(key, Arc::clone(&self.recorder)),
+                    value: Mutex::new(0.0),
+                })
+            })
+            .clone();
+        MetricsGauge::from_arc(gauge)
+    }
+
+    fn register_histogram(&self, key: &Key, _: &Metadata<'_>) -> MetricsHistogram {
+        let histogram = self
+            .histograms
+            .lock()
+            .entry(key.to_retained())
+            .or_insert_with(|| {
+                Arc::new(LegacyHistogram {
+                    emitter: LegacyEmitter::new(key, Arc::clone(&self.recorder)),
+                })
+            })
+            .clone();
+        MetricsHistogram::from_arc(histogram)
+    }
+}
+
+struct LegacyEmitter<R> {
+    recorder: Arc<R>,
+    name: String,
+    labels: Vec<(String, String)>,
+}
+
+impl<R: Recorder> LegacyEmitter<R> {
+    fn new(key: &Key, recorder: Arc<R>) -> Self {
+        Self {
+            recorder,
+            name: key.name().to_owned(),
+            labels: key
+                .labels()
+                .map(|label| (label.key().to_owned(), label.value().to_owned()))
+                .collect(),
+        }
+    }
+
+    fn emit(&self, ty: MetricType, value: MetricValue) {
+        let tags = self
+            .labels
+            .iter()
+            .map(|(key, value)| (key.as_str(), value as &dyn Display))
+            .collect::<Vec<_>>();
+
+        self.recorder.record_metric(Metric {
+            key: &self.name,
+            ty,
+            tags: &tags,
+            value,
+            __private: (),
+        });
+    }
+}
+
+struct LegacyCounter<R> {
+    emitter: LegacyEmitter<R>,
+    value: AtomicU64,
+}
+
+impl<R: Recorder> CounterFn for LegacyCounter<R> {
+    fn increment(&self, value: u64) {
+        self.value.fetch_add(value, Ordering::Relaxed);
+        self.emitter
+            .emit(MetricType::Counter, MetricValue::U64(value));
+    }
+
+    fn absolute(&self, value: u64) {
+        let previous = self.value.fetch_max(value, Ordering::Relaxed);
+        if value > previous {
+            self.emitter
+                .emit(MetricType::Counter, MetricValue::U64(value - previous));
+        }
+    }
+}
+
+struct LegacyGauge<R> {
+    emitter: LegacyEmitter<R>,
+    value: Mutex<f64>,
+}
+
+impl<R: Recorder> GaugeFn for LegacyGauge<R> {
+    fn increment(&self, value: f64) {
+        let current = {
+            let mut current = self.value.lock();
+            *current += value;
+            *current
+        };
+        self.emitter
+            .emit(MetricType::Gauge, MetricValue::F64(current));
+    }
+
+    fn decrement(&self, value: f64) {
+        let current = {
+            let mut current = self.value.lock();
+            *current -= value;
+            *current
+        };
+        self.emitter
+            .emit(MetricType::Gauge, MetricValue::F64(current));
+    }
+
+    fn set(&self, value: f64) {
+        *self.value.lock() = value;
+        self.emitter
+            .emit(MetricType::Gauge, MetricValue::F64(value));
+    }
+}
+
+struct LegacyHistogram<R> {
+    emitter: LegacyEmitter<R>,
+}
+
+impl<R: Recorder> HistogramFn for LegacyHistogram<R> {
+    fn record(&self, value: f64) {
+        self.emitter
+            .emit(MetricType::Timer, MetricValue::F64(value));
     }
 }
diff --git a/rust-arroyo/src/metrics/macros.rs b/rust-arroyo/src/metrics/macros.rs
index 9e47500d..48938859 100644
--- a/rust-arroyo/src/metrics/macros.rs
+++ b/rust-arroyo/src/metrics/macros.rs
@@ -1,7 +1,7 @@
 /// Create a [`Metric`].
 ///
 /// Instead of creating metrics directly, it is recommended to immediately record
-/// metrics using the [`counter!`], [`gauge!`] or [`distribution!`] macros.
+/// metrics using the [`counter!`], [`gauge!`] or [`timer!`] macros.
 ///
 /// This is the recommended way to create a [`Metric`], as the
 /// implementation details of it might change.
@@ -26,31 +26,116 @@ macro_rules! metric {
     }};
 }
 
-/// Records a counter [`Metric`](crate::metrics::Metric) with the global [`Recorder`](crate::metrics::Recorder).
+/// Increments a counter using the application's metrics recorder.
 #[macro_export]
 macro_rules! counter {
-    ($expr:expr) => {
-        $crate::__record_metric!(Counter: $expr, 1);
-    };
-    ($($tt:tt)+) => {
-        $crate::__record_metric!(Counter: $($tt)+);
-    };
+    ($key:literal) => {{
+        $crate::__metrics::counter!($key).increment(1);
+    }};
+    ($key:expr) => {{
+        let name = $crate::metrics::metric_name(&$key);
+        $crate::__metrics::counter!(name).increment(1);
+    }};
+    ($key:literal, $value:expr $(,)?) => {{
+        let value = $crate::metrics::counter_value($value);
+        $crate::__metrics::counter!($key).increment(value);
+    }};
+    ($key:expr, $value:expr $(,)?) => {{
+        let name = $crate::metrics::metric_name(&$key);
+        let value = $crate::metrics::counter_value($value);
+        $crate::__metrics::counter!(name).increment(value);
+    }};
+    ($key:literal, $value:expr, $($tag_key:expr => $tag_val:expr),+ $(,)?) => {{
+        let value = $crate::metrics::counter_value($value);
+        let labels = ::std::vec![
+            $($crate::__metrics::Label::new(
+                $tag_key,
+                $crate::metrics::metric_label(&$tag_val),
+            )),+
+        ];
+        $crate::__metrics::counter!($key, labels).increment(value);
+    }};
+    ($key:expr, $value:expr, $($tag_key:expr => $tag_val:expr),+ $(,)?) => {{
+        let name = $crate::metrics::metric_name(&$key);
+        let value = $crate::metrics::counter_value($value);
+        let labels = ::std::vec![
+            $($crate::__metrics::Label::new(
+                $tag_key,
+                $crate::metrics::metric_label(&$tag_val),
+            )),+
+        ];
+        $crate::__metrics::counter!(name, labels).increment(value);
+    }};
 }
 
-/// Records a gauge [`Metric`](crate::metrics::Metric) with the global [`Recorder`](crate::metrics::Recorder).
+/// Sets a gauge using the application's metrics recorder.
 #[macro_export]
 macro_rules! gauge {
-    ($($tt:tt)+) => {
-        $crate::__record_metric!(Gauge: $($tt)+);
-    };
+    ($key:literal, $value:expr $(,)?) => {{
+        let value = $crate::metrics::gauge_value($value);
+        $crate::__metrics::gauge!($key).set(value);
+    }};
+    ($key:expr, $value:expr $(,)?) => {{
+        let name = $crate::metrics::metric_name(&$key);
+        let value = $crate::metrics::gauge_value($value);
+        $crate::__metrics::gauge!(name).set(value);
+    }};
+    ($key:literal, $value:expr, $($tag_key:expr => $tag_val:expr),+ $(,)?) => {{
+        let value = $crate::metrics::gauge_value($value);
+        let labels = ::std::vec![
+            $($crate::__metrics::Label::new(
+                $tag_key,
+                $crate::metrics::metric_label(&$tag_val),
+            )),+
+        ];
+        $crate::__metrics::gauge!($key, labels).set(value);
+    }};
+    ($key:expr, $value:expr, $($tag_key:expr => $tag_val:expr),+ $(,)?) => {{
+        let name = $crate::metrics::metric_name(&$key);
+        let value = $crate::metrics::gauge_value($value);
+        let labels = ::std::vec![
+            $($crate::__metrics::Label::new(
+                $tag_key,
+                $crate::metrics::metric_label(&$tag_val),
+            )),+
+        ];
+        $crate::__metrics::gauge!(name, labels).set(value);
+    }};
 }
 
-/// Records a timer [`Metric`](crate::metrics::Metric) with the global [`Recorder`](crate::metrics::Recorder).
+/// Records a millisecond timer using the application's metrics recorder.
 #[macro_export]
 macro_rules! timer {
-    ($($tt:tt)+) => {
-        $crate::__record_metric!(Timer: $($tt)+);
-    };
+    ($key:literal, $value:expr $(,)?) => {{
+        let value = $crate::metrics::timer_milliseconds($value);
+        $crate::__metrics::histogram!($key).record(value);
+    }};
+    ($key:expr, $value:expr $(,)?) => {{
+        let name = $crate::metrics::metric_name(&$key);
+        let value = $crate::metrics::timer_milliseconds($value);
+        $crate::__metrics::histogram!(name).record(value);
+    }};
+    ($key:literal, $value:expr, $($tag_key:expr => $tag_val:expr),+ $(,)?) => {{
+        let value = $crate::metrics::timer_milliseconds($value);
+        let labels = ::std::vec![
+            $($crate::__metrics::Label::new(
+                $tag_key,
+                $crate::metrics::metric_label(&$tag_val),
+            )),+
+        ];
+        $crate::__metrics::histogram!($key, labels).record(value);
+    }};
+    ($key:expr, $value:expr, $($tag_key:expr => $tag_val:expr),+ $(,)?) => {{
+        let name = $crate::metrics::metric_name(&$key);
+        let value = $crate::metrics::timer_milliseconds($value);
+        let labels = ::std::vec![
+            $($crate::__metrics::Label::new(
+                $tag_key,
+                $crate::metrics::metric_label(&$tag_val),
+            )),+
+        ];
+        $crate::__metrics::histogram!(name, labels).record(value);
+    }};
 }
 
 #[macro_export]
diff --git a/rust-arroyo/src/metrics/mod.rs b/rust-arroyo/src/metrics/mod.rs
index 975426c3..329709a7 100644
--- a/rust-arroyo/src/metrics/mod.rs
+++ b/rust-arroyo/src/metrics/mod.rs
@@ -1,8 +1,43 @@
+//! Metrics emitted by Arroyo.
+//!
+//! Arroyo records metrics through the [`metrics-rs`](facade) facade. Applications
+//! are responsible for installing the recorder/exporter used by the entire process.
+//! The legacy [`Recorder`] and [`StatsdRecorder`] APIs remain available for
+//! compatibility and are adapted to the same facade.
+
 mod globals;
 mod macros;
 mod statsd;
 mod types;
 
+use std::fmt::Display;
+
+pub use ::metrics as facade;
 pub use globals::*;
 pub use statsd::*;
 pub use types::*;
+
+#[doc(hidden)]
+pub fn metric_name(value: &dyn Display) -> String {
+    value.to_string()
+}
+
+#[doc(hidden)]
+pub fn metric_label(value: &dyn Display) -> String {
+    value.to_string()
+}
+
+#[doc(hidden)]
+pub fn counter_value(value: impl Into<MetricValue>) -> u64 {
+    metric_counter_value(value.into())
+}
+
+#[doc(hidden)]
+pub fn gauge_value(value: impl Into<MetricValue>) -> f64 {
+    metric_number(value.into())
+}
+
+#[doc(hidden)]
+pub fn timer_milliseconds(value: impl Into<MetricValue>) -> f64 {
+    metric_milliseconds(value.into())
+}
diff --git a/rust-arroyo/src/metrics/types.rs b/rust-arroyo/src/metrics/types.rs
index 0cae7dbc..e8328126 100644
--- a/rust-arroyo/src/metrics/types.rs
+++ b/rust-arroyo/src/metrics/types.rs
@@ -85,7 +85,7 @@ into_metric_value!(Duration => Duration);
 into_metric_value!(coarsetime::Duration => Duration);
 
 /// An alias for a list of Metric tags.
-pub type MetricTags<'a> = &'a [(&'static str, &'a dyn Display)];
+pub type MetricTags<'a> = &'a [(&'a str, &'a dyn Display)];
 
 /// A fully types Metric.
 ///
diff --git a/tests/legacy_metrics_recorder.rs b/tests/legacy_metrics_recorder.rs
new file mode 100644
index 00000000..83e75094
--- /dev/null
+++ b/tests/legacy_metrics_recorder.rs
@@ -0,0 +1,36 @@
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use sentry_arroyo::metrics::{MetricSink, StatsdRecorder};
+
+#[derive(Clone, Default)]
+struct RecordingSink {
+    metrics: Arc<Mutex<Vec<String>>>,
+}
+
+impl MetricSink for RecordingSink {
+    fn emit(&self, metric: &str) {
+        self.metrics.lock().unwrap().push(metric.to_owned());
+    }
+}
+
+#[test]
+fn legacy_statsd_recorder_is_adapted_to_the_metrics_facade() {
+    let sink = RecordingSink::default();
+    let recorded = Arc::clone(&sink.metrics);
+
+    sentry_arroyo::metrics::init(StatsdRecorder::new("service", sink)).unwrap();
+
+    sentry_arroyo::counter!("arroyo.test.counter", 3, "status" => "ok");
+    sentry_arroyo::gauge!("arroyo.test.gauge", -2);
+    sentry_arroyo::timer!("arroyo.test.timer", Duration::from_micros(1_234_567));
+
+    assert_eq!(
+        *recorded.lock().unwrap(),
+        vec![
+            "service.arroyo.test.counter:3|c|#status:ok",
+            "service.arroyo.test.gauge:-2|g",
+            "service.arroyo.test.timer:1234|ms",
+        ]
+    );
+}
diff --git a/tests/metrics_facade.rs b/tests/metrics_facade.rs
new file mode 100644
index 00000000..9801325b
--- /dev/null
+++ b/tests/metrics_facade.rs
@@ -0,0 +1,173 @@
+use std::collections::BTreeMap;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use metrics::{
+    Counter, CounterFn, Gauge, GaugeFn, Histogram, HistogramFn, Key, KeyName, Metadata, Recorder,
+    SharedString, Unit,
+};
+
+#[derive(Debug, PartialEq)]
+enum MetricValue {
+    Counter(u64),
+    Gauge(f64),
+    Histogram(f64),
+}
+
+#[derive(Debug, PartialEq)]
+struct RecordedMetric {
+    name: String,
+    labels: BTreeMap<String, String>,
+    value: MetricValue,
+}
+
+#[derive(Clone)]
+struct RecordingHandle {
+    name: String,
+    labels: BTreeMap<String, String>,
+    metrics: Arc<Mutex<Vec<RecordedMetric>>>,
+}
+
+impl RecordingHandle {
+    fn new(key: &Key, metrics: Arc<Mutex<Vec<RecordedMetric>>>) -> Self {
+        Self {
+            name: key.name().to_owned(),
+            labels: key
+                .labels()
+                .map(|label| (label.key().to_owned(), label.value().to_owned()))
+                .collect(),
+            metrics,
+        }
+    }
+
+    fn record(&self, value: MetricValue) {
+        self.metrics.lock().unwrap().push(RecordedMetric {
+            name: self.name.clone(),
+            labels: self.labels.clone(),
+            value,
+        });
+    }
+}
+
+impl CounterFn for RecordingHandle {
+    fn increment(&self, value: u64) {
+        self.record(MetricValue::Counter(value));
+    }
+
+    fn absolute(&self, value: u64) {
+        self.record(MetricValue::Counter(value));
+    }
+}
+
+impl GaugeFn for RecordingHandle {
+    fn increment(&self, value: f64) {
+        self.record(MetricValue::Gauge(value));
+    }
+
+    fn decrement(&self, value: f64) {
+        self.record(MetricValue::Gauge(-value));
+    }
+
+    fn set(&self, value: f64) {
+        self.record(MetricValue::Gauge(value));
+    }
+}
+
+impl HistogramFn for RecordingHandle {
+    fn record(&self, value: f64) {
+        self.record(MetricValue::Histogram(value));
+    }
+}
+
+#[derive(Default)]
+struct RecordingRecorder {
+    metrics: Arc<Mutex<Vec<RecordedMetric>>>,
+}
+
+impl Recorder for RecordingRecorder {
+    fn describe_counter(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
+
+    fn describe_gauge(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
+
+    fn describe_histogram(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
+
+    fn register_counter(&self, key: &Key, _: &Metadata<'_>) -> Counter {
+        Counter::from_arc(Arc::new(RecordingHandle::new(
+            key,
+            Arc::clone(&self.metrics),
+        )))
+    }
+
+    fn register_gauge(&self, key: &Key, _: &Metadata<'_>) -> Gauge {
+        Gauge::from_arc(Arc::new(RecordingHandle::new(
+            key,
+            Arc::clone(&self.metrics),
+        )))
+    }
+
+    fn register_histogram(&self, key: &Key, _: &Metadata<'_>) -> Histogram {
+        Histogram::from_arc(Arc::new(RecordingHandle::new(
+            key,
+            Arc::clone(&self.metrics),
+        )))
+    }
+}
+
+#[test]
+fn arroyo_and_application_metrics_use_the_same_metrics_recorder() {
+    let recorder = RecordingRecorder::default();
+    let recorded = Arc::clone(&recorder.metrics);
+
+    metrics::with_local_recorder(&recorder, || {
+        metrics::counter!("application.counter").increment(2);
+
+        sentry_arroyo::counter!(
+            "arroyo.test.counter",
+            3,
+            "status" => "ok",
+            "partition" => 7
+        );
+        sentry_arroyo::gauge!("arroyo.test.gauge", -2, "producer_name" => "producer");
+
+        let timer_name = "arroyo.test.timer".to_owned();
+        sentry_arroyo::timer!(&timer_name, Duration::from_micros(1_234_567));
+
+        sentry_arroyo::metrics::record_metric(sentry_arroyo::metric!(
+            Counter: "legacy.counter", 4
+        ));
+    });
+
+    assert_eq!(
+        *recorded.lock().unwrap(),
+        vec![
+            RecordedMetric {
+                name: "application.counter".to_owned(),
+                labels: BTreeMap::new(),
+                value: MetricValue::Counter(2),
+            },
+            RecordedMetric {
+                name: "arroyo.test.counter".to_owned(),
+                labels: BTreeMap::from([
+                    ("partition".to_owned(), "7".to_owned()),
+                    ("status".to_owned(), "ok".to_owned()),
+                ]),
+                value: MetricValue::Counter(3),
+            },
+            RecordedMetric {
+                name: "arroyo.test.gauge".to_owned(),
+                labels: BTreeMap::from([("producer_name".to_owned(), "producer".to_owned(),)]),
+                value: MetricValue::Gauge(-2.0),
+            },
+            RecordedMetric {
+                name: "arroyo.test.timer".to_owned(),
+                labels: BTreeMap::new(),
+                value: MetricValue::Histogram(1_234.0),
+            },
+            RecordedMetric {
+                name: "legacy.counter".to_owned(),
+                labels: BTreeMap::new(),
+                value: MetricValue::Counter(4),
+            },
+        ]
+    );
+}