JAVA-2557: Accept any negative length when decoding elements of tuples and UDTs (#1368)

diff --git a/changelog/README.md b/changelog/README.md
index 805ce20..e4d26b3 100644
--- a/changelog/README.md
+++ b/changelog/README.md

@@ -2,8 +2,9 @@
 
 <!-- Note: contrary to 3.x, insert new entries *first* in their section -->
 
-### 4.4.0 (in progress)
+### 4.3.1 (in progress)
 
+- [bug] JAVA-2557: Accept any negative length when decoding elements of tuples and UDTs
 
 ### 4.3.0
 

diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/TupleCodec.java b/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/TupleCodec.java
index 2a900ce..ad008cd 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/TupleCodec.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/TupleCodec.java

@@ -109,7 +109,7 @@
         }
         int elementSize = input.getInt();
         ByteBuffer element;
-        if (elementSize == -1) {
+        if (elementSize < 0) {
           element = null;
         } else {
           element = input.slice();

diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodec.java b/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodec.java
index 2e2df95..f515c6b 100644
--- a/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodec.java
+++ b/core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodec.java

@@ -112,7 +112,7 @@
         }
         int elementSize = input.getInt();
         ByteBuffer element;
-        if (elementSize == -1) {
+        if (elementSize < 0) {
           element = null;
         } else {
           element = input.slice();

diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/TupleCodecTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/TupleCodecTest.java
index f7d609e..ab95dd4 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/TupleCodecTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/TupleCodecTest.java

@@ -122,6 +122,26 @@
     verify(textCodec).decode(Bytes.fromHexString("0x61"), ProtocolVersion.DEFAULT);
   }
 
+  /** Test for JAVA-2557. Ensures that the codec can decode null fields with any negative length. */
+  @Test
+  public void should_decode_negative_element_length_as_null_field() {
+    TupleValue tuple =
+        decode(
+            "0x"
+                + "ffffffff" // field1 has length -1
+                + "fffffffe" // field2 has length -2
+                + "80000000" // field3 has length Integer.MIN_VALUE (-2147483648)
+            );
+
+    assertThat(tuple.isNull(0)).isTrue();
+    assertThat(tuple.isNull(1)).isTrue();
+    assertThat(tuple.isNull(2)).isTrue();
+
+    verifyZeroInteractions(intCodec);
+    verifyZeroInteractions(doubleCodec);
+    verifyZeroInteractions(textCodec);
+  }
+
   @Test
   public void should_format_null_tuple() {
     assertThat(format(null)).isEqualTo("NULL");

diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodecTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodecTest.java
index 5947cff..a3bd4fe 100644
--- a/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodecTest.java
+++ b/core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodecTest.java

@@ -131,6 +131,39 @@
     verify(textCodec).decode(Bytes.fromHexString("0x61"), ProtocolVersion.DEFAULT);
   }
 
+  /** Test for JAVA-2557. Ensures that the codec can decode null fields with any negative length. */
+  @Test
+  public void should_decode_negative_element_length_as_null_field() {
+    UdtValue udt =
+        decode(
+            "0x"
+                + "ffffffff" // field1 has length -1
+                + "fffffffe" // field2 has length -2
+                + "80000000" // field3 has length Integer.MIN_VALUE (-2147483648)
+            );
+
+    assertThat(udt.isNull(0)).isTrue();
+    assertThat(udt.isNull(1)).isTrue();
+    assertThat(udt.isNull(2)).isTrue();
+
+    verifyZeroInteractions(intCodec);
+    verifyZeroInteractions(doubleCodec);
+    verifyZeroInteractions(textCodec);
+  }
+
+  @Test
+  public void should_decode_absent_element_as_null_field() {
+    UdtValue udt = decode("0x");
+
+    assertThat(udt.isNull(0)).isTrue();
+    assertThat(udt.isNull(1)).isTrue();
+    assertThat(udt.isNull(2)).isTrue();
+
+    verifyZeroInteractions(intCodec);
+    verifyZeroInteractions(doubleCodec);
+    verifyZeroInteractions(textCodec);
+  }
+
   @Test
   public void should_format_null_udt() {
     assertThat(format(null)).isEqualTo("NULL");