Catching IllegalStateException in case of failed deserialization

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

@@ -26,7 +26,7 @@ public List<LogRecordData> deserialize(byte[] source) throws DeserializationExce
     try {
       return ProtoLogsDataMapper.getInstance()
           .fromProto(ExportLogsServiceRequest.ADAPTER.decode(source));
-    } catch (IOException e) {
+    } catch (IOException | IllegalStateException e) {
       throw new DeserializationException(e);
     }
   }

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

@@ -26,7 +26,7 @@ public List<MetricData> deserialize(byte[] source) throws DeserializationExcepti
     try {
       return ProtoMetricsDataMapper.getInstance()
           .fromProto(ExportMetricsServiceRequest.ADAPTER.decode(source));
-    } catch (IOException e) {
+    } catch (IOException | IllegalStateException e) {
       throw new DeserializationException(e);
     }
   }

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

@@ -26,7 +26,7 @@ public List<SpanData> deserialize(byte[] source) throws DeserializationException
     try {
       return ProtoSpansDataMapper.getInstance()
           .fromProto(ExportTraceServiceRequest.ADAPTER.decode(source));
-    } catch (IOException e) {
+    } catch (IOException | IllegalStateException e) {
       throw new DeserializationException(e);
     }
   }

disk-buffering/src/test/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/LogRecordDataDeserializerTest.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import io.opentelemetry.api.common.Value;
+import io.opentelemetry.api.logs.Severity;
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.logs.models.LogRecordDataImpl;
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
+import io.opentelemetry.contrib.disk.buffering.testutils.BaseSignalSerializerTest;
+import io.opentelemetry.contrib.disk.buffering.testutils.TestData;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import org.junit.jupiter.api.Test;
+
+class LogRecordDataDeserializerTest extends BaseSignalSerializerTest<LogRecordData> {
+  private static final LogRecordData LOG_RECORD =
+      LogRecordDataImpl.builder()
+          .setResource(TestData.RESOURCE_FULL)
+          .setSpanContext(TestData.SPAN_CONTEXT)
+          .setInstrumentationScopeInfo(TestData.INSTRUMENTATION_SCOPE_INFO_FULL)
+          .setAttributes(TestData.ATTRIBUTES)
+          .setBodyValue(Value.of("Log body"))
+          .setSeverity(Severity.DEBUG)
+          .setSeverityText("Log severity text")
+          .setTimestampEpochNanos(100L)
+          .setObservedTimestampEpochNanos(200L)
+          .setTotalAttributeCount(3)
+          .setEventName("event")
+          .build();
+
+  @Test
+  void verifyDeserialization() {
+    assertSerializeDeserialize(LOG_RECORD, LOG_RECORD);
+  }
+
+  @Test
+  void whenDecodingMalformedMessage_wrapIntoDeserializationException() {
+    assertThrows(
+        DeserializationException.class,
+        () -> getDeserializer().deserialize(TestData.makeMalformedSignalBinary()));
+  }
+
+  @Test
+  void whenDecodingTooShortMessage_wrapIntoDeserializationException() {
+    assertThrows(
+        DeserializationException.class,
+        () -> getDeserializer().deserialize(TestData.makeTooShortSignalBinary()));
+  }
+
+  @Override
+  protected SignalSerializer<LogRecordData> getSerializer() {
+    return SignalSerializer.ofLogs();
+  }
+
+  @Override
+  protected SignalDeserializer<LogRecordData> getDeserializer() {
+    return SignalDeserializer.ofLogs();
+  }
+}

disk-buffering/src/test/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/MetricDataDeserializerTest.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
+import io.opentelemetry.contrib.disk.buffering.testutils.BaseSignalSerializerTest;
+import io.opentelemetry.contrib.disk.buffering.testutils.TestData;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import org.junit.jupiter.api.Test;
+
+class MetricDataDeserializerTest extends BaseSignalSerializerTest<MetricData> {
+
+  @Test
+  void whenDecodingMalformedMessage_wrapIntoDeserializationException() {
+    assertThrows(
+        DeserializationException.class,
+        () -> getDeserializer().deserialize(TestData.makeMalformedSignalBinary()));
+  }
+
+  @Test
+  void whenDecodingTooShortMessage_wrapIntoDeserializationException() {
+    assertThrows(
+        DeserializationException.class,
+        () -> getDeserializer().deserialize(TestData.makeTooShortSignalBinary()));
+  }
+
+  @Override
+  protected SignalSerializer<MetricData> getSerializer() {
+    return SignalSerializer.ofMetrics();
+  }
+
+  @Override
+  protected SignalDeserializer<MetricData> getDeserializer() {
+    return SignalDeserializer.ofMetrics();
+  }
+}

disk-buffering/src/test/java/io/opentelemetry/contrib/disk/buffering/internal/serialization/deserializers/SpanDataDeserializerTest.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.disk.buffering.internal.serialization.deserializers;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.mapping.spans.models.SpanDataImpl;
+import io.opentelemetry.contrib.disk.buffering.internal.serialization.serializers.SignalSerializer;
+import io.opentelemetry.contrib.disk.buffering.testutils.BaseSignalSerializerTest;
+import io.opentelemetry.contrib.disk.buffering.testutils.TestData;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.data.StatusData;
+import java.util.Collections;
+import org.junit.jupiter.api.Test;
+
+class SpanDataDeserializerTest extends BaseSignalSerializerTest<SpanData> {
+  private static final SpanData SPAN_DATA =
+      SpanDataImpl.builder()
+          .setSpanContext(TestData.SPAN_CONTEXT)
+          .setParentSpanContext(TestData.PARENT_SPAN_CONTEXT)
+          .setName("Test span")
+          .setKind(SpanKind.SERVER)
+          .setStartEpochNanos(100L)
+          .setEndEpochNanos(200L)
+          .setStatus(StatusData.ok())
+          .setAttributes(TestData.ATTRIBUTES)
+          .setResource(TestData.RESOURCE_FULL)
+          .setInstrumentationScopeInfo(TestData.INSTRUMENTATION_SCOPE_INFO_FULL)
+          .setTotalRecordedLinks(0)
+          .setTotalRecordedEvents(0)
+          .setTotalAttributeCount(0)
+          .setEvents(Collections.emptyList())
+          .setLinks(Collections.emptyList())
+          .build();
+
+  @Test
+  void verifyDeserialization() {
+    assertSerializeDeserialize(SPAN_DATA, SPAN_DATA);
+  }
+
+  @Test
+  void whenDecodingMalformedMessage_wrapIntoDeserializationException() {
+    assertThrows(
+        DeserializationException.class,
+        () -> getDeserializer().deserialize(TestData.makeMalformedSignalBinary()));
+  }
+
+  @Test
+  void whenDecodingTooShortMessage_wrapIntoDeserializationException() {
+    assertThrows(
+        DeserializationException.class,
+        () -> getDeserializer().deserialize(TestData.makeTooShortSignalBinary()));
+  }
+
+  @Override
+  protected SignalSerializer<SpanData> getSerializer() {
+    return SignalSerializer.ofSpans();
+  }
+
+  @Override
+  protected SignalDeserializer<SpanData> getDeserializer() {
+    return SignalDeserializer.ofSpans();
+  }
+}

disk-buffering/src/test/java/io/opentelemetry/contrib/disk/buffering/testutils/TestData.java

@@ -131,5 +131,24 @@ private static LongExemplarData makeLongExemplarData(TraceFlags flags) {
     return ImmutableLongExemplarData.create(ATTRIBUTES, 100L, context, 1L);
   }
 
+  @NotNull
+  public static byte[] makeTooShortSignalBinary() {
+    return new byte[] {
+      (byte) 0x0A, // type
+      (byte) 0xFF, // defining length 255, but message is shorter
+      (byte) 0x01 // content
+    };
+  }
+
+  @NotNull
+  public static byte[] makeMalformedSignalBinary() {
+    return new byte[] {
+      (byte) 0x0A, // type
+      (byte) 0x02, // length
+      (byte) 0x08, // field 1, wire type 0 (varint) - this should be a nested message but isn't
+      (byte) 0x01 // content
+    };
+  }
+
   private TestData() {}
 }