Make OpenLineageClient and Transports AutoCloseable (OpenLineage#3122)

Signed-off-by: Dominik Dębowczyk <[email protected]>

client/java/src/main/java/io/openlineage/client/OpenLineageClient.java

@@ -22,7 +22,7 @@
 
 /** HTTP client used to emit {@link OpenLineage.RunEvent}s to HTTP backend. */
 @Slf4j
-public final class OpenLineageClient {
+public final class OpenLineageClient implements AutoCloseable {
   final Transport transport;
   final Optional<CircuitBreaker> circuitBreaker;
   final MeterRegistry meterRegistry;
@@ -160,6 +160,12 @@ public static Builder builder() {
     return new Builder();
   }
 
+  @Override
+  public void close() throws Exception {
+    transport.close();
+    meterRegistry.close();
+  }
+
   /**
    * Builder for {@link OpenLineageClient} instances.
    *

client/java/src/main/java/io/openlineage/client/transports/CompositeTransport.java

@@ -6,6 +6,7 @@
 package io.openlineage.client.transports;
 
 import io.openlineage.client.OpenLineage;
+import io.openlineage.client.OpenLineageClientException;
 import java.util.ArrayList;
 import java.util.List;
 import lombok.NonNull;
@@ -73,4 +74,17 @@ private void handleEmissionFailure(Transport transport, Exception e) {
           "Transport " + transport.getClass().getSimpleName() + " failed to emit event", e);
     }
   }
+
+  @Override
+  public void close() throws Exception {
+    transports.forEach(
+        t -> {
+          try {
+            t.close();
+          } catch (Exception e) {
+            log.error("Failed to close {} transport", t.getClass().getSimpleName(), e);
+            throw new OpenLineageClientException(e);
+          }
+        });
+  }
 }

client/java/src/main/java/io/openlineage/client/transports/HttpTransport.java

@@ -14,7 +14,6 @@
 import io.openlineage.client.OpenLineage;
 import io.openlineage.client.OpenLineageClientException;
 import io.openlineage.client.OpenLineageClientUtils;
-import java.io.Closeable;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -39,7 +38,7 @@
 import org.apache.http.util.EntityUtils;
 
 @Slf4j
-public final class HttpTransport extends Transport implements Closeable {
+public final class HttpTransport extends Transport {
   private static final String API_V1 = "/api/v1";
 
   private final CloseableHttpClient http;

client/java/src/main/java/io/openlineage/client/transports/KafkaTransport.java

@@ -99,4 +99,9 @@ private void emit(String eventAsJson, String eventKey) {
       log.error("Failed to collect lineage event: {}", eventAsJson, e);
     }
   }
+
+  @Override
+  public void close() throws Exception {
+    producer.close();
+  }
 }

client/java/src/main/java/io/openlineage/client/transports/Transport.java

@@ -10,7 +10,7 @@
 import lombok.NonNull;
 
 @NoArgsConstructor
-public abstract class Transport {
+public abstract class Transport implements AutoCloseable {
   enum Type {
     CONSOLE,
     FILE,
@@ -28,4 +28,7 @@ enum Type {
   public abstract void emit(@NonNull OpenLineage.DatasetEvent datasetEvent);
 
   public abstract void emit(@NonNull OpenLineage.JobEvent jobEvent);
+
+  @Override
+  public void close() throws Exception {}
 }

client/java/transports-dataplex/src/main/java/io/openlineage/client/transports/dataplex/DataplexTransport.java

@@ -36,7 +36,7 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
-public class DataplexTransport extends Transport implements Closeable {
+public class DataplexTransport extends Transport {
 
   private final ProducerClientWrapper producerClientWrapper;
 

client/java/transports-gcs/src/main/java/io/openlineage/client/transports/gcs/GcsTransport.java

@@ -97,4 +97,9 @@ private void uploadObject(String objectName, String contents) {
     Blob blob = storage.create(blobInfo, content);
     log.debug("Stored event: {}", blob.asBlobInfo().getBlobId().toGsUtilUri());
   }
+
+  @Override
+  public void close() throws Exception {
+    storage.close();
+  }
 }