Creating DeserializationException

Author: LikeTheSalad

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/exporter/FromDiskExporterImpl.java

@@ -5,6 +5,7 @@
 
 package io.opentelemetry.contrib.disk.buffering.internal.exporter;
 
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.DeserializationException;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.Storage;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.files.reader.ProcessResult;
@@ -71,7 +72,7 @@ public boolean exportStoredBatch(long timeout, TimeUnit unit) throws IOException
                     "Now exporting batch of " + telemetry.size() + " " + deserializer.signalType());
                 CompletableResultCode join = exportFunction.apply(telemetry).join(timeout, unit);
                 return join.isSuccess() ? ProcessResult.SUCCEEDED : ProcessResult.TRY_LATER;
-              } catch (IllegalArgumentException e) {
+              } catch (DeserializationException e) {
                 return ProcessResult.CONTENT_INVALID;
               }
             });

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/DeserializationException.java

@@ -0,0 +1,10 @@
+package io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers;
+
+import java.io.IOException;
+
+@SuppressWarnings("serial")
+public class DeserializationException extends IOException {
+  public DeserializationException(Throwable cause) {
+    super(cause);
+  }
+}

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/LogRecordDataDeserializer.java

@@ -22,11 +22,11 @@ static LogRecordDataDeserializer getInstance() {
   }
 
   @Override
-  public List<LogRecordData> deserialize(byte[] source) {
+  public List<LogRecordData> deserialize(byte[] source) throws DeserializationException {
     try {
       return ProtoLogsDataMapper.getInstance().fromProto(LogsData.ADAPTER.decode(source));
     } catch (IOException e) {
-      throw new IllegalArgumentException(e);
+      throw new DeserializationException(e);
     }
   }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/MetricDataDeserializer.java

@@ -22,11 +22,11 @@ static MetricDataDeserializer getInstance() {
   }
 
   @Override
-  public List<MetricData> deserialize(byte[] source) {
+  public List<MetricData> deserialize(byte[] source) throws DeserializationException {
     try {
       return ProtoMetricsDataMapper.getInstance().fromProto(MetricsData.ADAPTER.decode(source));
     } catch (IOException e) {
-      throw new IllegalArgumentException(e);
+      throw new DeserializationException(e);
     }
   }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/SignalDeserializer.java

@@ -25,7 +25,7 @@ static SignalDeserializer<LogRecordData> ofLogs() {
   }
 
   /** Deserializes the given byte array into a list of telemetry items. */
-  List<SDK_ITEM> deserialize(byte[] source);
+  List<SDK_ITEM> deserialize(byte[] source) throws DeserializationException;
 
   /** Returns the name of the stored type of signal -- one of "metrics", "spans", or "logs". */
   default String signalType() {

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/SpanDataDeserializer.java

@@ -22,11 +22,11 @@ static SpanDataDeserializer getInstance() {
   }
 
   @Override
-  public List<SpanData> deserialize(byte[] source) {
+  public List<SpanData> deserialize(byte[] source) throws DeserializationException {
     try {
       return ProtoSpansDataMapper.getInstance().fromProto(TracesData.ADAPTER.decode(source));
     } catch (IOException e) {
-      throw new IllegalArgumentException(e);
+      throw new DeserializationException(e);
     }
   }
 

disk-buffering/src/test/java/io/opentelemetry/contrib/disk/buffering/FromDiskExporterImplTest.java

@@ -14,6 +14,7 @@
 import static org.mockito.Mockito.when;
 
 import io.opentelemetry.contrib.disk.buffering.internal.exporter.FromDiskExporterImpl;
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.DeserializationException;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.storage.TestData;
 import io.opentelemetry.sdk.common.Clock;
@@ -88,12 +89,23 @@ void verifyStorageFolderIsCreated() {
     assertThat(new File(rootDir, STORAGE_FOLDER_NAME).exists()).isTrue();
   }
 
+  @Test
+  void whenDeserializationFails_returnFalse() throws IOException {
+    when(deserializer.deserialize(any()))
+        .thenAnswer(
+            invocation -> {
+              throw new DeserializationException(new IOException("Some exception"));
+            });
+
+    assertThat(exporter.exportStoredBatch(1, TimeUnit.SECONDS)).isFalse();
+  }
+
   private void createDummyFile() throws IOException {
     File file = new File(rootDir, STORAGE_FOLDER_NAME + "/" + 1000L);
     Files.write(file.toPath(), singletonList("First line"));
   }
 
-  private void setUpSerializer() {
+  private void setUpSerializer() throws DeserializationException {
     deserializer = mock();
     when(deserializer.deserialize(any())).thenReturn(deserializedData);
   }

disk-buffering/src/test/java/io/opentelemetry/contrib/disk/buffering/internal/storage/files/ReadableFileTest.java

@@ -19,6 +19,7 @@
 import io.opentelemetry.api.common.Value;
 import io.opentelemetry.api.logs.Severity;
 import io.opentelemetry.contrib.disk.buffering.internal.files.TemporaryFileProvider;
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.DeserializationException;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers.SignalDeserializer;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.logs.models.LogRecordDataImpl;
 import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
@@ -140,8 +141,7 @@ void whenProcessingSucceeds_returnSuccessStatus() throws IOException {
   @Test
   void whenProcessingFails_returnTryLaterStatus() throws IOException {
     assertEquals(
-        ReadableResult.TRY_LATER,
-        readableFile.readAndProcess(bytes -> ProcessResult.TRY_LATER));
+        ReadableResult.TRY_LATER, readableFile.readAndProcess(bytes -> ProcessResult.TRY_LATER));
   }
 
   @Test
@@ -250,6 +250,10 @@ private static List<LogRecordData> getRemainingDataAndClose(ReadableFile readabl
   }
 
   private static LogRecordData deserialize(byte[] data) {
-    return DESERIALIZER.deserialize(data).get(0);
+    try {
+      return DESERIALIZER.deserialize(data).get(0);
+    } catch (DeserializationException e) {
+      throw new RuntimeException(e);
+    }
   }
 }