[#567] Initial implementation of SCYLLA_USE_METADATA_ID feature negotiation during driver startup 🚀

Author: nikagra

core/src/main/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandler.java

@@ -41,6 +41,7 @@
 import com.datastax.oss.driver.internal.core.protocol.BytesToSegmentDecoder;
 import com.datastax.oss.driver.internal.core.protocol.FrameToSegmentEncoder;
 import com.datastax.oss.driver.internal.core.protocol.LwtInfo;
+import com.datastax.oss.driver.internal.core.protocol.MetadataIdInfo;
 import com.datastax.oss.driver.internal.core.protocol.SegmentToBytesEncoder;
 import com.datastax.oss.driver.internal.core.protocol.SegmentToFrameDecoder;
 import com.datastax.oss.driver.internal.core.protocol.ShardingInfo;
@@ -98,6 +99,7 @@ class ProtocolInitHandler extends ConnectInitHandler {
   private final boolean querySupportedOptions;
   private LwtInfo lwtInfo;
   private TabletInfo tabletInfo;
+  private MetadataIdInfo metadataIdInfo;
 
   /**
    * @param querySupportedOptions whether to send OPTIONS as the first message, to request which
@@ -198,6 +200,9 @@ Message getRequest() {
           if (tabletInfo != null && tabletInfo.isEnabled()) {
             TabletInfo.addOption(startupOptions);
           }
+          if (metadataIdInfo != null && metadataIdInfo.isEnabled()) {
+            MetadataIdInfo.addOption(startupOptions);
+          }
           return request = new Startup(startupOptions);
         case GET_CLUSTER_NAME:
           return request = CLUSTER_NAME_QUERY;
@@ -227,17 +232,20 @@ void onResponse(Message response) {
           ProtocolUtils.opcodeString(response.opcode));
       try {
         if (step == Step.OPTIONS && response instanceof Supported) {
-          channel.attr(DriverChannel.OPTIONS_KEY).set(((Supported) response).options);
-          Supported res = (Supported) response;
-          ConnectionShardingInfo shardingInfo = ShardingInfo.parseShardingInfo(res.options);
+          Supported supported = (Supported) response;
+          channel.attr(DriverChannel.OPTIONS_KEY).set(supported.options);
+          ConnectionShardingInfo shardingInfo = ShardingInfo.parseShardingInfo(supported.options);
           if (shardingInfo != null) {
             channel.attr(DriverChannel.SHARDING_INFO_KEY).set(shardingInfo);
           }
-          lwtInfo = LwtInfo.parseLwtInfo(res.options);
+          lwtInfo = LwtInfo.parseLwtInfo(supported.options);
           if (lwtInfo != null) {
             channel.attr(LWT_INFO_KEY).set(lwtInfo);
           }
-          tabletInfo = TabletInfo.parseTabletInfo(res.options);
+          tabletInfo = TabletInfo.parseTabletInfo(supported.options);
+          metadataIdInfo =
+              MetadataIdInfo.parseMetadataId(
+                  supported.options); // TODO: Check if setting channel attribute is needed
           step = Step.STARTUP;
           send();
         } else if (step == Step.STARTUP && response instanceof Ready) {

core/src/main/java/com/datastax/oss/driver/internal/core/protocol/MetadataIdInfo.java

@@ -0,0 +1,35 @@
+package com.datastax.oss.driver.internal.core.protocol;
+
+import java.util.List;
+import java.util.Map;
+
+public class MetadataIdInfo {
+  private static final String SCYLLA_USE_METADATA_ID_STARTUP_OPTION_KEY = "SCYLLA_USE_METADATA_ID";
+  private static final String SCYLLA_USE_METADATA_ID_STARTUP_OPTION_VALUE = "";
+
+  private final boolean enabled;
+
+  private MetadataIdInfo(boolean enabled) {
+    this.enabled = enabled;
+  }
+
+  public boolean isEnabled() {
+    return enabled;
+  }
+
+  public static MetadataIdInfo parseMetadataId(Map<String, List<String>> supported) {
+    if (!supported.containsKey(SCYLLA_USE_METADATA_ID_STARTUP_OPTION_KEY)) {
+      return new MetadataIdInfo(false);
+    }
+    List<String> values = supported.get(SCYLLA_USE_METADATA_ID_STARTUP_OPTION_KEY);
+    return new MetadataIdInfo(
+        values != null
+            && values.size() == 1
+            && values.get(0).equals(SCYLLA_USE_METADATA_ID_STARTUP_OPTION_VALUE));
+  }
+
+  public static void addOption(Map<String, String> options) {
+    options.put(
+        SCYLLA_USE_METADATA_ID_STARTUP_OPTION_KEY, SCYLLA_USE_METADATA_ID_STARTUP_OPTION_VALUE);
+  }
+}