fix(config): allow deflate for OTLP HTTP exporters

Map declarative OTLP compression values through a shared helper that recognizes Deflate when the exporter enum supports it, while leaving gRPC validation unchanged.

Add regression coverage for the shared mapping helper plus tracer and meter provider HTTP exporter construction.

Signed-off-by: Asish Kumar <officialasishkumar@gmail.com>

Author: officialasishkumar

CHANGELOG.md

@@ -22,10 +22,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4907](https://github.com/open-telemetry/opentelemetry-python/issues/4907))
 - Drop Python 3.9 support
   ([#5076](https://github.com/open-telemetry/opentelemetry-python/pull/5076))
+- `opentelemetry-sdk`: Allow declarative OTLP HTTP exporters to map `compression: deflate` instead of rejecting it as unsupported
+  ([#5075](https://github.com/open-telemetry/opentelemetry-python/pull/5075))
 - `opentelemetry-semantic-conventions`: use `X | Y` union annotation
   ([#5096](https://github.com/open-telemetry/opentelemetry-python/pull/5096))
 
-
 ## Version 1.41.0/0.62b0 (2026-04-09)
 
 

opentelemetry-sdk/src/opentelemetry/sdk/_configuration/_common.py

@@ -17,6 +17,8 @@
 import logging
 from typing import Optional
 
+from opentelemetry.sdk._configuration._exceptions import ConfigurationError
+
 _logger = logging.getLogger(__name__)
 
 
@@ -47,3 +49,24 @@ def _parse_headers(
         for pair in headers:
             result[pair.name] = pair.value or ""
     return result
+
+
+def _map_compression(
+    value: Optional[str], compression_enum: type
+) -> Optional[object]:
+    """Map a compression string to the given Compression enum value."""
+    if value is None or value.lower() == "none":
+        return None
+    if value.lower() == "gzip":
+        return compression_enum.Gzip  # type: ignore[attr-defined]
+    if value.lower() == "deflate" and hasattr(compression_enum, "Deflate"):
+        return compression_enum.Deflate  # type: ignore[attr-defined]
+
+    supported_values = ["'gzip'", "'none'"]
+    if hasattr(compression_enum, "Deflate"):
+        supported_values.insert(1, "'deflate'")
+
+    raise ConfigurationError(
+        f"Unsupported compression value '{value}'. Supported values: "
+        f"{', '.join(supported_values)}."
+    )

opentelemetry-sdk/src/opentelemetry/sdk/_configuration/_meter_provider.py

@@ -18,7 +18,10 @@
 from typing import Optional, Set, Type
 
 from opentelemetry import metrics
-from opentelemetry.sdk._configuration._common import _parse_headers
+from opentelemetry.sdk._configuration._common import (
+    _map_compression,
+    _parse_headers,
+)
 from opentelemetry.sdk._configuration._exceptions import ConfigurationError
 from opentelemetry.sdk._configuration.models import (
     Aggregation as AggregationConfig,
@@ -265,19 +268,6 @@ def _create_console_metric_exporter(
     )
 
 
-def _map_compression_metric(
-    value: Optional[str], compression_enum: type
-) -> Optional[object]:
-    """Map a compression string to the given Compression enum value."""
-    if value is None or value.lower() == "none":
-        return None
-    if value.lower() == "gzip":
-        return compression_enum.Gzip  # type: ignore[attr-defined]
-    raise ConfigurationError(
-        f"Unsupported compression value '{value}'. Supported values: 'gzip', 'none'."
-    )
-
-
 def _create_otlp_http_metric_exporter(
     config: OtlpHttpMetricExporterConfig,
 ) -> MetricExporter:
@@ -296,7 +286,7 @@ def _create_otlp_http_metric_exporter(
             "Install it with: pip install opentelemetry-exporter-otlp-proto-http"
         ) from exc
 
-    compression = _map_compression_metric(config.compression, Compression)
+    compression = _map_compression(config.compression, Compression)
     headers = _parse_headers(config.headers, config.headers_list)
     timeout = (config.timeout / 1000.0) if config.timeout is not None else None
     preferred_temporality = _map_temporality(config.temporality_preference)
@@ -331,7 +321,7 @@ def _create_otlp_grpc_metric_exporter(
             "Install it with: pip install opentelemetry-exporter-otlp-proto-grpc"
         ) from exc
 
-    compression = _map_compression_metric(config.compression, grpc.Compression)
+    compression = _map_compression(config.compression, grpc.Compression)
     headers = _parse_headers(config.headers, config.headers_list)
     timeout = (config.timeout / 1000.0) if config.timeout is not None else None
     preferred_temporality = _map_temporality(config.temporality_preference)

opentelemetry-sdk/src/opentelemetry/sdk/_configuration/_tracer_provider.py

@@ -18,7 +18,10 @@
 from typing import Optional
 
 from opentelemetry import trace
-from opentelemetry.sdk._configuration._common import _parse_headers
+from opentelemetry.sdk._configuration._common import (
+    _map_compression,
+    _parse_headers,
+)
 from opentelemetry.sdk._configuration._exceptions import ConfigurationError
 from opentelemetry.sdk._configuration.models import (
     OtlpGrpcExporter as OtlpGrpcExporterConfig,
@@ -103,20 +106,6 @@ def _create_otlp_http_span_exporter(
         compression=compression,  # type: ignore[arg-type]
     )
 
-
-def _map_compression(
-    value: Optional[str], compression_enum: type
-) -> Optional[object]:
-    """Map a compression string to the given Compression enum value."""
-    if value is None or value.lower() == "none":
-        return None
-    if value.lower() == "gzip":
-        return compression_enum.Gzip  # type: ignore[attr-defined]
-    raise ConfigurationError(
-        f"Unsupported compression value '{value}'. Supported values: 'gzip', 'none'."
-    )
-
-
 def _create_otlp_grpc_span_exporter(
     config: OtlpGrpcExporterConfig,
 ) -> SpanExporter:

opentelemetry-sdk/tests/_configuration/test_common.py

@@ -15,7 +15,11 @@
 import unittest
 from types import SimpleNamespace
 
-from opentelemetry.sdk._configuration._common import _parse_headers
+from opentelemetry.sdk._configuration._common import (
+    _map_compression,
+    _parse_headers,
+)
+from opentelemetry.sdk._configuration._exceptions import ConfigurationError
 
 
 class TestParseHeaders(unittest.TestCase):
@@ -79,3 +83,53 @@ def test_struct_headers_override_headers_list(self):
 
     def test_both_empty_struct_and_none_list_returns_empty_dict(self):
         self.assertEqual(_parse_headers([], None), {})
+
+
+class _CompressionWithDeflate:
+    Gzip = "gzip"
+    Deflate = "deflate"
+
+
+class _CompressionWithoutDeflate:
+    Gzip = "gzip"
+
+
+class TestMapCompression(unittest.TestCase):
+    def test_none_returns_none(self):
+        self.assertIsNone(_map_compression(None, _CompressionWithDeflate))
+
+    def test_none_string_returns_none(self):
+        self.assertIsNone(
+            _map_compression("none", _CompressionWithDeflate)
+        )
+
+    def test_gzip_maps_to_gzip(self):
+        self.assertEqual(
+            _map_compression("gzip", _CompressionWithDeflate), "gzip"
+        )
+
+    def test_deflate_maps_when_supported(self):
+        self.assertEqual(
+            _map_compression("deflate", _CompressionWithDeflate),
+            "deflate",
+        )
+
+    def test_deflate_raises_when_unsupported(self):
+        with self.assertRaises(ConfigurationError) as ctx:
+            _map_compression("deflate", _CompressionWithoutDeflate)
+
+        self.assertEqual(
+            str(ctx.exception),
+            "Unsupported compression value 'deflate'. Supported values: "
+            "'gzip', 'none'.",
+        )
+
+    def test_http_error_message_includes_deflate(self):
+        with self.assertRaises(ConfigurationError) as ctx:
+            _map_compression("brotli", _CompressionWithDeflate)
+
+        self.assertEqual(
+            str(ctx.exception),
+            "Unsupported compression value 'brotli'. Supported values: "
+            "'gzip', 'deflate', 'none'.",
+        )

opentelemetry-sdk/tests/_configuration/test_meter_provider.py

@@ -274,6 +274,34 @@ def test_otlp_http_created_with_endpoint(self):
         self.assertIsNone(kwargs["timeout"])
         self.assertIsNone(kwargs["compression"])
 
+    def test_otlp_http_created_with_deflate_compression(self):
+        mock_exporter_cls = MagicMock()
+        mock_compression_cls = MagicMock()
+        mock_compression_cls.Deflate = "deflate_val"
+        mock_http_module = MagicMock()
+        mock_http_module.Compression = mock_compression_cls
+        mock_module = MagicMock()
+        mock_module.OTLPMetricExporter = mock_exporter_cls
+
+        with patch.dict(
+            sys.modules,
+            {
+                "opentelemetry.exporter.otlp.proto.http.metric_exporter": mock_module,
+                "opentelemetry.exporter.otlp.proto.http": mock_http_module,
+            },
+        ):
+            config = self._make_periodic_config(
+                PushMetricExporterConfig(
+                    otlp_http=OtlpHttpMetricExporterConfig(
+                        compression="deflate"
+                    )
+                )
+            )
+            create_meter_provider(config)
+
+        _, kwargs = mock_exporter_cls.call_args
+        self.assertEqual(kwargs["compression"], "deflate_val")
+
     def test_otlp_grpc_missing_package_raises(self):
         config = self._make_periodic_config(
             PushMetricExporterConfig(otlp_grpc=OtlpGrpcMetricExporterConfig())

opentelemetry-sdk/tests/_configuration/test_tracer_provider.py

@@ -309,6 +309,32 @@ def test_otlp_http_created_with_endpoint(self):
             compression=None,
         )
 
+    def test_otlp_http_created_with_deflate_compression(self):
+        mock_exporter_cls = MagicMock()
+        mock_compression_cls = MagicMock()
+        mock_compression_cls.Deflate = "deflate_val"
+        mock_module = MagicMock()
+        mock_module.OTLPSpanExporter = mock_exporter_cls
+        mock_http_module = MagicMock()
+        mock_http_module.Compression = mock_compression_cls
+
+        with patch.dict(
+            sys.modules,
+            {
+                "opentelemetry.exporter.otlp.proto.http.trace_exporter": mock_module,
+                "opentelemetry.exporter.otlp.proto.http": mock_http_module,
+            },
+        ):
+            config = self._make_batch_config(
+                SpanExporterConfig(
+                    otlp_http=OtlpHttpExporterConfig(compression="deflate")
+                )
+            )
+            create_tracer_provider(config)
+
+        _, kwargs = mock_exporter_cls.call_args
+        self.assertEqual(kwargs["compression"], "deflate_val")
+
     def test_otlp_http_headers_list(self):
         mock_exporter_cls = MagicMock()
         mock_http_module = MagicMock()