fix(metrics): use weakref in fork handler to allow GC after provider shutdown

 os.register_at_fork holds callbacks for the lifetime of the process with
 no unregister API. Passing a bound method created a permanent strong
 reference chain: fork callback → SynchronousMeasurementConsumer →
 _reader_storages → MetricReader → exporter, preventing garbage collection
 even after MeterProvider.shutdown().

 Replace the bound method with a weakref-based closure so the consumer and
 everything it owns can be collected normally. The fork handler is a no-op
 once the consumer is gone.

 Fixes test_meter_provider_shutdown_cleans_up_successfully.

Author: dshivashankar1994

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py

@@ -4,6 +4,7 @@
 # pylint: disable=unused-import
 
 import os
+import weakref
 from abc import ABC, abstractmethod
 from collections.abc import Mapping
 from threading import Lock
@@ -68,7 +69,14 @@ def __init__(
         ] = []
         self._needs_storage_reinit = False
         if hasattr(os, "register_at_fork"):
-            os.register_at_fork(after_in_child=self._at_fork_reinit)
+            weak_self = weakref.ref(self)
+
+            def _fork_handler():
+                obj = weak_self()
+                if obj is not None:
+                    obj._at_fork_reinit()  # pylint: disable=protected-access
+
+            os.register_at_fork(after_in_child=_fork_handler)
 
     def _at_fork_reinit(self):
         """Reinitialize lock in child process after fork"""

opentelemetry-sdk/tests/metrics/test_measurement_consumer.py

@@ -208,8 +208,10 @@ def test_collect_deadline(
     "opentelemetry.sdk.metrics._internal."
     "measurement_consumer.MetricReaderStorage"
 )
-class TestSynchronousMeasurementConsumerForkHandler(TestCase):  # pylint: disable=protected-access
-    """Exhaustive tests for fork handler, needs_storage_reinit, and lazy _reinit_storages."""
+class TestSynchronousMeasurementConsumerForkHandler(TestCase):
+     """Exhaustive tests for fork handler, needs_storage_reinit, and lazy _reinit_storages."""
+
+    # pylint: disable=protected-access
 
     def test_register_at_fork_called_when_available(
         self, MockMetricReaderStorage