Add compression option to OpenLineage HttpTransport

Author: dolfinus

docs/src/main/sphinx/admin/event-listeners-openlineage.md

@@ -210,6 +210,11 @@ event-listener.config-files=etc/openlineage-event-listener.properties,...
     - List of custom url params to be added to final HTTP Request. See
     [](openlineage-event-listener-custom-url-params) for more details.
     - Empty
+*
+    - openlineage-event-listener.transport.compression
+    - Compression codec used for reducing size of HTTP body.
+      Allowed values: `none`, `gzip`.
+    - `none`
 
 :::
 

plugin/trino-openlineage/src/main/java/io/trino/plugin/openlineage/transport/http/OpenLineageHttpTransport.java

@@ -23,6 +23,7 @@
 import java.net.URI;
 import java.util.Map;
 
+import static io.trino.plugin.openlineage.transport.http.OpenLineageHttpTransportConfig.Compression;
 import static java.lang.Math.toIntExact;
 
 public class OpenLineageHttpTransport
@@ -34,6 +35,7 @@ public class OpenLineageHttpTransport
     private final TokenProvider tokenProvider;
     private final Map<String, String> urlParams;
     private final Map<String, String> headers;
+    private final Compression compression;
 
     @Inject
     public OpenLineageHttpTransport(OpenLineageHttpTransportConfig config)
@@ -44,6 +46,7 @@ public OpenLineageHttpTransport(OpenLineageHttpTransportConfig config)
         this.tokenProvider = config.getApiKey().map(OpenLineageHttpTransport::createTokenProvider).orElse(null);
         this.urlParams = config.getUrlParams();
         this.headers = config.getHeaders();
+        this.compression = config.getCompression();
     }
 
     @Override
@@ -57,7 +60,7 @@ public HttpTransport buildTransport()
                         this.tokenProvider,
                         this.urlParams,
                         this.headers,
-                        null,
+                        this.compression == Compression.GZIP ? HttpConfig.Compression.GZIP : null,
                         new HttpSslContextConfig()));
     }
 

plugin/trino-openlineage/src/main/java/io/trino/plugin/openlineage/transport/http/OpenLineageHttpTransportConfig.java

@@ -33,12 +33,20 @@
 
 public class OpenLineageHttpTransportConfig
 {
+    public enum Compression
+    {
+        NONE, GZIP
+    }
+
     private URI url;
     private String endpoint;
     private Optional<String> apiKey = Optional.empty();
     private Duration timeout = new Duration(5000, TimeUnit.MILLISECONDS);
     private Map<String, String> headers = new HashMap<>();
     private Map<String, String> urlParams = new HashMap<>();
+    // Most of OpenLineage HTTP servers support GZIP compression,
+    // but users may use custom servers that don't. Use NONE for compatibility
+    private Compression compression = Compression.NONE;
 
     @NotNull
     public URI getUrl()
@@ -137,4 +145,18 @@ public OpenLineageHttpTransportConfig setUrlParams(List<String> urlParas)
         }
         return this;
     }
+
+    @NotNull
+    public Compression getCompression()
+    {
+        return compression;
+    }
+
+    @Config("openlineage-event-listener.transport.compression")
+    @ConfigDescription("Compression codec using for HTTP body")
+    public OpenLineageHttpTransportConfig setCompression(Compression compression)
+    {
+        this.compression = compression;
+        return this;
+    }
 }

plugin/trino-openlineage/src/test/java/io/trino/plugin/openlineage/TestOpenLineageEventListenerMarquezIntegration.java

@@ -42,6 +42,7 @@ protected QueryRunner createQueryRunner()
         return OpenLineageListenerQueryRunner.builder()
                 .addListenerProperty("openlineage-event-listener.transport.type", "HTTP")
                 .addListenerProperty("openlineage-event-listener.transport.url", server.getMarquezUri().toString())
+                .addListenerProperty("openlineage-event-listener.transport.compression", "gzip")
                 .addListenerProperty("openlineage-event-listener.trino.uri", TRINO_URI)
                 .build();
     }

plugin/trino-openlineage/src/test/java/io/trino/plugin/openlineage/http/TestOpenLineageHttpTransportConfig.java

@@ -37,7 +37,8 @@ void testDefaults()
                 .setTimeout(Duration.valueOf("5s"))
                 .setApiKey(null)
                 .setHeaders(ImmutableList.of())
-                .setUrlParams(ImmutableList.of()));
+                .setUrlParams(ImmutableList.of())
+                .setCompression(OpenLineageHttpTransportConfig.Compression.NONE));
     }
 
     @Test
@@ -51,7 +52,7 @@ void testExplicitPropertyMappings()
                 .put("openlineage-event-listener.transport.timeout", "30s")
                 .put("openlineage-event-listener.transport.headers", "header1:value1,header2:value2")
                 .put("openlineage-event-listener.transport.url-params", "urlParam1:urlVal1,urlParam2:urlVal2")
-
+                .put("openlineage-event-listener.transport.compression", "gzip")
                 .buildOrThrow();
 
         OpenLineageHttpTransportConfig expected = new OpenLineageHttpTransportConfig()
@@ -60,7 +61,8 @@ void testExplicitPropertyMappings()
                 .setApiKey("dummy")
                 .setTimeout(Duration.valueOf("30s"))
                 .setHeaders(ImmutableList.of("header1:value1", "header2:value2"))
-                .setUrlParams(ImmutableList.of("urlParam1:urlVal1", "urlParam2:urlVal2"));
+                .setUrlParams(ImmutableList.of("urlParam1:urlVal1", "urlParam2:urlVal2"))
+                .setCompression(OpenLineageHttpTransportConfig.Compression.GZIP);
 
         assertFullMapping(properties, expected);
     }