fix(sdk): force_flush returns meaningful bool on MetricReader

Fixes #5020

Signed-off-by: Ravi Theja <ravitheja4531@gmail.com>

Author: ravitheja4531-cell

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#5093](https://github.com/open-telemetry/opentelemetry-python/pull/5093))
 - `opentelemetry-sdk`: fix YAML structure injection via environment variable substitution in declarative file configuration; values containing newlines are now emitted as quoted YAML scalars per spec requirement
   ([#5091](https://github.com/open-telemetry/opentelemetry-python/pull/5091))
+- `opentelemetry-sdk`: Fix `force_flush` on `MetricReader` and `PeriodicExportingMetricReader` to return a meaningful `bool` reflecting actual export success/failure instead of always returning `True`. Also fixes `detach(token)` not being called when export raises an exception. ([#5020](https://github.com/open-telemetry/opentelemetry-python/issues/5020))
 - `opentelemetry-sdk`: Add `create_logger_provider`/`configure_logger_provider` to declarative file configuration, enabling LoggerProvider instantiation from config files without reading env vars
   ([#4990](https://github.com/open-telemetry/opentelemetry-python/pull/4990))
 - `opentelemetry-sdk`: Add `service` resource detector support to declarative file configuration via `detection_development.detectors[].service`

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py

@@ -336,7 +336,7 @@ def __init__(
         )
 
     @final
-    def collect(self, timeout_millis: float = 10_000) -> None:
+    def collect(self, timeout_millis: float = 10_000) -> Optional[bool]:
         """Collects the metrics from the internal SDK state and
         invokes the `_receive_metrics` with the collection.
 
@@ -361,10 +361,11 @@ def collect(self, timeout_millis: float = 10_000) -> None:
             self._metrics.record_collection(perf_counter() - start_time)
 
         if metrics is not None:
-            self._receive_metrics(
+            return self._receive_metrics(
                 metrics,
                 timeout_millis=timeout_millis,
             )
+        return None
 
     @final
     def _set_collect_callback(
@@ -386,17 +387,25 @@ def _receive_metrics(
         metrics_data: MetricsData,
         timeout_millis: float = 10_000,
         **kwargs,
-    ) -> None:
-        """Called by `MetricReader.collect` when it receives a batch of metrics"""
+    ) -> bool:
+        """Called by `MetricReader.collect` when it receives a batch of metrics.
+
+        Subclasses must return ``True`` on success and ``False`` on failure.
+
+        .. note::
+            Existing subclasses that return ``None`` (the old implicit default)
+            will be treated as vacuous success by ``force_flush``, preserving
+            backward-compatible behaviour.
+        """
 
     def _set_meter_provider(self, meter_provider: MeterProvider) -> None:
         self._metrics = MetricReaderMetrics(
             self._otel_component_type, meter_provider
         )
 
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
-        self.collect(timeout_millis=timeout_millis)
-        return True
+        result = self.collect(timeout_millis=timeout_millis)
+        return result is not False
 
     @abstractmethod
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
@@ -449,9 +458,10 @@ def _receive_metrics(
         metrics_data: MetricsData,
         timeout_millis: float = 10_000,
         **kwargs,
-    ) -> None:
+    ) -> bool:
         with self._lock:
             self._metrics_data = metrics_data
+        return True
 
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
         pass
@@ -567,17 +577,19 @@ def _receive_metrics(
         metrics_data: MetricsData,
         timeout_millis: float = 10_000,
         **kwargs,
-    ) -> None:
+    ) -> bool:
         token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
-        # pylint: disable=broad-exception-caught,invalid-name
         try:
             with self._export_lock:
-                self._exporter.export(
+                result = self._exporter.export(
                     metrics_data, timeout_millis=timeout_millis
                 )
+                return result is MetricExportResult.SUCCESS
         except Exception:
             _logger.exception("Exception while exporting metrics")
-        detach(token)
+            return False
+        finally:
+            detach(token)
 
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
         deadline_ns = time_ns() + timeout_millis * 10**6
@@ -596,6 +608,6 @@ def _shutdown():
         self._exporter.shutdown(timeout=(deadline_ns - time_ns()) / 10**6)
 
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
-        super().force_flush(timeout_millis=timeout_millis)
-        self._exporter.force_flush(timeout_millis=timeout_millis)
-        return True
+        collect_ok = super().force_flush(timeout_millis=timeout_millis)
+        exporter_ok = self._exporter.force_flush(timeout_millis=timeout_millis)
+        return collect_ok and exporter_ok