Replacing CompletableFuture by CompletableResultCode (#2670)

Author: LikeTheSalad

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/exporters/LogRecordToDiskExporter.java

@@ -22,7 +22,7 @@ public final class LogRecordToDiskExporter implements LogRecordExporter {
   private final ExporterCallback<LogRecordData> callback;
   private static final ExporterCallback<LogRecordData> DEFAULT_CALLBACK =
       new NoopExporterCallback<>();
-  private static final Duration DEFAULT_EXPORT_TIMEOUT = Duration.ofSeconds(10);
+  private static final Duration DEFAULT_WRITE_TIMEOUT = Duration.ofSeconds(10);
 
   private LogRecordToDiskExporter(
       SignalStorageExporter<LogRecordData> storageExporter,
@@ -54,7 +54,7 @@ public CompletableResultCode shutdown() {
   public static final class Builder {
     private final SignalStorage.LogRecord storage;
     private ExporterCallback<LogRecordData> callback = DEFAULT_CALLBACK;
-    private Duration writeTimeout = DEFAULT_EXPORT_TIMEOUT;
+    private Duration writeTimeout = DEFAULT_WRITE_TIMEOUT;
 
     @CanIgnoreReturnValue
     public Builder setExporterCallback(ExporterCallback<LogRecordData> value) {

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/exporters/MetricToDiskExporter.java

@@ -25,7 +25,7 @@ public final class MetricToDiskExporter implements MetricExporter {
   private final AggregationTemporalitySelector aggregationTemporalitySelector;
   private final ExporterCallback<MetricData> callback;
   private static final ExporterCallback<MetricData> DEFAULT_CALLBACK = new NoopExporterCallback<>();
-  private static final Duration DEFAULT_EXPORT_TIMEOUT = Duration.ofSeconds(10);
+  private static final Duration DEFAULT_WRITE_TIMEOUT = Duration.ofSeconds(10);
 
   private MetricToDiskExporter(
       SignalStorageExporter<MetricData> storageExporter,
@@ -66,7 +66,7 @@ public static final class Builder {
     private AggregationTemporalitySelector aggregationTemporalitySelector =
         AggregationTemporalitySelector.alwaysCumulative();
     private ExporterCallback<MetricData> callback = DEFAULT_CALLBACK;
-    private Duration writeTimeout = DEFAULT_EXPORT_TIMEOUT;
+    private Duration writeTimeout = DEFAULT_WRITE_TIMEOUT;
 
     @CanIgnoreReturnValue
     public Builder setExporterCallback(ExporterCallback<MetricData> value) {

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/exporters/SpanToDiskExporter.java

@@ -21,7 +21,7 @@ public final class SpanToDiskExporter implements SpanExporter {
   private final SignalStorageExporter<SpanData> storageExporter;
   private final ExporterCallback<SpanData> callback;
   private static final ExporterCallback<SpanData> DEFAULT_CALLBACK = new NoopExporterCallback<>();
-  private static final Duration DEFAULT_EXPORT_TIMEOUT = Duration.ofSeconds(10);
+  private static final Duration DEFAULT_WRITE_TIMEOUT = Duration.ofSeconds(10);
 
   private SpanToDiskExporter(
       SignalStorageExporter<SpanData> storageExporter, ExporterCallback<SpanData> callback) {
@@ -52,7 +52,7 @@ public CompletableResultCode shutdown() {
   public static final class Builder {
     private final SignalStorage.Span storage;
     private ExporterCallback<SpanData> callback = DEFAULT_CALLBACK;
-    private Duration writeTimeout = DEFAULT_EXPORT_TIMEOUT;
+    private Duration writeTimeout = DEFAULT_WRITE_TIMEOUT;
 
     private Builder(SignalStorage.Span storage) {
       this.storage = storage;

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/exporters/SignalStorageExporter.java

@@ -5,16 +5,12 @@
 
 package io.opentelemetry.contrib.disk.buffering.internal.exporters;
 
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-
 import io.opentelemetry.contrib.disk.buffering.exporters.callback.ExporterCallback;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
-import io.opentelemetry.contrib.disk.buffering.storage.result.WriteResult;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import java.time.Duration;
 import java.util.Collection;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 /** Internal utility for common export to disk operations across all exporters. */
@@ -31,23 +27,26 @@ public SignalStorageExporter(
   }
 
   public CompletableResultCode exportToStorage(Collection<T> items) {
-    CompletableFuture<WriteResult> future = storage.write(items);
-    try {
-      WriteResult operation = future.get(writeTimeout.toMillis(), MILLISECONDS);
-      if (operation.isSuccessful()) {
-        callback.onExportSuccess(items);
-        return CompletableResultCode.ofSuccess();
-      }
+    CompletableResultCode result =
+        storage.write(items).join(writeTimeout.toMillis(), TimeUnit.MILLISECONDS);
 
-      Throwable error = operation.getError();
+    if (!result.isDone()) {
+      TimeoutException timeout =
+          new TimeoutException("Storage write timed out after " + writeTimeout.toMillis() + "ms");
+      callback.onExportError(items, timeout);
+      return CompletableResultCode.ofExceptionalFailure(timeout);
+    }
+
+    if (!result.isSuccess()) {
+      Throwable error = result.getFailureThrowable();
       callback.onExportError(items, error);
       if (error != null) {
         return CompletableResultCode.ofExceptionalFailure(error);
       }
       return CompletableResultCode.ofFailure();
-    } catch (ExecutionException | InterruptedException | TimeoutException e) {
-      callback.onExportError(items, e);
-      return CompletableResultCode.ofExceptionalFailure(e);
     }
+
+    callback.onExportSuccess(items);
+    return CompletableResultCode.ofSuccess();
   }
 }

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/storage/FileSignalStorage.java

@@ -8,11 +8,10 @@
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
-import io.opentelemetry.contrib.disk.buffering.storage.result.WriteResult;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Iterator;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -46,34 +45,34 @@ public FileSignalStorage(
   }
 
   @Override
-  public CompletableFuture<WriteResult> write(Collection<T> items) {
+  public CompletableResultCode write(Collection<T> items) {
     logger.finer("Intercepting batch.");
     try {
       serializer.initialize(items);
       if (storage.write(serializer)) {
-        return CompletableFuture.completedFuture(WriteResult.successful());
+        return CompletableResultCode.ofSuccess();
       }
       logger.fine("Could not store batch in disk.");
-      return CompletableFuture.completedFuture(
-          WriteResult.error(new Exception("Could not store batch in disk for an unknown reason.")));
+      return CompletableResultCode.ofExceptionalFailure(
+          new Exception("Could not store batch in disk for an unknown reason."));
     } catch (IOException e) {
       logger.log(
           Level.WARNING,
           "An unexpected error happened while attempting to write the data in disk.",
           e);
-      return CompletableFuture.completedFuture(WriteResult.error(e));
+      return CompletableResultCode.ofExceptionalFailure(e);
     } finally {
       serializer.reset();
     }
   }
 
   @Override
-  public CompletableFuture<WriteResult> clear() {
+  public CompletableResultCode clear() {
     try {
       storage.clear();
-      return CompletableFuture.completedFuture(WriteResult.successful());
+      return CompletableResultCode.ofSuccess();
     } catch (IOException e) {
-      return CompletableFuture.completedFuture(WriteResult.error(e));
+      return CompletableResultCode.ofExceptionalFailure(e);
     }
   }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/storage/SignalStorage.java

@@ -5,13 +5,12 @@
 
 package io.opentelemetry.contrib.disk.buffering.storage;
 
-import io.opentelemetry.contrib.disk.buffering.storage.result.WriteResult;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import java.io.Closeable;
 import java.util.Collection;
-import java.util.concurrent.CompletableFuture;
 
 /**
  * Allows writing and iterating over written signal items.
@@ -24,16 +23,16 @@ public interface SignalStorage<T> extends Iterable<Collection<T>>, Closeable {
    * Stores signal items.
    *
    * @param items The items to be stored.
-   * @return A future with {@link WriteResult}.
+   * @return A {@link CompletableResultCode} representing the outcome of the write operation.
    */
-  CompletableFuture<WriteResult> write(Collection<T> items);
+  CompletableResultCode write(Collection<T> items);
 
   /**
    * Removes all the previously stored items.
    *
-   * @return A future with {@link WriteResult}.
+   * @return A {@link CompletableResultCode} representing the outcome of the clear operation.
    */
-  CompletableFuture<WriteResult> clear();
+  CompletableResultCode clear();
 
   /**
    * Abstraction for Spans. Implementations should use this instead of {@link SignalStorage}

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/storage/impl/FileLogRecordStorage.java

@@ -11,14 +11,13 @@
 import io.opentelemetry.contrib.disk.buffering.internal.storage.FolderManager;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
-import io.opentelemetry.contrib.disk.buffering.storage.result.WriteResult;
 import io.opentelemetry.sdk.common.Clock;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
 import java.io.File;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Iterator;
-import java.util.concurrent.CompletableFuture;
 import javax.annotation.Nonnull;
 
 public final class FileLogRecordStorage implements SignalStorage.LogRecord {
@@ -45,12 +44,12 @@ private FileLogRecordStorage(FileSignalStorage<LogRecordData> fileSignalStorage)
   }
 
   @Override
-  public CompletableFuture<WriteResult> write(Collection<LogRecordData> items) {
+  public CompletableResultCode write(Collection<LogRecordData> items) {
     return fileSignalStorage.write(items);
   }
 
   @Override
-  public CompletableFuture<WriteResult> clear() {
+  public CompletableResultCode clear() {
     return fileSignalStorage.clear();
   }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/storage/impl/FileMetricStorage.java

@@ -11,14 +11,13 @@
 import io.opentelemetry.contrib.disk.buffering.internal.storage.FolderManager;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
-import io.opentelemetry.contrib.disk.buffering.storage.result.WriteResult;
 import io.opentelemetry.sdk.common.Clock;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import java.io.File;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Iterator;
-import java.util.concurrent.CompletableFuture;
 import javax.annotation.Nonnull;
 
 public final class FileMetricStorage implements SignalStorage.Metric {
@@ -45,12 +44,12 @@ private FileMetricStorage(FileSignalStorage<MetricData> fileSignalStorage) {
   }
 
   @Override
-  public CompletableFuture<WriteResult> write(Collection<MetricData> items) {
+  public CompletableResultCode write(Collection<MetricData> items) {
     return fileSignalStorage.write(items);
   }
 
   @Override
-  public CompletableFuture<WriteResult> clear() {
+  public CompletableResultCode clear() {
     return fileSignalStorage.clear();
   }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/storage/impl/FileSpanStorage.java

@@ -11,14 +11,13 @@
 import io.opentelemetry.contrib.disk.buffering.internal.storage.FolderManager;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
-import io.opentelemetry.contrib.disk.buffering.storage.result.WriteResult;
 import io.opentelemetry.sdk.common.Clock;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import java.io.File;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Iterator;
-import java.util.concurrent.CompletableFuture;
 import javax.annotation.Nonnull;
 
 public final class FileSpanStorage implements SignalStorage.Span {
@@ -45,12 +44,12 @@ private FileSpanStorage(FileSignalStorage<SpanData> fileSignalStorage) {
   }
 
   @Override
-  public CompletableFuture<WriteResult> write(Collection<SpanData> items) {
+  public CompletableResultCode write(Collection<SpanData> items) {
     return fileSignalStorage.write(items);
   }
 
   @Override
-  public CompletableFuture<WriteResult> clear() {
+  public CompletableResultCode clear() {
     return fileSignalStorage.clear();
   }