Make UdtCodec logic tolerant to extra unknown fields at the tail of U…

…DT to support live schema upgrades

Patch by Dmitry Konstantinov; reviewed by TBD for CASSANDRA-19814

core/src/main/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodec.java

@@ -100,17 +100,14 @@ public UdtValue decode(@Nullable ByteBuffer bytes, @NonNull ProtocolVersion prot
     if (bytes == null) {
       return null;
     }
-    // empty byte buffers will result in empty values
     try {
       ByteBuffer input = bytes.duplicate();
       UdtValue value = cqlType.newValue();
       int i = 0;
       while (input.hasRemaining()) {
         if (i == cqlType.getFieldTypes().size()) {
-          throw new IllegalArgumentException(
-              String.format(
-                  "Too many fields in encoded UDT value, expected %d",
-                  cqlType.getFieldTypes().size()));
+          // ignores all unknown fields at the tail during a decoding
+          break;
         }
         int elementSize = input.getInt();
         ByteBuffer element;

core/src/test/java/com/datastax/oss/driver/internal/core/type/codec/UdtCodecTest.java

@@ -136,18 +136,23 @@ public void should_decode_udt() {
   }
 
   @Test
-  public void should_fail_to_decode_udt_when_too_many_fields() {
-    assertThatThrownBy(
-            () ->
-                decode(
-                    "0x"
-                        + ("00000004" + "00000001")
-                        + "ffffffff"
-                        + ("00000001" + "61")
-                        // extra contents
-                        + "ffffffff"))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessage("Too many fields in encoded UDT value, expected 3");
+  public void should_decode_udt_when_too_many_fields() {
+    UdtValue udt = decode("0x"
+            + ("00000004" + "00000001") // size and contents of int field 0
+            + "ffffffff" // null field 1
+            + ("00000001" + "61") // size and contents of String field 2
+            + ("00000004" + "00000002") // size and contents of int field 3, unknown for UDT version used to create the codec
+    );
+
+    assertThat(udt.getInt(0)).isEqualTo(1);
+    assertThat(udt.isNull(1)).isTrue();
+    assertThat(udt.getString(2)).isEqualTo("a");
+    assertThat(udt.size()).isEqualTo(3); // unknown field is not decoded
+
+
+    verify(intCodec).decodePrimitive(Bytes.fromHexString("0x00000001"), ProtocolVersion.DEFAULT);
+    verifyZeroInteractions(doubleCodec);
+    verify(textCodec).decode(Bytes.fromHexString("0x61"), ProtocolVersion.DEFAULT);
   }
 
   /** Test for JAVA-2557. Ensures that the codec can decode null fields with any negative length. */