[scylladb#597] Updating extension negotiation logic. Adding ProtocolFeatures to FrameEncoder and FrameDecoder constructors. Updating integration tests.

nikagra · dkropachev · commit 813c9d9fb44e · 2025-10-02T13:36:12.000-04:00
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ChannelFactory.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ChannelFactory.java
@@ -47,6 +47,7 @@
 import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
 import com.datastax.oss.driver.shaded.guava.common.base.Preconditions;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
+import com.datastax.oss.protocol.internal.ProtocolFeatures;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
@@ -256,7 +257,7 @@ private void connect(
         cf -> {
           if (connectFuture.isSuccess()) {
             Channel channel = connectFuture.channel();
-            DriverChannel driverChannel =
+            DriverChannel driverChannel = // driver channel init
                 new DriverChannel(endPoint, channel, context.getWriteCoalescer(), currentVersion);
             // If this is the first successful connection, remember the protocol version and
             // cluster name for future connections.
@@ -330,7 +331,7 @@ ChannelInitializer<Channel> initializer(
       CompletableFuture<DriverChannel> resultFuture) {
     return new ChannelFactoryInitializer(
         endPoint, protocolVersion, options, nodeMetricUpdater, resultFuture);
-  };
+  }
 
   class ChannelFactoryInitializer extends ChannelInitializer<Channel> {
 
@@ -422,10 +423,10 @@ protected void initChannel(Channel channel) {
         pipeline
             .addLast(
                 FRAME_TO_BYTES_ENCODER_NAME,
-                new FrameEncoder(context.getFrameCodec(), maxFrameLength))
+                new FrameEncoder(context.getFrameCodec(), new ProtocolFeatures(), maxFrameLength))
             .addLast(
                 BYTES_TO_FRAME_DECODER_NAME,
-                new FrameDecoder(context.getFrameCodec(), maxFrameLength))
+                new FrameDecoder(context.getFrameCodec(), new ProtocolFeatures(), maxFrameLength))
             // Note: HeartbeatHandler is inserted here once init completes
             .addLast(INFLIGHT_HANDLER_NAME, inFlightHandler)
             .addLast(INIT_HANDLER_NAME, initHandler);
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannel.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannel.java
@@ -38,6 +38,7 @@
 import com.datastax.oss.driver.internal.core.session.DefaultSession;
 import com.datastax.oss.driver.internal.core.util.concurrent.UncaughtExceptions;
 import com.datastax.oss.protocol.internal.Message;
+import com.datastax.oss.protocol.internal.ProtocolFeatures;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelFuture;
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/protocol/FrameDecoder.java b/core/src/main/java/com/datastax/oss/driver/internal/core/protocol/FrameDecoder.java
@@ -22,6 +22,7 @@
 import com.datastax.oss.protocol.internal.Frame;
 import com.datastax.oss.protocol.internal.FrameCodec;
 import com.datastax.oss.protocol.internal.ProtocolConstants;
+import com.datastax.oss.protocol.internal.ProtocolFeatures;
 import com.datastax.oss.protocol.internal.response.Error;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
@@ -41,11 +42,16 @@ public class FrameDecoder extends LengthFieldBasedFrameDecoder {
   private static final int LENGTH_FIELD_LENGTH = 4;
 
   private final FrameCodec<ByteBuf> frameCodec;
+  private final ProtocolFeatures protocolFeatures;
   private boolean isFirstResponse;
 
-  public FrameDecoder(FrameCodec<ByteBuf> frameCodec, int maxFrameLengthInBytes) {
+  public FrameDecoder(
+      FrameCodec<ByteBuf> frameCodec,
+      ProtocolFeatures protocolFeatures,
+      int maxFrameLengthInBytes) {
     super(maxFrameLengthInBytes, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH, 0, 0, true);
     this.frameCodec = frameCodec;
+    this.protocolFeatures = protocolFeatures;
   }
 
   @Override
@@ -87,7 +93,7 @@ protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception
       ByteBuf buffer = (ByteBuf) super.decode(ctx, in);
       return (buffer == null)
           ? null // did not receive whole frame yet, keep reading
-          : frameCodec.decode(buffer);
+          : frameCodec.decode(buffer, protocolFeatures);
     } catch (Exception e) {
       // If decoding failed, try to read at least the stream id, so that the error can be
       // propagated to the client request matching that id (otherwise we have to fail all
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/protocol/FrameEncoder.java b/core/src/main/java/com/datastax/oss/driver/internal/core/protocol/FrameEncoder.java
@@ -20,6 +20,7 @@
 import com.datastax.oss.driver.api.core.connection.FrameTooLongException;
 import com.datastax.oss.protocol.internal.Frame;
 import com.datastax.oss.protocol.internal.FrameCodec;
+import com.datastax.oss.protocol.internal.ProtocolFeatures;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -32,17 +33,20 @@
 public class FrameEncoder extends MessageToMessageEncoder<Frame> {
 
   private final FrameCodec<ByteBuf> frameCodec;
+  private final ProtocolFeatures protocolFeatures;
   private final int maxFrameLength;
 
-  public FrameEncoder(FrameCodec<ByteBuf> frameCodec, int maxFrameLength) {
+  public FrameEncoder(
+      FrameCodec<ByteBuf> frameCodec, ProtocolFeatures protocolFeatures, int maxFrameLength) {
     super(Frame.class);
     this.frameCodec = frameCodec;
+    this.protocolFeatures = protocolFeatures;
     this.maxFrameLength = maxFrameLength;
   }
 
   @Override
   protected void encode(ChannelHandlerContext ctx, Frame frame, List<Object> out) throws Exception {
-    ByteBuf buffer = frameCodec.encode(frame);
+    ByteBuf buffer = frameCodec.encode(frame, protocolFeatures);
     int actualLength = buffer.readableBytes();
     if (actualLength > maxFrameLength) {
       throw new FrameTooLongException(
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/protocol/MetadataIdInfo.java b/core/src/main/java/com/datastax/oss/driver/internal/core/protocol/MetadataIdInfo.java
@@ -7,25 +7,16 @@ 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() {}
 
-  private MetadataIdInfo(boolean enabled) {
-    this.enabled = enabled;
-  }
-
-  public boolean isEnabled() {
-    return enabled;
-  }
-
-  public static MetadataIdInfo parseMetadataId(Map<String, List<String>> supported) {
+  public static boolean parseMetadataId(Map<String, List<String>> supported) {
     if (!supported.containsKey(SCYLLA_USE_METADATA_ID_STARTUP_OPTION_KEY)) {
-      return new MetadataIdInfo(false);
+      return 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));
+    return values != null
+        && values.size() == 1
+        && values.get(0).equals(SCYLLA_USE_METADATA_ID_STARTUP_OPTION_VALUE);
   }
 
   public static void addOption(Map<String, String> options) {
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/protocol/SegmentToFrameDecoder.java b/core/src/main/java/com/datastax/oss/driver/internal/core/protocol/SegmentToFrameDecoder.java
@@ -19,6 +19,7 @@
 
 import com.datastax.oss.protocol.internal.Frame;
 import com.datastax.oss.protocol.internal.FrameCodec;
+import com.datastax.oss.protocol.internal.ProtocolFeatures;
 import com.datastax.oss.protocol.internal.Segment;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import io.netty.buffer.ByteBuf;
@@ -73,7 +74,7 @@ private void decodeSelfContained(Segment<ByteBuf> segment, List<Object> out) {
     int frameCount = 0;
     try {
       do {
-        Frame frame = frameCodec.decode(payload);
+        Frame frame = frameCodec.decode(payload, new ProtocolFeatures());
         LOG.trace(
             "[{}] Decoded response frame {} from self-contained segment",
             logPrefix,
@@ -110,7 +111,7 @@ private void decodeSlice(Segment<ByteBuf> segment, ByteBufAllocator allocator, L
       encodedFrame.addComponents(true, accumulatedSlices);
       Frame frame;
       try {
-        frame = frameCodec.decode(encodedFrame);
+        frame = frameCodec.decode(encodedFrame, new ProtocolFeatures());
       } finally {
         encodedFrame.release();
         // Reset our state
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerTest.java
@@ -84,7 +84,7 @@ public class ProtocolInitHandlerTest extends ChannelHandlerTestBase {
   @Mock private DriverExecutionProfile defaultProfile;
   @Mock private Appender<ILoggingEvent> appender;
 
-  private ProtocolVersionRegistry protocolVersionRegistry =
+  private final ProtocolVersionRegistry protocolVersionRegistry =
       new DefaultProtocolVersionRegistry("test");
   private HeartbeatHandler heartbeatHandler;
 
@@ -365,7 +365,7 @@ public void should_invoke_auth_provider_when_server_does_not_send_challenge() {
   }
 
   @Test
-  public void should_fail_to_initialize_if_server_sends_auth_error() throws Throwable {
+  public void should_fail_to_initialize_if_server_sends_auth_error() {
     channel
         .pipeline()
         .addLast(
@@ -443,7 +443,7 @@ public void should_check_cluster_name_if_provided() {
   }
 
   @Test
-  public void should_fail_to_initialize_if_cluster_name_does_not_match() throws Throwable {
+  public void should_fail_to_initialize_if_cluster_name_does_not_match() {
     channel
         .pipeline()
         .addLast(
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/protocol/FrameDecoderTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/protocol/FrameDecoderTest.java
@@ -26,6 +26,7 @@
 import com.datastax.oss.protocol.internal.Compressor;
 import com.datastax.oss.protocol.internal.Frame;
 import com.datastax.oss.protocol.internal.FrameCodec;
+import com.datastax.oss.protocol.internal.ProtocolFeatures;
 import com.datastax.oss.protocol.internal.response.AuthSuccess;
 import io.netty.buffer.ByteBuf;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
@@ -63,7 +64,7 @@ public void setup() {
   @Test
   public void should_decode_valid_payload() {
     // Given
-    FrameDecoder decoder = new FrameDecoder(frameCodec, 1024);
+    FrameDecoder decoder = new FrameDecoder(frameCodec, new ProtocolFeatures(), 1024);
     channel.pipeline().addLast(decoder);
 
     // When
@@ -83,7 +84,8 @@ public void should_decode_valid_payload() {
   @Test
   public void should_fail_to_decode_if_payload_is_valid_but_too_long() {
     // Given
-    FrameDecoder decoder = new FrameDecoder(frameCodec, VALID_PAYLOAD.readableBytes() - 1);
+    FrameDecoder decoder =
+        new FrameDecoder(frameCodec, new ProtocolFeatures(), VALID_PAYLOAD.readableBytes() - 1);
     channel.pipeline().addLast(decoder);
 
     // When
@@ -102,7 +104,7 @@ public void should_fail_to_decode_if_payload_is_valid_but_too_long() {
   @Test
   public void should_fail_to_decode_if_payload_cannot_be_decoded() {
     // Given
-    FrameDecoder decoder = new FrameDecoder(frameCodec, 1024);
+    FrameDecoder decoder = new FrameDecoder(frameCodec, new ProtocolFeatures(), 1024);
     channel.pipeline().addLast(decoder);
 
     // When
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/protocol/SegmentToFrameDecoderTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/protocol/SegmentToFrameDecoderTest.java
@@ -25,6 +25,7 @@
 import com.datastax.oss.protocol.internal.FrameCodec;
 import com.datastax.oss.protocol.internal.Message;
 import com.datastax.oss.protocol.internal.ProtocolConstants;
+import com.datastax.oss.protocol.internal.ProtocolFeatures;
 import com.datastax.oss.protocol.internal.ProtocolV5ClientCodecs;
 import com.datastax.oss.protocol.internal.ProtocolV5ServerCodecs;
 import com.datastax.oss.protocol.internal.Segment;
@@ -93,6 +94,6 @@ private static ByteBuf encodeFrame(Message message) {
             Collections.emptyMap(),
             Collections.emptyList(),
             message);
-    return FRAME_CODEC.encode(frame);
+    return FRAME_CODEC.encode(frame, new ProtocolFeatures());
   }
 }
diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/cql/PreparedStatementIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/cql/PreparedStatementIT.java
@@ -79,9 +79,9 @@
 @Category(ParallelizableTests.class)
 public class PreparedStatementIT {
 
-  private CcmRule ccmRule = CcmRule.getInstance();
+  private final CcmRule ccmRule = CcmRule.getInstance();
 
-  private SessionRule<CqlSession> sessionRule =
+  private final SessionRule<CqlSession> sessionRule =
       SessionRule.builder(ccmRule)
           .withConfigLoader(
               SessionUtils.configLoaderBuilder()
@@ -159,8 +159,8 @@ public void should_have_non_empty_variable_definitions_for_select_query_with_bou
 
   @Test
   @BackendRequirement(type = BackendType.CASSANDRA, minInclusive = "4.0")
-  @BackendRequirement(type = BackendType.SCYLLA)
-  public void should_update_metadata_when_schema_changed_across_executions() {
+  @BackendRequirement(type = BackendType.SCYLLA, maxExclusive = "2025.3")
+  public void should_update_metadata_when_schema_changed_across_executions_with_no_metadata_id_feature() {
     // Given
     CqlSession session = sessionRule.session();
     PreparedStatement ps = session.prepare("SELECT * FROM prepared_statement_test WHERE a = ?");
@@ -205,6 +205,37 @@ public void should_update_metadata_when_schema_changed_across_executions() {
     }
   }
 
+  @Test
+  @BackendRequirement(type = BackendType.SCYLLA, minInclusive = "2025.3", description = "SCYLLA_USE_METADATA_ID feature is added in Scylla 2025.3")
+  public void should_update_metadata_when_schema_changed_across_executions_with_metadata_id_feature_support() {
+    // Given
+    CqlSession session = sessionRule.session();
+    PreparedStatement ps = session.prepare("SELECT * FROM prepared_statement_test WHERE a = ?");
+    ByteBuffer idBefore = ps.getResultMetadataId();
+    assertThat(idBefore).isNotNull();
+
+    // When
+    session.execute(
+      SimpleStatement.builder("ALTER TABLE prepared_statement_test ADD d int")
+        .setExecutionProfile(sessionRule.slowProfile())
+        .build());
+    BoundStatement bs = ps.bind(1);
+    ResultSet rows = session.execute(bs);
+
+    // Then
+    ByteBuffer idAfter = ps.getResultMetadataId();
+    assertThat(idAfter).isNotNull();
+    assertThat(Bytes.toHexString(idAfter)).isNotEqualTo(Bytes.toHexString(idBefore));
+    for (ColumnDefinitions columnDefinitions :
+      ImmutableList.of(
+        ps.getResultSetDefinitions(),
+        bs.getPreparedStatement().getResultSetDefinitions(),
+        rows.getColumnDefinitions())) {
+      assertThat(columnDefinitions).hasSize(4);
+      assertThat(columnDefinitions.get("d").getType()).isEqualTo(DataTypes.INT);
+    }
+  }
+
   @Test
   @BackendRequirement(type = BackendType.CASSANDRA, minInclusive = "4.0")
   @BackendRequirement(type = BackendType.SCYLLA)
diff --git a/test-infra/src/main/java/com/datastax/oss/driver/api/testinfra/ccm/CcmBridge.java b/test-infra/src/main/java/com/datastax/oss/driver/api/testinfra/ccm/CcmBridge.java
@@ -284,7 +284,7 @@ private static Version parseCcmVersion() {
     return result;
   }
 
-  public Optional<Version> getScyllaVersion() {
+  public static Optional<Version> getScyllaVersion() {
     return isDistributionOf(BackendType.SCYLLA) ? Optional.of(VERSION) : Optional.empty();
   }
 
@@ -628,12 +628,12 @@ public String getNodeIpAddress(int nodeId) {
     return ipPrefix + nodeId;
   }
 
-  private static String IN_MS_STR = "_in_ms";
-  private static int IN_MS_STR_LENGTH = IN_MS_STR.length();
-  private static String ENABLE_STR = "enable_";
-  private static int ENABLE_STR_LENGTH = ENABLE_STR.length();
-  private static String IN_KB_STR = "_in_kb";
-  private static int IN_KB_STR_LENGTH = IN_KB_STR.length();
+  private static final String IN_MS_STR = "_in_ms";
+  private static final int IN_MS_STR_LENGTH = IN_MS_STR.length();
+  private static final String ENABLE_STR = "enable_";
+  private static final int ENABLE_STR_LENGTH = ENABLE_STR.length();
+  private static final String IN_KB_STR = "_in_kb";
+  private static final int IN_KB_STR_LENGTH = IN_KB_STR.length();
 
   @SuppressWarnings("unused")
   private String getConfigKey(String originalKey, Object originalValue, Version cassandraVersion) {
