diff --git a/api/src/main/java/io/grpc/DoubleCounterMetricInstrument.java b/api/src/main/java/io/grpc/DoubleCounterMetricInstrument.java
index 52e16476411..6657a6d8d42 100644
--- a/api/src/main/java/io/grpc/DoubleCounterMetricInstrument.java
+++ b/api/src/main/java/io/grpc/DoubleCounterMetricInstrument.java
@@ -23,7 +23,7 @@
  */
 @Internal
 public final class DoubleCounterMetricInstrument extends PartialMetricInstrument {
-  DoubleCounterMetricInstrument(long index, String name, String description, String unit,
+  DoubleCounterMetricInstrument(int index, String name, String description, String unit,
       List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);
   }
diff --git a/api/src/main/java/io/grpc/DoubleHistogramMetricInstrument.java b/api/src/main/java/io/grpc/DoubleHistogramMetricInstrument.java
index e9472e29faa..76d1b284daa 100644
--- a/api/src/main/java/io/grpc/DoubleHistogramMetricInstrument.java
+++ b/api/src/main/java/io/grpc/DoubleHistogramMetricInstrument.java
@@ -25,7 +25,7 @@
 public final class DoubleHistogramMetricInstrument extends PartialMetricInstrument {
   private final List<Double> bucketBoundaries;
 
-  DoubleHistogramMetricInstrument(long index, String name, String description, String unit,
+  DoubleHistogramMetricInstrument(int index, String name, String description, String unit,
       List<Double> bucketBoundaries, List<String> requiredLabelKeys, List<String> optionalLabelKeys,
       boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);
diff --git a/api/src/main/java/io/grpc/LongCounterMetricInstrument.java b/api/src/main/java/io/grpc/LongCounterMetricInstrument.java
index 233fb757ba4..ab29ba59266 100644
--- a/api/src/main/java/io/grpc/LongCounterMetricInstrument.java
+++ b/api/src/main/java/io/grpc/LongCounterMetricInstrument.java
@@ -23,7 +23,7 @@
  */
 @Internal
 public final class LongCounterMetricInstrument extends PartialMetricInstrument {
-  LongCounterMetricInstrument(long index, String name, String description, String unit,
+  LongCounterMetricInstrument(int index, String name, String description, String unit,
       List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);
   }
diff --git a/api/src/main/java/io/grpc/LongGaugeMetricInstrument.java b/api/src/main/java/io/grpc/LongGaugeMetricInstrument.java
index 1a98c490606..c8804e98832 100644
--- a/api/src/main/java/io/grpc/LongGaugeMetricInstrument.java
+++ b/api/src/main/java/io/grpc/LongGaugeMetricInstrument.java
@@ -23,7 +23,7 @@
  */
 @Internal
 public final class LongGaugeMetricInstrument extends PartialMetricInstrument {
-  LongGaugeMetricInstrument(long index, String name, String description, String unit,
+  LongGaugeMetricInstrument(int index, String name, String description, String unit,
       List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);
   }
diff --git a/api/src/main/java/io/grpc/LongHistogramMetricInstrument.java b/api/src/main/java/io/grpc/LongHistogramMetricInstrument.java
index 919e733c85a..87006021f4f 100644
--- a/api/src/main/java/io/grpc/LongHistogramMetricInstrument.java
+++ b/api/src/main/java/io/grpc/LongHistogramMetricInstrument.java
@@ -25,7 +25,7 @@
 public final class LongHistogramMetricInstrument extends PartialMetricInstrument {
   private final List<Long> bucketBoundaries;
 
-  LongHistogramMetricInstrument(long index, String name, String description, String unit,
+  LongHistogramMetricInstrument(int index, String name, String description, String unit,
       List<Long> bucketBoundaries, List<String> requiredLabelKeys, List<String> optionalLabelKeys,
       boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);
diff --git a/api/src/main/java/io/grpc/MetricInstrument.java b/api/src/main/java/io/grpc/MetricInstrument.java
index f4353d651f3..1930319060d 100644
--- a/api/src/main/java/io/grpc/MetricInstrument.java
+++ b/api/src/main/java/io/grpc/MetricInstrument.java
@@ -28,7 +28,7 @@ public interface MetricInstrument {
    *
    * @return the index of the metric instrument.
    */
-  public long getIndex();
+  public int getIndex();
 
   /**
    * Returns the name of the metric.
diff --git a/api/src/main/java/io/grpc/MetricInstrumentRegistry.java b/api/src/main/java/io/grpc/MetricInstrumentRegistry.java
index 0eda6f2d4d0..fd74eba281d 100644
--- a/api/src/main/java/io/grpc/MetricInstrumentRegistry.java
+++ b/api/src/main/java/io/grpc/MetricInstrumentRegistry.java
@@ -16,25 +16,36 @@
 
 package io.grpc;
 
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.CopyOnWriteArraySet;
+import javax.annotation.concurrent.GuardedBy;
 
 /**
  * A registry for globally registered metric instruments.
  */
 @Internal
 public final class MetricInstrumentRegistry {
+  static final int INITIAL_INSTRUMENT_CAPACITY = 5;
   private static MetricInstrumentRegistry instance;
-  private final List<MetricInstrument> metricInstruments;
-  private final Set<String> registeredMetricNames;
+  private final Object lock = new Object();
+  @GuardedBy("lock")
+  private final Set<String> registeredMetricNames = new HashSet<>();
+  @GuardedBy("lock")
+  private MetricInstrument[] metricInstruments =
+      new MetricInstrument[INITIAL_INSTRUMENT_CAPACITY];
+  @GuardedBy("lock")
+  private int nextAvailableMetricIndex;
 
-  private MetricInstrumentRegistry() {
-    this.metricInstruments = new CopyOnWriteArrayList<>();
-    this.registeredMetricNames = new CopyOnWriteArraySet<>();
-  }
+  @VisibleForTesting
+  MetricInstrumentRegistry() {}
 
   /**
    * Returns the default metric instrument registry.
@@ -50,7 +61,10 @@ public static synchronized MetricInstrumentRegistry getDefaultRegistry() {
    * Returns a list of registered metric instruments.
    */
   public List<MetricInstrument> getMetricInstruments() {
-    return Collections.unmodifiableList(metricInstruments);
+    synchronized (lock) {
+      return Collections.unmodifiableList(
+          Arrays.asList(Arrays.copyOfRange(metricInstruments, 0, nextAvailableMetricIndex)));
+    }
   }
 
   /**
@@ -65,20 +79,30 @@ public List<MetricInstrument> getMetricInstruments() {
    * @return the newly created DoubleCounterMetricInstrument
    * @throws IllegalStateException if a metric with the same name already exists
    */
-  // TODO(dnvindhya): Evaluate locks over synchronized methods and update if needed
-  public synchronized DoubleCounterMetricInstrument registerDoubleCounter(String name,
+  public DoubleCounterMetricInstrument registerDoubleCounter(String name,
       String description, String unit, List<String> requiredLabelKeys,
       List<String> optionalLabelKeys, boolean enableByDefault) {
-    if (registeredMetricNames.contains(name)) {
-      throw new IllegalStateException("Metric with name " + name + " already exists");
+    checkArgument(!Strings.isNullOrEmpty(name), "missing metric name");
+    checkNotNull(description, "description");
+    checkNotNull(unit, "unit");
+    checkNotNull(requiredLabelKeys, "requiredLabelKeys");
+    checkNotNull(optionalLabelKeys, "optionalLabelKeys");
+    synchronized (lock) {
+      if (registeredMetricNames.contains(name)) {
+        throw new IllegalStateException("Metric with name " + name + " already exists");
+      }
+      int index = nextAvailableMetricIndex;
+      if (index + 1 == metricInstruments.length) {
+        resizeMetricInstruments();
+      }
+      DoubleCounterMetricInstrument instrument = new DoubleCounterMetricInstrument(
+          index, name, description, unit, requiredLabelKeys, optionalLabelKeys,
+          enableByDefault);
+      metricInstruments[index] = instrument;
+      registeredMetricNames.add(name);
+      nextAvailableMetricIndex += 1;
+      return instrument;
     }
-    long instrumentIndex = metricInstruments.size();
-    DoubleCounterMetricInstrument instrument = new DoubleCounterMetricInstrument(
-        instrumentIndex, name, description, unit, requiredLabelKeys, optionalLabelKeys,
-        enableByDefault);
-    metricInstruments.add(instrument);
-    registeredMetricNames.add(name);
-    return instrument;
   }
 
   /**
@@ -93,20 +117,30 @@ public synchronized DoubleCounterMetricInstrument registerDoubleCounter(String n
    * @return the newly created LongCounterMetricInstrument
    * @throws IllegalStateException if a metric with the same name already exists
    */
-  public synchronized LongCounterMetricInstrument registerLongCounter(String name,
+  public LongCounterMetricInstrument registerLongCounter(String name,
       String description, String unit, List<String> requiredLabelKeys,
       List<String> optionalLabelKeys, boolean enableByDefault) {
-    if (registeredMetricNames.contains(name)) {
-      throw new IllegalStateException("Metric with name " + name + " already exists");
+    checkArgument(!Strings.isNullOrEmpty(name), "missing metric name");
+    checkNotNull(description, "description");
+    checkNotNull(unit, "unit");
+    checkNotNull(requiredLabelKeys, "requiredLabelKeys");
+    checkNotNull(optionalLabelKeys, "optionalLabelKeys");
+    synchronized (lock) {
+      if (registeredMetricNames.contains(name)) {
+        throw new IllegalStateException("Metric with name " + name + " already exists");
+      }
+      int index = nextAvailableMetricIndex;
+      if (index + 1 == metricInstruments.length) {
+        resizeMetricInstruments();
+      }
+      LongCounterMetricInstrument instrument = new LongCounterMetricInstrument(
+          index, name, description, unit, requiredLabelKeys, optionalLabelKeys,
+          enableByDefault);
+      metricInstruments[index] = instrument;
+      registeredMetricNames.add(name);
+      nextAvailableMetricIndex += 1;
+      return instrument;
     }
-    // Acquire lock?
-    long instrumentIndex = metricInstruments.size();
-    LongCounterMetricInstrument instrument = new LongCounterMetricInstrument(
-        instrumentIndex, name, description, unit, requiredLabelKeys, optionalLabelKeys,
-        enableByDefault);
-    metricInstruments.add(instrument);
-    registeredMetricNames.add(name);
-    return instrument;
   }
 
   /**
@@ -122,20 +156,32 @@ public synchronized LongCounterMetricInstrument registerLongCounter(String name,
    * @return the newly created DoubleHistogramMetricInstrument
    * @throws IllegalStateException if a metric with the same name already exists
    */
-  public synchronized DoubleHistogramMetricInstrument registerDoubleHistogram(String name,
+  public DoubleHistogramMetricInstrument registerDoubleHistogram(String name,
       String description, String unit, List<Double> bucketBoundaries,
       List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean enableByDefault) {
-    if (registeredMetricNames.contains(name)) {
-      throw new IllegalStateException("Metric with name " + name + " already exists");
+    checkArgument(!Strings.isNullOrEmpty(name), "missing metric name");
+    checkNotNull(description, "description");
+    checkNotNull(unit, "unit");
+    checkNotNull(bucketBoundaries, "bucketBoundaries");
+    checkNotNull(requiredLabelKeys, "requiredLabelKeys");
+    checkNotNull(optionalLabelKeys, "optionalLabelKeys");
+    synchronized (lock) {
+      if (registeredMetricNames.contains(name)) {
+        throw new IllegalStateException("Metric with name " + name + " already exists");
+      }
+      int index = nextAvailableMetricIndex;
+      if (index + 1 == metricInstruments.length) {
+        resizeMetricInstruments();
+      }
+      DoubleHistogramMetricInstrument instrument = new DoubleHistogramMetricInstrument(
+          index, name, description, unit, bucketBoundaries, requiredLabelKeys,
+          optionalLabelKeys,
+          enableByDefault);
+      metricInstruments[index] = instrument;
+      registeredMetricNames.add(name);
+      nextAvailableMetricIndex += 1;
+      return instrument;
     }
-    long indexToInsertInstrument = metricInstruments.size();
-    DoubleHistogramMetricInstrument instrument = new DoubleHistogramMetricInstrument(
-        indexToInsertInstrument, name, description, unit, bucketBoundaries, requiredLabelKeys,
-        optionalLabelKeys,
-        enableByDefault);
-    metricInstruments.add(instrument);
-    registeredMetricNames.add(name);
-    return instrument;
   }
 
   /**
@@ -151,20 +197,32 @@ public synchronized DoubleHistogramMetricInstrument registerDoubleHistogram(Stri
    * @return the newly created LongHistogramMetricInstrument
    * @throws IllegalStateException if a metric with the same name already exists
    */
-  public synchronized LongHistogramMetricInstrument registerLongHistogram(String name,
+  public LongHistogramMetricInstrument registerLongHistogram(String name,
       String description, String unit, List<Long> bucketBoundaries, List<String> requiredLabelKeys,
       List<String> optionalLabelKeys, boolean enableByDefault) {
-    if (registeredMetricNames.contains(name)) {
-      throw new IllegalStateException("Metric with name " + name + " already exists");
+    checkArgument(!Strings.isNullOrEmpty(name), "missing metric name");
+    checkNotNull(description, "description");
+    checkNotNull(unit, "unit");
+    checkNotNull(bucketBoundaries, "bucketBoundaries");
+    checkNotNull(requiredLabelKeys, "requiredLabelKeys");
+    checkNotNull(optionalLabelKeys, "optionalLabelKeys");
+    synchronized (lock) {
+      if (registeredMetricNames.contains(name)) {
+        throw new IllegalStateException("Metric with name " + name + " already exists");
+      }
+      int index = nextAvailableMetricIndex;
+      if (index + 1 == metricInstruments.length) {
+        resizeMetricInstruments();
+      }
+      LongHistogramMetricInstrument instrument = new LongHistogramMetricInstrument(
+          index, name, description, unit, bucketBoundaries, requiredLabelKeys,
+          optionalLabelKeys,
+          enableByDefault);
+      metricInstruments[index] = instrument;
+      registeredMetricNames.add(name);
+      nextAvailableMetricIndex += 1;
+      return instrument;
     }
-    long indexToInsertInstrument = metricInstruments.size();
-    LongHistogramMetricInstrument instrument = new LongHistogramMetricInstrument(
-        indexToInsertInstrument, name, description, unit, bucketBoundaries, requiredLabelKeys,
-        optionalLabelKeys,
-        enableByDefault);
-    metricInstruments.add(instrument);
-    registeredMetricNames.add(name);
-    return instrument;
   }
 
 
@@ -180,18 +238,38 @@ public synchronized LongHistogramMetricInstrument registerLongHistogram(String n
    * @return the newly created LongGaugeMetricInstrument
    * @throws IllegalStateException if a metric with the same name already exists
    */
-  public synchronized LongGaugeMetricInstrument registerLongGauge(String name, String description,
+  public LongGaugeMetricInstrument registerLongGauge(String name, String description,
       String unit, List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean
       enableByDefault) {
-    if (registeredMetricNames.contains(name)) {
-      throw new IllegalStateException("Metric with name " + name + " already exists");
+    checkArgument(!Strings.isNullOrEmpty(name), "missing metric name");
+    checkNotNull(description, "description");
+    checkNotNull(unit, "unit");
+    checkNotNull(requiredLabelKeys, "requiredLabelKeys");
+    checkNotNull(optionalLabelKeys, "optionalLabelKeys");
+    synchronized (lock) {
+      if (registeredMetricNames.contains(name)) {
+        throw new IllegalStateException("Metric with name " + name + " already exists");
+      }
+      int index = nextAvailableMetricIndex;
+      if (index + 1 == metricInstruments.length) {
+        resizeMetricInstruments();
+      }
+      LongGaugeMetricInstrument instrument = new LongGaugeMetricInstrument(
+          index, name, description, unit, requiredLabelKeys, optionalLabelKeys,
+          enableByDefault);
+      metricInstruments[index] = instrument;
+      registeredMetricNames.add(name);
+      nextAvailableMetricIndex += 1;
+      return instrument;
     }
-    long indexToInsertInstrument = metricInstruments.size();
-    LongGaugeMetricInstrument instrument = new LongGaugeMetricInstrument(
-        indexToInsertInstrument, name, description, unit, requiredLabelKeys, optionalLabelKeys,
-        enableByDefault);
-    metricInstruments.add(instrument);
-    registeredMetricNames.add(name);
-    return instrument;
+  }
+
+  @GuardedBy("lock")
+  private void resizeMetricInstruments() {
+    // Increase the capacity of the metricInstruments array by INITIAL_INSTRUMENT_CAPACITY
+    int newInstrumentsCapacity = metricInstruments.length + INITIAL_INSTRUMENT_CAPACITY;
+    MetricInstrument[] resizedMetricInstruments = Arrays.copyOf(metricInstruments,
+        newInstrumentsCapacity);
+    metricInstruments = resizedMetricInstruments;
   }
 }
diff --git a/api/src/main/java/io/grpc/MetricRecorder.java b/api/src/main/java/io/grpc/MetricRecorder.java
index bcf237f9328..24968b30854 100644
--- a/api/src/main/java/io/grpc/MetricRecorder.java
+++ b/api/src/main/java/io/grpc/MetricRecorder.java
@@ -16,11 +16,6 @@
 
 package io.grpc;
 
-import io.grpc.DoubleCounterMetricInstrument;
-import io.grpc.DoubleHistogramMetricInstrument;
-import io.grpc.Internal;
-import io.grpc.LongCounterMetricInstrument;
-import io.grpc.LongHistogramMetricInstrument;
 import java.util.List;
 
 /**
@@ -30,25 +25,25 @@
 @Internal
 public interface MetricRecorder {
   /**
-   * Records a value for a double-precision counter metric instrument.
+   * Adds a value for a double-precision counter metric instrument.
    *
-   * @param metricInstrument The counter metric instrument to record the value against.
-   * @param value The value to record.
+   * @param metricInstrument The counter metric instrument to add the value against.
+   * @param value The value to add.
    * @param requiredLabelValues A list of required label values for the metric.
    * @param optionalLabelValues A list of additional, optional label values for the metric.
    */
-  default void recordDoubleCounter(DoubleCounterMetricInstrument metricInstrument, double value,
+  default void addDoubleCounter(DoubleCounterMetricInstrument metricInstrument, double value,
       List<String> requiredLabelValues, List<String> optionalLabelValues) {}
 
   /**
-   * Records a value for a long valued counter metric instrument.
+   * Adds a value for a long valued counter metric instrument.
    *
-   * @param metricInstrument The counter metric instrument to record the value against.
-   * @param value The value to record.
+   * @param metricInstrument The counter metric instrument to add the value against.
+   * @param value The value to add.
    * @param requiredLabelValues A list of required label values for the metric.
    * @param optionalLabelValues A list of additional, optional label values for the metric.
    */
-  default void recordLongCounter(LongCounterMetricInstrument metricInstrument, long value,
+  default void addLongCounter(LongCounterMetricInstrument metricInstrument, long value,
       List<String> requiredLabelValues, List<String> optionalLabelValues) {}
 
   /**
diff --git a/api/src/main/java/io/grpc/MetricSink.java b/api/src/main/java/io/grpc/MetricSink.java
index c6d8aaa644c..c4b7b192596 100644
--- a/api/src/main/java/io/grpc/MetricSink.java
+++ b/api/src/main/java/io/grpc/MetricSink.java
@@ -98,4 +98,6 @@ default void recordDoubleHistogram(DoubleHistogramMetricInstrument metricInstrum
   default void recordLongHistogram(LongHistogramMetricInstrument metricInstrument, long value,
       List<String> requiredLabelValues, List<String> optionalLabelValues) {
   }
+
+  default void updateMeasures(List<MetricInstrument> instruments) {}
 }
diff --git a/api/src/main/java/io/grpc/PartialMetricInstrument.java b/api/src/main/java/io/grpc/PartialMetricInstrument.java
index 5cfecf3499c..c246b67f810 100644
--- a/api/src/main/java/io/grpc/PartialMetricInstrument.java
+++ b/api/src/main/java/io/grpc/PartialMetricInstrument.java
@@ -25,7 +25,7 @@
  */
 @Internal
 abstract class PartialMetricInstrument implements MetricInstrument {
-  protected final long index;
+  protected final int index;
   protected final String name;
   protected final String description;
   protected final String unit;
@@ -44,7 +44,7 @@ abstract class PartialMetricInstrument implements MetricInstrument {
    * @param optionalLabelKeys a list of optional label keys for the metric
    * @param enableByDefault   whether the metric should be enabled by default
    */
-  protected PartialMetricInstrument(long index, String name, String description, String unit,
+  protected PartialMetricInstrument(int index, String name, String description, String unit,
       List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean enableByDefault) {
     this.index = index;
     this.name = name;
@@ -56,7 +56,7 @@ protected PartialMetricInstrument(long index, String name, String description, S
   }
 
   @Override
-  public long getIndex() {
+  public int getIndex() {
     return index;
   }
 
diff --git a/api/src/test/java/io/grpc/MetricInstrumentRegistryTest.java b/api/src/test/java/io/grpc/MetricInstrumentRegistryTest.java
new file mode 100644
index 00000000000..b378f4aaef5
--- /dev/null
+++ b/api/src/test/java/io/grpc/MetricInstrumentRegistryTest.java
@@ -0,0 +1,193 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc;
+
+import static com.google.common.truth.Truth.assertThat;
+import static io.grpc.MetricInstrumentRegistry.INITIAL_INSTRUMENT_CAPACITY;
+
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Unit test for {@link MetricInstrumentRegistry}.
+ */
+@RunWith(JUnit4.class)
+public class MetricInstrumentRegistryTest {
+  private static final ImmutableList<String> REQUIRED_LABEL_KEYS = ImmutableList.of("KEY1", "KEY2");
+  private static final ImmutableList<String> OPTIONAL_LABEL_KEYS = ImmutableList.of(
+      "OPTIONAL_KEY_1");
+  private static final ImmutableList<Double> DOUBLE_HISTOGRAM_BUCKETS = ImmutableList.of(0.01, 0.1);
+  private static final ImmutableList<Long> LONG_HISTOGRAM_BUCKETS = ImmutableList.of(1L, 10L);
+  private static final String METRIC_NAME_1 = "testMetric1";
+  private static final String DESCRIPTION_1 = "description1";
+  private static final String DESCRIPTION_2 = "description2";
+  private static final String UNIT_1 = "unit1";
+  private static final String UNIT_2 = "unit2";
+  private static final boolean ENABLED = true;
+  private static final boolean DISABLED = false;
+  private MetricInstrumentRegistry registry = new MetricInstrumentRegistry();
+
+  @Test
+  public void registerDoubleCounterSuccess() {
+    DoubleCounterMetricInstrument instrument = registry.registerDoubleCounter(
+        METRIC_NAME_1, DESCRIPTION_1, UNIT_1, REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, ENABLED);
+    assertThat(registry.getMetricInstruments().contains(instrument)).isTrue();
+    assertThat(registry.getMetricInstruments().size()).isEqualTo(1);
+    assertThat(instrument.getName()).isEqualTo(METRIC_NAME_1);
+    assertThat(instrument.getDescription()).isEqualTo(DESCRIPTION_1);
+    assertThat(instrument.getUnit()).isEqualTo(UNIT_1);
+    assertThat(instrument.getRequiredLabelKeys()).isEqualTo(REQUIRED_LABEL_KEYS);
+    assertThat(instrument.getOptionalLabelKeys()).isEqualTo(OPTIONAL_LABEL_KEYS);
+    assertThat(instrument.isEnableByDefault()).isTrue();
+  }
+
+  @Test
+  public void registerLongCounterSuccess() {
+    LongCounterMetricInstrument instrument2 = registry.registerLongCounter(
+        METRIC_NAME_1, DESCRIPTION_1, UNIT_1, REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, ENABLED);
+    assertThat(registry.getMetricInstruments().contains(instrument2)).isTrue();
+    assertThat(registry.getMetricInstruments().size()).isEqualTo(1);
+    assertThat(instrument2.getName()).isEqualTo(METRIC_NAME_1);
+    assertThat(instrument2.getDescription()).isEqualTo(DESCRIPTION_1);
+    assertThat(instrument2.getUnit()).isEqualTo(UNIT_1);
+    assertThat(instrument2.getRequiredLabelKeys()).isEqualTo(REQUIRED_LABEL_KEYS);
+    assertThat(instrument2.getOptionalLabelKeys()).isEqualTo(OPTIONAL_LABEL_KEYS);
+    assertThat(instrument2.isEnableByDefault()).isTrue();
+  }
+
+  @Test
+  public void registerDoubleHistogramSuccess() {
+    DoubleHistogramMetricInstrument instrument3 = registry.registerDoubleHistogram(
+        METRIC_NAME_1, DESCRIPTION_1, UNIT_1, DOUBLE_HISTOGRAM_BUCKETS, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, ENABLED);
+    assertThat(registry.getMetricInstruments().contains(instrument3)).isTrue();
+    assertThat(registry.getMetricInstruments().size()).isEqualTo(1);
+    assertThat(instrument3.getName()).isEqualTo(METRIC_NAME_1);
+    assertThat(instrument3.getDescription()).isEqualTo(DESCRIPTION_1);
+    assertThat(instrument3.getUnit()).isEqualTo(UNIT_1);
+    assertThat(instrument3.getBucketBoundaries()).isEqualTo(DOUBLE_HISTOGRAM_BUCKETS);
+    assertThat(instrument3.getRequiredLabelKeys()).isEqualTo(REQUIRED_LABEL_KEYS);
+    assertThat(instrument3.getOptionalLabelKeys()).isEqualTo(OPTIONAL_LABEL_KEYS);
+    assertThat(instrument3.isEnableByDefault()).isTrue();
+  }
+
+  @Test
+  public void registerLongHistogramSuccess() {
+    LongHistogramMetricInstrument instrument4 = registry.registerLongHistogram(
+        METRIC_NAME_1, DESCRIPTION_1, UNIT_1, LONG_HISTOGRAM_BUCKETS, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, ENABLED);
+    assertThat(registry.getMetricInstruments().contains(instrument4)).isTrue();
+    assertThat(registry.getMetricInstruments().size()).isEqualTo(1);
+    assertThat(instrument4.getName()).isEqualTo(METRIC_NAME_1);
+    assertThat(instrument4.getDescription()).isEqualTo(DESCRIPTION_1);
+    assertThat(instrument4.getUnit()).isEqualTo(UNIT_1);
+    assertThat(instrument4.getBucketBoundaries()).isEqualTo(LONG_HISTOGRAM_BUCKETS);
+    assertThat(instrument4.getRequiredLabelKeys()).isEqualTo(REQUIRED_LABEL_KEYS);
+    assertThat(instrument4.getOptionalLabelKeys()).isEqualTo(OPTIONAL_LABEL_KEYS);
+    assertThat(instrument4.isEnableByDefault()).isTrue();
+  }
+
+  @Test
+  public void registerLongGaugeSuccess() {
+    LongGaugeMetricInstrument instrument4 = registry.registerLongGauge(
+        METRIC_NAME_1, DESCRIPTION_1, UNIT_1, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, ENABLED);
+    assertThat(registry.getMetricInstruments().contains(instrument4)).isTrue();
+    assertThat(registry.getMetricInstruments().size()).isEqualTo(1);
+    assertThat(instrument4.getName()).isEqualTo(METRIC_NAME_1);
+    assertThat(instrument4.getDescription()).isEqualTo(DESCRIPTION_1);
+    assertThat(instrument4.getUnit()).isEqualTo(UNIT_1);
+    assertThat(instrument4.getRequiredLabelKeys()).isEqualTo(REQUIRED_LABEL_KEYS);
+    assertThat(instrument4.getOptionalLabelKeys()).isEqualTo(OPTIONAL_LABEL_KEYS);
+    assertThat(instrument4.isEnableByDefault()).isTrue();
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void registerDoubleCounterDuplicateName() {
+    registry.registerDoubleCounter(METRIC_NAME_1, DESCRIPTION_1, UNIT_1, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, ENABLED);
+    registry.registerDoubleCounter(METRIC_NAME_1, DESCRIPTION_2, UNIT_2, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, DISABLED);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void registerLongCounterDuplicateName() {
+    registry.registerDoubleCounter(METRIC_NAME_1, DESCRIPTION_1, UNIT_1, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, ENABLED);
+    registry.registerLongCounter(METRIC_NAME_1, DESCRIPTION_2, UNIT_2, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, DISABLED);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void registerDoubleHistogramDuplicateName() {
+    registry.registerLongHistogram(METRIC_NAME_1, DESCRIPTION_1, UNIT_1, LONG_HISTOGRAM_BUCKETS,
+        REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, ENABLED);
+    registry.registerDoubleHistogram(METRIC_NAME_1, DESCRIPTION_2, UNIT_2, DOUBLE_HISTOGRAM_BUCKETS,
+        REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, DISABLED);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void registerLongHistogramDuplicateName() {
+    registry.registerLongCounter(METRIC_NAME_1, DESCRIPTION_1, UNIT_1, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, ENABLED);
+    registry.registerLongHistogram(METRIC_NAME_1, DESCRIPTION_2, UNIT_2, LONG_HISTOGRAM_BUCKETS,
+        REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, DISABLED);
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void registerLongGaugeDuplicateName() {
+    registry.registerDoubleHistogram(METRIC_NAME_1, DESCRIPTION_1, UNIT_1, DOUBLE_HISTOGRAM_BUCKETS,
+        REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, ENABLED);
+    registry.registerLongGauge(METRIC_NAME_1, DESCRIPTION_2, UNIT_2, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, DISABLED);
+  }
+
+  @Test
+  public void getMetricInstrumentsMultipleRegistered() {
+    DoubleCounterMetricInstrument instrument1 = registry.registerDoubleCounter(
+        "testMetric1", DESCRIPTION_1, UNIT_1, REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, ENABLED);
+    LongCounterMetricInstrument instrument2 = registry.registerLongCounter(
+        "testMetric2", DESCRIPTION_2, UNIT_2, REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, DISABLED);
+    DoubleHistogramMetricInstrument instrument3 = registry.registerDoubleHistogram(
+        "testMetric3", DESCRIPTION_2, UNIT_2, DOUBLE_HISTOGRAM_BUCKETS, REQUIRED_LABEL_KEYS,
+        OPTIONAL_LABEL_KEYS, DISABLED);
+
+    List<MetricInstrument> instruments = registry.getMetricInstruments();
+    assertThat(instruments.size()).isEqualTo(3);
+    assertThat(instruments.contains(instrument1)).isTrue();
+    assertThat(instruments.contains(instrument2)).isTrue();
+    assertThat(instruments.contains(instrument3)).isTrue();
+  }
+
+  @Test
+  public void resizeMetricInstrumentsCapacityIncrease() {
+    int initialCapacity = INITIAL_INSTRUMENT_CAPACITY;
+    MetricInstrumentRegistry testRegistry = new MetricInstrumentRegistry();
+
+    // Registering enough instruments to trigger resize
+    for (int i = 0; i < initialCapacity + 1; i++) {
+      testRegistry.registerLongHistogram("name" + i, "desc", "unit", ImmutableList.of(),
+          ImmutableList.of(), ImmutableList.of(), true);
+    }
+
+    assertThat(testRegistry.getMetricInstruments().size()).isGreaterThan(initialCapacity);
+  }
+
+}
diff --git a/api/src/testFixtures/java/io/grpc/MetricInstrumentRegistryAccessor.java b/api/src/testFixtures/java/io/grpc/MetricInstrumentRegistryAccessor.java
new file mode 100644
index 00000000000..bd17dccad58
--- /dev/null
+++ b/api/src/testFixtures/java/io/grpc/MetricInstrumentRegistryAccessor.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc;
+
+/**
+ * Accesses test-only methods of {@link MetricInstrumentRegistry}.
+ */
+public final class MetricInstrumentRegistryAccessor {
+
+  private MetricInstrumentRegistryAccessor() {
+  }
+
+  public static MetricInstrumentRegistry createMetricInstrumentRegistry() {
+    return new MetricInstrumentRegistry();
+  }
+}
diff --git a/core/src/main/java/io/grpc/internal/MetricRecorderImpl.java b/core/src/main/java/io/grpc/internal/MetricRecorderImpl.java
new file mode 100644
index 00000000000..eab3e84d09d
--- /dev/null
+++ b/core/src/main/java/io/grpc/internal/MetricRecorderImpl.java
@@ -0,0 +1,174 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.grpc.DoubleCounterMetricInstrument;
+import io.grpc.DoubleHistogramMetricInstrument;
+import io.grpc.LongCounterMetricInstrument;
+import io.grpc.LongHistogramMetricInstrument;
+import io.grpc.MetricInstrument;
+import io.grpc.MetricInstrumentRegistry;
+import io.grpc.MetricRecorder;
+import io.grpc.MetricSink;
+import java.util.List;
+
+/**
+ * Provides a central point  for gRPC components to record metric values. Metrics can be exported to
+ * monitoring systems by configuring one or more {@link MetricSink}s.
+ *
+ * <p>This class encapsulates the interaction with metric sinks, including updating them with
+ * the latest set of {@link MetricInstrument}s provided by the {@link MetricInstrumentRegistry}.
+ */
+final class MetricRecorderImpl implements MetricRecorder {
+
+  private final List<MetricSink> metricSinks;
+  private final MetricInstrumentRegistry registry;
+
+  @VisibleForTesting
+  MetricRecorderImpl(List<MetricSink> metricSinks, MetricInstrumentRegistry registry) {
+    this.metricSinks = metricSinks;
+    this.registry = registry;
+  }
+
+  /**
+   * Records a double counter value.
+   *
+   * @param metricInstrument the {@link DoubleCounterMetricInstrument} to record.
+   * @param value the value to record.
+   * @param requiredLabelValues the required label values for the metric.
+   * @param optionalLabelValues the optional label values for the metric.
+   */
+  @Override
+  public void addDoubleCounter(DoubleCounterMetricInstrument metricInstrument, double value,
+      List<String> requiredLabelValues, List<String> optionalLabelValues) {
+    checkArgument(requiredLabelValues != null
+            && requiredLabelValues.size() == metricInstrument.getRequiredLabelKeys().size(),
+        "Incorrect number of required labels provided. Expected: "
+            + metricInstrument.getRequiredLabelKeys().size());
+    checkArgument(optionalLabelValues != null
+            && optionalLabelValues.size() == metricInstrument.getOptionalLabelKeys().size(),
+        "Incorrect number of optional labels provided. Expected: "
+            + metricInstrument.getOptionalLabelKeys().size());
+    for (MetricSink sink : metricSinks) {
+      // TODO(dnvindhya): Move updating measures logic from sink to here
+      List<Object> measures = sink.getMetricsMeasures();
+      if (measures.size() <= metricInstrument.getIndex()) {
+        // Measures may need updating in two cases:
+        // 1. When the sink is initially created with an empty list of measures.
+        // 2. When new metric instruments are registered, requiring the sink to accommodate them.
+        sink.updateMeasures(registry.getMetricInstruments());
+      }
+      sink.recordDoubleCounter(metricInstrument, value, requiredLabelValues, optionalLabelValues);
+    }
+  }
+
+  /**
+   * Records a long counter value.
+   *
+   * @param metricInstrument the {@link LongCounterMetricInstrument} to record.
+   * @param value the value to record.
+   * @param requiredLabelValues the required label values for the metric.
+   * @param optionalLabelValues the optional label values for the metric.
+   */
+  @Override
+  public void addLongCounter(LongCounterMetricInstrument metricInstrument, long value,
+      List<String> requiredLabelValues, List<String> optionalLabelValues) {
+    checkArgument(requiredLabelValues != null
+            && requiredLabelValues.size() == metricInstrument.getRequiredLabelKeys().size(),
+        "Incorrect number of required labels provided. Expected: "
+            + metricInstrument.getRequiredLabelKeys().size());
+    checkArgument(optionalLabelValues != null
+            && optionalLabelValues.size() == metricInstrument.getOptionalLabelKeys().size(),
+        "Incorrect number of optional labels provided. Expected: "
+            + metricInstrument.getOptionalLabelKeys().size());
+    for (MetricSink sink : metricSinks) {
+      List<Object> measures = sink.getMetricsMeasures();
+      if (measures.size() <= metricInstrument.getIndex()) {
+        // Measures may need updating in two cases:
+        // 1. When the sink is initially created with an empty list of measures.
+        // 2. When new metric instruments are registered, requiring the sink to accommodate them.
+        sink.updateMeasures(registry.getMetricInstruments());
+      }
+      sink.recordLongCounter(metricInstrument, value, requiredLabelValues, optionalLabelValues);
+    }
+  }
+
+  /**
+   * Records a double histogram value.
+   *
+   * @param metricInstrument the {@link DoubleHistogramMetricInstrument} to record.
+   * @param value the value to record.
+   * @param requiredLabelValues the required label values for the metric.
+   * @param optionalLabelValues the optional label values for the metric.
+   */
+  @Override
+  public void recordDoubleHistogram(DoubleHistogramMetricInstrument metricInstrument, double value,
+      List<String> requiredLabelValues, List<String> optionalLabelValues) {
+    checkArgument(requiredLabelValues != null
+            && requiredLabelValues.size() == metricInstrument.getRequiredLabelKeys().size(),
+        "Incorrect number of required labels provided. Expected: "
+            + metricInstrument.getRequiredLabelKeys().size());
+    checkArgument(optionalLabelValues != null
+            && optionalLabelValues.size() == metricInstrument.getOptionalLabelKeys().size(),
+        "Incorrect number of optional labels provided. Expected: "
+            + metricInstrument.getOptionalLabelKeys().size());
+    for (MetricSink sink : metricSinks) {
+      List<Object> measures = sink.getMetricsMeasures();
+      if (measures.size() <= metricInstrument.getIndex()) {
+        // Measures may need updating in two cases:
+        // 1. When the sink is initially created with an empty list of measures.
+        // 2. When new metric instruments are registered, requiring the sink to accommodate them.
+        sink.updateMeasures(registry.getMetricInstruments());
+      }
+      sink.recordDoubleHistogram(metricInstrument, value, requiredLabelValues, optionalLabelValues);
+    }
+  }
+
+  /**
+   * Records a long histogram value.
+   *
+   * @param metricInstrument the {@link LongHistogramMetricInstrument} to record.
+   * @param value the value to record.
+   * @param requiredLabelValues the required label values for the metric.
+   * @param optionalLabelValues the optional label values for the metric.
+   */
+  @Override
+  public void recordLongHistogram(LongHistogramMetricInstrument metricInstrument, long value,
+      List<String> requiredLabelValues, List<String> optionalLabelValues) {
+    checkArgument(requiredLabelValues != null
+            && requiredLabelValues.size() == metricInstrument.getRequiredLabelKeys().size(),
+        "Incorrect number of required labels provided. Expected: "
+            + metricInstrument.getRequiredLabelKeys().size());
+    checkArgument(optionalLabelValues != null
+            && optionalLabelValues.size() == metricInstrument.getOptionalLabelKeys().size(),
+        "Incorrect number of optional labels provided. Expected: "
+            + metricInstrument.getOptionalLabelKeys().size());
+    for (MetricSink sink : metricSinks) {
+      List<Object> measures = sink.getMetricsMeasures();
+      if (measures.size() <= metricInstrument.getIndex()) {
+        // Measures may need updating in two cases:
+        // 1. When the sink is initially created with an empty list of measures.
+        // 2. When new metric instruments are registered, requiring the sink to accommodate them.
+        sink.updateMeasures(registry.getMetricInstruments());
+      }
+      sink.recordLongHistogram(metricInstrument, value, requiredLabelValues, optionalLabelValues);
+    }
+  }
+}
diff --git a/core/src/test/java/io/grpc/internal/MetricRecorderImplTest.java b/core/src/test/java/io/grpc/internal/MetricRecorderImplTest.java
new file mode 100644
index 00000000000..9f79732c7eb
--- /dev/null
+++ b/core/src/test/java/io/grpc/internal/MetricRecorderImplTest.java
@@ -0,0 +1,224 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableList;
+import io.grpc.DoubleCounterMetricInstrument;
+import io.grpc.DoubleHistogramMetricInstrument;
+import io.grpc.LongCounterMetricInstrument;
+import io.grpc.LongHistogramMetricInstrument;
+import io.grpc.MetricInstrumentRegistry;
+import io.grpc.MetricInstrumentRegistryAccessor;
+import io.grpc.MetricRecorder;
+import io.grpc.MetricSink;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Unit test for {@link MetricRecorderImpl}.
+ */
+@RunWith(JUnit4.class)
+public class MetricRecorderImplTest {
+  private static final String DESCRIPTION = "description";
+  private static final String UNIT = "unit";
+  private static final boolean ENABLED = true;
+  private static final ImmutableList<String> REQUIRED_LABEL_KEYS = ImmutableList.of("KEY1", "KEY2");
+  private static final ImmutableList<String> OPTIONAL_LABEL_KEYS = ImmutableList.of(
+      "OPTIONAL_KEY_1");
+  private static final ImmutableList<String> REQUIRED_LABEL_VALUES = ImmutableList.of("VALUE1",
+      "VALUE2");
+  private static final ImmutableList<String> OPTIONAL_LABEL_VALUES = ImmutableList.of(
+      "OPTIONAL_VALUE_1");
+  private MetricSink mockSink = mock(MetricSink.class);
+  private List<MetricSink> sinks = Arrays.asList(mockSink, mockSink);
+  private MetricInstrumentRegistry registry =
+      MetricInstrumentRegistryAccessor.createMetricInstrumentRegistry();
+  private final DoubleCounterMetricInstrument doubleCounterInstrument =
+      registry.registerDoubleCounter("counter0", DESCRIPTION, UNIT, REQUIRED_LABEL_KEYS,
+          OPTIONAL_LABEL_KEYS, ENABLED);
+  private final LongCounterMetricInstrument longCounterInstrument =
+      registry.registerLongCounter("counter1", DESCRIPTION, UNIT, REQUIRED_LABEL_KEYS,
+          OPTIONAL_LABEL_KEYS, ENABLED);
+  private final DoubleHistogramMetricInstrument doubleHistogramInstrument =
+      registry.registerDoubleHistogram("histogram1", DESCRIPTION, UNIT,
+          Collections.emptyList(), REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, ENABLED);
+  private final LongHistogramMetricInstrument longHistogramInstrument =
+      registry.registerLongHistogram("histogram2", DESCRIPTION, UNIT,
+          Collections.emptyList(), REQUIRED_LABEL_KEYS, OPTIONAL_LABEL_KEYS, ENABLED);
+  private MetricRecorder recorder;
+
+  @Before
+  public void setUp() {
+    recorder = new MetricRecorderImpl(sinks, registry);
+  }
+
+  @Test
+  public void recordCounter() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.addDoubleCounter(doubleCounterInstrument, 1.0, REQUIRED_LABEL_VALUES,
+        OPTIONAL_LABEL_VALUES);
+    verify(mockSink, times(2)).recordDoubleCounter(eq(doubleCounterInstrument), eq(1D),
+        eq(REQUIRED_LABEL_VALUES), eq(OPTIONAL_LABEL_VALUES));
+
+    recorder.addLongCounter(longCounterInstrument, 1, REQUIRED_LABEL_VALUES,
+        OPTIONAL_LABEL_VALUES);
+    verify(mockSink, times(2)).recordLongCounter(eq(longCounterInstrument), eq(1L),
+        eq(REQUIRED_LABEL_VALUES), eq(OPTIONAL_LABEL_VALUES));
+
+    verify(mockSink, never()).updateMeasures(registry.getMetricInstruments());
+  }
+
+  @Test
+  public void recordHistogram() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.recordDoubleHistogram(doubleHistogramInstrument, 99.0, REQUIRED_LABEL_VALUES,
+        OPTIONAL_LABEL_VALUES);
+    verify(mockSink, times(2)).recordDoubleHistogram(eq(doubleHistogramInstrument),
+        eq(99D), eq(REQUIRED_LABEL_VALUES), eq(OPTIONAL_LABEL_VALUES));
+
+    recorder.recordLongHistogram(longHistogramInstrument, 99, REQUIRED_LABEL_VALUES,
+        OPTIONAL_LABEL_VALUES);
+    verify(mockSink, times(2)).recordLongHistogram(eq(longHistogramInstrument), eq(99L),
+        eq(REQUIRED_LABEL_VALUES), eq(OPTIONAL_LABEL_VALUES));
+
+    verify(mockSink, never()).updateMeasures(registry.getMetricInstruments());
+  }
+
+  @Test
+  public void newRegisteredMetricUpdateMeasures() {
+    // Sink is initialized with zero measures, should trigger updateMeasures() on sinks
+    when(mockSink.getMetricsMeasures()).thenReturn(new ArrayList<>());
+
+    // Double Counter
+    recorder.addDoubleCounter(doubleCounterInstrument, 1.0, REQUIRED_LABEL_VALUES,
+        OPTIONAL_LABEL_VALUES);
+    verify(mockSink, times(2)).updateMeasures(anyList());
+    verify(mockSink, times(2)).recordDoubleCounter(eq(doubleCounterInstrument), eq(1D),
+        eq(REQUIRED_LABEL_VALUES), eq(OPTIONAL_LABEL_VALUES));
+
+    // Long Counter
+    recorder.addLongCounter(longCounterInstrument, 1, REQUIRED_LABEL_VALUES,
+        OPTIONAL_LABEL_VALUES);
+    verify(mockSink, times(4)).updateMeasures(anyList());
+    verify(mockSink, times(2)).recordLongCounter(eq(longCounterInstrument), eq(1L),
+        eq(REQUIRED_LABEL_VALUES), eq(OPTIONAL_LABEL_VALUES));
+
+    // Double Histogram
+    recorder.recordDoubleHistogram(doubleHistogramInstrument, 99.0, REQUIRED_LABEL_VALUES,
+        OPTIONAL_LABEL_VALUES);
+    verify(mockSink, times(6)).updateMeasures(anyList());
+    verify(mockSink, times(2)).recordDoubleHistogram(eq(doubleHistogramInstrument),
+        eq(99D), eq(REQUIRED_LABEL_VALUES), eq(OPTIONAL_LABEL_VALUES));
+
+    // Long Histogram
+    recorder.recordLongHistogram(longHistogramInstrument, 99, REQUIRED_LABEL_VALUES,
+        OPTIONAL_LABEL_VALUES);
+    verify(mockSink, times(8)).updateMeasures(registry.getMetricInstruments());
+    verify(mockSink, times(2)).recordLongHistogram(eq(longHistogramInstrument), eq(99L),
+        eq(REQUIRED_LABEL_VALUES), eq(OPTIONAL_LABEL_VALUES));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void recordDoubleCounterMismatchedRequiredLabelValues() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.addDoubleCounter(doubleCounterInstrument, 1.0, ImmutableList.of(),
+        OPTIONAL_LABEL_VALUES);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void recordLongCounterMismatchedRequiredLabelValues() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.addLongCounter(longCounterInstrument, 1, ImmutableList.of(),
+        OPTIONAL_LABEL_VALUES);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void recordDoubleHistogramMismatchedRequiredLabelValues() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.recordDoubleHistogram(doubleHistogramInstrument, 99.0, ImmutableList.of(),
+        OPTIONAL_LABEL_VALUES);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void recordLongHistogramMismatchedRequiredLabelValues() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.recordLongHistogram(longHistogramInstrument, 99, ImmutableList.of(),
+        OPTIONAL_LABEL_VALUES);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void recordDoubleCounterMismatchedOptionalLabelValues() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.addDoubleCounter(doubleCounterInstrument, 1.0, REQUIRED_LABEL_VALUES,
+        ImmutableList.of());
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void recordLongCounterMismatchedOptionalLabelValues() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.addLongCounter(longCounterInstrument, 1, REQUIRED_LABEL_VALUES,
+        ImmutableList.of());
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void recordDoubleHistogramMismatchedOptionalLabelValues() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.recordDoubleHistogram(doubleHistogramInstrument, 99.0, REQUIRED_LABEL_VALUES,
+        ImmutableList.of());
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void recordLongHistogramMismatchedOptionalLabelValues() {
+    when(mockSink.getMetricsMeasures()).thenReturn(
+        Arrays.asList(new Object(), new Object(), new Object(), new Object()));
+
+    recorder.recordLongHistogram(longHistogramInstrument, 99, REQUIRED_LABEL_VALUES,
+        ImmutableList.of());
+  }
+}