diff --git a/sdk/include/opentelemetry/sdk/metrics/observer_result.h b/sdk/include/opentelemetry/sdk/metrics/observer_result.h
index cc60cf5f64..827f371368 100644
--- a/sdk/include/opentelemetry/sdk/metrics/observer_result.h
+++ b/sdk/include/opentelemetry/sdk/metrics/observer_result.h
@@ -3,14 +3,13 @@
 
 #pragma once
 
-#include <unordered_map>
+#include <memory>
 
 #include "opentelemetry/common/attribute_value.h"
 #include "opentelemetry/common/key_value_iterable.h"
 #include "opentelemetry/metrics/observer_result.h"
 #include "opentelemetry/nostd/string_view.h"
-#include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
-#include "opentelemetry/sdk/metrics/view/attributes_processor.h"
+#include "opentelemetry/sdk/metrics/state/measurement_attributes_map.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -22,32 +21,33 @@ template <class T>
 class ObserverResultT final : public opentelemetry::metrics::ObserverResultT<T>
 {
 public:
-  explicit ObserverResultT(const AttributesProcessor *attributes_processor = nullptr)
-      : attributes_processor_(attributes_processor)
-  {}
+  explicit ObserverResultT() = default;
 
   ~ObserverResultT() override = default;
 
   void Observe(T value) noexcept override
   {
-    data_[MetricAttributes{{}, attributes_processor_}] = value;
+    std::unordered_map<nostd::string_view, opentelemetry::common::AttributeValue> empty;
+    data_[empty] += value;
   }
 
   void Observe(T value, const opentelemetry::common::KeyValueIterable &attributes) noexcept override
   {
-    data_[MetricAttributes{attributes, attributes_processor_}] =
-        value;  // overwrites the previous value if present
+    std::unordered_map<nostd::string_view, opentelemetry::common::AttributeValue> attr_map;
+    attributes.ForEachKeyValue(
+        [&](nostd::string_view key, opentelemetry::common::AttributeValue val) noexcept {
+          attr_map.emplace(key, val);
+          return true;
+        });
+    data_[attr_map] += value;
   }
 
-  const std::unordered_map<MetricAttributes, T, AttributeHashGenerator> &GetMeasurements()
-  {
-    return data_;
-  }
+  const Measurements<T> &GetMeasurements() { return data_; }
 
 private:
-  std::unordered_map<MetricAttributes, T, AttributeHashGenerator> data_;
-  const AttributesProcessor *attributes_processor_;
+  Measurements<T> data_;
 };
+
 }  // namespace metrics
 }  // namespace sdk
 
diff --git a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h
index 968de7aa82..a9ddde3ec2 100644
--- a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h
+++ b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h
@@ -17,9 +17,11 @@
 #  include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
 #endif
 
+#include "opentelemetry/common/key_value_iterable_view.h"
 #include "opentelemetry/sdk/metrics/instruments.h"
 #include "opentelemetry/sdk/metrics/observer_result.h"
 #include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
+#include "opentelemetry/sdk/metrics/state/measurement_attributes_map.h"
 #include "opentelemetry/sdk/metrics/state/metric_collector.h"
 #include "opentelemetry/sdk/metrics/state/metric_storage.h"
 #include "opentelemetry/sdk/metrics/state/temporal_metric_storage.h"
@@ -36,6 +38,7 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
 public:
   AsyncMetricStorage(InstrumentDescriptor instrument_descriptor,
                      const AggregationType aggregation_type,
+                     std::shared_ptr<const AttributesProcessor> attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
                      ExemplarFilterType exempler_filter_type,
                      nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
@@ -48,6 +51,7 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
             std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
         delta_hash_map_(
             std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
+        attributes_processor_(std::move(attributes_processor)),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
         exemplar_filter_type_(exempler_filter_type),
         exemplar_reservoir_(exemplar_reservoir),
@@ -95,26 +99,45 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
     }
   }
 
-  void RecordLong(
-      const std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> &measurements,
-      opentelemetry::common::SystemTimestamp observation_time) noexcept override
+  void RecordLong(const opentelemetry::sdk::metrics::Measurements<int64_t> &measurements,
+                  opentelemetry::common::SystemTimestamp observation_time) noexcept override
   {
     if (instrument_descriptor_.value_type_ != InstrumentValueType::kLong)
     {
       return;
     }
-    Record<int64_t>(measurements, observation_time);
+    std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> mp =
+        ConvertMeasurementsToMetricAttributes<int64_t>(measurements, observation_time);
+    Record<int64_t>(mp, observation_time);
   }
 
-  void RecordDouble(
-      const std::unordered_map<MetricAttributes, double, AttributeHashGenerator> &measurements,
-      opentelemetry::common::SystemTimestamp observation_time) noexcept override
+  void RecordDouble(const opentelemetry::sdk::metrics::Measurements<double> &measurements,
+                    opentelemetry::common::SystemTimestamp observation_time) noexcept override
   {
     if (instrument_descriptor_.value_type_ != InstrumentValueType::kDouble)
     {
       return;
     }
-    Record<double>(measurements, observation_time);
+    std::unordered_map<MetricAttributes, double, AttributeHashGenerator> mp =
+        ConvertMeasurementsToMetricAttributes<double>(measurements, observation_time);
+    Record<double>(mp, observation_time);
+  }
+
+  template <class T>
+  std::unordered_map<MetricAttributes, T, AttributeHashGenerator>
+  ConvertMeasurementsToMetricAttributes(
+      const opentelemetry::sdk::metrics::Measurements<T> &measurements,
+      opentelemetry::common::SystemTimestamp /* observation_time */) noexcept
+  {
+    std::unordered_map<MetricAttributes, T, AttributeHashGenerator> mp;
+    for (auto &measurement : measurements)
+    {
+      opentelemetry::common::KeyValueIterableView<
+          std::unordered_map<nostd::string_view, opentelemetry::common::AttributeValue>>
+          kv_view{measurement.first};
+      mp[MetricAttributes{kv_view, attributes_processor_.get()}] += measurement.second;
+    }
+    return mp;
   }
 
   bool Collect(CollectorHandle *collector,
@@ -144,6 +167,7 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
   const AggregationConfig *aggregation_config_;
   std::unique_ptr<AttributesHashMap> cumulative_hash_map_;
   std::unique_ptr<AttributesHashMap> delta_hash_map_;
+  std::shared_ptr<const AttributesProcessor> attributes_processor_;
   opentelemetry::common::SpinLockMutex hashmap_lock_;
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
   ExemplarFilterType exemplar_filter_type_;
diff --git a/sdk/include/opentelemetry/sdk/metrics/state/measurement_attributes_map.h b/sdk/include/opentelemetry/sdk/metrics/state/measurement_attributes_map.h
new file mode 100644
index 0000000000..82b6d549ac
--- /dev/null
+++ b/sdk/include/opentelemetry/sdk/metrics/state/measurement_attributes_map.h
@@ -0,0 +1,157 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <cstring>
+#include <string>
+#include <unordered_map>
+#include <utility>
+
+#include "opentelemetry/common/attribute_value.h"
+#include "opentelemetry/common/key_value_iterable.h"
+#include "opentelemetry/nostd/string_view.h"
+#include "opentelemetry/sdk/common/custom_hash_equality.h"
+#include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h"
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+// Hash function for AttributeValue (int and string only)
+struct AttributeValueHash
+{
+  size_t operator()(const opentelemetry::common::AttributeValue &value) const noexcept
+  {
+    if (nostd::holds_alternative<int32_t>(value))
+    {
+      return std::hash<int32_t>{}(nostd::get<int32_t>(value));
+    }
+    else if (nostd::holds_alternative<int64_t>(value))
+    {
+      return std::hash<int64_t>{}(nostd::get<int64_t>(value));
+    }
+    else if (nostd::holds_alternative<const char *>(value))
+    {
+      return std::hash<nostd::string_view>{}(nostd::string_view(nostd::get<const char *>(value)));
+    }
+    else if (nostd::holds_alternative<nostd::string_view>(value))
+    {
+      return std::hash<nostd::string_view>{}(nostd::get<nostd::string_view>(value));
+    }
+
+    return 0;  // fallback for other types
+  }
+};
+
+// Equality function for AttributeValue (int and string only)
+struct AttributeValueEqual
+{
+  bool operator()(const opentelemetry::common::AttributeValue &a,
+                  const opentelemetry::common::AttributeValue &b) const noexcept
+  {
+    if (a.index() != b.index())
+    {
+      return false;
+    }
+
+    // Compare int32_t
+    if (nostd::holds_alternative<int32_t>(a))
+    {
+      return nostd::get<int32_t>(a) == nostd::get<int32_t>(b);
+    }
+    // Compare int64_t
+    else if (nostd::holds_alternative<int64_t>(a))
+    {
+      return nostd::get<int64_t>(a) == nostd::get<int64_t>(b);
+    }
+    // Compare const char*
+    else if (nostd::holds_alternative<const char *>(a))
+    {
+      return nostd::string_view(nostd::get<const char *>(a)) ==
+             nostd::string_view(nostd::get<const char *>(b));
+    }
+    // Compare string_view
+    else if (nostd::holds_alternative<nostd::string_view>(a))
+    {
+      return nostd::get<nostd::string_view>(a) == nostd::get<nostd::string_view>(b);
+    }
+
+    return false;
+  }
+};
+
+// Hash function for unordered_map of key-value pairs
+// This Custom Hash is only applied to strings and int for now
+struct AttributeMapHash
+{
+  size_t operator()(
+      const std::unordered_map<nostd::string_view, opentelemetry::common::AttributeValue> &map)
+      const noexcept
+  {
+    size_t hash = 0;
+    AttributeValueHash value_hasher;
+
+    for (const auto &pair : map)
+    {
+      // Combine key hash
+      size_t key_hash = std::hash<nostd::string_view>{}(pair.first);
+      hash ^= key_hash + 0x9e3779b9 + (hash << 6) + (hash >> 2);
+
+      // Combine value hash
+      size_t value_hash = value_hasher(pair.second);
+      hash ^= value_hash + 0x9e3779b9 + (hash << 6) + (hash >> 2);
+    }
+
+    return hash;
+  }
+};
+
+// Equality function for unordered_map of key-value pairs
+// This Custom Equal is only applied to strings and int for now
+struct AttributeMapEqual
+{
+  bool operator()(
+      const std::unordered_map<nostd::string_view, opentelemetry::common::AttributeValue> &a,
+      const std::unordered_map<nostd::string_view, opentelemetry::common::AttributeValue> &b)
+      const noexcept
+  {
+    if (a.size() != b.size())
+    {
+      return false;
+    }
+
+    AttributeValueEqual value_equal;
+
+    for (const auto &pair_a : a)
+    {
+      auto it = b.find(pair_a.first);
+      if (it == b.end())
+      {
+        return false;  // key not found in b
+      }
+
+      // Compare values
+      if (!value_equal(pair_a.second, it->second))
+      {
+        return false;
+      }
+    }
+
+    return true;
+  }
+};
+
+template <class T>
+using Measurements = std::unordered_map<
+    std::unordered_map<nostd::string_view, opentelemetry::common::AttributeValue>,
+    T,
+    AttributeMapHash,
+    AttributeMapEqual>;
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
diff --git a/sdk/include/opentelemetry/sdk/metrics/state/metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/metric_storage.h
index 26adc37c08..0df7499ccd 100644
--- a/sdk/include/opentelemetry/sdk/metrics/state/metric_storage.h
+++ b/sdk/include/opentelemetry/sdk/metrics/state/metric_storage.h
@@ -13,6 +13,7 @@
 #include "opentelemetry/sdk/metrics/data/metric_data.h"
 #include "opentelemetry/sdk/metrics/instruments.h"
 #include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
+#include "opentelemetry/sdk/metrics/state/measurement_attributes_map.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -72,13 +73,11 @@ class AsyncWritableMetricStorage
   virtual ~AsyncWritableMetricStorage() = default;
 
   /* Records a batch of measurements */
-  virtual void RecordLong(
-      const std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> &measurements,
-      opentelemetry::common::SystemTimestamp observation_time) noexcept = 0;
+  virtual void RecordLong(const opentelemetry::sdk::metrics::Measurements<int64_t> &measurements,
+                          opentelemetry::common::SystemTimestamp observation_time) noexcept = 0;
 
-  virtual void RecordDouble(
-      const std::unordered_map<MetricAttributes, double, AttributeHashGenerator> &measurements,
-      opentelemetry::common::SystemTimestamp observation_time) noexcept = 0;
+  virtual void RecordDouble(const opentelemetry::sdk::metrics::Measurements<double> &measurements,
+                            opentelemetry::common::SystemTimestamp observation_time) noexcept = 0;
 };
 
 class NoopMetricStorage : public MetricStorage
@@ -119,13 +118,11 @@ class NoopWritableMetricStorage : public SyncWritableMetricStorage
 class NoopAsyncWritableMetricStorage : public AsyncWritableMetricStorage
 {
 public:
-  void RecordLong(const std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator>
-                      & /* measurements */,
+  void RecordLong(const opentelemetry::sdk::metrics::Measurements<int64_t> & /* measurements */,
                   opentelemetry::common::SystemTimestamp /* observation_time */) noexcept override
   {}
 
-  void RecordDouble(const std::unordered_map<MetricAttributes, double, AttributeHashGenerator>
-                        & /* measurements */,
+  void RecordDouble(const opentelemetry::sdk::metrics::Measurements<double> & /* measurements */,
                     opentelemetry::common::SystemTimestamp /* observation_time */) noexcept override
   {}
 };
diff --git a/sdk/include/opentelemetry/sdk/metrics/state/multi_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/multi_metric_storage.h
index 2a77b2f856..5a8c232571 100644
--- a/sdk/include/opentelemetry/sdk/metrics/state/multi_metric_storage.h
+++ b/sdk/include/opentelemetry/sdk/metrics/state/multi_metric_storage.h
@@ -12,6 +12,7 @@
 #include "opentelemetry/common/timestamp.h"
 #include "opentelemetry/context/context.h"
 #include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
+#include "opentelemetry/sdk/metrics/state/measurement_attributes_map.h"
 #include "opentelemetry/sdk/metrics/state/metric_storage.h"
 #include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 #include "opentelemetry/version.h"
@@ -80,9 +81,8 @@ class AsyncMultiMetricStorage : public AsyncWritableMetricStorage
     storages_.push_back(storage);
   }
 
-  void RecordLong(
-      const std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> &measurements,
-      opentelemetry::common::SystemTimestamp observation_time) noexcept override
+  void RecordLong(const opentelemetry::sdk::metrics::Measurements<int64_t> &measurements,
+                  opentelemetry::common::SystemTimestamp observation_time) noexcept override
   {
     for (auto &s : storages_)
     {
@@ -90,9 +90,8 @@ class AsyncMultiMetricStorage : public AsyncWritableMetricStorage
     }
   }
 
-  void RecordDouble(
-      const std::unordered_map<MetricAttributes, double, AttributeHashGenerator> &measurements,
-      opentelemetry::common::SystemTimestamp observation_time) noexcept override
+  void RecordDouble(const opentelemetry::sdk::metrics::Measurements<double> &measurements,
+                    opentelemetry::common::SystemTimestamp observation_time) noexcept override
   {
     for (auto &s : storages_)
     {
diff --git a/sdk/src/metrics/meter.cc b/sdk/src/metrics/meter.cc
index 475f8a9c13..d0125c618d 100644
--- a/sdk/src/metrics/meter.cc
+++ b/sdk/src/metrics/meter.cc
@@ -592,7 +592,7 @@ std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(
        ,
        exemplar_filter_type
 #endif
-  ](const View &view) {
+  ](const View &view) -> bool {
         auto view_instr_desc = instrument_descriptor;
         if (!view.GetName().empty())
         {
@@ -616,12 +616,14 @@ std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(
         {
           WarnOnDuplicateInstrument(GetInstrumentationScope(), storage_registry_, view_instr_desc);
           async_storage = std::shared_ptr<AsyncMetricStorage>(new AsyncMetricStorage(
-              view_instr_desc, view.GetAggregationType(),
+              view_instr_desc, view.GetAggregationType(), view.GetAttributesProcessor()
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+                                                              ,
               exemplar_filter_type,
               GetExemplarReservoir(view.GetAggregationType(), view.GetAggregationConfig(),
-                                   view_instr_desc),
+                                   view_instr_desc)
 #endif
+                  ,
               view.GetAggregationConfig()));
           storage_registry_.insert({view_instr_desc, async_storage});
         }
diff --git a/sdk/test/metrics/async_metric_storage_test.cc b/sdk/test/metrics/async_metric_storage_test.cc
index 380ddd0564..7019a2a103 100644
--- a/sdk/test/metrics/async_metric_storage_test.cc
+++ b/sdk/test/metrics/async_metric_storage_test.cc
@@ -22,16 +22,13 @@
 #include "opentelemetry/sdk/metrics/export/metric_producer.h"
 #include "opentelemetry/sdk/metrics/instruments.h"
 #include "opentelemetry/sdk/metrics/state/async_metric_storage.h"
-#include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
-#include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h"
+#include "opentelemetry/sdk/metrics/state/measurement_attributes_map.h"
 #include "opentelemetry/sdk/metrics/state/metric_collector.h"
+#include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
-#  include "opentelemetry/sdk/metrics/data/exemplar_data.h"
 #  include "opentelemetry/sdk/metrics/exemplar/filter_type.h"
 #  include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
-#else
-#  include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 #endif
 
 using namespace opentelemetry::sdk::metrics;
@@ -67,22 +64,25 @@ TEST_P(WritableMetricStorageTestFixture, TestAggregation)
   std::shared_ptr<CollectorHandle> collector(new MockCollectorHandle(temporality));
   std::vector<std::shared_ptr<CollectorHandle>> collectors;
   collectors.push_back(collector);
+  std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attrproc{
+      new opentelemetry::sdk::metrics::DefaultAttributesProcessor()};
 
   opentelemetry::sdk::metrics::AsyncMetricStorage storage(
-      instr_desc, AggregationType::kSum,
+      instr_desc, AggregationType::kSum, std::move(attrproc),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
       nullptr);
-  int64_t get_count1                                                                  = 20;
-  int64_t put_count1                                                                  = 10;
-  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements1 = {
+  int64_t get_count1                                               = 20;
+  int64_t put_count1                                               = 10;
+  opentelemetry::sdk::metrics::Measurements<int64_t> measurements1 = {
       {{{"RequestType", "GET"}}, get_count1}, {{{"RequestType", "PUT"}}, put_count1}};
   storage.RecordLong(measurements1,
                      opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
 
   storage.Collect(
       collector.get(), collectors, sdk_start_ts, collection_ts, [&](const MetricData &metric_data) {
+        EXPECT_EQ(metric_data.point_data_attr_.size(), 2);
         for (const auto &data_attr : metric_data.point_data_attr_)
         {
           const auto &data = opentelemetry::nostd::get<SumPointData>(data_attr.point_data);
@@ -104,7 +104,7 @@ TEST_P(WritableMetricStorageTestFixture, TestAggregation)
   int64_t get_count2 = 50;
   int64_t put_count2 = 70;
 
-  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements2 = {
+  opentelemetry::sdk::metrics::Measurements<int64_t> measurements2 = {
       {{{"RequestType", "GET"}}, get_count2}, {{{"RequestType", "PUT"}}, put_count2}};
   storage.RecordLong(measurements2,
                      opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
@@ -163,15 +163,18 @@ TEST_P(WritableMetricStorageTestUpDownFixture, TestAggregation)
   std::vector<std::shared_ptr<CollectorHandle>> collectors;
   collectors.push_back(collector);
 
+  std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attrproc{
+      new opentelemetry::sdk::metrics::DefaultAttributesProcessor()};
+
   opentelemetry::sdk::metrics::AsyncMetricStorage storage(
-      instr_desc, AggregationType::kDefault,
+      instr_desc, AggregationType::kDefault, std::move(attrproc),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
       nullptr);
-  int64_t get_count1                                                                  = 20;
-  int64_t put_count1                                                                  = 10;
-  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements1 = {
+  int64_t get_count1                                               = 20;
+  int64_t put_count1                                               = 10;
+  opentelemetry::sdk::metrics::Measurements<int64_t> measurements1 = {
       {{{"RequestType", "GET"}}, get_count1}, {{{"RequestType", "PUT"}}, put_count1}};
   storage.RecordLong(measurements1,
                      opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
@@ -201,7 +204,7 @@ TEST_P(WritableMetricStorageTestUpDownFixture, TestAggregation)
   int64_t get_count2 = -50;
   int64_t put_count2 = -70;
 
-  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements2 = {
+  opentelemetry::sdk::metrics::Measurements<int64_t> measurements2 = {
       {{{"RequestType", "GET"}}, get_count2}, {{{"RequestType", "PUT"}}, put_count2}};
   storage.RecordLong(measurements2,
                      opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
@@ -259,16 +262,19 @@ TEST_P(WritableMetricStorageTestObservableGaugeFixture, TestAggregation)
   std::vector<std::shared_ptr<CollectorHandle>> collectors;
   collectors.push_back(collector);
 
+  std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attrproc{
+      new opentelemetry::sdk::metrics::DefaultAttributesProcessor()};
+
   opentelemetry::sdk::metrics::AsyncMetricStorage storage(
-      instr_desc, AggregationType::kLastValue,
+      instr_desc, AggregationType::kLastValue, std::move(attrproc),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
       nullptr);
-  int64_t freq_cpu0                                                                   = 3;
-  int64_t freq_cpu1                                                                   = 5;
-  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements1 = {
-      {{{"CPU", "0"}}, freq_cpu0}, {{{"CPU", "1"}}, freq_cpu1}};
+  int64_t freq_cpu0                                                = 3;
+  int64_t freq_cpu1                                                = 5;
+  opentelemetry::sdk::metrics::Measurements<int64_t> measurements1 = {{{{"CPU", "0"}}, freq_cpu0},
+                                                                      {{{"CPU", "1"}}, freq_cpu1}};
   storage.RecordLong(measurements1,
                      opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
 
@@ -294,8 +300,8 @@ TEST_P(WritableMetricStorageTestObservableGaugeFixture, TestAggregation)
   freq_cpu0 = 6;
   freq_cpu1 = 8;
 
-  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements2 = {
-      {{{"CPU", "0"}}, freq_cpu0}, {{{"CPU", "1"}}, freq_cpu1}};
+  opentelemetry::sdk::metrics::Measurements<int64_t> measurements2 = {{{{"CPU", "0"}}, freq_cpu0},
+                                                                      {{{"CPU", "1"}}, freq_cpu1}};
   storage.RecordLong(measurements2,
                      opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
   storage.Collect(
diff --git a/sdk/test/metrics/observer_result_test.cc b/sdk/test/metrics/observer_result_test.cc
index 5f6e7fe3b2..4d467eff24 100644
--- a/sdk/test/metrics/observer_result_test.cc
+++ b/sdk/test/metrics/observer_result_test.cc
@@ -5,20 +5,15 @@
 #include <cstdint>
 #include <map>
 #include <string>
-#include <unordered_map>
 #include <utility>
 
 #include "opentelemetry/common/key_value_iterable_view.h"
 #include "opentelemetry/sdk/metrics/observer_result.h"
-#include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
-#include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 
 using namespace opentelemetry::sdk::metrics;
 TEST(ObserverResult, BasicTests)
 {
-  const AttributesProcessor *attributes_processor = new DefaultAttributesProcessor();
-
-  ObserverResultT<int64_t> observer_result(attributes_processor);
+  ObserverResultT<int64_t> observer_result;
 
   observer_result.Observe(10);
   observer_result.Observe(20);
@@ -37,6 +32,4 @@ TEST(ObserverResult, BasicTests)
   observer_result.Observe(
       40, opentelemetry::common::KeyValueIterableView<std::map<std::string, int64_t>>(m2));
   EXPECT_EQ(observer_result.GetMeasurements().size(), 3);
-
-  delete attributes_processor;
 }
diff --git a/sdk/test/metrics/sum_aggregation_test.cc b/sdk/test/metrics/sum_aggregation_test.cc
index 5c29fe43e8..ddc8a5b8cd 100644
--- a/sdk/test/metrics/sum_aggregation_test.cc
+++ b/sdk/test/metrics/sum_aggregation_test.cc
@@ -3,6 +3,7 @@
 
 #include <gtest/gtest.h>
 #include <stddef.h>
+#include <cstdint>
 #include <initializer_list>
 #include <map>
 #include <string>
@@ -13,7 +14,9 @@
 
 #include "opentelemetry/common/macros.h"
 #include "opentelemetry/context/context.h"
+#include "opentelemetry/metrics/async_instruments.h"
 #include "opentelemetry/metrics/meter.h"
+#include "opentelemetry/metrics/observer_result.h"
 #include "opentelemetry/metrics/sync_instruments.h"
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/nostd/shared_ptr.h"
@@ -322,6 +325,126 @@ TEST(CounterToSumFilterAttributes, Double)
   });
 }
 
+TEST(CounterToSumFilterAttributes, DoubleObservable)
+{
+  MeterProvider mp;
+  auto m                      = mp.GetMeter("meter2", "version2", "schema2");
+  std::string instrument_unit = "ms";
+  std::string instrument_name = "counter2";
+  std::string instrument_desc = "counter metrics2";
+
+  opentelemetry::sdk::metrics::FilterAttributeMap allowedattr;
+
+  std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attrproc{
+      new opentelemetry::sdk::metrics::FilteringAttributesProcessor(allowedattr)};
+
+  std::shared_ptr<opentelemetry::sdk::metrics::AggregationConfig> dummy_aggregation_config{
+      new opentelemetry::sdk::metrics::AggregationConfig};
+  std::unique_ptr<MockMetricExporter> exporter(new MockMetricExporter());
+  std::shared_ptr<MetricReader> reader{new MockMetricReader(std::move(exporter))};
+  mp.AddMetricReader(reader);
+
+  std::unique_ptr<View> view{new View("view2", "view2_description", AggregationType::kSum,
+                                      dummy_aggregation_config, std::move(attrproc))};
+  std::unique_ptr<InstrumentSelector> instrument_selector{
+      new InstrumentSelector(InstrumentType::kObservableCounter, instrument_name, instrument_unit)};
+  std::unique_ptr<MeterSelector> meter_selector{new MeterSelector("meter2", "version2", "schema2")};
+  mp.AddView(std::move(instrument_selector), std::move(meter_selector), std::move(view));
+
+  auto c = m->CreateDoubleObservableCounter(instrument_name, instrument_desc, instrument_unit);
+  c->AddCallback(
+      [](opentelemetry::metrics::ObserverResult result, void * /*state */) {
+        auto observer_double =
+            nostd::get<nostd::shared_ptr<opentelemetry::metrics::ObserverResultT<double>>>(result);
+        observer_double->Observe(10, {{"version", 1}});
+        observer_double->Observe(20, {{"version", 2}});
+      },
+      nullptr);
+
+  reader->Collect([&](ResourceMetrics &rm) {
+    for (const ScopeMetrics &smd : rm.scope_metric_data_)
+    {
+      EXPECT_EQ(1, smd.metric_data_.size());
+      for (const MetricData &md : smd.metric_data_)
+      {
+        EXPECT_EQ(1, md.point_data_attr_.size());
+        EXPECT_EQ(
+            30.0,
+            opentelemetry::nostd::get<double>(
+                opentelemetry::nostd::get<SumPointData>(md.point_data_attr_[0].point_data).value_));
+        EXPECT_EQ(0, md.point_data_attr_[0].attributes.size());
+        EXPECT_EQ(md.point_data_attr_[0].attributes.end(),
+                  md.point_data_attr_[0].attributes.find("version"));
+      }
+    }
+    return true;
+  });
+}
+
+TEST(CounterToSumFilterAttributes, LongObservable)
+{
+  MeterProvider mp;
+  auto m                      = mp.GetMeter("meter2", "version2", "schema2");
+  std::string instrument_unit = "ms";
+  std::string instrument_name = "counter2";
+  std::string instrument_desc = "counter metrics2";
+
+  opentelemetry::sdk::metrics::FilterAttributeMap allowedattr;
+  allowedattr["attr1"] = true;
+  std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attrproc{
+      new opentelemetry::sdk::metrics::FilteringAttributesProcessor(allowedattr)};
+
+  std::shared_ptr<opentelemetry::sdk::metrics::AggregationConfig> dummy_aggregation_config{
+      new opentelemetry::sdk::metrics::AggregationConfig};
+  std::unique_ptr<MockMetricExporter> exporter(new MockMetricExporter());
+  std::shared_ptr<MetricReader> reader{new MockMetricReader(std::move(exporter))};
+  mp.AddMetricReader(reader);
+
+  std::unique_ptr<View> view{new View("view2", "view2_description", AggregationType::kSum,
+                                      dummy_aggregation_config, std::move(attrproc))};
+  std::unique_ptr<InstrumentSelector> instrument_selector{
+      new InstrumentSelector(InstrumentType::kObservableCounter, instrument_name, instrument_unit)};
+  std::unique_ptr<MeterSelector> meter_selector{new MeterSelector("meter2", "version2", "schema2")};
+  mp.AddView(std::move(instrument_selector), std::move(meter_selector), std::move(view));
+
+  auto c = m->CreateInt64ObservableCounter(instrument_name, instrument_desc, instrument_unit);
+  c->AddCallback(
+      [](opentelemetry::metrics::ObserverResult result, void * /*state */) {
+        auto observer_int64 =
+            nostd::get<nostd::shared_ptr<opentelemetry::metrics::ObserverResultT<int64_t>>>(result);
+        observer_int64->Observe(10, {{"attr1", "val1"}, {"attr2", "val2"}});
+        observer_int64->Observe(20, {{"attr1", "val1"}});
+        observer_int64->Observe(5, {{"attr3", "val1"}, {"attr2", "val2"}});
+      },
+      nullptr);
+
+  reader->Collect([&](ResourceMetrics &rm) {
+    for (const ScopeMetrics &smd : rm.scope_metric_data_)
+    {
+      EXPECT_EQ(1, smd.metric_data_.size());
+      for (const MetricData &md : smd.metric_data_)
+      {
+        EXPECT_EQ(2, md.point_data_attr_.size());
+        EXPECT_EQ(
+            30,
+            opentelemetry::nostd::get<int64_t>(
+                opentelemetry::nostd::get<SumPointData>(md.point_data_attr_[0].point_data).value_));
+        EXPECT_EQ(
+            5,
+            opentelemetry::nostd::get<int64_t>(
+                opentelemetry::nostd::get<SumPointData>(md.point_data_attr_[1].point_data).value_));
+        EXPECT_EQ(1, md.point_data_attr_[0].attributes.size());
+        EXPECT_NE(md.point_data_attr_[0].attributes.end(),
+                  md.point_data_attr_[0].attributes.find("attr1"));
+        EXPECT_EQ(0, md.point_data_attr_[1].attributes.size());
+        EXPECT_EQ(md.point_data_attr_[1].attributes.end(),
+                  md.point_data_attr_[1].attributes.find("attr1"));
+      }
+    }
+    return true;
+  });
+}
+
 TEST(CounterToSumFilterAttributesWithCardinalityLimit, Double)
 {
   MeterProvider mp;