3.x: Handle empty metadataid when metadataid feature is on

driver-core/src/main/java/com/datastax/driver/core/Requests.java

@@ -224,6 +224,8 @@ public String toString() {
 
   static class Execute extends Message.Request {
 
+    public static final byte[] EMPTY_BYTES = new byte[0];
+
     static final Message.Coder<Execute> coder =
         new Message.Coder<Execute>() {
           @Override
@@ -233,17 +235,23 @@ public void encode(
               ProtocolVersion version,
               ProtocolFeatureStore featureStore) {
             CBUtil.writeShortBytes(msg.statementId.bytes, dest);
-            if (ProtocolFeatures.PREPARED_METADATA_CHANGES.isSupportedBy(version, featureStore))
-              CBUtil.writeShortBytes(msg.resultMetadataId.bytes, dest);
+            if (ProtocolFeatures.PREPARED_METADATA_CHANGES.isSupportedBy(version, featureStore)) {
+              byte[] bytes =
+                  msg.resultMetadataId != null ? msg.resultMetadataId.bytes : EMPTY_BYTES;
+              CBUtil.writeShortBytes(bytes, dest);
+            }
             msg.options.encode(dest, version);
           }
 
           @Override
           public int encodedSize(
               Execute msg, ProtocolVersion version, ProtocolFeatureStore featureStore) {
             int size = CBUtil.sizeOfShortBytes(msg.statementId.bytes);
-            if (ProtocolFeatures.PREPARED_METADATA_CHANGES.isSupportedBy(version, featureStore))
-              size += CBUtil.sizeOfShortBytes(msg.resultMetadataId.bytes);
+            if (ProtocolFeatures.PREPARED_METADATA_CHANGES.isSupportedBy(version, featureStore)) {
+              byte[] bytes =
+                  msg.resultMetadataId != null ? msg.resultMetadataId.bytes : EMPTY_BYTES;
+              size += CBUtil.sizeOfShortBytes(bytes);
+            }
             size += msg.options.encodedSize(version);
             return size;
           }

driver-core/src/test/java/com/datastax/driver/core/RequestsTest.java

@@ -0,0 +1,65 @@
+package com.datastax.driver.core;
+
+import static org.testng.Assert.assertEquals;
+
+import com.datastax.driver.core.utils.Bytes;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class RequestsTest {
+
+  public static class ExecuteTest {
+
+    @Test(groups = "unit", dataProvider = "shouldProperlyEncodeMessages")
+    void should_properly_encode_messages(
+        MD5Digest resultMetadataId,
+        ProtocolVersion protocolVersion,
+        ProtocolFeatureStore protocolFeatureStore,
+        int expectedSize) {
+      // given
+      ByteBuf destination = ByteBufAllocator.DEFAULT.buffer();
+      Requests.Execute request =
+          new Requests.Execute(
+              MD5Digest.wrap(Bytes.fromHexString("0xcafebabe").array()),
+              resultMetadataId,
+              Requests.QueryProtocolOptions.DEFAULT,
+              false);
+
+      // when
+      int size = Requests.Execute.coder.encodedSize(request, protocolVersion, protocolFeatureStore);
+      Requests.Execute.coder.encode(request, destination, protocolVersion, protocolFeatureStore);
+
+      // then
+      assertEquals(size, expectedSize);
+      assertEquals(destination.writerIndex(), expectedSize);
+    }
+
+    @DataProvider
+    Object[][] shouldProperlyEncodeMessages() {
+      return new Object[][] {
+        {
+          MD5Digest.wrap(Bytes.fromHexString("0xdeadbeef").array()),
+          ProtocolVersion.V4,
+          new ProtocolFeatureStore(null, null, null, false),
+          9
+        },
+        {
+          MD5Digest.wrap(Bytes.fromHexString("0xdeadbeef").array()),
+          ProtocolVersion.V4,
+          new ProtocolFeatureStore(null, null, null, true),
+          15
+        },
+        {null, ProtocolVersion.V4, new ProtocolFeatureStore(null, null, null, true), 11},
+        {
+          MD5Digest.wrap(Bytes.fromHexString("0xdeadbeef").array()),
+          ProtocolVersion.V5,
+          new ProtocolFeatureStore(null, null, null, false),
+          18
+        },
+        {null, ProtocolVersion.V5, new ProtocolFeatureStore(null, null, null, false), 14},
+      };
+    }
+  }
+}